wso2 · Tharsanan1 · Aug 13, 2026 · Aug 13, 2026 · Aug 13, 2026 · Aug 18, 2026
@@ -467,13 +467,21 @@ Most bugs surface in logs or config dumps without needing to step through code.
 | Envoy router (in Docker) — `[rtr]` lines | `cd <REPO_ROOT>/gateway && docker compose logs --no-log-prefix gateway-runtime 2>&1 \| grep '^\[rtr\]'` |
 | Python executor (Option 2B) | `/tmp/python_executor.log` |
 
-> Why the `grep`: the `gateway-runtime` container stamps every log line with
-> one of three prefixes — `[rtr]` (Envoy router), `[pol]` (in-container PE,
-> still receives xDS pushes even in debug mode), unprefixed (the entrypoint).
+> Why the `grep`: the `gateway-runtime` container stamps every human-readable log
+> line with one of these prefixes — `[rtr]` (Envoy router), `[pol]` (in-container PE,
+> still receives xDS pushes even in debug mode), `[pye]` (Python executor, started
+> only when compiled Python policies are present), and `[ent]` (the entrypoint's
+> own startup and shutdown messages).
 > When debugging traffic you only want `[rtr]` — Envoy's access log is where
 > each request's status, upstream, and policy verdict actually surface.
 > `--no-log-prefix` drops Docker's `gateway-runtime-1  |` per-line prefix so
-> the `[rtr]` anchor is at column 0.
+> the `[rtr]` anchor is at column 0. `[pye]` lines share this same container
+> stdout; under Option 2B the executor runs on the host instead and logs to
+> `/tmp/python_executor.log` (see the table above).
+>
+> JSON output on stdout is not prefixed — the policy engine's traffic log, and
+> Envoy's access log when `router.access_logs.format = "json"` — so `grep '^\['`
+> will not match it. Those records carry a `"component"` field instead.
 
 **Controller log lines** carry `correlation_id=<uuid>` — grep on it to follow
 one request end-to-end across handler → service → xDS push:

@@ -35,9 +35,14 @@ import (
 )
 
 func init() {
-	// Set up text logger early so init() logs match main() logs
-	slog.SetDefault(slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelInfo})))
-
+	// init() runs before main() installs the configured logger, so set one up here
+	// with the same tag. Format is fixed to text: policy_engine.logging.format is
+	// not known until config loads in main().
+	slog.SetDefault(slog.New(slog.NewTextHandler(
+		newComponentPrefixWriter(os.Stdout, logComponentPrefix),
+		&slog.HandlerOptions{Level: slog.LevelInfo},
+	)))
+
 	ctx := context.Background()
 	slog.InfoContext(ctx, "Registering policies from Builder compilation")
 

@@ -822,6 +822,21 @@ func defaultGRPCEventServerConfig() GRPCEventServerConfig {
 	}
 }
 
