influx fix
This commit is contained in:
@@ -4,8 +4,11 @@ import (
|
|||||||
"bytes"
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"io"
|
||||||
|
"math"
|
||||||
"net/http"
|
"net/http"
|
||||||
"sort"
|
"sort"
|
||||||
|
"strconv"
|
||||||
"strings"
|
"strings"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -65,9 +68,8 @@ func (e *InfluxDBExporter) Write(ctx context.Context, metrics []core.Metric) err
|
|||||||
defer resp.Body.Close()
|
defer resp.Body.Close()
|
||||||
|
|
||||||
if resp.StatusCode >= 300 {
|
if resp.StatusCode >= 300 {
|
||||||
buf := make([]byte, 256)
|
msg, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
|
||||||
n, _ := resp.Body.Read(buf)
|
return fmt.Errorf("influxdb: HTTP %d: %s", resp.StatusCode, string(msg))
|
||||||
return fmt.Errorf("influxdb: HTTP %d: %s", resp.StatusCode, string(buf[:n]))
|
|
||||||
}
|
}
|
||||||
|
|
||||||
return nil
|
return nil
|
||||||
@@ -80,23 +82,22 @@ func (e *InfluxDBExporter) Write(ctx context.Context, metrics []core.Metric) err
|
|||||||
// Each metric becomes one line:
|
// Each metric becomes one line:
|
||||||
//
|
//
|
||||||
// measurement,tag1=val1,tag2=val2 value=<float> <timestamp_ns>
|
// 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 {
|
func EncodeLineProtocol(metrics []core.Metric) string {
|
||||||
if len(metrics) == 0 {
|
lines := make([]string, 0, len(metrics))
|
||||||
return ""
|
|
||||||
}
|
|
||||||
|
|
||||||
var b strings.Builder
|
for _, m := range metrics {
|
||||||
b.Grow(len(metrics) * 128)
|
if m.Name == "" || math.IsNaN(m.Value) || math.IsInf(m.Value, 0) {
|
||||||
|
continue
|
||||||
for i, m := range metrics {
|
|
||||||
if i > 0 {
|
|
||||||
b.WriteByte('\n')
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Measurement
|
var b strings.Builder
|
||||||
b.WriteString(escapeMeasurement(m.Name))
|
b.WriteString(escapeMeasurement(m.Name))
|
||||||
|
|
||||||
// Tags (sorted for determinism)
|
|
||||||
if len(m.Labels) > 0 {
|
if len(m.Labels) > 0 {
|
||||||
keys := make([]string, 0, len(m.Labels))
|
keys := make([]string, 0, len(m.Labels))
|
||||||
for k := range m.Labels {
|
for k := range m.Labels {
|
||||||
@@ -105,62 +106,42 @@ func EncodeLineProtocol(metrics []core.Metric) string {
|
|||||||
sort.Strings(keys)
|
sort.Strings(keys)
|
||||||
|
|
||||||
for _, k := range keys {
|
for _, k := range keys {
|
||||||
|
v := m.Labels[k]
|
||||||
|
if k == "" || v == "" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
b.WriteByte(',')
|
b.WriteByte(',')
|
||||||
b.WriteString(escapeTagKey(k))
|
b.WriteString(escapeTag(k))
|
||||||
b.WriteByte('=')
|
b.WriteByte('=')
|
||||||
b.WriteString(escapeTagValue(m.Labels[k]))
|
b.WriteString(escapeTag(v))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Fields
|
|
||||||
b.WriteString(" value=")
|
b.WriteString(" value=")
|
||||||
b.WriteString(formatFloat(m.Value))
|
b.WriteString(formatFloat(m.Value))
|
||||||
|
|
||||||
// Timestamp (nanoseconds since epoch)
|
|
||||||
if !m.Timestamp.IsZero() {
|
if !m.Timestamp.IsZero() {
|
||||||
b.WriteByte(' ')
|
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 {
|
var (
|
||||||
if strings.ContainsAny(s, ", ") {
|
measurementEscaper = strings.NewReplacer(",", `\,`, " ", `\ `)
|
||||||
var b strings.Builder
|
tagEscaper = strings.NewReplacer(",", `\,`, "=", `\=`, " ", `\ `)
|
||||||
for _, c := range s {
|
)
|
||||||
if c == ',' || c == ' ' {
|
|
||||||
b.WriteByte('\\')
|
|
||||||
}
|
|
||||||
b.WriteRune(c)
|
|
||||||
}
|
|
||||||
return b.String()
|
|
||||||
}
|
|
||||||
return s
|
|
||||||
}
|
|
||||||
|
|
||||||
func escapeTagKey(s string) string {
|
func escapeMeasurement(s string) string { return measurementEscaper.Replace(s) }
|
||||||
return escapeTagValue(s)
|
|
||||||
}
|
|
||||||
|
|
||||||
func escapeTagValue(s string) string {
|
func escapeTag(s string) string { return tagEscaper.Replace(s) }
|
||||||
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
|
|
||||||
}
|
|
||||||
|
|
||||||
|
// 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 {
|
func formatFloat(f float64) string {
|
||||||
if f == float64(int64(f)) && f < 9.223372036854775807e18 && f > -9.223372036854775808e18 {
|
return strconv.FormatFloat(f, 'f', -1, 64)
|
||||||
return fmt.Sprintf("%di", int64(f))
|
|
||||||
}
|
|
||||||
return fmt.Sprintf("%g", f)
|
|
||||||
}
|
}
|
||||||
|
|||||||
6
main.go
6
main.go
@@ -2,6 +2,7 @@ package main
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"flag"
|
||||||
"log"
|
"log"
|
||||||
"os/signal"
|
"os/signal"
|
||||||
"syscall"
|
"syscall"
|
||||||
@@ -14,7 +15,10 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
func main() {
|
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 {
|
if err != nil {
|
||||||
log.Fatalf("config: %v", err)
|
log.Fatalf("config: %v", err)
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user