From 5e90a8fbc0b29b50b4af1306488464352f0f458e Mon Sep 17 00:00:00 2001 From: archandatta <35818003+archandatta@users.noreply.github.com> Date: Wed, 24 Jun 2026 14:33:06 +0000 Subject: [PATCH 01/11] Add telemetry event to OTLP log record converter Pure mapping from telemetry envelopes to OTLP log records: event type to EventName, structured JSON body, severity, and kernel.* attributes. Excludes screenshot/monitor categories from export. --- server/go.mod | 7 +- server/go.sum | 14 +-- server/lib/events/otlpconvert.go | 125 ++++++++++++++++++++++++++ server/lib/events/otlpconvert_test.go | 125 ++++++++++++++++++++++++++ 4 files changed, 262 insertions(+), 9 deletions(-) create mode 100644 server/lib/events/otlpconvert.go create mode 100644 server/lib/events/otlpconvert_test.go diff --git a/server/go.mod b/server/go.mod index f2518e7a..e37a7d71 100644 --- a/server/go.mod +++ b/server/go.mod @@ -27,6 +27,7 @@ require ( github.com/samber/lo v1.52.0 github.com/stretchr/testify v1.11.1 github.com/testcontainers/testcontainers-go v0.40.0 + go.opentelemetry.io/otel/log v0.20.0 golang.org/x/sync v0.18.0 golang.org/x/sys v0.39.0 golang.org/x/term v0.37.0 @@ -100,12 +101,12 @@ require ( github.com/yusufpapurcu/wmi v1.2.4 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 // indirect - go.opentelemetry.io/otel v1.39.0 // indirect + go.opentelemetry.io/otel v1.44.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.39.0 // indirect - go.opentelemetry.io/otel/metric v1.39.0 // indirect + go.opentelemetry.io/otel/metric v1.44.0 // indirect go.opentelemetry.io/otel/sdk v1.39.0 // indirect go.opentelemetry.io/otel/sdk/metric v1.39.0 // indirect - go.opentelemetry.io/otel/trace v1.39.0 // indirect + go.opentelemetry.io/otel/trace v1.44.0 // indirect golang.org/x/crypto v0.44.0 // indirect golang.org/x/mod v0.29.0 // indirect golang.org/x/net v0.47.0 // indirect diff --git a/server/go.sum b/server/go.sum index 1950c7a2..00214fb1 100644 --- a/server/go.sum +++ b/server/go.sum @@ -256,20 +256,22 @@ go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 h1:F7Jx+6hwnZ41NSFTO5q4LYDtJRXBf2PD0rNBkeB/lus= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0/go.mod h1:UHB22Z8QsdRDrnAtX4PntOl36ajSxcdUMt1sF7Y6E7Q= -go.opentelemetry.io/otel v1.39.0 h1:8yPrr/S0ND9QEfTfdP9V+SiwT4E0G7Y5MO7p85nis48= -go.opentelemetry.io/otel v1.39.0/go.mod h1:kLlFTywNWrFyEdH0oj2xK0bFYZtHRYUdv1NklR/tgc8= +go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU= +go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.39.0 h1:f0cb2XPmrqn4XMy9PNliTgRKJgS5WcL/u0/WRYGz4t0= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.39.0/go.mod h1:vnakAaFckOMiMtOIhFI2MNH4FYrZzXCYxmb1LlhoGz8= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.19.0 h1:IeMeyr1aBvBiPVYihXIaeIZba6b8E1bYp7lbdxK8CQg= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.19.0/go.mod h1:oVdCUtjq9MK9BlS7TtucsQwUcXcymNiEDjgDD2jMtZU= -go.opentelemetry.io/otel/metric v1.39.0 h1:d1UzonvEZriVfpNKEVmHXbdf909uGTOQjA0HF0Ls5Q0= -go.opentelemetry.io/otel/metric v1.39.0/go.mod h1:jrZSWL33sD7bBxg1xjrqyDjnuzTUB0x1nBERXd7Ftcs= +go.opentelemetry.io/otel/log v0.20.0 h1:/5i0vuHxCLWUfChWG41K9wkM0jafruPw9NU1/RCJirs= +go.opentelemetry.io/otel/log v0.20.0/go.mod h1:wOcMcjsZpG8x7Bak7IhSi/lg8wscV2C1VdrKCLPlt0E= +go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= +go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= go.opentelemetry.io/otel/sdk v1.39.0 h1:nMLYcjVsvdui1B/4FRkwjzoRVsMK8uL/cj0OyhKzt18= go.opentelemetry.io/otel/sdk v1.39.0/go.mod h1:vDojkC4/jsTJsE+kh+LXYQlbL8CgrEcwmt1ENZszdJE= go.opentelemetry.io/otel/sdk/metric v1.39.0 h1:cXMVVFVgsIf2YL6QkRF4Urbr/aMInf+2WKg+sEJTtB8= go.opentelemetry.io/otel/sdk/metric v1.39.0/go.mod h1:xq9HEVH7qeX69/JnwEfp6fVq5wosJsY1mt4lLfYdVew= -go.opentelemetry.io/otel/trace v1.39.0 h1:2d2vfpEDmCJ5zVYz7ijaJdOF59xLomrvj7bjt6/qCJI= -go.opentelemetry.io/otel/trace v1.39.0/go.mod h1:88w4/PnZSazkGzz/w84VHpQafiU4EtqqlVdxWy+rNOA= +go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= +go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= go.opentelemetry.io/proto/otlp v1.9.0 h1:l706jCMITVouPOqEnii2fIAuO3IVGBRPV5ICjceRb/A= go.opentelemetry.io/proto/otlp v1.9.0/go.mod h1:xE+Cx5E/eEHw+ISFkwPLwCZefwVjY+pqKg1qcK03+/4= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= diff --git a/server/lib/events/otlpconvert.go b/server/lib/events/otlpconvert.go new file mode 100644 index 00000000..5bf79332 --- /dev/null +++ b/server/lib/events/otlpconvert.go @@ -0,0 +1,125 @@ +package events + +import ( + "encoding/json" + "fmt" + "math" + "strings" + "time" + + oapi "github.com/kernel/kernel-images/server/lib/oapi" + "go.opentelemetry.io/otel/log" +) + +// otlpScopeName is the InstrumentationScope name stamped on exported records. +const otlpScopeName = "github.com/kernel/kernel-images/server/lib/events" + +// otlpExcludedCategories are telemetry categories that are not forwarded to the +// OTLP sink. Screenshots are base64 frames and Monitor is collector-health +// metadata; both bloat a customer's log backend with no analytical value. +var otlpExcludedCategories = map[oapi.TelemetryEventCategory]struct{}{ + Screenshot: {}, + Monitor: {}, +} + +// otlpCategoryExported reports whether events in cat are forwarded to OTLP. +func otlpCategoryExported(cat oapi.TelemetryEventCategory) bool { + _, excluded := otlpExcludedCategories[cat] + return !excluded +} + +// toLogRecord maps a telemetry envelope to an OTLP log record. It is pure (no +// I/O, no wall-clock reads) so it can be unit tested in isolation. +func toLogRecord(env Envelope) log.Record { + ev := env.Event + + var rec log.Record + rec.SetTimestamp(time.UnixMicro(ev.Ts)) + rec.SetEventName(ev.Type) + sev, sevText := otlpSeverity(ev.Type) + rec.SetSeverity(sev) + rec.SetSeverityText(sevText) + + if body, ok := jsonToLogValue(ev.Data); ok { + rec.SetBody(body) + } + + attrs := make([]log.KeyValue, 0, 8) + attrs = append(attrs, + log.String("kernel.event.category", string(ev.Category)), + log.Int64("kernel.event.seq", int64(env.Seq)), + log.String("kernel.source.kind", string(ev.Source.Kind)), + ) + if ev.Source.Event != nil { + attrs = append(attrs, log.String("kernel.source.event", *ev.Source.Event)) + } + if ev.Truncated { + attrs = append(attrs, log.Bool("kernel.truncated", true)) + } + if ev.Source.Metadata != nil { + for k, v := range *ev.Source.Metadata { + attrs = append(attrs, log.String("kernel."+k, v)) + } + } + rec.AddAttributes(attrs...) + return rec +} + +// otlpSeverity maps an event type to an OTLP severity. Console errors are +// errors; anything that reports a failure is a warning; everything else is +// informational. +func otlpSeverity(eventType string) (log.Severity, string) { + switch { + case eventType == "console_error": + return log.SeverityError, "ERROR" + case strings.HasSuffix(eventType, "_failed"): + return log.SeverityWarn, "WARN" + default: + return log.SeverityInfo, "INFO" + } +} + +// jsonToLogValue decodes raw event data into a structured OTLP value so the +// body stays navigable in the backend rather than an opaque string. Returns +// false when there is no data (e.g. truncated oversized envelopes). Invalid +// JSON falls back to the raw bytes as a string. +func jsonToLogValue(raw json.RawMessage) (log.Value, bool) { + if len(raw) == 0 { + return log.Value{}, false + } + var v any + if err := json.Unmarshal(raw, &v); err != nil { + return log.StringValue(string(raw)), true + } + return anyToLogValue(v), true +} + +func anyToLogValue(v any) log.Value { + switch t := v.(type) { + case nil: + return log.Value{} + case bool: + return log.BoolValue(t) + case float64: + if t == math.Trunc(t) && !math.IsInf(t, 0) && math.Abs(t) < math.MaxInt64 { + return log.Int64Value(int64(t)) + } + return log.Float64Value(t) + case string: + return log.StringValue(t) + case []any: + vals := make([]log.Value, len(t)) + for i, e := range t { + vals[i] = anyToLogValue(e) + } + return log.SliceValue(vals...) + case map[string]any: + kvs := make([]log.KeyValue, 0, len(t)) + for k, e := range t { + kvs = append(kvs, log.KeyValue{Key: k, Value: anyToLogValue(e)}) + } + return log.MapValue(kvs...) + default: + return log.StringValue(fmt.Sprintf("%v", t)) + } +} diff --git a/server/lib/events/otlpconvert_test.go b/server/lib/events/otlpconvert_test.go new file mode 100644 index 00000000..6b30a775 --- /dev/null +++ b/server/lib/events/otlpconvert_test.go @@ -0,0 +1,125 @@ +package events + +import ( + "encoding/json" + "testing" + "time" + + oapi "github.com/kernel/kernel-images/server/lib/oapi" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/log" +) + +func strptr(s string) *string { return &s } + +func attrsOf(rec log.Record) map[string]log.Value { + out := make(map[string]log.Value) + rec.WalkAttributes(func(kv log.KeyValue) bool { + out[kv.Key] = kv.Value + return true + }) + return out +} + +func TestToLogRecord_CoreFields(t *testing.T) { + meta := map[string]string{ + "telemetry_session_id": "cs_abc", + "cdp_session_id": "s1", + "target_id": "t1", + } + env := Envelope{ + Seq: 412, + Event: Event{ + Ts: 1_718_801_234_567_890, + Type: "network_response", + Category: Network, + Source: oapi.BrowserEventSource{ + Kind: oapi.BrowserEventSourceKind("cdp"), + Event: strptr("Network.loadingFinished"), + Metadata: &meta, + }, + Data: json.RawMessage(`{"status":200,"url":"https://api.foo.com/v1/x","ok":true}`), + }, + } + + rec := toLogRecord(env) + + assert.Equal(t, "network_response", rec.EventName()) + assert.Equal(t, time.UnixMicro(1_718_801_234_567_890).UTC(), rec.Timestamp().UTC()) + assert.Equal(t, log.SeverityInfo, rec.Severity()) + + attrs := attrsOf(rec) + assert.Equal(t, "network", attrs["kernel.event.category"].AsString()) + assert.Equal(t, int64(412), attrs["kernel.event.seq"].AsInt64()) + assert.Equal(t, "cdp", attrs["kernel.source.kind"].AsString()) + assert.Equal(t, "Network.loadingFinished", attrs["kernel.source.event"].AsString()) + assert.Equal(t, "cs_abc", attrs["kernel.telemetry_session_id"].AsString()) + assert.Equal(t, "s1", attrs["kernel.cdp_session_id"].AsString()) + assert.Equal(t, "t1", attrs["kernel.target_id"].AsString()) +} + +func TestToLogRecord_StructuredBody(t *testing.T) { + env := Envelope{Seq: 1, Event: Event{ + Ts: 1, + Type: "network_response", + Data: json.RawMessage(`{"status":200,"ok":true,"ratio":1.5,"tags":["a","b"]}`), + }} + + rec := toLogRecord(env) + body := rec.Body() + require.Equal(t, log.KindMap, body.Kind()) + + got := make(map[string]log.Value) + for _, kv := range body.AsMap() { + got[kv.Key] = kv.Value + } + assert.Equal(t, int64(200), got["status"].AsInt64(), "integral json number stays int64") + assert.Equal(t, true, got["ok"].AsBool()) + assert.Equal(t, 1.5, got["ratio"].AsFloat64()) + require.Equal(t, log.KindSlice, got["tags"].Kind()) + assert.Len(t, got["tags"].AsSlice(), 2) +} + +func TestToLogRecord_Severity(t *testing.T) { + cases := map[string]log.Severity{ + "console_error": log.SeverityError, + "network_loading_failed": log.SeverityWarn, + "monitor_init_failed": log.SeverityWarn, + "network_response": log.SeverityInfo, + "page_navigation": log.SeverityInfo, + } + for typ, want := range cases { + rec := toLogRecord(Envelope{Event: Event{Type: typ}}) + assert.Equalf(t, want, rec.Severity(), "severity for %q", typ) + } +} + +func TestToLogRecord_TruncatedNoData(t *testing.T) { + env := Envelope{Seq: 9, Event: Event{ + Ts: 1, + Type: "network_response", + Category: Network, + Truncated: true, + // Data nil: oversized envelope had its payload stripped at publish. + }} + + rec := toLogRecord(env) + assert.Equal(t, log.KindEmpty, rec.Body().Kind(), "no body when data is absent") + assert.Equal(t, true, attrsOf(rec)["kernel.truncated"].AsBool()) +} + +func TestToLogRecord_InvalidJSONFallsBackToString(t *testing.T) { + env := Envelope{Seq: 1, Event: Event{Ts: 1, Type: "x", Data: json.RawMessage(`not json`)}} + rec := toLogRecord(env) + body := rec.Body() + require.Equal(t, log.KindString, body.Kind()) + assert.Equal(t, "not json", body.AsString()) +} + +func TestOTLPCategoryExported(t *testing.T) { + assert.True(t, otlpCategoryExported(Network)) + assert.True(t, otlpCategoryExported(Console)) + assert.False(t, otlpCategoryExported(Screenshot)) + assert.False(t, otlpCategoryExported(Monitor)) +} From 55c250d4be9db526253f1ed6016fde61881c9510 Mon Sep 17 00:00:00 2001 From: archandatta <35818003+archandatta@users.noreply.github.com> Date: Wed, 24 Jun 2026 14:35:20 +0000 Subject: [PATCH 02/11] Add OTLP telemetry sink Mirrors S2StorageWriter: reads the event stream via an independent ring reader and exports records over OTLP/HTTP through the OTel log SDK. Configurable endpoint, path, headers, and resource attributes; drops excluded categories and oversized envelopes. --- server/go.mod | 35 ++--- server/go.sum | 76 ++++++----- server/lib/events/otlpstorage.go | 184 ++++++++++++++++++++++++++ server/lib/events/otlpstorage_test.go | 63 +++++++++ 4 files changed, 310 insertions(+), 48 deletions(-) create mode 100644 server/lib/events/otlpstorage.go create mode 100644 server/lib/events/otlpstorage_test.go diff --git a/server/go.mod b/server/go.mod index e37a7d71..b45dbe4e 100644 --- a/server/go.mod +++ b/server/go.mod @@ -27,10 +27,14 @@ require ( github.com/samber/lo v1.52.0 github.com/stretchr/testify v1.11.1 github.com/testcontainers/testcontainers-go v0.40.0 + go.opentelemetry.io/otel v1.44.0 + go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.20.0 go.opentelemetry.io/otel/log v0.20.0 - golang.org/x/sync v0.18.0 - golang.org/x/sys v0.39.0 - golang.org/x/term v0.37.0 + go.opentelemetry.io/otel/sdk v1.44.0 + go.opentelemetry.io/otel/sdk/log v0.20.0 + golang.org/x/sync v0.20.0 + golang.org/x/sys v0.45.0 + golang.org/x/term v0.43.0 gopkg.in/yaml.v3 v3.0.1 ) @@ -40,6 +44,7 @@ require ( github.com/Microsoft/go-winio v0.6.2 // indirect github.com/apapsch/go-jsonmerge/v2 v2.0.0 // indirect github.com/cenkalti/backoff/v4 v4.3.0 // indirect + github.com/cenkalti/backoff/v5 v5.0.3 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/containerd/errdefs v1.0.0 // indirect github.com/containerd/errdefs/pkg v0.3.0 // indirect @@ -59,7 +64,7 @@ require ( github.com/go-ole/go-ole v1.2.6 // indirect github.com/go-openapi/jsonpointer v0.21.0 // indirect github.com/go-openapi/swag v0.23.0 // indirect - github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 // indirect + github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 // indirect github.com/jinzhu/inflection v1.0.0 // indirect github.com/jinzhu/now v1.1.5 // indirect github.com/josharian/intern v1.0.0 // indirect @@ -101,21 +106,19 @@ require ( github.com/yusufpapurcu/wmi v1.2.4 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 // indirect - go.opentelemetry.io/otel v1.44.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.39.0 // indirect go.opentelemetry.io/otel/metric v1.44.0 // indirect - go.opentelemetry.io/otel/sdk v1.39.0 // indirect - go.opentelemetry.io/otel/sdk/metric v1.39.0 // indirect go.opentelemetry.io/otel/trace v1.44.0 // indirect - golang.org/x/crypto v0.44.0 // indirect - golang.org/x/mod v0.29.0 // indirect - golang.org/x/net v0.47.0 // indirect - golang.org/x/text v0.31.0 // indirect - golang.org/x/tools v0.38.0 // indirect - google.golang.org/genproto/googleapis/api v0.0.0-20250825161204-c5933d9347a5 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20250825161204-c5933d9347a5 // indirect - google.golang.org/grpc v1.75.1 // indirect - google.golang.org/protobuf v1.36.10 // indirect + go.opentelemetry.io/proto/otlp v1.10.0 // indirect + golang.org/x/crypto v0.51.0 // indirect + golang.org/x/mod v0.35.0 // indirect + golang.org/x/net v0.55.0 // indirect + golang.org/x/text v0.37.0 // indirect + golang.org/x/tools v0.44.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect + google.golang.org/grpc v1.81.1 // indirect + google.golang.org/protobuf v1.36.11 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect gorm.io/gorm v1.25.7 // indirect modernc.org/libc v1.22.5 // indirect diff --git a/server/go.sum b/server/go.sum index 00214fb1..8a394267 100644 --- a/server/go.sum +++ b/server/go.sum @@ -14,6 +14,8 @@ github.com/avast/retry-go/v5 v5.0.0/go.mod h1://d+usmKWio1agtZfS1H/ltTqwtIfBnRq9 github.com/bmatcuk/doublestar v1.1.1/go.mod h1:UD6OnuiIn0yFxxA2le/rnRU1G4RaI4UvFv1sNto9p6w= github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= +github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= +github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI= @@ -93,6 +95,8 @@ github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvq github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI= github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= @@ -105,8 +109,8 @@ github.com/google/pprof v0.0.0-20221118152302-e6195bd50e26 h1:Xim43kblpZXfIBQsbu github.com/google/pprof v0.0.0-20221118152302-e6195bd50e26/go.mod h1:dDKJzRmX4S37WGHujM7tX//fmj1uioxKzKxz3lo4HJo= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2 h1:8Tjv8EJ+pM1xP8mK6egEbD1OgnVTyacbefKhmbLhIhU= -github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.2/go.mod h1:pkJQ2tZHJ0aFOVEEot6oZmaVEZcRme73eIFmhiVuRWs= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0 h1:5VipnvEpbqr2gA2VbM+nYVbkIF28c5ZQfqCBQ5g2xfk= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.29.0/go.mod h1:Hyl3n6Twe1hvtd9XUXDec4pTvgMSEixRuQKPTMH2bNs= github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU= github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc= github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E= @@ -258,6 +262,8 @@ go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 h1:F7Jx+6h go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0/go.mod h1:UHB22Z8QsdRDrnAtX4PntOl36ajSxcdUMt1sF7Y6E7Q= go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU= go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc= +go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.20.0 h1:owlhcJ3QO3X0YTDTCcDZ4V+6aVDkWbNmBoQ5NUp7Oww= +go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.20.0/go.mod h1:MP4eemTiI9zC8fgg+DYynhYDYf3ba72S376TvP+Ye0Q= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.39.0 h1:f0cb2XPmrqn4XMy9PNliTgRKJgS5WcL/u0/WRYGz4t0= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.39.0/go.mod h1:vnakAaFckOMiMtOIhFI2MNH4FYrZzXCYxmb1LlhoGz8= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.19.0 h1:IeMeyr1aBvBiPVYihXIaeIZba6b8E1bYp7lbdxK8CQg= @@ -266,22 +272,26 @@ go.opentelemetry.io/otel/log v0.20.0 h1:/5i0vuHxCLWUfChWG41K9wkM0jafruPw9NU1/RCJ go.opentelemetry.io/otel/log v0.20.0/go.mod h1:wOcMcjsZpG8x7Bak7IhSi/lg8wscV2C1VdrKCLPlt0E= go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= -go.opentelemetry.io/otel/sdk v1.39.0 h1:nMLYcjVsvdui1B/4FRkwjzoRVsMK8uL/cj0OyhKzt18= -go.opentelemetry.io/otel/sdk v1.39.0/go.mod h1:vDojkC4/jsTJsE+kh+LXYQlbL8CgrEcwmt1ENZszdJE= -go.opentelemetry.io/otel/sdk/metric v1.39.0 h1:cXMVVFVgsIf2YL6QkRF4Urbr/aMInf+2WKg+sEJTtB8= -go.opentelemetry.io/otel/sdk/metric v1.39.0/go.mod h1:xq9HEVH7qeX69/JnwEfp6fVq5wosJsY1mt4lLfYdVew= +go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58= +go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0= +go.opentelemetry.io/otel/sdk/log v0.20.0 h1:vM3xI7TQgKPiSghe6urZtAkyFY7SodrSpC83CffDFuY= +go.opentelemetry.io/otel/sdk/log v0.20.0/go.mod h1:Knej2nmsTUzN79T2eeXdRsjjPcoxoq2pUyUHz9TFyyU= +go.opentelemetry.io/otel/sdk/log/logtest v0.20.0 h1:OqdRZ1guyzamK3M6LlRsmGqRrjkHWw6WZOKKli5ELpg= +go.opentelemetry.io/otel/sdk/log/logtest v0.20.0/go.mod h1:PuMIlm7zAt7c3z8zfOI5ox4iT1Z87We+PF6YoINux/M= +go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI= +go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA= go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= -go.opentelemetry.io/proto/otlp v1.9.0 h1:l706jCMITVouPOqEnii2fIAuO3IVGBRPV5ICjceRb/A= -go.opentelemetry.io/proto/otlp v1.9.0/go.mod h1:xE+Cx5E/eEHw+ISFkwPLwCZefwVjY+pqKg1qcK03+/4= +go.opentelemetry.io/proto/otlp v1.10.0 h1:IQRWgT5srOCYfiWnpqUYz9CVmbO8bFmKcwYxpuCSL2g= +go.opentelemetry.io/proto/otlp v1.10.0/go.mod h1:/CV4QoCR/S9yaPj8utp3lvQPoqMtxXdzn7ozvvozVqk= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= -golang.org/x/crypto v0.44.0 h1:A97SsFvM3AIwEEmTBiaxPPTYpDC47w720rdiiUvgoAU= -golang.org/x/crypto v0.44.0/go.mod h1:013i+Nw79BMiQiMsOPcVCB5ZIJbYkerPrGnOa00tvmc= +golang.org/x/crypto v0.51.0 h1:IBPXwPfKxY7cWQZ38ZCIRPI50YLeevDLlLnyC5wRGTI= +golang.org/x/crypto v0.51.0/go.mod h1:8AdwkbraGNABw2kOX6YFPs3WM22XqI4EXEd8g+x7Oc8= golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= -golang.org/x/mod v0.29.0 h1:HV8lRxZC4l2cr3Zq1LvtOsi/ThTgWnUk/y64QSs8GwA= -golang.org/x/mod v0.29.0/go.mod h1:NyhrlYXJ2H4eJiRy/WDBO6HMqZQ6q9nk4JzS3NuCK+w= +golang.org/x/mod v0.35.0 h1:Ww1D637e6Pg+Zb2KrWfHQUnH2dQRLBQyAtpr/haaJeM= +golang.org/x/mod v0.35.0/go.mod h1:+GwiRhIInF8wPm+4AoT6L0FA1QWAad3OMdTRx4tFYlU= golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= @@ -289,13 +299,13 @@ golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/ golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= golang.org/x/net v0.0.0-20210428140749-89ef3d95e781/go.mod h1:OJAsFXCWl8Ukc7SiCT/9KSuxbyM7479/AVlXFRxuMCk= golang.org/x/net v0.0.0-20220225172249-27dd8689420f/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk= -golang.org/x/net v0.47.0 h1:Mx+4dIFzqraBXUugkia1OOvlD6LemFo1ALMHjrXDOhY= -golang.org/x/net v0.47.0/go.mod h1:/jNxtkgq5yWUGYkaZGqo27cfGZ1c5Nen03aYrrKpVRU= +golang.org/x/net v0.55.0 h1:bcvxaJn3e1U6InsFWt1JUq1aSjnRxLzT2rtD2KfkDF8= +golang.org/x/net v0.55.0/go.mod h1:L5U2KuzuOe1lY7Z+aWVIKK6qEeJXnXV9yzGA+WCHJww= golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.18.0 h1:kr88TuHDroi+UVf+0hZnirlk8o8T+4MrK6mr60WkH/I= -golang.org/x/sync v0.18.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= +golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= @@ -317,35 +327,37 @@ golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.39.0 h1:CvCKL8MeisomCi6qNZ+wbb0DN9E5AATixKsvNtMoMFk= -golang.org/x/sys v0.39.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/sys v0.45.0 h1:dO4czNzziLiiXplLQgBCEpCvXQ3dnkn0SdaZSYdQ+FY= +golang.org/x/sys v0.45.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= -golang.org/x/term v0.37.0 h1:8EGAD0qCmHYZg6J17DvsMy9/wJ7/D/4pV/wfnld5lTU= -golang.org/x/term v0.37.0/go.mod h1:5pB4lxRNYYVZuTLmy8oR2BH8dflOR+IbTYFD8fi3254= +golang.org/x/term v0.43.0 h1:S4RLU2sB31O/NCl+zFN9Aru9A/Cq2aqKpTZJ6B+DwT4= +golang.org/x/term v0.43.0/go.mod h1:lrhlHNdQJHO+1qVYiHfFKVuVioJIheAc3fBSMFYEIsk= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ= -golang.org/x/text v0.31.0 h1:aC8ghyu4JhP8VojJ2lEHBnochRno1sgL6nEi9WGFGMM= -golang.org/x/text v0.31.0/go.mod h1:tKRAlv61yKIjGGHX/4tP1LTbc13YSec1pxVEWXzfoeM= +golang.org/x/text v0.37.0 h1:Cqjiwd9eSg8e0QAkyCaQTNHFIIzWtidPahFWR83rTrc= +golang.org/x/text v0.37.0/go.mod h1:a5sjxXGs9hsn/AJVwuElvCAo9v8QYLzvavO5z2PiM38= golang.org/x/time v0.5.0 h1:o7cqy6amK/52YcAKIPlM3a+Fpj35zvRj2TP+e1xFSfk= golang.org/x/time v0.5.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= -golang.org/x/tools v0.38.0 h1:Hx2Xv8hISq8Lm16jvBZ2VQf+RLmbd7wVUsALibYI/IQ= -golang.org/x/tools v0.38.0/go.mod h1:yEsQ/d/YK8cjh0L6rZlY8tgtlKiBNTL14pGDJPJpYQs= +golang.org/x/tools v0.44.0 h1:UP4ajHPIcuMjT1GqzDWRlalUEoY+uzoZKnhOjbIPD2c= +golang.org/x/tools v0.44.0/go.mod h1:KA0AfVErSdxRZIsOVipbv3rQhVXTnlU6UhKxHd1seDI= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= -google.golang.org/genproto/googleapis/api v0.0.0-20250825161204-c5933d9347a5 h1:BIRfGDEjiHRrk0QKZe3Xv2ieMhtgRGeLcZQ0mIVn4EY= -google.golang.org/genproto/googleapis/api v0.0.0-20250825161204-c5933d9347a5/go.mod h1:j3QtIyytwqGr1JUDtYXwtMXWPKsEa5LtzIFN1Wn5WvE= -google.golang.org/genproto/googleapis/rpc v0.0.0-20250825161204-c5933d9347a5 h1:eaY8u2EuxbRv7c3NiGK0/NedzVsCcV6hDuU5qPX5EGE= -google.golang.org/genproto/googleapis/rpc v0.0.0-20250825161204-c5933d9347a5/go.mod h1:M4/wBTSeyLxupu3W3tJtOgB14jILAS/XWPSSa3TAlJc= -google.golang.org/grpc v1.75.1 h1:/ODCNEuf9VghjgO3rqLcfg8fiOP0nSluljWFlDxELLI= -google.golang.org/grpc v1.75.1/go.mod h1:JtPAzKiq4v1xcAB2hydNlWI2RnF85XXcV0mhKXr2ecQ= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa h1:Kjn0N0tCrDgiAFW+lGO4JZ3ck44CehvJQMAwj9QF0G8= +google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:q4lMZS6kskjT5HvCPrnnypcDPVJqT/f4nfxmkE7gryY= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa h1:mZHHdPZl0dbGHCflZgAq/Q468DWVFcU2whhB2KAo8fk= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.81.1 h1:VnnIIZ88UzOOKLukQi+ImGz8O1Wdp8nAGGnvOfEIWQQ= +google.golang.org/grpc v1.81.1/go.mod h1:xGH9GfzOyMTGIOXBJmXt+BX/V0kcdQbdcuwQ/zNw42I= google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8= google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0= google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM= @@ -354,8 +366,8 @@ google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzi google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU= google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= -google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE= -google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= +google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= diff --git a/server/lib/events/otlpstorage.go b/server/lib/events/otlpstorage.go new file mode 100644 index 00000000..124bf5ff --- /dev/null +++ b/server/lib/events/otlpstorage.go @@ -0,0 +1,184 @@ +package events + +import ( + "context" + "fmt" + "log/slog" + "sync" + "sync/atomic" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp" + "go.opentelemetry.io/otel/log" + sdklog "go.opentelemetry.io/otel/sdk/log" + "go.opentelemetry.io/otel/sdk/resource" + semconv "go.opentelemetry.io/otel/semconv/v1.27.0" +) + +// defaultOTLPMaxBatchRecords bounds records per export request. The metro-api +// relay caps bodies at 10MB; envelopes are already capped at ~1MB each at +// publish, so this keeps the common case well under the limit. Byte-based +// batching (the real guard against many large records) is deferred. +const defaultOTLPMaxBatchRecords = 200 + +// maxOTLPRecordBytes drops a single envelope whose data alone would risk the +// relay body cap. A safety net above the publish-time truncation. +const maxOTLPRecordBytes = 4 * 1024 * 1024 + +// OTLPConfig configures the OTLP telemetry sink. +type OTLPConfig struct { + // Endpoint is the host[:port] of the OTLP/HTTP target: the metro-api relay + // in production, a local collector in development. + Endpoint string + // URLPath overrides the request path (the relay serves /otlp-relay/v1/logs). + // Empty uses the exporter default of /v1/logs. + URLPath string + // Insecure sends over plaintext HTTP. Development only. + Insecure bool + // Headers are attached to every export request (e.g. the instance JWT). + Headers map[string]string + + // ServiceName, InstanceName, and Metro populate the OTLP Resource. + ServiceName string + InstanceName string + Metro string + + // MaxBatchRecords bounds records per export. Defaults to 200. + MaxBatchRecords int +} + +// otlpStorage implements Storage by converting envelopes to OTLP log records +// and emitting them through the OTel log SDK, which batches and exports to the +// configured endpoint. +type otlpStorage struct { + provider *sdklog.LoggerProvider + logger log.Logger + log *slog.Logger + dropped atomic.Uint64 +} + +func newOTLPStorage(ctx context.Context, cfg OTLPConfig, logger *slog.Logger) (*otlpStorage, error) { + opts := []otlploghttp.Option{otlploghttp.WithEndpoint(cfg.Endpoint)} + if cfg.URLPath != "" { + opts = append(opts, otlploghttp.WithURLPath(cfg.URLPath)) + } + if cfg.Insecure { + opts = append(opts, otlploghttp.WithInsecure()) + } + if len(cfg.Headers) > 0 { + opts = append(opts, otlploghttp.WithHeaders(cfg.Headers)) + } + + exporter, err := otlploghttp.New(ctx, opts...) + if err != nil { + return nil, fmt.Errorf("otlp storage: new exporter: %w", err) + } + + res := resource.NewSchemaless( + semconv.ServiceName(cfg.ServiceName), + attribute.String("kernel.instance_name", cfg.InstanceName), + attribute.String("kernel.metro", cfg.Metro), + ) + + maxBatch := cfg.MaxBatchRecords + if maxBatch == 0 { + maxBatch = defaultOTLPMaxBatchRecords + } + processor := sdklog.NewBatchProcessor(exporter, sdklog.WithExportMaxBatchSize(maxBatch)) + provider := sdklog.NewLoggerProvider( + sdklog.WithResource(res), + sdklog.WithProcessor(processor), + ) + + return &otlpStorage{ + provider: provider, + logger: provider.Logger(otlpScopeName), + log: logger, + }, nil +} + +// Append converts an exported-category envelope to an OTLP record and hands it +// to the SDK. Excluded categories and oversized envelopes are dropped. +func (s *otlpStorage) Append(ctx context.Context, env Envelope) error { + if !otlpCategoryExported(env.Event.Category) { + return nil + } + if len(env.Event.Data) > maxOTLPRecordBytes { + n := s.dropped.Add(1) + s.log.Warn("otlp storage: dropped oversized event", "seq", env.Seq, "bytes", len(env.Event.Data), "total_dropped", n) + return nil + } + s.logger.Emit(ctx, toLogRecord(env)) + return nil +} + +// Close flushes buffered records and shuts down the exporter. +func (s *otlpStorage) Close(ctx context.Context) error { + return s.provider.Shutdown(ctx) +} + +// OTLPStorageWriter reads from an EventStream and forwards each event to an +// OTLP endpoint. Construct with NewOTLPStorageWriter, call Start to begin and +// Stop to drain and shut down. Mirrors S2StorageWriter. +type OTLPStorageWriter struct { + es *EventStream + cfg OTLPConfig + log *slog.Logger + + mu sync.Mutex + started bool + storage *otlpStorage + writer *StorageWriter + done chan struct{} +} + +func NewOTLPStorageWriter(es *EventStream, cfg OTLPConfig, log *slog.Logger) *OTLPStorageWriter { + return &OTLPStorageWriter{es: es, cfg: cfg, log: log} +} + +// Start builds the exporter and begins reading from the event stream. ctx +// governs the Run loop; cancel it (e.g. on SIGTERM) to stop reading. The +// exporter outlives ctx and is torn down by Stop after flushing. +func (w *OTLPStorageWriter) Start(ctx context.Context) error { + w.mu.Lock() + defer w.mu.Unlock() + if w.started { + return fmt.Errorf("otlpstoragewriter: Start called more than once") + } + storage, err := newOTLPStorage(ctx, w.cfg, w.log) + if err != nil { + return err + } + w.storage = storage + w.writer = NewStorageWriter(w.es, storage, w.log) + w.done = make(chan struct{}) + w.started = true + go func() { + defer close(w.done) + if err := w.writer.Run(ctx); err != nil && ctx.Err() == nil { + w.log.Error("otlp storage writer failed", "err", err) + } + }() + return nil +} + +// Stop waits for the Run goroutine to exit, drains remaining ring events, then +// flushes and shuts down the exporter. ctx bounds total shutdown time. +func (w *OTLPStorageWriter) Stop(ctx context.Context) error { + w.mu.Lock() + if !w.started { + w.mu.Unlock() + return nil + } + w.mu.Unlock() + + select { + case <-w.done: + case <-ctx.Done(): + return ctx.Err() + } + if err := w.writer.Drain(ctx); err != nil { + w.log.Warn("otlp storage writer: drain incomplete", "err", err) + } + return w.storage.Close(ctx) +} diff --git a/server/lib/events/otlpstorage_test.go b/server/lib/events/otlpstorage_test.go new file mode 100644 index 00000000..731b739a --- /dev/null +++ b/server/lib/events/otlpstorage_test.go @@ -0,0 +1,63 @@ +package events + +import ( + "context" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestOTLPStorageWriter_ExportsEvents drives the full sink against a local HTTP +// server standing in for the relay/collector, verifying records reach the +// configured path with the configured headers. +func TestOTLPStorageWriter_ExportsEvents(t *testing.T) { + var mu sync.Mutex + var paths, auths []string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + mu.Lock() + paths = append(paths, r.URL.Path) + auths = append(auths, r.Header.Get("Authorization")) + mu.Unlock() + w.WriteHeader(http.StatusOK) + })) + defer srv.Close() + + es, err := NewEventStream(EventStreamConfig{RingCapacity: 64}) + require.NoError(t, err) + + cfg := OTLPConfig{ + Endpoint: strings.TrimPrefix(srv.URL, "http://"), + URLPath: "/otlp-relay/v1/logs", + Insecure: true, + Headers: map[string]string{"Authorization": "Bearer test-jwt"}, + ServiceName: "kernel-browser", + InstanceName: "browser-1", + Metro: "dev-iad", + MaxBatchRecords: 10, + } + wtr := NewOTLPStorageWriter(es, cfg, slog.Default()) + + ctx, cancel := context.WithCancel(context.Background()) + require.NoError(t, wtr.Start(ctx)) + + es.Publish(Envelope{Event: Event{Ts: 1, Type: "page_navigation", Category: Page, Data: []byte(`{"url":"https://x"}`)}}) + es.Publish(Envelope{Event: Event{Ts: 2, Type: "screenshot", Category: Screenshot, Data: []byte(`{"png":"..."}`)}}) + + cancel() // stop the Run loop; Stop then drains the ring and flushes + stopCtx, stopCancel := context.WithTimeout(context.Background(), 5*time.Second) + defer stopCancel() + require.NoError(t, wtr.Stop(stopCtx)) + + mu.Lock() + defer mu.Unlock() + require.NotEmpty(t, paths, "expected at least one export request") + assert.Equal(t, "/otlp-relay/v1/logs", paths[0]) + assert.Equal(t, "Bearer test-jwt", auths[0]) +} From b9dc22deba7d2d929239d74cfb7017f7e5d15b04 Mon Sep 17 00:00:00 2001 From: archandatta <35818003+archandatta@users.noreply.github.com> Date: Wed, 24 Jun 2026 14:36:41 +0000 Subject: [PATCH 03/11] Wire OTLP export sink into config and startup Adds OTLP_RELAY_* config and starts the OTLP storage writer at boot when an endpoint is set, alongside the existing S2 sink, with matching graceful shutdown. JWT is redacted in logs. --- server/cmd/api/main.go | 31 +++++++++++++++++++++++++++++++ server/cmd/config/config.go | 20 ++++++++++++++++++++ 2 files changed, 51 insertions(+) diff --git a/server/cmd/api/main.go b/server/cmd/api/main.go index b3500a52..fee7dfad 100644 --- a/server/cmd/api/main.go +++ b/server/cmd/api/main.go @@ -125,6 +125,29 @@ func main() { } } + // Optional OTLP export sink. Independent of S2; both can run together. + var otlpWriter *events.OTLPStorageWriter + if config.OTLPEndpoint != "" { + headers := map[string]string{} + if config.OTLPInstanceJWT != "" { + headers["Authorization"] = "Bearer " + config.OTLPInstanceJWT + } + slogger.Info("OTLP export enabled", "endpoint", config.OTLPEndpoint, "path", config.OTLPPath) + otlpWriter = events.NewOTLPStorageWriter(eventStream, events.OTLPConfig{ + Endpoint: config.OTLPEndpoint, + URLPath: config.OTLPPath, + Insecure: config.OTLPInsecure, + Headers: headers, + ServiceName: config.OTLPServiceName, + InstanceName: config.OTLPInstanceName, + Metro: config.OTLPMetro, + }, slogger) + if err := otlpWriter.Start(ctx); err != nil { + slogger.Error("failed to start OTLP storage writer", "err", err) + os.Exit(1) + } + } + apiService, err := api.New( recorder.NewFFmpegManager(), recorder.NewFFmpegRecorderFactory(config.PathToFFmpeg, defaultParams, stz), @@ -306,6 +329,14 @@ func main() { slogger.Error("s2 storage writer stop failed", "err", err) } } + + if otlpWriter != nil { + stopCtx, stopCancel := context.WithTimeout(context.Background(), 10*time.Second) + defer stopCancel() + if err := otlpWriter.Stop(stopCtx); err != nil { + slogger.Error("otlp storage writer stop failed", "err", err) + } + } } func mustFFmpeg() { diff --git a/server/cmd/config/config.go b/server/cmd/config/config.go index aae38b9e..96782acd 100644 --- a/server/cmd/config/config.go +++ b/server/cmd/config/config.go @@ -47,6 +47,17 @@ type Config struct { S2Basin string `envconfig:"S2_BASIN" default:""` S2AccessToken string `envconfig:"S2_ACCESS_TOKEN" default:""` S2Stream string `envconfig:"S2_STREAM" default:""` + + // OTLP telemetry export. OTLPEndpoint (host[:port]) must be set to enable + // the OTLP sink, which forwards telemetry events to the metro-api relay (or + // a collector in development). + OTLPEndpoint string `envconfig:"OTLP_RELAY_ENDPOINT" default:""` + OTLPPath string `envconfig:"OTLP_RELAY_PATH" default:"/otlp-relay/v1/logs"` + OTLPInsecure bool `envconfig:"OTLP_RELAY_INSECURE" default:"false"` + OTLPInstanceJWT string `envconfig:"KERNEL_INSTANCE_JWT" default:""` + OTLPServiceName string `envconfig:"OTLP_SERVICE_NAME" default:"kernel-browser"` + OTLPInstanceName string `envconfig:"OTLP_INSTANCE_NAME" default:""` + OTLPMetro string `envconfig:"OTLP_METRO" default:""` } // LogValue implements slog.LogValuer, redacting secret fields. @@ -55,6 +66,10 @@ func (c *Config) LogValue() slog.Value { if c.S2AccessToken != "" { s2AccessToken = "[redacted]" } + otlpJWT := "" + if c.OTLPInstanceJWT != "" { + otlpJWT = "[redacted]" + } return slog.GroupValue( slog.Int("port", c.Port), slog.Int("frame_rate", c.FrameRate), @@ -73,6 +88,11 @@ func (c *Config) LogValue() slog.Value { slog.String("s2_basin", c.S2Basin), slog.String("s2_access_token", s2AccessToken), slog.String("s2_stream", c.S2Stream), + slog.String("otlp_endpoint", c.OTLPEndpoint), + slog.String("otlp_path", c.OTLPPath), + slog.Bool("otlp_insecure", c.OTLPInsecure), + slog.String("otlp_instance_jwt", otlpJWT), + slog.String("otlp_service_name", c.OTLPServiceName), ) } From 5ae881c090f5ee66456b2ce2f46c141cb9dc4d07 Mon Sep 17 00:00:00 2001 From: archandatta <35818003+archandatta@users.noreply.github.com> Date: Wed, 24 Jun 2026 14:38:40 +0000 Subject: [PATCH 04/11] Add local OTel collector config for OTLP sink testing Receives OTLP/HTTP on :4318 and prints decoded records to stdout for local validation of the telemetry export path. --- server/dev/otelcol/config.yaml | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) create mode 100644 server/dev/otelcol/config.yaml diff --git a/server/dev/otelcol/config.yaml b/server/dev/otelcol/config.yaml new file mode 100644 index 00000000..35d61fad --- /dev/null +++ b/server/dev/otelcol/config.yaml @@ -0,0 +1,24 @@ +# Local OTel Collector for testing the OTLP telemetry sink (Phase 1). +# Receives OTLP/HTTP on :4318 and prints decoded records to stdout. +# +# docker run --rm -p 4318:4318 \ +# -v "$PWD/server/dev/otelcol/config.yaml":/etc/otelcol-contrib/config.yaml \ +# otel/opentelemetry-collector-contrib:latest +# +# Then run the server with: +# OTLP_RELAY_ENDPOINT=localhost:4318 OTLP_RELAY_PATH=/v1/logs OTLP_RELAY_INSECURE=true +receivers: + otlp: + protocols: + http: + endpoint: 0.0.0.0:4318 + +exporters: + debug: + verbosity: detailed + +service: + pipelines: + logs: + receivers: [otlp] + exporters: [debug] From b49d593ce5ebed876f49d66f87d4b0c98afd768a Mon Sep 17 00:00:00 2001 From: archandatta <35818003+archandatta@users.noreply.github.com> Date: Thu, 25 Jun 2026 11:55:57 +0000 Subject: [PATCH 05/11] Promote event type and key payload fields to OTLP attributes EventName is dropped by some backends (Datadog, Loki), and a structured body is only flattened by some (Honeycomb), so mirror the event type into kernel.event.type and lift high-value network/console fields into queryable attributes using semantic-convention names. Full payload stays in the body. --- server/lib/events/otlpconvert.go | 60 +++++++++++++++++++-------- server/lib/events/otlpconvert_test.go | 28 +++++++++++++ 2 files changed, 70 insertions(+), 18 deletions(-) diff --git a/server/lib/events/otlpconvert.go b/server/lib/events/otlpconvert.go index 5bf79332..f8cc170d 100644 --- a/server/lib/events/otlpconvert.go +++ b/server/lib/events/otlpconvert.go @@ -40,12 +40,23 @@ func toLogRecord(env Envelope) log.Record { rec.SetSeverity(sev) rec.SetSeverityText(sevText) - if body, ok := jsonToLogValue(ev.Data); ok { - rec.SetBody(body) + // Decode the payload once: it becomes the structured body and the source of + // promoted attributes. + var data any + if len(ev.Data) > 0 { + if err := json.Unmarshal(ev.Data, &data); err != nil { + rec.SetBody(log.StringValue(string(ev.Data))) + data = nil + } else { + rec.SetBody(anyToLogValue(data)) + } } - attrs := make([]log.KeyValue, 0, 8) + attrs := make([]log.KeyValue, 0, 12) attrs = append(attrs, + // EventName is dropped by some backends (Datadog, Loki), so the event + // type is mirrored into an attribute to stay queryable everywhere. + log.String("kernel.event.type", ev.Type), log.String("kernel.event.category", string(ev.Category)), log.Int64("kernel.event.seq", int64(env.Seq)), log.String("kernel.source.kind", string(ev.Source.Kind)), @@ -61,10 +72,38 @@ func toLogRecord(env Envelope) log.Record { attrs = append(attrs, log.String("kernel."+k, v)) } } + if m, ok := data.(map[string]any); ok { + attrs = append(attrs, promotedAttributes(ev.Category, m)...) + } rec.AddAttributes(attrs...) return rec } +// promotedAttributes lifts high-value payload fields into typed, queryable +// attributes (OTel semantic conventions where they exist) so they stay +// filterable in backends that do not flatten a structured body (Datadog, Loki). +// The full payload remains in the body for fidelity. +func promotedAttributes(cat oapi.TelemetryEventCategory, data map[string]any) []log.KeyValue { + var out []log.KeyValue + switch cat { + case Network: + if v, ok := data["method"].(string); ok { + out = append(out, log.String("http.request.method", v)) + } + if v, ok := data["url"].(string); ok { + out = append(out, log.String("url.full", v)) + } + if v, ok := data["status"].(float64); ok { + out = append(out, log.Int64("http.response.status_code", int64(v))) + } + case Console: + if v, ok := data["level"].(string); ok { + out = append(out, log.String("kernel.console.level", v)) + } + } + return out +} + // otlpSeverity maps an event type to an OTLP severity. Console errors are // errors; anything that reports a failure is a warning; everything else is // informational. @@ -79,21 +118,6 @@ func otlpSeverity(eventType string) (log.Severity, string) { } } -// jsonToLogValue decodes raw event data into a structured OTLP value so the -// body stays navigable in the backend rather than an opaque string. Returns -// false when there is no data (e.g. truncated oversized envelopes). Invalid -// JSON falls back to the raw bytes as a string. -func jsonToLogValue(raw json.RawMessage) (log.Value, bool) { - if len(raw) == 0 { - return log.Value{}, false - } - var v any - if err := json.Unmarshal(raw, &v); err != nil { - return log.StringValue(string(raw)), true - } - return anyToLogValue(v), true -} - func anyToLogValue(v any) log.Value { switch t := v.(type) { case nil: diff --git a/server/lib/events/otlpconvert_test.go b/server/lib/events/otlpconvert_test.go index 6b30a775..a3e5b0f6 100644 --- a/server/lib/events/otlpconvert_test.go +++ b/server/lib/events/otlpconvert_test.go @@ -50,6 +50,7 @@ func TestToLogRecord_CoreFields(t *testing.T) { assert.Equal(t, log.SeverityInfo, rec.Severity()) attrs := attrsOf(rec) + assert.Equal(t, "network_response", attrs["kernel.event.type"].AsString(), "event type mirrored as attribute for backends that drop EventName") assert.Equal(t, "network", attrs["kernel.event.category"].AsString()) assert.Equal(t, int64(412), attrs["kernel.event.seq"].AsInt64()) assert.Equal(t, "cdp", attrs["kernel.source.kind"].AsString()) @@ -57,6 +58,33 @@ func TestToLogRecord_CoreFields(t *testing.T) { assert.Equal(t, "cs_abc", attrs["kernel.telemetry_session_id"].AsString()) assert.Equal(t, "s1", attrs["kernel.cdp_session_id"].AsString()) assert.Equal(t, "t1", attrs["kernel.target_id"].AsString()) + // High-value network fields promoted to queryable attributes. + assert.Equal(t, int64(200), attrs["http.response.status_code"].AsInt64()) + assert.Equal(t, "https://api.foo.com/v1/x", attrs["url.full"].AsString()) +} + +func TestToLogRecord_PromotedAttributes(t *testing.T) { + t.Run("network", func(t *testing.T) { + env := Envelope{Seq: 1, Event: Event{Type: "network_request", Category: Network, + Data: json.RawMessage(`{"method":"GET","url":"https://x/y","status":404}`)}} + rec := toLogRecord(env) + attrs := attrsOf(rec) + assert.Equal(t, "GET", attrs["http.request.method"].AsString()) + assert.Equal(t, "https://x/y", attrs["url.full"].AsString()) + assert.Equal(t, int64(404), attrs["http.response.status_code"].AsInt64()) + }) + t.Run("console", func(t *testing.T) { + env := Envelope{Seq: 2, Event: Event{Type: "console_error", Category: Console, + Data: json.RawMessage(`{"level":"error","text":"boom"}`)}} + rec := toLogRecord(env) + assert.Equal(t, "error", attrsOf(rec)["kernel.console.level"].AsString()) + }) + t.Run("non-object data does not panic", func(t *testing.T) { + env := Envelope{Seq: 3, Event: Event{Type: "x", Category: Network, + Data: json.RawMessage(`["a","b"]`)}} + rec := toLogRecord(env) + assert.Equal(t, "x", attrsOf(rec)["kernel.event.type"].AsString()) + }) } func TestToLogRecord_StructuredBody(t *testing.T) { From e654fa22c3241be5b7ab02b813c4146f188cab6c Mon Sep 17 00:00:00 2001 From: archandatta <35818003+archandatta@users.noreply.github.com> Date: Thu, 25 Jun 2026 12:29:52 +0000 Subject: [PATCH 06/11] Drop never-firing oversized-record guard Envelopes are already capped at ~1MB at publish, so a 4MB per-record check could never fire. Removes the dead guard, its drop counter, and the unused storage logger. --- server/lib/events/otlpstorage.go | 19 +++---------------- 1 file changed, 3 insertions(+), 16 deletions(-) diff --git a/server/lib/events/otlpstorage.go b/server/lib/events/otlpstorage.go index 124bf5ff..3c58b7e6 100644 --- a/server/lib/events/otlpstorage.go +++ b/server/lib/events/otlpstorage.go @@ -5,7 +5,6 @@ import ( "fmt" "log/slog" "sync" - "sync/atomic" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp" @@ -21,10 +20,6 @@ import ( // batching (the real guard against many large records) is deferred. const defaultOTLPMaxBatchRecords = 200 -// maxOTLPRecordBytes drops a single envelope whose data alone would risk the -// relay body cap. A safety net above the publish-time truncation. -const maxOTLPRecordBytes = 4 * 1024 * 1024 - // OTLPConfig configures the OTLP telemetry sink. type OTLPConfig struct { // Endpoint is the host[:port] of the OTLP/HTTP target: the metro-api relay @@ -53,11 +48,9 @@ type OTLPConfig struct { type otlpStorage struct { provider *sdklog.LoggerProvider logger log.Logger - log *slog.Logger - dropped atomic.Uint64 } -func newOTLPStorage(ctx context.Context, cfg OTLPConfig, logger *slog.Logger) (*otlpStorage, error) { +func newOTLPStorage(ctx context.Context, cfg OTLPConfig) (*otlpStorage, error) { opts := []otlploghttp.Option{otlploghttp.WithEndpoint(cfg.Endpoint)} if cfg.URLPath != "" { opts = append(opts, otlploghttp.WithURLPath(cfg.URLPath)) @@ -93,21 +86,15 @@ func newOTLPStorage(ctx context.Context, cfg OTLPConfig, logger *slog.Logger) (* return &otlpStorage{ provider: provider, logger: provider.Logger(otlpScopeName), - log: logger, }, nil } // Append converts an exported-category envelope to an OTLP record and hands it -// to the SDK. Excluded categories and oversized envelopes are dropped. +// to the SDK. Excluded categories are dropped. func (s *otlpStorage) Append(ctx context.Context, env Envelope) error { if !otlpCategoryExported(env.Event.Category) { return nil } - if len(env.Event.Data) > maxOTLPRecordBytes { - n := s.dropped.Add(1) - s.log.Warn("otlp storage: dropped oversized event", "seq", env.Seq, "bytes", len(env.Event.Data), "total_dropped", n) - return nil - } s.logger.Emit(ctx, toLogRecord(env)) return nil } @@ -145,7 +132,7 @@ func (w *OTLPStorageWriter) Start(ctx context.Context) error { if w.started { return fmt.Errorf("otlpstoragewriter: Start called more than once") } - storage, err := newOTLPStorage(ctx, w.cfg, w.log) + storage, err := newOTLPStorage(ctx, w.cfg) if err != nil { return err } From 59215c2eb50d146b8de51839a4fbfe83a81017d9 Mon Sep 17 00:00:00 2001 From: archandatta <35818003+archandatta@users.noreply.github.com> Date: Mon, 29 Jun 2026 14:39:03 +0000 Subject: [PATCH 07/11] Add otlpbench: shared benchmark rig for the delivery-substrate bake-off Substrate-independent harness: a measurement OTLP receiver (record count, end-to-end lag p50/p99, bytes, slow-upstream delay) and a relay-shape load driver (light/heavy profiles at a target rate) feeding the shared converter. Reused unchanged by the server-side substrates; only the driver differs. --- server/cmd/otlpbench/main.go | 213 +++++++++++++++++++++++++++++++++++ server/go.mod | 4 +- 2 files changed, 215 insertions(+), 2 deletions(-) create mode 100644 server/cmd/otlpbench/main.go diff --git a/server/cmd/otlpbench/main.go b/server/cmd/otlpbench/main.go new file mode 100644 index 00000000..ad98b51b --- /dev/null +++ b/server/cmd/otlpbench/main.go @@ -0,0 +1,213 @@ +// Command otlpbench is the shared benchmark rig for the OTLP delivery-substrate +// bake-off. It is substrate-independent: every candidate feeds the same +// converter and is measured by the same receiver. +// +// # terminal 1: measurement receiver (counts records, measures end-to-end lag) +// go run ./cmd/otlpbench -mode=receiver -addr=:4318 [-delay=50ms] +// +// # terminal 2: relay-shape load driver (EventStream -> OTLPStorageWriter -> receiver) +// go run ./cmd/otlpbench -mode=relay -endpoint=127.0.0.1:4318 -path=/v1/logs \ +// -rate=500 -dur=10s -profile=heavy +// +// The server-side substrates (Temporal activity, claim goroutine) reuse the +// same receiver and workload profiles; only the driver differs. +package main + +import ( + "bytes" + "compress/gzip" + "context" + "flag" + "fmt" + "io" + "log/slog" + "net/http" + "os" + "os/signal" + "sort" + "sync" + "syscall" + "time" + + "github.com/kernel/kernel-images/server/lib/events" + oapi "github.com/kernel/kernel-images/server/lib/oapi" + collogspb "go.opentelemetry.io/proto/otlp/collector/logs/v1" + "google.golang.org/protobuf/proto" +) + +func main() { + mode := flag.String("mode", "receiver", "receiver | relay") + addr := flag.String("addr", ":4318", "receiver listen address") + endpoint := flag.String("endpoint", "127.0.0.1:4318", "driver target host[:port]") + path := flag.String("path", "/v1/logs", "driver target path") + delay := flag.Duration("delay", 0, "receiver: artificial per-request delay (simulate slow customer)") + rate := flag.Int("rate", 500, "driver: events per second") + dur := flag.Duration("dur", 10*time.Second, "driver: run duration") + profile := flag.String("profile", "light", "driver: light | heavy") + flag.Parse() + + switch *mode { + case "receiver": + runReceiver(*addr, *delay) + case "relay": + runRelay(*endpoint, *path, *profile, *rate, *dur) + default: + fmt.Fprintf(os.Stderr, "unknown mode %q\n", *mode) + os.Exit(2) + } +} + +func runReceiver(addr string, delay time.Duration) { + var ( + mu sync.Mutex + total int + byteCount int64 + lags []time.Duration + ) + + http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + body, _ := io.ReadAll(r.Body) + data := body + if r.Header.Get("Content-Encoding") == "gzip" { + if gz, err := gzip.NewReader(bytes.NewReader(body)); err == nil { + if d, err := io.ReadAll(gz); err == nil { + data = d + } + } + } + var req collogspb.ExportLogsServiceRequest + if err := proto.Unmarshal(data, &req); err != nil { + http.Error(w, "bad protobuf", http.StatusBadRequest) + return + } + now := time.Now().UnixNano() + mu.Lock() + for _, rl := range req.ResourceLogs { + for _, sl := range rl.ScopeLogs { + for _, lr := range sl.LogRecords { + total++ + if lr.TimeUnixNano > 0 { + lags = append(lags, time.Duration(now-int64(lr.TimeUnixNano))) + } + } + } + } + byteCount += int64(len(data)) + mu.Unlock() + + if delay > 0 { + time.Sleep(delay) + } + resp, _ := proto.Marshal(&collogspb.ExportLogsServiceResponse{}) + w.Header().Set("Content-Type", "application/x-protobuf") + _, _ = w.Write(resp) + }) + + srv := &http.Server{Addr: addr} + go func() { + last := 0 + for range time.Tick(time.Second) { + mu.Lock() + n := total + mu.Unlock() + fmt.Printf("received total=%d (+%d/s)\n", n, n-last) + last = n + } + }() + + go func() { + fmt.Printf("otlpbench receiver listening on %s (delay=%s)\n", addr, delay) + if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { + fmt.Fprintln(os.Stderr, err) + } + }() + + sig := make(chan os.Signal, 1) + signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM) + <-sig + + mu.Lock() + defer mu.Unlock() + fmt.Println("\n=== otlpbench receiver summary ===") + fmt.Printf("records: %d\n", total) + fmt.Printf("bytes: %d\n", byteCount) + fmt.Printf("lag p50: %s\n", pct(lags, 0.50)) + fmt.Printf("lag p99: %s\n", pct(lags, 0.99)) +} + +func runRelay(endpoint, path, profile string, rate int, dur time.Duration) { + es, err := events.NewEventStream(events.EventStreamConfig{RingCapacity: 4096}) + if err != nil { + panic(err) + } + wctx, wcancel := context.WithCancel(context.Background()) + writer := events.NewOTLPStorageWriter(es, events.OTLPConfig{ + Endpoint: endpoint, URLPath: path, Insecure: true, + ServiceName: "kernel-browser", InstanceName: "bench", Metro: "dev-local", + }, slog.Default()) + if err := writer.Start(wctx); err != nil { + panic(err) + } + + const tick = 10 * time.Millisecond + perTick := rate * int(tick) / int(time.Second) + if perTick < 1 { + perTick = 1 + } + published := 0 + deadline := time.Now().Add(dur) + t := time.NewTicker(tick) + defer t.Stop() + for now := range t.C { + if !now.Before(deadline) { + break + } + for i := 0; i < perTick; i++ { + es.Publish(makeEnvelope(profile, published)) + published++ + } + } + + wcancel() + stopCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + _ = writer.Stop(stopCtx) + fmt.Printf("relay driver: published=%d over %s (target %d/s, profile=%s)\n", published, dur, rate, profile) +} + +var heavyBody = `{"status":200,"url":"https://example.com/api","mime_type":"application/json","body":"` + + stringOfLen(8*1024) + `"}` + +func makeEnvelope(profile string, i int) events.Envelope { + meta := map[string]string{"telemetry_session_id": "bench", "cdp_session_id": "s1", "target_id": "t1"} + ev := events.Event{ + Ts: time.Now().UnixMicro(), + Source: oapi.BrowserEventSource{Kind: oapi.BrowserEventSourceKind("cdp"), Metadata: &meta}, + } + if profile == "heavy" { + ev.Type = "network_response" + ev.Category = oapi.TelemetryEventCategory("network") + ev.Data = []byte(heavyBody) + } else { + ev.Type = "page_navigation" + ev.Category = oapi.TelemetryEventCategory("page") + ev.Data = []byte(fmt.Sprintf(`{"url":"https://example.com/p/%d","frame_id":"F1"}`, i)) + } + return events.Envelope{Event: ev} +} + +func stringOfLen(n int) string { + b := make([]byte, n) + for i := range b { + b[i] = 'x' + } + return string(b) +} + +func pct(d []time.Duration, p float64) time.Duration { + if len(d) == 0 { + return 0 + } + sort.Slice(d, func(i, j int) bool { return d[i] < d[j] }) + return d[int(float64(len(d)-1)*p)] +} diff --git a/server/go.mod b/server/go.mod index b45dbe4e..c2a7d8c6 100644 --- a/server/go.mod +++ b/server/go.mod @@ -32,9 +32,11 @@ require ( go.opentelemetry.io/otel/log v0.20.0 go.opentelemetry.io/otel/sdk v1.44.0 go.opentelemetry.io/otel/sdk/log v0.20.0 + go.opentelemetry.io/proto/otlp v1.10.0 golang.org/x/sync v0.20.0 golang.org/x/sys v0.45.0 golang.org/x/term v0.43.0 + google.golang.org/protobuf v1.36.11 gopkg.in/yaml.v3 v3.0.1 ) @@ -109,7 +111,6 @@ require ( go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.39.0 // indirect go.opentelemetry.io/otel/metric v1.44.0 // indirect go.opentelemetry.io/otel/trace v1.44.0 // indirect - go.opentelemetry.io/proto/otlp v1.10.0 // indirect golang.org/x/crypto v0.51.0 // indirect golang.org/x/mod v0.35.0 // indirect golang.org/x/net v0.55.0 // indirect @@ -118,7 +119,6 @@ require ( google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect google.golang.org/grpc v1.81.1 // indirect - google.golang.org/protobuf v1.36.11 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect gorm.io/gorm v1.25.7 // indirect modernc.org/libc v1.22.5 // indirect From 650c0399536780dd5d01a5e438a46923f0f63228 Mon Sep 17 00:00:00 2001 From: archandatta <35818003+archandatta@users.noreply.github.com> Date: Mon, 29 Jun 2026 17:56:52 +0000 Subject: [PATCH 08/11] Add read-latency knob to otlpbench to model server-side pushers The Temporal and claim-goroutine substrates share the relay's convert/batch/POST hot path but add an S2 read hop. The knob injects that hop so end-to-end lag can be compared against the relay baseline. --- server/cmd/otlpbench/main.go | 25 +++++++++++++++++++++---- 1 file changed, 21 insertions(+), 4 deletions(-) diff --git a/server/cmd/otlpbench/main.go b/server/cmd/otlpbench/main.go index ad98b51b..479edc88 100644 --- a/server/cmd/otlpbench/main.go +++ b/server/cmd/otlpbench/main.go @@ -44,13 +44,14 @@ func main() { rate := flag.Int("rate", 500, "driver: events per second") dur := flag.Duration("dur", 10*time.Second, "driver: run duration") profile := flag.String("profile", "light", "driver: light | heavy") + readLatency := flag.Duration("read-latency", 0, "driver: simulated S2 read-hop latency; models the server-side pusher substrates (shapes 2/3)") flag.Parse() switch *mode { case "receiver": runReceiver(*addr, *delay) case "relay": - runRelay(*endpoint, *path, *profile, *rate, *dur) + runRelay(*endpoint, *path, *profile, *rate, *dur, *readLatency) default: fmt.Fprintf(os.Stderr, "unknown mode %q\n", *mode) os.Exit(2) @@ -135,7 +136,7 @@ func runReceiver(addr string, delay time.Duration) { fmt.Printf("lag p99: %s\n", pct(lags, 0.99)) } -func runRelay(endpoint, path, profile string, rate int, dur time.Duration) { +func runRelay(endpoint, path, profile string, rate int, dur time.Duration, readLatency time.Duration) { es, err := events.NewEventStream(events.EventStreamConfig{RingCapacity: 4096}) if err != nil { panic(err) @@ -163,16 +164,32 @@ func runRelay(endpoint, path, profile string, rate int, dur time.Duration) { break } for i := 0; i < perTick; i++ { - es.Publish(makeEnvelope(profile, published)) + env := makeEnvelope(profile, published) + if readLatency > 0 { + // Models the server-side substrates: the record is emitted now + // but only reaches the pusher after the S2 read hop, so the + // measured end-to-end lag includes it. + time.AfterFunc(readLatency, func() { es.Publish(env) }) + } else { + es.Publish(env) + } published++ } } + if readLatency > 0 { + time.Sleep(readLatency + 200*time.Millisecond) + } wcancel() stopCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() _ = writer.Stop(stopCtx) - fmt.Printf("relay driver: published=%d over %s (target %d/s, profile=%s)\n", published, dur, rate, profile) + + shape := "relay (in-VM)" + if readLatency > 0 { + shape = fmt.Sprintf("server-side pusher (modeled, read-latency=%s)", readLatency) + } + fmt.Printf("%s: published=%d over %s (target %d/s, profile=%s)\n", shape, published, dur, rate, profile) } var heavyBody = `{"status":200,"url":"https://example.com/api","mime_type":"application/json","body":"` + From 4a2e2b56eb7682fbd411101ed6013670999d8770 Mon Sep 17 00:00:00 2001 From: archandatta <35818003+archandatta@users.noreply.github.com> Date: Tue, 30 Jun 2026 12:35:59 +0000 Subject: [PATCH 09/11] Add durable mode and fault injection to otlpbench for crash A/B Adds a durable-pusher mode (unbounded log + resumable cursor) and a delivery outage knob, plus unique-record counting in the receiver, so the relay's best-effort loss can be measured head-to-head against at-least-once recovery. --- server/cmd/otlpbench/main.go | 168 +++++++++++++++++++++++++++++++---- 1 file changed, 152 insertions(+), 16 deletions(-) diff --git a/server/cmd/otlpbench/main.go b/server/cmd/otlpbench/main.go index 479edc88..6156f5ec 100644 --- a/server/cmd/otlpbench/main.go +++ b/server/cmd/otlpbench/main.go @@ -45,13 +45,18 @@ func main() { dur := flag.Duration("dur", 10*time.Second, "driver: run duration") profile := flag.String("profile", "light", "driver: light | heavy") readLatency := flag.Duration("read-latency", 0, "driver: simulated S2 read-hop latency; models the server-side pusher substrates (shapes 2/3)") + pauseAt := flag.Duration("pause-at", 0, "driver: when to take the delivery path down (fault injection)") + pauseFor := flag.Duration("pause-for", 0, "driver: how long the delivery path stays down") flag.Parse() + fault := fault{at: *pauseAt, dur: *pauseFor} switch *mode { case "receiver": runReceiver(*addr, *delay) case "relay": - runRelay(*endpoint, *path, *profile, *rate, *dur, *readLatency) + runRelay(*endpoint, *path, *profile, *rate, *dur, *readLatency, fault) + case "durable": + runDurable(*endpoint, *path, *profile, *rate, *dur, fault) default: fmt.Fprintf(os.Stderr, "unknown mode %q\n", *mode) os.Exit(2) @@ -64,6 +69,7 @@ func runReceiver(addr string, delay time.Duration) { total int byteCount int64 lags []time.Duration + seen = map[int64]struct{}{} ) http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { @@ -90,6 +96,11 @@ func runReceiver(addr string, delay time.Duration) { if lr.TimeUnixNano > 0 { lags = append(lags, time.Duration(now-int64(lr.TimeUnixNano))) } + for _, kv := range lr.Attributes { + if kv.Key == "kernel.event.seq" { + seen[kv.Value.GetIntValue()] = struct{}{} + } + } } } } @@ -130,22 +141,41 @@ func runReceiver(addr string, delay time.Duration) { mu.Lock() defer mu.Unlock() fmt.Println("\n=== otlpbench receiver summary ===") - fmt.Printf("records: %d\n", total) - fmt.Printf("bytes: %d\n", byteCount) - fmt.Printf("lag p50: %s\n", pct(lags, 0.50)) - fmt.Printf("lag p99: %s\n", pct(lags, 0.99)) + fmt.Printf("records: %d\n", total) + fmt.Printf("unique: %d\n", len(seen)) + fmt.Printf("duplicates: %d\n", total-len(seen)) + fmt.Printf("bytes: %d\n", byteCount) + fmt.Printf("lag p50: %s\n", pct(lags, 0.50)) + fmt.Printf("lag p99: %s\n", pct(lags, 0.99)) +} + +// fault describes a delivery-path outage injected mid-run. +type fault struct { + at time.Duration + dur time.Duration +} + +func (f fault) active(elapsed time.Duration) bool { + return f.dur > 0 && elapsed >= f.at && elapsed < f.at+f.dur +} + +func newWriter(es *events.EventStream, endpoint, path string) *events.OTLPStorageWriter { + return events.NewOTLPStorageWriter(es, events.OTLPConfig{ + Endpoint: endpoint, URLPath: path, Insecure: true, + ServiceName: "kernel-browser", InstanceName: "bench", Metro: "dev-local", + }, slog.Default()) } -func runRelay(endpoint, path, profile string, rate int, dur time.Duration, readLatency time.Duration) { - es, err := events.NewEventStream(events.EventStreamConfig{RingCapacity: 4096}) +func runRelay(endpoint, path, profile string, rate int, dur time.Duration, readLatency time.Duration, f fault) { + // The relay's source is the volatile in-VM ring (1024 slots). While the + // delivery path is down it keeps filling and overflows; those records are + // lost with no replay. + es, err := events.NewEventStream(events.EventStreamConfig{RingCapacity: 1024}) if err != nil { panic(err) } wctx, wcancel := context.WithCancel(context.Background()) - writer := events.NewOTLPStorageWriter(es, events.OTLPConfig{ - Endpoint: endpoint, URLPath: path, Insecure: true, - ServiceName: "kernel-browser", InstanceName: "bench", Metro: "dev-local", - }, slog.Default()) + writer := newWriter(es, endpoint, path) if err := writer.Start(wctx); err != nil { panic(err) } @@ -156,19 +186,34 @@ func runRelay(endpoint, path, profile string, rate int, dur time.Duration, readL perTick = 1 } published := 0 - deadline := time.Now().Add(dur) + start := time.Now() + deadline := start.Add(dur) + paused, resumed := false, false t := time.NewTicker(tick) defer t.Stop() for now := range t.C { if !now.Before(deadline) { break } + elapsed := now.Sub(start) + if f.active(elapsed) && !paused { + wcancel() + sc, c := context.WithTimeout(context.Background(), 2*time.Second) + _ = writer.Stop(sc) + c() + paused = true + fmt.Printf(" [fault] relay delivery DOWN at %s\n", elapsed.Round(time.Millisecond)) + } + if paused && !resumed && elapsed >= f.at+f.dur { + wctx, wcancel = context.WithCancel(context.Background()) + writer = newWriter(es, endpoint, path) + _ = writer.Start(wctx) + resumed = true + fmt.Printf(" [fault] relay delivery RESTORED at %s\n", elapsed.Round(time.Millisecond)) + } for i := 0; i < perTick; i++ { env := makeEnvelope(profile, published) if readLatency > 0 { - // Models the server-side substrates: the record is emitted now - // but only reaches the pusher after the S2 read hop, so the - // measured end-to-end lag includes it. time.AfterFunc(readLatency, func() { es.Publish(env) }) } else { es.Publish(env) @@ -185,13 +230,104 @@ func runRelay(endpoint, path, profile string, rate int, dur time.Duration, readL defer cancel() _ = writer.Stop(stopCtx) - shape := "relay (in-VM)" + shape := "relay (in-VM, best-effort)" if readLatency > 0 { shape = fmt.Sprintf("server-side pusher (modeled, read-latency=%s)", readLatency) } fmt.Printf("%s: published=%d over %s (target %d/s, profile=%s)\n", shape, published, dur, rate, profile) } +// runDurable models the server-side durable pusher: a producer appends every +// record to an unbounded persistent log (this models the existing reliable +// VM->S2 write), and a pusher reads from a cursor and delivers. While the +// delivery path is down the log keeps growing, so on resume the pusher catches +// up from the cursor and loses nothing. +func runDurable(endpoint, path, profile string, rate int, dur time.Duration, f fault) { + es, err := events.NewEventStream(events.EventStreamConfig{RingCapacity: 262144}) + if err != nil { + panic(err) + } + wctx, wcancel := context.WithCancel(context.Background()) + writer := newWriter(es, endpoint, path) + if err := writer.Start(wctx); err != nil { + panic(err) + } + + const tick = 10 * time.Millisecond + perTick := rate * int(tick) / int(time.Second) + if perTick < 1 { + perTick = 1 + } + + var ( + mu sync.Mutex + logq []events.Envelope + produced int + ) + start := time.Now() + prodDone := make(chan struct{}) + go func() { + defer close(prodDone) + t := time.NewTicker(tick) + defer t.Stop() + deadline := start.Add(dur) + for now := range t.C { + if !now.Before(deadline) { + return + } + mu.Lock() + for i := 0; i < perTick; i++ { + logq = append(logq, makeEnvelope(profile, produced)) + produced++ + } + mu.Unlock() + } + }() + + cursor := 0 + producing := true + loggedDown := false + for producing || func() bool { mu.Lock(); defer mu.Unlock(); return cursor < len(logq) }() { + elapsed := time.Since(start) + if f.active(elapsed) { + if !loggedDown { + fmt.Printf(" [fault] pusher delivery DOWN at %s (log keeps growing)\n", elapsed.Round(time.Millisecond)) + loggedDown = true + } + time.Sleep(tick) + goto checkProd + } + if loggedDown { + fmt.Printf(" [fault] pusher RESUMED at %s, catching up from cursor\n", elapsed.Round(time.Millisecond)) + loggedDown = false + } + for { + mu.Lock() + if cursor >= len(logq) { + mu.Unlock() + break + } + env := logq[cursor] + mu.Unlock() + es.Publish(env) + cursor++ + } + checkProd: + select { + case <-prodDone: + producing = false + default: + time.Sleep(tick) + } + } + + wcancel() + stopCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) + defer cancel() + _ = writer.Stop(stopCtx) + fmt.Printf("durable pusher (at-least-once): produced=%d delivered-from-log=%d (profile=%s)\n", produced, cursor, profile) +} + var heavyBody = `{"status":200,"url":"https://example.com/api","mime_type":"application/json","body":"` + stringOfLen(8*1024) + `"}` From df56297fd8c596455a77c2e2d1e5381667d4cf23 Mon Sep 17 00:00:00 2001 From: archandatta <35818003+archandatta@users.noreply.github.com> Date: Thu, 2 Jul 2026 13:09:40 +0000 Subject: [PATCH 10/11] Surface OTLP export failures and address review nits Wraps the exporter to log delivery failures (the SDK otherwise reports them only through an uninstalled global logger, so loss was silent). Removes the unused MaxBatchRecords knob, omits empty resource attributes, logs the two missing OTLP config fields, decodes the exported payload in the sink test to prove category exclusion, and notes the producer-schema tie in the converter. --- server/cmd/config/config.go | 2 + server/lib/events/otlpconvert.go | 4 +- server/lib/events/otlpstorage.go | 58 ++++++++++++++++++--------- server/lib/events/otlpstorage_test.go | 38 +++++++++++++----- 4 files changed, 71 insertions(+), 31 deletions(-) diff --git a/server/cmd/config/config.go b/server/cmd/config/config.go index 96782acd..0b6f5fe2 100644 --- a/server/cmd/config/config.go +++ b/server/cmd/config/config.go @@ -93,6 +93,8 @@ func (c *Config) LogValue() slog.Value { slog.Bool("otlp_insecure", c.OTLPInsecure), slog.String("otlp_instance_jwt", otlpJWT), slog.String("otlp_service_name", c.OTLPServiceName), + slog.String("otlp_instance_name", c.OTLPInstanceName), + slog.String("otlp_metro", c.OTLPMetro), ) } diff --git a/server/lib/events/otlpconvert.go b/server/lib/events/otlpconvert.go index f8cc170d..24b2b89c 100644 --- a/server/lib/events/otlpconvert.go +++ b/server/lib/events/otlpconvert.go @@ -82,7 +82,9 @@ func toLogRecord(env Envelope) log.Record { // promotedAttributes lifts high-value payload fields into typed, queryable // attributes (OTel semantic conventions where they exist) so they stay // filterable in backends that do not flatten a structured body (Datadog, Loki). -// The full payload remains in the body for fidelity. +// The full payload remains in the body for fidelity. The data keys read here +// mirror the producer event-data schema (network/console event data in +// openapi.yaml); keep them in sync if that schema changes. func promotedAttributes(cat oapi.TelemetryEventCategory, data map[string]any) []log.KeyValue { var out []log.KeyValue switch cat { diff --git a/server/lib/events/otlpstorage.go b/server/lib/events/otlpstorage.go index 3c58b7e6..c5f9bf7f 100644 --- a/server/lib/events/otlpstorage.go +++ b/server/lib/events/otlpstorage.go @@ -5,6 +5,7 @@ import ( "fmt" "log/slog" "sync" + "sync/atomic" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp" @@ -14,10 +15,11 @@ import ( semconv "go.opentelemetry.io/otel/semconv/v1.27.0" ) -// defaultOTLPMaxBatchRecords bounds records per export request. The metro-api -// relay caps bodies at 10MB; envelopes are already capped at ~1MB each at -// publish, so this keeps the common case well under the limit. Byte-based -// batching (the real guard against many large records) is deferred. +// defaultOTLPMaxBatchRecords bounds records per export request. Envelopes are +// capped at ~1MB each at publish, so a small batch keeps typical requests +// under the metro-api relay's 10MB body cap. This is a count bound, not a byte +// bound: byte-based batching (the real guard against many large records) is +// deferred. const defaultOTLPMaxBatchRecords = 200 // OTLPConfig configures the OTLP telemetry sink. @@ -37,9 +39,24 @@ type OTLPConfig struct { ServiceName string InstanceName string Metro string +} + +// loggingExporter wraps the OTLP exporter to surface export failures. The SDK +// otherwise reports them only through a global logger that is not installed +// here, so delivery loss would be silent. +type loggingExporter struct { + sdklog.Exporter + log *slog.Logger + failures atomic.Uint64 +} - // MaxBatchRecords bounds records per export. Defaults to 200. - MaxBatchRecords int +func (e *loggingExporter) Export(ctx context.Context, records []sdklog.Record) error { + err := e.Exporter.Export(ctx, records) + if err != nil { + n := e.failures.Add(1) + e.log.Warn("otlp export failed", "records", len(records), "err", err, "total_failures", n) + } + return err } // otlpStorage implements Storage by converting envelopes to OTLP log records @@ -50,7 +67,7 @@ type otlpStorage struct { logger log.Logger } -func newOTLPStorage(ctx context.Context, cfg OTLPConfig) (*otlpStorage, error) { +func newOTLPStorage(ctx context.Context, cfg OTLPConfig, log *slog.Logger) (*otlpStorage, error) { opts := []otlploghttp.Option{otlploghttp.WithEndpoint(cfg.Endpoint)} if cfg.URLPath != "" { opts = append(opts, otlploghttp.WithURLPath(cfg.URLPath)) @@ -67,17 +84,19 @@ func newOTLPStorage(ctx context.Context, cfg OTLPConfig) (*otlpStorage, error) { return nil, fmt.Errorf("otlp storage: new exporter: %w", err) } - res := resource.NewSchemaless( - semconv.ServiceName(cfg.ServiceName), - attribute.String("kernel.instance_name", cfg.InstanceName), - attribute.String("kernel.metro", cfg.Metro), - ) - - maxBatch := cfg.MaxBatchRecords - if maxBatch == 0 { - maxBatch = defaultOTLPMaxBatchRecords + attrs := []attribute.KeyValue{semconv.ServiceName(cfg.ServiceName)} + if cfg.InstanceName != "" { + attrs = append(attrs, attribute.String("kernel.instance_name", cfg.InstanceName)) + } + if cfg.Metro != "" { + attrs = append(attrs, attribute.String("kernel.metro", cfg.Metro)) } - processor := sdklog.NewBatchProcessor(exporter, sdklog.WithExportMaxBatchSize(maxBatch)) + res := resource.NewSchemaless(attrs...) + + processor := sdklog.NewBatchProcessor( + &loggingExporter{Exporter: exporter, log: log}, + sdklog.WithExportMaxBatchSize(defaultOTLPMaxBatchRecords), + ) provider := sdklog.NewLoggerProvider( sdklog.WithResource(res), sdklog.WithProcessor(processor), @@ -90,7 +109,8 @@ func newOTLPStorage(ctx context.Context, cfg OTLPConfig) (*otlpStorage, error) { } // Append converts an exported-category envelope to an OTLP record and hands it -// to the SDK. Excluded categories are dropped. +// to the SDK. Excluded categories are dropped. Delivery failures surface via +// the loggingExporter, not this return value (the SDK batches asynchronously). func (s *otlpStorage) Append(ctx context.Context, env Envelope) error { if !otlpCategoryExported(env.Event.Category) { return nil @@ -132,7 +152,7 @@ func (w *OTLPStorageWriter) Start(ctx context.Context) error { if w.started { return fmt.Errorf("otlpstoragewriter: Start called more than once") } - storage, err := newOTLPStorage(ctx, w.cfg) + storage, err := newOTLPStorage(ctx, w.cfg, w.log) if err != nil { return err } diff --git a/server/lib/events/otlpstorage_test.go b/server/lib/events/otlpstorage_test.go index 731b739a..8f46a69a 100644 --- a/server/lib/events/otlpstorage_test.go +++ b/server/lib/events/otlpstorage_test.go @@ -2,6 +2,7 @@ package events import ( "context" + "io" "log/slog" "net/http" "net/http/httptest" @@ -12,18 +13,32 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + collogspb "go.opentelemetry.io/proto/otlp/collector/logs/v1" + "google.golang.org/protobuf/proto" ) // TestOTLPStorageWriter_ExportsEvents drives the full sink against a local HTTP -// server standing in for the relay/collector, verifying records reach the -// configured path with the configured headers. +// server standing in for the relay/collector. It decodes the exported OTLP +// payload to confirm the request lands with the configured path/headers and +// that an excluded category (screenshot) never reaches the receiver. func TestOTLPStorageWriter_ExportsEvents(t *testing.T) { var mu sync.Mutex - var paths, auths []string + var paths, auths, eventNames []string srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + body, err := io.ReadAll(r.Body) + require.NoError(t, err) + var req collogspb.ExportLogsServiceRequest + require.NoError(t, proto.Unmarshal(body, &req)) mu.Lock() paths = append(paths, r.URL.Path) auths = append(auths, r.Header.Get("Authorization")) + for _, rl := range req.ResourceLogs { + for _, sl := range rl.ScopeLogs { + for _, lr := range sl.LogRecords { + eventNames = append(eventNames, lr.EventName) + } + } + } mu.Unlock() w.WriteHeader(http.StatusOK) })) @@ -33,14 +48,13 @@ func TestOTLPStorageWriter_ExportsEvents(t *testing.T) { require.NoError(t, err) cfg := OTLPConfig{ - Endpoint: strings.TrimPrefix(srv.URL, "http://"), - URLPath: "/otlp-relay/v1/logs", - Insecure: true, - Headers: map[string]string{"Authorization": "Bearer test-jwt"}, - ServiceName: "kernel-browser", - InstanceName: "browser-1", - Metro: "dev-iad", - MaxBatchRecords: 10, + Endpoint: strings.TrimPrefix(srv.URL, "http://"), + URLPath: "/otlp-relay/v1/logs", + Insecure: true, + Headers: map[string]string{"Authorization": "Bearer test-jwt"}, + ServiceName: "kernel-browser", + InstanceName: "browser-1", + Metro: "dev-iad", } wtr := NewOTLPStorageWriter(es, cfg, slog.Default()) @@ -60,4 +74,6 @@ func TestOTLPStorageWriter_ExportsEvents(t *testing.T) { require.NotEmpty(t, paths, "expected at least one export request") assert.Equal(t, "/otlp-relay/v1/logs", paths[0]) assert.Equal(t, "Bearer test-jwt", auths[0]) + // The excluded screenshot must not reach the receiver; only the page event does. + assert.Equal(t, []string{"page_navigation"}, eventNames) } From aaac9a07e9f69ad6df33480ce835717722756262 Mon Sep 17 00:00:00 2001 From: archandatta <35818003+archandatta@users.noreply.github.com> Date: Thu, 2 Jul 2026 13:10:05 +0000 Subject: [PATCH 11/11] Remove benchmark rig from the feature PR The otlpbench rig (including the alternative-substrate durable-pusher model and fault injection) is delivery-substrate exploration, not part of the in-VM exporter. Moved to its own branch to keep this PR reviewable in isolation. --- server/cmd/otlpbench/main.go | 366 ----------------------------------- 1 file changed, 366 deletions(-) delete mode 100644 server/cmd/otlpbench/main.go diff --git a/server/cmd/otlpbench/main.go b/server/cmd/otlpbench/main.go deleted file mode 100644 index 6156f5ec..00000000 --- a/server/cmd/otlpbench/main.go +++ /dev/null @@ -1,366 +0,0 @@ -// Command otlpbench is the shared benchmark rig for the OTLP delivery-substrate -// bake-off. It is substrate-independent: every candidate feeds the same -// converter and is measured by the same receiver. -// -// # terminal 1: measurement receiver (counts records, measures end-to-end lag) -// go run ./cmd/otlpbench -mode=receiver -addr=:4318 [-delay=50ms] -// -// # terminal 2: relay-shape load driver (EventStream -> OTLPStorageWriter -> receiver) -// go run ./cmd/otlpbench -mode=relay -endpoint=127.0.0.1:4318 -path=/v1/logs \ -// -rate=500 -dur=10s -profile=heavy -// -// The server-side substrates (Temporal activity, claim goroutine) reuse the -// same receiver and workload profiles; only the driver differs. -package main - -import ( - "bytes" - "compress/gzip" - "context" - "flag" - "fmt" - "io" - "log/slog" - "net/http" - "os" - "os/signal" - "sort" - "sync" - "syscall" - "time" - - "github.com/kernel/kernel-images/server/lib/events" - oapi "github.com/kernel/kernel-images/server/lib/oapi" - collogspb "go.opentelemetry.io/proto/otlp/collector/logs/v1" - "google.golang.org/protobuf/proto" -) - -func main() { - mode := flag.String("mode", "receiver", "receiver | relay") - addr := flag.String("addr", ":4318", "receiver listen address") - endpoint := flag.String("endpoint", "127.0.0.1:4318", "driver target host[:port]") - path := flag.String("path", "/v1/logs", "driver target path") - delay := flag.Duration("delay", 0, "receiver: artificial per-request delay (simulate slow customer)") - rate := flag.Int("rate", 500, "driver: events per second") - dur := flag.Duration("dur", 10*time.Second, "driver: run duration") - profile := flag.String("profile", "light", "driver: light | heavy") - readLatency := flag.Duration("read-latency", 0, "driver: simulated S2 read-hop latency; models the server-side pusher substrates (shapes 2/3)") - pauseAt := flag.Duration("pause-at", 0, "driver: when to take the delivery path down (fault injection)") - pauseFor := flag.Duration("pause-for", 0, "driver: how long the delivery path stays down") - flag.Parse() - - fault := fault{at: *pauseAt, dur: *pauseFor} - switch *mode { - case "receiver": - runReceiver(*addr, *delay) - case "relay": - runRelay(*endpoint, *path, *profile, *rate, *dur, *readLatency, fault) - case "durable": - runDurable(*endpoint, *path, *profile, *rate, *dur, fault) - default: - fmt.Fprintf(os.Stderr, "unknown mode %q\n", *mode) - os.Exit(2) - } -} - -func runReceiver(addr string, delay time.Duration) { - var ( - mu sync.Mutex - total int - byteCount int64 - lags []time.Duration - seen = map[int64]struct{}{} - ) - - http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { - body, _ := io.ReadAll(r.Body) - data := body - if r.Header.Get("Content-Encoding") == "gzip" { - if gz, err := gzip.NewReader(bytes.NewReader(body)); err == nil { - if d, err := io.ReadAll(gz); err == nil { - data = d - } - } - } - var req collogspb.ExportLogsServiceRequest - if err := proto.Unmarshal(data, &req); err != nil { - http.Error(w, "bad protobuf", http.StatusBadRequest) - return - } - now := time.Now().UnixNano() - mu.Lock() - for _, rl := range req.ResourceLogs { - for _, sl := range rl.ScopeLogs { - for _, lr := range sl.LogRecords { - total++ - if lr.TimeUnixNano > 0 { - lags = append(lags, time.Duration(now-int64(lr.TimeUnixNano))) - } - for _, kv := range lr.Attributes { - if kv.Key == "kernel.event.seq" { - seen[kv.Value.GetIntValue()] = struct{}{} - } - } - } - } - } - byteCount += int64(len(data)) - mu.Unlock() - - if delay > 0 { - time.Sleep(delay) - } - resp, _ := proto.Marshal(&collogspb.ExportLogsServiceResponse{}) - w.Header().Set("Content-Type", "application/x-protobuf") - _, _ = w.Write(resp) - }) - - srv := &http.Server{Addr: addr} - go func() { - last := 0 - for range time.Tick(time.Second) { - mu.Lock() - n := total - mu.Unlock() - fmt.Printf("received total=%d (+%d/s)\n", n, n-last) - last = n - } - }() - - go func() { - fmt.Printf("otlpbench receiver listening on %s (delay=%s)\n", addr, delay) - if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed { - fmt.Fprintln(os.Stderr, err) - } - }() - - sig := make(chan os.Signal, 1) - signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM) - <-sig - - mu.Lock() - defer mu.Unlock() - fmt.Println("\n=== otlpbench receiver summary ===") - fmt.Printf("records: %d\n", total) - fmt.Printf("unique: %d\n", len(seen)) - fmt.Printf("duplicates: %d\n", total-len(seen)) - fmt.Printf("bytes: %d\n", byteCount) - fmt.Printf("lag p50: %s\n", pct(lags, 0.50)) - fmt.Printf("lag p99: %s\n", pct(lags, 0.99)) -} - -// fault describes a delivery-path outage injected mid-run. -type fault struct { - at time.Duration - dur time.Duration -} - -func (f fault) active(elapsed time.Duration) bool { - return f.dur > 0 && elapsed >= f.at && elapsed < f.at+f.dur -} - -func newWriter(es *events.EventStream, endpoint, path string) *events.OTLPStorageWriter { - return events.NewOTLPStorageWriter(es, events.OTLPConfig{ - Endpoint: endpoint, URLPath: path, Insecure: true, - ServiceName: "kernel-browser", InstanceName: "bench", Metro: "dev-local", - }, slog.Default()) -} - -func runRelay(endpoint, path, profile string, rate int, dur time.Duration, readLatency time.Duration, f fault) { - // The relay's source is the volatile in-VM ring (1024 slots). While the - // delivery path is down it keeps filling and overflows; those records are - // lost with no replay. - es, err := events.NewEventStream(events.EventStreamConfig{RingCapacity: 1024}) - if err != nil { - panic(err) - } - wctx, wcancel := context.WithCancel(context.Background()) - writer := newWriter(es, endpoint, path) - if err := writer.Start(wctx); err != nil { - panic(err) - } - - const tick = 10 * time.Millisecond - perTick := rate * int(tick) / int(time.Second) - if perTick < 1 { - perTick = 1 - } - published := 0 - start := time.Now() - deadline := start.Add(dur) - paused, resumed := false, false - t := time.NewTicker(tick) - defer t.Stop() - for now := range t.C { - if !now.Before(deadline) { - break - } - elapsed := now.Sub(start) - if f.active(elapsed) && !paused { - wcancel() - sc, c := context.WithTimeout(context.Background(), 2*time.Second) - _ = writer.Stop(sc) - c() - paused = true - fmt.Printf(" [fault] relay delivery DOWN at %s\n", elapsed.Round(time.Millisecond)) - } - if paused && !resumed && elapsed >= f.at+f.dur { - wctx, wcancel = context.WithCancel(context.Background()) - writer = newWriter(es, endpoint, path) - _ = writer.Start(wctx) - resumed = true - fmt.Printf(" [fault] relay delivery RESTORED at %s\n", elapsed.Round(time.Millisecond)) - } - for i := 0; i < perTick; i++ { - env := makeEnvelope(profile, published) - if readLatency > 0 { - time.AfterFunc(readLatency, func() { es.Publish(env) }) - } else { - es.Publish(env) - } - published++ - } - } - if readLatency > 0 { - time.Sleep(readLatency + 200*time.Millisecond) - } - - wcancel() - stopCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) - defer cancel() - _ = writer.Stop(stopCtx) - - shape := "relay (in-VM, best-effort)" - if readLatency > 0 { - shape = fmt.Sprintf("server-side pusher (modeled, read-latency=%s)", readLatency) - } - fmt.Printf("%s: published=%d over %s (target %d/s, profile=%s)\n", shape, published, dur, rate, profile) -} - -// runDurable models the server-side durable pusher: a producer appends every -// record to an unbounded persistent log (this models the existing reliable -// VM->S2 write), and a pusher reads from a cursor and delivers. While the -// delivery path is down the log keeps growing, so on resume the pusher catches -// up from the cursor and loses nothing. -func runDurable(endpoint, path, profile string, rate int, dur time.Duration, f fault) { - es, err := events.NewEventStream(events.EventStreamConfig{RingCapacity: 262144}) - if err != nil { - panic(err) - } - wctx, wcancel := context.WithCancel(context.Background()) - writer := newWriter(es, endpoint, path) - if err := writer.Start(wctx); err != nil { - panic(err) - } - - const tick = 10 * time.Millisecond - perTick := rate * int(tick) / int(time.Second) - if perTick < 1 { - perTick = 1 - } - - var ( - mu sync.Mutex - logq []events.Envelope - produced int - ) - start := time.Now() - prodDone := make(chan struct{}) - go func() { - defer close(prodDone) - t := time.NewTicker(tick) - defer t.Stop() - deadline := start.Add(dur) - for now := range t.C { - if !now.Before(deadline) { - return - } - mu.Lock() - for i := 0; i < perTick; i++ { - logq = append(logq, makeEnvelope(profile, produced)) - produced++ - } - mu.Unlock() - } - }() - - cursor := 0 - producing := true - loggedDown := false - for producing || func() bool { mu.Lock(); defer mu.Unlock(); return cursor < len(logq) }() { - elapsed := time.Since(start) - if f.active(elapsed) { - if !loggedDown { - fmt.Printf(" [fault] pusher delivery DOWN at %s (log keeps growing)\n", elapsed.Round(time.Millisecond)) - loggedDown = true - } - time.Sleep(tick) - goto checkProd - } - if loggedDown { - fmt.Printf(" [fault] pusher RESUMED at %s, catching up from cursor\n", elapsed.Round(time.Millisecond)) - loggedDown = false - } - for { - mu.Lock() - if cursor >= len(logq) { - mu.Unlock() - break - } - env := logq[cursor] - mu.Unlock() - es.Publish(env) - cursor++ - } - checkProd: - select { - case <-prodDone: - producing = false - default: - time.Sleep(tick) - } - } - - wcancel() - stopCtx, cancel := context.WithTimeout(context.Background(), 15*time.Second) - defer cancel() - _ = writer.Stop(stopCtx) - fmt.Printf("durable pusher (at-least-once): produced=%d delivered-from-log=%d (profile=%s)\n", produced, cursor, profile) -} - -var heavyBody = `{"status":200,"url":"https://example.com/api","mime_type":"application/json","body":"` + - stringOfLen(8*1024) + `"}` - -func makeEnvelope(profile string, i int) events.Envelope { - meta := map[string]string{"telemetry_session_id": "bench", "cdp_session_id": "s1", "target_id": "t1"} - ev := events.Event{ - Ts: time.Now().UnixMicro(), - Source: oapi.BrowserEventSource{Kind: oapi.BrowserEventSourceKind("cdp"), Metadata: &meta}, - } - if profile == "heavy" { - ev.Type = "network_response" - ev.Category = oapi.TelemetryEventCategory("network") - ev.Data = []byte(heavyBody) - } else { - ev.Type = "page_navigation" - ev.Category = oapi.TelemetryEventCategory("page") - ev.Data = []byte(fmt.Sprintf(`{"url":"https://example.com/p/%d","frame_id":"F1"}`, i)) - } - return events.Envelope{Event: ev} -} - -func stringOfLen(n int) string { - b := make([]byte, n) - for i := range b { - b[i] = 'x' - } - return string(b) -} - -func pct(d []time.Duration, p float64) time.Duration { - if len(d) == 0 { - return 0 - } - sort.Slice(d, func(i, j int) bool { return d[i] < d[j] }) - return d[int(float64(len(d)-1)*p)] -}