+// routerLogComponentTag prefixes the router's text access-log lines so they stay
+// separable from the policy engine's and the Python executor's output on the
+// container's shared stdout.
+const routerLogComponentTag = "[rtr] "
+
+// textAccessLogStartsWithComponentTag reports whether a text access-log format
+// opens with routerLogComponentTag. A deployer-supplied text_format replaces the
+// default outright, so the tag can go missing without anything else noticing.
+//
+// The tag has to lead the line, not merely appear in it: attribution works by
+// anchoring on it at column 0, so a tag placed mid-format is present but useless.
+func textAccessLogStartsWithComponentTag(textFormat string) bool {
+	return strings.HasPrefix(textFormat, routerLogComponentTag)
+}
+
 // defaultConfig returns a Config struct with default configuration values
 func defaultConfig() *Config {
 	return &Config{
@@ -943,6 +958,9 @@ func defaultConfig() *Config {
 				Enabled: true,
 				Format:  "text",
 				JSONFields: map[string]string{
+					// Deployer-supplied json_fields merge into these defaults rather than
+					// replacing them, so "component" survives unless explicitly overridden.
+					"component":  "rtr",
 					"t":          "%START_TIME%",
 					"meth":       "%REQ(:METHOD)%",
 					"path":       "%REQ(X-ENVOY-ORIGINAL-PATH?:PATH)%",
@@ -967,7 +985,10 @@ func defaultConfig() *Config {
 					"reqDur":     "%REQUEST_DURATION%",
 					"respDur":    "%RESPONSE_DURATION%",
 				},
-				TextFormat: "[%START_TIME%] \"%REQ(:METHOD)% %REQ(X-ENVOY-ORIGINAL-PATH?:PATH)% %PROTOCOL%\" " +
+				// routerLogComponentTag identifies the router on the container's shared
+				// stdout; keep it when overriding. The JSON variant uses the "component"
+				// field instead.
+				TextFormat: routerLogComponentTag + "[%START_TIME%] \"%REQ(:METHOD)% %REQ(X-ENVOY-ORIGINAL-PATH?:PATH)% %PROTOCOL%\" " +
 					"%REQ(:PATH)% %UPSTREAM_PROTOCOL% %RESPONSE_CODE% %RESPONSE_FLAGS% %RESPONSE_CODE_DETAILS% " +
 					"%CONNECTION_TERMINATION_DETAILS% %BYTES_RECEIVED% %BYTES_SENT% %DURATION% " +
 					"%REQUEST_TX_DURATION% %RESPONSE_TX_DURATION% %REQUEST_DURATION% %RESPONSE_DURATION% " +
@@ -1321,6 +1342,13 @@ func (c *Config) Validate() error {
 			if c.Router.AccessLogs.TextFormat == "" {
 				return fmt.Errorf("router.access_logs.text_format must be configured when format is 'text'")
 			}
+			// Warn rather than fail: an operator may have dropped the tag on purpose,
+			// and refusing to start over a log-formatting choice would be worse than
+			// the ambiguity it causes.
+			if !textAccessLogStartsWithComponentTag(c.Router.AccessLogs.TextFormat) {
+				slog.Warn("router.access_logs.text_format does not start with " + routerLogComponentTag +
+					"; router access-log lines will not be attributable on the container's shared stdout")
+			}
 		}
 	}
 

@@ -1915,3 +1915,40 @@ func TestConfig_CaseInsensitiveAlgorithm(t *testing.T) {
 	err := cfg.Validate()
 	assert.NoError(t, err, "Algorithm validation should be case insensitive")
 }
+
+func TestTextAccessLogStartsWithComponentTag(t *testing.T) {
+	// A deployer-supplied text_format replaces the default outright, so the router's
+	// component tag can go missing with nothing else to catch it.
+	tests := []struct {
+		name       string
+		textFormat string
+		want       bool
+	}{
+		{"shipped default", defaultConfig().Router.AccessLogs.TextFormat, true},
+		{"custom format keeping the tag", "[rtr] %RESPONSE_CODE%\n", true},
+		{"custom format dropping the tag", "[%START_TIME%] %RESPONSE_CODE%\n", false},
+		{"empty", "", false},
+		{"tag without its trailing space is not a match", "[rtr]%RESPONSE_CODE%\n", false},
+		// Present but not leading: attribution anchors on the tag at column 0, so a
+		// line reading "[2026-...] [rtr] 200" is not greppable by component.
+		{"tag appearing mid-format does not count", "[%START_TIME%] [rtr] %RESPONSE_CODE%\n", false},
+		{"leading space before the tag does not count", " [rtr] %RESPONSE_CODE%\n", false},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, textAccessLogStartsWithComponentTag(tt.textFormat))
+		})
+	}
+}
+
+func TestValidate_CustomTextAccessLogWithoutTagIsNotFatal(t *testing.T) {
+	// Dropping the tag is a log-formatting choice, not a misconfiguration: it must
+	// warn, never refuse to start.
+	cfg := defaultConfig()
+	cfg.Router.AccessLogs.Enabled = true
+	cfg.Router.AccessLogs.Format = "text"
+	cfg.Router.AccessLogs.TextFormat = "[%START_TIME%] %RESPONSE_CODE%\n"
+
+	assert.False(t, textAccessLogStartsWithComponentTag(cfg.Router.AccessLogs.TextFormat))
+	assert.NoError(t, cfg.Validate())
+}
@@ -225,7 +225,6 @@ log "Starting Policy Engine under dlv (listening on :2346, headless)..."
     --listen=:2346 --headless=true \
     --api-version=2 --accept-multiclient -- \
     -xds-server "${PE_XDS_SERVER}" "${PE_ARGS[@]}" \
-    > >(while IFS= read -r line; do echo "[pol] $line"; done) \
     2> >(while IFS= read -r line; do echo "[pol] $line" >&2; done) &
 PE_PID=$!
 log "Policy Engine (dlv) started (PID $PE_PID)"
@@ -267,7 +266,6 @@ log "Starting Envoy..."
     --log-level "${LOG_LEVEL}" \
     --concurrency "${ROUTER_CONCURRENCY}" \
     "${ROUTER_ARGS[@]}" \
