diff --git a/.secrets.baseline b/.secrets.baseline index 7cf529b9..3a0d96c8 100644 --- a/.secrets.baseline +++ b/.secrets.baseline @@ -3,7 +3,7 @@ "files": "Cargo\\.lock$|\\.lock$|target/|^.secrets.baseline$", "lines": null }, - "generated_at": "2026-08-11T12:22:16Z", + "generated_at": "2026-08-11T17:44:49Z", "plugins_used": [ { "name": "AWSKeyDetector" @@ -142,7 +142,7 @@ "hashed_secret": "4a4645604f0b9e29503be96a87f6f47a6e4a7890", "is_secret": false, "is_verified": false, - "line_number": 154, + "line_number": 131, "type": "Secret Keyword", "verified_result": null }, @@ -150,7 +150,7 @@ "hashed_secret": "427f5e1b530d4a544883308d876a11d724060c86", "is_secret": false, "is_verified": false, - "line_number": 157, + "line_number": 134, "type": "Secret Keyword", "verified_result": null }, @@ -158,7 +158,7 @@ "hashed_secret": "bfc6000db1195a9522813fc405c666dd4ce669ad", "is_secret": false, "is_verified": false, - "line_number": 250, + "line_number": 227, "type": "Secret Keyword", "verified_result": null } @@ -168,7 +168,7 @@ "hashed_secret": "3e4e260fceb19ea97eb49fdcc922cc84f052b0f6", "is_secret": false, "is_verified": false, - "line_number": 235, + "line_number": 267, "type": "Secret Keyword", "verified_result": null }, @@ -176,7 +176,7 @@ "hashed_secret": "2f3b648abbad8976facf96604c2f272845b2aa23", "is_secret": false, "is_verified": false, - "line_number": 241, + "line_number": 273, "type": "Secret Keyword", "verified_result": null }, @@ -184,7 +184,7 @@ "hashed_secret": "af70e9cc80b554e84e780440cd8fe46092a7e6e5", "is_secret": false, "is_verified": false, - "line_number": 255, + "line_number": 287, "type": "Secret Keyword", "verified_result": null } @@ -194,7 +194,7 @@ "hashed_secret": "0a24796d4c71ce722a92f450f69dc36c60b21de4", "is_secret": false, "is_verified": false, - "line_number": 87, + "line_number": 222, "type": "Hex High Entropy String", "verified_result": null } @@ -224,7 +224,7 @@ "hashed_secret": "58e7dc38ba3a7d4a720006d2f3cc4cda774d89dc", "is_secret": false, "is_verified": false, - "line_number": 19, + "line_number": 20, "type": "Hex High Entropy String", "verified_result": null } diff --git a/Cargo.lock b/Cargo.lock index 6abd30c3..16eb740d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -553,6 +553,7 @@ dependencies = [ "contextforge-data-plane-apis", "contextforge-data-plane-cpex", "contextforge-data-plane-lib", + "contextforge-data-plane-observability", "cpex-payload-marker", "cpex-secrets-detection", "cpex-text-prefixer", @@ -561,22 +562,14 @@ dependencies = [ "jsonwebtoken", "num_cpus", "openport", - "opentelemetry", - "opentelemetry-otlp", - "opentelemetry_sdk", "redis", "reqwest", "rmcp", "rmp-serde", "rustls", - "serde_json", "tikv-jemallocator", "tokio", - "tonic", "tracing", - "tracing-appender", - "tracing-opentelemetry", - "tracing-subscriber", ] [[package]] @@ -619,6 +612,7 @@ dependencies = [ "clap", "contextforge-data-plane-apis", "contextforge-data-plane-cpex", + "contextforge-data-plane-observability", "cpex", "cpex-secrets-detection", "futures", @@ -628,8 +622,6 @@ dependencies = [ "jsonwebtoken", "lru_time_cache", "openport", - "opentelemetry", - "opentelemetry_sdk", "redis", "reqwest", "rmcp", @@ -647,12 +639,35 @@ dependencies = [ "tower", "tower-http 0.7.0", "tracing", - "tracing-opentelemetry", "typed-builder", "url", "uuid", ] +[[package]] +name = "contextforge-data-plane-observability" +version = "0.1.0" +dependencies = [ + "axum", + "chrono", + "clap", + "http", + "opentelemetry", + "opentelemetry-otlp", + "opentelemetry_sdk", + "serde_json", + "sha2", + "tokio", + "tonic", + "tower", + "tower-http 0.7.0", + "tracing", + "tracing-appender", + "tracing-opentelemetry", + "tracing-subscriber", + "uuid", +] + [[package]] name = "contextforge-load-test" version = "0.1.0" @@ -3591,6 +3606,16 @@ dependencies = [ "web-time", ] +[[package]] +name = "tracing-serde" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "704b1aeb7be0d0a84fc9828cae51dab5970fee5088f83d1dd7ee6f6246fc6ff1" +dependencies = [ + "serde", + "tracing-core", +] + [[package]] name = "tracing-subscriber" version = "0.3.23" @@ -3601,12 +3626,15 @@ dependencies = [ "nu-ansi-term", "once_cell", "regex-automata", + "serde", + "serde_json", "sharded-slab", "smallvec", "thread_local", "tracing", "tracing-core", "tracing-log", + "tracing-serde", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index eacf3e93..8f7adbb7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,6 +4,7 @@ members = [ "crates/contextforge-data-plane-apis", "crates/contextforge-data-plane-cpex", "crates/contextforge-data-plane-lib", + "crates/contextforge-data-plane-observability", "crates/contextforge-load-test", "crates/plugins/cpex-secrets-detection", ] @@ -24,6 +25,7 @@ repository = "https://github.com/contextforge-org/contextforge-data-plane" # Keep dependencies here only when at least two workspace members inherit them. contextforge-data-plane-cpex = { path = "./crates/contextforge-data-plane-cpex" } contextforge-data-plane-apis = { path = "./crates/contextforge-data-plane-apis"} +contextforge-data-plane-observability = { path = "./crates/contextforge-data-plane-observability" } rmcp = { version = "3.1.1", features = [ "server", "client", @@ -35,11 +37,11 @@ rmcp = { version = "3.1.1", features = [ ] } serde = {version= "1.0"} serde_json = "1.0" +chrono = "0.4.44" tracing = "0.1" -tracing-opentelemetry = "0.33.0" -opentelemetry = "0.32" -opentelemetry_sdk = { version = "0.32", features = ["rt-tokio", "metrics"] } tokio = { version = "1.53.1", features = ["macros", "net", "rt-multi-thread", "signal", "sync", "time"] } +tower = "0.5.3" +tower-http = "0.7.0" http = "1.4.0" redis = { version = "1.2.1", features = [ "default", diff --git a/_context/wiki/architecture.md b/_context/wiki/architecture.md index 3ef364f8..ad70c31b 100644 --- a/_context/wiki/architecture.md +++ b/_context/wiki/architecture.md @@ -7,7 +7,8 @@ Tower layers execute outside-in. A request reaches MCP handlers with these exten ```text TCP/TLS listener -> HttpMetricsLayer - -> TraceLayer + -> correlation_layer → transaction/correlation request scope + response headers + -> TraceLayer → request span + structured status/latency event -> /contextforge-rs nested router -> mcp_origin_layer → validates Host then Origin (403 when disallowed) -> CORS layer @@ -30,6 +31,7 @@ MCP handlers read typed extensions — they never parse headers, paths, or Redis ```text downstream request + -> correlation + W3C trace extraction -> Host/Origin validation → virtual host extraction → JWT validation → session extraction -> user config lookup → MCP handler validation -> request plugin hooks @@ -37,16 +39,19 @@ downstream request upstream response -> response plugin hooks → merge/namespace/passthrough - -> metrics, tracing, logging → downstream response + -> structured status/latency log → correlation response headers → metrics → downstream response ``` ```mermaid flowchart TD - bin["binary\nCLI · logging · runtime"] + bin["binary\nCLI · observability init · runtime"] lib["lib\nrouting · middleware\nsessions · transports"] + obs["observability\nJSON logs · correlation\ntracing · performance"] apis["apis\nUserConfig · VirtualHost\nBackendMCPGateway"] cpex["cpex\nCPEX hook factories"] bin --> lib + bin --> obs + lib --> obs lib --> apis lib --> cpex ``` @@ -82,6 +87,11 @@ Order is invariant: auth/config before backend selection; request plugins before | `transports/` | Downstream TCP and TLS listener setup | | `tools.rs` | Local bootstrap helpers (`with_tools` feature only) | +Cross-cutting observability lives in the sibling `contextforge-data-plane-observability` +crate. The binary explicitly installs its subscriber and exporters at startup; +the library uses its request middleware, context propagation, user pseudonym, +and latency timers without owning global initialization. + ## State Ownership | State | Owner | Lifetime | @@ -90,6 +100,9 @@ Order is invariant: auth/config before backend selection; request plugins before | JWT decoders | `ContextForgeDataPlaneAppState` | Process | | User config | `RedisUserConfigStore` (LRU + Redis) | Request-path consumed; control-plane authored | | Request identity / VirtualHostId | Request extensions | One HTTP request | +| Transaction and correlation IDs | Request extension + task-local scope | One HTTP request; returned in response headers and snapshotted into backend transport during initialize | +| Trace and span IDs | Request context + request span | One HTTP request; parsed or generated even when OTLP export is disabled and propagated to backend initialization | +| Pseudonymous user ID | Request span | One authenticated request; first 12 hex characters of a SHA-256 digest, never the raw JWT subject | | Downstream session id | RMCP + `SessionId` extension | MCP session | | Backend RMCP services | `BackendTransports` map | Local process, per principal/backend/session | | Local user session mapping | `LocalUserSessionStore` | Local LRU, 50k entries, 1 hour | @@ -142,7 +155,7 @@ Startup sequence (`main.rs` → `Gateway::run_gateway`): ```text install rustls crypto provider -> Config::parse() - -> logging::init_tracing_logging(&config) + -> observability::init_observability(&logging_config) -> Runtime::from(&config) ← sets executor shape -> optional CpexRuntimeRegistry -> Gateway::builder() @@ -165,7 +178,7 @@ backend response -> session_id_layer response side ← on DELETE success: remove session + backend transports -> claims_layer response side -> virtual_host_id_layer response side - -> CORS, mcp_origin_layer, TraceLayer, HttpMetricsLayer + -> CORS, mcp_origin_layer, TraceLayer, correlation_layer, HttpMetricsLayer -> downstream response ``` diff --git a/_context/wiki/config.md b/_context/wiki/config.md index 304c9a3c..e1d174f9 100644 --- a/_context/wiki/config.md +++ b/_context/wiki/config.md @@ -84,9 +84,11 @@ origins such as `https://app.example.com`; Host entries are authorities such as | `--otlp-endpoint ` | `CONTEXTFORGE_DATA_PLANE_OTEL_EXPORTER_OTLP_ENDPOINT` | Protocol-specific | Trace endpoint; defaults to `http://127.0.0.1:4317` for gRPC or `http://127.0.0.1:4318/v1/traces` for HTTP. | | `--otlp-metrics-endpoint ` | `CONTEXTFORGE_DATA_PLANE_OTEL_EXPORTER_OTLP_METRICS_ENDPOINT` | Protocol-specific | Metrics endpoint; defaults to `http://127.0.0.1:4317` for gRPC or `http://127.0.0.1:4318/v1/metrics` for HTTP. | | `--otlp-headers ` | `CONTEXTFORGE_DATA_PLANE_OTEL_EXPORTER_OTLP_HEADERS` | None | Comma-separated `key=value` exporter headers. | -| `--otlp-service-name ` | `CONTEXTFORGE_DATA_PLANE_OTEL_SERVICE_NAME` | `CONTEXTFORGE-DATA-PLANE` | OpenTelemetry `service.name`. | +| `--otlp-service-name ` | `CONTEXTFORGE_DATA_PLANE_OTEL_SERVICE_NAME` | `contextforge-data-plane` | OpenTelemetry `service.name` and structured-log `service_name`. | | `--log-name ` | `CONTEXTFORGE_DATA_PLANE_LOG_NAME` | `contextforge-data-plane.log` | File log name in the current directory. | | `--log-rotation ` | `CONTEXTFORGE_DATA_PLANE_LOG_ROTATION` | `hourly` | `minutely`, `hourly`, `daily`, or `never`. | +| `--environment ` | `CONTEXTFORGE_DATA_PLANE_ENVIRONMENT` | `unknown` | Deployment environment included in every log event. | +| `--cluster-id ` | `CONTEXTFORGE_DATA_PLANE_CLUSTER_ID` | `unknown` | Cluster identifier included in every log event. | ## JWT Claims (validated by `claims_layer`) @@ -242,10 +244,23 @@ cargo nextest run --locked -p contextforge-data-plane-lib --test gateway_plugins | Var | Default | Controls | | --- | --- | --- | -| `RUST_LOG` | `debug` | Console filter | -| `RUST_FILE_LOG` | `debug` | File filter | +| `RUST_LOG` | `info` | Console filter | +| `RUST_FILE_LOG` | `info` | File filter | | `RUST_TRACE_LOG` | `info` | OTLP span filter (`debug` for local trace verification) | +Console and rolling-file logs are newline-delimited JSON with the common fields +listed in [Working Preferences](preferences.md#logging-tracing). Requests +preserve or generate `x-contextforge-transaction-id` and +`x-contextforge-correlation-id` and return both headers to the caller. +Correlation IDs are UUIDs; transaction IDs accept caller values up to 128 +characters. Valid W3C `traceparent` IDs are preserved and missing or invalid +trace/span IDs are generated even when OTLP export is disabled. During +`initialize`, both ContextForge IDs and standard W3C `traceparent` context are snapshotted +into the backend transport. As with other downstream headers, backend +propagation is session-scoped until transports become per-request. + +After JWT validation, `user_id` is recorded as a `sha256:`-prefixed, truncated +digest of the subject. The raw subject is never written to logs. ## Telemetry Debugging Notes @@ -253,27 +268,29 @@ cargo nextest run --locked -p contextforge-data-plane-lib --test gateway_plugins Metrics are pushed by a `PeriodicReader` every **30 seconds**. Allow ~35s after the first request before data appears downstream. -**Stable log prefixes for grepping** (use these to scope log searches by boundary): +**Stable structured fields for filtering** (use these to scope searches by boundary): -| Prefix | Boundary | +| Filter | Boundary | | --- | --- | -| `claims_layer` | JWT validation failures | -| `user_config_store_layer` | Config lookup / Redis errors | -| `virtual_host_config_layer` | Unknown virtual host | -| `AuthorizedCallValidator::validate` | Post-session MCP validation | -| `initialize:` | Backend session creation | -| `call_tool` | Tool routing and backend invocation | +| `component=Authorization` | JWT and MCP call validation | +| `component=UserConfig` | Config cache, serialization, and Redis access | +| `component=Routing` | Virtual-host routing and backend invocation | +| `component=Plugins` | Runtime plugin lifecycle and hook results | +| `component=HttpServer event_type=PERFORMANCE` | Request status and latency | +| `event_type=PERFORMANCE metric=database_latency` | Redis connection/read/write latency | +| `event_type=PERFORMANCE metric=external_call_latency` | Backend initialization and MCP call latency | +| `error_code=CFDP-*` | Stable operator-facing failures | **Debugging by symptom:** | Symptom | Where to look | | --- | --- | -| `401` | `claims_layer` logs: missing/invalid token, unsupported algorithm, no decoder key | -| `400` config error | `user_config_store_layer` logs + Redis content for the JWT subject | -| `404 Server not found` | `virtual_host_config_layer` debug: requested vhost id vs caller's config | -| MCP routing errors | `AuthorizedCallValidator::validate` debug, then `call_tool`/`read_resource`/`get_prompt` warns | -| Backend failures | `initialize:` warns for failed backends; routed-call warns name the failing backend | -| Plugin problems | CPEX pipeline error logs; invalid reload marks runtime failed | +| `401` | `component=Authorization`; inspect the stable message and rejection fields. | +| `400` config error | `component=UserConfig`; correlate with `transaction_id` rather than logging the JWT subject. | +| `404 Server not found` | `component=Routing operation=load_virtual_host_config`. | +| MCP routing errors | `component=Authorization` followed by `component=Routing` with the same correlation ID. | +| Backend failures | `component=Routing` and the relevant `backend_name`. | +| Plugin problems | `component=Plugins`; invalid reloads mark the runtime failed. | ## Local Telemetry Verification Stack @@ -315,12 +332,12 @@ flowchart TD SYM --> SBACK["Backend failure"] SYM --> SPLUG["Plugin problem"] - S401 --> L401["grep: claims_layer\nmissing/invalid token\nbad algorithm / no decoder key"] - S400 --> L400["grep: user_config_store_layer\n+ Redis content for JWT subject"] - S404 --> L404["grep: virtual_host_config_layer\nrequested vhost vs caller config"] - SMCP --> LMCP["grep: AuthorizedCallValidator::validate\nthen call_tool / read_resource / get_prompt warns"] - SBACK --> LBACK["grep: initialize: warns\nrouted-call warns name failing backend"] - SPLUG --> LPLUG["CPEX pipeline error logs\ninvalid reload marks runtime failed"] + S401 --> L401["filter: component=Authorization"] + S400 --> L400["filter: component=UserConfig\ncorrelate by transaction_id"] + S404 --> L404["filter: component=Routing\noperation=load_virtual_host_config"] + SMCP --> LMCP["filter: component=Routing\ncorrelation_id"] + SBACK --> LBACK["filter: component=Routing\nbackend_name"] + SPLUG --> LPLUG["filter: component=Plugins"] ``` diff --git a/_context/wiki/preferences.md b/_context/wiki/preferences.md index 55fc3b75..b7cb0965 100644 --- a/_context/wiki/preferences.md +++ b/_context/wiki/preferences.md @@ -32,11 +32,20 @@ CI additionally runs `cargo shear --check-test-targets --deny-warnings --locked` ## Logging (tracing) - Use `tracing` for all log output. -- **Prefer message-embedded fields**: `level!("method_name - event field = {val} other_field = {other}")`. - Do **not** use structured field syntax (`, field = val`) for dataplane logs. -- Keep method/event prefixes stable and reuse the same field names and order for related events. +- Emit queryable fields with structured syntax: + `level!(component = "Routing", operation = "call_tool", backend_name, "backend tool call completed")`. +- Console and rolling-file output are newline-delimited JSON. The formatter adds the common contract: + `timestamp`, `service_name`, `version`, `environment`, `cluster_id`, `transaction_id`, + `correlation_id`, `trace_id`, `span_id`, `user_id`, `log_level`, `error_code`, `message`, and `component`. +- The shared implementation lives in `contextforge-data-plane-observability`; applications explicitly initialize it. +- Request, Redis, and backend-call latency events use `event_type=PERFORMANCE` with `metric`, `outcome`, and + `latency_ms` fields. +- `error!` and fatal events must add a stable `CFDP-*` error code, root cause, impact scope, and retryability. + Add `http_status` and `stack_trace` when available; the formatter supplies explicit null/default values otherwise. +- Use short stable messages and stable field names. Put variable data in fields, not message text. - `warn!` is for unexpected conditions that need operator attention. Expected user/config misses → `debug!` or `info!`. -- **Never log**: tokens, authorization headers, secrets, Redis key/value bytes, full `UserConfig`, or backend credentials. +- **Never log**: tokens, authorization headers, secrets, Redis key/value bytes, full `UserConfig`, backend credentials, + raw user subjects, session IDs, resource URIs, progress tokens, request/response bodies, or prompt content. ## Change discipline diff --git a/_context/wiki/project.md b/_context/wiki/project.md index 9f45e2fa..5aff4f16 100644 --- a/_context/wiki/project.md +++ b/_context/wiki/project.md @@ -77,7 +77,8 @@ Architecture context lives in the wiki. Key pages: | Crate | Purpose | | --- | --- | | `contextforge-data-plane-lib` | All dataplane behavior: routing, middleware, sessions, transports. Almost everything goes here. | -| `contextforge-data-plane` (binary) | Process shell only: CLI flags, logging, runtime shape. No dataplane logic. | +| `contextforge-data-plane` (binary) | Process shell only: CLI flags, observability initialization, runtime shape. No dataplane logic. | +| `contextforge-data-plane-observability` | Structured JSON formatting, correlation and W3C trace propagation, OTLP exporters, pseudonymous user attribution, and performance timers. | | `contextforge-data-plane-apis` | Shared config shapes (`UserConfig`, `User`, plugin config). Regenerate JSON schemas after any change: `cargo run -p contextforge-data-plane-apis`. | | `contextforge-data-plane-cpex` | Plugin integration (CPEX hook factories). | | `contextforge-load-test` | Performance harness: end-to-end MCP traffic driver. | diff --git a/_context/wiki/security.md b/_context/wiki/security.md index 65eea3e9..7f734f77 100644 --- a/_context/wiki/security.md +++ b/_context/wiki/security.md @@ -79,4 +79,7 @@ These routes are registered **outside the authentication middleware** — unauth ## Secrets Handling - The HMAC secret is held as a `SecretString`; key and certificate material is read from disk paths at startup. -- Never log: tokens, authorization headers, secrets, Redis key/value bytes, full `UserConfig` documents, or backend credentials. +- Never log: tokens, authorization headers, secrets, Redis key/value bytes, full `UserConfig` documents, backend + credentials, raw user subjects, session IDs, resource URIs, or progress tokens. +- Use generated or caller-supplied transaction/correlation IDs to join request events. Authenticated requests record + `user_id` as a `sha256:`-prefixed, 12-hex-character pseudonym; never log the raw JWT subject. diff --git a/crates/contextforge-data-plane-cpex/src/handle.rs b/crates/contextforge-data-plane-cpex/src/handle.rs index c8c4dfe2..033c3f79 100644 --- a/crates/contextforge-data-plane-cpex/src/handle.rs +++ b/crates/contextforge-data-plane-cpex/src/handle.rs @@ -118,7 +118,12 @@ impl CpexRuntimeRegistry { match result { Ok(()) => last_applied_config = Some(fingerprint), Err(error) => { - tracing::warn!(%error, "failed to reload CPEX runtime plugin config"); + tracing::warn!( + component = "Plugins", + operation = "reload_config", + error = %error, + "runtime plugin config reload failed" + ); set_runtime_failed(&runtime, &error); last_applied_config = None; }, @@ -126,12 +131,22 @@ impl CpexRuntimeRegistry { }, Ok(None) => { let error = GatewayPluginRuntimeError::ConfigMissing; - tracing::warn!(%error, "failed to reload CPEX runtime plugin config"); + tracing::warn!( + component = "Plugins", + operation = "reload_config", + error = %error, + "runtime plugin config is missing" + ); set_runtime_failed(&runtime, &error); last_applied_config = None; }, Err(error) => { - tracing::warn!(%error, "failed to load CPEX runtime plugin config"); + tracing::warn!( + component = "Plugins", + operation = "load_config", + error = %error, + "runtime plugin config load failed" + ); set_runtime_failed(&runtime, &error); last_applied_config = None; }, @@ -292,7 +307,12 @@ impl GatewayPluginRuntimeHandle { fn runtime_failed_error(state: &RuntimeState) -> ErrorData { if let RuntimeState::Failed(error) = state { - tracing::warn!(%error, "rejecting tool call because CPEX runtime is failed"); + tracing::warn!( + component = "Plugins", + operation = "tool_call", + error = %error, + "tool call rejected because the runtime plugin is unavailable" + ); } ErrorData { code: ErrorCode::INTERNAL_ERROR, message: "Runtime plugin reload failed".into(), data: None } } diff --git a/crates/contextforge-data-plane-cpex/src/pipeline.rs b/crates/contextforge-data-plane-cpex/src/pipeline.rs index 5d6cdd97..58d94fe3 100644 --- a/crates/contextforge-data-plane-cpex/src/pipeline.rs +++ b/crates/contextforge-data-plane-cpex/src/pipeline.rs @@ -71,7 +71,13 @@ pub(crate) fn plugin_denied_error(result: PipelineResult) -> ErrorData { let code = result .violation .and_then(|violation| { - warn!("Plugin denied tool call: code={} plugin={:?}", violation.code, violation.plugin_name); + warn!( + component = "Plugins", + operation = "tool_call", + violation_code = violation.code, + plugin = ?violation.plugin_name, + "runtime plugin denied tool call" + ); violation.proto_error_code.and_then(|code| i32::try_from(code).ok()).map(ErrorCode) }) .unwrap_or(ErrorCode::INVALID_REQUEST); @@ -82,11 +88,13 @@ pub(crate) fn plugin_denied_error(result: PipelineResult) -> ErrorData { pub(crate) fn log_pipeline_errors(hook: &'static str, result: &PipelineResult) { for error in &result.errors { warn!( + component = "Plugins", + operation = "pipeline_hook", hook, plugin = error.plugin_name, - code = error.code.as_deref().unwrap_or(""), + plugin_error_code = error.code.as_deref().unwrap_or(""), proto_error_code = error.proto_error_code, - "CPEX plugin soft error" + "runtime plugin reported a recoverable error" ); } } diff --git a/crates/contextforge-data-plane-cpex/src/runtime.rs b/crates/contextforge-data-plane-cpex/src/runtime.rs index 21dc39c5..e106e65f 100644 --- a/crates/contextforge-data-plane-cpex/src/runtime.rs +++ b/crates/contextforge-data-plane-cpex/src/runtime.rs @@ -88,7 +88,12 @@ impl Drop for GatewayPluginRuntime { manager.shutdown().await; }); }, - Err(error) => tracing::warn!(%error, "skipping CPEX plugin shutdown outside a Tokio runtime"), + Err(error) => tracing::warn!( + component = "Plugins", + operation = "shutdown", + error = %error, + "runtime plugin shutdown skipped outside a Tokio runtime" + ), } } } diff --git a/crates/contextforge-data-plane-lib/Cargo.toml b/crates/contextforge-data-plane-lib/Cargo.toml index 6505887b..32283296 100644 --- a/crates/contextforge-data-plane-lib/Cargo.toml +++ b/crates/contextforge-data-plane-lib/Cargo.toml @@ -14,29 +14,27 @@ repository.workspace = true [dependencies] contextforge-data-plane-apis.workspace = true contextforge-data-plane-cpex.workspace = true +contextforge-data-plane-observability.workspace = true rmcp.workspace = true serde.workspace = true serde_json.workspace = true tracing.workspace = true -tracing-opentelemetry.workspace = true -opentelemetry.workspace = true tokio.workspace = true tokio-util = "0.7" axum.workspace = true axum-otel-metrics = "0.14" -tower-http = { version = "0.7.0", features = ["full"] } -tower = "0.5.3" +tower-http = { workspace = true, features = ["full"] } +tower.workspace = true http.workspace = true futures = { version = "0.3", features = ["std", "alloc"] } jsonwebtoken.workspace = true -chrono = "0.4.44" +chrono.workspace = true redis.workspace = true clap.workspace = true thiserror.workspace = true rmp-serde.workspace = true async-trait.workspace = true reqwest.workspace = true -uuid.workspace = true lru_time_cache = "0.11.11" hyper-util = "0.1.20" hyper = "1.4.0" @@ -54,11 +52,11 @@ with_tools = [] [dev-dependencies] -opentelemetry_sdk.workspace = true cpex.workspace = true openport.workspace = true cpex-secrets-detection.workspace = true test-log = "0.2.20" +uuid.workspace = true axum-server = { version = "0.8.0", features = ["tls-rustls"] } [lints] diff --git a/crates/contextforge-data-plane-lib/src/common.rs b/crates/contextforge-data-plane-lib/src/common.rs index b736b35e..f95c5202 100644 --- a/crates/contextforge-data-plane-lib/src/common.rs +++ b/crates/contextforge-data-plane-lib/src/common.rs @@ -1,4 +1,5 @@ use clap::{Parser, ValueEnum}; +use contextforge_data_plane_observability::{LogRotation, OtlpProtocol}; use http::uri::Authority; use jsonwebtoken::DecodingKey; use redis::{ConnectionAddr, IntoConnectionInfo, RedisError}; @@ -115,30 +116,6 @@ pub enum RedisConnectionMode { Mtls, } -#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, ValueEnum)] -#[derive(Default)] -pub enum LogRotation { - Minutely, - #[default] - Hourly, - Daily, - Never, -} - -/// Wire protocol used to export OpenTelemetry data to the collector / backend. -/// -/// `Grpc` targets the standard OTLP/gRPC port (e.g. `4317`) used by -/// collectors such as the OpenTelemetry Collector and Tempo. -/// `HttpProtobuf` targets OTLP over HTTP/1.1 with a protobuf payload -/// (e.g. `4318/v1/traces`) and is the only protocol supported by Langfuse's -/// OTel ingestion endpoint (`/api/public/otel/v1/traces`). -#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, ValueEnum, Default)] -pub enum OtlpProtocol { - #[default] - Grpc, - HttpProtobuf, -} - #[derive(Debug, Clone, Parser, Default)] #[command(name = "contextforge-data-plane")] #[command(about = "Minimal, fast, experimental data plane for ContextForge")] @@ -178,7 +155,7 @@ pub struct Config { pub otlp_headers: Option, /// Overrides the `service.name` OpenTelemetry resource attribute. - /// Defaults to `CONTEXTFORGE-DATA-PLANE`. + /// Defaults to `contextforge-data-plane`. #[arg(long, env = "CONTEXTFORGE_DATA_PLANE_OTEL_SERVICE_NAME")] pub otlp_service_name: Option, @@ -258,6 +235,14 @@ pub struct Config { #[arg(long, env = "CONTEXTFORGE_DATA_PLANE_LOG_ROTATION")] pub log_rotation: Option, + /// Deployment environment included in every structured log event. + #[arg(long, env = "CONTEXTFORGE_DATA_PLANE_ENVIRONMENT")] + pub environment: Option, + + /// Cluster identifier included in every structured log event. + #[arg(long, env = "CONTEXTFORGE_DATA_PLANE_CLUSTER_ID")] + pub cluster_id: Option, + #[arg( long, env = "CONTEXTFORGE_GATEWAY_RS_MCP_ALLOWED_ORIGINS", diff --git a/crates/contextforge-data-plane-lib/src/gateway/backend_client.rs b/crates/contextforge-data-plane-lib/src/gateway/backend_client.rs index 034bef24..8234421a 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/backend_client.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/backend_client.rs @@ -63,7 +63,14 @@ impl GatewayBackendClient { downstream: Peer, post_state: Option, ) -> Result, ServiceError> { - debug!("track_tool_call {tool_name} {downstream_progress_token:?} {post_state:?}"); + debug!( + component = "Routing", + operation = "track_tool_call", + tool_name, + has_progress_token = downstream_progress_token.is_some(), + has_post_hook_state = post_state.is_some(), + "tracking backend tool call" + ); let Some(downstream_progress_token) = downstream_progress_token else { return start_backend_tool_call(peer, request).await; }; @@ -80,7 +87,7 @@ impl GatewayBackendClient { } pub(crate) async fn stop_tracking_tool_call(&self, backend_progress_token: &ProgressToken) { - debug!("stop_tracking_tool_call {backend_progress_token:?}"); + debug!(component = "Routing", operation = "stop_tracking_tool_call", "stopped tracking backend tool call"); let mut calls = self.in_flight_calls.write().await; calls.remove(backend_progress_token); } @@ -91,13 +98,23 @@ impl GatewayBackendClient { } pub(crate) async fn track_resource_subscription(&self, resource_uri: &str, downstream: Peer) { - debug!("track_resource_subscription backend {} uri {resource_uri}", self.backend_name); + debug!( + component = "Routing", + operation = "track_resource_subscription", + backend_name = self.backend_name, + "tracking backend resource subscription" + ); let mut subscriptions = self.resource_subscriptions.lock().await; subscriptions.insert(resource_uri.to_owned(), downstream); } pub(crate) async fn stop_tracking_resource_subscription(&self, resource_uri: &str) { - debug!("stop_tracking_resource_subscription backend {} uri {resource_uri}", self.backend_name); + debug!( + component = "Routing", + operation = "stop_tracking_resource_subscription", + backend_name = self.backend_name, + "stopped tracking backend resource subscription" + ); let mut subscriptions = self.resource_subscriptions.lock().await; subscriptions.remove(resource_uri); } @@ -117,7 +134,12 @@ impl GatewayBackendClient { match plugin_runtime.after_stream_event(&call.tool_name, event, call.post_state.clone()).await { Ok(event) => event, Err(error) => { - warn!("call_tool: plugin rejected backend notification: {error:?}"); + warn!( + component = "Plugins", + operation = "filter_backend_notification", + error = ?error, + "runtime plugin rejected backend notification" + ); None }, } @@ -140,18 +162,24 @@ impl ClientHandler for GatewayBackendClient { async fn on_progress(&self, mut progress: ProgressNotificationParam, _context: NotificationContext) { let Some(call) = self.progress_call(&progress.progress_token).await else { debug!( - "call_tool: dropping backend progress notification with unknown token {:?}", - progress.progress_token + component = "Routing", + operation = "forward_progress", + "backend progress notification dropped because its token is unknown" ); return; }; progress.progress_token.clone_from(&call.downstream_progress_token); - debug!("Processing Progress Notification {progress:?} {call:?}"); + debug!(component = "Routing", operation = "forward_progress", "processing backend progress notification"); let Some(progress) = self.stream_event_post_hook(&call, progress).await else { return; }; if let Err(error) = call.downstream.notify_progress(progress).await { - warn!("call_tool: unable to forward backend progress notification downstream: {error:?}"); + warn!( + component = "Routing", + operation = "forward_progress", + error = ?error, + "backend progress notification could not be forwarded" + ); } } @@ -161,13 +189,24 @@ impl ClientHandler for GatewayBackendClient { _context: NotificationContext, ) { let Some(downstream) = self.resource_subscription(¶ms.uri).await else { - debug!("resource_updated: dropping backend notification for unsubscribed uri {}", params.uri); + debug!( + component = "Routing", + operation = "forward_resource_update", + backend_name = self.backend_name, + "backend resource notification dropped because there is no subscription" + ); return; }; params.uri = resource_uri_for_downstream(&self.backend_name, params.uri, self.namespace_identifiers); if let Err(error) = downstream.notify_resource_updated(params).await { - warn!("resource_updated: unable to forward backend notification downstream: {error:?}"); + warn!( + component = "Routing", + operation = "forward_resource_update", + backend_name = self.backend_name, + error = ?error, + "backend resource notification could not be forwarded" + ); } } } @@ -202,7 +241,12 @@ pub(crate) async fn call_backend_tool( let Some(response) = response else { let reason = "tool call cancelled by the downstream client".to_owned(); if let Err(error) = handle.cancel(Some(reason.clone())).await { - warn!("call_tool: unable to relay cancellation to the backend: {error:?}"); + warn!( + component = "Routing", + operation = "cancel_tool_call", + error = ?error, + "tool call cancellation could not be relayed to the backend" + ); } return Err(ServiceError::Cancelled { reason: Some(reason) }); }; diff --git a/crates/contextforge-data-plane-lib/src/gateway/identifier_routing.rs b/crates/contextforge-data-plane-lib/src/gateway/identifier_routing.rs index c02b9946..9d888da9 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/identifier_routing.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/identifier_routing.rs @@ -66,7 +66,13 @@ pub(super) fn exposed_tool_name(virtual_host: &VirtualHost, backend_name: &str, } pub(super) fn backend_forward_error(op: &str, backend_name: &str, error: &ServiceError) -> ErrorData { - warn!("{op}: backend {backend_name} error = {error:?}"); + warn!( + component = "Routing", + operation = op, + backend_name, + error = ?error, + "backend request failed" + ); match error { ServiceError::McpError(mcp_error) => mcp_error.to_owned(), @@ -104,13 +110,19 @@ pub(super) async fn resolve_backend( backend_name: &str, ) -> Result<(String, McpClientService), ErrorData> { let backend_transports = session_manager.borrow_transports().await; - debug!("{op}: resolving backend {backend_name} from {backend_transports:?}"); + debug!( + component = "Routing", + operation = op, + backend_name, + backend_count = backend_transports.len(), + "resolving backend transport" + ); let mut target = None; for service_holder in backend_transports { if service_holder.name == backend_name { if target.is_some() { - warn!("{op}: more than one backend matching {backend_name}"); + warn!(component = "Routing", operation = op, backend_name, "duplicate backend transports matched"); session_manager.cleanup_backends("invalid session.. duplicate backends detected").await; return Err(ErrorData { code: ErrorCode::INVALID_REQUEST, @@ -130,7 +142,7 @@ pub(super) async fn resolve_backend( }); }; let Some(service) = running_service else { - warn!("{op}: no running backend for {backend_name}"); + warn!(component = "Routing", operation = op, backend_name, "backend transport is not running"); return Err(ErrorData { code: ErrorCode::INTERNAL_ERROR, message: "Routing problem... got no responses from backends".into(), diff --git a/crates/contextforge-data-plane-lib/src/gateway/list_aggregation.rs b/crates/contextforge-data-plane-lib/src/gateway/list_aggregation.rs index deeb2430..b151f85f 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/list_aggregation.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/list_aggregation.rs @@ -4,6 +4,7 @@ use std::{ }; use contextforge_data_plane_apis::user_store::VirtualHost; +use contextforge_data_plane_observability::PerformanceTimer; use rmcp::{ ErrorData, model::{ @@ -83,7 +84,12 @@ where let call = &call; async move { let response = match service_holder.running_service { - Some(service) => Some(call(service_holder.name.clone(), service).await), + Some(service) => { + let mut timer = PerformanceTimer::external_call("Routing", op); + let response = call(service_holder.name.clone(), service).await; + timer.record_result(&response); + Some(response) + }, None => None, }; (service_holder.name, response) @@ -110,9 +116,23 @@ fn log_backend_response( item_count: impl Fn(&T) -> usize, ) { match response { - Some(Ok(response)) => info!("{kind}: backend {name} completed ({} items)", item_count(response)), - Some(Err(error)) => warn!("{kind}: backend {name} {error:?}"), - None => info!("{kind}: backend {name} unavailable"), + Some(Ok(response)) => info!( + component = "Routing", + operation = kind, + backend_name = name, + item_count = item_count(response), + "backend list request completed" + ), + Some(Err(error)) => warn!( + component = "Routing", + operation = kind, + backend_name = name, + error = ?error, + "backend list request failed" + ), + None => { + info!(component = "Routing", operation = kind, backend_name = name, "backend unavailable for list request"); + }, } } diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs index 026de314..e62aaa5e 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_call_validator.rs @@ -31,7 +31,15 @@ impl<'a> AuthorizedCallValidator<'a> { let has_claims = maybe_claims.is_some(); let virtual_host_id = maybe_virtual_host_id.map_or("", |id| id.value().as_str()); debug!( - "AuthorizedCallValidator::validate - mcp call validation call_name = {call_name} has_user_config = {has_user_config} virtual_hosts = {virtual_hosts} has_session_id = {has_session_id} has_claims = {has_claims} virtual_host_id = {virtual_host_id}" + component = "Authorization", + operation = "validate_call", + call_name, + has_user_config, + virtual_hosts, + has_session_id, + has_claims, + virtual_host_id, + "validated MCP call context" ); let Some(session_id) = maybe_session_id else { @@ -63,7 +71,12 @@ impl<'a> AuthorizedCallValidator<'a> { let virtual_host_id = virtual_host_id.value(); let virtual_hosts = user_config.virtual_hosts.len(); debug!( - "AuthorizedCallValidator::validate - mcp virtual host config missing call_name = {call_name} virtual_host_id = {virtual_host_id} virtual_hosts = {virtual_hosts}" + component = "Authorization", + operation = "validate_call", + call_name, + virtual_host_id, + virtual_hosts, + "virtual host config is missing" ); return Err(ErrorData { code: ErrorCode::RESOURCE_NOT_FOUND, @@ -106,7 +119,15 @@ impl<'a> InitializeCallValidator<'a> { let has_claims = maybe_claims.is_some(); let virtual_host_id = maybe_virtual_host_id.map_or("", |id| id.value().as_str()); debug!( - "InitializeCallValidator::validate - mcp call validation call_name = {call_name} has_user_config = {has_user_config} virtual_hosts = {virtual_hosts} has_session_id = {has_session_id} has_claims = {has_claims} virtual_host_id = {virtual_host_id}" + component = "Authorization", + operation = "validate_initialize", + call_name, + has_user_config, + virtual_hosts, + has_session_id, + has_claims, + virtual_host_id, + "validated MCP initialize context" ); let Some(user_config) = maybe_user_config else { @@ -130,7 +151,12 @@ impl<'a> InitializeCallValidator<'a> { let virtual_host_id = virtual_host_id.value(); let virtual_hosts = user_config.virtual_hosts.len(); debug!( - "InitializeCallValidator::validate - mcp virtual host config missing call_name = {call_name} virtual_host_id = {virtual_host_id} virtual_hosts = {virtual_hosts}" + component = "Authorization", + operation = "validate_initialize", + call_name, + virtual_host_id, + virtual_hosts, + "virtual host config is missing" ); return Err(ErrorData { code: ErrorCode::RESOURCE_NOT_FOUND, diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/completion.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/completion.rs index 7fe5211e..a97bfedb 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/completion.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/completion.rs @@ -1,3 +1,4 @@ +use contextforge_data_plane_observability::PerformanceTimer; use rmcp::{ ErrorData, RoleServer, model::{CompleteRequestParams, CompleteResult, Reference}, @@ -45,10 +46,16 @@ where Reference::Resource(resource) => resource.uri = routed_identifier, _ => return Err(ErrorData::invalid_params("Unsupported completion reference", None)), } - let response = service - .complete(routed_request) - .await - .map_err(|error| backend_forward_error("complete", &service_name, &error))?; - info!("complete: backend {service_name} returned {} values", response.completion.values.len()); + let mut timer = PerformanceTimer::external_call("Routing", "complete"); + let response = service.complete(routed_request).await; + timer.record_result(&response); + let response = response.map_err(|error| backend_forward_error("complete", &service_name, &error))?; + info!( + component = "Routing", + operation = "complete", + backend_name = service_name, + value_count = response.completion.values.len(), + "backend completion request completed" + ); Ok(response) } diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/initialization.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/initialization.rs index 705c6c3b..9896933b 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/initialization.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/initialization.rs @@ -1,6 +1,7 @@ use std::{collections::HashMap, sync::Arc}; use contextforge_data_plane_apis::user_store::BackendMCPGateway; +use contextforge_data_plane_observability::PerformanceTimer; use http::request::Parts; use rmcp::{ ErrorData, RoleClient, RoleServer, ServiceExt, @@ -65,44 +66,59 @@ where let backend_url = backend.url.clone(); let backend_cfg = backend.clone(); let downstream_headers = downstream_headers.clone(); - let downstream_session_id = downstream_session_id.clone(); Box::pin(async move { let mut headers = HashMap::new(); if let Some(host) = backend_url.host_str() && backend_url.scheme() == "https" { - let host = if let Some(port) = backend_url.port() { - format!("{host}:{port}") - } else { - host.to_owned() - }; + let host = + if let Some(port) = backend_url.port() { format!("{host}:{port}") } else { host.to_owned() }; if let Ok(value) = http::HeaderValue::from_str(&host) { headers.insert(http::header::HOST, value); } else { - warn!("Really can't set the host header for {:?}", backend_url.host_str()); + warn!( + component = "Routing", + operation = "initialize_backend", + backend_name = name, + "backend Host header could not be constructed" + ); } } apply_header_config(&mut headers, &backend_cfg, downstream_headers.as_ref()); - // Propagate the active W3C trace context to the backend so the - // gateway span links to the downstream MCP server's spans. - crate::telemetry::inject_current_context(&mut headers); + // Propagate request correlation and W3C trace context to the + // backend so logs and spans remain linked across services. + contextforge_data_plane_observability::inject_current_context(&mut headers); let config = StreamableHttpClientTransportConfig::with_uri(backend_url.to_string()).custom_headers(headers); let transport = StreamableHttpClientTransport::with_client(client, config); - let maybe_running_service = backend_client.serve(transport).await; - if let Ok(running_service) = maybe_running_service { - info!("initialize: intialized for {downstream_session_id:?} {name:?}"); - (name, Some(running_service)) - } else { - warn!( - "initialize: Unable to initialize for {downstream_session_id:?} {name:?} {maybe_running_service:?}", - ); - (name, None) + let mut timer = PerformanceTimer::external_call("Routing", "initialize_backend"); + let result = backend_client.serve(transport).await; + timer.record_result(&result); + match result { + Ok(running_service) => { + info!( + component = "Routing", + operation = "initialize_backend", + backend_name = name, + "backend initialized" + ); + (name, Some(running_service)) + }, + Err(error) => { + warn!( + component = "Routing", + operation = "initialize_backend", + backend_name = name, + error = ?error, + "backend initialization failed" + ); + (name, None) + }, } }) }) @@ -115,7 +131,11 @@ where .into_iter() .map(|(name, running_service): (_, _)| { info!( - "initialize: Adding transport: session_id {downstream_session_id:#?} backend {name} {running_service:?}" + component = "Routing", + operation = "register_backend_transport", + backend_name = name, + backend_connected = running_service.is_some(), + "backend transport registered" ); let server_capabilities = running_service diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/prompts.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/prompts.rs index 14c0654d..095fdb10 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/prompts.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/prompts.rs @@ -1,3 +1,4 @@ +use contextforge_data_plane_observability::PerformanceTimer; use rmcp::{ ErrorData, RoleServer, model::{GetPromptRequestParams, GetPromptResponse, ListPromptsResult, PaginatedRequestParams}, @@ -86,10 +87,16 @@ where let mut routed_request = request; routed_request.name = prompt_name; - let response = service - .get_prompt(routed_request) - .await - .map_err(|error| backend_forward_error("get_prompt", &service_name, &error))?; - info!("get_prompt: backend {service_name} returned {} messages", response.messages.len()); + let mut timer = PerformanceTimer::external_call("Routing", "get_prompt"); + let response = service.get_prompt(routed_request).await; + timer.record_result(&response); + let response = response.map_err(|error| backend_forward_error("get_prompt", &service_name, &error))?; + info!( + component = "Routing", + operation = "get_prompt", + backend_name = service_name, + message_count = response.messages.len(), + "backend prompt request completed" + ); Ok(response.into()) } diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/resources.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/resources.rs index d75dea04..16c94c13 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/resources.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/resources.rs @@ -1,3 +1,4 @@ +use contextforge_data_plane_observability::PerformanceTimer; use rmcp::{ ErrorData, RoleServer, model::{ @@ -89,11 +90,17 @@ where let mut routed_request = request; routed_request.uri = resource_uri; - let response = service - .read_resource(routed_request) - .await - .map_err(|error| backend_forward_error("read_resource", &service_name, &error))?; - info!("read_resource: backend {service_name} returned {} contents", response.contents.len()); + let mut timer = PerformanceTimer::external_call("Routing", "read_resource"); + let response = service.read_resource(routed_request).await; + timer.record_result(&response); + let response = response.map_err(|error| backend_forward_error("read_resource", &service_name, &error))?; + info!( + component = "Routing", + operation = "read_resource", + backend_name = service_name, + content_count = response.contents.len(), + "backend resource request completed" + ); Ok(response.into()) } @@ -174,11 +181,19 @@ where routed_request.uri = resource_uri.clone(); service.service().track_resource_subscription(&resource_uri, cx.peer.clone()).await; - if let Err(error) = service.subscribe(routed_request).await { + let mut timer = PerformanceTimer::external_call("Routing", "subscribe"); + let result = service.subscribe(routed_request).await; + timer.record_result(&result); + if let Err(error) = result { service.service().stop_tracking_resource_subscription(&resource_uri).await; return Err(backend_forward_error("subscribe", &service_name, &error)); } - info!("subscribe: backend {service_name} completed"); + info!( + component = "Routing", + operation = "subscribe", + backend_name = service_name, + "backend resource subscription completed" + ); Ok(()) } @@ -205,11 +220,16 @@ where let mut routed_request = request; routed_request.uri = resource_uri.clone(); - service - .unsubscribe(routed_request) - .await - .map_err(|error| backend_forward_error("unsubscribe", &service_name, &error))?; + let mut timer = PerformanceTimer::external_call("Routing", "unsubscribe"); + let result = service.unsubscribe(routed_request).await; + timer.record_result(&result); + result.map_err(|error| backend_forward_error("unsubscribe", &service_name, &error))?; service.service().stop_tracking_resource_subscription(&resource_uri).await; - info!("unsubscribe: backend {service_name} completed"); + info!( + component = "Routing", + operation = "unsubscribe", + backend_name = service_name, + "backend resource unsubscribe completed" + ); Ok(()) } diff --git a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/tools.rs b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/tools.rs index b442fb7d..d5a5ce1e 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/mcp_service/tools.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/mcp_service/tools.rs @@ -1,4 +1,5 @@ use contextforge_data_plane_cpex::ToolPreCallResult; +use contextforge_data_plane_observability::PerformanceTimer; use rmcp::{ ErrorData, RoleServer, model::{CallToolRequestParams, CallToolResponse, ErrorCode, ListToolsResult, PaginatedRequestParams}, @@ -103,7 +104,8 @@ where pre_result.arguments.apply_to_request(&mut routed_request, &tool_name); let progress_token = cx.meta.get_progress_token(); - let handle = backend_service + let mut timer = PerformanceTimer::external_call("Routing", "call_tool"); + let handle_result = backend_service .service() .start_tool_call( backend_service.peer(), @@ -113,10 +115,14 @@ where cx.peer.clone(), post_state.clone(), ) - .await - .map_err(|error| backend_forward_error("call_tool", &service_name, &error))?; + .await; + if handle_result.is_err() { + timer.failed(); + } + let handle = handle_result.map_err(|error| backend_forward_error("call_tool", &service_name, &error))?; let backend_progress_token = handle.progress_token.clone(); let response = call_backend_tool(handle, cx.ct.clone()).await; + timer.record_result(&response); backend_service.service().stop_tracking_tool_call(&backend_progress_token).await; let response = response.map_err(|error| backend_forward_error("call_tool", &service_name, &error))?; @@ -126,6 +132,6 @@ where }, _ => response, }; - info!("call_tool: backend {service_name} completed"); + info!(component = "Routing", operation = "call_tool", backend_name = service_name, "backend tool call completed"); Ok(response.into()) } diff --git a/crates/contextforge-data-plane-lib/src/gateway/session_manager.rs b/crates/contextforge-data-plane-lib/src/gateway/session_manager.rs index 20aa828a..ad2741a1 100644 --- a/crates/contextforge-data-plane-lib/src/gateway/session_manager.rs +++ b/crates/contextforge-data-plane-lib/src/gateway/session_manager.rs @@ -40,7 +40,6 @@ impl<'a> SessionManager<'a> { // pub async fn return_transports(&self, backend_transports: impl Iterator) { // let backend_transports = backend_transports.collect::>(); - // info!("Returning transports {:?} {backend_transports:?}", self.session_id); // let mut transports = self.transports.inner().lock().await; // for svc_holder in backend_transports { // transports @@ -51,11 +50,23 @@ impl<'a> SessionManager<'a> { pub async fn cleanup_backends(&self, reason: &'static str) { let names: Vec<_> = self.virtual_host.backends.keys().cloned().collect(); - info!("Cleaning up backends {:?}", self.session_id); + info!( + component = "Session", + operation = "cleanup_backends", + backend_count = names.len(), + reason, + "cleaning up backend transports" + ); let mut transports = self.transports.inner().lock().await; for name in names { let key = BackendTransportKey::from((&name, self.session_id, self.principal)); - debug!("session_manager: removing transport for {key:?} {reason}"); + debug!( + component = "Session", + operation = "cleanup_backend", + backend_name = name, + reason, + "removing backend transport" + ); transports.remove(&key); } } diff --git a/crates/contextforge-data-plane-lib/src/layers/claims_id.rs b/crates/contextforge-data-plane-lib/src/layers/claims_id.rs index 02ff0488..f4e7a70a 100644 --- a/crates/contextforge-data-plane-lib/src/layers/claims_id.rs +++ b/crates/contextforge-data-plane-lib/src/layers/claims_id.rs @@ -66,6 +66,7 @@ pub async fn claims_layer( }; let claims: ContextForgeClaims = claims.claims; + contextforge_data_plane_observability::record_authenticated_user(&claims.sub); parts.extensions.insert(claims.clone()); let request = Request::from_parts(parts, body); next.run(request).await diff --git a/crates/contextforge-data-plane-lib/src/layers/mcp_origin.rs b/crates/contextforge-data-plane-lib/src/layers/mcp_origin.rs index 302bcf88..b7b53c37 100644 --- a/crates/contextforge-data-plane-lib/src/layers/mcp_origin.rs +++ b/crates/contextforge-data-plane-lib/src/layers/mcp_origin.rs @@ -53,43 +53,75 @@ pub async fn mcp_origin_layer(State(config): State, request: http::Reque if let Some(ref allowed_hosts) = config.mcp_allowed_hosts { match request_authority(&request) { None => { - warn!("mcp_origin_layer - rejected request: Host header missing or unparseable"); + warn!( + component = "Security", + operation = "validate_origin", + rejection_reason = "missing_or_invalid_host", + "request rejected by origin validation" + ); return forbidden_response(); }, Some(ref authority) if !authority_in_allowlist(authority, allowed_hosts) => { - warn!("mcp_origin_layer - rejected request: Host not in allowlist host = {authority}"); + warn!( + component = "Security", + operation = "validate_origin", + host = %authority, + rejection_reason = "host_not_allowed", + "request rejected by origin validation" + ); return forbidden_response(); }, - Some(_) => debug!("mcp_origin_layer - Host is in allowlist"), + Some(_) => debug!(component = "Security", operation = "validate_origin", "request host accepted"), } } let Some(origin_header) = request.headers().get(header::ORIGIN) else { - debug!("mcp_origin_layer - no Origin header, allowing request"); + debug!(component = "Security", operation = "validate_origin", "request has no Origin header"); return next.run(request).await; }; let Ok(origin_str) = origin_header.to_str() else { - warn!("mcp_origin_layer - rejected non-UTF-8 Origin header"); + warn!( + component = "Security", + operation = "validate_origin", + rejection_reason = "invalid_origin_encoding", + "request rejected by origin validation" + ); return forbidden_response(); }; let Some(request_origin) = parse_origin(origin_str) else { - warn!("mcp_origin_layer - rejected malformed Origin header origin = {origin_str}"); + warn!( + component = "Security", + operation = "validate_origin", + rejection_reason = "malformed_origin", + "request rejected by origin validation" + ); return forbidden_response(); }; let Some(ref allowed_origins) = config.mcp_allowed_origins else { - warn!("mcp_origin_layer - rejected Origin: no allowed origins configured origin = {origin_str}"); + warn!( + component = "Security", + operation = "validate_origin", + rejection_reason = "origin_allowlist_missing", + "request rejected by origin validation" + ); return forbidden_response(); }; let allowed = allowed_origins.iter().map(Url::origin).any(|o| o == request_origin); if allowed { - debug!("mcp_origin_layer - Origin accepted via allowlist origin = {origin_str}"); + debug!(component = "Security", operation = "validate_origin", origin = origin_str, "request origin accepted"); next.run(request).await } else { - warn!("mcp_origin_layer - rejected Origin not in allowlist origin = {origin_str}"); + warn!( + component = "Security", + operation = "validate_origin", + origin = origin_str, + rejection_reason = "origin_not_allowed", + "request rejected by origin validation" + ); forbidden_response() } } diff --git a/crates/contextforge-data-plane-lib/src/layers/session_id.rs b/crates/contextforge-data-plane-lib/src/layers/session_id.rs index 62f84264..def519f2 100644 --- a/crates/contextforge-data-plane-lib/src/layers/session_id.rs +++ b/crates/contextforge-data-plane-lib/src/layers/session_id.rs @@ -39,7 +39,7 @@ pub async fn session_id_layer(State(state): State, mut request: .map(|_| MOCK_SESSION_ID.to_owned()); if let Some(session_id) = &session_id { - info!("MCP Session ID {session_id}"); + info!(component = "Session", operation = "attach_session", "MCP session attached to request"); request.extensions_mut().insert(SessionId { value: session_id.clone() }); } diff --git a/crates/contextforge-data-plane-lib/src/layers/user_config_store.rs b/crates/contextforge-data-plane-lib/src/layers/user_config_store.rs index 4d5804c2..fcfff0b9 100644 --- a/crates/contextforge-data-plane-lib/src/layers/user_config_store.rs +++ b/crates/contextforge-data-plane-lib/src/layers/user_config_store.rs @@ -2,7 +2,7 @@ use axum::{body::Body, extract::State, middleware::Next, response::Response}; use contextforge_data_plane_apis::User; use http::{StatusCode, header}; //use openid::Claims; -use tracing::{debug, info, warn}; +use tracing::{debug, error, info, warn}; use crate::{ common::{ContextForgeClaims, ContextForgeDataPlaneAppState}, @@ -20,21 +20,27 @@ pub async fn user_config_store_layer( if let Some(claims) = maybe_claims { let subject = claims.sub.clone(); debug!( - "user_config_store_layer - getting user config for request subject = {subject} method = {method} path = {path}" + component = "UserConfig", + operation = "load", + method = %method, + path, + "loading user config" ); match state.config_store.get_config(&User::new(&subject)).await { Ok(user_config) => { let virtual_hosts = user_config.virtual_hosts.len(); - info!( - "user_config_store_layer - loaded user config subject = {subject} virtual_hosts = {virtual_hosts}" - ); + info!(component = "UserConfig", operation = "load", virtual_hosts, "user config loaded"); request.extensions_mut().insert(user_config); next.run(request).await }, Err(ConfigStoreError::NoDataForKey) => { debug!( - "user_config_store_layer - user config lookup returned no data subject = {subject} method = {method} path = {path}" + component = "UserConfig", + operation = "load", + method = %method, + path, + "user config was not found" ); Response::builder() .status(StatusCode::BAD_REQUEST) @@ -44,8 +50,18 @@ pub async fn user_config_store_layer( }, Err(error) => { - debug!( - "user_config_store_layer - user config lookup failed subject = {subject} method = {method} path = {path} error = {error}" + error!( + component = "UserConfig", + operation = "load", + method = %method, + path, + error_code = "CFDP-USER-CONFIG-LOAD", + root_cause = %error, + impact_scope = "request", + retryable = true, + http_status = 500_u16, + error = %error, + "user config lookup failed" ); Response::builder() .status(StatusCode::INTERNAL_SERVER_ERROR) @@ -55,7 +71,13 @@ pub async fn user_config_store_layer( }, } } else { - warn!("user_config_store_layer - no claims found in request extensions method = {method} path = {path}"); + warn!( + component = "Authorization", + operation = "load_user_config", + method = %method, + path, + "request has no authorization claims" + ); Response::builder() .status(StatusCode::BAD_REQUEST) .header(header::CONTENT_TYPE, "text/plain") diff --git a/crates/contextforge-data-plane-lib/src/layers/virtual_host_config.rs b/crates/contextforge-data-plane-lib/src/layers/virtual_host_config.rs index 7634e58d..fa070e6b 100644 --- a/crates/contextforge-data-plane-lib/src/layers/virtual_host_config.rs +++ b/crates/contextforge-data-plane-lib/src/layers/virtual_host_config.rs @@ -17,7 +17,11 @@ pub async fn virtual_host_config_layer(request: http::Request, let virtual_host_id = virtual_host_id.value(); let virtual_hosts = user_config.virtual_hosts.len(); debug!( - "virtual_host_config_layer - virtual host config missing virtual_host_id = {virtual_host_id} virtual_hosts = {virtual_hosts}" + component = "Routing", + operation = "load_virtual_host_config", + virtual_host_id, + virtual_hosts, + "virtual host config is missing" ); return server_not_found_response(); } diff --git a/crates/contextforge-data-plane-lib/src/layers/virtual_host_id.rs b/crates/contextforge-data-plane-lib/src/layers/virtual_host_id.rs index ccdf55cd..d2900dbf 100644 --- a/crates/contextforge-data-plane-lib/src/layers/virtual_host_id.rs +++ b/crates/contextforge-data-plane-lib/src/layers/virtual_host_id.rs @@ -20,12 +20,17 @@ impl VirtualHostId { pub async fn virtual_host_id_layer(mut request: http::Request, next: Next) -> Response { let path = request.uri().path().to_owned(); - debug!("virtual_host_id_layer - extracting virtual host from path path = {path}"); + debug!(component = "Routing", operation = "extract_virtual_host", path, "extracting virtual host from path"); if let Some(virtual_host_id) = extract_virtual_host_id(&path) { request.extensions_mut().insert(virtual_host_id); next.run(request).await } else { - debug!("virtual_host_id_layer - failed to extract virtual host id from request path path = {path}"); + debug!( + component = "Routing", + operation = "extract_virtual_host", + path, + "virtual host ID could not be extracted from the request path" + ); Response::builder() .status(StatusCode::BAD_REQUEST) .header(header::CONTENT_TYPE, "text/plain") diff --git a/crates/contextforge-data-plane-lib/src/lib.rs b/crates/contextforge-data-plane-lib/src/lib.rs index 4a7f9b32..1cc27fb3 100644 --- a/crates/contextforge-data-plane-lib/src/lib.rs +++ b/crates/contextforge-data-plane-lib/src/lib.rs @@ -14,7 +14,6 @@ mod common; mod const_values; mod gateway; mod layers; -mod telemetry; mod transports; #[cfg(feature = "with_tools")] @@ -31,7 +30,8 @@ use typed_builder::TypedBuilder; pub use user_config_store::RedisUserConfigStore; pub use user_config_store::{ConfigStoreError, UserConfigStore}; -pub use crate::common::{Config, LogRotation, OtlpProtocol}; +pub use crate::common::Config; +pub use contextforge_data_plane_observability::{LogRotation, OtlpProtocol}; pub type Error = Box; pub type Result = std::result::Result; @@ -156,7 +156,12 @@ impl Gateway { let app = app.with_state(mcp_add_state); let app = axum::Router::new() .nest("/contextforge-rs", app) - .layer(TraceLayer::new_for_http().make_span_with(telemetry::ExtractingMakeSpan)) + .layer( + TraceLayer::new_for_http() + .make_span_with(contextforge_data_plane_observability::ExtractingMakeSpan) + .on_response(contextforge_data_plane_observability::LogOnResponse), + ) + .layer(middleware::from_fn(contextforge_data_plane_observability::correlation_layer)) .layer(HttpMetricsLayerBuilder::new().build()); let mut handlers = vec![]; diff --git a/crates/contextforge-data-plane-lib/src/telemetry.rs b/crates/contextforge-data-plane-lib/src/telemetry.rs deleted file mode 100644 index b866a136..00000000 --- a/crates/contextforge-data-plane-lib/src/telemetry.rs +++ /dev/null @@ -1,103 +0,0 @@ -//! W3C trace-context propagation glue. -//! -//! The OTLP exporter is configured in the binary crate; this module wires the -//! two request seams so the gateway continues the caller's distributed trace -//! and passes it on to backend MCP servers. All functions here are no-ops -//! unless a global text-map propagator is installed (see the binary's -//! `init_tracing_logging`), so they are safe to call when OTel is disabled. - -use std::collections::HashMap; - -use opentelemetry::global; -use opentelemetry::propagation::{Extractor, Injector}; -use tower_http::trace::MakeSpan; -use tracing::Span; -use tracing_opentelemetry::OpenTelemetrySpanExt; - -/// Reads inbound `http::HeaderMap` for the text-map propagator (extract side). -struct HeaderExtractor<'a>(&'a http::HeaderMap); - -impl Extractor for HeaderExtractor<'_> { - fn get(&self, key: &str) -> Option<&str> { - self.0.get(key).and_then(|value| value.to_str().ok()) - } - - fn keys(&self) -> Vec<&str> { - self.0.keys().map(http::HeaderName::as_str).collect() - } -} - -/// Writes the outbound backend header map for the text-map propagator (inject -/// side). Malformed keys/values are dropped rather than propagated. -struct HeaderInjector<'a>(&'a mut HashMap); - -impl Injector for HeaderInjector<'_> { - fn set(&mut self, key: &str, value: String) { - if let (Ok(name), Ok(value)) = - (http::HeaderName::from_bytes(key.as_bytes()), http::HeaderValue::from_str(&value)) - { - self.0.insert(name, value); - } - } -} - -/// [`MakeSpan`] that opens the per-request span and re-parents it onto any W3C -/// trace context found in the inbound headers, so the gateway span continues -/// the caller's trace instead of starting a fresh one. -/// -/// Body-generic on purpose: the outermost `TraceLayer` sees whatever body type -/// the server hands it, and a struct impl avoids pinning that down. -#[derive(Clone, Copy, Debug, Default)] -pub struct ExtractingMakeSpan; - -impl MakeSpan for ExtractingMakeSpan { - fn make_span(&mut self, request: &http::Request) -> Span { - let span = tracing::info_span!( - "http-request", - method = %request.method(), - uri = %request.uri(), - version = ?request.version(), - ); - let parent = - global::get_text_map_propagator(|propagator| propagator.extract(&HeaderExtractor(request.headers()))); - // Errors only when no OTel layer is registered (OTel disabled); ignore. - let _ = span.set_parent(parent); - span - } -} - -/// Injects the current span's trace context into `headers` so the trace -/// propagates to the backend MCP server. No-op when there is no valid active -/// context (e.g. OTel disabled): the propagator writes nothing. -pub fn inject_current_context(headers: &mut HashMap) { - let context = Span::current().context(); - global::get_text_map_propagator(|propagator| propagator.inject_context(&context, &mut HeaderInjector(headers))); -} - -#[cfg(test)] -mod tests { - use opentelemetry::propagation::TextMapPropagator; - use opentelemetry_sdk::propagation::TraceContextPropagator; - - use super::*; - - #[test] - fn round_trips_traceparent_through_extract_and_inject() { - // A well-formed W3C traceparent carrying a known trace id. - let trace_id = "0af7651916cd43dd8448eb211c80319c"; - let traceparent = format!("00-{trace_id}-b7ad6b7169203331-01"); - - let mut inbound = http::HeaderMap::new(); - inbound.insert("traceparent", http::HeaderValue::from_str(&traceparent).unwrap()); - - let propagator = TraceContextPropagator::new(); - let parent = propagator.extract(&HeaderExtractor(&inbound)); - - let mut outbound = HashMap::new(); - propagator.inject_context(&parent, &mut HeaderInjector(&mut outbound)); - - let injected = outbound.get(&http::HeaderName::from_static("traceparent")).expect("traceparent injected"); - // Same trace id must survive extract -> inject (span id differs). - assert!(injected.to_str().unwrap().contains(trace_id)); - } -} diff --git a/crates/contextforge-data-plane-lib/src/transports/tcp.rs b/crates/contextforge-data-plane-lib/src/transports/tcp.rs index cc468bbd..cceeab4e 100644 --- a/crates/contextforge-data-plane-lib/src/transports/tcp.rs +++ b/crates/contextforge-data-plane-lib/src/transports/tcp.rs @@ -16,13 +16,19 @@ impl Tcp { } pub async fn handle_tcp(self, service: Router) -> crate::Result<()> { - info!("Starting TCP listener at {}", self.address); + info!( + component = "Transport", + operation = "listen", + transport = "tcp", + address = %self.address, + "listener starting" + ); let tcp_listener: TcpListener = self.try_into()?; Ok(axum::serve(tcp_listener, service) .with_graceful_shutdown(async { tokio::signal::ctrl_c().await.ok(); - info!("Shutting down..."); + info!(component = "Transport", operation = "shutdown", transport = "tcp", "listener shutting down"); }) .await?) } diff --git a/crates/contextforge-data-plane-lib/src/transports/tls.rs b/crates/contextforge-data-plane-lib/src/transports/tls.rs index 0f227e5b..b826fad4 100644 --- a/crates/contextforge-data-plane-lib/src/transports/tls.rs +++ b/crates/contextforge-data-plane-lib/src/transports/tls.rs @@ -9,7 +9,7 @@ use rustls_pki_types::{self, CertificateDer, PrivateKeyDer, pem::PemObject}; use tokio::net::TcpListener; use tokio_rustls::TlsAcceptor; use tower::Service; -use tracing::{error, info, warn}; +use tracing::{info, warn}; use crate::{Config, Error, transports::tcp::Tcp}; @@ -48,7 +48,13 @@ impl TryFrom<&Config> for Option { impl DownstreamTls { pub async fn handle_tls(self, service: Router) -> crate::Result<()> { let DownstreamTls { tcp, server_config } = self; - info!("Starting TLS listener at {}", tcp.address); + info!( + component = "Transport", + operation = "listen", + transport = "tls", + address = %tcp.address, + "listener starting" + ); let tcp_listener: TcpListener = tcp.try_into()?; let tls_acceptor = TlsAcceptor::from(Arc::new(server_config)); @@ -59,11 +65,24 @@ impl DownstreamTls { let tower_service = service.clone(); let tls_acceptor = tls_acceptor.clone(); - if let Ok((tcp_stream, addr)) = maybe_stream { + if let Ok((tcp_stream, _addr)) = maybe_stream { tokio::spawn(async move { - let Ok(stream) = tls_acceptor.accept(tcp_stream).await else { - error!("error during tls handshake connection from {}", addr); - return; + let stream = match tls_acceptor.accept(tcp_stream).await { + Ok(stream) => stream, + Err(error) => { + tracing::error!( + component = "Transport", + operation = "tls_handshake", + transport = "tls", + error_code = "CFDP-TLS-HANDSHAKE", + root_cause = %error, + impact_scope = "connection", + retryable = true, + error = %error, + "TLS handshake failed" + ); + return; + }, }; let stream = TokioIo::new(stream); @@ -77,11 +96,23 @@ impl DownstreamTls { .await; if let Err(err) = ret { - warn!("error serving connection from {addr}: {err}"); + warn!( + component = "Transport", + operation = "serve_connection", + transport = "tls", + error = %err, + "TLS connection terminated with an error" + ); } }) } else { - warn!("Problem during TCP handshake {maybe_stream:?}"); + warn!( + component = "Transport", + operation = "accept_connection", + transport = "tcp", + error = ?maybe_stream, + "TCP connection accept failed" + ); return Err(maybe_stream.expect_err("Expect this to work").into()); }; } diff --git a/crates/contextforge-data-plane-lib/src/user_config_store/redis_config_store.rs b/crates/contextforge-data-plane-lib/src/user_config_store/redis_config_store.rs index 442675bb..30642c6f 100644 --- a/crates/contextforge-data-plane-lib/src/user_config_store/redis_config_store.rs +++ b/crates/contextforge-data-plane-lib/src/user_config_store/redis_config_store.rs @@ -5,6 +5,7 @@ use std::{ use async_trait::async_trait; use contextforge_data_plane_apis::{User, user_store::UserConfig}; +use contextforge_data_plane_observability::PerformanceTimer; use lru_time_cache::LruCache; use redis::{ AsyncCommands, RedisError, @@ -12,7 +13,7 @@ use redis::{ cmd, }; use tokio::sync::Mutex; -use tracing::{debug, warn}; +use tracing::{debug, error}; use super::{ConfigStoreError, UserConfigStore}; use crate::{ @@ -47,16 +48,26 @@ pub struct RedisUserConfigStore { impl RedisUserConfigStore { pub async fn new(redis_client: &RedisClient, cache_expiry: Duration) -> crate::Result { + let mut timer = PerformanceTimer::database("UserConfig", "connect"); + let connection = redis_client + .get_connection_manager_with_config(ConnectionManagerConfig::default().set_number_of_retries(REDIS_RETRIES)) + .await; + timer.record_result(&connection); Ok(Self { - connection: redis_client - .get_connection_manager_with_config( - ConnectionManagerConfig::default().set_number_of_retries(REDIS_RETRIES), - ) - .await - .map_err(|error| { - warn!("RedisUserConfigStore::new - failed to create Redis user config connection error = {error}"); - ConfigStoreError::InvalidConnection - })?, + connection: connection.map_err(|error| { + tracing::error!( + component = "UserConfig", + operation = "connect", + dependency = "redis", + error_code = "CFDP-USER-CONFIG-CONNECT", + root_cause = %error, + impact_scope = "service-startup", + retryable = true, + error = %error, + "user config store connection failed" + ); + ConfigStoreError::InvalidConnection + })?, cache: (!cache_expiry.is_zero()).then(|| { Arc::new(Mutex::new(LruCache::with_expiry_duration_and_capacity(cache_expiry, LRU_CACHE_ENTRIES))) }), @@ -75,41 +86,77 @@ impl UserConfigStore for RedisUserConfigStore { if entry.is_fresh(self.cache_expiry) { let virtual_hosts = entry.config.virtual_hosts.len(); debug!( - "RedisUserConfigStore::get_config - user config cache hit subject = {subject} virtual_hosts = {virtual_hosts}" + component = "UserConfig", + operation = "cache_get", + cache_result = "hit", + virtual_hosts, + "user config cache lookup completed" ); return Ok(entry.config.clone()); } - debug!("RedisUserConfigStore::get_config - user config cache entry expired subject = {subject}"); + debug!( + component = "UserConfig", + operation = "cache_get", + cache_result = "expired", + "user config cache lookup completed" + ); } else { - debug!("RedisUserConfigStore::get_config - user config cache miss subject = {subject}"); + debug!( + component = "UserConfig", + operation = "cache_get", + cache_result = "miss", + "user config cache lookup completed" + ); } } let Ok(key) = rmp_serde::encode::to_vec::(user_key) else { - warn!("RedisUserConfigStore::get_config - failed to encode Redis user config key subject = {subject}"); + error!( + component = "UserConfig", + operation = "encode_key", + error_code = "CFDP-USER-CONFIG-ENCODE", + root_cause = "user config key serialization failed", + impact_scope = "request", + retryable = false, + "user config key encoding failed" + ); return Err(ConfigStoreError::DataEncoding); }; let mut connection = self.connection.clone(); + let mut timer = PerformanceTimer::database("UserConfig", "read"); let maybe_user_config: Result>, RedisError> = cmd("GET").arg(key).take().query_async(&mut connection).await; + timer.record_result(&maybe_user_config); let user_config = match maybe_user_config { Ok(Some(user_config)) => { let bytes = user_config.len(); debug!( - "RedisUserConfigStore::get_config - loaded user config blob from Redis subject = {subject} bytes = {bytes}" + component = "UserConfig", + operation = "read", + dependency = "redis", + bytes, + "user config blob loaded" ); user_config }, Ok(None) => { - debug!("RedisUserConfigStore::get_config - no user config found in Redis subject = {subject}"); + debug!(component = "UserConfig", operation = "read", dependency = "redis", "user config was not found"); return Err(ConfigStoreError::NoDataForKey); }, Err(error) => { - warn!( - "RedisUserConfigStore::get_config - failed to load user config from Redis subject = {subject} error = {error}" + error!( + component = "UserConfig", + operation = "read", + dependency = "redis", + error_code = "CFDP-USER-CONFIG-READ", + root_cause = %error, + impact_scope = "request", + retryable = true, + error = %error, + "user config store read failed" ); return Err(ConfigStoreError::NoDataForKey); }, @@ -118,17 +165,22 @@ impl UserConfigStore for RedisUserConfigStore { let user_config = match rmp_serde::decode::from_slice::(&user_config) { Ok(user_config) => user_config, Err(error) => { - warn!( - "RedisUserConfigStore::get_config - failed to decode Redis user config blob subject = {subject} error = {error}" + error!( + component = "UserConfig", + operation = "decode", + error_code = "CFDP-USER-CONFIG-DECODE", + root_cause = %error, + impact_scope = "request", + retryable = false, + error = %error, + "user config blob decoding failed" ); return Err(ConfigStoreError::DataWrongFormat); }, }; let virtual_hosts = user_config.virtual_hosts.len(); - debug!( - "RedisUserConfigStore::get_config - decoded user config subject = {subject} virtual_hosts = {virtual_hosts}" - ); + debug!(component = "UserConfig", operation = "decode", virtual_hosts, "user config decoded"); if let Some(cache) = &self.cache { cache.lock().await.insert(subject.to_owned(), CachedUserConfig::new(user_config.clone())); @@ -140,26 +192,49 @@ impl UserConfigStore for RedisUserConfigStore { let subject = user_key.key(); let Ok(key) = rmp_serde::encode::to_vec::(user_key) else { - warn!("RedisUserConfigStore::set_config - failed to encode Redis user config key subject = {subject}"); + error!( + component = "UserConfig", + operation = "encode_key", + error_code = "CFDP-USER-CONFIG-ENCODE", + root_cause = "user config key serialization failed", + impact_scope = "request", + retryable = false, + "user config key encoding failed" + ); return Err(ConfigStoreError::DataEncoding); }; let Ok(encoded) = rmp_serde::encode::to_vec::(config) else { let virtual_hosts = config.virtual_hosts.len(); - warn!( - "RedisUserConfigStore::set_config - failed to encode user config subject = {subject} virtual_hosts = {virtual_hosts}" + error!( + component = "UserConfig", + operation = "encode", + virtual_hosts, + error_code = "CFDP-USER-CONFIG-ENCODE", + root_cause = "user config serialization failed", + impact_scope = "request", + retryable = false, + "user config encoding failed" ); return Err(ConfigStoreError::DataEncoding); }; let mut connection = self.connection.clone(); + let mut timer = PerformanceTimer::database("UserConfig", "write"); + let result = connection.set::<&[u8], &[u8], String>(&key, &encoded).await; + timer.record_result(&result); - match connection.set::<&[u8], &[u8], String>(&key, &encoded).await { + match result { Ok(_) => { let bytes = encoded.len(); let virtual_hosts = config.virtual_hosts.len(); debug!( - "RedisUserConfigStore::set_config - wrote user config to Redis subject = {subject} bytes = {bytes} virtual_hosts = {virtual_hosts}" + component = "UserConfig", + operation = "write", + dependency = "redis", + bytes, + virtual_hosts, + "user config written" ); if let Some(cache) = &self.cache { cache.lock().await.insert(subject.to_owned(), CachedUserConfig::new(config.clone())); @@ -167,8 +242,16 @@ impl UserConfigStore for RedisUserConfigStore { Ok(()) }, Err(error) => { - warn!( - "RedisUserConfigStore::set_config - failed to write user config to Redis subject = {subject} error = {error}" + error!( + component = "UserConfig", + operation = "write", + dependency = "redis", + error_code = "CFDP-USER-CONFIG-WRITE", + root_cause = %error, + impact_scope = "request", + retryable = true, + error = %error, + "user config store write failed" ); Err(ConfigStoreError::CantWriteData) }, diff --git a/crates/contextforge-data-plane-lib/tests/gateway_completions.rs b/crates/contextforge-data-plane-lib/tests/gateway_completions.rs index 4cd2f68c..121a1e96 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_completions.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_completions.rs @@ -63,7 +63,7 @@ async fn plaintext_complete_for_unrouted_reference_errors() -> Result<()> { } async fn assert_prompt_completion(gateway_url: String, client: reqwest::Client) -> Result<()> { - info!("Sending request to {gateway_url}"); + info!(component = "Test", operation = "complete", gateway_url, "sending gateway request"); let running_service = connect_client(gateway_url, client).await?; // Spec-compliant clients only issue completion/complete when the server advertises the diff --git a/crates/contextforge-data-plane-lib/tests/gateway_list_tools.rs b/crates/contextforge-data-plane-lib/tests/gateway_list_tools.rs index 04a77859..ed0b8fe4 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_list_tools.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_list_tools.rs @@ -37,9 +37,9 @@ async fn plaintext_lists_prefixed_backend_tools() -> Result<()> { handle.abort(); if maybe_passed.is_ok() { - info!("Test passed"); + info!(component = "Test", operation = "list_tools", outcome = "success", "test completed"); } else { - info!("Test NOT passed {maybe_passed:?}"); + info!(component = "Test", operation = "list_tools", outcome = "error", error = ?maybe_passed, "test completed"); panic!() } @@ -87,9 +87,9 @@ async fn tls_lists_prefixed_backend_tools() -> Result<()> { handle.abort(); if maybe_passed.is_ok() { - info!("Test passed"); + info!(component = "Test", operation = "list_tools", outcome = "success", "test completed"); } else { - info!("Test NOT passed {maybe_passed:?}"); + info!(component = "Test", operation = "list_tools", outcome = "error", error = ?maybe_passed, "test completed"); panic!() } @@ -101,23 +101,29 @@ async fn assert_list_tools( client: reqwest::Client, expected_tool_names: Vec, ) -> Result<()> { - info!("Seding request to {gateway_url}"); + info!(component = "Test", operation = "list_tools", gateway_url, "sending gateway request"); let running_service = connect_client(gateway_url, client).await?; let list_tools = running_service.list_tools(None).await; let Ok(list_tools) = list_tools else { let msg = format!("List tools returned error {list_tools:?}"); - warn!(msg); + warn!(component = "Test", operation = "list_tools", error = %msg, "gateway request failed"); return Err(msg.into()); }; let mut names: Vec = list_tools.tools.iter().map(|t| t.name.to_string()).collect(); names.sort(); - info!("Tool names {names:#?}"); + info!(component = "Test", operation = "list_tools", tool_names = ?names, "tool names received"); if expected_tool_names != names { - warn!("Actual {names:#?} Expected {expected_tool_names:#?}"); + warn!( + component = "Test", + operation = "list_tools", + actual = ?names, + expected = ?expected_tool_names, + "tool names did not match" + ); return Err("Expected tool names don't match actual".into()); } diff --git a/crates/contextforge-data-plane-lib/tests/gateway_pagination.rs b/crates/contextforge-data-plane-lib/tests/gateway_pagination.rs index 558f98d5..64a71de8 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_pagination.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_pagination.rs @@ -71,7 +71,7 @@ async fn start_gateway(config: Config, virtual_host_id: &str, user_config: UserC tokio::spawn(async move { let res = gateway.run_gateway().await; - warn!("Gateway exited {res:?}"); + warn!(component = "Test", operation = "run_gateway", result = ?res, "gateway exited"); }); gateway_url diff --git a/crates/contextforge-data-plane-lib/tests/gateway_prompts.rs b/crates/contextforge-data-plane-lib/tests/gateway_prompts.rs index 53b022a3..24c4af36 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_prompts.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_prompts.rs @@ -73,14 +73,14 @@ async fn assert_list_prompts( client: reqwest::Client, expected_prompt_names: Vec, ) -> Result<()> { - info!("Sending request to {gateway_url}"); + info!(component = "Test", operation = "list_prompts", gateway_url, "sending gateway request"); let running_service = connect_client(gateway_url, client).await?; let list_prompts = running_service.list_prompts(None).await; let Ok(list_prompts) = list_prompts else { let msg = format!("List prompts returned error {list_prompts:?}"); - warn!(msg); + warn!(component = "Test", operation = "list_prompts", error = %msg, "gateway request failed"); return Err(msg.into()); }; @@ -88,7 +88,13 @@ async fn assert_list_prompts( names.sort(); if expected_prompt_names != names { - warn!("Actual {names:#?} Expected {expected_prompt_names:#?}"); + warn!( + component = "Test", + operation = "list_prompts", + actual = ?names, + expected = ?expected_prompt_names, + "prompt names did not match" + ); return Err("Expected prompt names don't match actual".into()); } diff --git a/crates/contextforge-data-plane-lib/tests/gateway_resource_templates.rs b/crates/contextforge-data-plane-lib/tests/gateway_resource_templates.rs index 9ae3284a..2022a76d 100644 --- a/crates/contextforge-data-plane-lib/tests/gateway_resource_templates.rs +++ b/crates/contextforge-data-plane-lib/tests/gateway_resource_templates.rs @@ -78,14 +78,14 @@ async fn assert_list_resource_templates( expected_template_names: Vec, expected_template_uris: Vec, ) -> Result<()> { - info!("Sending request to {gateway_url}"); + info!(component = "Test", operation = "list_resource_templates", gateway_url, "sending gateway request"); let running_service = connect_client(gateway_url, client).await?; let list_templates = running_service.list_resource_templates(None).await; let Ok(list_templates) = list_templates else { let msg = format!("List resource templates returned error {list_templates:?}"); - warn!(msg); + warn!(component = "Test", operation = "list_resource_templates", error = %msg, "gateway request failed"); return Err(msg.into()); }; @@ -93,7 +93,13 @@ async fn assert_list_resource_templates( names.sort(); if expected_template_names != names { - warn!("Actual {names:#?} Expected {expected_template_names:#?}"); + warn!( + component = "Test", + operation = "list_resource_templates", + actual = ?names, + expected = ?expected_template_names, + "resource template names did not match" + ); return Err("Expected resource template names don't match actual".into()); } @@ -101,7 +107,13 @@ async fn assert_list_resource_templates( uris.sort(); if expected_template_uris != uris { - warn!("Actual {uris:#?} Expected {expected_template_uris:#?}"); + warn!( + component = "Test", + operation = "list_resource_templates", + actual = ?uris, + expected = ?expected_template_uris, + "resource template URIs did not match" + ); return Err("Expected resource template uris don't match actual".into()); } diff --git a/crates/contextforge-data-plane-lib/tests/support/client.rs b/crates/contextforge-data-plane-lib/tests/support/client.rs index d0d7e59c..17d33936 100644 --- a/crates/contextforge-data-plane-lib/tests/support/client.rs +++ b/crates/contextforge-data-plane-lib/tests/support/client.rs @@ -61,11 +61,11 @@ where match handler.clone().serve(transport).await { Ok(running_service) => return Ok(running_service), Err(error) if Instant::now() < deadline => { - warn!("No Service {error:?}"); + warn!(component = "Test", operation = "connect_client", error = ?error, "gateway service unavailable"); tokio::time::sleep(TEST_POLL_INTERVAL).await; }, Err(error) => { - warn!("No Service {error:?}"); + warn!(component = "Test", operation = "connect_client", error = ?error, "gateway service unavailable"); return Err("Couldn't get a service".into()); }, } diff --git a/crates/contextforge-data-plane-lib/tests/support/list_tools_gateway.rs b/crates/contextforge-data-plane-lib/tests/support/list_tools_gateway.rs index 074bd9b1..59ffe973 100644 --- a/crates/contextforge-data-plane-lib/tests/support/list_tools_gateway.rs +++ b/crates/contextforge-data-plane-lib/tests/support/list_tools_gateway.rs @@ -94,7 +94,7 @@ pub(crate) async fn create_gateway_with_four_counters(user: &str, config: Config let gateway = async move { let res = gateway.run_gateway().await; - warn!("Gateway exited with result {res:?}"); + warn!(component = "Test", operation = "run_gateway", result = ?res, "gateway exited"); Ok(()) } .boxed(); @@ -173,7 +173,7 @@ pub(crate) async fn create_tls_gateway_with_four_tls_counters( let gateway = async move { let res = gateway.run_gateway().await; - warn!("Gateway exited with result {res:?}"); + warn!(component = "Test", operation = "run_gateway", result = ?res, "gateway exited"); Ok(()) } .boxed(); diff --git a/crates/contextforge-data-plane-lib/tests/support/mock_counter.rs b/crates/contextforge-data-plane-lib/tests/support/mock_counter.rs index 08dcd545..cc6ad1a4 100644 --- a/crates/contextforge-data-plane-lib/tests/support/mock_counter.rs +++ b/crates/contextforge-data-plane-lib/tests/support/mock_counter.rs @@ -297,9 +297,13 @@ impl ServerHandler for Counter { context: RequestContext, ) -> Result { if let Some(http_request_part) = context.extensions.get::() { - let initialize_headers = &http_request_part.headers; - let initialize_uri = &http_request_part.uri; - tracing::info!(?initialize_headers, %initialize_uri, "initialize from http server"); + tracing::info!( + component = "TestBackend", + operation = "initialize", + header_count = http_request_part.headers.len(), + path = http_request_part.uri.path(), + "backend initialization received" + ); } Ok(self.get_info()) } diff --git a/crates/contextforge-data-plane-observability/Cargo.toml b/crates/contextforge-data-plane-observability/Cargo.toml new file mode 100644 index 00000000..a9d819dd --- /dev/null +++ b/crates/contextforge-data-plane-observability/Cargo.toml @@ -0,0 +1,38 @@ +[package] +name = "contextforge-data-plane-observability" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true +keywords.workspace = true +homepage.workspace = true +readme.workspace = true +repository.workspace = true + +[dependencies] +axum.workspace = true +chrono.workspace = true +clap.workspace = true +http.workspace = true +opentelemetry = "0.32" +opentelemetry_sdk = { version = "0.32", features = ["rt-tokio", "metrics"] } +serde_json.workspace = true +sha2 = "0.10.9" +tokio.workspace = true +tonic = "0.14" +tower-http = { workspace = true, features = ["trace"] } +tracing.workspace = true +tracing-appender = "0.2.3" +tracing-opentelemetry = "0.33.0" +tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] } +opentelemetry-otlp = { version = "0.32", features = ["grpc-tonic", "http-proto", "reqwest-blocking-client", "metrics"] } +uuid.workspace = true + +[dev-dependencies] +tower.workspace = true + +[lints] +workspace = true + +[lib] +doctest = false diff --git a/crates/contextforge-data-plane-observability/src/context.rs b/crates/contextforge-data-plane-observability/src/context.rs new file mode 100644 index 00000000..1e6a8ab9 --- /dev/null +++ b/crates/contextforge-data-plane-observability/src/context.rs @@ -0,0 +1,418 @@ +//! HTTP request correlation and W3C trace-context propagation. + +use std::{collections::HashMap, fmt::Write as _, time::Duration}; + +use axum::{extract::Request, middleware::Next, response::Response}; +use opentelemetry::global; +use opentelemetry::propagation::{Extractor, Injector}; +use opentelemetry::trace::TraceContextExt; +use sha2::{Digest, Sha256}; +use tower_http::trace::{MakeSpan, OnResponse}; +use tracing::Span; +use tracing_opentelemetry::OpenTelemetrySpanExt; +use uuid::Uuid; + +pub const TRANSACTION_ID_HEADER: http::HeaderName = http::HeaderName::from_static("x-contextforge-transaction-id"); +pub const CORRELATION_ID_HEADER: http::HeaderName = http::HeaderName::from_static("x-contextforge-correlation-id"); + +tokio::task_local! { + static CURRENT_CONTEXT: RequestContext; +} + +#[derive(Clone, Debug, Eq, PartialEq)] +struct TraceFields { + trace_id: String, + span_id: String, + trace_flags: String, +} + +impl TraceFields { + fn from_headers(headers: &http::HeaderMap) -> Self { + headers + .get(http::header::HeaderName::from_static("traceparent")) + .and_then(|value| value.to_str().ok()) + .and_then(parse_traceparent) + .unwrap_or_else(Self::generated) + } + + fn generated() -> Self { + let trace_id = Uuid::new_v4().simple().to_string(); + let span_id = Uuid::new_v4().simple().to_string()[..16].to_owned(); + Self { trace_id, span_id, trace_flags: "01".to_owned() } + } + + fn traceparent(&self) -> String { + format!("00-{}-{}-{}", self.trace_id, self.span_id, self.trace_flags) + } +} + +fn parse_traceparent(value: &str) -> Option { + let mut parts = value.trim().split('-'); + let version = parts.next()?; + let trace_id = parts.next()?; + let span_id = parts.next()?; + let trace_flags = parts.next()?; + if parts.next().is_some() + || version.len() != 2 + || version.eq_ignore_ascii_case("ff") + || trace_id.len() != 32 + || span_id.len() != 16 + || trace_flags.len() != 2 + || ![version, trace_id, span_id, trace_flags] + .iter() + .all(|part| part.bytes().all(|byte| byte.is_ascii_hexdigit())) + || trace_id.bytes().all(|byte| byte == b'0') + || span_id.bytes().all(|byte| byte == b'0') + { + return None; + } + Some(TraceFields { + trace_id: trace_id.to_ascii_lowercase(), + span_id: span_id.to_ascii_lowercase(), + trace_flags: trace_flags.to_ascii_lowercase(), + }) +} + +#[derive(Clone, Debug, Eq, PartialEq)] +struct RequestContext { + transaction_id: String, + correlation_id: String, + trace: TraceFields, +} + +impl RequestContext { + fn from_headers(headers: &http::HeaderMap) -> Self { + let incoming_correlation = uuid_header(headers, &CORRELATION_ID_HEADER); + let transaction_id = non_empty_header(headers, &TRANSACTION_ID_HEADER) + .or_else(|| incoming_correlation.clone()) + .unwrap_or_else(|| Uuid::new_v4().to_string()); + let correlation_id = incoming_correlation.unwrap_or_else(|| Uuid::new_v4().to_string()); + Self { transaction_id, correlation_id, trace: TraceFields::from_headers(headers) } + } + + fn insert_response_headers(&self, headers: &mut http::HeaderMap) { + if let Ok(value) = http::HeaderValue::from_str(&self.transaction_id) { + headers.insert(TRANSACTION_ID_HEADER, value); + } + if let Ok(value) = http::HeaderValue::from_str(&self.correlation_id) { + headers.insert(CORRELATION_ID_HEADER, value); + } + } + + fn insert_outbound_headers(&self, headers: &mut HashMap) + where + S: std::hash::BuildHasher, + { + insert_header(headers, TRANSACTION_ID_HEADER, &self.transaction_id); + insert_header(headers, CORRELATION_ID_HEADER, &self.correlation_id); + insert_header(headers, http::header::HeaderName::from_static("traceparent"), &self.trace.traceparent()); + } +} + +fn insert_header(headers: &mut HashMap, name: http::HeaderName, value: &str) +where + S: std::hash::BuildHasher, +{ + if let Ok(value) = http::HeaderValue::from_str(value) { + headers.insert(name, value); + } +} + +fn non_empty_header(headers: &http::HeaderMap, name: &http::HeaderName) -> Option { + headers + .get(name) + .and_then(|value| value.to_str().ok()) + .map(str::trim) + .filter(|value| !value.is_empty() && value.len() <= 128) + .map(ToOwned::to_owned) +} + +fn uuid_header(headers: &http::HeaderMap, name: &http::HeaderName) -> Option { + non_empty_header(headers, name).and_then(|value| Uuid::parse_str(&value).ok().map(|id| id.to_string())) +} + +pub async fn correlation_layer(mut request: Request, next: Next) -> Response { + let context = RequestContext::from_headers(request.headers()); + request.extensions_mut().insert(context.clone()); + let mut response = CURRENT_CONTEXT.scope(context.clone(), next.run(request)).await; + context.insert_response_headers(response.headers_mut()); + response +} + +struct HeaderExtractor<'a>(&'a http::HeaderMap); + +impl Extractor for HeaderExtractor<'_> { + fn get(&self, key: &str) -> Option<&str> { + self.0.get(key).and_then(|value| value.to_str().ok()) + } + + fn keys(&self) -> Vec<&str> { + self.0.keys().map(http::HeaderName::as_str).collect() + } +} + +struct HeaderInjector<'a, S>(&'a mut HashMap); + +impl Injector for HeaderInjector<'_, S> +where + S: std::hash::BuildHasher, +{ + fn set(&mut self, key: &str, value: String) { + if let (Ok(name), Ok(value)) = + (http::HeaderName::from_bytes(key.as_bytes()), http::HeaderValue::from_str(&value)) + { + self.0.insert(name, value); + } + } +} + +#[derive(Clone, Copy, Debug, Default)] +pub struct ExtractingMakeSpan; + +impl MakeSpan for ExtractingMakeSpan { + fn make_span(&mut self, request: &http::Request) -> Span { + let request_context = request + .extensions() + .get::() + .cloned() + .unwrap_or_else(|| RequestContext::from_headers(request.headers())); + let span = tracing::info_span!( + "http-request", + transaction_id = %request_context.transaction_id, + correlation_id = %request_context.correlation_id, + trace_id = %request_context.trace.trace_id, + span_id = %request_context.trace.span_id, + user_id = tracing::field::Empty, + component = "HttpServer", + operation = "http_request", + http_method = %request.method(), + http_path = request.uri().path(), + http_version = ?request.version(), + ); + let parent = + global::get_text_map_propagator(|propagator| propagator.extract(&HeaderExtractor(request.headers()))); + let _ = span.set_parent(parent); + let context = span.context(); + let otel_span = context.span(); + let span_context = otel_span.span_context(); + if span_context.is_valid() { + span.record("trace_id", span_context.trace_id().to_string()); + span.record("span_id", span_context.span_id().to_string()); + } + span + } +} + +#[derive(Clone, Copy, Debug, Default)] +pub struct LogOnResponse; + +impl OnResponse for LogOnResponse { + fn on_response(self, response: &http::Response, latency: Duration, _span: &Span) { + let http_status = response.status().as_u16(); + let latency_ms = u64::try_from(latency.as_millis()).unwrap_or(u64::MAX); + if response.status().is_server_error() { + tracing::error!( + component = "HttpServer", + operation = "http_request", + event_type = "PERFORMANCE", + metric = "request_latency", + outcome = "error", + http_status, + latency_ms, + error_code = "CFDP-HTTP-SERVER", + root_cause = "request returned a server error", + impact_scope = "single-request", + retryable = false, + "HTTP request failed" + ); + } else if response.status().is_client_error() { + tracing::warn!( + component = "HttpServer", + operation = "http_request", + event_type = "PERFORMANCE", + metric = "request_latency", + outcome = "client_error", + http_status, + latency_ms, + "HTTP request rejected" + ); + } else { + tracing::info!( + component = "HttpServer", + operation = "http_request", + event_type = "PERFORMANCE", + metric = "request_latency", + outcome = "success", + http_status, + latency_ms, + "HTTP request completed" + ); + } + } +} + +/// Adds request correlation and trace context to an outbound backend request. +pub fn inject_current_context(headers: &mut HashMap) +where + S: std::hash::BuildHasher, +{ + if let Ok(context) = CURRENT_CONTEXT.try_with(Clone::clone) { + context.insert_outbound_headers(headers); + } + let context = Span::current().context(); + global::get_text_map_propagator(|propagator| propagator.inject_context(&context, &mut HeaderInjector(headers))); +} + +/// Records a stable pseudonym for an authenticated subject on the current request span. +pub fn record_authenticated_user(subject: &str) { + Span::current().record("user_id", pseudonymous_user_id(subject)); +} + +fn pseudonymous_user_id(subject: &str) -> String { + let digest = Sha256::digest(subject.as_bytes()); + let mut pseudonym = String::with_capacity(19); + pseudonym.push_str("sha256:"); + for byte in &digest[..6] { + let _ = write!(pseudonym, "{byte:02x}"); + } + pseudonym +} + +#[cfg(test)] +mod tests { + use axum::{Router, body::Body, middleware, routing::get}; + use http::{Request, StatusCode}; + use opentelemetry::propagation::TextMapPropagator; + use opentelemetry_sdk::propagation::TraceContextPropagator; + use tower::ServiceExt; + + use super::*; + + const CORRELATION_ID: &str = "a663c5c5-a4b2-4f01-97a0-4aee5749d41e"; + + #[test] + fn round_trips_traceparent_through_extract_and_inject() { + let trace_id = "0af7651916cd43dd8448eb211c80319c"; // pragma: allowlist secret + let traceparent = format!("00-{trace_id}-b7ad6b7169203331-01"); + let mut inbound = http::HeaderMap::new(); + inbound.insert("traceparent", http::HeaderValue::from_str(&traceparent).unwrap()); + + let propagator = TraceContextPropagator::new(); + let parent = propagator.extract(&HeaderExtractor(&inbound)); + let mut outbound = HashMap::new(); + propagator.inject_context(&parent, &mut HeaderInjector(&mut outbound)); + + let injected = outbound.get(&http::HeaderName::from_static("traceparent")).expect("traceparent injected"); + assert!(injected.to_str().unwrap().contains(trace_id)); + } + + #[tokio::test] + async fn correlation_layer_preserves_valid_incoming_ids() { + let app = Router::new() + .route("/", get(|| async { StatusCode::NO_CONTENT })) + .layer(middleware::from_fn(correlation_layer)); + let request = Request::builder() + .header(TRANSACTION_ID_HEADER, "transaction-1") + .header(CORRELATION_ID_HEADER, CORRELATION_ID) + .body(Body::empty()) + .expect("request should build"); + + let response = app.oneshot(request).await.expect("request should complete"); + + assert_eq!(response.headers()[TRANSACTION_ID_HEADER], "transaction-1"); + assert_eq!(response.headers()[CORRELATION_ID_HEADER], CORRELATION_ID); + } + + #[tokio::test] + async fn invalid_correlation_id_is_replaced() { + let app = Router::new() + .route("/", get(|| async { StatusCode::NO_CONTENT })) + .layer(middleware::from_fn(correlation_layer)); + let request = Request::builder() + .header(TRANSACTION_ID_HEADER, "transaction-1") + .header(CORRELATION_ID_HEADER, "not-a-uuid") + .body(Body::empty()) + .expect("request should build"); + + let response = app.oneshot(request).await.expect("request should complete"); + + assert_eq!(response.headers()[TRANSACTION_ID_HEADER], "transaction-1"); + assert!(Uuid::parse_str(response.headers()[CORRELATION_ID_HEADER].to_str().unwrap()).is_ok()); + } + + #[tokio::test] + async fn correlation_layer_generates_ids_when_missing() { + let app = Router::new() + .route("/", get(|| async { StatusCode::NO_CONTENT })) + .layer(middleware::from_fn(correlation_layer)); + + let response = app.oneshot(Request::new(Body::empty())).await.expect("request should complete"); + + let transaction_id = response.headers()[TRANSACTION_ID_HEADER].to_str().expect("valid transaction ID"); + let correlation_id = response.headers()[CORRELATION_ID_HEADER].to_str().expect("valid correlation ID"); + assert!(Uuid::parse_str(transaction_id).is_ok()); + assert!(Uuid::parse_str(correlation_id).is_ok()); + } + + #[tokio::test] + async fn valid_correlation_id_becomes_transaction_id_when_transaction_id_is_missing() { + let app = Router::new() + .route("/", get(|| async { StatusCode::NO_CONTENT })) + .layer(middleware::from_fn(correlation_layer)); + let request = Request::builder() + .header(CORRELATION_ID_HEADER, CORRELATION_ID) + .body(Body::empty()) + .expect("request should build"); + + let response = app.oneshot(request).await.expect("request should complete"); + + assert_eq!(response.headers()[TRANSACTION_ID_HEADER], CORRELATION_ID); + assert_eq!(response.headers()[CORRELATION_ID_HEADER], CORRELATION_ID); + } + + #[tokio::test] + async fn outbound_requests_receive_context_without_an_exporter() { + let context = RequestContext { + transaction_id: "transaction-1".to_owned(), + correlation_id: CORRELATION_ID.to_owned(), + trace: TraceFields { + trace_id: "0af7651916cd43dd8448eb211c80319c".to_owned(), // pragma: allowlist secret + span_id: "b7ad6b7169203331".to_owned(), // pragma: allowlist secret + trace_flags: "01".to_owned(), + }, + }; + + let headers = CURRENT_CONTEXT + .scope(context, async { + let mut headers = HashMap::new(); + inject_current_context(&mut headers); + headers + }) + .await; + + assert_eq!(headers[&TRANSACTION_ID_HEADER], "transaction-1"); + assert_eq!(headers[&CORRELATION_ID_HEADER], CORRELATION_ID); + assert_eq!( + headers[&http::HeaderName::from_static("traceparent")], + "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01" + ); + } + + #[test] + fn traceparent_parser_rejects_zero_and_malformed_ids() { + assert!(parse_traceparent("00-00000000000000000000000000000000-b7ad6b7169203331-01").is_none()); + assert!(parse_traceparent("00-0af7651916cd43dd8448eb211c80319c-0000000000000000-01").is_none()); + assert!(parse_traceparent("malformed").is_none()); + } + + #[test] + fn authenticated_subjects_receive_stable_non_reversible_ids() { + let subject = "private-user-subject"; + let first = pseudonymous_user_id(subject); + + assert_eq!(first, pseudonymous_user_id(subject)); + assert!(first.starts_with("sha256:")); + assert_eq!(first.len(), 19); + assert!(!first.contains(subject)); + } +} diff --git a/crates/contextforge-data-plane-observability/src/formatter.rs b/crates/contextforge-data-plane-observability/src/formatter.rs new file mode 100644 index 00000000..d493b39d --- /dev/null +++ b/crates/contextforge-data-plane-observability/src/formatter.rs @@ -0,0 +1,302 @@ +use std::fmt; + +use chrono::{SecondsFormat, Utc}; +use serde_json::{Map, Value}; +use tracing::{Event, Subscriber}; +use tracing_subscriber::{ + fmt::{ + FmtContext, + format::{FormatEvent, FormatFields, Json, Writer}, + }, + registry::LookupSpan, +}; + +use crate::LoggingConfig; + +pub(crate) const DEFAULT_SERVICE_NAME: &str = "contextforge-data-plane"; +const UNCLASSIFIED_ERROR_CODE: &str = "CFDP-UNCLASSIFIED"; +const SPAN_FIELDS: &[&str] = + &["transaction_id", "correlation_id", "trace_id", "span_id", "user_id", "component", "operation"]; + +fn configured_value(value: Option<&str>, fallback: &str) -> String { + value.map(str::trim).filter(|value| !value.is_empty()).unwrap_or(fallback).to_owned() +} + +fn has_meaningful_value(value: Option<&Value>) -> bool { + value.is_some_and(|value| match value { + Value::Null => false, + Value::String(value) => !value.trim().is_empty(), + _ => true, + }) +} + +#[derive(Clone, Debug)] +pub(crate) struct LoggingMetadata { + service_name: String, + version: String, + environment: String, + cluster_id: String, +} + +impl LoggingMetadata { + pub(crate) fn from_config(configuration: &LoggingConfig) -> Self { + Self { + service_name: configured_value(configuration.service_name.as_deref(), DEFAULT_SERVICE_NAME), + version: configured_value(Some(&configuration.version), "unknown"), + environment: configured_value(configuration.environment.as_deref(), "unknown"), + cluster_id: configured_value(configuration.cluster_id.as_deref(), "unknown"), + } + } + + pub(crate) fn service_name(&self) -> &str { + &self.service_name + } + + pub(crate) fn version(&self) -> &str { + &self.version + } + + pub(crate) fn environment(&self) -> &str { + &self.environment + } + + pub(crate) fn cluster_id(&self) -> &str { + &self.cluster_id + } +} + +#[derive(Clone, Debug)] +pub(crate) struct StructuredJsonFormatter { + inner: tracing_subscriber::fmt::format::Format, + metadata: LoggingMetadata, +} + +impl StructuredJsonFormatter { + pub(crate) fn new(metadata: LoggingMetadata) -> Self { + Self { + inner: tracing_subscriber::fmt::format() + .json() + .flatten_event(true) + .with_current_span(true) + .with_span_list(true), + metadata, + } + } + + fn merge_span_fields(object: &mut Map, span: &Map) { + for key in SPAN_FIELDS { + if let Some(value) = span.get(*key) { + object.entry((*key).to_owned()).or_insert_with(|| value.clone()); + } + } + } + + fn inherit_span_fields(object: &mut Map) { + let span_list = object.get("spans").and_then(Value::as_array).cloned().unwrap_or_default(); + for span in span_list.iter().filter_map(Value::as_object) { + Self::merge_span_fields(object, span); + } + if let Some(span) = object.get("span").and_then(Value::as_object).cloned() { + Self::merge_span_fields(object, &span); + } + object.remove("span"); + object.remove("spans"); + } + + fn apply_error_contract(object: &mut Map) { + if !has_meaningful_value(object.get("error_code")) { + object.insert("error_code".to_owned(), Value::String(UNCLASSIFIED_ERROR_CODE.to_owned())); + } + if !has_meaningful_value(object.get("root_cause")) { + let root_cause = object + .get("error") + .filter(|value| has_meaningful_value(Some(value))) + .cloned() + .or_else(|| object.get("message").filter(|value| has_meaningful_value(Some(value))).cloned()) + .unwrap_or_else(|| Value::String("root cause unavailable".to_owned())); + object.insert("root_cause".to_owned(), root_cause); + } + object.entry("impact_scope".to_owned()).or_insert_with(|| Value::String("unknown".to_owned())); + object.entry("retryable".to_owned()).or_insert(Value::Bool(false)); + object.entry("http_status".to_owned()).or_insert(Value::Null); + object.entry("stack_trace".to_owned()).or_insert(Value::Null); + } + + fn apply_contract(&self, object: &mut Map) { + Self::inherit_span_fields(object); + + let mut log_level = object.remove("level").unwrap_or_else(|| Value::String("INFO".to_owned())); + if object.remove("fatal").and_then(|value| value.as_bool()) == Some(true) { + log_level = Value::String("FATAL".to_owned()); + } + object.insert("log_level".to_owned(), log_level); + object.insert("timestamp".to_owned(), Value::String(Utc::now().to_rfc3339_opts(SecondsFormat::Micros, true))); + object.insert("service_name".to_owned(), Value::String(self.metadata.service_name.clone())); + object.insert("version".to_owned(), Value::String(self.metadata.version.clone())); + object.insert("environment".to_owned(), Value::String(self.metadata.environment.clone())); + object.insert("cluster_id".to_owned(), Value::String(self.metadata.cluster_id.clone())); + + for key in ["transaction_id", "correlation_id", "trace_id", "span_id", "user_id"] { + object.entry(key.to_owned()).or_insert(Value::Null); + } + object.entry("message".to_owned()).or_insert_with(|| Value::String(String::new())); + let target = object.get("target").and_then(Value::as_str).unwrap_or(DEFAULT_SERVICE_NAME).to_owned(); + object.entry("component".to_owned()).or_insert_with(|| Value::String(target)); + + let is_error = matches!(object.get("log_level").and_then(Value::as_str), Some("ERROR" | "FATAL")); + if is_error { + Self::apply_error_contract(object); + } else { + object.entry("error_code".to_owned()).or_insert(Value::Null); + } + } +} + +impl FormatEvent for StructuredJsonFormatter +where + S: Subscriber + for<'lookup> LookupSpan<'lookup>, + N: for<'writer> FormatFields<'writer> + 'static, +{ + fn format_event(&self, context: &FmtContext<'_, S, N>, mut writer: Writer<'_>, event: &Event<'_>) -> fmt::Result { + let mut encoded = String::new(); + self.inner.format_event(context, Writer::new(&mut encoded), event)?; + let mut value: Value = serde_json::from_str(encoded.trim()).map_err(|_| fmt::Error)?; + let object = value.as_object_mut().ok_or(fmt::Error)?; + self.apply_contract(object); + let encoded = serde_json::to_string(object).map_err(|_| fmt::Error)?; + writer.write_str(&encoded)?; + writer.write_char('\n') + } +} + +#[cfg(test)] +mod tests { + use std::{ + io::Write, + sync::{Arc, Mutex}, + }; + + use serde_json::Value; + use tracing_subscriber::{Registry, fmt::MakeWriter, layer::SubscriberExt}; + + use super::{LoggingMetadata, StructuredJsonFormatter, UNCLASSIFIED_ERROR_CODE}; + use crate::LoggingConfig; + + #[derive(Clone, Default)] + struct SharedBuffer(Arc>>); + + struct BufferWriter(SharedBuffer); + + impl Write for BufferWriter { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.0.lock().unwrap_or_else(std::sync::PoisonError::into_inner).extend_from_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + impl<'writer> MakeWriter<'writer> for SharedBuffer { + type Writer = BufferWriter; + + fn make_writer(&'writer self) -> Self::Writer { + BufferWriter(self.clone()) + } + } + + fn metadata() -> LoggingMetadata { + LoggingMetadata { + service_name: "contextforge-data-plane".to_owned(), + version: "1.2.3".to_owned(), + environment: "onprem".to_owned(), + cluster_id: "cluster-test-01".to_owned(), + } + } + + fn event_from(run: impl FnOnce()) -> Value { + let output = SharedBuffer::default(); + let subscriber = Registry::default().with( + tracing_subscriber::fmt::layer() + .event_format(StructuredJsonFormatter::new(metadata())) + .fmt_fields(tracing_subscriber::fmt::format::JsonFields::new()) + .with_ansi(false) + .with_writer(output.clone()), + ); + tracing::subscriber::with_default(subscriber, run); + let bytes = output.0.lock().unwrap_or_else(std::sync::PoisonError::into_inner).clone(); + serde_json::from_slice(&bytes).expect("log line should be valid JSON") + } + + #[test] + fn metadata_uses_safe_defaults() { + let metadata = LoggingMetadata::from_config(&LoggingConfig::default()); + + assert_eq!(metadata.service_name(), "contextforge-data-plane"); + assert_eq!(metadata.version(), "unknown"); + assert_eq!(metadata.environment(), "unknown"); + assert_eq!(metadata.cluster_id(), "unknown"); + } + + #[test] + fn every_event_contains_the_structured_contract() { + let event = event_from(|| { + let span = tracing::info_span!( + "request", + transaction_id = "txn-1", + correlation_id = "corr-1", + trace_id = "00000000000000000000000000000001", + span_id = "0000000000000001", + user_id = "sha256:001122334455" + ); + let _entered = span.enter(); + tracing::info!(component = "Routing", operation = "test_logging", "structured event"); + }); + + assert!(event["timestamp"].as_str().is_some_and(|value| value.ends_with('Z'))); + assert_eq!(event["service_name"], "contextforge-data-plane"); + assert_eq!(event["version"], "1.2.3"); + assert_eq!(event["environment"], "onprem"); + assert_eq!(event["cluster_id"], "cluster-test-01"); + assert_eq!(event["transaction_id"], "txn-1"); + assert_eq!(event["correlation_id"], "corr-1"); + assert_eq!(event["trace_id"], "00000000000000000000000000000001"); + assert_eq!(event["span_id"], "0000000000000001"); + assert_eq!(event["user_id"], "sha256:001122334455"); + assert_eq!(event["log_level"], "INFO"); + assert_eq!(event["error_code"], Value::Null); + assert_eq!(event["message"], "structured event"); + assert_eq!(event["component"], "Routing"); + assert!(event.get("span").is_none()); + assert!(event.get("spans").is_none()); + } + + #[test] + fn error_events_receive_safe_defaults() { + let event = event_from(|| tracing::error!(error = "serialization failed")); + + assert_eq!(event["error_code"], UNCLASSIFIED_ERROR_CODE); + assert_eq!(event["root_cause"], "serialization failed"); + assert_eq!(event["impact_scope"], "unknown"); + assert_eq!(event["retryable"], false); + assert!(event.get("http_status").is_some()); + assert!(event.get("stack_trace").is_some()); + } + + #[test] + fn fatal_marker_sets_fatal_level() { + let event = event_from(|| { + tracing::error!( + fatal = true, + error_code = "CFDP-BOOTSTRAP", + root_cause = "startup failed", + impact_scope = "service-wide", + "service terminated" + ); + }); + + assert_eq!(event["log_level"], "FATAL"); + assert!(event.get("fatal").is_none()); + } +} diff --git a/crates/contextforge-data-plane-observability/src/lib.rs b/crates/contextforge-data-plane-observability/src/lib.rs new file mode 100644 index 00000000..6b59a7b7 --- /dev/null +++ b/crates/contextforge-data-plane-observability/src/lib.rs @@ -0,0 +1,328 @@ +//! Shared structured logging, request correlation, and telemetry support for +//! ContextForge data-plane processes. + +mod context; +mod formatter; +mod performance; + +use std::collections::HashMap; + +use chrono::{SecondsFormat, Utc}; +use clap::ValueEnum; +use formatter::{LoggingMetadata, StructuredJsonFormatter}; +use opentelemetry::global; +use opentelemetry::trace::TracerProvider; +use opentelemetry_otlp::{MetricExporter, Protocol, SpanExporter, WithExportConfig, WithHttpConfig, WithTonicConfig}; +use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; +use opentelemetry_sdk::trace::{RandomIdGenerator, Sampler}; +use tonic::metadata::{MetadataKey, MetadataMap, MetadataValue}; +use tracing_appender::non_blocking::WorkerGuard; +use tracing_subscriber::{Layer, Registry, fmt, layer::SubscriberExt, util::SubscriberInitExt}; + +pub use context::{ + CORRELATION_ID_HEADER, ExtractingMakeSpan, LogOnResponse, TRANSACTION_ID_HEADER, correlation_layer, + inject_current_context, record_authenticated_user, +}; +pub use performance::PerformanceTimer; + +type Error = Box; + +const DEFAULT_GRPC_TRACES_ENDPOINT: &str = "http://127.0.0.1:4317"; +const DEFAULT_GRPC_METRICS_ENDPOINT: &str = "http://127.0.0.1:4317"; +const DEFAULT_HTTP_TRACES_ENDPOINT: &str = "http://127.0.0.1:4318/v1/traces"; +const DEFAULT_HTTP_METRICS_ENDPOINT: &str = "http://127.0.0.1:4318/v1/metrics"; +const METRICS_EXPORT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(30); + +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, ValueEnum, Default)] +pub enum LogRotation { + Minutely, + #[default] + Hourly, + Daily, + Never, +} + +/// Wire protocol used to export OpenTelemetry data. +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, ValueEnum, Default)] +pub enum OtlpProtocol { + #[default] + Grpc, + HttpProtobuf, +} + +/// Application-owned settings consumed by [`init_observability`]. +#[derive(Clone, Debug)] +pub struct LoggingConfig { + pub service_name: Option, + pub version: String, + pub environment: Option, + pub cluster_id: Option, + pub log_name: Option, + pub log_rotation: LogRotation, + pub enable_open_telemetry: bool, + pub enable_otel_metrics: bool, + pub otlp_endpoint: Option, + pub otlp_metrics_endpoint: Option, + pub otlp_protocol: OtlpProtocol, + pub otlp_headers: Option, +} + +impl Default for LoggingConfig { + fn default() -> Self { + Self { + service_name: None, + version: "unknown".to_owned(), + environment: None, + cluster_id: None, + log_name: None, + log_rotation: LogRotation::default(), + enable_open_telemetry: false, + enable_otel_metrics: false, + otlp_endpoint: None, + otlp_metrics_endpoint: None, + otlp_protocol: OtlpProtocol::default(), + otlp_headers: None, + } + } +} + +/// Holds background exporter and file-writer handles for the process lifetime. +#[allow(dead_code)] +pub struct Guard { + appender: WorkerGuard, + meter_provider: Option, + tracer_provider: Option, +} + +impl Drop for Guard { + fn drop(&mut self) { + if let Some(provider) = self.meter_provider.take() { + let _ = provider.shutdown(); + } + if let Some(provider) = self.tracer_provider.take() { + let _ = provider.shutdown(); + } + } +} + +/// Installs the process-wide subscriber. Applications should call this once at startup. +pub fn init_observability(configuration: &LoggingConfig) -> Result { + let registry = Registry::default(); + let metadata = LoggingMetadata::from_config(configuration); + let log_name = configuration.log_name.clone().unwrap_or("contextforge-data-plane.log".to_owned()); + + let file_appender = match configuration.log_rotation { + LogRotation::Minutely => tracing_appender::rolling::minutely(".", log_name), + LogRotation::Hourly => tracing_appender::rolling::hourly(".", log_name), + LogRotation::Daily => tracing_appender::rolling::daily(".", log_name), + LogRotation::Never => tracing_appender::rolling::never(".", log_name), + }; + + let (non_blocking_appender, guard) = tracing_appender::non_blocking(file_appender); + let file_filter = + tracing_subscriber::EnvFilter::new(std::env::var("RUST_FILE_LOG").unwrap_or_else(|_| "info".to_owned())); + let console_filter = + tracing_subscriber::EnvFilter::new(std::env::var("RUST_LOG").unwrap_or_else(|_| "info".to_owned())); + let tracing_filter = + tracing_subscriber::EnvFilter::new(std::env::var("RUST_TRACE_LOG").unwrap_or_else(|_| "info".to_owned())); + + let console_layer = fmt::layer() + .event_format(StructuredJsonFormatter::new(metadata.clone())) + .fmt_fields(fmt::format::JsonFields::new()) + .with_ansi(false) + .with_filter(console_filter); + + let file_layer = fmt::layer() + .with_writer(non_blocking_appender) + .event_format(StructuredJsonFormatter::new(metadata.clone())) + .fmt_fields(fmt::format::JsonFields::new()) + .with_ansi(false) + .with_filter(file_filter); + + global::set_text_map_propagator(opentelemetry_sdk::propagation::TraceContextPropagator::new()); + + if configuration.enable_open_telemetry { + let service_name = metadata.service_name().to_owned(); + let headers = parse_otlp_headers(configuration.otlp_headers.as_deref())?; + let exporter = build_span_exporter(configuration, &headers)?; + let tracer_provider = opentelemetry_sdk::trace::SdkTracerProvider::builder() + .with_batch_exporter(exporter) + .with_id_generator(RandomIdGenerator::default()) + .with_sampler(Sampler::AlwaysOn) + .with_resource( + opentelemetry_sdk::Resource::builder() + .with_attributes(vec![opentelemetry::KeyValue::new("service.name", service_name.clone())]) + .build(), + ) + .build(); + let tracer = tracer_provider.tracer(service_name.clone()); + let telemetry = tracing_opentelemetry::layer().with_tracer(tracer); + let meter_provider = init_meter_provider(configuration, &service_name)?; + + registry.with(console_layer).with(file_layer).with(telemetry.with_filter(tracing_filter)).init(); + Ok(Guard { appender: guard, meter_provider, tracer_provider: Some(tracer_provider) }) + } else { + let meter_provider = init_meter_provider(configuration, metadata.service_name())?; + registry.with(console_layer).with(file_layer).init(); + Ok(Guard { appender: guard, meter_provider, tracer_provider: None }) + } +} + +fn build_span_exporter( + configuration: &LoggingConfig, + headers: &HashMap, +) -> Result { + match configuration.otlp_protocol { + OtlpProtocol::Grpc => { + let endpoint = configuration + .otlp_endpoint + .as_ref() + .map_or_else(|| DEFAULT_GRPC_TRACES_ENDPOINT.to_owned(), ToString::to_string); + Ok(SpanExporter::builder() + .with_tonic() + .with_endpoint(endpoint) + .with_metadata(headers_to_metadata(headers)?) + .with_timeout(std::time::Duration::from_secs(3)) + .build()?) + }, + OtlpProtocol::HttpProtobuf => { + let endpoint = configuration + .otlp_endpoint + .as_ref() + .map_or_else(|| DEFAULT_HTTP_TRACES_ENDPOINT.to_owned(), ToString::to_string); + let mut builder = SpanExporter::builder() + .with_http() + .with_endpoint(endpoint) + .with_protocol(Protocol::HttpBinary) + .with_timeout(std::time::Duration::from_secs(10)); + if !headers.is_empty() { + builder = builder.with_headers(headers.clone()); + } + Ok(builder.build()?) + }, + } +} + +#[allow(clippy::print_stderr)] +pub fn emit_bootstrap_failure(configuration: &LoggingConfig, error: &dyn std::fmt::Display) { + let metadata = LoggingMetadata::from_config(configuration); + let event = serde_json::json!({ + "timestamp": Utc::now().to_rfc3339_opts(SecondsFormat::Micros, true), + "service_name": metadata.service_name(), + "version": metadata.version(), + "environment": metadata.environment(), + "cluster_id": metadata.cluster_id(), + "transaction_id": serde_json::Value::Null, + "correlation_id": serde_json::Value::Null, + "trace_id": serde_json::Value::Null, + "span_id": serde_json::Value::Null, + "user_id": serde_json::Value::Null, + "log_level": "FATAL", + "error_code": "CFDP-BOOTSTRAP", + "message": "logging initialization failed", + "component": "Bootstrap", + "root_cause": error.to_string(), + "impact_scope": "service-wide", + "retryable": false, + "http_status": serde_json::Value::Null, + "stack_trace": serde_json::Value::Null, + }); + eprintln!("{event}"); +} + +fn init_meter_provider(configuration: &LoggingConfig, service_name: &str) -> Result, Error> { + if !configuration.enable_otel_metrics { + return Ok(None); + } + + let headers = parse_otlp_headers(configuration.otlp_headers.as_deref())?; + let exporter = match configuration.otlp_protocol { + OtlpProtocol::Grpc => { + let endpoint = configuration + .otlp_metrics_endpoint + .as_ref() + .map_or_else(|| DEFAULT_GRPC_METRICS_ENDPOINT.to_owned(), ToString::to_string); + MetricExporter::builder() + .with_tonic() + .with_endpoint(endpoint) + .with_metadata(headers_to_metadata(&headers)?) + .with_timeout(std::time::Duration::from_secs(3)) + .build()? + }, + OtlpProtocol::HttpProtobuf => { + let endpoint = configuration + .otlp_metrics_endpoint + .as_ref() + .map_or_else(|| DEFAULT_HTTP_METRICS_ENDPOINT.to_owned(), ToString::to_string); + let mut builder = MetricExporter::builder() + .with_http() + .with_endpoint(endpoint) + .with_protocol(Protocol::HttpBinary) + .with_timeout(std::time::Duration::from_secs(10)); + if !headers.is_empty() { + builder = builder.with_headers(headers); + } + builder.build()? + }, + }; + + let reader = PeriodicReader::builder(exporter).with_interval(METRICS_EXPORT_INTERVAL).build(); + let provider = SdkMeterProvider::builder() + .with_reader(reader) + .with_resource( + opentelemetry_sdk::Resource::builder() + .with_attributes(vec![opentelemetry::KeyValue::new("service.name", service_name.to_owned())]) + .build(), + ) + .build(); + + global::set_meter_provider(provider.clone()); + Ok(Some(provider)) +} + +fn headers_to_metadata(headers: &HashMap) -> Result { + let mut map = MetadataMap::new(); + for (key, value) in headers { + let key = MetadataKey::from_bytes(key.as_bytes()) + .map_err(|error| format!("invalid gRPC metadata key {key:?}: {error}"))?; + let value = + MetadataValue::try_from(value.as_str()).map_err(|error| format!("invalid gRPC metadata value: {error}"))?; + map.insert(key, value); + } + Ok(map) +} + +fn parse_otlp_headers(raw: Option<&str>) -> Result, Error> { + let mut headers = HashMap::new(); + let Some(raw) = raw else { return Ok(headers) }; + for entry in raw.split(',').map(str::trim).filter(|entry| !entry.is_empty()) { + match entry.split_once('=') { + None => return Err(format!("malformed OTLP header entry (missing '=' separator): {entry:?}").into()), + Some((key, _)) if key.trim().is_empty() => { + return Err(format!("malformed OTLP header entry (empty key): {entry:?}").into()); + }, + Some((key, value)) => { + headers.insert(key.trim().to_owned(), value.trim().to_owned()); + }, + } + } + Ok(headers) +} + +#[cfg(test)] +mod tests { + use super::parse_otlp_headers; + + #[test] + fn parses_export_headers() { + let parsed = parse_otlp_headers(Some(" Authorization = Basic abc , X-Project=demo ")).unwrap(); + assert_eq!(parsed.get("Authorization"), Some(&"Basic abc".to_owned())); + assert_eq!(parsed.get("X-Project"), Some(&"demo".to_owned())); + } + + #[test] + fn rejects_malformed_export_headers() { + assert!(parse_otlp_headers(Some("no-equals")).is_err()); + assert!(parse_otlp_headers(Some("=missing-key")).is_err()); + } +} diff --git a/crates/contextforge-data-plane-observability/src/performance.rs b/crates/contextforge-data-plane-observability/src/performance.rs new file mode 100644 index 00000000..fe189b26 --- /dev/null +++ b/crates/contextforge-data-plane-observability/src/performance.rs @@ -0,0 +1,66 @@ +use std::time::Instant; + +/// Emits one structured latency event when it leaves scope. +#[must_use = "the timer must remain in scope for the operation being measured"] +pub struct PerformanceTimer { + started_at: Instant, + component: String, + operation: String, + metric: &'static str, + outcome: &'static str, +} + +impl PerformanceTimer { + pub fn database(component: impl Into, operation: impl Into) -> Self { + Self::new(component, operation, "database_latency") + } + + pub fn external_call(component: impl Into, operation: impl Into) -> Self { + Self::new(component, operation, "external_call_latency") + } + + pub fn queue_wait(component: impl Into, operation: impl Into) -> Self { + Self::new(component, operation, "queue_wait_latency") + } + + fn new(component: impl Into, operation: impl Into, metric: &'static str) -> Self { + Self { + started_at: Instant::now(), + component: component.into(), + operation: operation.into(), + metric, + outcome: "unknown", + } + } + + pub fn succeeded(&mut self) { + self.outcome = "success"; + } + + pub fn failed(&mut self) { + self.outcome = "error"; + } + + pub fn record_result(&mut self, result: &Result) { + if result.is_ok() { + self.succeeded(); + } else { + self.failed(); + } + } +} + +impl Drop for PerformanceTimer { + fn drop(&mut self) { + let latency_ms = u64::try_from(self.started_at.elapsed().as_millis()).unwrap_or(u64::MAX); + tracing::info!( + component = self.component, + operation = self.operation, + event_type = "PERFORMANCE", + metric = self.metric, + outcome = self.outcome, + latency_ms, + "operation latency recorded" + ); + } +} diff --git a/crates/contextforge-data-plane/Cargo.toml b/crates/contextforge-data-plane/Cargo.toml index d287c2ba..083a7d4b 100644 --- a/crates/contextforge-data-plane/Cargo.toml +++ b/crates/contextforge-data-plane/Cargo.toml @@ -13,6 +13,7 @@ repository.workspace = true [dependencies] contextforge-data-plane-cpex.workspace = true contextforge-data-plane-lib = { path = "../contextforge-data-plane-lib" } +contextforge-data-plane-observability.workspace = true # These demo plugins remain in their independently hosted legacy repository. cpex-payload-marker = { git = "https://github.com/contextforge-gateway-rs/cpex-plugins-rs", rev = "ab47801daccfbba44ea07b033034a347e7b5afdd", optional = true } cpex-text-prefixer = { git = "https://github.com/contextforge-gateway-rs/cpex-plugins-rs", rev = "ab47801daccfbba44ea07b033034a347e7b5afdd", optional = true } @@ -20,14 +21,7 @@ cpex-tool-namespace = { git = "https://github.com/contextforge-gateway-rs/cpex-p cpex-secrets-detection = { workspace = true, optional = true } clap.workspace = true tracing.workspace = true -tracing-appender = "0.2.3" -tracing-opentelemetry.workspace = true -tracing-subscriber = { version = "0.3", features = ["env-filter"] } tokio.workspace = true -opentelemetry.workspace = true -opentelemetry-otlp = { version = "0.32", features = ["grpc-tonic", "http-proto", "reqwest-blocking-client", "metrics"] } -opentelemetry_sdk.workspace = true -tonic = "0.14" num_cpus = "1.17.0" rmcp.workspace = true tikv-jemallocator = "0.7.0" @@ -46,7 +40,6 @@ openport.workspace = true redis.workspace = true reqwest.workspace = true rmp-serde.workspace = true -serde_json.workspace = true [lints] workspace = true diff --git a/crates/contextforge-data-plane/src/logging.rs b/crates/contextforge-data-plane/src/logging.rs deleted file mode 100644 index a695f7f8..00000000 --- a/crates/contextforge-data-plane/src/logging.rs +++ /dev/null @@ -1,278 +0,0 @@ -use std::collections::HashMap; - -use contextforge_data_plane_lib::{Config, LogRotation, OtlpProtocol}; -use opentelemetry::global; -use opentelemetry::trace::TracerProvider; -use opentelemetry_otlp::{MetricExporter, Protocol, SpanExporter, WithExportConfig, WithHttpConfig, WithTonicConfig}; -use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; -use opentelemetry_sdk::trace::{RandomIdGenerator, Sampler}; -use tonic::metadata::{MetadataKey, MetadataMap, MetadataValue}; -use tracing_appender::non_blocking::WorkerGuard; -use tracing_subscriber::{ - Layer, Registry, filter, - fmt::{self, format::FmtSpan}, - layer::SubscriberExt, - util::SubscriberInitExt, -}; - -/// Holds RAII handles whose lifetime must match the process so background -/// telemetry tasks keep running. The file appender's worker thread needs -/// the [`WorkerGuard`] to stay alive to flush logs on shutdown, and the -/// metrics [`SdkMeterProvider`] needs to stay alive so its -/// [`PeriodicReader`] task keeps exporting at the configured interval. -#[allow(dead_code)] -pub struct Guard { - appender: WorkerGuard, - meter_provider: Option, -} - -const CONTROLLER_NAME: &str = "CONTEXTFORGE-DATA-PLANE"; -const DEFAULT_GRPC_TRACES_ENDPOINT: &str = "http://127.0.0.1:4317"; -const DEFAULT_GRPC_METRICS_ENDPOINT: &str = "http://127.0.0.1:4317"; -const DEFAULT_HTTP_TRACES_ENDPOINT: &str = "http://127.0.0.1:4318/v1/traces"; -const DEFAULT_HTTP_METRICS_ENDPOINT: &str = "http://127.0.0.1:4318/v1/metrics"; -const METRICS_EXPORT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(30); - -pub fn init_tracing_logging(configuration: &Config) -> Result> { - let registry = Registry::default(); - - let log_name = configuration.log_name.clone().unwrap_or("contextforge-data-plane.log".to_owned()); - - let file_appender = match configuration.log_rotation.clone().unwrap_or_default() { - LogRotation::Minutely => tracing_appender::rolling::minutely(".", log_name), - LogRotation::Hourly => tracing_appender::rolling::hourly(".", log_name), - LogRotation::Daily => tracing_appender::rolling::daily(".", log_name), - LogRotation::Never => tracing_appender::rolling::never(".", log_name), - }; - - let (non_blocking_appender, guard) = tracing_appender::non_blocking(file_appender); - let file_filter = - tracing_subscriber::EnvFilter::new(std::env::var("RUST_FILE_LOG").unwrap_or_else(|_| "debug".to_owned())); - let console_filter = - tracing_subscriber::EnvFilter::new(std::env::var("RUST_LOG").unwrap_or_else(|_| "debug".to_owned())); - let tracing_filter = - tracing_subscriber::EnvFilter::new(std::env::var("RUST_TRACE_LOG").unwrap_or_else(|_| "info".to_owned())); - - let console_layer = fmt::layer() - .event_format(fmt::format().compact()) - .with_target(true) - .with_span_events(FmtSpan::NONE) - .with_ansi(false) - .with_filter(filter::filter_fn(|meta| !meta.is_span())) - .with_filter(console_filter); - - let file_layer = fmt::layer() - .with_writer(non_blocking_appender) - .with_target(true) - .with_span_events(FmtSpan::NONE) - .with_ansi(false) - .with_filter(filter::filter_fn(|meta| !meta.is_span())) - .with_filter(file_filter); - - if let Some(true) = configuration.enable_open_telemetry { - let protocol = configuration.otlp_protocol.clone().unwrap_or_default(); - let service_name = configuration.otlp_service_name.clone().unwrap_or_else(|| CONTROLLER_NAME.to_owned()); - let headers = parse_otlp_headers(configuration.otlp_headers.as_deref())?; - - let exporter = match protocol { - OtlpProtocol::Grpc => { - let endpoint = configuration - .otlp_endpoint - .as_ref() - .map_or_else(|| DEFAULT_GRPC_TRACES_ENDPOINT.to_owned(), ToString::to_string); - SpanExporter::builder() - .with_tonic() - .with_endpoint(endpoint) - .with_metadata(headers_to_metadata(&headers)?) - .with_timeout(std::time::Duration::from_secs(3)) - .build()? - }, - OtlpProtocol::HttpProtobuf => { - let endpoint = configuration - .otlp_endpoint - .as_ref() - .map_or_else(|| DEFAULT_HTTP_TRACES_ENDPOINT.to_owned(), ToString::to_string); - let mut builder = SpanExporter::builder() - .with_http() - .with_endpoint(endpoint) - .with_protocol(Protocol::HttpBinary) - .with_timeout(std::time::Duration::from_secs(10)); - if !headers.is_empty() { - builder = builder.with_headers(headers); - } - builder.build()? - }, - }; - - let tracer_provider = opentelemetry_sdk::trace::SdkTracerProvider::builder() - .with_batch_exporter(exporter) - .with_id_generator(RandomIdGenerator::default()) - .with_sampler(Sampler::AlwaysOn) - .with_resource( - opentelemetry_sdk::Resource::builder() - .with_attributes(vec![opentelemetry::KeyValue::new("service.name", service_name.clone())]) - .build(), - ) - .build(); - - // Install the W3C propagator so inbound `traceparent` is extracted and - // outbound requests carry it. Without this, inject/extract are no-ops. - global::set_text_map_propagator(opentelemetry_sdk::propagation::TraceContextPropagator::new()); - - let tracer = tracer_provider.tracer(CONTROLLER_NAME); - let telemetry = tracing_opentelemetry::layer().with_tracer(tracer); - - let meter_provider = init_meter_provider(configuration, &service_name)?; - - registry.with(console_layer).with(file_layer).with(telemetry.with_filter(tracing_filter)).init(); - - Ok(Guard { appender: guard, meter_provider }) - } else { - registry.with(console_layer).with(file_layer).init(); - Ok(Guard { appender: guard, meter_provider: None }) - } -} - -/// Builds an OTLP metrics pipeline and installs it as the process-wide -/// [`global::meter_provider`] when `enable_otel_metrics = true`. -/// -/// Mirrors the trace exporter's gRPC / HTTP protocol branching and reuses -/// the same `service.name` resource attribute so traces and metrics show up -/// under one identity. -/// -/// Returns `None` when metrics are disabled; the returned provider must be -/// held alive (via [`Guard`]) for the [`PeriodicReader`]'s background task -/// to keep exporting. -fn init_meter_provider( - configuration: &Config, - service_name: &str, -) -> Result, Box> { - if configuration.enable_otel_metrics != Some(true) { - return Ok(None); - } - - let protocol = configuration.otlp_protocol.clone().unwrap_or_default(); - let headers = parse_otlp_headers(configuration.otlp_headers.as_deref())?; - - let exporter = match protocol { - OtlpProtocol::Grpc => { - let endpoint = configuration - .otlp_metrics_endpoint - .as_ref() - .map_or_else(|| DEFAULT_GRPC_METRICS_ENDPOINT.to_owned(), ToString::to_string); - MetricExporter::builder() - .with_tonic() - .with_endpoint(endpoint) - .with_metadata(headers_to_metadata(&headers)?) - .with_timeout(std::time::Duration::from_secs(3)) - .build()? - }, - OtlpProtocol::HttpProtobuf => { - let endpoint = configuration - .otlp_metrics_endpoint - .as_ref() - .map_or_else(|| DEFAULT_HTTP_METRICS_ENDPOINT.to_owned(), ToString::to_string); - let mut builder = MetricExporter::builder() - .with_http() - .with_endpoint(endpoint) - .with_protocol(Protocol::HttpBinary) - .with_timeout(std::time::Duration::from_secs(10)); - if !headers.is_empty() { - builder = builder.with_headers(headers); - } - builder.build()? - }, - }; - - let reader = PeriodicReader::builder(exporter).with_interval(METRICS_EXPORT_INTERVAL).build(); - - let provider = SdkMeterProvider::builder() - .with_reader(reader) - .with_resource( - opentelemetry_sdk::Resource::builder() - .with_attributes(vec![opentelemetry::KeyValue::new("service.name", service_name.to_owned())]) - .build(), - ) - .build(); - - global::set_meter_provider(provider.clone()); - Ok(Some(provider)) -} - -/// Converts a header [`HashMap`] to a tonic [`MetadataMap`] for gRPC metadata -/// attachment. Returns an error (and aborts startup) for any entry whose key -/// or value cannot be encoded as valid ASCII gRPC metadata, so invalid -/// configuration is surfaced rather than silently dropped. -fn headers_to_metadata( - headers: &HashMap, -) -> Result> { - let mut map = MetadataMap::new(); - for (k, v) in headers { - let key = MetadataKey::from_bytes(k.as_bytes()).map_err(|e| format!("invalid gRPC metadata key {k:?}: {e}"))?; - let val = MetadataValue::try_from(v.as_str()) - .map_err(|e| format!("invalid gRPC metadata value for key {k:?}: {e}"))?; - map.insert(key, val); - } - Ok(map) -} - -/// Parses a comma-separated `key=value` header string into a [`HashMap`]. -/// -/// Whitespace around keys and values is trimmed. Empty segments (from -/// trailing commas or whitespace-only entries) are silently skipped. -/// Any non-empty segment that is missing the `=` separator, or has an -/// empty key, is treated as a configuration error so the application -/// fails fast rather than silently dropping user-supplied headers. -fn parse_otlp_headers(raw: Option<&str>) -> Result, Box> { - let mut out = HashMap::new(); - let Some(raw) = raw else { return Ok(out) }; - for entry in raw.split(',') { - let entry = entry.trim(); - if entry.is_empty() { - continue; - } - match entry.split_once('=') { - None => { - return Err(format!("malformed OTLP header entry (missing '=' separator): {entry:?}").into()); - }, - Some((key, _)) if key.trim().is_empty() => { - return Err(format!("malformed OTLP header entry (empty key): {entry:?}").into()); - }, - Some((key, value)) => { - out.insert(key.trim().to_owned(), value.trim().to_owned()); - }, - } - } - Ok(out) -} - -#[cfg(test)] -mod tests { - use super::parse_otlp_headers; - - #[test] - fn parse_otlp_headers_handles_empty_and_missing_input() { - assert!(parse_otlp_headers(None).unwrap().is_empty()); - assert!(parse_otlp_headers(Some("")).unwrap().is_empty()); - assert!(parse_otlp_headers(Some(" , , ")).unwrap().is_empty()); - } - - #[test] - fn parse_otlp_headers_parses_multiple_entries_and_trims_whitespace() { - let parsed = parse_otlp_headers(Some(" Authorization = Basic abc , X-Project=demo ")).unwrap(); - assert_eq!(parsed.get("Authorization"), Some(&"Basic abc".to_owned())); - assert_eq!(parsed.get("X-Project"), Some(&"demo".to_owned())); - assert_eq!(parsed.len(), 2); - } - - #[test] - fn parse_otlp_headers_rejects_entry_without_separator() { - assert!(parse_otlp_headers(Some("no-equals")).is_err()); - assert!(parse_otlp_headers(Some("good=value,no-equals")).is_err()); - } - - #[test] - fn parse_otlp_headers_rejects_entry_with_empty_key() { - assert!(parse_otlp_headers(Some("=missing-key")).is_err()); - } -} diff --git a/crates/contextforge-data-plane/src/main.rs b/crates/contextforge-data-plane/src/main.rs index 9af223e2..ff2a57da 100644 --- a/crates/contextforge-data-plane/src/main.rs +++ b/crates/contextforge-data-plane/src/main.rs @@ -1,13 +1,13 @@ -mod logging; mod runtime; #[cfg(feature = "test-plugins")] mod test_plugins; -use std::sync::Arc; +use std::{process::ExitCode, sync::Arc}; use clap::Parser; use contextforge_data_plane_cpex::CpexRuntimeRegistry; use contextforge_data_plane_lib::{Config, Gateway, RedisClient, RedisConfig, UserConfigStoreType}; +use contextforge_data_plane_observability::{LoggingConfig, emit_bootstrap_failure, init_observability}; use rmcp::transport::streamable_http_server::session::local::LocalSessionManager; use rustls::crypto; use tikv_jemallocator::Jemalloc; @@ -15,13 +15,67 @@ use tracing::info; #[global_allocator] static GLOBAL: Jemalloc = Jemalloc; -fn main() -> Result<(), Box> { +fn main() -> ExitCode { let provider = crypto::ring::default_provider(); _ = provider.install_default(); let config = Config::parse(); - let _guard = logging::init_tracing_logging(&config)?; + let logging_config = logging_config(&config); + let _guard = match init_observability(&logging_config) { + Ok(guard) => guard, + Err(error) => { + emit_bootstrap_failure(&logging_config, error.as_ref()); + return ExitCode::FAILURE; + }, + }; + + match run(config) { + Ok(()) => ExitCode::SUCCESS, + Err(error) => { + tracing::error!( + fatal = true, + component = "Bootstrap", + operation = "startup", + error = %error, + error_code = "CFDP-BOOTSTRAP", + root_cause = %error, + impact_scope = "service-wide", + retryable = false, + "service startup failed" + ); + ExitCode::FAILURE + }, + } +} + +fn logging_config(config: &Config) -> LoggingConfig { + LoggingConfig { + service_name: config.otlp_service_name.clone(), + version: env!("CARGO_PKG_VERSION").to_owned(), + environment: config.environment.clone(), + cluster_id: config.cluster_id.clone(), + log_name: config.log_name.clone(), + log_rotation: config.log_rotation.clone().unwrap_or_default(), + enable_open_telemetry: config.enable_open_telemetry.unwrap_or(false), + enable_otel_metrics: config.enable_otel_metrics.unwrap_or(false), + otlp_endpoint: config.otlp_endpoint.clone(), + otlp_metrics_endpoint: config.otlp_metrics_endpoint.clone(), + otlp_protocol: config.otlp_protocol.clone().unwrap_or_default(), + otlp_headers: config.otlp_headers.clone(), + } +} + +fn run(config: Config) -> Result<(), Box> { + let feature_flags = + [cfg!(feature = "plugins").then_some("plugins"), cfg!(feature = "test-plugins").then_some("test-plugins")] + .into_iter() + .flatten() + .collect::>() + .join(","); + let feature_flags = if feature_flags.is_empty() { "none" } else { &feature_flags }; info!( + component = "Bootstrap", + operation = "startup", address = ?config.address, tls_address = ?config.tls_address, redis_mode = ?config.redis_mode, @@ -31,6 +85,12 @@ fn main() -> Result<(), Box> { otel_metrics_enabled = config.enable_otel_metrics.unwrap_or(false), single_runtime = config.single_runtime.unwrap_or(true), configured_cpus = ?config.number_of_cpus, + git_commit_sha = option_env!("GIT_COMMIT_SHA").unwrap_or("unknown"), + build_timestamp = option_env!("BUILD_TIMESTAMP").unwrap_or("unknown"), + config_profile = config.environment.as_deref().unwrap_or("unknown"), + feature_flags, + db_version = "not_applicable", + external_dependencies_reachable = "not_checked", "starting contextforge-data-plane" ); diff --git a/crates/contextforge-data-plane/src/runtime.rs b/crates/contextforge-data-plane/src/runtime.rs index 71e87b55..73fb0c4e 100644 --- a/crates/contextforge-data-plane/src/runtime.rs +++ b/crates/contextforge-data-plane/src/runtime.rs @@ -98,13 +98,23 @@ impl Runtime { for i in 1..self.number_of_threads { match Self::spawn_gateway_thread(format!("{}{i}", self.thread_name), gateway.clone(), None, None) { Ok(handle) => handles.push(handle), - Err(error) => warn!("Thread terminated at start with {error:?}"), + Err(error) => warn!( + component = "Runtime", + operation = "spawn_gateway_thread", + error = %error, + "gateway thread failed to start" + ), } } for handle in handles { let res = handle.join(); - info!("Thread terminated with {res:?}"); + info!( + component = "Runtime", + operation = "join_gateway_thread", + succeeded = res.is_ok(), + "gateway thread terminated" + ); } Ok(()) } @@ -122,7 +132,12 @@ impl Runtime { let runtime = match builder.build_local(LocalOptions::default()) { Ok(runtime) => runtime, Err(error) => { - warn!("Can't build thread {error:?}"); + warn!( + component = "Runtime", + operation = "build_gateway_thread", + error = %error, + "gateway runtime could not be built" + ); return Err::<(), contextforge_data_plane_lib::Error>(error.into()); }, }; @@ -154,16 +169,25 @@ impl Runtime { }; match cpex_runtime.initialize().await { Ok(Some(handle)) => { - debug!("CPEX Plugins initialization successful"); + debug!(component = "Plugins", operation = "initialize", "runtime plugins initialized"); Ok(Some(handle)) }, Ok(None) => { - debug!("CPEX Plugins initialization skipped"); + debug!(component = "Plugins", operation = "initialize", "runtime plugin initialization skipped"); Ok(None) }, - Err(e) => { - error!("CPEX Plugins initialization failed {e:?}"); - Err(e) + Err(error) => { + error!( + component = "Plugins", + operation = "initialize", + error_code = "CFDP-PLUGIN-INIT", + root_cause = %error, + impact_scope = "service-startup", + retryable = false, + error = ?error, + "runtime plugin initialization failed" + ); + Err(error) }, } } @@ -171,9 +195,19 @@ impl Runtime { async fn run_gateway(gateway: Gateway) -> contextforge_data_plane_lib::Result<()> { let res = gateway.run_gateway().await; if res.is_ok() { - debug!("Gateway process terminated"); + debug!(component = "Gateway", operation = "run", "gateway process terminated"); } else { - error!("Gateway process terminated {res:?}"); + let error = res.as_ref().expect_err("checked error result"); + error!( + component = "Gateway", + operation = "run", + error_code = "CFDP-GATEWAY-TERMINATED", + root_cause = %error, + impact_scope = "service-wide", + retryable = true, + error = ?error, + "gateway process terminated unexpectedly" + ); } Ok(()) } diff --git a/crates/contextforge-load-test/src/main.rs b/crates/contextforge-load-test/src/main.rs index 3e8b916d..8a42d11b 100644 --- a/crates/contextforge-load-test/src/main.rs +++ b/crates/contextforge-load-test/src/main.rs @@ -88,7 +88,7 @@ async fn counter_call(user: &mut GooseUser) -> TransactionResult { let session = user.get_session_data_unchecked::().clone(); if session.mcp_session_id.is_none() { - user.log_debug("counter_call: No Session id ", None, None, None)?; + user.log_debug("counter call has no session", None, None, None)?; return Err(Box::new(TransactionError::Custom("no session id".to_owned()))); } @@ -103,7 +103,7 @@ async fn counter_call(user: &mut GooseUser) -> TransactionResult { let payload = r.text().await?; if status == http::StatusCode::OK { if !payload.contains("\"isError\":false") && !payload.contains(&format!("\"text\":\"{}\"", i + 1)) { - let _ = user.log_debug("Counter has a problem ", None, None, Some(&payload)); + let _ = user.log_debug("counter response validation failed", None, None, None); return Err(Box::new(TransactionError::Custom("counting problem".to_owned()))); } } else { @@ -144,7 +144,7 @@ async fn initialize_session(user: &mut GooseUser) -> TransactionResult { let session = user.get_session_data_unchecked::(); let mut session = session.clone(); if session.mcp_session_id.is_some() { - _ = user.log_debug("intialize has session id ", None, None, None); + _ = user.log_debug("initialize called with an existing session", None, None, None); return Err(Box::new(TransactionError::Custom("initialize has session id ".to_owned()))); } let init_request = {