fix: write every influx field on every point

InfluxDB 3 creates a column on its first write. A dashboard query that
names ttft_ms or error failed until a streamed or failed request had
been exported. Write 0 and empty strings instead of omitting the field.

README.md

@@ -136,14 +136,18 @@ the exporter never delays a response. Each point has these tags and fields:
 
 | Kind | Name | Notes |
 |---|---|---|
-| tag | `provider`, `model`, `key_prefix` | group by provider, model address, or key |
+| tag | `provider`, `model`, `key_prefix` | group by provider, model address, or key. `key_prefix` is `(no key)` for the dashboard chat |
 | tag | `user_id`, `provider_id`, `provider_model_id` | UUIDs, set when known |
 | tag | `status_code`, `streamed` | HTTP status and `true`/`false` |
 | field | `requests`, `errors` | always `1` and `0` or `1`, so `sum()` gives counts |
 | field | `prompt_tokens`, `completion_tokens`, `cached_tokens`, `total_tokens` | integers |
 | field | `cost` | dollars, float |
-| field | `latency_ms`, `ttft_ms` | integers, `ttft_ms` only for streamed chat |
-| field | `request_id`, `finish_reason`, `error` | strings, set when known |
+| field | `latency_ms`, `ttft_ms` | integers. `ttft_ms` is `0` when the response was not streamed |
+| field | `request_id`, `finish_reason`, `error` | strings, empty when not known |
+
+Every field is written on every point. InfluxDB 3 creates a column on its first
+write, and a query that names a missing column fails, so optional values are
+written as `0` or an empty string.
 
 Generate an encryption key:
 

backend/internal/influx/influx.go

@@ -133,12 +133,18 @@ func (e *Exporter) Enqueue(p Point) {
 }
 
 // requestPoint maps a log row to the request measurement. Bounded values are
-// tags, so dashboards can group by them. Per-request values are fields.
+// tags, so dashboards can group by them. Per-request values are fields. Every
+// field is written on every point, because InfluxDB 3 creates a column on its
+// first write and a query that names a missing column fails.
 func (e *Exporter) requestPoint(rec store.RequestLog) Point {
+	keyPrefix := rec.KeyPrefix
+	if keyPrefix == "" {
+		keyPrefix = "(no key)"
+	}
 	tags := map[string]string{
 		"provider":    rec.ProviderName,
 		"model":       rec.ModelName,
-		"key_prefix":  rec.KeyPrefix,
+		"key_prefix":  keyPrefix,
 		"status_code": strconv.Itoa(rec.StatusCode),
 		"streamed":    strconv.FormatBool(rec.Streamed),
 	}
@@ -156,6 +162,15 @@ func (e *Exporter) requestPoint(rec store.RequestLog) Point {
 	if rec.StatusCode >= 400 || rec.Error != "" {
 		isErr = 1
 	}
+	// ttft_ms is 0 when the response was not streamed.
+	ttft := 0
+	if rec.TTFTMs != nil {
+		ttft = *rec.TTFTMs
+	}
+	msg := rec.Error
+	if len(msg) > maxErrorLen {
+		msg = msg[:maxErrorLen]
+	}
 	fields := map[string]any{
 		"requests":          1,
 		"prompt_tokens":     rec.PromptTokens,
@@ -164,23 +179,11 @@ func (e *Exporter) requestPoint(rec store.RequestLog) Point {
 		"total_tokens":      rec.TotalTokens,
 		"cost":              rec.Cost,
 		"latency_ms":        rec.LatencyMS,
+		"ttft_ms":           ttft,
 		"errors":            isErr,
-	}
-	if rec.TTFTMs != nil {
-		fields["ttft_ms"] = *rec.TTFTMs
-	}
-	if rec.RequestID != "" {
-		fields["request_id"] = rec.RequestID
-	}
-	if rec.FinishReason != "" {
-		fields["finish_reason"] = rec.FinishReason
-	}
-	if rec.Error != "" {
-		msg := rec.Error
-		if len(msg) > maxErrorLen {
-			msg = msg[:maxErrorLen]
-		}
-		fields["error"] = msg
+		"request_id":        rec.RequestID,
+		"finish_reason":     rec.FinishReason,
+		"error":             msg,
 	}
 	return Point{Measurement: e.cfg.Measurement, Tags: tags, Fields: fields, Time: e.now()}
 }

backend/internal/influx/influx_test.go

@@ -166,7 +166,7 @@ func TestRecordWritesRequestPoint(t *testing.T) {
 		t.Fatalf("auth = %q", srv.auth)
 	}
 	want := `lmr_request,key_prefix=lmr_abc,model=openai/gpt-4o,provider=openai,status_code=200,streamed=true,user_id=11111111-1111-1111-1111-111111111111 ` +
-		`cached_tokens=2i,completion_tokens=5i,cost=0.00125,errors=0i,latency_ms=321i,prompt_tokens=10i,request_id="req-1",requests=1i,total_tokens=15i,ttft_ms=42i 100000000000` + "\n"
+		`cached_tokens=2i,completion_tokens=5i,cost=0.00125,error="",errors=0i,finish_reason="",latency_ms=321i,prompt_tokens=10i,request_id="req-1",requests=1i,total_tokens=15i,ttft_ms=42i 100000000000` + "\n"
 	if len(srv.bodies) != 1 || srv.bodies[0] != want {
 		t.Fatalf("bodies = %q\nwant %q", srv.bodies, want)
 	}
@@ -189,6 +189,25 @@ func TestRecordMarksErrors(t *testing.T) {
 	}
 }
 
+// Every field must be present on every point so the columns exist from the
+// first write. Optional values get an empty string or zero.
+func TestRequestPointWritesAllFields(t *testing.T) {
+	srv := newCaptureServer(t)
+	e := newTestExporter(t, srv, Config{})
+	p := e.requestPoint(store.RequestLog{StatusCode: 200})
+	for _, k := range []string{"ttft_ms", "error", "finish_reason", "request_id"} {
+		if _, ok := p.Fields[k]; !ok {
+			t.Errorf("field %s missing", k)
+		}
+	}
+	if p.Fields["ttft_ms"] != 0 || p.Fields["error"] != "" {
+		t.Errorf("fields = %v", p.Fields)
+	}
+	if p.Tags["key_prefix"] != "(no key)" {
+		t.Errorf("key_prefix = %q", p.Tags["key_prefix"])
+	}
+}
+
 func TestBatchAndFlushOnClose(t *testing.T) {
 	srv := newCaptureServer(t)
 	e := newTestExporter(t, srv, Config{FlushInterval: time.Hour, BatchSize: 2})