-    > >(while IFS= read -r line; do echo "[rtr] $line"; done) \
     2> >(while IFS= read -r line; do echo "[rtr] $line" >&2; done) &
 ENVOY_PID=$!
 log "Envoy started (PID $ENVOY_PID)"

@@ -272,7 +272,6 @@ if [ -f /app/python-executor/python_policy_registry.py ]; then
     log "Starting Python Executor..."
     unset PYTHON_EXECUTOR_LISTEN
     python3 /app/python-executor/main.py --listen "${PYTHON_EXECUTOR_SOCKET}" "${PY_ARGS[@]}" \
-        > >(while IFS= read -r line; do echo "[pye] $line"; done) \
         2> >(while IFS= read -r line; do echo "[pye] $line" >&2; done) &
     PY_PID=$!
     log "Python Executor started (PID $PY_PID)"
@@ -298,10 +297,12 @@ else
     log "No Python policies detected, skipping Python Executor"
 fi
 
-# Start Policy Engine with [pol] log prefix
+# Start Policy Engine. Only stderr is prefixed: stdout carries the JSON traffic
+# log, which a line prefix would make unparseable, and the policy engine tags its
+# own stdout lines. stderr keeps the prefix because panics and stack dumps bypass
+# its logger.
 log "Starting Policy Engine..."
 /app/policy-engine -xds-server "${PE_XDS_SERVER}" "${PE_ARGS[@]}" \
-    > >(while IFS= read -r line; do echo "[pol] $line"; done) \
     2> >(while IFS= read -r line; do echo "[pol] $line" >&2; done) &
 PE_PID=$!
 log "Policy Engine started (PID $PE_PID)"
@@ -338,7 +339,6 @@ log "Starting Envoy..."
     --log-level "${LOG_LEVEL}" \
     --concurrency "${ROUTER_CONCURRENCY}" \
     "${ROUTER_ARGS[@]}" \
-    > >(while IFS= read -r line; do echo "[rtr] $line"; done) \
     2> >(while IFS= read -r line; do echo "[rtr] $line" >&2; done) &
 ENVOY_PID=$!
 log "Envoy started (PID $ENVOY_PID)"

