diff --git a/gateway/configs/config-template.toml b/gateway/configs/config-template.toml index d11d5ebb66..936475b881 100644 --- a/gateway/configs/config-template.toml +++ b/gateway/configs/config-template.toml @@ -281,18 +281,62 @@ port = 9010 host = "localhost" # ============================================================================= -# ANALYTICS CONFIGURATION +# COLLECTOR CONFIGURATION +# ============================================================================= +# The collector is the shared data-capture pipeline: it injects the analytics +# system policy and ships request/response headers and bodies to the policy-engine +# over ALS. It has NO on/off switch of its own — it is implicit, turning on +# automatically whenever a consumer below is enabled ([analytics] or +# [traffic_logging]) and staying off otherwise. This section only tunes what the +# collector captures. +[collector] +# Capture request/response payloads (bodies) into the collected event. Bodies are +# captured in full; per-consumer size caps are applied on output (see +# traffic_logging.max_payload_size). +send_request_body = false +send_response_body = false +# Capture ALL request/response headers into the collected event (request_headers/ +# response_headers). Applies to every API via the collector system policy, so no +# per-API header policy is needed. Sensitive values are redacted on output by +# consumers that support it (e.g. traffic_logging.masked_headers) — the Moesif +# publisher does not mask headers. Off by default; enable deliberately per consumer. +send_request_headers = false +send_response_headers = false + +# ALS transport tuning (advanced; defaults are sensible). One section read by BOTH +# ends of the Envoy → policy-engine access-log stream: +# • gateway-controller — configures Envoy's gRPC access-log sink (the sender) +# • policy-engine — runs the receiving gRPC server +# Shared keys (server_port, TLS, message/header limits) must match on both ends and +# live here once. Envoy-sender-only keys (buffer_*, grpc_request_timeout) are read +# only by the controller and ignored by the policy-engine. +[collector.als] +mode = "uds" # "uds" (default) or "tcp" +server_port = 18090 # engine listens here; Envoy dials it (tcp mode) +buffer_flush_interval = 1000000000 # Envoy sender only (nanoseconds) +buffer_size_bytes = 16384 # Envoy sender only +grpc_request_timeout = 20000000000 # Envoy sender only (nanoseconds) +shutdown_timeout = "600s" +public_key_path = "" # TLS for the ALS connection (both ends) +private_key_path = "" +als_plain_text = true # plaintext gRPC (skip TLS) +max_message_size = 1000000000 +max_header_limit = 8192 + +# ============================================================================= +# ANALYTICS CONFIGURATION (consumer — enabling it activates the collector) # ============================================================================= [analytics] enabled = false -# Deprecated: allow_payloads is preserved for backward compatibility. When true -# and both send_request_body and send_response_body are false, both directions -# are enabled (bool fields cannot distinguish "unset" from explicitly false). -# To disable payloads entirely, set allow_payloads = false or remove it; do not -# rely on send_*_body = false while allow_payloads remains true. +# Deprecated: allow_payloads / send_request_body / send_response_body are preserved +# for backward compatibility and are mapped onto the [collector] body-capture flags +# during validation (with a warning). Prefer setting [collector] directly. allow_payloads = false -send_request_body = false -send_response_body = false +# send_request_body = false +# send_response_body = false +# Publishers that receive analytics events. Supported: "moesif" (sends to the +# Moesif SaaS). For stdout JSON logging use [traffic_logging] instead of a "log" +# publisher here. enabled_publishers = ["moesif"] [analytics.publishers.moesif] @@ -303,17 +347,32 @@ event_queue_size = 10000 batch_size = 50 timer_wakeup_seconds = 3 -[analytics.grpc_event_server] -buffer_flush_interval = 1000000000 -buffer_size_bytes = 16384 -grpc_request_timeout = 20000000000 -server_port = 18090 -shutdown_timeout = "600s" -public_key_path = "" -private_key_path = "" -als_plain_text = true -max_message_size = 1000000000 -max_header_limit = 8192 +# ============================================================================= +# TRAFFIC LOGGING (consumer — enabling it activates the collector) +# ============================================================================= +# Writes collected events (rich API/application/AI/latency metadata, plus +# request/response headers and — when the collector's send_*_body is enabled — +# payloads) to stdout as a JSON line. Useful for log-scraping pipelines such as +# Fluent Bit/Loki/ELK; no external SaaS needed. +# +# Per-API (opt-in): a line is emitted only for APIs that attach the `log-message` +# policy with `enableTrafficLogging: true` — enabling this section alone does NOT +# log every API. So an API is logged only when both hold: [traffic_logging].enabled +# = true, and the `log-message` policy (enableTrafficLogging: true) is attached to +# that API. The policy's request/response parameters (per-flow payload/headers +# toggles and excludeHeaders) shape each API's line; they filter/mask what the +# collector captured (they cannot enable capture the collector was not configured +# to perform). +[traffic_logging] +enabled = false +# Header names (case-insensitive) whose values are redacted as "****" in the +# logged requestHeaders/responseHeaders. Applies globally; the `log-message` +# policy's per-API excludeHeaders drop additional headers entirely. +masked_headers = ["authorization", "x-api-key", "x-jwt-assertion"] +# Max bytes of request/response payload written to each log line (0 = no limit). +# Applied output-side, so the collector still captures full bodies and other +# consumers (e.g. Moesif) are unaffected. +max_payload_size = 0 # ============================================================================= # POLICY CONFIGURATIONS diff --git a/gateway/configs/config.toml b/gateway/configs/config.toml index cd4ea4897f..fbabb75d6d 100644 --- a/gateway/configs/config.toml +++ b/gateway/configs/config.toml @@ -1,16 +1,34 @@ -[analytics] -enabled = true -# Deprecated: allow_payloads is preserved for backward compatibility. When true -# and when send_request_body/send_response_body are not explicitly set, both -# request and response bodies will be included in analytics events. -# allow_payloads = true +# The collector is the shared data-capture pipeline. It has no on/off switch of its +# own: it turns on automatically whenever a consumer (analytics or traffic_logging) +# is enabled, and stays off otherwise. This section only tunes what is captured. +[collector] +# Capture request/response payloads (bodies) into the collected event (full bodies; +# per-consumer size caps are applied on output). send_request_body = false send_response_body = false +# Capture ALL request/response headers (redacted on output by per-consumer masking). +# Off by default; the Moesif publisher does not mask headers, so only enable this +# for consumers you know redact sensitive values (e.g. traffic_logging.masked_headers). +send_request_headers = false +send_response_headers = false + +# Analytics consumer (e.g. Moesif). Enabling it activates the collector automatically. +[analytics] +enabled = false enabled_publishers = ["moesif"] -[analytics.publishers.moesif] -application_id = "" +# [analytics.publishers.moesif] +# application_id = "" + +# Traffic-logging consumer: writes each collected event to stdout as a JSON line +# (no external service needed). Enabling it activates the collector automatically. +[traffic_logging] +enabled = true +# Header names (case-insensitive) whose values are redacted as "****". +masked_headers = ["authorization", "x-api-key", "x-jwt-assertion"] +# Max bytes of request/response payload written to each log line (0 = no limit). +max_payload_size = 2048 [router] gateway_host = "*" diff --git a/gateway/gateway-controller/pkg/config/config.go b/gateway/gateway-controller/pkg/config/config.go index 6402427984..bc4bb7cfd0 100644 --- a/gateway/gateway-controller/pkg/config/config.go +++ b/gateway/gateway-controller/pkg/config/config.go @@ -48,7 +48,9 @@ type Config struct { Router RouterConfig `koanf:"router"` PolicyEngine map[string]interface{} `koanf:"policy_engine"` PolicyConfigurations map[string]interface{} `koanf:"policy_configurations"` + Collector CollectorConfig `koanf:"collector"` Analytics AnalyticsConfig `koanf:"analytics"` + TrafficLogging TrafficLoggingConfig `koanf:"traffic_logging"` TracingConfig TracingConfig `koanf:"tracing"` APIKey APIKeyConfig `koanf:"api_key"` // Subscriptions controls application-level subscription behaviour for APIs. @@ -57,24 +59,60 @@ type Config struct { ImmutableGateway ImmutableGatewayConfig `koanf:"immutable_gateway"` } +// CollectorConfig holds the data-collection ("collector") configuration. The +// collector is the shared capture pipeline (the analytics system policy plus the +// Envoy→policy-engine ALS transport) that gathers request/response headers and +// bodies. It underpins every consumer of that data (analytics and traffic logging) +// and is implicitly active whenever a consumer is enabled — see +// Config.IsCollectorEnabled. This section tunes capture and transport; it has no +// on/off flag of its own. +type CollectorConfig struct { + // SendRequestBody / SendResponseBody capture request/response bodies into the + // collected event. + SendRequestBody bool `koanf:"send_request_body"` + SendResponseBody bool `koanf:"send_response_body"` + // SendRequestHeaders / SendResponseHeaders, when true, make the collector + // capture ALL request / response headers, so every API's headers flow into + // the collected event without attaching a per-API header policy. + SendRequestHeaders bool `koanf:"send_request_headers"` + SendResponseHeaders bool `koanf:"send_response_headers"` + // GRPCEventServerCfg tunes the Envoy→policy-engine gRPC access-log (ALS) + // transport that ships collected data. It is part of the collector and is + // configured under the shared [collector.als] section (the policy-engine reads + // the same section to configure its receiving ALS server). + GRPCEventServerCfg GRPCEventServerConfig `koanf:"als"` +} + // AnalyticsConfig holds analytics configuration type AnalyticsConfig struct { - Enabled bool `koanf:"enabled"` - EnabledPublishers []string `koanf:"enabled_publishers"` - Publishers AnalyticsPublishersConfig `koanf:"publishers"` - GRPCEventServerCfg GRPCEventServerConfig `koanf:"grpc_event_server"` - // AllowPayloads controls whether request and response bodies are captured - // into analytics metadata and forwarded to analytics publishers. - // Deprecated: use SendRequestBody and SendResponseBody instead. - // When true, validateAnalyticsConfig maps both SendRequestBody and SendResponseBody - // to true if both are false. Because bools cannot represent "unset", this also - // applies when both new flags are explicitly false; remove allow_payloads when - // migrating and set the directional flags directly. + Enabled bool `koanf:"enabled"` + EnabledPublishers []string `koanf:"enabled_publishers"` + Publishers AnalyticsPublishersConfig `koanf:"publishers"` + // GRPCEventServerCfg is a deprecated alias. ALS transport tuning moved to + // [collector.als]; when set here it is migrated onto the collector during + // validation (with a warning). Prefer [collector.als]. + GRPCEventServerCfg GRPCEventServerConfig `koanf:"grpc_event_server"` + // AllowPayloads, SendRequestBody and SendResponseBody are deprecated aliases. + // Body/header capture now lives under [collector]. When set, these are mapped + // onto collector.send_request_body / collector.send_response_body during + // validation (with a warning) for backward compatibility. Prefer the + // [collector] fields directly. AllowPayloads bool `koanf:"allow_payloads"` SendRequestBody bool `koanf:"send_request_body"` SendResponseBody bool `koanf:"send_response_body"` } +// TrafficLoggingConfig mirrors the policy-engine's stdout traffic-logging consumer. +// The controller only needs to know whether it is enabled, so that the collector +// (system policy + ALS sink) is activated when traffic logging is on even if +// analytics is off. Presentation keys (masked_headers, max_payload_size) are +// policy-engine-only and intentionally not bound here. +type TrafficLoggingConfig struct { + // Enabled turns stdout JSON traffic logging on. Enabling it implicitly activates + // the collector (see Config.IsCollectorEnabled). + Enabled bool `koanf:"enabled"` +} + // SubscriptionsConfig holds configuration for application-level subscriptions. type SubscriptionsConfig struct { // EnableValidation toggles automatic injection of the subscriptionValidation @@ -690,6 +728,26 @@ func LoadConfig(configPath string) (*Config, error) { return cfg, nil } +// defaultGRPCEventServerConfig returns the default Envoy→policy-engine ALS +// transport tuning. Shared by the collector (canonical) and the deprecated +// [analytics].grpc_event_server alias so a partial alias override migrates cleanly. +func defaultGRPCEventServerConfig() GRPCEventServerConfig { + return GRPCEventServerConfig{ + Mode: "uds", // UDS mode by default + Port: 18090, // Only used in TCP mode + ServerPort: 18090, // ALS server port + BufferFlushInterval: 1000000000, // 1 second + BufferSizeBytes: 16384, // 16 KiB + GRPCRequestTimeout: 20000000000, // 20 seconds + ShutdownTimeout: 600 * time.Second, + PublicKeyPath: "", + PrivateKeyPath: "", + ALSPlainText: true, + MaxMessageSize: 1000000000, + MaxHeaderLimit: 8192, + } +} + // defaultConfig returns a Config struct with default configuration values func defaultConfig() *Config { return &Config{ @@ -908,23 +966,19 @@ func defaultConfig() *Config { TimerWakeupSeconds: 3, }, }, - GRPCEventServerCfg: GRPCEventServerConfig{ - Mode: "uds", // UDS mode by default - Port: 18090, // Only used in TCP mode - ServerPort: 18090, // ALS server port - BufferFlushInterval: 1000000000, // 1 second - BufferSizeBytes: 16384, // 16 KiB - GRPCRequestTimeout: 20000000000, // 20 seconds - ShutdownTimeout: 600 * time.Second, - PublicKeyPath: "", - PrivateKeyPath: "", - ALSPlainText: true, - MaxMessageSize: 1000000000, - MaxHeaderLimit: 8192, - }, - AllowPayloads: false, - SendRequestBody: false, - SendResponseBody: false, + // Deprecated alias: default mirrors the collector so a partial + // [analytics.grpc_event_server] override migrates cleanly. + GRPCEventServerCfg: defaultGRPCEventServerConfig(), + AllowPayloads: false, + SendRequestBody: false, + SendResponseBody: false, + }, + Collector: CollectorConfig{ + SendRequestBody: false, + SendResponseBody: false, + SendRequestHeaders: false, + SendResponseHeaders: false, + GRPCEventServerCfg: defaultGRPCEventServerConfig(), }, TracingConfig: TracingConfig{ Enabled: false, @@ -1291,7 +1345,7 @@ func (c *Config) Validate() error { return err } - if err := c.validateAnalyticsConfig(); err != nil { + if err := c.validateCollectorConfig(); err != nil { return err } @@ -1697,51 +1751,104 @@ func validateDomains(field string, domains []string) error { return nil } -// validateAnalyticsConfig validates the analytics configuration -func (c *Config) validateAnalyticsConfig() error { - // Validate analytics configuration - if c.Analytics.Enabled { - // Migration path for deprecated analytics.allow_payloads. - // Runs when both directional flags are false, which is indistinguishable - // from "not set" because bool fields cannot represent unset vs explicit false. - if c.Analytics.AllowPayloads { - slog.Warn("analytics.allow_payloads is deprecated; use analytics.send_request_body and analytics.send_response_body instead") - if !c.Analytics.SendRequestBody && !c.Analytics.SendResponseBody { - c.Analytics.SendRequestBody = true - c.Analytics.SendResponseBody = true - } +// validateCollectorConfig migrates deprecated analytics aliases onto the collector +// and validates the ALS transport tuning when the collector is active. The collector +// has no on/off flag of its own: it is implicitly active whenever a consumer is +// enabled (analytics or traffic logging) — see IsCollectorEnabled. +func (c *Config) validateCollectorConfig() error { + c.migrateDeprecatedAnalyticsCapture() + c.migrateDeprecatedAnalyticsTransport() + + if c.IsCollectorEnabled() { + if err := validateGRPCEventServerConfig(c.Collector.GRPCEventServerCfg); err != nil { + return err } + } + return nil +} - // Validate gRPC event server configuration - grpcEventServerCfg := c.Analytics.GRPCEventServerCfg +// IsCollectorEnabled reports whether the collector should run. The collector is +// implicit: it is active whenever any consumer of the collected data is enabled +// (analytics or stdout traffic logging), and off otherwise. When active, the +// controller injects the analytics system policy and configures Envoy's ALS sink. +func (c *Config) IsCollectorEnabled() bool { + return c.Analytics.Enabled || c.TrafficLogging.Enabled +} - // Validate connection mode - switch grpcEventServerCfg.Mode { - case "uds", "": - // UDS mode (default) - port is unused for Envoy connection - case "tcp": - // TCP mode - validate port (host is derived from policy_engine.host) - if grpcEventServerCfg.Port <= 0 || grpcEventServerCfg.Port > 65535 { - return fmt.Errorf("analytics.grpc_event_server.port must be between 1 and 65535 when mode is tcp, got %d", grpcEventServerCfg.Port) - } - default: - return fmt.Errorf("analytics.grpc_event_server.mode must be 'uds' or 'tcp', got: %s", grpcEventServerCfg.Mode) +// migrateDeprecatedAnalyticsTransport maps a deprecated [analytics].grpc_event_server +// override onto the collector when the collector's transport tuning is still at its +// default, so existing configs keep working after the transport moved to [collector]. +func (c *Config) migrateDeprecatedAnalyticsTransport() { + def := defaultGRPCEventServerConfig() + if c.Analytics.GRPCEventServerCfg != def { + if c.Collector.GRPCEventServerCfg == def { + slog.Warn("analytics.grpc_event_server is deprecated; migrating it to collector.als") + c.Collector.GRPCEventServerCfg = c.Analytics.GRPCEventServerCfg + } else { + slog.Warn("analytics.grpc_event_server is deprecated and collector.als is already configured; ignoring the analytics.grpc_event_server override") } + } +} - // Validate buffer and timeout settings - if grpcEventServerCfg.BufferFlushInterval <= 0 || grpcEventServerCfg.BufferSizeBytes <= 0 || grpcEventServerCfg.GRPCRequestTimeout <= 0 { - return fmt.Errorf( - "invalid gRPC event server configuration: bufferFlushInterval=%d, bufferSizeBytes=%d, grpcRequestTimeout=%d (all must be > 0)", - grpcEventServerCfg.BufferFlushInterval, - grpcEventServerCfg.BufferSizeBytes, - grpcEventServerCfg.GRPCRequestTimeout, - ) +// migrateDeprecatedAnalyticsCapture maps the deprecated analytics.allow_payloads / +// analytics.send_request_body / analytics.send_response_body onto the collector's +// body-capture flags (when the collector flag is not already set), so existing +// configs keep working after capture settings moved under [collector]. These are +// analytics's own deprecated flags, so they are only honored while analytics is +// enabled — otherwise a stale value left over from a disabled analytics setup +// could silently turn on body capture for an unrelated consumer (e.g. +// traffic_logging) enabled later. +func (c *Config) migrateDeprecatedAnalyticsCapture() { + if !c.Analytics.Enabled { + return + } + // Directional aliases take precedence over allow_payloads. + if c.Analytics.SendRequestBody && !c.Collector.SendRequestBody { + slog.Warn("analytics.send_request_body is deprecated; use collector.send_request_body instead") + c.Collector.SendRequestBody = true + } + if c.Analytics.SendResponseBody && !c.Collector.SendResponseBody { + slog.Warn("analytics.send_response_body is deprecated; use collector.send_response_body instead") + c.Collector.SendResponseBody = true + } + // allow_payloads only fills in when no directional body capture is configured. + if c.Analytics.AllowPayloads { + slog.Warn("analytics.allow_payloads is deprecated; use collector.send_request_body and collector.send_response_body instead") + if !c.Collector.SendRequestBody && !c.Collector.SendResponseBody { + c.Collector.SendRequestBody = true + c.Collector.SendResponseBody = true } + } +} - // Validate server port - if grpcEventServerCfg.ServerPort <= 0 || grpcEventServerCfg.ServerPort > 65535 { - return fmt.Errorf("analytics.grpc_event_server.server_port must be between 1 and 65535, got %d", grpcEventServerCfg.ServerPort) +// validateGRPCEventServerConfig validates the Envoy→policy-engine ALS transport tuning. +func validateGRPCEventServerConfig(cfg GRPCEventServerConfig) error { + // Validate connection mode + switch cfg.Mode { + case "uds", "": + // UDS mode (default) - port is unused for Envoy connection + case "tcp": + // TCP mode - validate port (host is derived from policy_engine.host) + if cfg.Port <= 0 || cfg.Port > 65535 { + return fmt.Errorf("collector.als.port must be between 1 and 65535 when mode is tcp, got %d", cfg.Port) } + default: + return fmt.Errorf("collector.als.mode must be 'uds' or 'tcp', got: %s", cfg.Mode) + } + + // Validate buffer and timeout settings + if cfg.BufferFlushInterval <= 0 || cfg.BufferSizeBytes <= 0 || cfg.GRPCRequestTimeout <= 0 { + return fmt.Errorf( + "invalid gRPC event server configuration: bufferFlushInterval=%d, bufferSizeBytes=%d, grpcRequestTimeout=%d (all must be > 0)", + cfg.BufferFlushInterval, + cfg.BufferSizeBytes, + cfg.GRPCRequestTimeout, + ) + } + + // Validate server port + if cfg.ServerPort <= 0 || cfg.ServerPort > 65535 { + return fmt.Errorf("collector.als.server_port must be between 1 and 65535, got %d", cfg.ServerPort) } return nil } diff --git a/gateway/gateway-controller/pkg/config/config_test.go b/gateway/gateway-controller/pkg/config/config_test.go index 91e53733a4..7c4b117be9 100644 --- a/gateway/gateway-controller/pkg/config/config_test.go +++ b/gateway/gateway-controller/pkg/config/config_test.go @@ -102,6 +102,14 @@ func validConfig() *Config { ServerHeaderTransformation: commonconstants.OVERWRITE, }, }, + // Transport defaults mirror production so collector-gated grpc validation + // passes and the deprecated [analytics] alias stays neutral. + Collector: CollectorConfig{ + GRPCEventServerCfg: defaultGRPCEventServerConfig(), + }, + Analytics: AnalyticsConfig{ + GRPCEventServerCfg: defaultGRPCEventServerConfig(), + }, } } @@ -1205,11 +1213,11 @@ func TestConfig_ValidateAnalyticsConfig(t *testing.T) { name: "Analytics enabled with valid UDS config (default mode)", enabled: true, setupConfig: func(cfg *Config) { - cfg.Analytics.GRPCEventServerCfg.Mode = "uds" - cfg.Analytics.GRPCEventServerCfg.BufferFlushInterval = 1000 - cfg.Analytics.GRPCEventServerCfg.BufferSizeBytes = 16384 - cfg.Analytics.GRPCEventServerCfg.GRPCRequestTimeout = 5000 - cfg.Analytics.GRPCEventServerCfg.ServerPort = 18090 + cfg.Collector.GRPCEventServerCfg.Mode = "uds" + cfg.Collector.GRPCEventServerCfg.BufferFlushInterval = 1000 + cfg.Collector.GRPCEventServerCfg.BufferSizeBytes = 16384 + cfg.Collector.GRPCEventServerCfg.GRPCRequestTimeout = 5000 + cfg.Collector.GRPCEventServerCfg.ServerPort = 18090 }, wantErr: false, }, @@ -1217,12 +1225,12 @@ func TestConfig_ValidateAnalyticsConfig(t *testing.T) { name: "Analytics enabled with valid TCP config", enabled: true, setupConfig: func(cfg *Config) { - cfg.Analytics.GRPCEventServerCfg.Mode = "tcp" - cfg.Analytics.GRPCEventServerCfg.Port = 18090 - cfg.Analytics.GRPCEventServerCfg.BufferFlushInterval = 1000 - cfg.Analytics.GRPCEventServerCfg.BufferSizeBytes = 16384 - cfg.Analytics.GRPCEventServerCfg.GRPCRequestTimeout = 5000 - cfg.Analytics.GRPCEventServerCfg.ServerPort = 18090 + cfg.Collector.GRPCEventServerCfg.Mode = "tcp" + cfg.Collector.GRPCEventServerCfg.Port = 18090 + cfg.Collector.GRPCEventServerCfg.BufferFlushInterval = 1000 + cfg.Collector.GRPCEventServerCfg.BufferSizeBytes = 16384 + cfg.Collector.GRPCEventServerCfg.GRPCRequestTimeout = 5000 + cfg.Collector.GRPCEventServerCfg.ServerPort = 18090 }, wantErr: false, }, @@ -1230,11 +1238,11 @@ func TestConfig_ValidateAnalyticsConfig(t *testing.T) { name: "Analytics enabled with empty mode defaults to UDS", enabled: true, setupConfig: func(cfg *Config) { - cfg.Analytics.GRPCEventServerCfg.Mode = "" - cfg.Analytics.GRPCEventServerCfg.BufferFlushInterval = 1000 - cfg.Analytics.GRPCEventServerCfg.BufferSizeBytes = 16384 - cfg.Analytics.GRPCEventServerCfg.GRPCRequestTimeout = 5000 - cfg.Analytics.GRPCEventServerCfg.ServerPort = 18090 + cfg.Collector.GRPCEventServerCfg.Mode = "" + cfg.Collector.GRPCEventServerCfg.BufferFlushInterval = 1000 + cfg.Collector.GRPCEventServerCfg.BufferSizeBytes = 16384 + cfg.Collector.GRPCEventServerCfg.GRPCRequestTimeout = 5000 + cfg.Collector.GRPCEventServerCfg.ServerPort = 18090 }, wantErr: false, }, @@ -1242,49 +1250,49 @@ func TestConfig_ValidateAnalyticsConfig(t *testing.T) { name: "Invalid mode value", enabled: true, setupConfig: func(cfg *Config) { - cfg.Analytics.GRPCEventServerCfg.Mode = "invalid" - cfg.Analytics.GRPCEventServerCfg.BufferFlushInterval = 1000 - cfg.Analytics.GRPCEventServerCfg.BufferSizeBytes = 16384 - cfg.Analytics.GRPCEventServerCfg.GRPCRequestTimeout = 5000 - cfg.Analytics.GRPCEventServerCfg.ServerPort = 18090 + cfg.Collector.GRPCEventServerCfg.Mode = "invalid" + cfg.Collector.GRPCEventServerCfg.BufferFlushInterval = 1000 + cfg.Collector.GRPCEventServerCfg.BufferSizeBytes = 16384 + cfg.Collector.GRPCEventServerCfg.GRPCRequestTimeout = 5000 + cfg.Collector.GRPCEventServerCfg.ServerPort = 18090 }, wantErr: true, - errContains: "grpc_event_server.mode must be 'uds' or 'tcp'", + errContains: "collector.als.mode must be 'uds' or 'tcp'", }, { name: "TCP mode - invalid port", enabled: true, setupConfig: func(cfg *Config) { - cfg.Analytics.GRPCEventServerCfg.Mode = "tcp" - cfg.Analytics.GRPCEventServerCfg.Port = 0 - cfg.Analytics.GRPCEventServerCfg.BufferFlushInterval = 1000 - cfg.Analytics.GRPCEventServerCfg.BufferSizeBytes = 16384 - cfg.Analytics.GRPCEventServerCfg.GRPCRequestTimeout = 5000 - cfg.Analytics.GRPCEventServerCfg.ServerPort = 18090 + cfg.Collector.GRPCEventServerCfg.Mode = "tcp" + cfg.Collector.GRPCEventServerCfg.Port = 0 + cfg.Collector.GRPCEventServerCfg.BufferFlushInterval = 1000 + cfg.Collector.GRPCEventServerCfg.BufferSizeBytes = 16384 + cfg.Collector.GRPCEventServerCfg.GRPCRequestTimeout = 5000 + cfg.Collector.GRPCEventServerCfg.ServerPort = 18090 }, wantErr: true, - errContains: "grpc_event_server.port must be between 1 and 65535", + errContains: "collector.als.port must be between 1 and 65535", }, { name: "Invalid server port", enabled: true, setupConfig: func(cfg *Config) { - cfg.Analytics.GRPCEventServerCfg.Mode = "uds" - cfg.Analytics.GRPCEventServerCfg.BufferFlushInterval = 1000 - cfg.Analytics.GRPCEventServerCfg.BufferSizeBytes = 16384 - cfg.Analytics.GRPCEventServerCfg.GRPCRequestTimeout = 5000 - cfg.Analytics.GRPCEventServerCfg.ServerPort = 0 + cfg.Collector.GRPCEventServerCfg.Mode = "uds" + cfg.Collector.GRPCEventServerCfg.BufferFlushInterval = 1000 + cfg.Collector.GRPCEventServerCfg.BufferSizeBytes = 16384 + cfg.Collector.GRPCEventServerCfg.GRPCRequestTimeout = 5000 + cfg.Collector.GRPCEventServerCfg.ServerPort = 0 }, wantErr: true, - errContains: "grpc_event_server.server_port must be between 1 and 65535", + errContains: "collector.als.server_port must be between 1 and 65535", }, { name: "Invalid buffer flush interval", enabled: true, setupConfig: func(cfg *Config) { - cfg.Analytics.GRPCEventServerCfg.Mode = "uds" - cfg.Analytics.GRPCEventServerCfg.BufferFlushInterval = 0 - cfg.Analytics.GRPCEventServerCfg.ServerPort = 18090 + cfg.Collector.GRPCEventServerCfg.Mode = "uds" + cfg.Collector.GRPCEventServerCfg.BufferFlushInterval = 0 + cfg.Collector.GRPCEventServerCfg.ServerPort = 18090 }, wantErr: true, errContains: "invalid gRPC event server configuration", @@ -1294,6 +1302,8 @@ func TestConfig_ValidateAnalyticsConfig(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { cfg := validConfig() + // Analytics is a consumer; enabling it makes the collector implicitly + // active so these tests exercise the collector transport validation. cfg.Analytics.Enabled = tt.enabled if tt.setupConfig != nil { tt.setupConfig(cfg) @@ -1309,9 +1319,34 @@ func TestConfig_ValidateAnalyticsConfig(t *testing.T) { } } +// TestConfig_IsCollectorEnabled covers the implicit collector: it is active iff a +// consumer (analytics or traffic logging) is enabled, and off otherwise. +func TestConfig_IsCollectorEnabled(t *testing.T) { + t.Run("no consumers -> off", func(t *testing.T) { + cfg := validConfig() + assert.False(t, cfg.IsCollectorEnabled()) + require.NoError(t, cfg.Validate()) + }) + + t.Run("analytics on -> collector on", func(t *testing.T) { + cfg := validConfig() + cfg.Analytics.Enabled = true + cfg.Analytics.EnabledPublishers = []string{} + assert.True(t, cfg.IsCollectorEnabled()) + require.NoError(t, cfg.Validate()) + }) + + t.Run("traffic logging on -> collector on", func(t *testing.T) { + cfg := validConfig() + cfg.TrafficLogging.Enabled = true + assert.True(t, cfg.IsCollectorEnabled()) + require.NoError(t, cfg.Validate()) + }) +} + func TestConfig_ValidateAnalyticsPayloadMigration(t *testing.T) { setValidAnalyticsGRPC := func(cfg *Config) { - cfg.Analytics.Enabled = true + cfg.Analytics.Enabled = true // a consumer being on makes the collector implicit cfg.Analytics.GRPCEventServerCfg.Mode = "uds" cfg.Analytics.GRPCEventServerCfg.BufferFlushInterval = 1000 cfg.Analytics.GRPCEventServerCfg.BufferSizeBytes = 16384 @@ -1371,12 +1406,30 @@ func TestConfig_ValidateAnalyticsPayloadMigration(t *testing.T) { err := cfg.Validate() require.NoError(t, err) - assert.Equal(t, tt.wantSendReq, cfg.Analytics.SendRequestBody) - assert.Equal(t, tt.wantSendResp, cfg.Analytics.SendResponseBody) + // Deprecated analytics body aliases now migrate onto the collector. + assert.Equal(t, tt.wantSendReq, cfg.Collector.SendRequestBody) + assert.Equal(t, tt.wantSendResp, cfg.Collector.SendResponseBody) }) } } +// TestConfig_ValidateAnalyticsPayloadMigration_SkippedWhenAnalyticsDisabled guards +// against a stale analytics.allow_payloads left over from a disabled analytics +// setup silently turning on body capture for an unrelated consumer (traffic +// logging) enabled later. The deprecated capture aliases belong to analytics, so +// they must only be honored while analytics itself is enabled. +func TestConfig_ValidateAnalyticsPayloadMigration_SkippedWhenAnalyticsDisabled(t *testing.T) { + cfg := validConfig() + cfg.Analytics.Enabled = false + cfg.TrafficLogging.Enabled = true // an unrelated consumer activates the collector + cfg.Analytics.AllowPayloads = true + + err := cfg.Validate() + require.NoError(t, err) + assert.False(t, cfg.Collector.SendRequestBody) + assert.False(t, cfg.Collector.SendResponseBody) +} + func TestConfig_ValidateAuthConfig(t *testing.T) { tests := []struct { name string diff --git a/gateway/gateway-controller/pkg/utils/system_policies.go b/gateway/gateway-controller/pkg/utils/system_policies.go index 8ad1da067f..8985f53388 100644 --- a/gateway/gateway-controller/pkg/utils/system_policies.go +++ b/gateway/gateway-controller/pkg/utils/system_policies.go @@ -84,14 +84,15 @@ var defaultSystemPolicies = []systemPolicyConfig{ if cfg == nil { return false } - slog.Debug("Analytics state -> ", "state", cfg.Analytics.Enabled) - return cfg.Analytics.Enabled - }, - // Default parameters (can be overridden via additionalProps) - Parameters: map[string]interface{}{ - "send_request_body": false, - "send_response_body": false, + // The analytics system policy is the collector: it is injected whenever + // the collector is active — i.e. whenever any consumer (analytics or + // traffic logging) is enabled. + slog.Debug("Collector state -> ", "state", cfg.IsCollectorEnabled()) + return cfg.IsCollectorEnabled() }, + // No static defaults — all four capture flags are set at injection time + // from cfg.Collector (see InjectSystemPolicies below). + Parameters: nil, ExecutionCondition: nil, }, } @@ -197,10 +198,13 @@ func InjectSystemPolicies(policies []policyenginev1.PolicyInstance, cfg *config. for k, v := range sysPol.Parameters { effectiveDefaults[k] = v } - // For the analytics system policy, propagate the payload flags from runtime config. + // For the analytics (collector) system policy, propagate the payload and + // header capture flags from the collector config. if sysPol.Name == constants.ANALYTICS_SYSTEM_POLICY_NAME { - effectiveDefaults["send_request_body"] = cfg.Analytics.SendRequestBody - effectiveDefaults["send_response_body"] = cfg.Analytics.SendResponseBody + effectiveDefaults["send_request_body"] = cfg.Collector.SendRequestBody + effectiveDefaults["send_response_body"] = cfg.Collector.SendResponseBody + effectiveDefaults["send_request_headers"] = cfg.Collector.SendRequestHeaders + effectiveDefaults["send_response_headers"] = cfg.Collector.SendResponseHeaders } // Merge parameters efficiently diff --git a/gateway/gateway-controller/pkg/utils/system_policies_test.go b/gateway/gateway-controller/pkg/utils/system_policies_test.go index d0a0f1a61e..8ed2e284cb 100644 --- a/gateway/gateway-controller/pkg/utils/system_policies_test.go +++ b/gateway/gateway-controller/pkg/utils/system_policies_test.go @@ -124,12 +124,9 @@ func TestInjectSystemPolicies_NilConfig(t *testing.T) { assert.Equal(t, policies, result) } -func TestInjectSystemPolicies_AnalyticsDisabled(t *testing.T) { - cfg := &config.Config{ - Analytics: config.AnalyticsConfig{ - Enabled: false, - }, - } +func TestInjectSystemPolicies_CollectorDisabled(t *testing.T) { + // No consumer enabled -> collector implicitly off -> no system policy injected. + cfg := &config.Config{} policies := []policyenginev1.PolicyInstance{ {Name: "existing", Version: "v1.0.0"}, } @@ -139,11 +136,10 @@ func TestInjectSystemPolicies_AnalyticsDisabled(t *testing.T) { assert.Equal(t, "existing", result[0].Name) } -func TestInjectSystemPolicies_AnalyticsEnabled(t *testing.T) { +func TestInjectSystemPolicies_CollectorEnabled(t *testing.T) { + // A consumer enabled -> collector implicitly on -> system policy injected. cfg := &config.Config{ - Analytics: config.AnalyticsConfig{ - Enabled: true, - }, + Analytics: config.AnalyticsConfig{Enabled: true}, } policies := []policyenginev1.PolicyInstance{ {Name: "existing", Version: "v1.0.0"}, @@ -159,11 +155,30 @@ func TestInjectSystemPolicies_AnalyticsEnabled(t *testing.T) { assert.Equal(t, "existing", result[1].Name) } -func TestInjectSystemPolicies_AllowPayloadsTrue(t *testing.T) { +func TestInjectSystemPolicies_TrafficLoggingOnlyEnablesCollector(t *testing.T) { + // Traffic logging alone (no analytics) -> collector implicitly on -> system + // policy injected. Guards against a regression that gates injection on + // cfg.Analytics.Enabled alone instead of cfg.IsCollectorEnabled(). cfg := &config.Config{ - Analytics: config.AnalyticsConfig{ - Enabled: true, - SendRequestBody: true, + TrafficLogging: config.TrafficLoggingConfig{Enabled: true}, + } + policies := []policyenginev1.PolicyInstance{ + {Name: "existing", Version: "v1.0.0"}, + } + + result := InjectSystemPolicies(policies, cfg, nil) + assert.Len(t, result, 2) + assert.Equal(t, constants.ANALYTICS_SYSTEM_POLICY_NAME, result[0].Name) + assert.Equal(t, constants.ANALYTICS_SYSTEM_POLICY_VERSION, result[0].Version) + assert.True(t, result[0].Enabled) + assert.Equal(t, "existing", result[1].Name) +} + +func TestInjectSystemPolicies_BodyFlagsPropagated(t *testing.T) { + cfg := &config.Config{ + Analytics: config.AnalyticsConfig{Enabled: true}, + Collector: config.CollectorConfig{ + SendRequestBody: true, SendResponseBody: true, }, } @@ -175,11 +190,11 @@ func TestInjectSystemPolicies_AllowPayloadsTrue(t *testing.T) { assert.Equal(t, true, result[0].Parameters["send_response_body"]) } -func TestInjectSystemPolicies_AllowPayloadsFalse(t *testing.T) { +func TestInjectSystemPolicies_BodyFlagsDefaultFalse(t *testing.T) { cfg := &config.Config{ - Analytics: config.AnalyticsConfig{ - Enabled: true, - SendRequestBody: false, + Analytics: config.AnalyticsConfig{Enabled: true}, + Collector: config.CollectorConfig{ + SendRequestBody: false, SendResponseBody: false, }, } @@ -190,12 +205,39 @@ func TestInjectSystemPolicies_AllowPayloadsFalse(t *testing.T) { assert.Equal(t, false, result[0].Parameters["send_response_body"]) } -func TestInjectSystemPolicies_WithAdditionalProps(t *testing.T) { +func TestInjectSystemPolicies_HeaderFlagsPropagated(t *testing.T) { cfg := &config.Config{ - Analytics: config.AnalyticsConfig{ - Enabled: true, + Analytics: config.AnalyticsConfig{Enabled: true}, + Collector: config.CollectorConfig{ + SendRequestHeaders: true, + SendResponseHeaders: true, }, } + + result := InjectSystemPolicies(nil, cfg, nil) + assert.Len(t, result, 1) + assert.Equal(t, constants.ANALYTICS_SYSTEM_POLICY_NAME, result[0].Name) + assert.Equal(t, true, result[0].Parameters["send_request_headers"]) + assert.Equal(t, true, result[0].Parameters["send_response_headers"]) +} + +func TestInjectSystemPolicies_HeaderFlagsDefaultFalse(t *testing.T) { + // Zero-value collector (headers unset) with a consumer on: propagation passes the + // struct's false through, matching the production default (see defaultConfig). + cfg := &config.Config{ + Analytics: config.AnalyticsConfig{Enabled: true}, + } + + result := InjectSystemPolicies(nil, cfg, nil) + assert.Len(t, result, 1) + assert.Equal(t, false, result[0].Parameters["send_request_headers"]) + assert.Equal(t, false, result[0].Parameters["send_response_headers"]) +} + +func TestInjectSystemPolicies_WithAdditionalProps(t *testing.T) { + cfg := &config.Config{ + Analytics: config.AnalyticsConfig{Enabled: true}, + } additionalProps := map[string]any{ constants.ANALYTICS_SYSTEM_POLICY_NAME: map[string]interface{}{ "custom_param": "custom_value", @@ -209,9 +251,7 @@ func TestInjectSystemPolicies_WithAdditionalProps(t *testing.T) { func TestInjectSystemPolicies_WithSharedParams(t *testing.T) { cfg := &config.Config{ - Analytics: config.AnalyticsConfig{ - Enabled: true, - }, + Analytics: config.AnalyticsConfig{Enabled: true}, } additionalProps := map[string]any{ SharedParamsKey: map[string]interface{}{ @@ -226,9 +266,7 @@ func TestInjectSystemPolicies_WithSharedParams(t *testing.T) { func TestInjectSystemPolicies_EmptyPolicies(t *testing.T) { cfg := &config.Config{ - Analytics: config.AnalyticsConfig{ - Enabled: true, - }, + Analytics: config.AnalyticsConfig{Enabled: true}, } result := InjectSystemPolicies([]policyenginev1.PolicyInstance{}, cfg, nil) @@ -238,9 +276,7 @@ func TestInjectSystemPolicies_EmptyPolicies(t *testing.T) { func TestInjectSystemPolicies_PreservesExistingPolicies(t *testing.T) { cfg := &config.Config{ - Analytics: config.AnalyticsConfig{ - Enabled: true, - }, + Analytics: config.AnalyticsConfig{Enabled: true}, } policies := []policyenginev1.PolicyInstance{ {Name: "policy1", Version: "v1.0.0"}, diff --git a/gateway/gateway-controller/pkg/xds/translator.go b/gateway/gateway-controller/pkg/xds/translator.go index 408a6dda7a..f0b5708d8f 100644 --- a/gateway/gateway-controller/pkg/xds/translator.go +++ b/gateway/gateway-controller/pkg/xds/translator.go @@ -561,10 +561,10 @@ func (t *Translator) TranslateConfigs( policyEngineCluster := t.createPolicyEngineCluster() clusters = append(clusters, policyEngineCluster) - // Add ALS cluster if gRPC access log is enabled - log.Debug("gRPC event server config", slog.Any("config", t.config.Analytics.GRPCEventServerCfg)) - if t.config.Analytics.Enabled { - log.Info("gRPC access log is enabled, creating ALS cluster") + // Add ALS cluster if the collector is active (it ships access logs over gRPC) + log.Debug("gRPC event server config", slog.Any("config", t.config.Collector.GRPCEventServerCfg)) + if t.config.IsCollectorEnabled() { + log.Info("collector is enabled, creating ALS cluster") alsCluster := t.createALSCluster() clusters = append(clusters, alsCluster) } @@ -2156,7 +2156,7 @@ func (t *Translator) createPolicyEngineCluster() *cluster.Cluster { // createALSCluster creates an Envoy cluster for the gRPC access log service func (t *Translator) createALSCluster() *cluster.Cluster { - grpcConfig := t.config.Analytics.GRPCEventServerCfg + grpcConfig := t.config.Collector.GRPCEventServerCfg // Build the endpoint address (UDS or TCP) var address *core.Address @@ -2785,8 +2785,8 @@ func (t *Translator) createAccessLogConfig() ([]*accesslog.AccessLog, error) { }, }) - // If gRPC access log is enabled, create the configuration and append to existing access logs - if t.config.Analytics.Enabled { + // If the collector is active, create the gRPC access log config and append to existing access logs + if t.config.IsCollectorEnabled() { t.logger.Info("Creating gRPC access log configuration") grpcAccessLog, err := t.createGRPCAccessLog() if err != nil { @@ -2802,8 +2802,8 @@ func (t *Translator) createAccessLogConfig() ([]*accesslog.AccessLog, error) { // createGRPCAccessLog creates a gRPC access log configuration for the gateway controller func (t *Translator) createGRPCAccessLog() (*accesslog.AccessLog, error) { - grpcConfig := t.config.Analytics.GRPCEventServerCfg - bufferSizeBytes, err := checkedUInt32FromPositiveInt("analytics.grpc_event_server.buffer_size_bytes", grpcConfig.BufferSizeBytes) + grpcConfig := t.config.Collector.GRPCEventServerCfg + bufferSizeBytes, err := checkedUInt32FromPositiveInt("collector.als.buffer_size_bytes", grpcConfig.BufferSizeBytes) if err != nil { return nil, err } diff --git a/gateway/gateway-controller/pkg/xds/translator_test.go b/gateway/gateway-controller/pkg/xds/translator_test.go index dc637583e8..4a99e0036d 100644 --- a/gateway/gateway-controller/pkg/xds/translator_test.go +++ b/gateway/gateway-controller/pkg/xds/translator_test.go @@ -1330,7 +1330,7 @@ func TestTranslator_CreateALSCluster(t *testing.T) { routerCfg := testRouterConfig() cfg := testConfig() cfg.Analytics.Enabled = true - cfg.Analytics.GRPCEventServerCfg = config.GRPCEventServerConfig{ + cfg.Collector.GRPCEventServerCfg = config.GRPCEventServerConfig{ Mode: "uds", BufferFlushInterval: 1000000000, BufferSizeBytes: 16384, @@ -1357,7 +1357,7 @@ func TestTranslator_CreateALSCluster(t *testing.T) { routerCfg := testRouterConfig() cfg := testConfig() cfg.Analytics.Enabled = true - cfg.Analytics.GRPCEventServerCfg = config.GRPCEventServerConfig{ + cfg.Collector.GRPCEventServerCfg = config.GRPCEventServerConfig{ Mode: "", BufferFlushInterval: 1000000000, BufferSizeBytes: 16384, @@ -1383,7 +1383,7 @@ func TestTranslator_CreateALSCluster(t *testing.T) { routerCfg := testRouterConfig() cfg := testConfig() cfg.Analytics.Enabled = true - cfg.Analytics.GRPCEventServerCfg = config.GRPCEventServerConfig{ + cfg.Collector.GRPCEventServerCfg = config.GRPCEventServerConfig{ Mode: "tcp", Port: 18090, BufferFlushInterval: 1000000000, @@ -1415,7 +1415,7 @@ func TestTranslator_CreateGRPCAccessLog(t *testing.T) { logger := createTestLogger() routerCfg := testRouterConfig() cfg := testConfig() - cfg.Analytics.GRPCEventServerCfg = config.GRPCEventServerConfig{ + cfg.Collector.GRPCEventServerCfg = config.GRPCEventServerConfig{ Mode: "tcp", Port: 18090, BufferFlushInterval: 1000, @@ -1433,7 +1433,7 @@ func TestTranslator_CreateGRPCAccessLog_BufferSizeOverflow(t *testing.T) { logger := createTestLogger() routerCfg := testRouterConfig() cfg := testConfig() - cfg.Analytics.GRPCEventServerCfg = config.GRPCEventServerConfig{ + cfg.Collector.GRPCEventServerCfg = config.GRPCEventServerConfig{ Mode: "tcp", Port: 18090, BufferFlushInterval: 1000, diff --git a/gateway/gateway-runtime/policy-engine/cmd/policy-engine/main.go b/gateway/gateway-runtime/policy-engine/cmd/policy-engine/main.go index 031f527745..1281a7a519 100644 --- a/gateway/gateway-runtime/policy-engine/cmd/policy-engine/main.go +++ b/gateway/gateway-runtime/policy-engine/cmd/policy-engine/main.go @@ -268,10 +268,12 @@ func main() { metrics.StartMemoryMetricsUpdater(ctx, 15*time.Second) } - // Start access log service server if enabled + // Start the access log service server when the collector is enabled. The + // collector is the shared transport that carries collected data to its + // consumers (analytics, traffic logging). var alsServer *grpc.Server - slog.DebugContext(ctx, "Policy engine ALS server config", "config", cfg.Analytics.AccessLogsServiceCfg) - if cfg.Analytics.Enabled { + slog.DebugContext(ctx, "Policy engine ALS server config", "config", cfg.Collector.AccessLogsServiceCfg) + if cfg.IsCollectorEnabled() { // Start the access log service server slog.Info("Starting the ALS gRPC server...") alsServer = utils.StartAccessLogServiceServer(cfg) diff --git a/gateway/gateway-runtime/policy-engine/internal/analytics/analytics.go b/gateway/gateway-runtime/policy-engine/internal/analytics/analytics.go index 56d541c190..da8820fafb 100644 --- a/gateway/gateway-runtime/policy-engine/internal/analytics/analytics.go +++ b/gateway/gateway-runtime/policy-engine/internal/analytics/analytics.go @@ -23,6 +23,7 @@ import ( "log/slog" "maps" "strconv" + "sync" "time" v3 "github.com/envoyproxy/go-control-plane/envoy/data/accesslog/v3" @@ -64,6 +65,12 @@ const ( RequestHeadersKey = "request_headers" ResponseHeadersKey = "response_headers" + // TrafficLogMetadataKey is the analytics-metadata key under which the + // log-message policy (in access-log mode) stamps its per-API opt-in marker. + // Its presence gates the stdout traffic-logging publisher; its value (a JSON + // directive) shapes the emitted line. Must match the key used by that policy. + TrafficLogMetadataKey = "traffic_log" + // PromptTokenCountMetadataKey represents the prompt token count metadata key. PromptTokenCountMetadataKey string = "aitoken:prompttokencount" // CompletionTokenCountMetadataKey represents the completion token count metadata key. @@ -88,16 +95,24 @@ type Analytics struct { cfg *config.Config // publishers represents the publishers. publishers []analytics_publisher.Publisher + // directiveCache caches parsed TrafficLogDirectives keyed by raw JSON string. + // The directive is static per API deployment, so this eliminates per-request + // allocations after the first parse. + directiveCache sync.Map } -// NewAnalytics creates a new instance of Analytics. +// NewAnalytics creates a new instance of Analytics. Publishers are assembled from +// each independently-configured consumer of the collected data: the analytics +// consumer ([analytics], e.g. Moesif) and the traffic-logging consumer +// ([traffic_logging], stdout JSON). Both rely on the collector being enabled to +// receive any events. func NewAnalytics(cfg *config.Config) *Analytics { analyticsCfg := cfg.Analytics publishers := make([]analytics_publisher.Publisher, 0) if analyticsCfg.Enabled { for _, publisherName := range analyticsCfg.EnabledPublishers { switch publisherName { - case "moesif": + case MoesifAnalyticsPublisher: publisher := analytics_publisher.NewMoesif(&analyticsCfg.Publishers.Moesif) if publisher != nil { publishers = append(publishers, publisher) @@ -109,8 +124,14 @@ func NewAnalytics(cfg *config.Config) *Analytics { } } + // Traffic logging is a standalone consumer, independent of analytics. + if cfg.TrafficLogging.Enabled { + publishers = append(publishers, analytics_publisher.NewLog(&cfg.TrafficLogging)) + slog.Info("Traffic logging (stdout) publisher added") + } + if len(publishers) == 0 { - slog.Debug("No analytics publishers found. Analytics will not be published.") + slog.Debug("No analytics publishers found. Collected events will not be published.") } return &Analytics{ cfg: cfg, @@ -133,7 +154,6 @@ func (c *Analytics) Process(event *v3.HTTPAccessLogEntry) { return } - // Add logic to publish the event analyticEvent := c.prepareAnalyticEvent(event) for _, publisher := range c.publishers { publisher.Publish(analyticEvent) @@ -197,6 +217,17 @@ func (c *Analytics) prepareAnalyticEvent(logEntry *v3.HTTPAccessLogEntry) *dto.E for key, value := range keyValuePairsFromMetadata { slog.Debug(fmt.Sprintf("Metadata key: %v -> value: %+v", key, value)) } + + // Extract the per-API traffic-log opt-in marker stamped by the log-message + // policy (access-log mode). Its presence marks the API as opted in to stdout + // traffic logging; + // its value shapes the emitted line. It is intentionally kept off + // event.Properties so it never reaches the serialized output or other + // publishers (e.g. Moesif). A malformed marker still opts the API in with + // default (empty) presentation, since attachment alone signals intent. + if raw, exists := keyValuePairsFromMetadata[TrafficLogMetadataKey]; exists { + event.TrafficLog = c.parseTrafficLogDirective(raw) + } // Prepare extended API extendedAPI := dto.ExtendedAPI{} extendedAPI.APIType = keyValuePairsFromMetadata[APITypeKey] @@ -247,33 +278,36 @@ func (c *Analytics) prepareAnalyticEvent(logEntry *v3.HTTPAccessLogEntry) *dto.E properties.TimeToFirstUpstreamTxByte != nil && properties.TimeToFirstUpstreamRxByte != nil && properties.TimeToLastUpstreamRxByte != nil && properties.TimeToLastDownstreamTxByte != nil { - lastRx := - (properties.TimeToLastRxByte.Seconds * 1000) + - (int64(properties.TimeToLastRxByte.Nanos) / 1_000_000) - - firstUpTx := - (properties.TimeToFirstUpstreamTxByte.Seconds * 1000) + - (int64(properties.TimeToFirstUpstreamTxByte.Nanos) / 1_000_000) - - firstUpRx := - (properties.TimeToFirstUpstreamRxByte.Seconds * 1000) + - (int64(properties.TimeToFirstUpstreamRxByte.Nanos) / 1_000_000) - - lastUpRx := - (properties.TimeToLastUpstreamRxByte.Seconds * 1000) + - (int64(properties.TimeToLastUpstreamRxByte.Nanos) / 1_000_000) + toMs := func(secs int64, nanos int32) int64 { + return (secs * 1000) + int64(nanos)/1_000_000 + } - lastDownTx := - (properties.TimeToLastDownstreamTxByte.Seconds * 1000) + - (int64(properties.TimeToLastDownstreamTxByte.Nanos) / 1_000_000) + lastRx := toMs(properties.TimeToLastRxByte.Seconds, properties.TimeToLastRxByte.Nanos) + firstUpTx := toMs(properties.TimeToFirstUpstreamTxByte.Seconds, properties.TimeToFirstUpstreamTxByte.Nanos) + firstUpRx := toMs(properties.TimeToFirstUpstreamRxByte.Seconds, properties.TimeToFirstUpstreamRxByte.Nanos) + lastUpRx := toMs(properties.TimeToLastUpstreamRxByte.Seconds, properties.TimeToLastUpstreamRxByte.Nanos) + lastDownTx := toMs(properties.TimeToLastDownstreamTxByte.Seconds, properties.TimeToLastDownstreamTxByte.Nanos) latencies := dto.Latencies{ + Duration: lastDownTx, BackendLatency: lastUpRx - firstUpTx, RequestMediationLatency: firstUpTx - lastRx, ResponseLatency: lastDownTx - firstUpRx, ResponseMediationLatency: lastDownTx - lastUpRx, } + // US_TX_END → US_RX_BEG: time the backend spent before sending the first response byte (TTFB). + if properties.TimeToLastUpstreamTxByte != nil { + lastUpTx := toMs(properties.TimeToLastUpstreamTxByte.Seconds, properties.TimeToLastUpstreamTxByte.Nanos) + latencies.BackendProcDuration = firstUpRx - lastUpTx + } + + // US_RX_BEG → DS_TX_BEG: gateway overhead processing the first response byte before writing downstream. + if properties.TimeToFirstDownstreamTxByte != nil { + firstDownTx := toMs(properties.TimeToFirstDownstreamTxByte.Seconds, properties.TimeToFirstDownstreamTxByte.Nanos) + latencies.ResponseProcDuration = firstDownTx - firstUpRx + } + event.Latencies = &latencies } @@ -428,22 +462,22 @@ func (c *Analytics) prepareAnalyticEvent(logEntry *v3.HTTPAccessLogEntry) *dto.E //Adding request and response headers for the analytics event if requestHeaders, exists := keyValuePairsFromMetadata[RequestHeadersKey]; exists { - event.Properties["requestHeaders"] = requestHeaders + event.Properties[dto.PropKeyRequestHeaders] = requestHeaders } if responseHeaders, exists := keyValuePairsFromMetadata[ResponseHeadersKey]; exists { - event.Properties["responseHeaders"] = responseHeaders + event.Properties[dto.PropKeyResponseHeaders] = responseHeaders } - // Optionally attach request and response payloads when enabled via configuration. - if c.cfg.Analytics.SendRequestBody { - if requestPayload, ok := keyValuePairsFromMetadata["request_payload"]; ok && requestPayload != "" { - event.Properties["request_payload"] = requestPayload + // Optionally attach request and response payloads when enabled via the collector. + if c.cfg.Collector.SendRequestBody { + if requestPayload, ok := keyValuePairsFromMetadata[dto.PropKeyRequestPayload]; ok && requestPayload != "" { + event.Properties[dto.PropKeyRequestPayload] = requestPayload slog.Debug("Analytics request payload captured", "size_bytes", len(requestPayload)) } } - if c.cfg.Analytics.SendResponseBody { - if responsePayload, ok := keyValuePairsFromMetadata["response_payload"]; ok && responsePayload != "" { - event.Properties["response_payload"] = responsePayload + if c.cfg.Collector.SendResponseBody { + if responsePayload, ok := keyValuePairsFromMetadata[dto.PropKeyResponsePayload]; ok && responsePayload != "" { + event.Properties[dto.PropKeyResponsePayload] = responsePayload slog.Debug("Analytics response payload captured", "size_bytes", len(responsePayload)) } } @@ -494,6 +528,24 @@ func (c *Analytics) prepareAnalyticEvent(logEntry *v3.HTTPAccessLogEntry) *dto.E return event } +// parseTrafficLogDirective returns the parsed TrafficLogDirective for the given +// raw JSON string, using a cache to avoid re-parsing the same static per-API +// directive on every request. Two concurrent first-parses of the same string are +// benign — the winner's value is stored and both callers get an equivalent result. +func (c *Analytics) parseTrafficLogDirective(raw string) *dto.TrafficLogDirective { + if cached, ok := c.directiveCache.Load(raw); ok { + return cached.(*dto.TrafficLogDirective) + } + dir := &dto.TrafficLogDirective{} + if raw != "" { + if err := json.Unmarshal([]byte(raw), dir); err != nil { + slog.Warn("Failed to parse traffic_log directive; opting in with defaults", "error", err) + } + } + c.directiveCache.Store(raw, dir) + return dir +} + func (c *Analytics) getAnonymousApp() *dto.Application { application := &dto.Application{} application.ApplicationID = anonymousValue 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 46d20cf61b..05afdaceab 100644 --- a/gateway/gateway-runtime/policy-engine/internal/analytics/analytics_test.go +++ b/gateway/gateway-runtime/policy-engine/internal/analytics/analytics_test.go @@ -63,8 +63,8 @@ func validAnalyticsConfigForValidation(analytics config.AnalyticsConfig) *config }, Analytics: analytics, } - cfg.Analytics.Enabled = true - cfg.Analytics.AccessLogsServiceCfg = config.AccessLogsServiceConfig{ + cfg.Analytics.Enabled = true // a consumer being on makes the collector implicit + cfg.Collector.AccessLogsServiceCfg = config.AccessLogsServiceConfig{ Mode: "uds", ShutdownTimeout: 600 * time.Second, ExtProcMaxMessageSize: 1000000, @@ -138,6 +138,18 @@ func TestNewAnalytics_EnabledWithUnknownPublisherType(t *testing.T) { assert.Empty(t, analytics.publishers) // Unknown type should not be added } +func TestNewAnalytics_TrafficLoggingEnabled(t *testing.T) { + // Traffic logging is a standalone consumer, independent of analytics. + cfg := &config.Config{ + TrafficLogging: config.TrafficLoggingConfig{Enabled: true}, + } + + analytics := NewAnalytics(cfg) + + require.NotNil(t, analytics) + assert.Len(t, analytics.publishers, 1) // traffic-logging publisher should be registered +} + // ============================================================================= // isInvalid Tests // ============================================================================= @@ -253,6 +265,24 @@ func TestProcess_WithMockPublisher(t *testing.T) { assert.Equal(t, "TestAPI", mockPub.event.API.APIName) } +// Path-based suppression moved to the traffic-logging publisher, so +// Analytics.Process no longer filters by path — every valid event reaches the +// publishers (each decides what to do with it). Ignored-path coverage lives in +// publishers/log_test.go. +func TestProcess_PublishesEvent(t *testing.T) { + analytics := NewAnalytics(&config.Config{}) + mockPub := &mockPublisher{} + analytics.publishers = append(analytics.publishers, mockPub) + + logEntry := &v3.HTTPAccessLogEntry{ + Response: &v3.HTTPResponseProperties{ResponseCode: wrapperspb.UInt32(200)}, + Request: &v3.HTTPRequestProperties{OriginalPath: "/api/v1/orders", RequestMethod: corev3.RequestMethod_GET}, + } + analytics.Process(logEntry) + + assert.True(t, mockPub.called, "publisher must be called") +} + func TestProcess_PanicRecovery(t *testing.T) { cfg := &config.Config{} analytics := NewAnalytics(cfg) @@ -373,6 +403,73 @@ func TestPrepareAnalyticEvent_WithFilterMetadata(t *testing.T) { assert.Equal(t, "TestApp", event.Application.ApplicationName) } +func TestPrepareAnalyticEvent_TrafficLogMarker(t *testing.T) { + cfg := &config.Config{} + analytics := NewAnalytics(cfg) + + logEntry := createLogEntryWithMetadata(map[string]string{ + APINameKey: "TestAPI", + TrafficLogMetadataKey: `{"request":{"payload":false,"headers":true}}`, + }) + + event := analytics.prepareAnalyticEvent(logEntry) + + require.NotNil(t, event) + require.NotNil(t, event.TrafficLog, "traffic_log marker should populate event.TrafficLog") + require.NotNil(t, event.TrafficLog.Request) + assert.True(t, event.TrafficLog.Request.Headers) + assert.False(t, event.TrafficLog.Request.Payload) + assert.Nil(t, event.TrafficLog.Response, "unset flow stays nil") + + // The marker must never leak into serialized properties (or other publishers). + _, inProps := event.Properties[TrafficLogMetadataKey] + assert.False(t, inProps, "traffic_log must not appear in event.Properties") +} + +func TestPrepareAnalyticEvent_NoTrafficLogMarker(t *testing.T) { + cfg := &config.Config{} + analytics := NewAnalytics(cfg) + + logEntry := createLogEntryWithMetadata(map[string]string{APINameKey: "TestAPI"}) + + event := analytics.prepareAnalyticEvent(logEntry) + + require.NotNil(t, event) + assert.Nil(t, event.TrafficLog, "no marker -> event.TrafficLog stays nil (publisher skips)") +} + +func TestPrepareAnalyticEvent_MalformedTrafficLogMarker(t *testing.T) { + cfg := &config.Config{} + analytics := NewAnalytics(cfg) + + logEntry := createLogEntryWithMetadata(map[string]string{ + TrafficLogMetadataKey: "not-json", + }) + + event := analytics.prepareAnalyticEvent(logEntry) + + require.NotNil(t, event) + // Presence alone is the opt-in signal; a malformed marker still opts in with + // default (empty) presentation. + require.NotNil(t, event.TrafficLog) + assert.Nil(t, event.TrafficLog.Request) + assert.Nil(t, event.TrafficLog.Response) +} + +func TestPrepareAnalyticEvent_TrafficLogDirectiveCached(t *testing.T) { + cfg := &config.Config{} + a := NewAnalytics(cfg) + raw := `{"request":{"payload":false,"headers":true}}` + logEntry := createLogEntryWithMetadata(map[string]string{TrafficLogMetadataKey: raw}) + + event1 := a.prepareAnalyticEvent(logEntry) + event2 := a.prepareAnalyticEvent(logEntry) + + require.NotNil(t, event1.TrafficLog) + // Same pointer — second call must return the cached directive, not a new allocation. + assert.Same(t, event1.TrafficLog, event2.TrafficLog) +} + func TestPrepareAnalyticEvent_WithAnonymousApp(t *testing.T) { cfg := &config.Config{} analytics := NewAnalytics(cfg) @@ -525,7 +622,7 @@ func TestPrepareAnalyticEvent_WithRequestResponseHeaders(t *testing.T) { func TestPrepareAnalyticEvent_WithPayloadsEnabled(t *testing.T) { cfg := &config.Config{ - Analytics: config.AnalyticsConfig{ + Collector: config.CollectorConfig{ SendRequestBody: true, SendResponseBody: true, }, @@ -551,7 +648,7 @@ func TestPrepareAnalyticEvent_WithPayloadsEnabled(t *testing.T) { func TestPrepareAnalyticEvent_WithPayloadsDisabled(t *testing.T) { cfg := &config.Config{ - Analytics: config.AnalyticsConfig{ + Collector: config.CollectorConfig{ SendRequestBody: false, SendResponseBody: false, }, @@ -574,7 +671,7 @@ func TestPrepareAnalyticEvent_WithPayloadsDisabled(t *testing.T) { func TestPrepareAnalyticEvent_RequestPayloadOnly(t *testing.T) { cfg := &config.Config{ - Analytics: config.AnalyticsConfig{ + Collector: config.CollectorConfig{ SendRequestBody: true, SendResponseBody: false, }, @@ -598,7 +695,7 @@ func TestPrepareAnalyticEvent_RequestPayloadOnly(t *testing.T) { func TestPrepareAnalyticEvent_ResponsePayloadOnly(t *testing.T) { cfg := &config.Config{ - Analytics: config.AnalyticsConfig{ + Collector: config.CollectorConfig{ SendRequestBody: false, SendResponseBody: true, }, diff --git a/gateway/gateway-runtime/policy-engine/internal/analytics/dto/event.go b/gateway/gateway-runtime/policy-engine/internal/analytics/dto/event.go index 6b2f05829b..fe5d7a89c4 100644 --- a/gateway/gateway-runtime/policy-engine/internal/analytics/dto/event.go +++ b/gateway/gateway-runtime/policy-engine/internal/analytics/dto/event.go @@ -19,6 +19,16 @@ package dto import "time" +// Property keys written into Event.Properties by the analytics pipeline and read +// back by publishers (e.g. the Log publisher's masking and field-projection paths). +// Both sides must use these constants so a rename stays in one place. +const ( + PropKeyRequestHeaders = "requestHeaders" + PropKeyResponseHeaders = "responseHeaders" + PropKeyRequestPayload = "request_payload" + PropKeyResponsePayload = "response_payload" +) + // Event represents analytics event data. type Event struct { API *ExtendedAPI `json:"api,omitempty" bson:"api"` @@ -36,4 +46,46 @@ type Event struct { UserIP string `json:"userIp,omitempty" bson:"user_ip"` ErrorType string `json:"errorType,omitempty" bson:"error_type"` Properties map[string]interface{} `json:"properties,omitempty" bson:"properties"` + + // TrafficLog carries the per-API stdout traffic-logging opt-in marker stamped + // by the log-message policy (access-log mode). When nil, the API has not opted + // in and the stdout traffic-logging publisher skips the event. It is + // gating/presentation state only and is never serialized (json:"-") nor sent + // to other publishers. + TrafficLog *TrafficLogDirective `json:"-" bson:"-"` } + +// TrafficLogDirective is the presentation config carried in the traffic-log +// marker. Field names mirror the policy's marker JSON so it round-trips. A nil +// flow means that flow was not configured. +type TrafficLogDirective struct { + Request *TrafficLogFlow `json:"request,omitempty"` + Response *TrafficLogFlow `json:"response,omitempty"` + Fields *TrafficLogFields `json:"fields,omitempty"` + // Labels holds the policy's resolved labels (context references already + // expanded at request time). The Log publisher emits them as a top-level + // "labels" object on the log line. + Labels map[string]interface{} `json:"labels,omitempty"` + // MaskedHeaders lists lower-cased header names whose values are redacted + // in the emitted log line. Merged with the global masked_headers config. + MaskedHeaders []string `json:"maskedHeaders,omitempty"` +} + +// TrafficLogFlow is the per-flow (request or response) presentation config. +type TrafficLogFlow struct { + Payload bool `json:"payload"` + Headers bool `json:"headers"` +} + +// TrafficLogFields selects which fields appear in the emitted line. Exactly one +// of Only or Exclude should be set. Only keeps exactly the named fields; Exclude +// drops the named fields and keeps everything else. Names are top-level keys +// (e.g. "latencies", "requestHeaders") or dotted sub-key paths within map fields +// (e.g. "requestHeaders.authorization", "labels.env"). When set, this is +// authoritative over field presence; per-flow Payload/Headers booleans are ignored +// (global header masking still applies). If both are set, Only takes precedence. +type TrafficLogFields struct { + Only []string `json:"only,omitempty"` + Exclude []string `json:"exclude,omitempty"` +} + diff --git a/gateway/gateway-runtime/policy-engine/internal/analytics/dto/latencies.go b/gateway/gateway-runtime/policy-engine/internal/analytics/dto/latencies.go index 412f95f460..e9e4818f0e 100644 --- a/gateway/gateway-runtime/policy-engine/internal/analytics/dto/latencies.go +++ b/gateway/gateway-runtime/policy-engine/internal/analytics/dto/latencies.go @@ -23,7 +23,12 @@ type Latencies struct { BackendLatency int64 `json:"backendLatency"` RequestMediationLatency int64 `json:"requestMediationLatency"` ResponseMediationLatency int64 `json:"responseMediationLatency"` - Duration int64 `json:"duration"` + // Duration is the total request duration: downstream request received → downstream response sent (ms). + Duration int64 `json:"duration"` + // BackendProcDuration is the backend TTFB: upstream request fully sent → first upstream response byte (ms). + BackendProcDuration int64 `json:"backendProcDuration"` + // ResponseProcDuration is the gateway response overhead: first upstream response byte → first downstream response byte (ms). + ResponseProcDuration int64 `json:"responseProcDuration"` } // GetResponseLatency returns the response latency. diff --git a/gateway/gateway-runtime/policy-engine/internal/analytics/publishers/log.go b/gateway/gateway-runtime/policy-engine/internal/analytics/publishers/log.go new file mode 100644 index 0000000000..0718c1aec7 --- /dev/null +++ b/gateway/gateway-runtime/policy-engine/internal/analytics/publishers/log.go @@ -0,0 +1,225 @@ +/* + * Copyright (c) 2025, WSO2 LLC. (http://www.wso2.org) All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package publishers + +import ( + "encoding/json" + "fmt" + "log/slog" + "os" + "strings" + "sync" + + "github.com/wso2/api-platform/gateway/gateway-runtime/policy-engine/internal/analytics/dto" + "github.com/wso2/api-platform/gateway/gateway-runtime/policy-engine/internal/config" +) + +// maskedHeaderValue is the placeholder written in place of a masked header value. +const maskedHeaderValue = "****" + +// Log is an analytics publisher that writes each enriched analytics event to +// stdout as a single JSON line. It is intended for log-scraping pipelines +// (Fluent Bit, Loki, ELK, etc.) and as a lightweight alternative to a SaaS +// analytics backend. The event already carries the rich metadata, headers and +// (when send_request_body/send_response_body are enabled) payloads attached by +// the analytics engine, so this publisher only serializes it. +type Log struct { + // maskedHeaders holds lower-cased header names whose values are redacted in + // the requestHeaders/responseHeaders properties before logging. + maskedHeaders map[string]bool + // maxPayloadSize caps the number of request/response payload bytes written to + // the log line (0 = no limit). Truncation is output-side only. + maxPayloadSize int + // mu serializes writes to stdout so concurrent ALS streams do not interleave. + mu sync.Mutex + // out is the destination writer; defaults to os.Stdout (overridable in tests). + out *os.File +} + +// NewLog creates a new stdout traffic-logging publisher. +func NewLog(logCfg *config.TrafficLoggingConfig) *Log { + if logCfg == nil { + logCfg = &config.TrafficLoggingConfig{} + } + + masked := make(map[string]bool, len(logCfg.MaskedHeaders)) + for _, h := range logCfg.MaskedHeaders { + h = strings.ToLower(strings.TrimSpace(h)) + if h != "" { + masked[h] = true + } + } + + return &Log{ + maskedHeaders: masked, + maxPayloadSize: logCfg.MaxPayloadSize, + out: os.Stdout, + } +} + +// Publish writes the event to stdout as JSON. Traffic logging is per-API: only +// events carrying a traffic-log directive (stamped by the log-message policy in +// access-log mode on APIs that opted in) are emitted; all others are skipped. +func (l *Log) Publish(event *dto.Event) { + if event == nil || event.TrafficLog == nil { + return + } + + dir := event.TrafficLog + tl := l.toTrafficLogEvent(event, dir) + + data, err := json.Marshal(tl) + if err != nil { + slog.Error("Failed to marshal traffic-log event", "error", err) + return + } + + if fields := dir.Fields; fields != nil && (len(fields.Only) > 0 || len(fields.Exclude) > 0) { + var m map[string]interface{} + if err := json.Unmarshal(data, &m); err != nil { + slog.Error("Failed to unmarshal for field projection; emitting as-is", "error", err) + } else { + applyFieldsProjection(m, fields) + if projected, merr := json.Marshal(m); merr == nil { + data = projected + } else { + slog.Error("Failed to remarshal after field projection; emitting as-is", "error", merr) + } + } + } + + l.write(data) +} + +func (l *Log) write(data []byte) { + l.mu.Lock() + defer l.mu.Unlock() + if _, err := fmt.Fprintln(l.out, string(data)); err != nil { + slog.Error("Failed to write analytics event to stdout", "error", err) + } +} + +// parseHeadersFromString converts the JSON-encoded header value stored in +// event.Properties (a map[string]string or map[string][]string serialized by the +// ext_proc layer) into a map[string]string so it embeds as a plain JSON object +// in the log line. Other publishers (e.g. Moesif) read the raw string directly; +// the Log publisher calls this only on the local TrafficLogEvent it builds, so +// the shared event is never modified. Multi-value headers are flattened to their +// first value. Returns nil on empty input or parse failure. +func parseHeadersFromString(raw string) map[string]string { + if raw == "" { + return nil + } + var single map[string]string + if err := json.Unmarshal([]byte(raw), &single); err == nil { + return single + } + // Fallback: multi-value wire format — flatten to first value. + var multi map[string][]string + if err := json.Unmarshal([]byte(raw), &multi); err == nil { + out := make(map[string]string, len(multi)) + for k, vs := range multi { + if len(vs) > 0 { + out[k] = vs[0] + } + } + return out + } + return nil +} + +// truncatePayload returns up to maxPayloadSize bytes of the payload (0 = no +// limit). Truncation is on a byte boundary, matching the previous capture-time +// behavior. +func (l *Log) truncatePayload(s string) string { + if l.maxPayloadSize <= 0 || len(s) <= l.maxPayloadSize { + return s + } + return s[:l.maxPayloadSize] +} + +// applyFieldsProjection mutates m in place to restrict it to the configured +// fields. Names are top-level keys (e.g. "latencies", "requestHeaders") or +// dotted sub-key paths within map fields (e.g. "requestHeaders.authorization", +// "labels.env"). Only keeps exactly the named fields; Exclude drops the named +// fields and keeps everything else. If both are set, Only takes precedence. +func applyFieldsProjection(m map[string]interface{}, fields *dto.TrafficLogFields) { + if len(fields.Only) > 0 { + directKeys := make(map[string]bool) + subKeys := make(map[string][]string) // topKey → sub-keys to keep + for _, name := range fields.Only { + if top, sub, found := strings.Cut(name, "."); found { + subKeys[top] = append(subKeys[top], sub) + } else { + directKeys[name] = true + } + } + for key := range m { + if !directKeys[key] && subKeys[key] == nil { + delete(m, key) + } + } + for top, subs := range subKeys { + if directKeys[top] { + continue // whole key kept; don't filter sub-keys + } + if nested, ok := m[top].(map[string]interface{}); ok { + keep := make(map[string]bool, len(subs)) + for _, s := range subs { + keep[s] = true + } + for k := range nested { + if !keep[k] { + delete(nested, k) + } + } + if len(nested) == 0 { + delete(m, top) + } + } + } + return + } + for _, name := range fields.Exclude { + if top, sub, found := strings.Cut(name, "."); found { + if nested, ok := m[top].(map[string]interface{}); ok { + delete(nested, sub) + if len(nested) == 0 { + delete(m, top) + } + } + } else { + delete(m, name) + } + } +} + +// maskHeaders redacts header values whose names appear in mask (case-insensitive). +// Returns a new map; the input is not modified. Per-header exclusion is handled +// downstream by applyFieldsProjection via dotted paths (e.g. "requestHeaders.authorization"). +func (l *Log) maskHeaders(headers map[string]string, mask map[string]bool) map[string]string { + result := make(map[string]string, len(headers)) + for name, value := range headers { + if mask[strings.ToLower(name)] { + result[name] = maskedHeaderValue + } else { + result[name] = value + } + } + return result +} diff --git a/gateway/gateway-runtime/policy-engine/internal/analytics/publishers/log_test.go b/gateway/gateway-runtime/policy-engine/internal/analytics/publishers/log_test.go new file mode 100644 index 0000000000..fa4fb8c5e0 --- /dev/null +++ b/gateway/gateway-runtime/policy-engine/internal/analytics/publishers/log_test.go @@ -0,0 +1,468 @@ +/* + * Copyright (c) 2026, WSO2 LLC. (http://www.wso2.org) All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package publishers + +import ( + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/wso2/api-platform/gateway/gateway-runtime/policy-engine/internal/analytics/dto" + "github.com/wso2/api-platform/gateway/gateway-runtime/policy-engine/internal/config" +) + +// newLogToFile builds a Log publisher that writes to a temp file, returning the +// publisher and a function that reads back what was written. +func newLogToFile(t *testing.T, cfg *config.TrafficLoggingConfig) (*Log, func() string) { + t.Helper() + path := filepath.Join(t.TempDir(), "out.log") + f, err := os.Create(path) + require.NoError(t, err) + t.Cleanup(func() { _ = f.Close() }) + + l := NewLog(cfg) + l.out = f + return l, func() string { + require.NoError(t, f.Sync()) + data, err := os.ReadFile(path) + require.NoError(t, err) + return string(data) + } +} + +// bothFlows returns a directive that opts in to logging both request and response +// headers and payloads — the "log everything captured" case. +func bothFlows() *dto.TrafficLogDirective { + return &dto.TrafficLogDirective{ + Request: &dto.TrafficLogFlow{Payload: true, Headers: true}, + Response: &dto.TrafficLogFlow{Payload: true, Headers: true}, + } +} + +// decodeLine parses the single JSON line emitted by the Log publisher. +func decodeLine(t *testing.T, out string) map[string]interface{} { + t.Helper() + var decoded map[string]interface{} + require.NoError(t, json.Unmarshal([]byte(out), &decoded)) + return decoded +} + +// headerMap returns the JSON-decoded header object as map[string]interface{} for +// easy key/value assertions. Fails the test if v is not the expected type. +func headerMap(t *testing.T, v interface{}) map[string]interface{} { + t.Helper() + m, ok := v.(map[string]interface{}) + require.True(t, ok, "expected header object (map[string]interface{}), got %T", v) + return m +} + +func TestNewLog_NilConfig(t *testing.T) { + l := NewLog(nil) + require.NotNil(t, l) + assert.Empty(t, l.maskedHeaders) +} + +// Per-API gating: an event without a traffic-log directive is never emitted. +func TestLog_Publish_SkipsWhenNoDirective(t *testing.T) { + l, read := newLogToFile(t, &config.TrafficLoggingConfig{}) + event := createBaseEvent() // no TrafficLog set + event.Properties["requestHeaders"] = `{"x-foo":"bar"}` + + l.Publish(event) + + assert.Empty(t, read(), "event without a traffic-log directive must not be logged") +} + +func TestLog_Publish_WritesJSONLineWithLatencies(t *testing.T) { + l, read := newLogToFile(t, &config.TrafficLoggingConfig{}) + event := createBaseEvent() + event.TrafficLog = bothFlows() + event.Properties["requestHeaders"] = `{"x-foo":"bar"}` + + l.Publish(event) + + out := read() + // Single line (one trailing newline). + assert.Equal(t, 1, strings.Count(strings.TrimRight(out, "\n"), "\n")+1) + + decoded := decodeLine(t, out) + api := decoded["api"].(map[string]interface{}) + assert.Equal(t, "test-api", api["name"]) + reqH := headerMap(t, decoded["requestHeaders"]) + assert.Equal(t, "bar", reqH["x-foo"]) + + // ALS-derived latencies are always present in the line — the key improvement + // over the inline log-message policy, which could never see them. + latencies := decoded["latencies"].(map[string]interface{}) + assert.Equal(t, float64(100), latencies["responseLatency"]) +} + +// Labels from the directive are emitted as a top-level "labels" object. +func TestLog_Publish_LabelsTopLevel(t *testing.T) { + l, read := newLogToFile(t, &config.TrafficLoggingConfig{}) + event := createBaseEvent() + event.TrafficLog = &dto.TrafficLogDirective{ + Request: &dto.TrafficLogFlow{Headers: true}, + Labels: map[string]interface{}{ + "who": "alice", + "authType": "jwt", + "retryCount": float64(3), + }, + } + event.Properties["requestHeaders"] = `{"x-foo":"bar"}` + + l.Publish(event) + + decoded := decodeLine(t, read()) + labels, ok := decoded["labels"].(map[string]interface{}) + require.True(t, ok, "expected top-level labels object, got %T", decoded["labels"]) + assert.Equal(t, "alice", labels["who"]) + assert.Equal(t, "jwt", labels["authType"]) + assert.Equal(t, float64(3), labels["retryCount"]) + reqH := headerMap(t, decoded["requestHeaders"]) + assert.Equal(t, "bar", reqH["x-foo"]) +} + +// A directive with no labels emits no "labels" key. +func TestLog_Publish_NoLabelsWhenAbsent(t *testing.T) { + l, read := newLogToFile(t, &config.TrafficLoggingConfig{}) + event := createBaseEvent() + event.TrafficLog = bothFlows() + + l.Publish(event) + + decoded := decodeLine(t, read()) + _, present := decoded["labels"] + assert.False(t, present, "no labels key expected when directive has no labels") +} + +// The fields projection can select "labels" like any other top-level key. +func TestLog_Publish_LabelsProjectableViaFields(t *testing.T) { + l, read := newLogToFile(t, &config.TrafficLoggingConfig{}) + event := createBaseEvent() + event.TrafficLog = &dto.TrafficLogDirective{ + Labels: map[string]interface{}{"who": "alice"}, + Fields: &dto.TrafficLogFields{Only: []string{"labels"}}, + } + event.Properties["requestHeaders"] = `{"x-foo":"bar"}` + + l.Publish(event) + + decoded := decodeLine(t, read()) + labels, ok := decoded["labels"].(map[string]interface{}) + require.True(t, ok, "expected labels retained by include projection") + assert.Equal(t, "alice", labels["who"]) + _, hasHeaders := decoded["requestHeaders"] + assert.False(t, hasHeaders, "requestHeaders not in Only list -> dropped") +} + +func TestLog_Publish_MasksHeaders(t *testing.T) { + l, read := newLogToFile(t, &config.TrafficLoggingConfig{MaskedHeaders: []string{"Authorization"}}) + event := createBaseEvent() + event.TrafficLog = bothFlows() + event.Properties["requestHeaders"] = `{"Authorization":"Bearer secret","x-foo":"bar"}` + event.Properties["responseHeaders"] = `{"authorization":"Bearer secret2"}` + + l.Publish(event) + + decoded := decodeLine(t, read()) + reqH := headerMap(t, decoded["requestHeaders"]) + assert.Equal(t, "****", reqH["Authorization"]) // masked + assert.Equal(t, "bar", reqH["x-foo"]) // untouched + + resH := headerMap(t, decoded["responseHeaders"]) + assert.Equal(t, "****", resH["authorization"]) // case-insensitive match +} + +// Per-API maskedHeaders are merged with the global config mask; either source alone redacts. +func TestLog_Publish_PerAPIMaskedHeadersMergedWithGlobal(t *testing.T) { + // Global config masks "authorization"; per-API directive adds "x-secret". + l, read := newLogToFile(t, &config.TrafficLoggingConfig{MaskedHeaders: []string{"authorization"}}) + event := createBaseEvent() + event.TrafficLog = &dto.TrafficLogDirective{ + Request: &dto.TrafficLogFlow{Headers: true}, + MaskedHeaders: []string{"x-secret"}, + } + event.Properties["requestHeaders"] = `{"Authorization":"Bearer s","X-Secret":"top","x-foo":"bar"}` + + l.Publish(event) + + decoded := decodeLine(t, read()) + reqH := headerMap(t, decoded["requestHeaders"]) + assert.Equal(t, "****", reqH["Authorization"], "global masked header still redacted") + assert.Equal(t, "****", reqH["X-Secret"], "per-API masked header redacted") + assert.Equal(t, "bar", reqH["x-foo"], "unmasked header unchanged") +} + +// Per-API maskedHeaders work even when no global headers are configured. +func TestLog_Publish_PerAPIMaskedHeadersNoGlobal(t *testing.T) { + l, read := newLogToFile(t, &config.TrafficLoggingConfig{}) + event := createBaseEvent() + event.TrafficLog = &dto.TrafficLogDirective{ + Request: &dto.TrafficLogFlow{Headers: true}, + MaskedHeaders: []string{"X-Token"}, + } + event.Properties["requestHeaders"] = `{"X-Token":"secret","x-foo":"bar"}` + + l.Publish(event) + + decoded := decodeLine(t, read()) + reqH := headerMap(t, decoded["requestHeaders"]) + assert.Equal(t, "****", reqH["X-Token"]) + assert.Equal(t, "bar", reqH["x-foo"]) +} + +// fields.exclude with a dotted path drops a specific header entirely (vs global masking which redacts). +func TestLog_Publish_ExcludeHeadersDrops(t *testing.T) { + l, read := newLogToFile(t, &config.TrafficLoggingConfig{MaskedHeaders: []string{"authorization"}}) + event := createBaseEvent() + event.TrafficLog = &dto.TrafficLogDirective{ + Request: &dto.TrafficLogFlow{Headers: true}, + Fields: &dto.TrafficLogFields{Exclude: []string{"requestHeaders.X-Secret"}}, + } + event.Properties["requestHeaders"] = `{"Authorization":"Bearer s","X-Secret":"top","x-foo":"bar"}` + + l.Publish(event) + + decoded := decodeLine(t, read()) + reqH := headerMap(t, decoded["requestHeaders"]) + + _, hasSecret := reqH["X-Secret"] + assert.False(t, hasSecret, "excluded header must be dropped") + assert.Equal(t, "****", reqH["Authorization"], "masked header still redacted") + assert.Equal(t, "bar", reqH["x-foo"]) +} + +// headers:false omits the headers property; a nil flow omits its whole side. +func TestLog_Publish_DisabledFieldsOmitted(t *testing.T) { + l, read := newLogToFile(t, &config.TrafficLoggingConfig{}) + event := createBaseEvent() + event.TrafficLog = &dto.TrafficLogDirective{ + Request: &dto.TrafficLogFlow{Headers: false, Payload: true}, + // Response flow nil -> both response props dropped. + } + event.Properties["requestHeaders"] = `{"x-foo":"bar"}` + event.Properties["request_payload"] = "req-body" + event.Properties["responseHeaders"] = `{"x-bar":"baz"}` + event.Properties["response_payload"] = "resp-body" + + l.Publish(event) + + decoded := decodeLine(t, read()) + _, hasReqHeaders := decoded["requestHeaders"] + assert.False(t, hasReqHeaders, "request headers disabled -> omitted") + assert.Equal(t, "req-body", decoded["requestBody"], "request payload enabled -> kept") + + _, hasRespHeaders := decoded["responseHeaders"] + _, hasRespPayload := decoded["responseBody"] + assert.False(t, hasRespHeaders, "nil response flow -> response headers omitted") + assert.False(t, hasRespPayload, "nil response flow -> response payload omitted") +} + +func TestLog_Publish_DoesNotMutateSharedEvent(t *testing.T) { + l, _ := newLogToFile(t, &config.TrafficLoggingConfig{MaskedHeaders: []string{"authorization"}}) + event := createBaseEvent() + event.TrafficLog = bothFlows() + original := `{"authorization":"Bearer secret"}` + event.Properties["requestHeaders"] = original + + l.Publish(event) + + // The shared event (read by other publishers) must be untouched. + assert.Equal(t, original, event.Properties["requestHeaders"]) +} + +func TestLog_Publish_NilEvent(t *testing.T) { + l, read := newLogToFile(t, &config.TrafficLoggingConfig{}) + assert.NotPanics(t, func() { l.Publish(nil) }) + assert.Empty(t, read()) +} + +// Field selection (include): only named top-level keys survive; fields.only is +// authoritative over presence (request.headers boolean is ignored, masking still applies). +func TestLog_Publish_FieldsInclude(t *testing.T) { + l, read := newLogToFile(t, &config.TrafficLoggingConfig{MaskedHeaders: []string{"authorization"}}) + event := createBaseEvent() + event.Properties["requestHeaders"] = `{"Authorization":"Bearer s","X-Keep":"k"}` + event.Properties["responseHeaders"] = `{"x-bar":"baz"}` + event.TrafficLog = &dto.TrafficLogDirective{ + Request: &dto.TrafficLogFlow{Headers: false}, // boolean ignored when fields set + Fields: &dto.TrafficLogFields{Only: []string{"latencies", "requestHeaders"}}, + } + + l.Publish(event) + decoded := decodeLine(t, read()) + + _, hasAPI := decoded["api"] + _, hasOp := decoded["operation"] + assert.False(t, hasAPI, "api not listed -> dropped") + assert.False(t, hasOp, "operation not listed -> dropped") + assert.Contains(t, decoded, "latencies", "latencies listed -> kept") + + _, hasResp := decoded["responseHeaders"] + assert.False(t, hasResp, "responseHeaders not listed -> dropped") + + require.NotNil(t, decoded["requestHeaders"], "requestHeaders present (fields authoritative, boolean ignored)") + reqH := headerMap(t, decoded["requestHeaders"]) + assert.Equal(t, "****", reqH["Authorization"], "masking still applies") + assert.Equal(t, "k", reqH["X-Keep"]) +} + +// Field selection (exclude): named keys are dropped, everything else remains. +func TestLog_Publish_FieldsExclude(t *testing.T) { + l, read := newLogToFile(t, &config.TrafficLoggingConfig{}) + event := createBaseEvent() + event.Properties["requestHeaders"] = `{"x-foo":"bar"}` + event.Properties["request_payload"] = "secret-body" + event.TrafficLog = &dto.TrafficLogDirective{ + Fields: &dto.TrafficLogFields{Exclude: []string{"operation", "requestBody"}}, + } + + l.Publish(event) + decoded := decodeLine(t, read()) + + _, hasOp := decoded["operation"] + assert.False(t, hasOp, "operation excluded") + assert.Contains(t, decoded, "api", "api kept") + assert.Contains(t, decoded, "latencies", "latencies kept") + _, hasPayload := decoded["requestBody"] + assert.False(t, hasPayload, "requestBody excluded") + assert.Contains(t, decoded, "requestHeaders", "requestHeaders kept (not excluded)") +} + +// requestBody and labels are top-level keys like any other and can be selected +// explicitly via fields.only. +func TestLog_Publish_FieldsIncludeRequestBodyAndLabels(t *testing.T) { + l, read := newLogToFile(t, &config.TrafficLoggingConfig{}) + event := createBaseEvent() + event.Properties["requestHeaders"] = `{"x-foo":"bar"}` + event.Properties["request_payload"] = "body-data" + event.TrafficLog = &dto.TrafficLogDirective{ + Labels: map[string]interface{}{"env": "prod"}, + Fields: &dto.TrafficLogFields{Only: []string{"requestBody", "labels"}}, + } + + l.Publish(event) + decoded := decodeLine(t, read()) + + _, hasAPI := decoded["api"] + assert.False(t, hasAPI, "api not listed -> dropped") + _, hasHeaders := decoded["requestHeaders"] + assert.False(t, hasHeaders, "requestHeaders not in Only list -> dropped") + assert.Equal(t, "body-data", decoded["requestBody"]) + labels, ok := decoded["labels"].(map[string]interface{}) + require.True(t, ok) + assert.Equal(t, "prod", labels["env"]) +} + +// Output-side payload truncation (0 = no limit). +func TestLog_Publish_TruncatesPayload(t *testing.T) { + l, read := newLogToFile(t, &config.TrafficLoggingConfig{MaxPayloadSize: 5}) + event := createBaseEvent() + event.TrafficLog = bothFlows() + event.Properties["request_payload"] = "hello world" + event.Properties["response_payload"] = "goodbye world" + + l.Publish(event) + decoded := decodeLine(t, read()) + assert.Equal(t, "hello", decoded["requestBody"]) + assert.Equal(t, "goodb", decoded["responseBody"]) +} + +func TestLog_Publish_NoTruncationWhenZero(t *testing.T) { + l, read := newLogToFile(t, &config.TrafficLoggingConfig{MaxPayloadSize: 0}) + event := createBaseEvent() + event.TrafficLog = bothFlows() + event.Properties["request_payload"] = "hello world" + + l.Publish(event) + decoded := decodeLine(t, read()) + assert.Equal(t, "hello world", decoded["requestBody"]) +} + +func TestLog_Publish_UnparseableHeadersDropped(t *testing.T) { + l, read := newLogToFile(t, &config.TrafficLoggingConfig{MaskedHeaders: []string{"authorization"}}) + event := createBaseEvent() + event.TrafficLog = bothFlows() + event.Properties["requestHeaders"] = "not-json" + + l.Publish(event) + + decoded := decodeLine(t, read()) + _, hasHeaders := decoded["requestHeaders"] + assert.False(t, hasHeaders, "unparseable header value must be silently dropped") +} + +// Application is omitted entirely for unauthenticated requests (all fields empty). +func TestLog_Publish_UnauthenticatedRequestOmitsApplication(t *testing.T) { + l, read := newLogToFile(t, &config.TrafficLoggingConfig{}) + event := createBaseEvent() + event.Application = &dto.Application{} // all fields are "" + event.TrafficLog = bothFlows() + + l.Publish(event) + + decoded := decodeLine(t, read()) + _, hasApp := decoded["application"] + assert.False(t, hasApp, "application with all-empty fields must be absent") +} + +// TrafficLogAPI uses clean field names (id/name/kind) not the Moesif apiId/apiName/apiType. +func TestLog_Publish_APIFieldNames(t *testing.T) { + l, read := newLogToFile(t, &config.TrafficLoggingConfig{}) + event := createBaseEvent() + event.TrafficLog = bothFlows() + + l.Publish(event) + + decoded := decodeLine(t, read()) + api, ok := decoded["api"].(map[string]interface{}) + require.True(t, ok, "api object must be present") + assert.Equal(t, "api-123", api["id"]) + assert.Equal(t, "test-api", api["name"]) + assert.Equal(t, "v1.0", api["version"]) + assert.Equal(t, "Rest", api["kind"]) + assert.Equal(t, "/test", api["context"]) + assert.Equal(t, "project-123", api["projectId"]) + // Moesif-specific keys must not bleed into traffic logs. + for _, moesifKey := range []string{"apiId", "apiName", "apiCreator", "apiCreatorTenantDomain", "organizationId"} { + _, present := api[moesifKey] + assert.False(t, present, "Moesif field %q must not appear in traffic log", moesifKey) + } +} + +// Top-level fields — status, correlationId, client — are always present when set. +func TestLog_Publish_TopLevelFieldsPresent(t *testing.T) { + l, read := newLogToFile(t, &config.TrafficLoggingConfig{}) + event := createBaseEvent() + event.TrafficLog = bothFlows() + + l.Publish(event) + + decoded := decodeLine(t, read()) + assert.Equal(t, float64(200), decoded["status"], "proxy response code must appear as status") + assert.Equal(t, "corr-123", decoded["correlationId"]) + client, ok := decoded["client"].(map[string]interface{}) + require.True(t, ok, "client object must be present") + assert.Equal(t, "192.168.1.1", client["ip"]) + assert.Equal(t, "test-agent", client["userAgent"]) +} diff --git a/gateway/gateway-runtime/policy-engine/internal/analytics/publishers/traffic_log_event.go b/gateway/gateway-runtime/policy-engine/internal/analytics/publishers/traffic_log_event.go new file mode 100644 index 0000000000..da6b251a70 --- /dev/null +++ b/gateway/gateway-runtime/policy-engine/internal/analytics/publishers/traffic_log_event.go @@ -0,0 +1,197 @@ +/* + * Copyright (c) 2026, WSO2 LLC. (http://www.wso2.org) All Rights Reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * + */ + +package publishers + +import ( + "strings" + + "github.com/wso2/api-platform/gateway/gateway-runtime/policy-engine/internal/analytics/dto" +) + +// trafficLogTimestampFormat is RFC 3339 with millisecond precision. +const trafficLogTimestampFormat = "2006-01-02T15:04:05.000Z07:00" + +// TrafficLogEvent is the JSON shape written to stdout by the Log publisher. +// It is intentionally separate from dto.Event (shaped for Moesif) so its field +// names, schema, and presence rules can evolve independently. All string fields +// carry omitempty so absent or unknown values produce no key rather than "". +type TrafficLogEvent struct { + Timestamp string `json:"timestamp,omitempty"` + CorrelationID string `json:"correlationId,omitempty"` + Status int `json:"status,omitempty"` + API *TrafficLogAPI `json:"api,omitempty"` + Operation *TrafficLogOperation `json:"operation,omitempty"` + Target *TrafficLogTarget `json:"target,omitempty"` + Application *TrafficLogApplication `json:"application,omitempty"` + Client *TrafficLogClient `json:"client,omitempty"` + Latencies *dto.Latencies `json:"latencies,omitempty"` + RequestHeaders map[string]string `json:"requestHeaders,omitempty"` + ResponseHeaders map[string]string `json:"responseHeaders,omitempty"` + RequestBody string `json:"requestBody,omitempty"` + ResponseBody string `json:"responseBody,omitempty"` + Labels map[string]interface{} `json:"labels,omitempty"` +} + +// TrafficLogAPI identifies the API that processed the request. +type TrafficLogAPI struct { + ID string `json:"id,omitempty"` + Name string `json:"name,omitempty"` + Version string `json:"version,omitempty"` + Context string `json:"context,omitempty"` + Kind string `json:"kind,omitempty"` + ProjectID string `json:"projectId,omitempty"` +} + +// TrafficLogOperation describes the matched operation within the API. +type TrafficLogOperation struct { + Method string `json:"method,omitempty"` + Path string `json:"path,omitempty"` +} + +// TrafficLogTarget holds upstream response information. +type TrafficLogTarget struct { + StatusCode int `json:"statusCode,omitempty"` + Destination string `json:"destination,omitempty"` +} + +// TrafficLogApplication is present only for authenticated requests. +type TrafficLogApplication struct { + ID string `json:"id,omitempty"` + Name string `json:"name,omitempty"` + Owner string `json:"owner,omitempty"` + KeyType string `json:"keyType,omitempty"` +} + +// TrafficLogClient holds downstream caller information. +type TrafficLogClient struct { + IP string `json:"ip,omitempty"` + UserAgent string `json:"userAgent,omitempty"` +} + +// toTrafficLogEvent translates a dto.Event and its traffic-log directive into the +// traffic-log-specific output shape, applying per-flow header filtering, global +// header masking, and payload truncation. +func (l *Log) toTrafficLogEvent(event *dto.Event, dir *dto.TrafficLogDirective) *TrafficLogEvent { + tl := &TrafficLogEvent{ + Status: event.ProxyResponseCode, + Latencies: event.Latencies, + } + + if !event.RequestTimestamp.IsZero() { + tl.Timestamp = event.RequestTimestamp.UTC().Format(trafficLogTimestampFormat) + } + + if event.MetaInfo != nil { + tl.CorrelationID = event.MetaInfo.CorrelationID + } + + if event.API != nil { + tl.API = &TrafficLogAPI{ + ID: event.API.APIID, + Name: event.API.APIName, + Version: event.API.APIVersion, + Context: event.API.APIContext, + Kind: event.API.APIType, + ProjectID: event.API.ProjectID, + } + } + + if event.Operation != nil { + tl.Operation = &TrafficLogOperation{ + Method: event.Operation.APIMethod, + Path: event.Operation.APIResourceTemplate, + } + } + + if event.Target != nil { + tl.Target = &TrafficLogTarget{ + StatusCode: event.Target.TargetResponseCode, + Destination: event.Target.Destination, + } + } + + // Application is only meaningful for authenticated requests. + if a := event.Application; a != nil && (a.ApplicationID != "" || a.ApplicationName != "") { + tl.Application = &TrafficLogApplication{ + ID: a.ApplicationID, + Name: a.ApplicationName, + Owner: a.ApplicationOwner, + KeyType: a.KeyType, + } + } + + if event.UserIP != "" || event.UserAgentHeader != "" { + tl.Client = &TrafficLogClient{ + IP: event.UserIP, + UserAgent: event.UserAgentHeader, + } + } + + // When a fields selection is configured it is authoritative over presence: + // the per-flow Headers/Payload booleans are ignored and applyFieldsProjection + // in Publish decides what survives. Global masking always applies; per-API + // maskedHeaders are merged in here and passed down. + hasFieldsSelection := dir.Fields != nil && (len(dir.Fields.Only) > 0 || len(dir.Fields.Exclude) > 0) + + // Effective header mask: global config merged with any per-API additions. + mask := l.maskedHeaders + if len(dir.MaskedHeaders) > 0 { + merged := make(map[string]bool, len(l.maskedHeaders)+len(dir.MaskedHeaders)) + for k, v := range l.maskedHeaders { + merged[k] = v + } + for _, h := range dir.MaskedHeaders { + merged[strings.ToLower(h)] = true + } + mask = merged + } + + // Request flow + if raw, ok := event.Properties[dto.PropKeyRequestHeaders].(string); ok { + if hasFieldsSelection || (dir.Request != nil && dir.Request.Headers) { + if headers := parseHeadersFromString(raw); headers != nil { + tl.RequestHeaders = l.maskHeaders(headers, mask) + } + } + } + if p, ok := event.Properties[dto.PropKeyRequestPayload].(string); ok && p != "" { + if hasFieldsSelection || (dir.Request != nil && dir.Request.Payload) { + tl.RequestBody = l.truncatePayload(p) + } + } + + // Response flow + if raw, ok := event.Properties[dto.PropKeyResponseHeaders].(string); ok { + if hasFieldsSelection || (dir.Response != nil && dir.Response.Headers) { + if headers := parseHeadersFromString(raw); headers != nil { + tl.ResponseHeaders = l.maskHeaders(headers, mask) + } + } + } + if p, ok := event.Properties[dto.PropKeyResponsePayload].(string); ok && p != "" { + if hasFieldsSelection || (dir.Response != nil && dir.Response.Payload) { + tl.ResponseBody = l.truncatePayload(p) + } + } + + if len(dir.Labels) > 0 { + tl.Labels = dir.Labels + } + + return tl +} diff --git a/gateway/gateway-runtime/policy-engine/internal/config/config.go b/gateway/gateway-runtime/policy-engine/internal/config/config.go index 5dfdc688ed..9580c733a2 100644 --- a/gateway/gateway-runtime/policy-engine/internal/config/config.go +++ b/gateway/gateway-runtime/policy-engine/internal/config/config.go @@ -42,24 +42,44 @@ type Config struct { PolicyEngine PolicyEngine `koanf:"policy_engine"` GatewayController map[string]interface{} `koanf:"gateway_controller"` PolicyConfigurations map[string]interface{} `koanf:"policy_configurations"` + Collector CollectorConfig `koanf:"collector"` Analytics AnalyticsConfig `koanf:"analytics"` + TrafficLogging TrafficLoggingConfig `koanf:"traffic_logging"` TracingConfig TracingConfig `koanf:"tracing"` } +// CollectorConfig holds the data-collection ("collector") configuration. The +// collector is the shared capture pipeline that gathers request/response headers +// and bodies and ships them to the policy-engine over ALS. It underpins every +// consumer of that data (analytics and traffic logging) and is implicitly active +// whenever a consumer is enabled — see Config.IsCollectorEnabled. This section +// tunes capture and transport; it has no on/off flag of its own. +type CollectorConfig struct { + // SendRequestBody / SendResponseBody attach captured request/response bodies + // onto the collected event. + SendRequestBody bool `koanf:"send_request_body"` + SendResponseBody bool `koanf:"send_response_body"` + // AccessLogsServiceCfg tunes the policy-engine ALS receiver (the gRPC server + // that ingests collected access logs). It is part of the collector transport + // and is configured under the shared [collector.als] section (the controller + // reads the same section to configure Envoy's sender side). + AccessLogsServiceCfg AccessLogsServiceConfig `koanf:"als"` +} + // AnalyticsConfig holds analytics configuration type AnalyticsConfig struct { - Enabled bool `koanf:"enabled"` - EnabledPublishers []string `koanf:"enabled_publishers"` - Publishers AnalyticsPublishersConfig `koanf:"publishers"` - GRPCEventServerCfg map[string]interface{} `koanf:"grpc_event_server"` - AccessLogsServiceCfg AccessLogsServiceConfig `koanf:"access_logs_service"` - // AllowPayloads controls whether request and response bodies are captured - // into analytics metadata and forwarded to analytics publishers. - // Deprecated: use SendRequestBody and SendResponseBody instead. - // When true, validateAnalyticsConfig maps both SendRequestBody and SendResponseBody - // to true if both are false. Because bools cannot represent "unset", this also - // applies when both new flags are explicitly false; remove allow_payloads when - // migrating and set the directional flags directly. + Enabled bool `koanf:"enabled"` + EnabledPublishers []string `koanf:"enabled_publishers"` + Publishers AnalyticsPublishersConfig `koanf:"publishers"` + GRPCEventServerCfg map[string]interface{} `koanf:"grpc_event_server"` + // AccessLogsServiceCfg is a deprecated alias. ALS receiver tuning moved to + // [collector.als]; when set here it is migrated onto the collector during + // validation (with a warning). Prefer [collector.als]. + AccessLogsServiceCfg AccessLogsServiceConfig `koanf:"access_logs_service"` + // AllowPayloads, SendRequestBody and SendResponseBody are deprecated aliases. + // Body capture now lives under [collector]. When set, these are mapped onto + // collector.send_request_body / collector.send_response_body during validation + // (with a warning). Prefer the [collector] fields directly. AllowPayloads bool `koanf:"allow_payloads"` SendRequestBody bool `koanf:"send_request_body"` SendResponseBody bool `koanf:"send_response_body"` @@ -70,6 +90,22 @@ type AnalyticsPublishersConfig struct { Moesif MoesifPublisherConfig `koanf:"moesif"` } +// TrafficLoggingConfig holds configuration for the stdout traffic-logging feature, +// which writes each collected event to stdout as a JSON line. It is a consumer of +// the collector; enabling it implicitly activates the collector. +type TrafficLoggingConfig struct { + // Enabled turns stdout JSON traffic logging on. + Enabled bool `koanf:"enabled"` + // MaskedHeaders lists header names (case-insensitive) whose values are + // redacted in the logged requestHeaders/responseHeaders. + MaskedHeaders []string `koanf:"masked_headers"` + // MaxPayloadSize caps the number of bytes of request/response payload written + // to the log line (0 = no limit). Truncation is applied at the publisher, so + // the collector still captures the full body and other consumers (e.g. Moesif) + // are unaffected. + MaxPayloadSize int `koanf:"max_payload_size"` +} + // MoesifPublisherConfig holds Moesif-specific configuration type MoesifPublisherConfig struct { ApplicationID string `koanf:"application_id"` @@ -304,6 +340,22 @@ func Load(configPath string) (*Config, error) { return cfg, nil } +// defaultAccessLogsServiceConfig returns the default policy-engine ALS receiver tuning. +// Shared by the collector (canonical) and the deprecated [analytics].access_logs_service +// alias so a partial alias override migrates cleanly. +func defaultAccessLogsServiceConfig() AccessLogsServiceConfig { + return AccessLogsServiceConfig{ + Mode: "", + ServerPort: 18090, + ShutdownTimeout: 600 * time.Second, + PublicKeyPath: "", + PrivateKeyPath: "", + ALSPlainText: true, + ExtProcMaxMessageSize: 1000000000, + ExtProcMaxHeaderLimit: 8192, + } +} + // defaultConfig returns a Config struct with default configuration values func defaultConfig() *Config { return &Config{ @@ -350,6 +402,16 @@ func defaultConfig() *Config { }, TracingServiceName: "policy-engine", }, + Collector: CollectorConfig{ + SendRequestBody: false, + SendResponseBody: false, + AccessLogsServiceCfg: defaultAccessLogsServiceConfig(), + }, + TrafficLogging: TrafficLoggingConfig{ + Enabled: false, + MaskedHeaders: []string{}, + MaxPayloadSize: 0, + }, Analytics: AnalyticsConfig{ Enabled: false, EnabledPublishers: []string{"moesif"}, @@ -369,19 +431,12 @@ func defaultConfig() *Config { "buffer_size_bytes": 16384, "grpc_request_timeout": 20000000000, }, - AccessLogsServiceCfg: AccessLogsServiceConfig{ - Mode: "", - ServerPort: 18090, - ShutdownTimeout: 600 * time.Second, - PublicKeyPath: "", - PrivateKeyPath: "", - ALSPlainText: true, - ExtProcMaxMessageSize: 1000000000, - ExtProcMaxHeaderLimit: 8192, - }, - AllowPayloads: false, - SendRequestBody: false, - SendResponseBody: false, + // Deprecated alias: default mirrors the collector so a partial + // [analytics.access_logs_service] override migrates cleanly. + AccessLogsServiceCfg: defaultAccessLogsServiceConfig(), + AllowPayloads: false, + SendRequestBody: false, + SendResponseBody: false, }, TracingConfig: TracingConfig{ Enabled: false, @@ -480,6 +535,12 @@ func (c *Config) Validate() error { return fmt.Errorf("invalid logging.format: %s (must be json or text)", c.PolicyEngine.Logging.Format) } + if err := c.validateCollectorConfig(); err != nil { + return err + } + if c.TrafficLogging.MaxPayloadSize < 0 { + return fmt.Errorf("traffic_logging.max_payload_size must be >= 0, got %d", c.TrafficLogging.MaxPayloadSize) + } if c.Analytics.Enabled { if err := c.validateAnalyticsConfig(); err != nil { return fmt.Errorf("analytics configuration validation failed: %v", err) @@ -536,49 +597,107 @@ func (c *Config) validateXDSConfig() error { return nil } -// validateAnalyticsConfig validates the analytics configuration -func (c *Config) validateAnalyticsConfig() error { - // Validate analytics configuration - if c.Analytics.Enabled { - // Migration path for deprecated analytics.allow_payloads. - // Runs when both directional flags are false, which is indistinguishable - // from "not set" because bool fields cannot represent unset vs explicit false. - if c.Analytics.AllowPayloads { - slog.Warn("analytics.allow_payloads is deprecated; use analytics.send_request_body and analytics.send_response_body instead") - if !c.Analytics.SendRequestBody && !c.Analytics.SendResponseBody { - c.Analytics.SendRequestBody = true - c.Analytics.SendResponseBody = true - } +// validateCollectorConfig migrates deprecated analytics capture aliases onto the +// collector and enforces the collector prerequisite: a consumer (analytics or +// traffic logging) requires the collector that feeds it. The collector has no +// on/off flag of its own: it is implicitly active whenever a consumer is enabled +// (see IsCollectorEnabled), so its transport is validated only in that case. +func (c *Config) validateCollectorConfig() error { + c.migrateDeprecatedAnalyticsCapture() + c.migrateDeprecatedAnalyticsTransport() + + if c.IsCollectorEnabled() { + if err := validateAccessLogsServiceConfig(c.Collector.AccessLogsServiceCfg); err != nil { + return err } + } + return nil +} - // Validate ALS server config (policy-engine side) - als := c.Analytics.AccessLogsServiceCfg - - // Validate ALS connection mode - switch als.Mode { - case "uds", "": - // UDS mode (default) - port is unused - case "tcp": - // TCP mode - validate port - if als.ServerPort <= 0 || als.ServerPort > 65535 { - return fmt.Errorf("analytics.access_logs_service.server_port must be between 1 and 65535, got %d", als.ServerPort) - } - default: - return fmt.Errorf("analytics.access_logs_service.mode must be 'uds' or 'tcp', got: %s", als.Mode) - } - if als.ShutdownTimeout <= 0 { - return fmt.Errorf("analytics.access_logs_service.shutdown_timeout must be positive, got %s", als.ShutdownTimeout) - } - if als.ExtProcMaxMessageSize <= 0 { - return fmt.Errorf("analytics.access_logs_service.max_message_size must be positive, got %d", als.ExtProcMaxMessageSize) +// IsCollectorEnabled reports whether the collector should run. The collector is +// implicit: it is active whenever any consumer of the collected data is enabled +// (analytics or stdout traffic logging), and off otherwise. +func (c *Config) IsCollectorEnabled() bool { + return c.Analytics.Enabled || c.TrafficLogging.Enabled +} + +// migrateDeprecatedAnalyticsTransport maps a deprecated [analytics].access_logs_service +// override onto the collector when the collector's receiver tuning is still at its +// default, so existing configs keep working after the transport moved to [collector]. +func (c *Config) migrateDeprecatedAnalyticsTransport() { + def := defaultAccessLogsServiceConfig() + if c.Analytics.AccessLogsServiceCfg != def { + if c.Collector.AccessLogsServiceCfg == def { + slog.Warn("analytics.access_logs_service is deprecated; migrating it to collector.als") + c.Collector.AccessLogsServiceCfg = c.Analytics.AccessLogsServiceCfg + } else { + slog.Warn("analytics.access_logs_service is deprecated and collector.als is already configured; ignoring the analytics.access_logs_service override") } - if als.ExtProcMaxHeaderLimit <= 0 { - return fmt.Errorf("analytics.access_logs_service.max_header_limit must be positive, got %d", als.ExtProcMaxHeaderLimit) + } +} + +// validateAccessLogsServiceConfig validates the policy-engine ALS receiver tuning. +func validateAccessLogsServiceConfig(als AccessLogsServiceConfig) error { + switch als.Mode { + case "uds", "": + // UDS mode (default) - port is unused + case "tcp": + if als.ServerPort <= 0 || als.ServerPort > 65535 { + return fmt.Errorf("collector.als.server_port must be between 1 and 65535, got %d", als.ServerPort) } - if als.ExtProcMaxHeaderLimit > math.MaxUint32 { - return fmt.Errorf("analytics.access_logs_service.max_header_limit must be <= %d, got %d", uint64(math.MaxUint32), als.ExtProcMaxHeaderLimit) + default: + return fmt.Errorf("collector.als.mode must be 'uds' or 'tcp', got: %s", als.Mode) + } + if als.ShutdownTimeout <= 0 { + return fmt.Errorf("collector.als.shutdown_timeout must be positive, got %s", als.ShutdownTimeout) + } + if als.ExtProcMaxMessageSize <= 0 { + return fmt.Errorf("collector.als.max_message_size must be positive, got %d", als.ExtProcMaxMessageSize) + } + if als.ExtProcMaxHeaderLimit <= 0 { + return fmt.Errorf("collector.als.max_header_limit must be positive, got %d", als.ExtProcMaxHeaderLimit) + } + if als.ExtProcMaxHeaderLimit > math.MaxUint32 { + return fmt.Errorf("collector.als.max_header_limit must be <= %d, got %d", uint64(math.MaxUint32), als.ExtProcMaxHeaderLimit) + } + return nil +} + +// migrateDeprecatedAnalyticsCapture maps the deprecated analytics.allow_payloads / +// analytics.send_request_body / analytics.send_response_body onto the collector's +// body-capture flags (when the collector flag is not already set), so existing +// configs keep working after capture settings moved under [collector]. These are +// analytics's own deprecated flags, so they are only honored while analytics is +// enabled — otherwise a stale value left over from a disabled analytics setup +// could silently turn on body capture for an unrelated consumer (e.g. +// traffic_logging) enabled later. +func (c *Config) migrateDeprecatedAnalyticsCapture() { + if !c.Analytics.Enabled { + return + } + // Directional aliases take precedence over allow_payloads. + if c.Analytics.SendRequestBody && !c.Collector.SendRequestBody { + slog.Warn("analytics.send_request_body is deprecated; use collector.send_request_body instead") + c.Collector.SendRequestBody = true + } + if c.Analytics.SendResponseBody && !c.Collector.SendResponseBody { + slog.Warn("analytics.send_response_body is deprecated; use collector.send_response_body instead") + c.Collector.SendResponseBody = true + } + // allow_payloads only fills in when no directional body capture is configured. + if c.Analytics.AllowPayloads { + slog.Warn("analytics.allow_payloads is deprecated; use collector.send_request_body and collector.send_response_body instead") + if !c.Collector.SendRequestBody && !c.Collector.SendResponseBody { + c.Collector.SendRequestBody = true + c.Collector.SendResponseBody = true } + } +} +// validateAnalyticsConfig validates the analytics consumer configuration (publishers). +// ALS transport validation lives in validateCollectorConfig. +func (c *Config) validateAnalyticsConfig() error { + if c.Analytics.Enabled { // Validate enabled publishers for _, publisherName := range c.Analytics.EnabledPublishers { switch publisherName { diff --git a/gateway/gateway-runtime/policy-engine/internal/config/config_test.go b/gateway/gateway-runtime/policy-engine/internal/config/config_test.go index 3470b998c9..914b6853b3 100644 --- a/gateway/gateway-runtime/policy-engine/internal/config/config_test.go +++ b/gateway/gateway-runtime/policy-engine/internal/config/config_test.go @@ -69,8 +69,15 @@ func validConfig() *Config { Timeout: 30 * time.Second, }, }, + // The collector is implicit (active whenever a consumer is enabled). ALS + // receiver defaults mirror production so transport validation passes and the + // deprecated alias stays neutral (no spurious migration). + Collector: CollectorConfig{ + AccessLogsServiceCfg: defaultAccessLogsServiceConfig(), + }, Analytics: AnalyticsConfig{ - Enabled: false, + Enabled: false, + AccessLogsServiceCfg: defaultAccessLogsServiceConfig(), }, TracingConfig: TracingConfig{ Enabled: false, @@ -948,7 +955,7 @@ func TestValidate_AnalyticsConfig(t *testing.T) { name: "analytics enabled - valid UDS config (default)", setup: func(cfg *Config) { cfg.Analytics.Enabled = true - cfg.Analytics.AccessLogsServiceCfg = AccessLogsServiceConfig{ + cfg.Collector.AccessLogsServiceCfg = AccessLogsServiceConfig{ Mode: "uds", ShutdownTimeout: 600 * time.Second, ExtProcMaxMessageSize: 1000000, @@ -961,7 +968,7 @@ func TestValidate_AnalyticsConfig(t *testing.T) { name: "analytics enabled - valid UDS config (empty mode defaults to uds)", setup: func(cfg *Config) { cfg.Analytics.Enabled = true - cfg.Analytics.AccessLogsServiceCfg = AccessLogsServiceConfig{ + cfg.Collector.AccessLogsServiceCfg = AccessLogsServiceConfig{ Mode: "", ShutdownTimeout: 600 * time.Second, ExtProcMaxMessageSize: 1000000, @@ -974,7 +981,7 @@ func TestValidate_AnalyticsConfig(t *testing.T) { name: "analytics enabled - valid TCP config", setup: func(cfg *Config) { cfg.Analytics.Enabled = true - cfg.Analytics.AccessLogsServiceCfg = AccessLogsServiceConfig{ + cfg.Collector.AccessLogsServiceCfg = AccessLogsServiceConfig{ Mode: "tcp", ServerPort: 18090, ShutdownTimeout: 600 * time.Second, @@ -988,7 +995,7 @@ func TestValidate_AnalyticsConfig(t *testing.T) { name: "analytics enabled - invalid mode", setup: func(cfg *Config) { cfg.Analytics.Enabled = true - cfg.Analytics.AccessLogsServiceCfg = AccessLogsServiceConfig{ + cfg.Collector.AccessLogsServiceCfg = AccessLogsServiceConfig{ Mode: "invalid", ShutdownTimeout: 600 * time.Second, ExtProcMaxMessageSize: 1000000, @@ -1002,7 +1009,7 @@ func TestValidate_AnalyticsConfig(t *testing.T) { name: "analytics enabled - TCP mode invalid ALS port", setup: func(cfg *Config) { cfg.Analytics.Enabled = true - cfg.Analytics.AccessLogsServiceCfg = AccessLogsServiceConfig{ + cfg.Collector.AccessLogsServiceCfg = AccessLogsServiceConfig{ Mode: "tcp", ServerPort: 0, ShutdownTimeout: 600 * time.Second, @@ -1017,7 +1024,7 @@ func TestValidate_AnalyticsConfig(t *testing.T) { name: "analytics enabled - UDS mode skips port validation", setup: func(cfg *Config) { cfg.Analytics.Enabled = true - cfg.Analytics.AccessLogsServiceCfg = AccessLogsServiceConfig{ + cfg.Collector.AccessLogsServiceCfg = AccessLogsServiceConfig{ Mode: "uds", ServerPort: 0, // Invalid port, but irrelevant in UDS mode ShutdownTimeout: 600 * time.Second, @@ -1031,7 +1038,7 @@ func TestValidate_AnalyticsConfig(t *testing.T) { name: "analytics enabled - invalid shutdown timeout", setup: func(cfg *Config) { cfg.Analytics.Enabled = true - cfg.Analytics.AccessLogsServiceCfg = AccessLogsServiceConfig{ + cfg.Collector.AccessLogsServiceCfg = AccessLogsServiceConfig{ ShutdownTimeout: 0, ExtProcMaxMessageSize: 1000000, ExtProcMaxHeaderLimit: 8192, @@ -1044,7 +1051,7 @@ func TestValidate_AnalyticsConfig(t *testing.T) { name: "analytics enabled - invalid max message size", setup: func(cfg *Config) { cfg.Analytics.Enabled = true - cfg.Analytics.AccessLogsServiceCfg = AccessLogsServiceConfig{ + cfg.Collector.AccessLogsServiceCfg = AccessLogsServiceConfig{ ShutdownTimeout: 600 * time.Second, ExtProcMaxMessageSize: 0, ExtProcMaxHeaderLimit: 8192, @@ -1057,7 +1064,7 @@ func TestValidate_AnalyticsConfig(t *testing.T) { name: "analytics enabled - invalid max header limit", setup: func(cfg *Config) { cfg.Analytics.Enabled = true - cfg.Analytics.AccessLogsServiceCfg = AccessLogsServiceConfig{ + cfg.Collector.AccessLogsServiceCfg = AccessLogsServiceConfig{ ShutdownTimeout: 600 * time.Second, ExtProcMaxMessageSize: 1000000, ExtProcMaxHeaderLimit: 0, @@ -1070,7 +1077,7 @@ func TestValidate_AnalyticsConfig(t *testing.T) { name: "analytics enabled - max header limit exceeds uint32", setup: func(cfg *Config) { cfg.Analytics.Enabled = true - cfg.Analytics.AccessLogsServiceCfg = AccessLogsServiceConfig{ + cfg.Collector.AccessLogsServiceCfg = AccessLogsServiceConfig{ ShutdownTimeout: 600 * time.Second, ExtProcMaxMessageSize: 1000000, ExtProcMaxHeaderLimit: math.MaxInt, @@ -1099,8 +1106,8 @@ func TestValidate_AnalyticsConfig(t *testing.T) { func TestValidate_AnalyticsPayloadMigration(t *testing.T) { setValidAnalyticsALS := func(cfg *Config) { - cfg.Analytics.Enabled = true - cfg.Analytics.AccessLogsServiceCfg = AccessLogsServiceConfig{ + cfg.Analytics.Enabled = true // a consumer being on makes the collector implicit + cfg.Collector.AccessLogsServiceCfg = AccessLogsServiceConfig{ Mode: "uds", ShutdownTimeout: 600 * time.Second, ExtProcMaxMessageSize: 1000000, @@ -1160,12 +1167,68 @@ func TestValidate_AnalyticsPayloadMigration(t *testing.T) { err := cfg.Validate() require.NoError(t, err) - assert.Equal(t, tt.wantSendReq, cfg.Analytics.SendRequestBody) - assert.Equal(t, tt.wantSendResp, cfg.Analytics.SendResponseBody) + // Deprecated analytics body aliases now migrate onto the collector. + assert.Equal(t, tt.wantSendReq, cfg.Collector.SendRequestBody) + assert.Equal(t, tt.wantSendResp, cfg.Collector.SendResponseBody) }) } } +// TestValidate_AnalyticsPayloadMigration_SkippedWhenAnalyticsDisabled guards +// against a stale analytics.allow_payloads left over from a disabled analytics +// setup silently turning on body capture for an unrelated consumer (traffic +// logging) enabled later. The deprecated capture aliases belong to analytics, so +// they must only be honored while analytics itself is enabled. +func TestValidate_AnalyticsPayloadMigration_SkippedWhenAnalyticsDisabled(t *testing.T) { + cfg := validConfig() + cfg.Analytics.Enabled = false + cfg.TrafficLogging.Enabled = true // an unrelated consumer activates the collector + cfg.Analytics.AllowPayloads = true + + err := cfg.Validate() + require.NoError(t, err) + assert.False(t, cfg.Collector.SendRequestBody) + assert.False(t, cfg.Collector.SendResponseBody) +} + +// TestValidate_CollectorPrerequisite verifies that enabling a consumer (analytics, +// traffic logging) without the collector auto-enables the collector (a +// backward-compat soft prerequisite) rather than failing. +// TestIsCollectorEnabled covers the implicit collector: it is active iff a consumer +// (analytics or traffic logging) is enabled, and off otherwise. +func TestIsCollectorEnabled(t *testing.T) { + t.Run("no consumers -> off", func(t *testing.T) { + cfg := validConfig() + cfg.Analytics.Enabled = false + cfg.TrafficLogging.Enabled = false + assert.False(t, cfg.IsCollectorEnabled()) + }) + + t.Run("analytics on -> collector on", func(t *testing.T) { + cfg := validConfig() + cfg.Analytics.Enabled = true + cfg.Analytics.EnabledPublishers = []string{} + assert.True(t, cfg.IsCollectorEnabled()) + require.NoError(t, cfg.Validate()) + }) + + t.Run("traffic logging on -> collector on", func(t *testing.T) { + cfg := validConfig() + cfg.TrafficLogging.Enabled = true + assert.True(t, cfg.IsCollectorEnabled()) + require.NoError(t, cfg.Validate()) + }) +} + +func TestValidate_TrafficLoggingMaxPayloadSize(t *testing.T) { + cfg := validConfig() + cfg.TrafficLogging.Enabled = true + cfg.TrafficLogging.MaxPayloadSize = -1 + err := cfg.Validate() + require.Error(t, err) + assert.Contains(t, err.Error(), "traffic_logging.max_payload_size") +} + // TestValidate_AnalyticsPublishers tests analytics publisher validation func TestValidate_AnalyticsPublishers(t *testing.T) { tests := []struct { @@ -1178,7 +1241,7 @@ func TestValidate_AnalyticsPublishers(t *testing.T) { name: "no publishers enabled", setup: func(cfg *Config) { cfg.Analytics.Enabled = true - cfg.Analytics.AccessLogsServiceCfg = AccessLogsServiceConfig{ + cfg.Collector.AccessLogsServiceCfg = AccessLogsServiceConfig{ ServerPort: 18090, ShutdownTimeout: 600 * time.Second, ExtProcMaxMessageSize: 1000000, @@ -1192,7 +1255,7 @@ func TestValidate_AnalyticsPublishers(t *testing.T) { name: "unknown publisher type", setup: func(cfg *Config) { cfg.Analytics.Enabled = true - cfg.Analytics.AccessLogsServiceCfg = AccessLogsServiceConfig{ + cfg.Collector.AccessLogsServiceCfg = AccessLogsServiceConfig{ ServerPort: 18090, ShutdownTimeout: 600 * time.Second, ExtProcMaxMessageSize: 1000000, @@ -1207,7 +1270,7 @@ func TestValidate_AnalyticsPublishers(t *testing.T) { name: "moesif publisher - missing application_id", setup: func(cfg *Config) { cfg.Analytics.Enabled = true - cfg.Analytics.AccessLogsServiceCfg = AccessLogsServiceConfig{ + cfg.Collector.AccessLogsServiceCfg = AccessLogsServiceConfig{ ServerPort: 18090, ShutdownTimeout: 600 * time.Second, ExtProcMaxMessageSize: 1000000, @@ -1223,7 +1286,7 @@ func TestValidate_AnalyticsPublishers(t *testing.T) { name: "moesif publisher - invalid publish_interval", setup: func(cfg *Config) { cfg.Analytics.Enabled = true - cfg.Analytics.AccessLogsServiceCfg = AccessLogsServiceConfig{ + cfg.Collector.AccessLogsServiceCfg = AccessLogsServiceConfig{ ServerPort: 18090, ShutdownTimeout: 600 * time.Second, ExtProcMaxMessageSize: 1000000, @@ -1240,7 +1303,7 @@ func TestValidate_AnalyticsPublishers(t *testing.T) { name: "moesif publisher - invalid base_url", setup: func(cfg *Config) { cfg.Analytics.Enabled = true - cfg.Analytics.AccessLogsServiceCfg = AccessLogsServiceConfig{ + cfg.Collector.AccessLogsServiceCfg = AccessLogsServiceConfig{ ServerPort: 18090, ShutdownTimeout: 600 * time.Second, ExtProcMaxMessageSize: 1000000, @@ -1258,7 +1321,7 @@ func TestValidate_AnalyticsPublishers(t *testing.T) { name: "moesif publisher - valid config", setup: func(cfg *Config) { cfg.Analytics.Enabled = true - cfg.Analytics.AccessLogsServiceCfg = AccessLogsServiceConfig{ + cfg.Collector.AccessLogsServiceCfg = AccessLogsServiceConfig{ ServerPort: 18090, ShutdownTimeout: 600 * time.Second, ExtProcMaxMessageSize: 1000000, diff --git a/gateway/gateway-runtime/policy-engine/internal/utils/access_logger_server.go b/gateway/gateway-runtime/policy-engine/internal/utils/access_logger_server.go index 87049d7d8a..b68babbe08 100644 --- a/gateway/gateway-runtime/policy-engine/internal/utils/access_logger_server.go +++ b/gateway/gateway-runtime/policy-engine/internal/utils/access_logger_server.go @@ -90,15 +90,15 @@ func StartAccessLogServiceServer(cfg *config.Config) *grpc.Server { Timeout: 20 * time.Second, } maxHeaderListSize, err := checkedUInt32FromPositiveInt( - "analytics.access_logs_service.max_header_limit", - cfg.Analytics.AccessLogsServiceCfg.ExtProcMaxHeaderLimit, + "collector.als.max_header_limit", + cfg.Collector.AccessLogsServiceCfg.ExtProcMaxHeaderLimit, ) if err != nil { panic(err) } - server, err := CreateGRPCServer(cfg.Analytics.AccessLogsServiceCfg.PublicKeyPath, - cfg.Analytics.AccessLogsServiceCfg.PrivateKeyPath, cfg.Analytics.AccessLogsServiceCfg.ALSPlainText, - grpc.MaxRecvMsgSize(cfg.Analytics.AccessLogsServiceCfg.ExtProcMaxMessageSize), + server, err := CreateGRPCServer(cfg.Collector.AccessLogsServiceCfg.PublicKeyPath, + cfg.Collector.AccessLogsServiceCfg.PrivateKeyPath, cfg.Collector.AccessLogsServiceCfg.ALSPlainText, + grpc.MaxRecvMsgSize(cfg.Collector.AccessLogsServiceCfg.ExtProcMaxMessageSize), grpc.MaxHeaderListSize(maxHeaderListSize), grpc.KeepaliveParams(kaParams)) if err != nil { @@ -109,7 +109,7 @@ func StartAccessLogServiceServer(cfg *config.Config) *grpc.Server { // Create listener based on mode (same pattern as ext_proc in main.go) var listener net.Listener - alsMode := cfg.Analytics.AccessLogsServiceCfg.Mode + alsMode := cfg.Collector.AccessLogsServiceCfg.Mode if alsMode == "" { alsMode = "uds" } @@ -139,14 +139,14 @@ func StartAccessLogServiceServer(cfg *config.Config) *grpc.Server { } }() case "tcp": - listener, err = net.Listen("tcp", fmt.Sprintf(":%d", cfg.Analytics.AccessLogsServiceCfg.ServerPort)) + listener, err = net.Listen("tcp", fmt.Sprintf(":%d", cfg.Collector.AccessLogsServiceCfg.ServerPort)) if err != nil { - slog.Error("Failed to listen on ALS TCP port", "port", cfg.Analytics.AccessLogsServiceCfg.ServerPort) + slog.Error("Failed to listen on ALS TCP port", "port", cfg.Collector.AccessLogsServiceCfg.ServerPort) panic(err) } go func() { - slog.Info("Starting to serve access log service server", "mode", "tcp", "port", cfg.Analytics.AccessLogsServiceCfg.ServerPort) + slog.Info("Starting to serve access log service server", "mode", "tcp", "port", cfg.Collector.AccessLogsServiceCfg.ServerPort) if err := server.Serve(listener); err != nil { slog.Error("ALS server exited", "error", err) } diff --git a/gateway/gateway-runtime/policy-engine/internal/utils/access_logger_server_test.go b/gateway/gateway-runtime/policy-engine/internal/utils/access_logger_server_test.go index fd6dba08d7..3176524241 100644 --- a/gateway/gateway-runtime/policy-engine/internal/utils/access_logger_server_test.go +++ b/gateway/gateway-runtime/policy-engine/internal/utils/access_logger_server_test.go @@ -210,10 +210,7 @@ func TestStreamAccessLogs_MultipleMessages(t *testing.T) { func TestStartAccessLogServiceServer_TCP(t *testing.T) { cfg := &config.Config{ - Analytics: config.AnalyticsConfig{ - Enabled: false, - EnabledPublishers: []string{}, - Publishers: config.AnalyticsPublishersConfig{}, + Collector: config.CollectorConfig{ AccessLogsServiceCfg: config.AccessLogsServiceConfig{ Mode: "tcp", ServerPort: 19001, // Use non-standard port to avoid conflicts diff --git a/gateway/system-policies/analytics/analytics.go b/gateway/system-policies/analytics/analytics.go index e0771e103c..7827cd34b0 100644 --- a/gateway/system-policies/analytics/analytics.go +++ b/gateway/system-policies/analytics/analytics.go @@ -152,6 +152,14 @@ func (a *AnalyticsPolicy) OnRequestHeaders(_ context.Context, reqCtx *policy.Req } } + // Capture all request headers when enabled, so they flow into analytics events + // (and the stdout/log publisher) without attaching a per-API header policy. + if sendReqHeaders, _ := getHeaderFlags(params); sendReqHeaders && reqCtx.Headers != nil { + if headers := serializeHeaders(reqCtx.Headers); headers != "" { + analyticsMetadata["request_headers"] = headers + } + } + if len(analyticsMetadata) > 0 { return policy.UpstreamRequestHeaderModifications{AnalyticsMetadata: analyticsMetadata} } @@ -198,6 +206,13 @@ func (a *AnalyticsPolicy) OnResponseHeaders(_ context.Context, respCtx *policy.R } } + // Capture all response headers when enabled. + if _, sendRespHeaders := getHeaderFlags(params); sendRespHeaders && respCtx.ResponseHeaders != nil { + if headers := serializeHeaders(respCtx.ResponseHeaders); headers != "" { + analyticsMetadata["response_headers"] = headers + } + } + if len(analyticsMetadata) > 0 { return policy.DownstreamResponseHeaderModifications{AnalyticsMetadata: analyticsMetadata} } @@ -883,18 +898,6 @@ func getPayloadFlags(params map[string]interface{}) (sendRequestBody, sendRespon return false, false } - parseBoolLike := func(v interface{}) bool { - switch val := v.(type) { - case bool: - return val - case string: - lower := strings.ToLower(strings.TrimSpace(val)) - return lower == "true" || lower == "1" || lower == "yes" - default: - return false - } - } - hasReq, hasResp := false, false if raw, ok := params["send_request_body"]; ok { @@ -922,6 +925,55 @@ func getPayloadFlags(params map[string]interface{}) (sendRequestBody, sendRespon return false, false } +// parseBoolLike interprets bool and common string ("true"/"1"/"yes") representations +// of a boolean policy parameter. +func parseBoolLike(v interface{}) bool { + switch val := v.(type) { + case bool: + return val + case string: + lower := strings.ToLower(strings.TrimSpace(val)) + return lower == "true" || lower == "1" || lower == "yes" + default: + return false + } +} + +// getHeaderFlags derives per-direction header capture flags from policy parameters. +func getHeaderFlags(params map[string]interface{}) (sendRequestHeaders, sendResponseHeaders bool) { + if params == nil { + return false, false + } + if raw, ok := params["send_request_headers"]; ok { + sendRequestHeaders = parseBoolLike(raw) + } + if raw, ok := params["send_response_headers"]; ok { + sendResponseHeaders = parseBoolLike(raw) + } + return sendRequestHeaders, sendResponseHeaders +} + +// serializeHeaders renders all headers as a JSON object string ({"name":"v1, v2"}), +// matching the request_headers/response_headers format the analytics engine reads. +// Returns "" when there are no headers. Sensitive values are not masked here; the +// stdout/log publisher applies masked_headers on output. +func serializeHeaders(headers *policy.Headers) string { + all := headers.GetAll() + if len(all) == 0 { + return "" + } + flat := make(map[string]string, len(all)) + for name, values := range all { + flat[name] = strings.Join(values, ", ") + } + data, err := json.Marshal(flat) + if err != nil { + slog.Error("Failed to marshal headers for analytics", "error", err) + return "" + } + return string(data) +} + // Helper to extract string values via JSONPath func extractStringFromJsonpath(payload map[string]interface{}, path string) string { val, err := utils.ExtractValueFromJsonpath(payload, path) diff --git a/gateway/system-policies/analytics/analytics_headers_test.go b/gateway/system-policies/analytics/analytics_headers_test.go new file mode 100644 index 0000000000..aed92c9f6b --- /dev/null +++ b/gateway/system-policies/analytics/analytics_headers_test.go @@ -0,0 +1,63 @@ +package analytics + +import ( + "encoding/json" + "testing" + + policy "github.com/wso2/api-platform/sdk/core/policy/v1alpha2" +) + +func TestGetHeaderFlags(t *testing.T) { + cases := []struct { + name string + params map[string]interface{} + wantReq bool + wantResp bool + }{ + {"nil params", nil, false, false}, + {"absent", map[string]interface{}{}, false, false}, + {"bool true", map[string]interface{}{"send_request_headers": true, "send_response_headers": true}, true, true}, + {"string true", map[string]interface{}{"send_request_headers": "true"}, true, false}, + {"mixed", map[string]interface{}{"send_request_headers": false, "send_response_headers": "yes"}, false, true}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + gotReq, gotResp := getHeaderFlags(c.params) + if gotReq != c.wantReq || gotResp != c.wantResp { + t.Fatalf("getHeaderFlags(%v) = (%v, %v), want (%v, %v)", c.params, gotReq, gotResp, c.wantReq, c.wantResp) + } + }) + } +} + +func TestSerializeHeaders(t *testing.T) { + // Empty headers -> empty string. + if got := serializeHeaders(policy.NewHeaders(nil)); got != "" { + t.Fatalf("serializeHeaders(empty) = %q, want \"\"", got) + } + + h := policy.NewHeaders(map[string][]string{ + "Authorization": {"Bearer secret"}, + "X-Foo": {"a", "b"}, + }) + got := serializeHeaders(h) + if got == "" { + t.Fatal("serializeHeaders returned empty for non-empty headers") + } + + var decoded map[string]string + if err := json.Unmarshal([]byte(got), &decoded); err != nil { + t.Fatalf("output is not valid JSON: %v (%q)", err, got) + } + // NewHeaders lower-cases keys; multi-value headers are joined with ", ". + if decoded["authorization"] != "Bearer secret" { + t.Errorf("authorization = %q, want %q", decoded["authorization"], "Bearer secret") + } + if decoded["x-foo"] != "a, b" { + t.Errorf("x-foo = %q, want %q", decoded["x-foo"], "a, b") + } +} + +// Note: payload-size capping moved out of the capture path (the collector now +// captures full bodies); truncation is applied output-side by the traffic-logging +// publisher (traffic_logging.max_payload_size) and tested there. diff --git a/gateway/system-policies/analytics/policy-definition.yaml b/gateway/system-policies/analytics/policy-definition.yaml index 7fe915469d..c3c223bd8a 100644 --- a/gateway/system-policies/analytics/policy-definition.yaml +++ b/gateway/system-policies/analytics/policy-definition.yaml @@ -25,6 +25,22 @@ parameters: into analytics metadata as response_payload, for both buffered and streaming responses, which is then forwarded through the analytics pipeline (including publishers like Moesif). + send_request_headers: + type: boolean + default: false + description: > + When true, the analytics system policy captures ALL request headers + into analytics metadata as request_headers, which is forwarded through + the analytics pipeline (including the stdout/log and Moesif publishers). + Sensitive values are not masked here; the log publisher's masked_headers + setting redacts them on output. + send_response_headers: + type: boolean + default: false + description: > + When true, the analytics system policy captures ALL response headers + into analytics metadata as response_headers, which is forwarded through + the analytics pipeline (including the stdout/log and Moesif publishers). allow_payloads: type: boolean default: false