From f510718d46cfb4cafacbb0de54e90b7ce3c7b45d Mon Sep 17 00:00:00 2001 From: Matt Wisner Date: Mon, 10 Aug 2026 12:51:56 -0400 Subject: [PATCH] feat(router): observe subscription delivery outcomes --- router/core/graphql_handler.go | 36 +++++++- .../graphql_handler_observability_test.go | 26 ++++++ router/core/websocket.go | 62 ++++++++++++- router/core/websocket_observability_test.go | 86 +++++++++++++++++ router/pkg/metric/engine_metrics.go | 92 ++++++++++++++----- router/pkg/metric/noop_stream_metrics.go | 1 + router/pkg/metric/oltp_stream_metric_store.go | 4 + router/pkg/metric/prom_stream_metric_store.go | 4 + router/pkg/metric/stream_measurements.go | 26 ++++-- router/pkg/metric/stream_metric_store.go | 36 ++++++++ router/pkg/otel/attributes.go | 19 +++- router/pkg/pubsub/datasource/mocks.go | 22 +++-- .../datasource/subscription_event_updater.go | 44 ++++++--- ...event_updater_beforeeventsdispatch_test.go | 29 +++++- router/pkg/pubsub/kafka/adapter.go | 35 ++++++- router/pkg/pubsub/nats/adapter.go | 21 ++++- router/pkg/pubsub/redis/adapter.go | 18 +++- router/pkg/pubsub/redis/adapter_test.go | 10 +- router/pkg/statistics/engine_stats.go | 86 +++++++++++++++-- router/pkg/statistics/engine_stats_test.go | 21 +++++ 20 files changed, 601 insertions(+), 77 deletions(-) create mode 100644 router/core/graphql_handler_observability_test.go create mode 100644 router/core/websocket_observability_test.go create mode 100644 router/pkg/statistics/engine_stats_test.go diff --git a/router/core/graphql_handler.go b/router/core/graphql_handler.go index 4ef92da46b..bbf5cc54ab 100644 --- a/router/core/graphql_handler.go +++ b/router/core/graphql_handler.go @@ -477,6 +477,17 @@ func (h *GraphQLHandler) writeError(ctx *resolve.Context, err error, res *resolv } requestLogger := reqContext.logger + errorKind := getErrorType(err) + observer, observesSubscriptions := h.engineStats.(statistics.SubscriptionObserver) + if _, isSubscription := w.(resolve.SubscriptionResponseWriter); isSubscription && observesSubscriptions { + reason := subscriptionResolutionErrorReason(errorKind) + observer.SubscriptionResolutionError(reason) + requestLogger.Warn("Subscription resolution failed", + zap.String("reason", string(reason)), + zap.Bool("terminal", terminal || isTerminalSubscriptionError(err)), + zap.Error(err), + ) + } httpWriter, isHttpResponseWriter := w.(http.ResponseWriter) response := GraphQLErrorResponse{ @@ -484,7 +495,7 @@ func (h *GraphQLHandler) writeError(ctx *resolve.Context, err error, res *resolv Data: nil, } - switch getErrorType(err) { + switch errorKind { case errorTypeMergeResult: var errMerge resolve.ErrMergeResult if !errors.As(err, &errMerge) { @@ -638,6 +649,29 @@ func (h *GraphQLHandler) writeError(ctx *resolve.Context, err error, res *resolv } } +func subscriptionResolutionErrorReason(kind errorType) statistics.SubscriptionResolutionErrorReason { + switch kind { + case errorTypeRateLimit: + return statistics.SubscriptionResolutionErrorRateLimit + case errorTypeUnauthorized: + return statistics.SubscriptionResolutionErrorAuthorization + case errorTypeContextCanceled: + return statistics.SubscriptionResolutionErrorContextCanceled + case errorTypeContextTimeout: + return statistics.SubscriptionResolutionErrorTimeout + case errorTypeUpgradeFailed, errorTypeEDFS: + return statistics.SubscriptionResolutionErrorFetch + case errorTypeStreamsHandlerError: + return statistics.SubscriptionResolutionErrorHandler + case errorTypeEDFSInvalidMessage, errorTypeInvalidWsSubprotocol: + return statistics.SubscriptionResolutionErrorInvalidMessage + case errorTypeMergeResult: + return statistics.SubscriptionResolutionErrorResolve + default: + return statistics.SubscriptionResolutionErrorUnknown + } +} + func (h *GraphQLHandler) setDebugCacheHeaders(w http.ResponseWriter, opCtx *operationContext) { if h.enableCacheResponseHeaders { if opCtx.normalizationCacheHit { diff --git a/router/core/graphql_handler_observability_test.go b/router/core/graphql_handler_observability_test.go new file mode 100644 index 0000000000..662e3712a2 --- /dev/null +++ b/router/core/graphql_handler_observability_test.go @@ -0,0 +1,26 @@ +package core + +import ( + "testing" + + "github.com/stretchr/testify/require" + "github.com/wundergraph/cosmo/router/pkg/statistics" +) + +func TestSubscriptionResolutionErrorReason(t *testing.T) { + t.Parallel() + tests := map[errorType]statistics.SubscriptionResolutionErrorReason{ + errorTypeUnauthorized: statistics.SubscriptionResolutionErrorAuthorization, + errorTypeContextCanceled: statistics.SubscriptionResolutionErrorContextCanceled, + errorTypeContextTimeout: statistics.SubscriptionResolutionErrorTimeout, + errorTypeUpgradeFailed: statistics.SubscriptionResolutionErrorFetch, + errorTypeEDFS: statistics.SubscriptionResolutionErrorFetch, + errorTypeStreamsHandlerError: statistics.SubscriptionResolutionErrorHandler, + errorTypeEDFSInvalidMessage: statistics.SubscriptionResolutionErrorInvalidMessage, + errorTypeMergeResult: statistics.SubscriptionResolutionErrorResolve, + errorTypeUnknown: statistics.SubscriptionResolutionErrorUnknown, + } + for kind, want := range tests { + require.Equal(t, want, subscriptionResolutionErrorReason(kind)) + } +} diff --git a/router/core/websocket.go b/router/core/websocket.go index 3fe6ec3345..f7082bc9c8 100644 --- a/router/core/websocket.go +++ b/router/core/websocket.go @@ -669,8 +669,12 @@ func (rw *websocketResponseWriter) Complete() { } err := rw.protocol.Complete(rw.id) if err != nil { - rw.logger.Debug("Sending complete message", zap.Error(err)) + reason := webSocketWriteFailureReason(err) + rw.observeFrame("complete", "none", "failure", reason) + rw.logger.Warn("Sending WebSocket complete message failed", zap.String("reason", reason), zap.Error(err)) + return } + rw.observeFrame("complete", "none", "success", "none") } // Heartbeat is a no-op function for WebSocket subscriptions. @@ -700,15 +704,22 @@ func (rw *websocketResponseWriter) Error(data []byte) { errors = json.RawMessage(`[{"message":"Unable to subscribe"}]`) } if err := rw.protocol.WriteGraphQLErrors(rw.id, errors, nil); err != nil { - rw.logger.Debug("Sending error message", zap.Error(err)) + reason := webSocketWriteFailureReason(err) + rw.observeFrame("terminal_error", "errors_only", "failure", reason) + rw.logger.Warn("Sending terminal GraphQL error frame failed", zap.String("reason", reason), zap.Error(err)) return } + rw.observeFrame("terminal_error", "errors_only", "success", "none") // subscriptions-transport-ws clients rely on an explicit "complete" to end // the stream after a data+errors frame. graphql-transport-ws treats the // "error" frame as terminal per spec, so no follow-up is needed there. if rw.protocol.Subprotocol() == wsproto.SubscriptionsTransportWSSubprotocol { if err := rw.protocol.Complete(rw.id); err != nil { - rw.logger.Debug("Sending complete after error", zap.Error(err)) + reason := webSocketWriteFailureReason(err) + rw.observeFrame("complete", "none", "failure", reason) + rw.logger.Warn("Sending WebSocket complete after error failed", zap.String("reason", reason), zap.Error(err)) + } else { + rw.observeFrame("complete", "none", "success", "none") } } } @@ -728,6 +739,7 @@ func (rw *websocketResponseWriter) Flush() error { "response_headers": rw.header, }) if err != nil { + rw.observeFrame("data", webSocketPayloadType(payload), "failure", "serialization_error") rw.logger.Warn("Serializing response headers", zap.Error(err)) return err } @@ -746,12 +758,56 @@ func (rw *websocketResponseWriter) Flush() error { err = rw.protocol.WriteGraphQLData(rw.id, payload, extensions) rw.buf.Reset() if err != nil { + reason := webSocketWriteFailureReason(err) + rw.observeFrame("data", webSocketPayloadType(payload), "failure", reason) + rw.logger.Warn("Sending GraphQL data frame failed", zap.String("reason", reason), zap.Error(err)) return err } + rw.observeFrame("data", webSocketPayloadType(payload), "success", "none") } return nil } +func (rw *websocketResponseWriter) observeFrame(frameType, payloadType, result, reason string) { + observer, ok := rw.stats.(statistics.SubscriptionObserver) + if !ok { + return + } + observer.WebSocketFrame(statistics.WebSocketFrameObservation{ + FrameType: frameType, + PayloadType: payloadType, + Result: result, + Reason: reason, + }) +} + +func webSocketPayloadType(payload []byte) string { + data := gjson.GetBytes(payload, "data") + hasData := data.Exists() && data.Type != gjson.Null + hasErrors := gjson.GetBytes(payload, "errors").Type == gjson.JSON + switch { + case hasData && hasErrors: + return "data_with_errors" + case hasErrors: + return "errors_only" + case hasData: + return "data" + default: + return "none" + } +} + +func webSocketWriteFailureReason(err error) string { + switch { + case errors.Is(err, context.Canceled): + return "context_canceled" + case errors.Is(err, net.ErrClosed), errors.Is(err, syscall.EPIPE), errors.Is(err, syscall.ECONNRESET): + return "client_disconnected" + default: + return "protocol_error" + } +} + func (rw *websocketResponseWriter) SubscriptionResponseWriter() resolve.SubscriptionResponseWriter { return rw } diff --git a/router/core/websocket_observability_test.go b/router/core/websocket_observability_test.go new file mode 100644 index 0000000000..e8c10c1b25 --- /dev/null +++ b/router/core/websocket_observability_test.go @@ -0,0 +1,86 @@ +package core + +import ( + "context" + "encoding/json" + "errors" + "testing" + + "github.com/stretchr/testify/require" + "github.com/wundergraph/cosmo/router/internal/wsproto" + "github.com/wundergraph/cosmo/router/pkg/statistics" + "go.uber.org/zap" +) + +type observabilityTestProtocol struct { + dataErr error + errorErr error + completeErr error +} + +func (p *observabilityTestProtocol) Subprotocol() string { return wsproto.GraphQLWSSubprotocol } +func (p *observabilityTestProtocol) Initialize() (json.RawMessage, error) { return nil, nil } +func (p *observabilityTestProtocol) ReadMessage() (*wsproto.Message, error) { return nil, nil } +func (p *observabilityTestProtocol) Pong(*wsproto.Message) error { return nil } +func (p *observabilityTestProtocol) WriteGraphQLData(string, json.RawMessage, json.RawMessage) error { + return p.dataErr +} +func (p *observabilityTestProtocol) WriteGraphQLErrors(string, json.RawMessage, json.RawMessage) error { + return p.errorErr +} +func (p *observabilityTestProtocol) Complete(string) error { return p.completeErr } + +func TestWebsocketResponseWriterObservesFrameOutcomes(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + payload string + protocolErr error + wantResult string + wantPayload string + }{ + {name: "data success", payload: `{"data":{"value":1}}`, wantResult: "success", wantPayload: "data"}, + {name: "data with errors success", payload: `{"data":{"value":1},"errors":[{"message":"partial"}]}`, wantResult: "success", wantPayload: "data_with_errors"}, + {name: "errors only success", payload: `{"errors":[{"message":"failed"}],"data":null}`, wantResult: "success", wantPayload: "errors_only"}, + {name: "write failure", payload: `{"data":{"value":1}}`, protocolErr: errors.New("write failed"), wantResult: "failure", wantPayload: "data"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + stats := statistics.NewEngineStats(t.Context(), zap.NewNop(), false) + writer := newWebsocketResponseWriter("1", &observabilityTestProtocol{dataErr: tt.protocolErr}, true, zap.NewNop(), stats, nil) + _, err := writer.Write([]byte(tt.payload)) + require.NoError(t, err) + flushErr := writer.Flush() + if tt.protocolErr == nil { + require.NoError(t, flushErr) + } else { + require.ErrorIs(t, flushErr, tt.protocolErr) + } + + report := stats.GetReport() + require.Len(t, report.WebSocketFrames, 1) + require.Equal(t, statistics.WebSocketFrameObservation{ + FrameType: "data", PayloadType: tt.wantPayload, Result: tt.wantResult, + Reason: map[bool]string{true: "protocol_error", false: "none"}[tt.protocolErr != nil], + }, report.WebSocketFrames[0].Observation) + require.Equal(t, uint64(1), report.WebSocketFrames[0].Count) + }) + } +} + +func TestWebsocketResponseWriterObservesTerminalErrorWriteFailure(t *testing.T) { + t.Parallel() + stats := statistics.NewEngineStats(context.Background(), zap.NewNop(), false) + writer := newWebsocketResponseWriter("1", &observabilityTestProtocol{errorErr: errors.New("client gone")}, true, zap.NewNop(), stats, nil) + + writer.Error([]byte(`{"errors":[{"message":"timeout"}]}`)) + + report := stats.GetReport() + require.Len(t, report.WebSocketFrames, 1) + require.Equal(t, statistics.WebSocketFrameObservation{ + FrameType: "terminal_error", PayloadType: "errors_only", Result: "failure", Reason: "protocol_error", + }, report.WebSocketFrames[0].Observation) +} diff --git a/router/pkg/metric/engine_metrics.go b/router/pkg/metric/engine_metrics.go index 204331c8ef..00f307d096 100644 --- a/router/pkg/metric/engine_metrics.go +++ b/router/pkg/metric/engine_metrics.go @@ -5,6 +5,7 @@ import ( "errors" "fmt" + rotel "github.com/wundergraph/cosmo/router/pkg/otel" "github.com/wundergraph/cosmo/router/pkg/statistics" "go.opentelemetry.io/otel/attribute" otelmetric "go.opentelemetry.io/otel/metric" @@ -16,22 +17,26 @@ const ( cosmoEngineMeterName = "cosmo.router.engine" cosmoEngineMeterVersion = "0.0.1" - engineMetricBaseKey = "router.engine." - engineConnectionCountKey = engineMetricBaseKey + "connections" - engineSubscriptionCountKey = engineMetricBaseKey + "subscriptions" - engineTriggerCountKey = engineMetricBaseKey + "triggers" - engineMessagesSentKey = engineMetricBaseKey + "messages.sent" - engineResolversMaxConcurrentKey = engineMetricBaseKey + "resolvers.max_concurrent" - engineResolversInflightKey = engineMetricBaseKey + "resolvers.inflight" + engineMetricBaseKey = "router.engine." + engineConnectionCountKey = engineMetricBaseKey + "connections" + engineSubscriptionCountKey = engineMetricBaseKey + "subscriptions" + engineTriggerCountKey = engineMetricBaseKey + "triggers" + engineMessagesSentKey = engineMetricBaseKey + "messages.sent" + engineSubscriptionResolutionErrorsKey = engineMetricBaseKey + "subscription.resolution.errors" + engineWebSocketFramesKey = engineMetricBaseKey + "websocket.frames" + engineResolversMaxConcurrentKey = engineMetricBaseKey + "resolvers.max_concurrent" + engineResolversInflightKey = engineMetricBaseKey + "resolvers.inflight" ) type engineInstruments struct { - connectionCount otelmetric.Int64ObservableUpDownCounter - subscriptionCount otelmetric.Int64ObservableUpDownCounter - triggerCount otelmetric.Int64ObservableUpDownCounter - messagesSent otelmetric.Int64ObservableCounter - resolversMaxConcurrent otelmetric.Int64ObservableUpDownCounter - resolversInflight otelmetric.Int64ObservableUpDownCounter + connectionCount otelmetric.Int64ObservableUpDownCounter + subscriptionCount otelmetric.Int64ObservableUpDownCounter + triggerCount otelmetric.Int64ObservableUpDownCounter + messagesSent otelmetric.Int64ObservableCounter + subscriptionResolutionErrors otelmetric.Int64ObservableCounter + webSocketFrames otelmetric.Int64ObservableCounter + resolversMaxConcurrent otelmetric.Int64ObservableUpDownCounter + resolversInflight otelmetric.Int64ObservableUpDownCounter } func (i *engineInstruments) toList() []otelmetric.Observable { @@ -52,6 +57,12 @@ func (i *engineInstruments) toList() []otelmetric.Observable { if i.messagesSent != nil { result = append(result, i.messagesSent) } + if i.subscriptionResolutionErrors != nil { + result = append(result, i.subscriptionResolutionErrors) + } + if i.webSocketFrames != nil { + result = append(result, i.webSocketFrames) + } if i.resolversMaxConcurrent != nil { result = append(result, i.resolversMaxConcurrent) @@ -111,12 +122,14 @@ func setupInstruments(m otelmetric.Meter, statConfig *EngineStatsConfig, resolve var ( err error - connectionCount otelmetric.Int64ObservableUpDownCounter - subscriptionCount otelmetric.Int64ObservableUpDownCounter - triggerCount otelmetric.Int64ObservableUpDownCounter - messagesSent otelmetric.Int64ObservableCounter - resolversMaxConcurrent otelmetric.Int64ObservableUpDownCounter - resolversInflight otelmetric.Int64ObservableUpDownCounter + connectionCount otelmetric.Int64ObservableUpDownCounter + subscriptionCount otelmetric.Int64ObservableUpDownCounter + triggerCount otelmetric.Int64ObservableUpDownCounter + messagesSent otelmetric.Int64ObservableCounter + subscriptionResolutionErrors otelmetric.Int64ObservableCounter + webSocketFrames otelmetric.Int64ObservableCounter + resolversMaxConcurrent otelmetric.Int64ObservableUpDownCounter + resolversInflight otelmetric.Int64ObservableUpDownCounter ) if statConfig.Subscription { @@ -144,6 +157,18 @@ func setupInstruments(m otelmetric.Meter, statConfig *EngineStatsConfig, resolve if err != nil { return nil, err } + + subscriptionResolutionErrors, err = m.Int64ObservableCounter(engineSubscriptionResolutionErrorsKey, + otelmetric.WithDescription("Number of subscription resolution errors by reason.")) + if err != nil { + return nil, err + } + + webSocketFrames, err = m.Int64ObservableCounter(engineWebSocketFramesKey, + otelmetric.WithDescription("Number of subscription WebSocket frame write outcomes.")) + if err != nil { + return nil, err + } } if resolverStats { @@ -161,12 +186,14 @@ func setupInstruments(m otelmetric.Meter, statConfig *EngineStatsConfig, resolve } return &engineInstruments{ - connectionCount: connectionCount, - subscriptionCount: subscriptionCount, - triggerCount: triggerCount, - messagesSent: messagesSent, - resolversMaxConcurrent: resolversMaxConcurrent, - resolversInflight: resolversInflight, + connectionCount: connectionCount, + subscriptionCount: subscriptionCount, + triggerCount: triggerCount, + messagesSent: messagesSent, + subscriptionResolutionErrors: subscriptionResolutionErrors, + webSocketFrames: webSocketFrames, + resolversMaxConcurrent: resolversMaxConcurrent, + resolversInflight: resolversInflight, }, nil } @@ -200,6 +227,21 @@ func (e *EngineMetrics) observeInstruments(o otelmetric.Observer, stats statisti o.ObserveInt64(e.instruments.subscriptionCount, int64(report.Subscriptions), otelmetric.WithAttributes(e.baseAttributes...)) o.ObserveInt64(e.instruments.triggerCount, int64(report.Triggers), otelmetric.WithAttributes(e.baseAttributes...)) o.ObserveInt64(e.instruments.messagesSent, int64(report.MessagesSent), otelmetric.WithAttributes(e.baseAttributes...)) + for _, item := range report.SubscriptionResolutionErrors { + attrs := append([]attribute.KeyValue{}, e.baseAttributes...) + attrs = append(attrs, rotel.WgSubscriptionReason.String(string(item.Reason))) + o.ObserveInt64(e.instruments.subscriptionResolutionErrors, int64(item.Count), otelmetric.WithAttributes(attrs...)) + } + for _, item := range report.WebSocketFrames { + attrs := append([]attribute.KeyValue{}, e.baseAttributes...) + attrs = append(attrs, + rotel.WgSubscriptionFrameType.String(item.Observation.FrameType), + rotel.WgSubscriptionPayloadType.String(item.Observation.PayloadType), + rotel.WgSubscriptionResult.String(item.Observation.Result), + rotel.WgSubscriptionReason.String(item.Observation.Reason), + ) + o.ObserveInt64(e.instruments.webSocketFrames, int64(item.Count), otelmetric.WithAttributes(attrs...)) + } } if e.instruments.resolversMaxConcurrent != nil { diff --git a/router/pkg/metric/noop_stream_metrics.go b/router/pkg/metric/noop_stream_metrics.go index c312cc2472..c6ea52b437 100644 --- a/router/pkg/metric/noop_stream_metrics.go +++ b/router/pkg/metric/noop_stream_metrics.go @@ -8,6 +8,7 @@ type NoopStreamMetricStore struct{} func (n *NoopStreamMetricStore) Produce(ctx context.Context, event StreamsEvent) {} func (n *NoopStreamMetricStore) Consume(ctx context.Context, event StreamsEvent) {} +func (n *NoopStreamMetricStore) Process(ctx context.Context, event StreamsEvent) {} func (n *NoopStreamMetricStore) Flush(ctx context.Context) error { return nil } func (n *NoopStreamMetricStore) Shutdown(ctx context.Context) error { return nil } diff --git a/router/pkg/metric/oltp_stream_metric_store.go b/router/pkg/metric/oltp_stream_metric_store.go index 8d30c15364..e192c2c747 100644 --- a/router/pkg/metric/oltp_stream_metric_store.go +++ b/router/pkg/metric/oltp_stream_metric_store.go @@ -46,3 +46,7 @@ func (o *otlpStreamEventMetrics) Produce(ctx context.Context, opts ...otelmetric func (o *otlpStreamEventMetrics) Consume(ctx context.Context, opts ...otelmetric.AddOption) { o.instruments.consumedMessages.Add(ctx, 1, opts...) } + +func (o *otlpStreamEventMetrics) Process(ctx context.Context, count int64, opts ...otelmetric.AddOption) { + o.instruments.processedMessages.Add(ctx, count, opts...) +} diff --git a/router/pkg/metric/prom_stream_metric_store.go b/router/pkg/metric/prom_stream_metric_store.go index 30309f2444..70727c9068 100644 --- a/router/pkg/metric/prom_stream_metric_store.go +++ b/router/pkg/metric/prom_stream_metric_store.go @@ -47,6 +47,10 @@ func (p *promStreamEventMetrics) Consume(ctx context.Context, opts ...otelmetric p.instruments.consumedMessages.Add(ctx, 1, opts...) } +func (p *promStreamEventMetrics) Process(ctx context.Context, count int64, opts ...otelmetric.AddOption) { + p.instruments.processedMessages.Add(ctx, count, opts...) +} + func (p *promStreamEventMetrics) Flush(ctx context.Context) error { return p.meterProvider.ForceFlush(ctx) } diff --git a/router/pkg/metric/stream_measurements.go b/router/pkg/metric/stream_measurements.go index a5e1dadfb1..06400ad31d 100644 --- a/router/pkg/metric/stream_measurements.go +++ b/router/pkg/metric/stream_measurements.go @@ -7,8 +7,9 @@ import ( ) const ( - messagingSentMessages = "router.streams.sent.messages" - messagingConsumedMessages = "router.streams.received.messages" + messagingSentMessages = "router.streams.sent.messages" + messagingConsumedMessages = "router.streams.received.messages" + messagingProcessedMessages = "router.streams.processed.messages" ) var ( @@ -18,11 +19,15 @@ var ( messagingConsumedMessagesOptions = []otelmetric.Int64CounterOption{ otelmetric.WithDescription("Number of stream consumed messages"), } + messagingProcessedMessagesOptions = []otelmetric.Int64CounterOption{ + otelmetric.WithDescription("Number of stream messages dispatched or dropped before subscription fan-out"), + } ) type eventInstruments struct { - producedMessages otelmetric.Int64Counter - consumedMessages otelmetric.Int64Counter + producedMessages otelmetric.Int64Counter + consumedMessages otelmetric.Int64Counter + processedMessages otelmetric.Int64Counter } func newStreamEventInstruments(meter otelmetric.Meter) (*eventInstruments, error) { @@ -42,8 +47,17 @@ func newStreamEventInstruments(meter otelmetric.Meter) (*eventInstruments, error return nil, fmt.Errorf("failed to create received messages counter: %w", err) } + processedCounter, err := meter.Int64Counter( + messagingProcessedMessages, + messagingProcessedMessagesOptions..., + ) + if err != nil { + return nil, fmt.Errorf("failed to create processed messages counter: %w", err) + } + return &eventInstruments{ - producedMessages: producedCounter, - consumedMessages: consumedCounter, + producedMessages: producedCounter, + consumedMessages: consumedCounter, + processedMessages: processedCounter, }, nil } diff --git a/router/pkg/metric/stream_metric_store.go b/router/pkg/metric/stream_metric_store.go index 361f49388d..63779d0669 100644 --- a/router/pkg/metric/stream_metric_store.go +++ b/router/pkg/metric/stream_metric_store.go @@ -27,17 +27,23 @@ type StreamsEvent struct { ProviderType ProviderType // The messaging system type that are supported ErrorType string // Optional error type, e.g., "publish_error" or "receive_error". If empty, the attribute is not set DestinationName string // The name of the destination queue / topic / channel + RootFieldName string // The GraphQL subscription root field + Result string // Optional processing result, e.g. "dispatched" or "dropped" + Reason string // Optional processing reason + Count int64 // Optional event count; defaults to one } // StreamMetricProvider is the interface that wraps the basic Event metric methods. type StreamMetricProvider interface { Produce(ctx context.Context, opts ...otelmetric.AddOption) Consume(ctx context.Context, opts ...otelmetric.AddOption) + Process(ctx context.Context, count int64, opts ...otelmetric.AddOption) } type StreamMetricStore interface { Produce(ctx context.Context, event StreamsEvent) Consume(ctx context.Context, event StreamsEvent) + Process(ctx context.Context, event StreamsEvent) } // StreamMetrics is the store for Event (Kafka/Redis/NATS) metrics. @@ -121,3 +127,33 @@ func (e *StreamMetrics) Consume(ctx context.Context, event StreamsEvent) { provider.Consume(ctx, opt) } } + +func (e *StreamMetrics) Process(ctx context.Context, event StreamsEvent) { + attrs := []attribute.KeyValue{ + otel.WgStreamOperationName.String(event.StreamOperationName), + otel.WgProviderType.String(string(event.ProviderType)), + } + if event.ProviderId != "" { + attrs = append(attrs, otel.WgProviderId.String(event.ProviderId)) + } + if event.DestinationName != "" { + attrs = append(attrs, otel.WgDestinationName.String(event.DestinationName)) + } + if event.RootFieldName != "" { + attrs = append(attrs, otel.WgGraphQLFieldName.String(event.RootFieldName)) + } + if event.Result != "" { + attrs = append(attrs, otel.WgStreamProcessingResult.String(event.Result)) + } + if event.Reason != "" { + attrs = append(attrs, otel.WgStreamProcessingReason.String(event.Reason)) + } + count := event.Count + if count == 0 { + count = 1 + } + opt := e.withAttrs(attrs...) + for _, provider := range e.providers { + provider.Process(ctx, count, opt) + } +} diff --git a/router/pkg/otel/attributes.go b/router/pkg/otel/attributes.go index 07e0deaab2..643af9f8fa 100644 --- a/router/pkg/otel/attributes.go +++ b/router/pkg/otel/attributes.go @@ -66,11 +66,20 @@ const ( // Messaging metrics attributes const ( - WgStreamOperationName = attribute.Key("wg.stream.operation.name") - WgProviderType = attribute.Key("wg.provider.type") - WgDestinationName = attribute.Key("wg.destination.name") - WgProviderId = attribute.Key("wg.provider.id") - WgErrorType = attribute.Key("wg.error.type") + WgStreamOperationName = attribute.Key("wg.stream.operation.name") + WgProviderType = attribute.Key("wg.provider.type") + WgDestinationName = attribute.Key("wg.destination.name") + WgProviderId = attribute.Key("wg.provider.id") + WgErrorType = attribute.Key("wg.error.type") + WgStreamProcessingResult = attribute.Key("wg.stream.processing.result") + WgStreamProcessingReason = attribute.Key("wg.stream.processing.reason") +) + +const ( + WgSubscriptionFrameType = attribute.Key("wg.subscription.frame.type") + WgSubscriptionPayloadType = attribute.Key("wg.subscription.payload.type") + WgSubscriptionResult = attribute.Key("wg.subscription.result") + WgSubscriptionReason = attribute.Key("wg.subscription.reason") ) const ( diff --git a/router/pkg/pubsub/datasource/mocks.go b/router/pkg/pubsub/datasource/mocks.go index cbada9235a..c65f876b00 100644 --- a/router/pkg/pubsub/datasource/mocks.go +++ b/router/pkg/pubsub/datasource/mocks.go @@ -1209,9 +1209,17 @@ func (_c *MockSubscriptionEventUpdater_SetHooks_Call) RunAndReturn(run func(hook } // Update provides a mock function for the type MockSubscriptionEventUpdater -func (_mock *MockSubscriptionEventUpdater) Update(events []StreamEvent) { - _mock.Called(events) - return +func (_mock *MockSubscriptionEventUpdater) Update(events []StreamEvent) SubscriptionEventUpdateResult { + ret := _mock.Called(events) + + if len(ret) == 0 { + return SubscriptionEventUpdateResult{} + } + + if returnFunc, ok := ret.Get(0).(func([]StreamEvent) SubscriptionEventUpdateResult); ok { + return returnFunc(events) + } + return ret.Get(0).(SubscriptionEventUpdateResult) } // MockSubscriptionEventUpdater_Update_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Update' @@ -1238,12 +1246,12 @@ func (_c *MockSubscriptionEventUpdater_Update_Call) Run(run func(events []Stream return _c } -func (_c *MockSubscriptionEventUpdater_Update_Call) Return() *MockSubscriptionEventUpdater_Update_Call { - _c.Call.Return() +func (_c *MockSubscriptionEventUpdater_Update_Call) Return(result SubscriptionEventUpdateResult) *MockSubscriptionEventUpdater_Update_Call { + _c.Call.Return(result) return _c } -func (_c *MockSubscriptionEventUpdater_Update_Call) RunAndReturn(run func(events []StreamEvent)) *MockSubscriptionEventUpdater_Update_Call { - _c.Run(run) +func (_c *MockSubscriptionEventUpdater_Update_Call) RunAndReturn(run func(events []StreamEvent) SubscriptionEventUpdateResult) *MockSubscriptionEventUpdater_Update_Call { + _c.Call.Return(run) return _c } diff --git a/router/pkg/pubsub/datasource/subscription_event_updater.go b/router/pkg/pubsub/datasource/subscription_event_updater.go index 77d01c407a..b2f69d9580 100644 --- a/router/pkg/pubsub/datasource/subscription_event_updater.go +++ b/router/pkg/pubsub/datasource/subscription_event_updater.go @@ -17,12 +17,18 @@ const defaultTimeout = 5 * time.Second // that provides a way to send the event struct instead of the raw data // It is used to give access to the event additional fields to the hooks. type SubscriptionEventUpdater interface { - Update(events []StreamEvent) + Update(events []StreamEvent) SubscriptionEventUpdateResult Complete() Done() SetHooks(hooks Hooks) } +type SubscriptionEventUpdateResult struct { + InputCount int + DispatchedCount int + DropReason string +} + type subscriptionEventUpdater struct { eventUpdater resolve.SubscriptionUpdater subscriptionEventConfiguration SubscriptionEventConfiguration @@ -34,10 +40,20 @@ type subscriptionEventUpdater struct { semaphore *semaphore.Weighted } -func (s *subscriptionEventUpdater) Update(events []StreamEvent) { - events, ok := s.runBeforeEventsDispatchHooks(events) +func (s *subscriptionEventUpdater) Update(events []StreamEvent) SubscriptionEventUpdateResult { + result := SubscriptionEventUpdateResult{InputCount: len(events)} + events, ok, dropReason := s.runBeforeEventsDispatchHooks(events) if !ok { - return + result.DropReason = dropReason + return result + } + for _, event := range events { + if event != nil { + result.DispatchedCount++ + } + } + if result.DispatchedCount < result.InputCount { + result.DropReason = "before_dispatch_removed" } if len(s.hooks.OnReceiveEvents.Handlers) == 0 { @@ -47,7 +63,7 @@ func (s *subscriptionEventUpdater) Update(events []StreamEvent) { } s.eventUpdater.Update(event.GetData()) } - return + return result } subscriptions := s.eventUpdater.Subscriptions() @@ -90,14 +106,15 @@ func (s *subscriptionEventUpdater) Update(events []StreamEvent) { "max_concurrent_handlers or reduce handler execution time." + "Events may arrive out of order.") } + return result } // runBeforeEventsDispatchHooks runs the BeforeEventsDispatch hooks once per received batch, // before any per-subscriber fan-out. It returns the (possibly transformed) events and // false if a hook failed and the batch should be dropped. -func (s *subscriptionEventUpdater) runBeforeEventsDispatchHooks(events []StreamEvent) ([]StreamEvent, bool) { +func (s *subscriptionEventUpdater) runBeforeEventsDispatchHooks(events []StreamEvent) ([]StreamEvent, bool, string) { if len(s.hooks.BeforeEventsDispatch.Handlers) == 0 { - return events, true + return events, true, "" } ctx, cancel := context.WithTimeout(context.Background(), s.beforeEventsDispatchTimeout) @@ -106,11 +123,12 @@ func (s *subscriptionEventUpdater) runBeforeEventsDispatchHooks(events []StreamE type hookResult struct { events []StreamEvent ok bool + reason string } done := make(chan hookResult, 1) go func() { - res := hookResult{nil, false} + res := hookResult{events: nil, ok: false} defer func() { if r := recover(); r != nil { s.logger. @@ -119,7 +137,8 @@ func (s *subscriptionEventUpdater) runBeforeEventsDispatchHooks(events []StreamE zap.String("handler_name", "BeforeEventsDispatch"), zap.Any("error", r), ) - res = hookResult{nil, false} + res = hookResult{events: nil, ok: false} + res.reason = "before_dispatch_panic" } done <- res }() @@ -132,19 +151,20 @@ func (s *subscriptionEventUpdater) runBeforeEventsDispatchHooks(events []StreamE s.logger. With(zap.Int("handler_index", i)). Warn("BeforeEventsDispatch handler failed, dropping event batch", zap.Error(err)) + res.reason = "before_dispatch_error" return } } - res = hookResult{evts, true} + res = hookResult{events: evts, ok: true} }() select { case res := <-done: - return res.events, res.ok + return res.events, res.ok, res.reason case <-ctx.Done(): s.logger.Warn("BeforeEventsDispatch handler timeout exceeded, dropping event batch. " + "Consider increasing events.handler.before_events_dispatch.handler_timeout or reduce handler execution time.") - return nil, false + return nil, false, "before_dispatch_timeout" } } diff --git a/router/pkg/pubsub/datasource/subscription_event_updater_beforeeventsdispatch_test.go b/router/pkg/pubsub/datasource/subscription_event_updater_beforeeventsdispatch_test.go index 333207ca7d..ad36523e77 100644 --- a/router/pkg/pubsub/datasource/subscription_event_updater_beforeeventsdispatch_test.go +++ b/router/pkg/pubsub/datasource/subscription_event_updater_beforeeventsdispatch_test.go @@ -55,7 +55,8 @@ func TestSubscriptionEventUpdater_Update_WithBeforeEventsDispatchHooks_Success(t testEventBuilder, ) - updater.Update(originalEvents) + result := updater.Update(originalEvents) + assert.Equal(t, SubscriptionEventUpdateResult{InputCount: 1, DispatchedCount: 1}, result) select { case receivedArgs := <-receivedArgs: @@ -99,7 +100,8 @@ func TestSubscriptionEventUpdater_Update_WithBeforeEventsDispatchHooks_Error(t * testEventBuilder, ) - updater.Update(events) + result := updater.Update(events) + assert.Equal(t, SubscriptionEventUpdateResult{InputCount: 1, DropReason: "before_dispatch_error"}, result) // Assert that the whole batch was dropped mockUpdater.AssertNotCalled(t, "Update") @@ -108,6 +110,29 @@ func TestSubscriptionEventUpdater_Update_WithBeforeEventsDispatchHooks_Error(t * mockUpdater.AssertNotCalled(t, "CloseSubscription") } +func TestSubscriptionEventUpdater_Update_ReportsEventsRemovedByBeforeDispatchHook(t *testing.T) { + mockUpdater := NewMockSubscriptionUpdater(t) + config := &testSubscriptionEventConfig{providerID: "test-provider", providerType: ProviderTypeKafka, fieldName: "testField"} + events := []StreamEvent{ + &testEvent{mutableTestEvent("keep")}, + &testEvent{mutableTestEvent("drop")}, + } + mockUpdater.On("Update", []byte("keep")).Return() + updater := NewSubscriptionEventUpdater(config, Hooks{ + BeforeEventsDispatch: BeforeEventsDispatchHooks{Handlers: []BeforeEventsDispatchFn{ + func(context.Context, SubscriptionEventConfiguration, EventBuilderFn, []StreamEvent) ([]StreamEvent, error) { + return events[:1], nil + }, + }}, + }, mockUpdater, zap.NewNop(), testEventBuilder) + + result := updater.Update(events) + + assert.Equal(t, SubscriptionEventUpdateResult{ + InputCount: 2, DispatchedCount: 1, DropReason: "before_dispatch_removed", + }, result) +} + func TestSubscriptionEventUpdater_Update_WithMultipleBeforeEventsDispatchHooks_Success(t *testing.T) { mockUpdater := NewMockSubscriptionUpdater(t) config := &testSubscriptionEventConfig{ diff --git a/router/pkg/pubsub/kafka/adapter.go b/router/pkg/pubsub/kafka/adapter.go index 3ee51437f4..093d6fe031 100644 --- a/router/pkg/pubsub/kafka/adapter.go +++ b/router/pkg/pubsub/kafka/adapter.go @@ -48,7 +48,8 @@ type ProviderAdapter struct { } type PollerOpts struct { - providerId string + providerId string + rootFieldName string } // topicPoller polls the Kafka topic for new records and calls the updateTriggers function. @@ -109,7 +110,7 @@ func (p *ProviderAdapter) topicPoller(ctx context.Context, client *kgo.Client, u DestinationName: r.Topic, }) - updater.Update([]datasource.StreamEvent{ + updateResult := updater.Update([]datasource.StreamEvent{ &Event{ evt: &MutableEvent{ Data: r.Value, @@ -118,6 +119,31 @@ func (p *ProviderAdapter) topicPoller(ctx context.Context, client *kgo.Client, u }, }, }) + if updateResult.DispatchedCount > 0 { + p.streamMetricStore.Process(ctx, metric.StreamsEvent{ + ProviderId: pollerOpts.providerId, + StreamOperationName: kafkaReceive, + ProviderType: metric.ProviderTypeKafka, + DestinationName: r.Topic, + RootFieldName: pollerOpts.rootFieldName, + Result: "dispatched", + Reason: "none", + Count: int64(updateResult.DispatchedCount), + }) + } + droppedCount := updateResult.InputCount - updateResult.DispatchedCount + if droppedCount > 0 { + p.streamMetricStore.Process(ctx, metric.StreamsEvent{ + ProviderId: pollerOpts.providerId, + StreamOperationName: kafkaReceive, + ProviderType: metric.ProviderTypeKafka, + DestinationName: r.Topic, + RootFieldName: pollerOpts.rootFieldName, + Result: "dropped", + Reason: updateResult.DropReason, + Count: int64(droppedCount), + }) + } } } } @@ -173,7 +199,10 @@ func (p *ProviderAdapter) Subscribe(ctx context.Context, conf datasource.Subscri stop := context.AfterFunc(p.ctx, cancel) defer stop() - err := p.topicPoller(pollerCtx, client, updater, PollerOpts{providerId: conf.ProviderID()}) + err := p.topicPoller(pollerCtx, client, updater, PollerOpts{ + providerId: conf.ProviderID(), + rootFieldName: conf.RootFieldName(), + }) if err != nil { if errors.Is(err, errClientClosed) || errors.Is(err, context.Canceled) { log.Debug("poller canceled", zap.Error(err)) diff --git a/router/pkg/pubsub/nats/adapter.go b/router/pkg/pubsub/nats/adapter.go index a8ba5c3c7e..486cf05fe8 100644 --- a/router/pkg/pubsub/nats/adapter.go +++ b/router/pkg/pubsub/nats/adapter.go @@ -158,12 +158,13 @@ func (p *ProviderAdapter) Subscribe(ctx context.Context, cfg datasource.Subscrip DestinationName: msg.Subject(), }) - updater.Update([]datasource.StreamEvent{ + updateResult := updater.Update([]datasource.StreamEvent{ &Event{evt: &MutableEvent{ Data: msg.Data(), Headers: map[string][]string(msg.Headers()), }}, }) + p.recordProcessingResult(p.ctx, subConf, msg.Subject(), updateResult) // Acknowledge the message after it has been processed ackErr := msg.Ack() @@ -208,12 +209,13 @@ func (p *ProviderAdapter) Subscribe(ctx context.Context, cfg datasource.Subscrip ProviderType: metric.ProviderTypeNats, DestinationName: msg.Subject, }) - updater.Update([]datasource.StreamEvent{ + updateResult := updater.Update([]datasource.StreamEvent{ &Event{evt: &MutableEvent{ Data: msg.Data, Headers: map[string][]string(msg.Header), }}, }) + p.recordProcessingResult(p.ctx, subConf, msg.Subject, updateResult) case <-p.ctx.Done(): // When the application context is done, we stop the subscriptions for _, subscription := range subscriptions { @@ -241,6 +243,21 @@ func (p *ProviderAdapter) Subscribe(ctx context.Context, cfg datasource.Subscrip return nil } +func (p *ProviderAdapter) recordProcessingResult(ctx context.Context, conf datasource.SubscriptionEventConfiguration, destination string, result datasource.SubscriptionEventUpdateResult) { + if result.DispatchedCount > 0 { + p.streamMetricStore.Process(ctx, metric.StreamsEvent{ + ProviderId: conf.ProviderID(), StreamOperationName: natsReceive, ProviderType: metric.ProviderTypeNats, + DestinationName: destination, RootFieldName: conf.RootFieldName(), Result: "dispatched", Reason: "none", Count: int64(result.DispatchedCount), + }) + } + if dropped := result.InputCount - result.DispatchedCount; dropped > 0 { + p.streamMetricStore.Process(ctx, metric.StreamsEvent{ + ProviderId: conf.ProviderID(), StreamOperationName: natsReceive, ProviderType: metric.ProviderTypeNats, + DestinationName: destination, RootFieldName: conf.RootFieldName(), Result: "dropped", Reason: result.DropReason, Count: int64(dropped), + }) + } +} + func (p *ProviderAdapter) Publish(ctx context.Context, conf datasource.PublishEventConfiguration, events []datasource.StreamEvent) error { pubConf, ok := conf.(*PublishAndRequestEventConfiguration) if !ok { diff --git a/router/pkg/pubsub/redis/adapter.go b/router/pkg/pubsub/redis/adapter.go index 606a473e96..3c1bad8f22 100644 --- a/router/pkg/pubsub/redis/adapter.go +++ b/router/pkg/pubsub/redis/adapter.go @@ -161,11 +161,12 @@ func (p *ProviderAdapter) Subscribe(ctx context.Context, conf datasource.Subscri ProviderType: metric.ProviderTypeRedis, DestinationName: msg.Channel, }) - updater.Update([]datasource.StreamEvent{ + updateResult := updater.Update([]datasource.StreamEvent{ &Event{evt: &MutableEvent{ Data: []byte(msg.Payload), }}, }) + p.recordProcessingResult(ctx, conf, msg.Channel, updateResult) case <-p.ctx.Done(): // When the application context is done, we stop the subscription if it is not already done log.Debug("application context done, stopping subscription") @@ -181,6 +182,21 @@ func (p *ProviderAdapter) Subscribe(ctx context.Context, conf datasource.Subscri return nil } +func (p *ProviderAdapter) recordProcessingResult(ctx context.Context, conf datasource.SubscriptionEventConfiguration, destination string, result datasource.SubscriptionEventUpdateResult) { + if result.DispatchedCount > 0 { + p.streamMetricStore.Process(ctx, metric.StreamsEvent{ + ProviderId: conf.ProviderID(), StreamOperationName: redisReceive, ProviderType: metric.ProviderTypeRedis, + DestinationName: destination, RootFieldName: conf.RootFieldName(), Result: "dispatched", Reason: "none", Count: int64(result.DispatchedCount), + }) + } + if dropped := result.InputCount - result.DispatchedCount; dropped > 0 { + p.streamMetricStore.Process(ctx, metric.StreamsEvent{ + ProviderId: conf.ProviderID(), StreamOperationName: redisReceive, ProviderType: metric.ProviderTypeRedis, + DestinationName: destination, RootFieldName: conf.RootFieldName(), Result: "dropped", Reason: result.DropReason, Count: int64(dropped), + }) + } +} + func (p *ProviderAdapter) Publish(ctx context.Context, conf datasource.PublishEventConfiguration, events []datasource.StreamEvent) error { pubConf, ok := conf.(*PublishEventConfiguration) if !ok { diff --git a/router/pkg/pubsub/redis/adapter_test.go b/router/pkg/pubsub/redis/adapter_test.go index 261cd2d7e1..bdcfef515a 100644 --- a/router/pkg/pubsub/redis/adapter_test.go +++ b/router/pkg/pubsub/redis/adapter_test.go @@ -14,10 +14,12 @@ import ( type noopUpdater struct{} -func (n *noopUpdater) Update(_ []datasource.StreamEvent) {} -func (n *noopUpdater) Complete() {} -func (n *noopUpdater) Done() {} -func (n *noopUpdater) SetHooks(_ datasource.Hooks) {} +func (n *noopUpdater) Update(events []datasource.StreamEvent) datasource.SubscriptionEventUpdateResult { + return datasource.SubscriptionEventUpdateResult{InputCount: len(events), DispatchedCount: len(events)} +} +func (n *noopUpdater) Complete() {} +func (n *noopUpdater) Done() {} +func (n *noopUpdater) SetHooks(_ datasource.Hooks) {} func TestProviderAdapter_SubscribeWithoutStartupReturnsError(t *testing.T) { t.Parallel() diff --git a/router/pkg/statistics/engine_stats.go b/router/pkg/statistics/engine_stats.go index 80b159afe6..b047554184 100644 --- a/router/pkg/statistics/engine_stats.go +++ b/router/pkg/statistics/engine_stats.go @@ -30,6 +30,45 @@ type EngineStatistics interface { UnregisterResolver(r ResolverConcurrencyReporter) } +// SubscriptionObserver is implemented by engine statistics collectors that expose +// detailed, low-cardinality subscription outcomes. It is optional so custom +// EngineStatistics implementations remain source compatible. +type SubscriptionObserver interface { + SubscriptionResolutionError(reason SubscriptionResolutionErrorReason) + WebSocketFrame(observation WebSocketFrameObservation) +} + +type SubscriptionResolutionErrorReason string + +const ( + SubscriptionResolutionErrorAuthorization SubscriptionResolutionErrorReason = "authorization_error" + SubscriptionResolutionErrorContextCanceled SubscriptionResolutionErrorReason = "context_canceled" + SubscriptionResolutionErrorFetch SubscriptionResolutionErrorReason = "fetch_error" + SubscriptionResolutionErrorHandler SubscriptionResolutionErrorReason = "handler_error" + SubscriptionResolutionErrorInvalidMessage SubscriptionResolutionErrorReason = "invalid_message" + SubscriptionResolutionErrorRateLimit SubscriptionResolutionErrorReason = "rate_limit" + SubscriptionResolutionErrorResolve SubscriptionResolutionErrorReason = "resolve_error" + SubscriptionResolutionErrorTimeout SubscriptionResolutionErrorReason = "fetch_timeout" + SubscriptionResolutionErrorUnknown SubscriptionResolutionErrorReason = "unknown" +) + +type WebSocketFrameObservation struct { + FrameType string + PayloadType string + Result string + Reason string +} + +type SubscriptionResolutionErrorCount struct { + Reason SubscriptionResolutionErrorReason + Count uint64 +} + +type WebSocketFrameCount struct { + Observation WebSocketFrameObservation + Count uint64 +} + type EngineStats struct { ctx context.Context logger *zap.Logger @@ -39,17 +78,22 @@ type EngineStats struct { messagesSent atomic.Uint64 triggers atomic.Uint64 + resolutionErrors sync.Map // map[SubscriptionResolutionErrorReason]*atomic.Uint64 + webSocketFrames sync.Map // map[WebSocketFrameObservation]*atomic.Uint64 + resolverMu sync.RWMutex resolverReporters map[ResolverConcurrencyReporter]struct{} } type UsageReport struct { - Connections uint64 - Subscriptions uint64 - MessagesSent uint64 - Triggers uint64 - ResolverMaxConcurrent uint64 - ResolverInflight uint64 + Connections uint64 + Subscriptions uint64 + MessagesSent uint64 + Triggers uint64 + ResolverMaxConcurrent uint64 + ResolverInflight uint64 + SubscriptionResolutionErrors []SubscriptionResolutionErrorCount + WebSocketFrames []WebSocketFrameCount } // NewEngineStats creates a new EngineStats instance. If reportStats is true, the stats will be reported every 5 seconds. @@ -79,6 +123,20 @@ func (s *EngineStats) GetReport() *UsageReport { report.ResolverInflight += uint64(r.InflightResolves()) } s.resolverMu.RUnlock() + s.resolutionErrors.Range(func(key, value any) bool { + report.SubscriptionResolutionErrors = append(report.SubscriptionResolutionErrors, SubscriptionResolutionErrorCount{ + Reason: key.(SubscriptionResolutionErrorReason), + Count: value.(*atomic.Uint64).Load(), + }) + return true + }) + s.webSocketFrames.Range(func(key, value any) bool { + report.WebSocketFrames = append(report.WebSocketFrames, WebSocketFrameCount{ + Observation: key.(WebSocketFrameObservation), + Count: value.(*atomic.Uint64).Load(), + }) + return true + }) return report } @@ -108,6 +166,16 @@ func (s *EngineStats) SubscriptionUpdateSent() { s.messagesSent.Inc() } +func (s *EngineStats) SubscriptionResolutionError(reason SubscriptionResolutionErrorReason) { + counter, _ := s.resolutionErrors.LoadOrStore(reason, &atomic.Uint64{}) + counter.(*atomic.Uint64).Inc() +} + +func (s *EngineStats) WebSocketFrame(observation WebSocketFrameObservation) { + counter, _ := s.webSocketFrames.LoadOrStore(observation, &atomic.Uint64{}) + counter.(*atomic.Uint64).Inc() +} + func (s *EngineStats) ConnectionsInc() { s.connections.Inc() } @@ -166,6 +234,10 @@ func (s *NoopEngineStats) GetReport() *UsageReport { func (s *NoopEngineStats) SubscriptionUpdateSent() {} +func (s *NoopEngineStats) SubscriptionResolutionError(_ SubscriptionResolutionErrorReason) {} + +func (s *NoopEngineStats) WebSocketFrame(_ WebSocketFrameObservation) {} + func (s *NoopEngineStats) ConnectionsInc() {} func (s *NoopEngineStats) ConnectionsDec() {} @@ -187,3 +259,5 @@ func (s *NoopEngineStats) UnregisterResolver(_ ResolverConcurrencyReporter) {} var _ EngineStatistics = &EngineStats{} var _ EngineStatistics = &NoopEngineStats{} +var _ SubscriptionObserver = &EngineStats{} +var _ SubscriptionObserver = &NoopEngineStats{} diff --git a/router/pkg/statistics/engine_stats_test.go b/router/pkg/statistics/engine_stats_test.go new file mode 100644 index 0000000000..e1e715892e --- /dev/null +++ b/router/pkg/statistics/engine_stats_test.go @@ -0,0 +1,21 @@ +package statistics + +import ( + "testing" + + "github.com/stretchr/testify/require" + "go.uber.org/zap" +) + +func TestEngineStatsSubscriptionObservations(t *testing.T) { + t.Parallel() + stats := NewEngineStats(t.Context(), zap.NewNop(), false) + stats.SubscriptionResolutionError(SubscriptionResolutionErrorTimeout) + stats.SubscriptionResolutionError(SubscriptionResolutionErrorTimeout) + observation := WebSocketFrameObservation{FrameType: "data", PayloadType: "data", Result: "failure", Reason: "client_disconnected"} + stats.WebSocketFrame(observation) + + report := stats.GetReport() + require.Equal(t, []SubscriptionResolutionErrorCount{{Reason: SubscriptionResolutionErrorTimeout, Count: 2}}, report.SubscriptionResolutionErrors) + require.Equal(t, []WebSocketFrameCount{{Observation: observation, Count: 1}}, report.WebSocketFrames) +}