@@ -22,6 +22,7 @@ import (
 	"context"
 	"flag"
 	"fmt"
+	"io"
 	"log/slog"
 	"net"
 	"os"
@@ -417,16 +418,50 @@ func setupLogger(cfg *config.Config) *slog.Logger {
 
 	opts := &slog.HandlerOptions{Level: level}
 
+	// Tagging is per logger, never process-wide on stdout: the traffic-log
+	// publisher writes bare JSON to os.Stdout on the same descriptor, and a
+	// process-wide prefix would corrupt it.
 	var handler slog.Handler
 	if cfg.PolicyEngine.Logging.Format == "json" {
 		handler = slog.NewJSONHandler(os.Stdout, opts)
-	} else {
-		handler = slog.NewTextHandler(os.Stdout, opts)
+		return slog.New(handler).With(slog.String(logComponentField, logComponentValue))
 	}
+	handler = slog.NewTextHandler(newComponentPrefixWriter(os.Stdout, logComponentPrefix), opts)
 
 	return slog.New(handler)
 }
 
+const (
+	logComponentPrefix = "[pol] "
+	logComponentField  = "component"
+	logComponentValue  = "pol"
+)
+
+// componentPrefixWriter prepends a tag to every write. slog emits one complete
+// record per Write, so this yields one tag per line. slog serializes writes, so
+// no lock is needed here.
+type componentPrefixWriter struct {
+	w      io.Writer
+	prefix []byte
+}
+
+func newComponentPrefixWriter(w io.Writer, prefix string) *componentPrefixWriter {
+	return &componentPrefixWriter{w: w, prefix: []byte(prefix)}
+}
+
+// Write emits the tag and p in one underlying Write so another writer sharing
+// the descriptor cannot interleave between them. The returned count excludes
+// the tag, which is framing rather than bytes consumed from the caller.
+func (c *componentPrefixWriter) Write(p []byte) (int, error) {
+	buf := make([]byte, 0, len(c.prefix)+len(p))
+	buf = append(buf, c.prefix...)
+	buf = append(buf, p...)
+	if _, err := c.w.Write(buf); err != nil {
+		return 0, err
+	}
+	return len(p), nil
+}
+
 // initializeXDSClient initializes and starts the xDS client
 func initializeXDSClient(ctx context.Context, cfg *config.Config, serverAddr string, k *kernel.Kernel, reg *registry.PolicyRegistry, resolvers resolver.ResolverRegistry) (*xdsclient.Client, error) {
 	slog.InfoContext(ctx, "Initializing xDS client",

@@ -19,10 +19,14 @@
 package main
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
+	"errors"
 	"log/slog"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"
 
@@ -334,6 +338,119 @@ func TestInitializeXDSClient_ValidConfig(t *testing.T) {
 	// The client will be cleaned up when the test exits
 }
 
+// =============================================================================
+// componentPrefixWriter Tests
+// =============================================================================
+
+func TestComponentPrefixWriter_PrefixesEachWrite(t *testing.T) {
+	var buf bytes.Buffer
+	w := newComponentPrefixWriter(&buf, "[pol] ")
+
+	n, err := w.Write([]byte("time=... level=INFO msg=hello\n"))
+
+	require.NoError(t, err)
+	// n must equal len(p); io.Writer callers treat any other count as a failed
+	// write.
+	assert.Equal(t, len("time=... level=INFO msg=hello\n"), n)
+	assert.Equal(t, "[pol] time=... level=INFO msg=hello\n", buf.String())
+}
+
+func TestComponentPrefixWriter_OneUnderlyingWritePerRecord(t *testing.T) {
+	counting := &countingWriter{}
+	w := newComponentPrefixWriter(counting, "[pol] ")
+
+	_, err := w.Write([]byte("first\n"))
+	require.NoError(t, err)
+	_, err = w.Write([]byte("second\n"))
+	require.NoError(t, err)
+
+	// One underlying write per record, so another writer on the same descriptor
+	// cannot interleave between tag and line.
+	assert.Equal(t, 2, counting.writes)
+	assert.Equal(t, "[pol] first\n[pol] second\n", counting.buf.String())
+}
+
+func TestComponentPrefixWriter_PropagatesError(t *testing.T) {
+	w := newComponentPrefixWriter(failingWriter{}, "[pol] ")
+
+	n, err := w.Write([]byte("boom\n"))
+
+	require.Error(t, err)
+	assert.Zero(t, n)
+}
+
+func TestSetupLogger_TextFormatIsComponentTagged(t *testing.T) {
+	cfg := &config.Config{
+		PolicyEngine: config.PolicyEngine{
+			Logging: config.LoggingConfig{Level: "info", Format: "text"},
+		},
+	}
+
+	out := captureStdout(t, func() {
+		setupLogger(cfg).Info("hello")
+	})
+
+	assert.True(t, strings.HasPrefix(out, "[pol] "), "text log line must be [pol]-tagged, got: %q", out)
+	assert.Contains(t, out, "msg=hello")
+}
+
+func TestSetupLogger_JSONFormatCarriesComponentField(t *testing.T) {
+	cfg := &config.Config{
+		PolicyEngine: config.PolicyEngine{
+			Logging: config.LoggingConfig{Level: "info", Format: "json"},
+		},
+	}
+
+	out := captureStdout(t, func() {
+		setupLogger(cfg).Info("hello")
+	})
+
+	assert.False(t, strings.HasPrefix(out, "[pol] "), "JSON log line must not be text-prefixed, got: %q", out)
+
+	var rec map[string]any
+	require.NoError(t, json.Unmarshal([]byte(strings.TrimSpace(out)), &rec), "JSON log line must parse, got: %q", out)
+	assert.Equal(t, "pol", rec["component"])
+	assert.Equal(t, "hello", rec["msg"])
+}
+
+type countingWriter struct {
+	buf    bytes.Buffer
+	writes int
+}
+
+func (c *countingWriter) Write(p []byte) (int, error) {
+	c.writes++
+	return c.buf.Write(p)
+}
+
+type failingWriter struct{}
+
+func (failingWriter) Write([]byte) (int, error) {
+	return 0, errors.New("write failed")
+}
+
+// captureStdout redirects os.Stdout for the duration of fn. setupLogger binds to
+// os.Stdout at construction, so the swap must precede the call.
+func captureStdout(t *testing.T, fn func()) string {
+	t.Helper()
+
+	r, w, err := os.Pipe()
+	require.NoError(t, err)
+	defer r.Close()
+
+	orig := os.Stdout
+	os.Stdout = w
+	defer func() { os.Stdout = orig }()
+
+	fn()
+	require.NoError(t, w.Close())
+
+	var buf bytes.Buffer
+	_, err = buf.ReadFrom(r)
+	require.NoError(t, err)
+	return buf.String()
+}
+
 // The ext_proc server must be constructed with all three bounds set. The values
 // themselves are validated in internal/config; what this pins is that none of the three
 // options is dropped from the construction, which is how this server silently ran on