diff --git a/.changeset/quiet-span-context.md b/.changeset/quiet-span-context.md new file mode 100644 index 00000000..70daf005 --- /dev/null +++ b/.changeset/quiet-span-context.md @@ -0,0 +1,5 @@ +--- +"posthog-go": minor +--- + +Add an optional OpenTelemetry context processor and HTTP middleware for session and person IDs on application spans. Both take the PostHog client and add no IDs when that client is nil, disabled, or closed. diff --git a/noop_posthog.go b/noop_posthog.go index 1f4c5c0c..4383ca2b 100644 --- a/noop_posthog.go +++ b/noop_posthog.go @@ -61,6 +61,9 @@ func (c *noopClient) ReloadFeatureFlags() error { return ErrSDKDisabled } +// IsEnabled returns false because a client without a project API key sends no data. +func (c *noopClient) IsEnabled() bool { return false } + func (c *noopClient) Close() error { return ErrSDKDisabled } diff --git a/otel/README.md b/otel/README.md index d2c65094..3c382aaf 100644 --- a/otel/README.md +++ b/otel/README.md @@ -101,3 +101,17 @@ variable, not as span attributes. This bridge forwards spans only, so prompt and response fields stay empty for ADK Go generations. See [`example/`](example) for a runnable program. + +## General application spans + +Register `NewContextSpanProcessor(client)` on your existing OpenTelemetry tracer provider to add `sessionId` and `posthogDistinctId` to application spans. A nil, disabled (empty API key), or closed client adds no IDs; a custom `posthog.Client` wrapper counts as enabled unless it implements `IsEnabled() bool`. Place it before processors that read attributes in `OnStart`. Keep your existing exporter and sampler. + +Use `posthogotel.NewRequestContextMiddleware(client, handler, options...)` for HTTP requests. This wrapper uses the core SDK's header extraction and options. Place it inside the OpenTelemetry HTTP middleware so the request span already exists. It also works outside the OpenTelemetry middleware, because the processor reads context when that middleware starts the span. Pass `r.Context()` when you start child spans. + +The bridge adds only missing IDs and preserves explicit span attributes. It does not create IDs or export spans. Register your normal application exporter to send general traces to PostHog. The AI `SpanProcessor` and `Exporter` above keep their separate AI-only behavior. + +The HTTP wrapper stops new spans from using the request IDs after the handler returns or panics. Spans already started keep their IDs when they end later. Detached work that needs its own IDs must use a fresh context, such as `context.Background()` with `posthog.WithFreshRequestContext`. A context derived from an ended request remains expired. + +For manual operations, `posthog.WithRequestContext` supplies IDs to the processor. The processor cannot determine when a manual operation ends. Use a new context for each operation and do not reuse its context for unrelated work. + +OpenTelemetry remains optional through this separate module. This bridge adds no OpenTelemetry dependency to the core SDK. Non-recording spans and spans without readable public attributes receive no changes. Span attribute limits can omit new IDs while preserving existing attributes. diff --git a/otel/context_span_processor.go b/otel/context_span_processor.go new file mode 100644 index 00000000..722803c1 --- /dev/null +++ b/otel/context_span_processor.go @@ -0,0 +1,114 @@ +package posthogotel + +import ( + "context" + "net/http" + "sync/atomic" + + "github.com/posthog/posthog-go" + "go.opentelemetry.io/otel/attribute" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/trace" +) + +// ContextSpanProcessor adds PostHog IDs to spans from the application's provider. +// Register it before processors that consume attributes in OnStart. It does not export spans. +type ContextSpanProcessor struct { + client contextClient + shutdown atomic.Bool +} + +var _ sdktrace.SpanProcessor = (*ContextSpanProcessor)(nil) + +// NewContextSpanProcessor creates an optional bridge for general application spans. +// It adds IDs only while client is enabled, so a nil, disabled, or closed client adds none. +// A client wrapper can implement IsEnabled() bool to report its state; otherwise it counts as enabled. +func NewContextSpanProcessor(client posthog.Client) *ContextSpanProcessor { + return &ContextSpanProcessor{client: newContextClient(client)} +} + +func (p *ContextSpanProcessor) OnStart(ctx context.Context, span sdktrace.ReadWriteSpan) { + if !p.shutdown.Load() { + enrichContextSpan(ctx, span, p.client) + } +} + +func (*ContextSpanProcessor) OnEnd(sdktrace.ReadOnlySpan) {} +func (p *ContextSpanProcessor) Shutdown(context.Context) error { + p.shutdown.Store(true) + return nil +} +func (*ContextSpanProcessor) ForceFlush(context.Context) error { return nil } + +type contextClient struct { + present bool + enabled interface{ IsEnabled() bool } +} + +func newContextClient(client posthog.Client) contextClient { + enabled, _ := client.(interface{ IsEnabled() bool }) + return contextClient{present: client != nil, enabled: enabled} +} + +func (c contextClient) isEnabled() bool { + return c.present && (c.enabled == nil || c.enabled.IsEnabled()) +} + +type contextSpanRequestKey struct{} +type contextSpanRequest struct{ active atomic.Bool } + +func enrichContextSpan(ctx context.Context, span trace.Span, client contextClient) { + if !client.present || ctx == nil || span == nil || !span.IsRecording() { + return + } + if request, ok := ctx.Value(contextSpanRequestKey{}).(*contextSpanRequest); ok && !request.active.Load() { + return + } + if !client.isEnabled() { + return + } + readable, ok := span.(interface{ Attributes() []attribute.KeyValue }) + if !ok { + return + } + request, ok := posthog.RequestContextFromContext(ctx) + if !ok { + return + } + hasDistinctID, hasSessionID := false, false + for _, attr := range readable.Attributes() { + switch string(attr.Key) { + case "posthogDistinctId": + hasDistinctID = true + case "sessionId": + hasSessionID = true + } + } + attrs := make([]attribute.KeyValue, 0, 2) + if request.DistinctId != "" && !hasDistinctID { + attrs = append(attrs, attribute.String("posthogDistinctId", request.DistinctId)) + } + if request.SessionId != "" && !hasSessionID { + attrs = append(attrs, attribute.String("sessionId", request.SessionId)) + } + span.SetAttributes(attrs...) +} + +// NewRequestContextMiddleware also enriches the existing request span after header extraction. +// It adds IDs to spans only while client is enabled, like NewContextSpanProcessor. +// It accepts the core middleware options. Spans started from an inherited request context +// after the handler returns omit request IDs. Register ContextSpanProcessor for child spans. +func NewRequestContextMiddleware(client posthog.Client, next http.Handler, opts ...posthog.RequestContextOption) http.Handler { + spanClient := newContextClient(client) + if next == nil { + return posthog.NewRequestContextMiddleware(nil, opts...) + } + return posthog.NewRequestContextMiddleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + request := &contextSpanRequest{} + request.active.Store(true) + defer request.active.Store(false) + ctx := context.WithValue(r.Context(), contextSpanRequestKey{}, request) + enrichContextSpan(ctx, trace.SpanFromContext(ctx), spanClient) + next.ServeHTTP(w, r.WithContext(ctx)) + }), opts...) +} diff --git a/otel/context_span_processor_test.go b/otel/context_span_processor_test.go new file mode 100644 index 00000000..9cc3ec62 --- /dev/null +++ b/otel/context_span_processor_test.go @@ -0,0 +1,328 @@ +package posthogotel + +import ( + "context" + "fmt" + "net/http" + "net/http/httptest" + "sync" + "testing" + + "github.com/posthog/posthog-go" + "go.opentelemetry.io/otel/attribute" + sdktrace "go.opentelemetry.io/otel/sdk/trace" + "go.opentelemetry.io/otel/sdk/trace/tracetest" + "go.opentelemetry.io/otel/trace" +) + +func TestContextSpanProcessor(t *testing.T) { + cases := []struct { + name string + request posthog.RequestContext + own []attribute.KeyValue + limit int + want map[string]string + }{ + {"both IDs", posthog.RequestContext{DistinctId: "user", SessionId: "session"}, nil, 128, map[string]string{"posthogDistinctId": "user", "sessionId": "session"}}, + {"session only", posthog.RequestContext{SessionId: "session"}, nil, 128, map[string]string{"sessionId": "session"}}, + {"no IDs", posthog.RequestContext{}, nil, 128, map[string]string{}}, + {"explicit IDs", posthog.RequestContext{DistinctId: "user", SessionId: "session"}, []attribute.KeyValue{attribute.String("posthogDistinctId", "app-user"), attribute.String("sessionId", "app-session")}, 128, map[string]string{"posthogDistinctId": "app-user", "sessionId": "app-session"}}, + {"full attributes", posthog.RequestContext{DistinctId: "user", SessionId: "session"}, []attribute.KeyValue{attribute.String("app.key", "keep")}, 1, map[string]string{"app.key": "keep"}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + recorder := tracetest.NewSpanRecorder() + limits := sdktrace.NewSpanLimits() + limits.AttributeCountLimit = tc.limit + provider := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(NewContextSpanProcessor(newEnabledClient(t))), sdktrace.WithSpanProcessor(recorder), sdktrace.WithRawSpanLimits(limits)) + defer provider.Shutdown(context.Background()) + ctx := posthog.WithFreshRequestContext(context.Background(), tc.request) + _, span := provider.Tracer("test").Start(ctx, "operation", trace.WithAttributes(tc.own...)) + span.End() + assertContextAttributes(t, recorder.Ended()[0], tc.want) + }) + } +} + +func TestContextHTTPMiddleware(t *testing.T) { + for _, panics := range []bool{false, true} { + t.Run(fmt.Sprintf("panic=%t", panics), func(t *testing.T) { + recorder := tracetest.NewSpanRecorder() + provider := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(NewContextSpanProcessor(newEnabledClient(t))), sdktrace.WithSpanProcessor(recorder)) + defer provider.Shutdown(context.Background()) + tracer := provider.Tracer("test") + ctx, incoming := tracer.Start(context.Background(), "incoming") + var detached context.Context + var ongoing trace.Span + next := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + detached = context.WithValue(r.Context(), "test-key", "value") + _, ongoing = tracer.Start(r.Context(), "child") + if panics { + panic("failure") + } + w.WriteHeader(http.StatusAccepted) + }) + handler := NewRequestContextMiddleware(newEnabledClient(t), next) + request := httptest.NewRequest(http.MethodGet, "https://example.com/", nil).WithContext(ctx) + request.Header.Set(posthog.HeaderPostHogDistinctID, "user") + request.Header.Set(posthog.HeaderPostHogSessionID, "session") + response := httptest.NewRecorder() + func() { + defer func() { + if got := recover(); panics && got != "failure" { + t.Errorf("panic=%v", got) + } else if !panics && got != nil { + t.Errorf("unexpected panic=%v", got) + } + }() + handler.ServeHTTP(response, request) + }() + if !panics && response.Code != http.StatusAccepted { + t.Errorf("status=%d", response.Code) + } + _, late := tracer.Start(detached, "late") + late.End() + ongoing.End() + incoming.End() + for _, span := range recorder.Ended() { + want := map[string]string{"posthogDistinctId": "user", "sessionId": "session"} + if span.Name() == "late" { + want = map[string]string{} + } + assertContextAttributes(t, span, want) + } + fresh := posthog.WithFreshRequestContext(context.Background(), posthog.RequestContext{DistinctId: "job-user"}) + _, job := tracer.Start(fresh, "job") + job.End() + spans := recorder.Ended() + assertContextAttributes(t, spans[len(spans)-1], map[string]string{"posthogDistinctId": "job-user"}) + }) + } +} + +func TestContextHTTPMiddlewareOptions(t *testing.T) { + cases := []struct { + name, distinctID, sessionID string + disabled bool + want map[string]string + }{ + {"disabled", "user", "session", true, map[string]string{}}, + {"absent", "", "", false, map[string]string{}}, + {"invalid", "\x00", " ", false, map[string]string{}}, + {"sanitized", "bad\x00user", " session ", false, map[string]string{"posthogDistinctId": "baduser", "sessionId": "session"}}, + {"session only", "", "session", false, map[string]string{"sessionId": "session"}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + recorder := tracetest.NewSpanRecorder() + provider := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(NewContextSpanProcessor(newEnabledClient(t))), sdktrace.WithSpanProcessor(recorder)) + defer provider.Shutdown(context.Background()) + ctx, incoming := provider.Tracer("test").Start(context.Background(), "incoming") + request := httptest.NewRequest(http.MethodGet, "https://example.com/", nil).WithContext(ctx) + request.Header.Set(posthog.HeaderPostHogDistinctID, tc.distinctID) + request.Header.Set(posthog.HeaderPostHogSessionID, tc.sessionID) + NewRequestContextMiddleware(newEnabledClient(t), http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {}), posthog.WithCaptureTracingHeaders(!tc.disabled)).ServeHTTP(httptest.NewRecorder(), request) + incoming.End() + assertContextAttributes(t, recorder.Ended()[0], tc.want) + }) + } +} + +func TestContextHTTPMiddlewareConcurrentRequests(t *testing.T) { + recorder := tracetest.NewSpanRecorder() + provider := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(NewContextSpanProcessor(newEnabledClient(t))), sdktrace.WithSpanProcessor(recorder)) + defer provider.Shutdown(context.Background()) + tracer := provider.Tracer("test") + ready := make(chan struct{}, 2) + release := make(chan struct{}) + handler := NewRequestContextMiddleware(newEnabledClient(t), http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + ready <- struct{}{} + <-release + _, span := tracer.Start(r.Context(), r.Header.Get(posthog.HeaderPostHogDistinctID)) + span.End() + })) + var workers sync.WaitGroup + for _, id := range []string{"first", "second"} { + workers.Add(1) + go func(id string) { + defer workers.Done() + request := httptest.NewRequest(http.MethodGet, "https://example.com/", nil) + request.Header.Set(posthog.HeaderPostHogDistinctID, id) + request.Header.Set(posthog.HeaderPostHogSessionID, id+"-session") + handler.ServeHTTP(httptest.NewRecorder(), request) + }(id) + } + <-ready + <-ready + close(release) + workers.Wait() + spans := recorder.Ended() + if len(spans) != 2 { + t.Fatalf("recorded spans=%d, want 2", len(spans)) + } + for _, span := range spans { + assertContextAttributes(t, span, map[string]string{"posthogDistinctId": span.Name(), "sessionId": span.Name() + "-session"}) + } +} + +func assertContextAttributes(t *testing.T, span sdktrace.ReadOnlySpan, want map[string]string) { + t.Helper() + got := make(map[string]string) + for _, attr := range span.Attributes() { + got[string(attr.Key)] = attr.Value.AsString() + } + if len(got) != len(want) { + t.Fatalf("span %s attributes=%v, want %v", span.Name(), got, want) + } + for key, value := range want { + if got[key] != value { + t.Errorf("span %s %s=%q, want %q", span.Name(), key, got[key], value) + } + } +} + +func TestContextHTTPMiddlewareWithoutRecording(t *testing.T) { + for _, sampledOut := range []bool{false, true} { + t.Run(fmt.Sprintf("sampledOut=%t", sampledOut), func(t *testing.T) { + ctx := context.Background() + if sampledOut { + provider := sdktrace.NewTracerProvider(sdktrace.WithSampler(sdktrace.NeverSample()), sdktrace.WithSpanProcessor(NewContextSpanProcessor(newEnabledClient(t)))) + defer provider.Shutdown(context.Background()) + var span trace.Span + ctx, span = provider.Tracer("test").Start(ctx, "unsampled") + defer span.End() + } + request := httptest.NewRequest(http.MethodGet, "https://example.com/", nil).WithContext(ctx) + request.Header.Set(posthog.HeaderPostHogSessionID, "session") + response := httptest.NewRecorder() + NewRequestContextMiddleware(newEnabledClient(t), http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + metadata, ok := posthog.RequestContextFromContext(r.Context()) + if !ok || metadata.SessionId != "session" { + t.Errorf("event context=%+v", metadata) + } + w.WriteHeader(http.StatusNoContent) + })).ServeHTTP(response, request) + if response.Code != http.StatusNoContent { + t.Errorf("status=%d", response.Code) + } + }) + } +} + +type contextOpaqueSpan struct{ trace.Span } + +func TestContextHTTPMiddlewareKeepsUnreadableSpan(t *testing.T) { + recorder := tracetest.NewSpanRecorder() + provider := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(recorder)) + defer provider.Shutdown(context.Background()) + ctx, span := provider.Tracer("test").Start(context.Background(), "opaque", trace.WithAttributes(attribute.String("sessionId", "application-session"))) + ctx = trace.ContextWithSpan(ctx, contextOpaqueSpan{span}) + request := httptest.NewRequest(http.MethodGet, "https://example.com/", nil).WithContext(ctx) + request.Header.Set(posthog.HeaderPostHogSessionID, "header-session") + NewRequestContextMiddleware(newEnabledClient(t), http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})).ServeHTTP(httptest.NewRecorder(), request) + span.End() + assertContextAttributes(t, recorder.Ended()[0], map[string]string{"sessionId": "application-session"}) +} + +func TestContextSpanProcessorShutdown(t *testing.T) { + for _, calls := range []int{1, 2} { + t.Run(fmt.Sprintf("shutdownCalls=%d", calls), func(t *testing.T) { + processor := NewContextSpanProcessor(newEnabledClient(t)) + recorder := tracetest.NewSpanRecorder() + provider := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(processor), sdktrace.WithSpanProcessor(recorder)) + defer provider.Shutdown(context.Background()) + ctx := posthog.WithFreshRequestContext(context.Background(), posthog.RequestContext{DistinctId: "user", SessionId: "session"}) + tracer := provider.Tracer("test") + _, before := tracer.Start(ctx, "before") + for range calls { + if err := processor.Shutdown(context.Background()); err != nil { + t.Fatal(err) + } + } + _, after := tracer.Start(ctx, "after") + after.End() + before.End() + spans := recorder.Ended() + if len(spans) != 2 { + t.Fatalf("recorded spans=%d, want 2", len(spans)) + } + assertContextAttributes(t, spans[0], map[string]string{}) + assertContextAttributes(t, spans[1], map[string]string{"posthogDistinctId": "user", "sessionId": "session"}) + }) + } +} + +type contextClientWrapper struct{ posthog.Client } + +func newEnabledClient(t *testing.T) posthog.Client { + t.Helper() + client := posthog.New("test-key") + t.Cleanup(func() { _ = client.Close() }) + return client +} + +func TestContextSpansFollowClientState(t *testing.T) { + cases := []struct { + name string + client func(*testing.T) posthog.Client + want map[string]string + }{ + {"missing client", func(*testing.T) posthog.Client { return nil }, map[string]string{}}, + {"empty API key", func(*testing.T) posthog.Client { return posthog.New(" ") }, map[string]string{}}, + {"closed client", func(t *testing.T) posthog.Client { + client := posthog.New("test-key") + if err := client.Close(); err != nil { + t.Fatal(err) + } + return client + }, map[string]string{}}, + {"enabled client", newEnabledClient, map[string]string{"posthogDistinctId": "user", "sessionId": "session"}}, + {"wrapper without IsEnabled", func(*testing.T) posthog.Client { return contextClientWrapper{posthog.New("")} }, map[string]string{"posthogDistinctId": "user", "sessionId": "session"}}, + } + for _, tc := range cases { + t.Run(tc.name+"/processor", func(t *testing.T) { + recorder := tracetest.NewSpanRecorder() + provider := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(NewContextSpanProcessor(tc.client(t))), sdktrace.WithSpanProcessor(recorder)) + defer provider.Shutdown(context.Background()) + ctx := posthog.WithFreshRequestContext(context.Background(), posthog.RequestContext{DistinctId: "user", SessionId: "session"}) + _, span := provider.Tracer("test").Start(ctx, "operation") + span.End() + assertContextAttributes(t, recorder.Ended()[0], tc.want) + }) + t.Run(tc.name+"/middleware", func(t *testing.T) { + recorder := tracetest.NewSpanRecorder() + provider := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(recorder)) + defer provider.Shutdown(context.Background()) + ctx, incoming := provider.Tracer("test").Start(context.Background(), "incoming") + request := httptest.NewRequest(http.MethodGet, "https://example.com/", nil).WithContext(ctx) + request.Header.Set(posthog.HeaderPostHogDistinctID, "user") + request.Header.Set(posthog.HeaderPostHogSessionID, "session") + NewRequestContextMiddleware(tc.client(t), http.HandlerFunc(func(http.ResponseWriter, *http.Request) {})).ServeHTTP(httptest.NewRecorder(), request) + incoming.End() + assertContextAttributes(t, recorder.Ended()[0], tc.want) + }) + } +} + +func TestContextSpanProcessorClientsAreIndependent(t *testing.T) { + closed, open := posthog.New("first-key"), newEnabledClient(t) + if err := closed.Close(); err != nil { + t.Fatal(err) + } + ctx := posthog.WithFreshRequestContext(context.Background(), posthog.RequestContext{DistinctId: "user", SessionId: "session"}) + for _, tc := range []struct { + name string + client posthog.Client + want map[string]string + }{ + {"closed", closed, map[string]string{}}, + {"open", open, map[string]string{"posthogDistinctId": "user", "sessionId": "session"}}, + } { + recorder := tracetest.NewSpanRecorder() + provider := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(NewContextSpanProcessor(tc.client)), sdktrace.WithSpanProcessor(recorder)) + _, span := provider.Tracer("test").Start(ctx, tc.name) + span.End() + _ = provider.Shutdown(context.Background()) + assertContextAttributes(t, recorder.Ended()[0], tc.want) + } +} diff --git a/otel/doc.go b/otel/doc.go index a1540afe..6d5abb50 100644 --- a/otel/doc.go +++ b/otel/doc.go @@ -18,6 +18,12 @@ // frameworks that accept only a span exporter. It filters spans and // delegates the AI spans to an OTLP/HTTP exporter. // +// ContextSpanProcessor adds IDs to general application spans without exporting +// them. NewRequestContextMiddleware also enriches the existing incoming span and +// limits request IDs to the handler lifetime. These integrations use the +// application's provider and exporter, and add no IDs when the PostHog client +// is nil, disabled, or closed. +// // This is a separate Go module so that the core posthog-go SDK does not depend // on OpenTelemetry. This module depends on the core module for the request // context. diff --git a/posthog.go b/posthog.go index abbac4bc..35020174 100644 --- a/posthog.go +++ b/posthog.go @@ -1442,6 +1442,9 @@ func (c *client) featureFlagEvaluationsHostWithContext(ctx context.Context) feat } } +// IsEnabled reports whether the client can still send data. It returns false after Close. +func (c *client) IsEnabled() bool { return !c.closed.Load() } + // Close gracefully shuts down the client, flushing any pending messages. // If ShutdownTimeout is set to a positive duration, Close waits up to that // duration for in-flight requests to complete. Otherwise, it waits indefinitely.