109 lines
2.4 KiB
Go
109 lines
2.4 KiB
Go
package logs
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"fmt"
|
|
"net/http"
|
|
"time"
|
|
)
|
|
|
|
type LokiWriter struct {
|
|
url string
|
|
labels map[string]string
|
|
httpClient *http.Client
|
|
}
|
|
|
|
type LokiPayload struct {
|
|
Streams []LokiStream `json:"streams"`
|
|
}
|
|
|
|
type LokiStream struct {
|
|
Stream map[string]string `json:"stream"`
|
|
Values [][]string `json:"values"`
|
|
}
|
|
|
|
func NewLokiWriter(url string, labels map[string]string) *LokiWriter {
|
|
return &LokiWriter{
|
|
url: url,
|
|
labels: labels,
|
|
httpClient: &http.Client{},
|
|
}
|
|
}
|
|
|
|
func (w *LokiWriter) Write(p []byte) (n int, err error) {
|
|
// Use zerolog to parse the log level
|
|
var event map[string]interface{}
|
|
if err := json.Unmarshal(p, &event); err != nil {
|
|
return 0, fmt.Errorf("failed to unmarshal log event: %w", err)
|
|
}
|
|
|
|
level := ""
|
|
if l, ok := event["level"].(string); ok {
|
|
level = l
|
|
}
|
|
|
|
message := ""
|
|
if m, ok := event["message"].(string); ok {
|
|
message = m
|
|
}
|
|
|
|
// Add log level to labels
|
|
labels := make(map[string]string)
|
|
for k, v := range w.labels {
|
|
labels[k] = v
|
|
}
|
|
labels["level"] = level
|
|
|
|
// Add label that have been added to the event
|
|
// A bit unsafe since we don't know what could be stored in the event
|
|
// but we can't access this object once passed to the multilevel writter
|
|
|
|
for k,v := range(event){
|
|
if k != "level" && k != "time" && k != "message"{
|
|
labels[k] = v.(string)
|
|
}
|
|
}
|
|
|
|
|
|
// Format the timestamp in nanoseconds
|
|
timestamp := fmt.Sprintf("%d000000", time.Now().UnixNano()/int64(time.Millisecond))
|
|
|
|
stream := LokiStream{
|
|
Stream: labels,
|
|
Values: [][]string{
|
|
{timestamp, message},
|
|
},
|
|
}
|
|
|
|
payload := LokiPayload{
|
|
Streams: []LokiStream{stream},
|
|
}
|
|
|
|
payloadBytes, err := json.Marshal(payload)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("failed to marshal payload: %w", err)
|
|
}
|
|
|
|
//fmt.Printf("Sending payload to Loki: %s\n", string(payloadBytes))
|
|
|
|
req, err := http.NewRequest("POST", w.url + "/loki/api/v1/push", bytes.NewReader(payloadBytes))
|
|
if err != nil {
|
|
return 0, fmt.Errorf("failed to create HTTP request: %w", err)
|
|
}
|
|
req.Header.Set("Content-Type", "application/json")
|
|
|
|
resp, err := w.httpClient.Do(req)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("failed to send log to Loki: %w", err)
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
//fmt.Printf("Loki response status: %d\n", resp.StatusCode)
|
|
if resp.StatusCode != http.StatusNoContent {
|
|
return 0, fmt.Errorf("received non-204 response from Loki: %d", resp.StatusCode)
|
|
}
|
|
|
|
return len(p), nil
|
|
}
|