Skip to content

fix(policy-engine): scope loopback suppression per publisher #3107

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
sanjulaonline wants to merge 1 commit into
base: main
Choose a base branch
from
+127 −42
Open

fix(policy-engine): scope loopback suppression per publisher #3107

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
87 changes: 60 additions & 27 deletions gateway/gateway-runtime/policy-engine/internal/analytics/analytics.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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:
Expand All @@ -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")
}

Expand Down Expand Up @@ -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)
}

}
Expand Down
82 changes: 67 additions & 15 deletions gateway/gateway-runtime/policy-engine/internal/analytics/analytics_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
}

// =============================================================================
Expand Down Expand Up @@ -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",
Expand All @@ -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)},
Expand Down Expand Up @@ -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))

Expand Down Expand Up @@ -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",
Expand All @@ -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))

Expand All @@ -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"))
Expand All @@ -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"))

Expand All @@ -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",
Expand Down