diff --git a/server/cmd/api/main.go b/server/cmd/api/main.go index b3500a52..5a0778f9 100644 --- a/server/cmd/api/main.go +++ b/server/cmd/api/main.go @@ -17,6 +17,8 @@ import ( "github.com/ghodss/yaml" "github.com/go-chi/chi/v5" chiMiddleware "github.com/go-chi/chi/v5/middleware" + "github.com/go-logr/logr" + "go.opentelemetry.io/otel" "golang.org/x/sync/errgroup" serverpkg "github.com/kernel/kernel-images/server" @@ -125,6 +127,36 @@ 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.InstanceJWT != "" { + headers["Authorization"] = "Bearer " + config.InstanceJWT + } + slogger.Info("OTLP export enabled", "endpoint", config.OTLPEndpoint, "path", config.OTLPPath) + // The OTel log SDK reports batch-queue drops through its global logger at + // logr V(1), which slog renders just below Info, so a default Info handler + // would swallow it. Wire it to a handler that admits that level so records + // dropped under sustained backpressure are observable, not silently lost. + otelDiag := slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo - 1}) + otel.SetLogger(logr.FromSlogHandler(otelDiag)) + otlpWriter = events.NewOTLPStorageWriter(eventStream, events.OTLPConfig{ + Endpoint: config.OTLPEndpoint, + URLPath: config.OTLPPath, + Insecure: config.OTLPInsecure, + Headers: headers, + ServiceName: config.OTLPServiceName, + InstanceName: config.InstanceName, + Metro: config.MetroName, + }, slogger) + if err := otlpWriter.Start(ctx); err != nil { + // Best-effort sink: a failed exporter must not take down the browser. + slogger.Error("OTLP export disabled: storage writer failed to start", "err", err) + otlpWriter = nil + } + } + apiService, err := api.New( recorder.NewFFmpegManager(), recorder.NewFFmpegRecorderFactory(config.PathToFFmpeg, defaultParams, stz), @@ -306,6 +338,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..7bd25cc1 100644 --- a/server/cmd/config/config.go +++ b/server/cmd/config/config.go @@ -47,6 +47,19 @@ type Config struct { S2Basin string `envconfig:"S2_BASIN" default:""` S2AccessToken string `envconfig:"S2_ACCESS_TOKEN" default:""` S2Stream string `envconfig:"S2_STREAM" default:""` + + // Browser-telemetry OTLP export. OTLPEndpoint (host[:port]) must be set to + // enable the OTLP sink. The BTEL_ prefix avoids collision with the standard + // OTEL_ vars that configure the API server's own telemetry. + OTLPEndpoint string `envconfig:"BTEL_OTLP_ENDPOINT" default:""` + OTLPPath string `envconfig:"BTEL_OTLP_PATH" default:"/v1/logs"` + OTLPInsecure bool `envconfig:"BTEL_OTLP_INSECURE" default:"false"` + OTLPServiceName string `envconfig:"BTEL_OTLP_SERVICE_NAME" default:"kernel-browser"` + // Platform-injected identity, reused to stamp the OTLP Resource. These are + // the same envs the VM already receives. + InstanceJWT string `envconfig:"KERNEL_INSTANCE_JWT" default:""` + InstanceName string `envconfig:"INST_NAME" default:""` + MetroName string `envconfig:"METRO_NAME" default:""` } // LogValue implements slog.LogValuer, redacting secret fields. @@ -55,6 +68,10 @@ func (c *Config) LogValue() slog.Value { if c.S2AccessToken != "" { s2AccessToken = "[redacted]" } + otlpJWT := "" + if c.InstanceJWT != "" { + otlpJWT = "[redacted]" + } return slog.GroupValue( slog.Int("port", c.Port), slog.Int("frame_rate", c.FrameRate), @@ -73,6 +90,13 @@ 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), + slog.String("otlp_instance_name", c.InstanceName), + slog.String("otlp_metro", c.MetroName), ) } diff --git a/server/cmd/config/config_test.go b/server/cmd/config/config_test.go index 27dd8b3d..f3d21bf2 100644 --- a/server/cmd/config/config_test.go +++ b/server/cmd/config/config_test.go @@ -31,6 +31,8 @@ func TestLoad(t *testing.T) { ChromeDriverProxyPort: 9224, ChromeDriverUpstreamAddr: "127.0.0.1:9225", DevToolsProxyAddr: "127.0.0.1:9222", + OTLPPath: "/v1/logs", + OTLPServiceName: "kernel-browser", }, }, { @@ -63,6 +65,8 @@ func TestLoad(t *testing.T) { ChromeDriverProxyPort: 5432, ChromeDriverUpstreamAddr: "127.0.0.1:9999", DevToolsProxyAddr: "127.0.0.1:9876", + OTLPPath: "/v1/logs", + OTLPServiceName: "kernel-browser", }, }, { @@ -85,6 +89,8 @@ func TestLoad(t *testing.T) { ChromeDriverProxyPort: 9224, ChromeDriverUpstreamAddr: "127.0.0.1:9225", DevToolsProxyAddr: "10.0.0.1:1234", + OTLPPath: "/v1/logs", + OTLPServiceName: "kernel-browser", }, }, { diff --git a/server/go.mod b/server/go.mod index f2518e7a..5786a9c0 100644 --- a/server/go.mod +++ b/server/go.mod @@ -16,6 +16,7 @@ require ( github.com/ghodss/yaml v1.0.0 github.com/glebarez/sqlite v1.11.0 github.com/go-chi/chi/v5 v5.2.1 + github.com/go-logr/logr v1.4.3 github.com/google/uuid v1.6.0 github.com/kelseyhightower/envconfig v1.4.0 github.com/kernel/hypeman-go v0.20.0 @@ -27,9 +28,16 @@ require ( github.com/samber/lo v1.52.0 github.com/stretchr/testify v1.11.1 github.com/testcontainers/testcontainers-go v0.40.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 v1.44.0 + go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.20.0 + 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 ) @@ -39,6 +47,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 @@ -53,12 +62,11 @@ require ( github.com/ebitengine/purego v0.8.4 // indirect github.com/felixge/httpsnoop v1.0.4 // indirect github.com/glebarez/go-sqlite v1.21.2 // indirect - github.com/go-logr/logr v1.4.3 // indirect github.com/go-logr/stdr v1.2.2 // indirect 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 @@ -100,21 +108,17 @@ 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/exporters/otlp/otlptrace v1.39.0 // indirect - go.opentelemetry.io/otel/metric v1.39.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 - 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/otel/metric v1.44.0 // indirect + go.opentelemetry.io/otel/trace v1.44.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 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 1950c7a2..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= @@ -256,30 +260,38 @@ 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/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= 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/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/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/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.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.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= @@ -287,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= @@ -315,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= @@ -352,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/otlpconvert.go b/server/lib/events/otlpconvert.go new file mode 100644 index 00000000..a49ed4fa --- /dev/null +++ b/server/lib/events/otlpconvert.go @@ -0,0 +1,165 @@ +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) + + // 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) + 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)), + ) + 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)) + } + // Producer metadata keys (telemetry_session_id, cdp ids) pass through under + // the kernel. prefix as-is: they are producer snake_case, unlike the dotted + // converter-owned keys above. + if ev.Source.Metadata != nil { + for k, v := range *ev.Source.Metadata { + 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 +} + +// Producer event-data keys promoted to queryable attributes. They mirror the +// network/console event-data fields in openapi.yaml; keep them in sync if that +// schema changes. +const ( + dataKeyMethod = "method" + dataKeyURL = "url" + dataKeyStatus = "status" + dataKeyLevel = "level" +) + +// 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[dataKeyMethod].(string); ok { + out = append(out, log.String("http.request.method", v)) + } + if v, ok := data[dataKeyURL].(string); ok { + out = append(out, log.String("url.full", v)) + } + if v, ok := data[dataKeyStatus].(float64); ok { + out = append(out, log.Int64("http.response.status_code", int64(v))) + } + case Console: + if v, ok := data[dataKeyLevel].(string); ok { + out = append(out, log.String("kernel.console.level", v)) + } + } + return out +} + +// otlpSeverity maps an event type to an OTLP severity. It is a deliberately +// coarse, best-effort mapping keyed off the producer's event-type string +// conventions (the literal "console_error" and the "_failed" suffix), not the +// generated event-type constants: those live in the producer package +// (cdpmonitor), which this package cannot import without a cycle. If those type +// names change, update this mapping in lockstep. +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" + } +} + +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..83d8718f --- /dev/null +++ b/server/lib/events/otlpconvert_test.go @@ -0,0 +1,175 @@ +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_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()) + 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()) + // 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) { + // network_response carries method/url/status; network_request has no status. + env := Envelope{Seq: 1, Event: Event{Type: "network_response", 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) { + 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_Truncated(t *testing.T) { + // Absent data (never set) and the production truncation output (payload + // replaced with JSON null at publish, not stripped to nil) both yield an + // empty body, but via different branches; cover both. + t.Run("data absent", func(t *testing.T) { + env := Envelope{Seq: 9, Event: Event{Ts: 1, Type: "network_response", Category: Network, Truncated: true}} + 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()) + }) + t.Run("data replaced with json null", func(t *testing.T) { + env := Envelope{Seq: 9, Event: Event{Ts: 1, Type: "network_response", Category: Network, Truncated: true, Data: json.RawMessage(`null`)}} + rec := toLogRecord(env) + assert.Equal(t, log.KindEmpty, rec.Body().Kind(), "json null payload yields an empty body") + assert.Equal(t, true, attrsOf(rec)["kernel.truncated"].AsBool()) + }) +} + +func TestToLogRecord_NumberCoercion(t *testing.T) { + env := Envelope{Seq: 1, Event: Event{Ts: 1, Type: "x", Category: Network, + Data: json.RawMessage(`{"small":200,"neg":-5,"frac":1.5,"big":1e19,"nullish":null}`)}} + rec := toLogRecord(env) + got := make(map[string]log.Value) + for _, kv := range rec.Body().AsMap() { + got[kv.Key] = kv.Value + } + assert.Equal(t, log.KindInt64, got["small"].Kind()) + assert.Equal(t, int64(-5), got["neg"].AsInt64()) + assert.Equal(t, log.KindFloat64, got["frac"].Kind()) + // 1e19 exceeds MaxInt64, so it stays a float rather than overflowing int64. + assert.Equal(t, log.KindFloat64, got["big"].Kind()) + // A JSON null inside the payload maps to an empty value, not "". + assert.Equal(t, log.KindEmpty, got["nullish"].Kind()) +} + +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)) +} diff --git a/server/lib/events/otlpstorage.go b/server/lib/events/otlpstorage.go new file mode 100644 index 00000000..54c1103f --- /dev/null +++ b/server/lib/events/otlpstorage.go @@ -0,0 +1,196 @@ +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. Envelopes are +// capped at ~1MB each at publish, so a small batch keeps typical requests +// under a common 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. +type OTLPConfig struct { + // Endpoint is the host[:port] of the OTLP/HTTP target: a collector in + // development, or a forwarding relay in production. + Endpoint string + // URLPath overrides the request path. 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 +} + +// loggingExporter wraps the OTLP exporter to surface export-time failures +// (network / customer-endpoint errors) with a running failure count. Queue +// drops under sustained backpressure are best-effort (the SDK drops the oldest +// record, bounded by the batch queue size) and are reported separately by the +// SDK's global logger, which main wires to our logger so they are observable. +type loggingExporter struct { + sdklog.Exporter + log *slog.Logger + failures atomic.Uint64 +} + +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 +// 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 +} + +func newOTLPStorage(ctx context.Context, cfg OTLPConfig, log *slog.Logger) (*otlpStorage, error) { + if cfg.Endpoint == "" { + return nil, fmt.Errorf("otlp storage: endpoint is required") + } + + 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) + } + + 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)) + } + res := resource.NewSchemaless(attrs...) + + processor := sdklog.NewBatchProcessor( + &loggingExporter{Exporter: exporter, log: log}, + sdklog.WithExportMaxBatchSize(defaultOTLPMaxBatchRecords), + ) + provider := sdklog.NewLoggerProvider( + sdklog.WithResource(res), + sdklog.WithProcessor(processor), + ) + + return &otlpStorage{ + provider: provider, + logger: provider.Logger(otlpScopeName), + }, nil +} + +// Append converts an exported-category envelope to an OTLP record and hands it +// 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 + } + 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..ea64b21a --- /dev/null +++ b/server/lib/events/otlpstorage_test.go @@ -0,0 +1,120 @@ +package events + +import ( + "context" + "errors" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + sdklog "go.opentelemetry.io/otel/sdk/log" + collogspb "go.opentelemetry.io/proto/otlp/collector/logs/v1" + "google.golang.org/protobuf/proto" +) + +// stubExporter is a controllable sdklog.Exporter for testing the wrapper. +type stubExporter struct{ err error } + +func (s stubExporter) Export(context.Context, []sdklog.Record) error { return s.err } +func (s stubExporter) Shutdown(context.Context) error { return nil } +func (s stubExporter) ForceFlush(context.Context) error { return nil } + +// TestLoggingExporter_CountsExportFailures confirms the wrapper surfaces export +// failures (the reason it exists) and leaves the counter untouched on success. +func TestLoggingExporter_CountsExportFailures(t *testing.T) { + failing := &loggingExporter{Exporter: stubExporter{err: errors.New("boom")}, log: slog.Default()} + require.Error(t, failing.Export(context.Background(), nil)) + require.Error(t, failing.Export(context.Background(), nil)) + assert.Equal(t, uint64(2), failing.failures.Load()) + + ok := &loggingExporter{Exporter: stubExporter{}, log: slog.Default()} + require.NoError(t, ok.Export(context.Background(), nil)) + assert.Equal(t, uint64(0), ok.failures.Load()) +} + +// TestOTLPStorageWriter_ExportsEvents drives the full sink against a local HTTP +// 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, eventNames []string + attrStr := map[string]string{} + var statusAttr int64 + var bodyIsMap bool + 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) + for _, kv := range lr.Attributes { + if kv.Key == "http.response.status_code" { + statusAttr = kv.Value.GetIntValue() + } else { + attrStr[kv.Key] = kv.Value.GetStringValue() + } + } + if lr.Body.GetKvlistValue() != nil { + bodyIsMap = true + } + } + } + } + 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", + } + 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: "network_response", Category: Network, + Data: []byte(`{"method":"GET","url":"https://x","status":200}`)}}) + 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]) + // The excluded screenshot must not reach the receiver; only the network event does. + assert.Equal(t, []string{"network_response"}, eventNames) + // Promoted attributes and the structured body survive the SDK to protobuf translation. + assert.Equal(t, "https://x", attrStr["url.full"]) + assert.Equal(t, "GET", attrStr["http.request.method"]) + assert.Equal(t, int64(200), statusAttr) + assert.True(t, bodyIsMap, "structured body should arrive as a kvlist") +}