67 lines
1.3 KiB
Go
67 lines
1.3 KiB
Go
|
package loki
|
||
|
|
|
||
|
|
import (
|
||
|
|
"bytes"
|
||
|
|
"encoding/json"
|
||
|
|
"fmt"
|
||
|
|
"net/http"
|
||
|
|
"strings"
|
||
|
|
)
|
||
|
|
|
||
|
|
func (client *Client) sendLogs(entries []logEntry) {
|
||
|
|
if len(entries) == 0 {
|
||
|
|
return
|
||
|
|
}
|
||
|
|
|
||
|
|
var streams []stream
|
||
|
|
stream := stream{
|
||
|
|
Labels: client.config.Labels,
|
||
|
|
Entries: entries,
|
||
|
|
}
|
||
|
|
streams = append(streams, stream)
|
||
|
|
|
||
|
|
msg := promtailMsg{Streams: streams}
|
||
|
|
jsonMsg, err := json.Marshal(msg)
|
||
|
|
if err != nil {
|
||
|
|
return
|
||
|
|
}
|
||
|
|
|
||
|
|
client.mutex.RLock()
|
||
|
|
endpoint := client.config.Endpoint
|
||
|
|
client.mutex.RUnlock()
|
||
|
|
|
||
|
|
client.sendRequest("POST", endpoint, "application/json", jsonMsg)
|
||
|
|
}
|
||
|
|
|
||
|
|
func (client *Client) sendRequest(method, url string, ctype string, reqBody []byte) {
|
||
|
|
req, err := http.NewRequest(method, url, bytes.NewBuffer(reqBody))
|
||
|
|
if err != nil {
|
||
|
|
return
|
||
|
|
}
|
||
|
|
|
||
|
|
req.Header.Set("Content-Type", ctype)
|
||
|
|
|
||
|
|
resp, _ := client.client.Do(req)
|
||
|
|
if resp != nil {
|
||
|
|
defer resp.Body.Close()
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func (p *stream) MarshalJSON() ([]byte, error) {
|
||
|
|
return json.Marshal(&struct {
|
||
|
|
Labels string `json:"labels"`
|
||
|
|
Entries []logEntry `json:"entries"`
|
||
|
|
}{
|
||
|
|
Labels: p.Labels.String(),
|
||
|
|
Entries: p.Entries,
|
||
|
|
})
|
||
|
|
}
|
||
|
|
|
||
|
|
func (l labels) String() string {
|
||
|
|
var labelPairs []string
|
||
|
|
for key, value := range l {
|
||
|
|
labelPairs = append(labelPairs, fmt.Sprintf(`%s="%s"`, key, value))
|
||
|
|
}
|
||
|
|
return fmt.Sprintf("{%s}", strings.Join(labelPairs, ","))
|
||
|
|
}
|