From 5807cc5d20292a5a9a58036c1b331851e8804106 Mon Sep 17 00:00:00 2001 From: LovroG05 Date: Sun, 11 Oct 2026 15:08:52 +0200 Subject: [PATCH] influx fix --- exporters/influxdb.go | 87 +++++++++++++++++-------------------------- main.go | 6 ++- 2 files changed, 39 insertions(+), 54 deletions(-) diff --git a/exporters/influxdb.go b/exporters/influxdb.go index 8795ed6..3637dd7 100644 --- a/exporters/influxdb.go +++ b/exporters/influxdb.go @@ -4,8 +4,11 @@ import ( "bytes" "context" "fmt" + "io" + "math" "net/http" "sort" + "strconv" "strings" "time" @@ -65,9 +68,8 @@ func (e *InfluxDBExporter) Write(ctx context.Context, metrics []core.Metric) err defer resp.Body.Close() if resp.StatusCode >= 300 { - buf := make([]byte, 256) - n, _ := resp.Body.Read(buf) - return fmt.Errorf("influxdb: HTTP %d: %s", resp.StatusCode, string(buf[:n])) + msg, _ := io.ReadAll(io.LimitReader(resp.Body, 512)) + return fmt.Errorf("influxdb: HTTP %d: %s", resp.StatusCode, string(msg)) } return nil @@ -80,23 +82,22 @@ func (e *InfluxDBExporter) Write(ctx context.Context, metrics []core.Metric) err // Each metric becomes one line: // // measurement,tag1=val1,tag2=val2 value= +// +// 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 { - if len(metrics) == 0 { - return "" - } + lines := make([]string, 0, len(metrics)) - var b strings.Builder - b.Grow(len(metrics) * 128) - - for i, m := range metrics { - if i > 0 { - b.WriteByte('\n') + for _, m := range metrics { + if m.Name == "" || math.IsNaN(m.Value) || math.IsInf(m.Value, 0) { + continue } - // Measurement + var b strings.Builder b.WriteString(escapeMeasurement(m.Name)) - // Tags (sorted for determinism) if len(m.Labels) > 0 { keys := make([]string, 0, len(m.Labels)) for k := range m.Labels { @@ -105,62 +106,42 @@ func EncodeLineProtocol(metrics []core.Metric) string { sort.Strings(keys) for _, k := range keys { + v := m.Labels[k] + if k == "" || v == "" { + continue + } b.WriteByte(',') - b.WriteString(escapeTagKey(k)) + b.WriteString(escapeTag(k)) b.WriteByte('=') - b.WriteString(escapeTagValue(m.Labels[k])) + b.WriteString(escapeTag(v)) } } - // Fields b.WriteString(" value=") b.WriteString(formatFloat(m.Value)) - // Timestamp (nanoseconds since epoch) if !m.Timestamp.IsZero() { b.WriteByte(' ') - fmt.Fprintf(&b, "%d", m.Timestamp.UnixNano()) + b.WriteString(strconv.FormatInt(m.Timestamp.UnixNano(), 10)) } + + lines = append(lines, b.String()) } - return b.String() + return strings.Join(lines, "\n") } -func escapeMeasurement(s string) string { - if strings.ContainsAny(s, ", ") { - var b strings.Builder - for _, c := range s { - if c == ',' || c == ' ' { - b.WriteByte('\\') - } - b.WriteRune(c) - } - return b.String() - } - return s -} +var ( + measurementEscaper = strings.NewReplacer(",", `\,`, " ", `\ `) + tagEscaper = strings.NewReplacer(",", `\,`, "=", `\=`, " ", `\ `) +) -func escapeTagKey(s string) string { - return escapeTagValue(s) -} +func escapeMeasurement(s string) string { return measurementEscaper.Replace(s) } -func escapeTagValue(s string) string { - if strings.ContainsAny(s, ", =") { - var b strings.Builder - for _, c := range s { - if c == ',' || c == '=' || c == ' ' { - b.WriteByte('\\') - } - b.WriteRune(c) - } - return b.String() - } - return 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 { - if f == float64(int64(f)) && f < 9.223372036854775807e18 && f > -9.223372036854775808e18 { - return fmt.Sprintf("%di", int64(f)) - } - return fmt.Sprintf("%g", f) + return strconv.FormatFloat(f, 'f', -1, 64) } diff --git a/main.go b/main.go index 30f926c..84c7bc3 100644 --- a/main.go +++ b/main.go @@ -2,6 +2,7 @@ package main import ( "context" + "flag" "log" "os/signal" "syscall" @@ -14,7 +15,10 @@ import ( ) func main() { - cfg, err := config.Load("config/config.yaml") + cfgPath := flag.String("config", "config/config.yaml", "path to the YAML config file") + flag.Parse() + + cfg, err := config.Load(*cfgPath) if err != nil { log.Fatalf("config: %v", err) }