[APIP] Add Access Logging support

common/collector/collector.go

@@ -0,0 +1,121 @@
+/*
+ * Copyright (c) 2026, WSO2 LLC. (https://www.wso2.com).
+ *
+ * WSO2 LLC. licenses this file to you 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 collector holds logic shared by the gateway-controller and
+// policy-engine's [collector] configuration: the collector is the shared
+// data-capture pipeline (system policy + ALS transport) that gathers
+// request/response headers and bodies for every consumer (analytics, stdout
+// traffic logging). It has no on/off flag of its own — it is implicit,
+// derived from whether a consumer is enabled — and both modules migrate the
+// same set of deprecated [analytics] aliases onto it at load time. The two
+// modules keep their own Config/CollectorConfig struct shapes (their transport
+// types differ: GRPCEventServerConfig vs AccessLogsServiceConfig), so this
+// package holds only the derivation/migration logic, not the structs.
+package collector
+
+import "log/slog"
+
+// ServerPort is the fixed TCP port for the Envoy → policy-engine ALS
+// (access-log service) gRPC stream in "tcp" mode: the gateway-controller
+// dials it (translator.go, Envoy cluster config) and the policy-engine
+// listens on it (access_logger_server.go). It is intentionally
+// non-configurable — the two sides previously had independent config fields
+// that defaulted to the same value but could drift apart, silently breaking
+// the connection; a single shared constant makes that impossible.
+const ServerPort = 18090
+
+// IsEnabled reports whether the collector should run: implicitly active
+// whenever any consumer of the collected data (analytics or stdout traffic
+// logging) is enabled, and off otherwise.
+func IsEnabled(analyticsEnabled, trafficLoggingEnabled bool) bool {
+	return analyticsEnabled || trafficLoggingEnabled
+}
+
+// CaptureFlags holds the deprecated [analytics] body-capture aliases being
+// migrated onto [collector].
+type CaptureFlags struct {
+	SendRequestBody  bool
+	SendResponseBody bool
+	AllowPayloads    bool
+}
+
+// MigrateDeprecatedCapture 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 analyticsEnabled is true — 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. Directional
+// aliases (send_request_body/send_response_body) take precedence over
+// allow_payloads, which only fills in when neither directional flag is
+// already set on the collector.
+func MigrateDeprecatedCapture(analyticsEnabled bool, deprecated CaptureFlags, collectorSendRequestBody, collectorSendResponseBody *bool) {
+	if !analyticsEnabled {
+		return
+	}
+	if deprecated.SendRequestBody {
+		if !*collectorSendRequestBody {
+			slog.Warn("analytics.send_request_body is deprecated; use collector.request_body instead")
+			*collectorSendRequestBody = true
+		} else {
+			slog.Warn("analytics.send_request_body is deprecated and collector.request_body is already configured; ignoring the analytics.send_request_body override")
+		}
+	}
+	if deprecated.SendResponseBody {
+		if !*collectorSendResponseBody {
+			slog.Warn("analytics.send_response_body is deprecated; use collector.response_body instead")
+			*collectorSendResponseBody = true
+		} else {
+			slog.Warn("analytics.send_response_body is deprecated and collector.response_body is already configured; ignoring the analytics.send_response_body override")
+		}
+	}
+	if deprecated.AllowPayloads {
+		slog.Warn("analytics.allow_payloads is deprecated; use collector.request_body and collector.response_body instead")
+		if !*collectorSendRequestBody && !*collectorSendResponseBody {
+			*collectorSendRequestBody = true
+			*collectorSendResponseBody = true
+		}
+	}
+}
+
+// MigrateDeprecatedTransport maps a deprecated [analytics] transport-tuning
+// alias (e.g. analytics.grpc_event_server on the controller,
+// analytics.access_logs_service on the policy-engine) onto the collector's
+// transport config when the collector's copy is still at its default, so
+// existing configs keep working after transport tuning moved to
+// [collector.server]. It is generic over the two modules' differing transport
+// struct shapes (T is only ever compared and assigned, never inspected). This
+// is analytics's own deprecated field, so it is only honored while
+// analyticsEnabled is true — otherwise a stale value left over from a
+// disabled analytics setup could silently reconfigure the transport for an
+// unrelated consumer (e.g. traffic_logging) enabled later.
+//
+// deprecatedKey names the deprecated config key for the warning message
+// (e.g. "analytics.grpc_event_server").
+func MigrateDeprecatedTransport[T comparable](analyticsEnabled bool, deprecated T, collectorCfg *T, def T, deprecatedKey string) {
+	if !analyticsEnabled || deprecated == def {
+		return
+	}
+	if *collectorCfg == def {
+		slog.Warn(deprecatedKey + " is deprecated; migrating it to collector.server")
+		*collectorCfg = deprecated
+	} else {
+		slog.Warn(deprecatedKey + " is deprecated and collector.server is already configured; ignoring the " + deprecatedKey + " override")
+	}
+}

common/collector/collector_test.go

@@ -0,0 +1,105 @@
+/*
+ * Copyright (c) 2026, WSO2 LLC. (https://www.wso2.com).
+ *
+ * WSO2 LLC. licenses this file to you 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 collector
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestIsEnabled(t *testing.T) {
+	assert.False(t, IsEnabled(false, false))
+	assert.True(t, IsEnabled(true, false))
+	assert.True(t, IsEnabled(false, true))
+	assert.True(t, IsEnabled(true, true))
+}
+
+func TestMigrateDeprecatedCapture_SkippedWhenAnalyticsDisabled(t *testing.T) {
+	collectorReq, collectorResp := false, false
+	MigrateDeprecatedCapture(false, CaptureFlags{SendRequestBody: true, SendResponseBody: true, AllowPayloads: true}, &collectorReq, &collectorResp)
+	assert.False(t, collectorReq)
+	assert.False(t, collectorResp)
+}
+
+func TestMigrateDeprecatedCapture_DirectionalFlags(t *testing.T) {
+	collectorReq, collectorResp := false, false
+	MigrateDeprecatedCapture(true, CaptureFlags{SendRequestBody: true, SendResponseBody: false}, &collectorReq, &collectorResp)
+	assert.True(t, collectorReq)
+	assert.False(t, collectorResp)
+}
+
+func TestMigrateDeprecatedCapture_AllowPayloadsFillsInWhenNoDirectionalFlags(t *testing.T) {
+	collectorReq, collectorResp := false, false
+	MigrateDeprecatedCapture(true, CaptureFlags{AllowPayloads: true}, &collectorReq, &collectorResp)
+	assert.True(t, collectorReq)
+	assert.True(t, collectorResp)
+}
+
+func TestMigrateDeprecatedCapture_DirectionalFlagsWinOverAllowPayloads(t *testing.T) {
+	collectorReq, collectorResp := false, true
+	MigrateDeprecatedCapture(true, CaptureFlags{SendRequestBody: true, AllowPayloads: true}, &collectorReq, &collectorResp)
+	assert.True(t, collectorReq)
+	assert.True(t, collectorResp, "already true; allow_payloads must not be re-applied since a directional flag is set")
+}
+
+type testTransportConfig struct {
+	Port int
+	Mode string
+}
+
+func TestMigrateDeprecatedTransport_SkippedWhenAnalyticsDisabled(t *testing.T) {
+	def := testTransportConfig{Port: 18090, Mode: "uds"}
+	collectorCfg := def
+	deprecated := testTransportConfig{Port: 9999, Mode: "tcp"}
+
+	MigrateDeprecatedTransport(false, deprecated, &collectorCfg, def, "analytics.grpc_event_server")
+
+	assert.Equal(t, def, collectorCfg, "collector transport must stay at default when analytics is disabled")
+}
+
+func TestMigrateDeprecatedTransport_SkippedWhenDeprecatedAtDefault(t *testing.T) {
+	def := testTransportConfig{Port: 18090, Mode: "uds"}
+	collectorCfg := def
+
+	MigrateDeprecatedTransport(true, def, &collectorCfg, def, "analytics.grpc_event_server")
+
+	assert.Equal(t, def, collectorCfg)
+}
+
+func TestMigrateDeprecatedTransport_MigratesWhenCollectorAtDefault(t *testing.T) {
+	def := testTransportConfig{Port: 18090, Mode: "uds"}
+	collectorCfg := def
+	deprecated := testTransportConfig{Port: 9999, Mode: "tcp"}
+
+	MigrateDeprecatedTransport(true, deprecated, &collectorCfg, def, "analytics.grpc_event_server")
+
+	assert.Equal(t, deprecated, collectorCfg)
+}
+
+func TestMigrateDeprecatedTransport_DoesNotClobberExplicitCollectorConfig(t *testing.T) {
+	def := testTransportConfig{Port: 18090, Mode: "uds"}
+	explicit := testTransportConfig{Port: 7000, Mode: "tcp"}
+	collectorCfg := explicit
+	deprecated := testTransportConfig{Port: 9999, Mode: "tcp"}
+
+	MigrateDeprecatedTransport(true, deprecated, &collectorCfg, def, "analytics.grpc_event_server")
+
+	assert.Equal(t, explicit, collectorCfg, "an explicit [collector.server] override must never be clobbered by the deprecated alias")
+}

gateway/configs/config-template.toml

@@ -281,7 +281,64 @@ 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).
+request_body = false
+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)
+# Off by default; enable deliberately per consumer.
+request_headers = false
+response_headers = false
+
+# Path prefixes for which the collector produces no analytics event and no
+# traffic-log line at all (as if disabled for that one request) — e.g. health
+# checks or metrics scrapes. Enforced by Envoy itself via an access-log filter,
+# so the policy-engine never even receives the request. Matched case-sensitively
+# against the path the client called (Envoy's x-envoy-original-path, set whenever
+# a route rewrite occurred), never against the rewritten backend path. Requests
+# without that header (no rewrite occurred) are never suppressed by this filter —
+# matching bare :path would risk suppressing traffic whose backend path merely
+# happens to share the prefix. Empty by default (no suppression).
+ignore_path_prefixes = []
+# ignore_path_prefixes = ["/health", "/metrics"]
+
+# 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 (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.
+# NOTE: the TCP port (18090) is not configurable here — it is a fixed constant
+# (collector.ServerPort in common/collector) shared by both ends, so they can
+# never disagree on it. See ANALYTICS_LOG_PUBLISHER_DESIGN.md changelog.
+[collector.server]
+mode = "uds"                       # "uds" (default) or "tcp"
+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
@@ -291,6 +348,9 @@ enabled = false
 # To disable payloads entirely, set allow_payloads = false or remove it; do not
 # rely on send_*_body = false while allow_payloads remains true.
 allow_payloads = false
+# Deprecated: 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.\
 send_request_body = false
 send_response_body = false
 enabled_publishers = ["moesif"]
@@ -303,17 +363,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 are unaffected.
+max_payload_size = 0
 
 # =============================================================================
 # POLICY CONFIGURATIONS

gateway/configs/config.toml

@@ -1,17 +1,21 @@
+[collector]
+request_body = false
+response_body = false
+request_headers = false
+response_headers = false
 
 [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
-send_request_body = false
-send_response_body = false
 enabled_publishers = ["moesif"]
 
 [analytics.publishers.moesif]
 application_id = "<MOESIF_APPLICATION_ID>"
 
+[traffic_logging]
+enabled = false
+masked_headers = ["authorization", "x-api-key", "x-jwt-assertion"]
+max_payload_size = 2048
+
 [router]
 gateway_host = "*"