148 lines
3.7 KiB
Go
148 lines
3.7 KiB
Go
package exporters
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"math"
|
|
"net/http"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"git.govekar.net/LovroG05/nvllama-watchdog/core"
|
|
)
|
|
|
|
type InfluxDBExporter struct {
|
|
url string
|
|
token string
|
|
user string
|
|
pass string
|
|
client *http.Client
|
|
}
|
|
|
|
// NewInfluxDBExporter creates an InfluxDB exporter.
|
|
//
|
|
// Version-specific URL examples:
|
|
//
|
|
// v1: http://localhost:8086/write?db=nvllama&rp=autogen
|
|
// v2: http://localhost:8086/api/v2/write?org=myorg&bucket=nvllama
|
|
// v3: http://localhost:8181/api/v3/write_lp?db=nvllama
|
|
func NewInfluxDBExporter(url, token, user, pass string) *InfluxDBExporter {
|
|
return &InfluxDBExporter{
|
|
url: url,
|
|
token: token,
|
|
user: user,
|
|
pass: pass,
|
|
client: &http.Client{Timeout: 10 * time.Second},
|
|
}
|
|
}
|
|
|
|
func (e *InfluxDBExporter) Name() string { return "influxdb" }
|
|
|
|
func (e *InfluxDBExporter) Write(ctx context.Context, metrics []core.Metric) error {
|
|
body := EncodeLineProtocol(metrics)
|
|
if body == "" {
|
|
return nil
|
|
}
|
|
|
|
req, err := http.NewRequestWithContext(ctx, http.MethodPost, e.url, bytes.NewReader([]byte(body)))
|
|
if err != nil {
|
|
return fmt.Errorf("influxdb: %w", err)
|
|
}
|
|
req.Header.Set("Content-Type", "text/plain; charset=utf-8")
|
|
|
|
// Auth: Bearer token (v2/v3) or basic auth (v1)
|
|
if e.token != "" {
|
|
req.Header.Set("Authorization", "Bearer "+e.token)
|
|
} else if e.user != "" {
|
|
req.SetBasicAuth(e.user, e.pass)
|
|
}
|
|
|
|
resp, err := e.client.Do(req)
|
|
if err != nil {
|
|
return fmt.Errorf("influxdb: %w", err)
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
if resp.StatusCode >= 300 {
|
|
msg, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
|
|
return fmt.Errorf("influxdb: HTTP %d: %s", resp.StatusCode, string(msg))
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
// ─── Line Protocol Encoder ──────────────────────────────────────────────────
|
|
|
|
// EncodeLineProtocol converts []core.Metric to InfluxDB line protocol.
|
|
//
|
|
// Each metric becomes one line:
|
|
//
|
|
// measurement,tag1=val1,tag2=val2 value=<float> <timestamp_ns>
|
|
//
|
|
// Values are always written as floats. InfluxDB fixes the type of a column
|
|
// at the first write, so a value that flips between integer and float
|
|
// syntax would be rejected later. NaN and Inf have no line protocol form
|
|
// and are skipped, as are empty tag values.
|
|
func EncodeLineProtocol(metrics []core.Metric) string {
|
|
lines := make([]string, 0, len(metrics))
|
|
|
|
for _, m := range metrics {
|
|
if m.Name == "" || math.IsNaN(m.Value) || math.IsInf(m.Value, 0) {
|
|
continue
|
|
}
|
|
|
|
var b strings.Builder
|
|
b.WriteString(escapeMeasurement(m.Name))
|
|
|
|
if len(m.Labels) > 0 {
|
|
keys := make([]string, 0, len(m.Labels))
|
|
for k := range m.Labels {
|
|
keys = append(keys, k)
|
|
}
|
|
sort.Strings(keys)
|
|
|
|
for _, k := range keys {
|
|
v := m.Labels[k]
|
|
if k == "" || v == "" {
|
|
continue
|
|
}
|
|
b.WriteByte(',')
|
|
b.WriteString(escapeTag(k))
|
|
b.WriteByte('=')
|
|
b.WriteString(escapeTag(v))
|
|
}
|
|
}
|
|
|
|
b.WriteString(" value=")
|
|
b.WriteString(formatFloat(m.Value))
|
|
|
|
if !m.Timestamp.IsZero() {
|
|
b.WriteByte(' ')
|
|
b.WriteString(strconv.FormatInt(m.Timestamp.UnixNano(), 10))
|
|
}
|
|
|
|
lines = append(lines, b.String())
|
|
}
|
|
|
|
return strings.Join(lines, "\n")
|
|
}
|
|
|
|
var (
|
|
measurementEscaper = strings.NewReplacer(",", `\,`, " ", `\ `)
|
|
tagEscaper = strings.NewReplacer(",", `\,`, "=", `\=`, " ", `\ `)
|
|
)
|
|
|
|
func escapeMeasurement(s string) string { return measurementEscaper.Replace(s) }
|
|
|
|
func escapeTag(s string) string { return tagEscaper.Replace(s) }
|
|
|
|
// formatFloat renders a float in line protocol float syntax. A value such
|
|
// as 100 is written as "100", which InfluxDB reads as a float.
|
|
func formatFloat(f float64) string {
|
|
return strconv.FormatFloat(f, 'f', -1, 64)
|
|
}
|