From 64409fc068e8526c357c9511b7c1264b79f7dc77 Mon Sep 17 00:00:00 2001 From: Sanjula Herath Date: Tue, 4 Aug 2026 02:44:39 +0530 Subject: [PATCH] fix(policy-engine): scope loopback suppression per publisher The internal provider hop of an LLM proxy call was returned before publisher fan-out, dropping it for both Moesif and traffic logging. Only Moesif needs that hop suppressed to avoid counting one client call twice; traffic logging needs it to record the vendor round-trip. Register each publisher with its delivery rule and skip only consumers that opt into loopback suppression. Keep the suppression debug trace single-shot and emit it only when a publisher is actually skipped. Fixes #2992 --- .../internal/analytics/analytics.go | 87 +++++++++++++------ .../internal/analytics/analytics_test.go | 82 +++++++++++++---- 2 files changed, 127 insertions(+), 42 deletions(-) diff --git a/gateway/gateway-runtime/policy-engine/internal/analytics/analytics.go b/gateway/gateway-runtime/policy-engine/internal/analytics/analytics.go index ef600d12db..ca0e9d0dfe 100644 --- a/gateway/gateway-runtime/policy-engine/internal/analytics/analytics.go +++ b/gateway/gateway-runtime/policy-engine/internal/analytics/analytics.go @@ -87,17 +87,28 @@ const ( // internal loopback request to the provider. InternalLoopbackMetadataKey string = "x-wso2-internal-loopback" - // PropInternalLoopbackProvider marks the provider-side loopback hop of a proxy call - // so Process can drop the duplicate event before publisher fan-out. + // PropInternalLoopbackProvider marks the provider-side loopback hop of a proxy call so + // Process can apply each consumer's delivery rule to it during publisher fan-out. PropInternalLoopbackProvider string = "isInternalLoopbackProvider" ) +// registeredPublisher pairs a publisher with the per-consumer delivery rules that only +// the collector knows about, keeping them out of the Publisher interface. +type registeredPublisher struct { + // publisher receives the prepared event. + publisher analytics_publisher.Publisher + // suppressInternalLoopback drops the provider-side loopback hop of a proxy call for + // this consumer only. Set for consumers that count a client call once (Moesif) and + // left unset for consumers that record every hop (traffic logging). + suppressInternalLoopback bool +} + // Analytics represents analytics collector service. type Analytics struct { // cfg represents the server configuration. cfg *config.Config - // publishers represents the publishers. - publishers []analytics_publisher.Publisher + // publishers represents the publishers together with their delivery rules. + publishers []registeredPublisher // missingDirectPeerWarn limits the "direct remote address unavailable" warning to one // line per process which otherwise repeating it once per request would flood the logs missingDirectPeerWarn sync.Once @@ -110,14 +121,18 @@ type Analytics struct { // receive any events. func NewAnalytics(cfg *config.Config) *Analytics { analyticsCfg := cfg.Analytics - publishers := make([]analytics_publisher.Publisher, 0) + publishers := make([]registeredPublisher, 0) if analyticsCfg.Enabled { for _, publisherName := range analyticsCfg.EnabledPublishers { switch publisherName { case MoesifAnalyticsPublisher: publisher := analytics_publisher.NewMoesif(&analyticsCfg.Publishers.Moesif) if publisher != nil { - publishers = append(publishers, publisher) + // Moesif counts one client call once, so it must not see the loopback hop. + publishers = append(publishers, registeredPublisher{ + publisher: publisher, + suppressInternalLoopback: true, + }) slog.Info("Moesif publisher added") } default: @@ -126,9 +141,16 @@ func NewAnalytics(cfg *config.Config) *Analytics { } } - // Traffic logging is a standalone consumer, independent of analytics. + // Traffic logging is a standalone consumer, independent of analytics. It records every + // hop, so the provider hop is the only measure of the real vendor round-trip and must + // not be suppressed. if cfg.TrafficLogging.Enabled { - publishers = append(publishers, analytics_publisher.NewLog(&cfg.TrafficLogging)) + publishers = append(publishers, registeredPublisher{ + publisher: analytics_publisher.NewLog(&cfg.TrafficLogging), + // Stated explicitly rather than left to the zero value: delivering the provider + // hop here is the behaviour this registration exists to guarantee. + suppressInternalLoopback: false, + }) slog.Info("Traffic logging (stdout) publisher added") } @@ -158,27 +180,38 @@ func (c *Analytics) Process(event *v3.HTTPAccessLogEntry) { analyticEvent := c.prepareAnalyticEvent(event) - // Suppress the internal loopback provider hop of an LLM proxy call so a single client - // call is counted once, detecting using the marker header set by the proxy - // when carrying on its loopback forward + // The internal loopback provider hop of an LLM proxy call is a duplicate only for + // consumers that count a client call once, detected using the marker header set by the + // proxy when carrying on its loopback forward. It is filtered per consumer during + // fan-out rather than globally, so consumers that record every hop still receive it. + isInternalLoopbackProvider := false if v, ok := analyticEvent.Properties[PropInternalLoopbackProvider].(bool); ok && v { - correlationID := "" - if analyticEvent.MetaInfo != nil { - correlationID = analyticEvent.MetaInfo.CorrelationID - } - apiType := "" - if analyticEvent.API != nil { - apiType = analyticEvent.API.APIType + isInternalLoopbackProvider = true + } + + // Traced once per event, and only once a consumer has actually been skipped: a + // deployment with no suppressing consumer drops nothing and must not claim otherwise. + suppressionLogged := false + for _, registered := range c.publishers { + if isInternalLoopbackProvider && registered.suppressInternalLoopback { + if !suppressionLogged { + suppressionLogged = true + correlationID := "" + if analyticEvent.MetaInfo != nil { + correlationID = analyticEvent.MetaInfo.CorrelationID + } + apiType := "" + if analyticEvent.API != nil { + apiType = analyticEvent.API.APIType + } + slog.Debug("Suppressing internal loopback provider analytics event", + "apiType", apiType, + "correlationId", correlationID, + ) + } + continue } - slog.Debug("Suppressing internal loopback provider analytics event", - "apiType", apiType, - "correlationId", correlationID, - ) - return - } - - for _, publisher := range c.publishers { - publisher.Publish(analyticEvent) + registered.publisher.Publish(analyticEvent) } } diff --git a/gateway/gateway-runtime/policy-engine/internal/analytics/analytics_test.go b/gateway/gateway-runtime/policy-engine/internal/analytics/analytics_test.go index 2969693f9c..33b62d47a6 100644 --- a/gateway/gateway-runtime/policy-engine/internal/analytics/analytics_test.go +++ b/gateway/gateway-runtime/policy-engine/internal/analytics/analytics_test.go @@ -159,7 +159,11 @@ func TestNewAnalytics_TrafficLoggingEnabled(t *testing.T) { analytics := NewAnalytics(cfg) require.NotNil(t, analytics) - assert.Len(t, analytics.publishers, 1) // traffic-logging publisher should be registered + require.Len(t, analytics.publishers, 1, "traffic-logging publisher should be registered") + // Asserted on the real constructor mapping, not a hand-assembled registration: traffic + // logging must receive the internal loopback provider hop. + assert.False(t, analytics.publishers[0].suppressInternalLoopback, + "traffic logging must be registered as non-suppressing") } // ============================================================================= @@ -263,7 +267,7 @@ func TestProcess_WithMockPublisher(t *testing.T) { // Inject a mock publisher mockPub := &mockPublisher{} - analytics.publishers = append(analytics.publishers, mockPub) + analytics.publishers = append(analytics.publishers, registeredPublisher{publisher: mockPub}) logEntry := createLogEntryWithMetadata(map[string]string{ APINameKey: "TestAPI", @@ -284,7 +288,7 @@ func TestProcess_WithMockPublisher(t *testing.T) { func TestProcess_PublishesEvent(t *testing.T) { analytics := NewAnalytics(&config.Config{}) mockPub := &mockPublisher{} - analytics.publishers = append(analytics.publishers, mockPub) + analytics.publishers = append(analytics.publishers, registeredPublisher{publisher: mockPub}) logEntry := &v3.HTTPAccessLogEntry{ Response: &v3.HTTPResponseProperties{ResponseCode: wrapperspb.UInt32(200)}, @@ -379,23 +383,65 @@ func loopbackEntry(apiType string, marker bool) *v3.HTTPAccessLogEntry { return e } -func TestProcess_SuppressesFlaggedLoopbackProvider(t *testing.T) { - // Marker + loopback + LlmProvider → the internal loopback hop is suppressed. +func TestProcess_SuppressesFlaggedLoopbackProviderForSuppressingPublisher(t *testing.T) { + // Marker + loopback + LlmProvider → the internal loopback hop is suppressed, but only + // for a consumer registered as counting a client call once (the Moesif shape). analytics := NewAnalytics(&config.Config{}) mockPub := &mockPublisher{} - analytics.publishers = append(analytics.publishers, mockPub) + analytics.publishers = append(analytics.publishers, registeredPublisher{ + publisher: mockPub, + suppressInternalLoopback: true, + }) analytics.Process(loopbackEntry("LlmProvider", true)) assert.False(t, mockPub.called, "flagged internal loopback provider event must be suppressed") } +func TestProcess_FlaggedLoopbackProviderSuppressedPerPublisher(t *testing.T) { + // The provider hop is the only record of the real vendor round-trip, so suppression is + // scoped to the consumer that needs it and every other consumer still receives it. + analytics := NewAnalytics(&config.Config{}) + suppressing := &mockPublisher{} + nonSuppressing := &mockPublisher{} + analytics.publishers = append(analytics.publishers, + registeredPublisher{publisher: suppressing, suppressInternalLoopback: true}, + registeredPublisher{publisher: nonSuppressing, suppressInternalLoopback: false}, + ) + + analytics.Process(loopbackEntry("LlmProvider", true)) + + assert.False(t, suppressing.called, "suppressing consumer must not receive the internal loopback provider hop") + assert.Equal(t, 0, suppressing.count, "suppressing consumer must receive zero events") + assert.True(t, nonSuppressing.called, "non-suppressing consumer must still receive the internal loopback provider hop") + assert.Equal(t, 1, nonSuppressing.count, "non-suppressing consumer must receive the hop exactly once") +} + +func TestProcess_NonSuppressingLoopbackDoesNotLogSuppression(t *testing.T) { + // With no suppressing consumer registered nothing is dropped, so claiming suppression + // in the log would send an operator hunting for an event that was in fact delivered. + buf := captureLogs(t) + analytics := NewAnalytics(&config.Config{}) + nonSuppressing := &mockPublisher{} + analytics.publishers = append(analytics.publishers, registeredPublisher{ + publisher: nonSuppressing, + suppressInternalLoopback: false, + }) + + analytics.Process(loopbackEntry("LlmProvider", true)) + + assert.True(t, nonSuppressing.called, "the only registered consumer must receive the flagged hop") + assert.Equal(t, 1, nonSuppressing.count, "the flagged hop must be published exactly once") + assert.NotContains(t, buf.String(), "Suppressing internal loopback provider analytics event", + "suppression must not be logged when no consumer was skipped") +} + func TestProcess_PublishesDirectProviderOverLoopback(t *testing.T) { // No marker (a direct provider call) is never suppressed, even when it arrives over // loopback — the sidecar/port-forward case the reviewer raised. analytics := NewAnalytics(&config.Config{}) mockPub := &mockPublisher{} - analytics.publishers = append(analytics.publishers, mockPub) + analytics.publishers = append(analytics.publishers, registeredPublisher{publisher: mockPub}) analytics.Process(loopbackEntry("LlmProvider", false)) @@ -441,7 +487,7 @@ func TestProcess_PublishesWhenDirectRemoteIPMissing(t *testing.T) { // duplicate) rather than suppressed on the strength of a forgeable marker alone. analytics := NewAnalytics(&config.Config{}) mockPub := &mockPublisher{} - analytics.publishers = append(analytics.publishers, mockPub) + analytics.publishers = append(analytics.publishers, registeredPublisher{publisher: mockPub}) entry := createLogEntryWithMetadata(map[string]string{ APITypeKey: "LlmProvider", @@ -463,7 +509,10 @@ func TestProcess_SuppressionLogsDebug(t *testing.T) { buf := captureLogs(t) analytics := NewAnalytics(&config.Config{}) mockPub := &mockPublisher{} - analytics.publishers = append(analytics.publishers, mockPub) + analytics.publishers = append(analytics.publishers, registeredPublisher{ + publisher: mockPub, + suppressInternalLoopback: true, + }) analytics.Process(loopbackEntry("LlmProvider", true)) @@ -473,13 +522,16 @@ func TestProcess_SuppressionLogsDebug(t *testing.T) { // TestProcess_ProxyCallPublishesExactlyOneEvent is the end-to-end shape of the bug this // suppression exists for: a single client call to an LLM proxy traverses the listener twice, so -// the access-log service delivers two entries. Exactly one event must reach the publishers, and -// it must be the proxy's — that is the hop carrying the user identity, application and -// subscription; the provider hop is anonymous. +// the access-log service delivers two entries. Exactly one event must reach a consumer that +// counts a client call once, and it must be the proxy's — that is the hop carrying the user +// identity, application and subscription; the provider hop is anonymous. func TestProcess_ProxyCallPublishesExactlyOneEvent(t *testing.T) { analytics := NewAnalytics(&config.Config{}) mockPub := &mockPublisher{} - analytics.publishers = append(analytics.publishers, mockPub) + analytics.publishers = append(analytics.publishers, registeredPublisher{ + publisher: mockPub, + suppressInternalLoopback: true, + }) // Hop 1 — the proxy's own event: real client peer, no marker. analytics.Process(createLogEntryWithAPITypeAndAddr("LlmProxy", "203.0.113.7")) @@ -498,7 +550,7 @@ func TestProcess_ProxyCallPublishesExactlyOneEvent(t *testing.T) { func TestProcess_DirectProviderCallPublishesExactlyOneEvent(t *testing.T) { analytics := NewAnalytics(&config.Config{}) mockPub := &mockPublisher{} - analytics.publishers = append(analytics.publishers, mockPub) + analytics.publishers = append(analytics.publishers, registeredPublisher{publisher: mockPub}) analytics.Process(createLogEntryWithAPITypeAndAddr("LlmProvider", "203.0.113.7")) @@ -523,7 +575,7 @@ func TestProcess_PanicRecovery(t *testing.T) { // Inject a panicking publisher panicPub := &panicPublisher{} - analytics.publishers = append(analytics.publishers, panicPub) + analytics.publishers = append(analytics.publishers, registeredPublisher{publisher: panicPub}) logEntry := createLogEntryWithMetadata(map[string]string{ APINameKey: "TestAPI",