diff --git a/crates/aisix-core/src/models/mod.rs b/crates/aisix-core/src/models/mod.rs index f8754339..f09022d2 100644 --- a/crates/aisix-core/src/models/mod.rs +++ b/crates/aisix-core/src/models/mod.rs @@ -39,8 +39,8 @@ pub use model::{ Adapter, BackgroundModelCheck, CooldownConfig, Model, DEFAULT_COOLDOWN_TRIGGER_STATUSES, }; pub use observability_exporter::{ - AliyunSlsConfig, ExporterKind, ObjectStoreCompression, ObjectStoreConfig, ObjectStoreProvider, - ObservabilityExporter, OtlpHttpConfig, SlsContentMode, + AliyunSlsConfig, DatadogConfig, ExporterKind, ObjectStoreCompression, ObjectStoreConfig, + ObjectStoreProvider, ObservabilityExporter, OtlpHttpConfig, SlsContentMode, }; pub use provider_key::{ ParamConstraints, ProviderKey, RequestOverrides, ResponseOverrides, StreamDoneMarker, diff --git a/crates/aisix-core/src/models/observability_exporter.rs b/crates/aisix-core/src/models/observability_exporter.rs index 6cb8b7de..849ab149 100644 --- a/crates/aisix-core/src/models/observability_exporter.rs +++ b/crates/aisix-core/src/models/observability_exporter.rs @@ -35,9 +35,10 @@ use serde::{Deserialize, Serialize}; use crate::resource::Resource; /// Discriminated union of exporter back-ends. Ships `otlp_http`, -/// `aliyun_sls`, and `object_store` (S3 / GCS / Azure Blob, one variant); -/// Datadog / … land in follow-ups, each as a new variant whose serde tag -/// matches the wire-side `kind` discriminator. +/// `aliyun_sls`, `object_store` (S3 / GCS / Azure Blob, one variant), and +/// `datadog` (Datadog native Logs intake); further targets land in +/// follow-ups, each as a new variant whose serde tag matches the wire-side +/// `kind` discriminator. /// /// `tag = "kind"` puts the variant tag inline with the inner struct's /// fields — same shape as `GuardrailKind` so the kine wire stays @@ -48,6 +49,7 @@ pub enum ExporterKind { OtlpHttp(OtlpHttpConfig), AliyunSls(AliyunSlsConfig), ObjectStore(ObjectStoreConfig), + Datadog(DatadogConfig), } #[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema, PartialEq, Eq)] @@ -124,10 +126,11 @@ pub struct AliyunSlsConfig { pub content_max_bytes: u32, } -/// Content-capture mode for an SLS exporter. Defaults to the -/// privacy-preserving `metadata_only`. +/// Content-capture mode for an SLS / Datadog exporter. Defaults to the +/// privacy-preserving `metadata_only`. (`Hash` so a Datadog exporter's +/// fingerprint can cover its content config; see `fingerprint_datadog`.) #[derive( - Debug, Clone, Copy, Default, Serialize, Deserialize, schemars::JsonSchema, PartialEq, Eq, + Debug, Clone, Copy, Default, Serialize, Deserialize, schemars::JsonSchema, PartialEq, Eq, Hash, )] #[serde(rename_all = "snake_case")] pub enum SlsContentMode { @@ -143,6 +146,99 @@ const fn default_content_max_bytes() -> u32 { 128 * 1024 } +/// Datadog native **Logs HTTP intake** target. Each request event becomes one +/// Datadog log object, gzip-compressed and POSTed to +/// `https://http-intake.logs./api/v2/logs`. Like `aliyun_sls` (and +/// unlike `otlp_http`), the Datadog API key is **never** part of this config: +/// it would otherwise sit in plaintext on the kine path, and a Datadog API key +/// grants broad org access. Instead the config carries a [`credential_ref`] +/// pointer that the customer-side DP resolves to the real key locally (env / +/// mounted secret); API7's control plane stores only the reference. This +/// deliberately diverges from issue #688's `api_key: SecretRef` draft to stay +/// consistent with SLS / object_store and avoid the #692 credential-encryption +/// dependency. +/// +/// [`credential_ref`]: DatadogConfig::credential_ref +#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema, PartialEq, Eq)] +#[serde(deny_unknown_fields)] +pub struct DatadogConfig { + /// Datadog site, validated against the allow-list in the loader schema + /// (`datadoghq.com`, `us3`/`us5`/`ap1`/`ap2` regions, `datadoghq.eu`, + /// `ddog-gov.com`). The intake host is `http-intake.logs.`. A + /// scheme-qualified loopback host (the e2e's `http://mock-datadog:*`) is + /// admitted only for vetted local mocks, never to redirect real traffic. + pub site: String, + + /// Opaque pointer to the Datadog API key, resolved locally by the DP at + /// delivery time. The plaintext key MUST NOT live in etcd/kine — the + /// control plane stores only this reference, never the key itself. The DP + /// reads `DD_CRED__API_KEY` from its own environment, where `` + /// upper-cases the reference with non-alphanumerics folded to `_`. + pub credential_ref: String, + + /// Datadog `service` reserved attribute — the service name every log from + /// this exporter is tagged with in Datadog's Log Explorer. + pub service: String, + + /// Datadog `ddsource` reserved attribute — the integration/source name. + /// Defaults to `aisix-ai-gateway`. + #[serde(default = "default_ddsource")] + pub ddsource: String, + + /// Operator-defined tags rendered into Datadog's comma-joined `ddtags` + /// reserved attribute (e.g. `["team:platform", "tier:prod"]` → + /// `team:platform,tier:prod`). Empty by default. + #[serde(default)] + pub tags: Vec, + + /// Whether captured request/response content is delivered to Datadog. + /// `metadata_only` (default) ships only operational metadata — never a + /// prompt or response. `full` additionally captures the request prompt and + /// the assembled response, each truncated to [`content_max_bytes`]. + /// Enabling `full` writes end-user prompt / response text into the + /// customer's Datadog org, so the dashboard must surface the privacy + /// implication when an operator turns it on. Reuses [`SlsContentMode`] — + /// the codebase's existing `metadata_only | full` model — so the shared + /// content-capture plumbing (`content_record` / `content_capture_cap`) + /// stays single-sourced. + /// + /// [`content_max_bytes`]: DatadogConfig::content_max_bytes + #[serde(default)] + pub content_mode: SlsContentMode, + + /// Per-field byte cap for captured content under `content_mode = full`. + /// The prompt and the response are each truncated to this many bytes + /// (UTF-8-boundary safe), and the log carries a `content_truncated` marker + /// when either was cut. Ignored under `metadata_only`. Defaults to 128 KiB. + /// + /// This bounds each field *independently*: a single log carries BOTH the + /// prompt and the response plus metadata, so the encoded log can reach + /// ~2× this cap. Datadog rejects any single log over 1 MB and any request + /// over 5 MB / 1000 logs; byte-aware per-log/per-request splitting to those + /// limits is not yet enforced (tracked in api7/ai-gateway#556) — until it + /// lands, a large cap on a busy `full` exporter risks Datadog rejecting an + /// oversized batch (a `Permanent` delivery error surfaced via `last_error`). + /// The 128 KiB default keeps a log well under the per-log limit. + /// + /// `0` is rejected — `full` with a zero cap captures nothing, which is a + /// misconfiguration. The `range(min = 1, max = …)` keeps the generated JSON + /// schema in step with the runtime validator so the CP and DP agree on the + /// bounds. + #[serde(default = "default_dd_content_max_bytes")] + #[schemars(range(min = 1, max = 1_048_576))] + pub content_max_bytes: u32, +} + +/// Default Datadog `ddsource` reserved attribute. +fn default_ddsource() -> String { + "aisix-ai-gateway".to_string() +} + +/// Default per-field content cap for a Datadog exporter: 128 KiB. +const fn default_dd_content_max_bytes() -> u32 { + 128 * 1024 +} + /// Object-storage sink — ONE config covering S3 / GCS / Azure Blob (and /// S3-compatible MinIO / Cloudflare R2 via `endpoint`) behind a single /// backend, so the sink is written once rather than per provider. Batched @@ -555,4 +651,89 @@ mod tests { ); } } + + const VALID_DATADOG: &str = r#"{ + "name": "datadog-prod", + "enabled": true, + "kind": "datadog", + "site": "datadoghq.com", + "credential_ref": "datadog-prod", + "service": "ai-gateway" + }"#; + + #[test] + fn deserialises_datadog() { + let e: ObservabilityExporter = serde_json::from_str(VALID_DATADOG).unwrap(); + assert_eq!(e.name, "datadog-prod"); + assert!(e.enabled); + match &e.kind { + ExporterKind::Datadog(c) => { + assert_eq!(c.site, "datadoghq.com"); + assert_eq!(c.credential_ref, "datadog-prod"); + assert_eq!(c.service, "ai-gateway"); + // ddsource defaults; tags empty. + assert_eq!(c.ddsource, "aisix-ai-gateway"); + assert!(c.tags.is_empty()); + // Content capture is off by default (privacy-preserving). + assert_eq!(c.content_mode, SlsContentMode::MetadataOnly); + assert_eq!(c.content_max_bytes, 128 * 1024); + } + other => panic!("expected datadog, got {other:?}"), + } + } + + #[test] + fn datadog_opts_into_full_content_capture_and_tags() { + let json = r#"{ + "name": "datadog-content", + "kind": "datadog", + "site": "datadoghq.eu", + "credential_ref": "r", + "service": "ai-gateway", + "ddsource": "custom-source", + "tags": ["team:platform", "tier:prod"], + "content_mode": "full", + "content_max_bytes": 4096 + }"#; + let e: ObservabilityExporter = serde_json::from_str(json).unwrap(); + match &e.kind { + ExporterKind::Datadog(c) => { + assert_eq!(c.site, "datadoghq.eu"); + assert_eq!(c.ddsource, "custom-source"); + assert_eq!(c.tags, vec!["team:platform", "tier:prod"]); + assert_eq!(c.content_mode, SlsContentMode::Full); + assert_eq!(c.content_max_bytes, 4096); + } + other => panic!("expected datadog, got {other:?}"), + } + } + + #[test] + fn datadog_round_trips_flat() { + let e: ObservabilityExporter = serde_json::from_str(VALID_DATADOG).unwrap(); + let v = serde_json::to_value(&e).unwrap(); + // Flat wire — kind tag + fields at the top level, never nested. + assert_eq!(v["kind"], "datadog"); + assert_eq!(v["site"], "datadoghq.com"); + assert_eq!(v["credential_ref"], "datadog-prod"); + assert_eq!(v["service"], "ai-gateway"); + assert!(v.get("datadog").is_none(), "kind block must not nest"); + } + + #[test] + fn rejects_plaintext_api_key_in_datadog_config() { + // The Datadog API key must NEVER be a config field — only a + // `credential_ref`. `deny_unknown_fields` on the inner config rejects + // any attempt to smuggle a plaintext key onto the kine path. + for key in ["api_key", "apikey", "dd_api_key", "key"] { + let json = format!( + r#"{{"name":"x","kind":"datadog","site":"datadoghq.com","credential_ref":"r","service":"s","{key}":"DDSECRET"}}"# + ); + let r: Result = serde_json::from_str(&json); + assert!( + r.is_err(), + "plaintext credential field `{key}` must be rejected" + ); + } + } } diff --git a/crates/aisix-core/src/models/schema.rs b/crates/aisix-core/src/models/schema.rs index 811014a4..262cfea9 100644 --- a/crates/aisix-core/src/models/schema.rs +++ b/crates/aisix-core/src/models/schema.rs @@ -591,8 +591,7 @@ fn observability_exporter_schema() -> Value { // considers THIS object's `properties` (not those inside `allOf`/`then`), // so every kind's fields are listed at the top level as the union; // per-kind required-fields and the endpoint pattern live in the - // `if`/`then` branches. Phase 2 adds `datadog_logs` / `s3_ndjson` the - // same way. + // `if`/`then` branches. Further kinds (`s3_ndjson`, …) land the same way. json!({ "$schema": "https://json-schema.org/draft/2020-12/schema", "type": "object", @@ -601,7 +600,7 @@ fn observability_exporter_schema() -> Value { "properties": { "name": { "type": "string", "minLength": 1, "maxLength": 120 }, "enabled": { "type": "boolean" }, - "kind": { "type": "string", "enum": ["otlp_http", "aliyun_sls", "object_store"] }, + "kind": { "type": "string", "enum": ["otlp_http", "aliyun_sls", "object_store", "datadog"] }, // Shared field; the per-kind pattern is enforced in the branches. "endpoint": { "type": "string" }, // otlp_http field. @@ -615,10 +614,14 @@ fn observability_exporter_schema() -> Value { "project": { "type": "string", "minLength": 1 }, "logstore": { "type": "string", "minLength": 1 }, "credential_ref": { "type": "string", "minLength": 1 }, - // Content capture (opt-in). `full` writes captured prompt / - // response into the logstore; `content_max_bytes` truncates each. + // Content capture (opt-in), shared by aliyun_sls + datadog. `full` + // writes captured prompt / response to the sink; `content_max_bytes` + // truncates each FIELD. It is not a per-log bound — a datadog log + // carries both prompt and response, so byte-aware splitting to + // Datadog's 1 MB-per-log / 5 MB-per-request intake limits is tracked + // separately (api7/ai-gateway#556), not enforced by this cap. "content_mode": { "type": "string", "enum": ["metadata_only", "full"] }, - "content_max_bytes": { "type": "integer", "minimum": 1 }, + "content_max_bytes": { "type": "integer", "minimum": 1, "maximum": 1048576 }, // object_store fields (S3 / GCS / Azure Blob, one variant). Cloud // credentials are NEVER here — only the shared `credential_ref`. "provider": { "type": "string", "enum": ["s3", "gcs", "azure_blob"] }, @@ -627,7 +630,17 @@ fn observability_exporter_schema() -> Value { "region": { "type": "string", "minLength": 1 }, "compression": { "type": "string", "enum": ["gzip", "none"] }, // object_store auth mode: how the DP reaches the bucket. - "auth_mode": { "type": "string", "enum": ["credential_ref", "cloud_identity"] } + "auth_mode": { "type": "string", "enum": ["credential_ref", "cloud_identity"] }, + // datadog fields. The Datadog API key is NEVER here — only the + // shared `credential_ref` the DP resolves locally. `site` is + // constrained to the allow-list in the per-kind branch below. + "site": { "type": "string", "minLength": 1 }, + "service": { "type": "string", "minLength": 1 }, + "ddsource": { "type": "string", "minLength": 1 }, + "tags": { + "type": "array", + "items": { "type": "string" } + } }, "allOf": [ { @@ -696,6 +709,29 @@ fn observability_exporter_schema() -> Value { } ] } + }, + { + "if": { "properties": { "kind": { "const": "datadog" } } }, + "then": { + "required": ["site", "credential_ref", "service"], + "properties": { + // The Datadog site, constrained to the supported intake + // sites; the sink posts to `https://http-intake.logs.`. + // Loopback bypass for e2e: a bare mock-datadog / 127.0.0.1 + // / localhost host, OPTIONALLY with a `:port`, which the + // sink posts to over http:// directly (a local mock intake + // needs no TLS) — never a way to redirect real traffic to + // an arbitrary host. The `:port` is allowed ONLY on the + // loopback hosts (the e2e harness binds a free port); the + // real sites match exactly, no port. Mirrors the + // aliyun_sls / object_store loopback patterns — the prior + // exact-enum rejected the harness's free-port host while + // the sink's `is_loopback_site` accepted it (#548). + "site": { + "pattern": "^(datadoghq\\.com|us3\\.datadoghq\\.com|us5\\.datadoghq\\.com|datadoghq\\.eu|ap1\\.datadoghq\\.com|ap2\\.datadoghq\\.com|ddog-gov\\.com)$|^(mock-datadog|127\\.0\\.0\\.1|localhost)(:[0-9]+)?$" + } + } + } } ] }) @@ -1598,6 +1634,154 @@ mod tests { assert!(validate_observability_exporter(&v).is_err()); } + #[test] + fn exporter_datadog_happy_path() { + let v = json!({ + "name": "datadog-prod", + "kind": "datadog", + "site": "datadoghq.com", + "credential_ref": "datadog-prod", + "service": "ai-gateway", + "ddsource": "aisix-ai-gateway", + "tags": ["team:platform", "tier:prod"] + }); + validate_observability_exporter(&v).unwrap(); + } + + #[test] + fn exporter_datadog_accepts_every_allow_list_site() { + for site in [ + "datadoghq.com", + "us3.datadoghq.com", + "us5.datadoghq.com", + "datadoghq.eu", + "ap1.datadoghq.com", + "ap2.datadoghq.com", + "ddog-gov.com", + ] { + let v = json!({ + "name": "x", + "kind": "datadog", + "site": site, + "credential_ref": "r", + "service": "s" + }); + validate_observability_exporter(&v) + .unwrap_or_else(|e| panic!("site {site:?} must validate: {e:?}")); + } + } + + #[test] + fn exporter_datadog_rejects_non_allow_list_site() { + // A plausible-looking but unsupported / spoofed site must be rejected — + // no exfil to an arbitrary `http-intake.logs.`. + for bad in [ + "evil.datadoghq.com.attacker.test", + "datadoghq.org", + "us9.datadoghq.com", + "datadog.com", + "datadoghq.com:443", // a port is NOT allowed on a real site + "", + ] { + let v = json!({ + "name": "x", + "kind": "datadog", + "site": bad, + "credential_ref": "r", + "service": "s" + }); + assert!( + validate_observability_exporter(&v).is_err(), + "site {bad:?} must be rejected by the allow-list" + ); + } + } + + #[test] + fn exporter_datadog_allows_loopback_mock_site() { + // The e2e points the DP at a local mock Datadog intake — bare host OR + // host:port. The harness binds a FREE port, so `:port` must validate + // (the prior exact-enum rejected it while the sink accepted it — #548). + for site in ["mock-datadog", "127.0.0.1:54321", "localhost:8080"] { + let v = json!({ + "name": "datadog-e2e", + "kind": "datadog", + "site": site, + "credential_ref": "mock", + "service": "ai-gateway" + }); + validate_observability_exporter(&v) + .unwrap_or_else(|e| panic!("loopback site {site:?} must validate: {e:?}")); + } + } + + #[test] + fn exporter_datadog_requires_site_credential_service() { + for missing in ["site", "credential_ref", "service"] { + let mut v = json!({ + "name": "x", + "kind": "datadog", + "site": "datadoghq.com", + "credential_ref": "r", + "service": "s" + }); + v.as_object_mut().unwrap().remove(missing); + assert!( + validate_observability_exporter(&v).is_err(), + "missing `{missing}` must be rejected" + ); + } + } + + #[test] + fn exporter_datadog_rejects_plaintext_api_key() { + // No API-key field is allowed at the schema layer either — + // `additionalProperties: false` rejects it before serde runs. + let v = json!({ + "name": "x", + "kind": "datadog", + "site": "datadoghq.com", + "credential_ref": "r", + "service": "s", + "api_key": "DDSECRET" + }); + assert!(validate_observability_exporter(&v).is_err()); + } + + #[test] + fn exporter_datadog_content_capture_fields() { + let base = |extra: serde_json::Value| { + let mut v = json!({ + "name": "x", + "kind": "datadog", + "site": "datadoghq.com", + "credential_ref": "r", + "service": "s" + }); + let obj = v.as_object_mut().unwrap(); + for (k, val) in extra.as_object().unwrap() { + obj.insert(k.clone(), val.clone()); + } + v + }; + // Opt-in content capture validates. + validate_observability_exporter(&base( + json!({ "content_mode": "full", "content_max_bytes": 4096 }), + )) + .unwrap(); + // Unknown content_mode is rejected. + assert!( + validate_observability_exporter(&base(json!({ "content_mode": "verbose" }))).is_err() + ); + // content_max_bytes must be a positive integer (min 1). + assert!(validate_observability_exporter(&base(json!({ "content_max_bytes": 0 }))).is_err()); + // content_max_bytes is capped at 1 MiB (Datadog per-log limit). + assert!( + validate_observability_exporter(&base(json!({ "content_max_bytes": 1_048_577 }))) + .is_err() + ); + } + // ---- rate_limit_policy schema tests ---- #[test] diff --git a/crates/aisix-obs/src/lib.rs b/crates/aisix-obs/src/lib.rs index e4af703b..75d87b6d 100644 --- a/crates/aisix-obs/src/lib.rs +++ b/crates/aisix-obs/src/lib.rs @@ -33,10 +33,10 @@ pub use metrics::{ pub use otlp::{install_otlp_tracer, shutdown_otlp, OtlpError, OtlpHandle}; pub use otlp_http_sink::{content_capture_cap, OtlpHttpFanOut, OtlpSink}; pub use sink::{ - AliyunSlsSink, BatchUnit, CapturedContent, ChannelKey, EventBatch, ExporterPipelines, - IdempotencyMarker, IdempotencyScheme, ObservabilitySink, OrderingScope, PipelineConfig, - SinkAck, SinkCapabilities, SinkContent, SinkError, SinkHandle, SinkHealth, SinkPipeline, - SinkRecord, SinkResult, SinkStatsSnapshot, SCHEMA_VERSION, + AliyunSlsSink, BatchUnit, CapturedContent, ChannelKey, DatadogSink, EventBatch, + ExporterPipelines, IdempotencyMarker, IdempotencyScheme, ObservabilitySink, OrderingScope, + PipelineConfig, SinkAck, SinkCapabilities, SinkContent, SinkError, SinkHandle, SinkHealth, + SinkPipeline, SinkRecord, SinkResult, SinkStatsSnapshot, SCHEMA_VERSION, }; pub use usage::{UsageEvent, UsageSink}; diff --git a/crates/aisix-obs/src/otlp_http_sink.rs b/crates/aisix-obs/src/otlp_http_sink.rs index d9a6b749..8b235935 100644 --- a/crates/aisix-obs/src/otlp_http_sink.rs +++ b/crates/aisix-obs/src/otlp_http_sink.rs @@ -33,17 +33,17 @@ use std::sync::Arc; use std::time::Duration; use aisix_core::models::{ - AliyunSlsConfig, ExporterKind, ObjectStoreConfig, ObservabilityExporter, OtlpHttpConfig, - SlsContentMode, + AliyunSlsConfig, DatadogConfig, ExporterKind, ObjectStoreConfig, ObservabilityExporter, + OtlpHttpConfig, SlsContentMode, }; use async_trait::async_trait; use serde_json::{json, Value}; use crate::sink::{ - build_object_store_sink, resolve_sls_credential, AliyunSlsSink, BatchUnit, CapturedContent, - EventBatch, ExporterPipelines, IdempotencyMarker, IdempotencyScheme, ObservabilitySink, - OrderingScope, PipelineConfig, SinkAck, SinkCapabilities, SinkContent, SinkError, SinkHealth, - SinkRecord, SinkResult, + build_object_store_sink, resolve_datadog_credential, resolve_sls_credential, AliyunSlsSink, + BatchUnit, CapturedContent, DatadogSink, EventBatch, ExporterPipelines, IdempotencyMarker, + IdempotencyScheme, ObservabilitySink, OrderingScope, PipelineConfig, SinkAck, SinkCapabilities, + SinkContent, SinkError, SinkHealth, SinkRecord, SinkResult, }; use crate::usage::UsageEvent; @@ -116,9 +116,10 @@ impl OtlpHttpFanOut { /// /// `content` is the request's captured prompt/response, or `None` when the /// handler captured none (the default). It is attached ONLY to an - /// `aliyun_sls` exporter whose `content_mode = full`; every other exporter - /// — and the CP telemetry path, which is not in this loop — receives the - /// shared metadata-only record, so prompt/response can never leak there. + /// `aliyun_sls` or `datadog` exporter whose `content_mode = full`; every + /// other exporter — and the CP telemetry path, which is not in this loop — + /// receives the shared metadata-only record, so prompt/response can never + /// leak there. pub fn fan_out<'a, I>( &self, event: &UsageEvent, @@ -194,6 +195,31 @@ impl OtlpHttpFanOut { build_object_store_sink(name, &cfg) }) } + ExporterKind::Datadog(cfg) => { + let fingerprint = fingerprint_datadog(cfg); + let name = exp.name.clone(); + let cfg = cfg.clone(); + self.inner + .exporters + .get_or_create(&exp.name, fingerprint, move || { + // Resolve the Datadog API key from the DP's local + // env at build time (the key never rode the kine + // path). Missing key → empty → Datadog 403 surfaces + // as a delivery-health auth error, not a silent drop + // — mirroring the SLS path. + let api_key = + resolve_datadog_credential(&cfg.credential_ref).unwrap_or_default(); + Arc::new(DatadogSink::new( + name, + &cfg.site, + api_key, + &cfg.ddsource, + &cfg.tags, + &cfg.service, + client, + )) as Arc + }) + } }; // A content-bearing record for an SLS exporter that opted into @@ -258,7 +284,7 @@ fn fingerprint_sls(cfg: &AliyunSlsConfig) -> u64 { /// The content-bearing [`SinkRecord`] for one exporter, or `None` to fall back /// to the shared metadata-only record. /// -/// Content is attached ONLY to an `aliyun_sls` exporter whose +/// Content is attached ONLY to an `aliyun_sls` OR `datadog` exporter whose /// `content_mode = full`, and ONLY when the handler captured content — every /// other exporter (and the CP telemetry path, which never enters the fan-out) /// gets metadata only. The captured prompt/response are truncated to the @@ -269,18 +295,19 @@ fn content_record( event: &UsageEvent, content: Option<&CapturedContent>, ) -> Option> { - let ExporterKind::AliyunSls(cfg) = kind else { - return None; + // The exporters that opt into full content capture share the + // `SlsContentMode` model; pull each one's (mode, cap). Any other kind + // never carries content, so prompt/response can't leak into it. + let (mode, max_bytes) = match kind { + ExporterKind::AliyunSls(cfg) => (cfg.content_mode, cfg.content_max_bytes), + ExporterKind::Datadog(cfg) => (cfg.content_mode, cfg.content_max_bytes), + _ => return None, }; - if cfg.content_mode != SlsContentMode::Full { + if mode != SlsContentMode::Full { return None; } let captured = content?; - let mut sc = SinkContent::capture( - &captured.prompt, - &captured.response, - cfg.content_max_bytes as usize, - ); + let mut sc = SinkContent::capture(&captured.prompt, &captured.response, max_bytes as usize); sc.truncated = sc.truncated || captured.truncated; Some(Arc::new( SinkRecord::metadata_only(event.clone()).with_content(sc), @@ -305,6 +332,9 @@ pub fn content_capture_cap<'a>( ExporterKind::AliyunSls(cfg) if cfg.content_mode == SlsContentMode::Full => { Some(cfg.content_max_bytes) } + ExporterKind::Datadog(cfg) if cfg.content_mode == SlsContentMode::Full => { + Some(cfg.content_max_bytes) + } _ => None, }) .max() @@ -328,6 +358,24 @@ fn fingerprint_object_store(cfg: &ObjectStoreConfig) -> u64 { hasher.finish() } +/// Hash a `datadog` exporter's delivery-relevant config. Covers only +/// kine-visible fields (site / credential_ref / service / ddsource / tags / +/// content config), never the resolved API key — rotating the secret under the +/// *same* reference therefore takes effect on the next DP restart, not live. A +/// ref change (or any other field change) rebuilds the pipeline. +fn fingerprint_datadog(cfg: &DatadogConfig) -> u64 { + use std::hash::{Hash, Hasher}; + let mut hasher = std::collections::hash_map::DefaultHasher::new(); + cfg.site.hash(&mut hasher); + cfg.credential_ref.hash(&mut hasher); + cfg.service.hash(&mut hasher); + cfg.ddsource.hash(&mut hasher); + cfg.tags.hash(&mut hasher); + cfg.content_mode.hash(&mut hasher); + cfg.content_max_bytes.hash(&mut hasher); + hasher.finish() +} + /// An [`ObservabilitySink`] over the OTLP/HTTP-JSON traces protocol — the /// same wire shape as [`OtlpHttpFanOut`], but driven by the shared /// [`crate::sink::SinkPipeline`] (batched, retried, backpressured) rather @@ -769,6 +817,18 @@ mod tests { }) } + fn datadog_kind(content_mode: SlsContentMode, max_bytes: u32) -> ExporterKind { + ExporterKind::Datadog(DatadogConfig { + site: "datadoghq.com".into(), + credential_ref: "r".into(), + service: "ai-gateway".into(), + ddsource: "aisix-ai-gateway".into(), + tags: vec![], + content_mode, + content_max_bytes: max_bytes, + }) + } + #[test] fn content_record_targets_only_full_capture_sls() { let event = sample_event(); @@ -842,6 +902,24 @@ mod tests { rec.content.as_ref().unwrap().truncated, "source truncation must propagate" ); + + // datadog behaves identically to sls: metadata_only → no content, + // full + captured content → a content-bearing record (same shared + // `SlsContentMode` plumbing). + let dd_meta = datadog_kind(SlsContentMode::MetadataOnly, 1024); + assert!(content_record(&dd_meta, &event, Some(&captured)).is_none()); + let dd_full = datadog_kind(SlsContentMode::Full, 1024); + assert!(content_record(&dd_full, &event, None).is_none()); + let rec = content_record(&dd_full, &event, Some(&captured)) + .expect("full-capture datadog with content yields a content record"); + let c = rec.content.as_ref().expect("content attached"); + assert_eq!(c.prompt, "the prompt"); + assert_eq!(c.response, "the response"); + // Per-exporter cap truncates a datadog record too. + let rec = + content_record(&datadog_kind(SlsContentMode::Full, 16), &event, Some(&big)).unwrap(); + assert_eq!(rec.content.as_ref().unwrap().prompt.len(), 16); + assert!(rec.content.as_ref().unwrap().truncated); } #[test] @@ -877,6 +955,27 @@ mod tests { // A disabled full-capture exporter is ignored. let disabled = sls("a", false, "full", 4096); assert_eq!(content_capture_cap([&disabled]), None); + + // A full-capture datadog exporter counts toward the cap too, and the + // max is taken across both kinds. + fn datadog(name: &str, enabled: bool, mode: &str, max: u32) -> ObservabilityExporter { + serde_json::from_value(serde_json::json!({ + "name": name, + "enabled": enabled, + "kind": "datadog", + "site": "datadoghq.com", + "credential_ref": "r", + "service": "ai-gateway", + "content_mode": mode, + "content_max_bytes": max, + })) + .unwrap() + } + let dd_full = datadog("dd", true, "full", 16384); + assert_eq!(content_capture_cap([&dd_full]), Some(16384)); + assert_eq!(content_capture_cap([&full, &dd_full]), Some(16384)); + let dd_meta = datadog("dd", true, "metadata_only", 4096); + assert_eq!(content_capture_cap([&dd_meta]), None); } #[test] diff --git a/crates/aisix-obs/src/sink/datadog.rs b/crates/aisix-obs/src/sink/datadog.rs new file mode 100644 index 00000000..04bb9993 --- /dev/null +++ b/crates/aisix-obs/src/sink/datadog.rs @@ -0,0 +1,737 @@ +//! Datadog native **Logs HTTP intake** sink — the `http_batch` family's +//! Datadog vendor (ai-gateway#688). +//! +//! A batch becomes one Datadog **logs intake** request: every [`SinkRecord`] +//! maps to one JSON log object (the canonical usage metadata flattened into +//! sibling fields under OTel GenAI semconv names, plus the Datadog reserved +//! attributes `ddsource` / `ddtags` / `service` / `message`, plus opt-in +//! captured prompt/response), the JSON array is gzip-compressed, and POSTed +//! to `https://http-intake.logs./api/v2/logs` over the crate's shared +//! rustls client. +//! +//! Wire details are taken from Datadog's official "Send logs" HTTP API +//! reference (): +//! the endpoint, the `DD-API-KEY` header, the `Content-Encoding: gzip` +//! support, the JSON-array body, and the intake limits (1000 logs and 5 MB +//! uncompressed per request; 1 MB per log). GenAI attribute names follow the +//! OpenTelemetry GenAI semantic conventions, matching the OTLP span builder +//! the other sinks emit, so a Datadog log and an OTLP span carry the same +//! `gen_ai.*` keys. + +use std::io::Write as _; + +use async_trait::async_trait; +use http::header::{CONTENT_ENCODING, CONTENT_TYPE}; +use http::{HeaderMap, HeaderName, HeaderValue}; +use serde::Deserialize; +use serde_json::{json, Map, Value}; + +use super::{ + BatchUnit, EventBatch, IdempotencyMarker, IdempotencyScheme, ObservabilitySink, OrderingScope, + SinkAck, SinkCapabilities, SinkError, SinkHealth, SinkRecord, SinkResult, +}; + +/// `DD-API-KEY`: Datadog's API-key header. The resolved key rides here and +/// nowhere else (never in the body, the URL, logs, or error text). +const DD_API_KEY: HeaderName = HeaderName::from_static("dd-api-key"); + +/// Cap on a masked error-detail string surfaced to logs / health. +const DETAIL_MAX_CHARS: usize = 200; + +/// A delivery target for one Datadog Logs intake. +pub struct DatadogSink { + name: String, + /// Full POST target — `https://http-intake.logs./api/v2/logs` + /// (or `http:///api/v2/logs` for a vetted mock intake). + endpoint_url: String, + /// Resolved Datadog API key. Sent only as the `DD-API-KEY` header. + api_key: String, + /// Datadog `ddsource` reserved attribute. + ddsource: String, + /// Comma-joined `ddtags` reserved attribute (empty string = omitted). + ddtags: String, + /// Datadog `service` reserved attribute. + service: String, + client: reqwest::Client, +} + +impl DatadogSink { + /// Build a sink for one Datadog site. + /// + /// `site` is a bare Datadog site host, e.g. `datadoghq.com`; the request + /// host is `http-intake.logs.` over https. A `site` that is a vetted + /// loopback host (the e2e's `mock-datadog` / `127.0.0.1` / `localhost`, + /// optionally with a `:port`) is posted to over http directly, so a local + /// mock intake needs no TLS. `tags` are rendered into a comma-joined + /// `ddtags` value once at build time. The `client` is shared across sinks + /// so connection pools and TLS sessions are reused. + #[allow(clippy::too_many_arguments)] + pub fn new( + name: impl Into, + site: &str, + api_key: impl Into, + ddsource: impl Into, + tags: &[String], + service: impl Into, + client: reqwest::Client, + ) -> Self { + Self { + name: name.into(), + endpoint_url: intake_url_for(site), + api_key: api_key.into(), + ddsource: ddsource.into(), + ddtags: tags.join(","), + service: service.into(), + client, + } + } +} + +#[async_trait] +impl ObservabilitySink for DatadogSink { + fn name(&self) -> &str { + &self.name + } + + fn capabilities(&self) -> SinkCapabilities { + SinkCapabilities { + // Datadog logs intake is at-least-once: no server-side dedup token. + idempotency: IdempotencyScheme::None, + // Independent posts; Datadog does not require cross-record ordering. + ordering: OrderingScope::None, + // Count-bounded today (parity with the SLS / OTLP sinks). Datadog + // caps the intake body (1000 logs and 5 MB uncompressed per + // request, 1 MB per log), but byte-aware chunking is only needed + // once full prompt/response content rides these records; on the + // metadata-only path the pipeline's per-batch record cap keeps + // bodies far under the limit. Declaring a byte ceiling the sink + // does not yet self-enforce would be a false promise (and a + // silent-drop bug under content) — so it stays `None` until the + // chunking lands (api7/ai-gateway#556), matching the SLS sink's + // resolved shape. + batch_unit: BatchUnit::Records, + max_batch_bytes: None, + // The intake accepts or rejects the whole request. + supports_partial_batch: false, + supports_streaming_ingest: false, + } + } + + async fn append_batch(&self, batch: &EventBatch, _marker: &IdempotencyMarker) -> SinkResult { + if batch.is_empty() { + return Ok(SinkAck::default()); + } + + // 1. One JSON log object per record → a single JSON array body. + let logs: Vec = batch.records.iter().map(|r| self.to_log(r)).collect(); + let raw = serde_json::to_vec(&Value::Array(logs)) + .map_err(|e| SinkError::Permanent(format!("datadog: json encode: {e}")))?; + + // 2. gzip the JSON (Datadog accepts `Content-Encoding: gzip`). + let compressed = gzip(&raw)?; + + // 3. Headers: JSON body, gzip encoding, and the API key. + let mut headers = HeaderMap::new(); + headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + headers.insert(CONTENT_ENCODING, HeaderValue::from_static("gzip")); + // The API key is the only place the secret appears. An un-resolvable + // key (empty) is sent as-is so Datadog answers 403 and the failure + // surfaces as a delivery-health auth error rather than a silent drop. + let key = HeaderValue::from_str(&self.api_key).map_err(|_| { + SinkError::Permanent("datadog: api key has invalid header bytes".into()) + })?; + headers.insert(DD_API_KEY, key); + + // 4. Deliver. + let resp = match self + .client + .post(&self.endpoint_url) + .headers(headers) + .body(compressed) + .send() + .await + { + Ok(resp) => resp, + // Connect / DNS / timeout — transient by nature. The endpoint URL + // carries no secret, so it is safe in the error detail. + Err(e) => { + return Err(SinkError::Transient(format!( + "datadog: POST {}: {e}", + self.endpoint_url + ))) + } + }; + + let status = resp.status(); + if status.is_success() { + return Ok(SinkAck { + accepted: batch.len(), + ..SinkAck::default() + }); + } + + let body = resp.text().await.unwrap_or_default(); + let detail = parse_datadog_error(status, &body); + // 429 (rate limit) and 5xx (502/503/504, transient server faults) are + // worth retrying; other 4xx (400 malformed / 401/403 auth / 413 too + // large) are config/auth/payload errors that fail identically on retry. + if is_transient_status(status) { + Err(SinkError::Transient(detail)) + } else { + Err(SinkError::Permanent(detail)) + } + } + + async fn healthcheck(&self) -> SinkHealth { + // A real connectivity probe (and the control-plane "test connection" + // affordance) lands with the health/metrics surface; until then a sink + // reports healthy and its delivery errors surface via + // `SinkStats::last_error`. (Mirrors the SLS / OTLP sinks.) + SinkHealth::healthy() + } +} + +impl DatadogSink { + /// Map one canonical record to a Datadog log JSON object. + /// + /// The metadata is produced by *serializing* [`crate::usage::UsageEvent`] + /// rather than hand-listing its 30-plus fields: that single-sources the + /// schema and inherits the exact `skip_serializing_if` emptiness rules the + /// cp-api wire uses (empty strings / zero counters are omitted). Field + /// names are remapped to the OTel GenAI semconv where one exists + /// (`gen_ai.system`, `gen_ai.request.model` / `gen_ai.response.model`, + /// `gen_ai.usage.input_tokens` / `gen_ai.usage.output_tokens`, …) so a + /// Datadog log and an OTLP span carry the same keys; the remainder land + /// under an `aisix.` prefix. The Datadog reserved attributes (`ddsource`, + /// `ddtags`, `service`, `message`) are set as siblings. + fn to_log(&self, record: &SinkRecord) -> Value { + let mut obj = Map::new(); + + // Datadog reserved attributes. + obj.insert("ddsource".into(), json!(self.ddsource)); + if !self.ddtags.is_empty() { + obj.insert("ddtags".into(), json!(self.ddtags)); + } + obj.insert("service".into(), json!(self.service)); + obj.insert("message".into(), json!(summary_message(&record.usage))); + obj.insert("schema_version".into(), json!(record.schema_version)); + + // Flatten the usage metadata, remapping each field to its GenAI semconv + // (or `aisix.`) key. Empty strings are skipped; numeric 0 / false stay. + if let Ok(Value::Object(fields)) = serde_json::to_value(&record.usage) { + for (key, value) in fields { + let Some(rendered) = render_scalar(&value) else { + continue; + }; + obj.insert(map_field_name(&key), rendered); + } + } + + // Opt-in captured content as flat, queryable fields. Absent on the + // default metadata-only path, so prompts never leak there. + if let Some(content) = &record.content { + obj.insert("gen_ai.prompt".into(), json!(content.prompt)); + obj.insert("gen_ai.completion".into(), json!(content.response)); + if content.truncated { + obj.insert("content_truncated".into(), json!(true)); + } + } + + Value::Object(obj) + } +} + +/// Render a JSON scalar as a Datadog log field value; `None` skips the field. +/// +/// Empty strings are skipped so the log omits blank fields uniformly: most +/// optional `UsageEvent` fields already drop out via `skip_serializing_if`, +/// but a few (`model_id`, `api_key_id`) carry only `#[serde(default)]` and +/// would otherwise serialize as `""`. Numeric `0` and `false` are kept — a +/// zero token count or status code is real data. Nested fields (e.g. +/// `applied_guardrails`) ride through structurally so Datadog can facet them. +fn render_scalar(value: &Value) -> Option { + match value { + Value::Null => None, + Value::String(s) if s.is_empty() => None, + other => Some(other.clone()), + } +} + +/// Map a `UsageEvent` field name to the key it lands under in the Datadog log. +/// +/// The OTel GenAI semantic conventions own the names for the LLM dimensions +/// that have a convention; everything else keeps an `aisix.`-prefixed custom +/// key so the field set is self-describing and collision-free in Datadog's +/// attribute namespace. This is the same naming the OTLP span builder uses. +fn map_field_name(field: &str) -> String { + match field { + // ── OTel GenAI semconv ── + "provider_model_version" => "gen_ai.response.model", + "provider_request_id" => "gen_ai.response.id", + "finish_reason" => "gen_ai.response.finish_reason", + "prompt_tokens" => "gen_ai.usage.input_tokens", + "completion_tokens" => "gen_ai.usage.output_tokens", + // ── HTTP semconv ── + "status_code" => "http.response.status_code", + // ── AISIX custom dimensions (no semconv) ── + other => return format!("aisix.{other}"), + } + .to_string() +} + +/// A short human-readable summary used as the Datadog log `message`. Datadog's +/// Log Explorer shows `message` as the row text, so a compact one-liner keyed +/// on the request makes the log readable without expanding attributes. +fn summary_message(usage: &crate::usage::UsageEvent) -> String { + let model = if !usage.provider_model_version.is_empty() { + usage.provider_model_version.as_str() + } else if !usage.model_id.is_empty() { + usage.model_id.as_str() + } else { + "-" + }; + format!( + "ai-gateway request {} model={} status={}", + usage.request_id, model, usage.status_code + ) +} + +/// The Datadog logs-intake error envelope — `{"errors": ["...", ...]}`. +#[derive(Deserialize)] +struct DatadogErrorBody { + #[serde(default)] + errors: Vec, +} + +/// Parse the Datadog error body into a masked detail, falling back to the raw +/// (truncated) body when it isn't the JSON envelope. The detail never contains +/// the API key — Datadog error messages echo neither the key nor the header. +fn parse_datadog_error(status: reqwest::StatusCode, body: &str) -> String { + if let Ok(parsed) = serde_json::from_str::(body) { + if !parsed.errors.is_empty() { + return truncate(&format!("HTTP {status}: {}", parsed.errors.join("; "))); + } + } + truncate(&format!("HTTP {status}: {body}")) +} + +/// Whether a Datadog intake HTTP status means "retry with backoff". +/// +/// Datadog returns `429 Too Many Requests` on rate-limit and `5xx` +/// (`502`/`503`/`504`) on transient server faults; both are retried. Other +/// `4xx` (`400` malformed payload, `401`/`403` bad API key, `413` payload too +/// large) are permanent — a retry of the same batch fails identically. +fn is_transient_status(status: reqwest::StatusCode) -> bool { + status.is_server_error() || status == reqwest::StatusCode::TOO_MANY_REQUESTS +} + +/// gzip a byte slice (RFC 1952). Permanent on the rare encode failure. +fn gzip(data: &[u8]) -> Result, SinkError> { + let mut enc = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default()); + enc.write_all(data) + .and_then(|_| enc.finish()) + .map_err(|e| SinkError::Permanent(format!("datadog: gzip: {e}"))) +} + +/// Truncate a masked detail string to a bounded length for logs / health. +fn truncate(s: &str) -> String { + s.chars().take(DETAIL_MAX_CHARS).collect() +} + +/// Compute the full intake POST URL for a Datadog `site`. +/// +/// A real Datadog site host becomes +/// `https://http-intake.logs./api/v2/logs`. A vetted loopback host (the +/// e2e's `mock-datadog` / `127.0.0.1` / `localhost`, optionally with a +/// `host:port`) is posted to over http directly, so a local mock receiver +/// needs no TLS. The exporter schema only admits a site from the allow-list +/// (the seven real sites plus those three loopback hosts), so this branch +/// can't be used to redirect real traffic to an arbitrary host. +fn intake_url_for(site: &str) -> String { + let site = site.trim().trim_end_matches('/'); + if is_loopback_site(site) { + format!("http://{site}/api/v2/logs") + } else { + format!("https://http-intake.logs.{site}/api/v2/logs") + } +} + +/// Whether a site token is a vetted loopback mock host (optionally `host:port`) +/// rather than a real Datadog site. Matches the schema's loopback allow-list. +fn is_loopback_site(site: &str) -> bool { + let host = site.split(':').next().unwrap_or(site); + matches!(host, "mock-datadog" | "127.0.0.1" | "localhost") +} + +/// Resolve an exporter's `credential_ref` to the Datadog API key from the DP's +/// local environment. +/// +/// The API key never travels on the kine path (the control plane stores only +/// the reference), so the DP looks it up where it actually runs. The reference +/// is upper-cased with non-alphanumerics folded to `_`, then read from +/// `DD_CRED__API_KEY`. The prefix is deliberately NOT `AISIX_`: that +/// namespace is owned by the config loader (`Environment::with_prefix("AISIX")`), +/// so an `AISIX_`-named secret would be reinterpreted as a config override. +/// Returns `None` when the key is unset or blank — the caller then lets the +/// misconfiguration surface as a delivery-health auth error rather than POST +/// with an empty key. (Mirrors `resolve_sls_credential` / +/// `resolve_object_store_credential`.) +pub fn resolve_datadog_credential(credential_ref: &str) -> Option { + resolve_datadog_credential_with(credential_ref, |key| std::env::var(key).ok()) +} + +/// Reference-resolution core, parameterized over the variable source so it is +/// testable without mutating the process environment. +fn resolve_datadog_credential_with( + credential_ref: &str, + lookup: impl Fn(&str) -> Option, +) -> Option { + let slug: String = credential_ref + .chars() + .map(|c| { + if c.is_ascii_alphanumeric() { + c.to_ascii_uppercase() + } else { + '_' + } + }) + .collect(); + let key = lookup(&format!("DD_CRED_{slug}_API_KEY"))?; + if key.is_empty() { + return None; + } + Some(key) +} + +#[cfg(test)] +mod tests { + use super::{ + intake_url_for, is_transient_status, parse_datadog_error, resolve_datadog_credential_with, + DatadogSink, + }; + use crate::sink::{EventBatch, IdempotencyMarker, ObservabilitySink, SinkContent, SinkRecord}; + use crate::usage::UsageEvent; + use serde_json::Value; + use std::sync::Arc; + use wiremock::matchers::{method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + const POST_PATH: &str = "/api/v2/logs"; + + /// Decode the request body a mock server captured back into the JSON array + /// of log objects (gunzip, then JSON-parse). + fn decode_logs(req: &wiremock::Request) -> Vec { + assert_eq!( + req.headers + .get("content-encoding") + .expect("content-encoding header") + .to_str() + .unwrap(), + "gzip" + ); + let mut gz = flate2::read::GzDecoder::new(&req.body[..]); + let mut text = String::new(); + std::io::Read::read_to_string(&mut gz, &mut text).expect("gunzip"); + match serde_json::from_str(&text).expect("valid json") { + Value::Array(a) => a, + other => panic!("expected a JSON array body, got {other}"), + } + } + + fn sink_for(server: &MockServer, tags: &[String]) -> DatadogSink { + // The mock server URI is `http://127.0.0.1:` — a vetted loopback + // site, so the sink posts to it directly over http. + let host = server.uri().strip_prefix("http://").unwrap().to_string(); + DatadogSink::new( + "datadog-test", + &host, + "test-dd-api-key", + "aisix-ai-gateway", + tags, + "ai-gateway", + reqwest::Client::new(), + ) + } + + fn batch_of(records: Vec) -> EventBatch { + EventBatch::new(records.into_iter().map(Arc::new).collect()) + } + + #[tokio::test] + async fn appends_batch_posts_gzipped_json_with_api_key_header() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path(POST_PATH)) + .respond_with(ResponseTemplate::new(202)) + .mount(&server) + .await; + + let sink = sink_for(&server, &["team:platform".into(), "tier:prod".into()]); + let event = UsageEvent { + request_id: "req-42".into(), + occurred_at: "2026-05-01T12:00:00Z".into(), + model_id: "gpt-4o".into(), + status_code: 200, + prompt_tokens: 5, + completion_tokens: 7, + latency_ms: 123, + provider_model_version: "gpt-4o-2024-08-06".into(), + finish_reason: "stop".into(), + ..UsageEvent::default() + }; + let ack = sink + .append_batch( + &batch_of(vec![SinkRecord::metadata_only(event)]), + &IdempotencyMarker::None, + ) + .await + .expect("delivery succeeds"); + assert_eq!(ack.accepted, 1); + + let requests = server.received_requests().await.unwrap(); + assert_eq!(requests.len(), 1); + let req = &requests[0]; + + // Wire shape: gzipped JSON to /api/v2/logs with the DD-API-KEY header. + assert_eq!(req.method.as_str(), "POST"); + assert_eq!(req.url.path(), POST_PATH); + assert_eq!( + req.headers.get("content-type").unwrap().to_str().unwrap(), + "application/json" + ); + assert_eq!( + req.headers.get("dd-api-key").unwrap().to_str().unwrap(), + "test-dd-api-key" + ); + + // Body round-trips: one log object with the reserved attrs + GenAI + // semconv token fields. + let logs = decode_logs(req); + assert_eq!(logs.len(), 1, "exactly one log object"); + let log = &logs[0]; + assert_eq!(log["ddsource"], "aisix-ai-gateway"); + assert_eq!(log["ddtags"], "team:platform,tier:prod"); + assert_eq!(log["service"], "ai-gateway"); + assert!(log["message"].as_str().unwrap().contains("req-42")); + assert_eq!(log["schema_version"], "1.0"); + // GenAI semconv names for the LLM dimensions. + assert_eq!(log["gen_ai.usage.input_tokens"], 5); + assert_eq!(log["gen_ai.usage.output_tokens"], 7); + assert_eq!(log["gen_ai.response.model"], "gpt-4o-2024-08-06"); + assert_eq!(log["gen_ai.response.finish_reason"], "stop"); + assert_eq!(log["http.response.status_code"], 200); + // AISIX custom dimensions under the `aisix.` prefix. + assert_eq!(log["aisix.request_id"], "req-42"); + assert_eq!(log["aisix.model_id"], "gpt-4o"); + assert_eq!(log["aisix.latency_ms"], 123); + + // The API key must NEVER appear in the body anywhere. + let body_text = serde_json::to_string(&logs).unwrap(); + assert!( + !body_text.contains("test-dd-api-key"), + "api key must never be on the wire body: {body_text}" + ); + // Empty metadata is omitted uniformly (no blank `aisix.api_key_id`). + assert!(log.get("aisix.api_key_id").is_none()); + // Metadata-only path never carries a prompt. + assert!(log.get("gen_ai.prompt").is_none()); + } + + #[tokio::test] + async fn full_content_record_emits_prompt_and_response() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path(POST_PATH)) + .respond_with(ResponseTemplate::new(202)) + .mount(&server) + .await; + + let sink = sink_for(&server, &[]); + let record = SinkRecord::metadata_only(UsageEvent { + request_id: "req-content".into(), + status_code: 200, + ..UsageEvent::default() + }) + .with_content(SinkContent { + prompt: "what is 2+2?".into(), + response: "4".into(), + truncated: true, + }); + sink.append_batch(&batch_of(vec![record]), &IdempotencyMarker::None) + .await + .expect("delivery succeeds"); + + let requests = server.received_requests().await.unwrap(); + let logs = decode_logs(&requests[0]); + let log = &logs[0]; + assert_eq!(log["gen_ai.prompt"], "what is 2+2?"); + assert_eq!(log["gen_ai.completion"], "4"); + assert_eq!(log["content_truncated"], true); + // ddtags is omitted entirely when no tags are configured. + assert!(log.get("ddtags").is_none(), "empty ddtags must be omitted"); + } + + #[tokio::test] + async fn server_error_is_transient() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path(POST_PATH)) + .respond_with( + ResponseTemplate::new(503).set_body_string(r#"{"errors":["service unavailable"]}"#), + ) + .mount(&server) + .await; + + let err = sink_for(&server, &[]) + .append_batch( + &batch_of(vec![SinkRecord::metadata_only(UsageEvent::default())]), + &IdempotencyMarker::None, + ) + .await + .expect_err("5xx fails"); + assert!(err.is_transient(), "5xx must be retried: {err}"); + } + + #[tokio::test] + async fn rate_limited_429_is_transient() { + // Datadog signals back-pressure with 429; a log sink must back off and + // retry, not drop the batch. + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path(POST_PATH)) + .respond_with( + ResponseTemplate::new(429).set_body_string(r#"{"errors":["rate limit"]}"#), + ) + .mount(&server) + .await; + + let err = sink_for(&server, &[]) + .append_batch( + &batch_of(vec![SinkRecord::metadata_only(UsageEvent::default())]), + &IdempotencyMarker::None, + ) + .await + .expect_err("429 fails this attempt"); + assert!(err.is_transient(), "429 must be retried: {err}"); + } + + #[tokio::test] + async fn auth_error_403_is_permanent() { + let server = MockServer::start().await; + Mock::given(method("POST")) + .and(path(POST_PATH)) + .respond_with(ResponseTemplate::new(403).set_body_string(r#"{"errors":["Forbidden"]}"#)) + .mount(&server) + .await; + + let err = sink_for(&server, &[]) + .append_batch( + &batch_of(vec![SinkRecord::metadata_only(UsageEvent::default())]), + &IdempotencyMarker::None, + ) + .await + .expect_err("bad api key fails"); + assert!(!err.is_transient(), "auth error must not be retried: {err}"); + } + + #[test] + fn transient_status_classification() { + use reqwest::StatusCode; + for s in [ + StatusCode::TOO_MANY_REQUESTS, + StatusCode::BAD_GATEWAY, + StatusCode::SERVICE_UNAVAILABLE, + StatusCode::GATEWAY_TIMEOUT, + ] { + assert!(is_transient_status(s), "{s} should be transient"); + } + for s in [ + StatusCode::BAD_REQUEST, + StatusCode::UNAUTHORIZED, + StatusCode::FORBIDDEN, + StatusCode::PAYLOAD_TOO_LARGE, + ] { + assert!(!is_transient_status(s), "{s} should be permanent"); + } + } + + #[test] + fn parse_error_extracts_messages_and_masks_body() { + let detail = parse_datadog_error( + reqwest::StatusCode::BAD_REQUEST, + r#"{"errors":["Invalid log format","bad ddsource"]}"#, + ); + assert!(detail.contains("Invalid log format")); + assert!(detail.contains("bad ddsource")); + assert!(detail.contains("400")); + + // Non-envelope body falls back to the raw (truncated) text. + let detail = parse_datadog_error( + reqwest::StatusCode::BAD_GATEWAY, + "502 Bad Gateway", + ); + assert!(detail.contains("502")); + } + + #[test] + fn intake_url_builds_https_for_a_real_site() { + assert_eq!( + intake_url_for("datadoghq.com"), + "https://http-intake.logs.datadoghq.com/api/v2/logs" + ); + assert_eq!( + intake_url_for("ap1.datadoghq.com"), + "https://http-intake.logs.ap1.datadoghq.com/api/v2/logs" + ); + // Trailing slash trimmed. + assert_eq!( + intake_url_for("datadoghq.eu/"), + "https://http-intake.logs.datadoghq.eu/api/v2/logs" + ); + } + + #[test] + fn intake_url_uses_http_for_a_loopback_site() { + // A loopback mock is posted to directly over http — no `http-intake` + // prefix, no TLS. + assert_eq!( + intake_url_for("mock-datadog:8080"), + "http://mock-datadog:8080/api/v2/logs" + ); + assert_eq!( + intake_url_for("127.0.0.1:9001"), + "http://127.0.0.1:9001/api/v2/logs" + ); + assert_eq!(intake_url_for("localhost"), "http://localhost/api/v2/logs"); + } + + #[test] + fn resolve_credential_maps_reference_to_env_key() { + // The ref folds to an upper `_`-joined slug; the key comes from + // `DD_CRED__API_KEY`. + let store = |key: &str| match key { + "DD_CRED_DATADOG_PROD_API_KEY" => Some("dd-key-123".to_string()), + _ => None, + }; + assert_eq!( + resolve_datadog_credential_with("datadog-prod", store), + Some("dd-key-123".to_string()) + ); + // Case-insensitive: `Datadog.Prod` folds to the same slug. + assert_eq!( + resolve_datadog_credential_with("Datadog.Prod", store), + Some("dd-key-123".to_string()) + ); + } + + #[test] + fn resolve_credential_is_none_when_unset_or_blank() { + // Key absent → None. + assert_eq!(resolve_datadog_credential_with("missing", |_| None), None); + // Blank value is treated as unset (never POST with an empty key). + assert_eq!( + resolve_datadog_credential_with("x", |_| Some(String::new())), + None + ); + } +} diff --git a/crates/aisix-obs/src/sink/mod.rs b/crates/aisix-obs/src/sink/mod.rs index 762c4357..e02bf237 100644 --- a/crates/aisix-obs/src/sink/mod.rs +++ b/crates/aisix-obs/src/sink/mod.rs @@ -13,6 +13,7 @@ //! and the concrete sinks (SLS, …) build on it. mod capabilities; +mod datadog; mod manager; mod object_store; mod pipeline; @@ -22,6 +23,7 @@ mod sls; pub use capabilities::{ BatchUnit, ChannelKey, IdempotencyMarker, IdempotencyScheme, OrderingScope, SinkCapabilities, }; +pub use datadog::{resolve_datadog_credential, DatadogSink}; pub use manager::ExporterPipelines; pub use object_store::{build_object_store_sink, ObjectStoreSink}; pub use pipeline::{PipelineConfig, SinkHandle, SinkPipeline, SinkStatsSnapshot}; diff --git a/schemas/resources/observability_exporter.schema.json b/schemas/resources/observability_exporter.schema.json index e8b64d61..41916f34 100644 --- a/schemas/resources/observability_exporter.schema.json +++ b/schemas/resources/observability_exporter.schema.json @@ -151,6 +151,66 @@ ] } } + }, + { + "description": "Datadog native **Logs HTTP intake** target. Each request event becomes one Datadog log object, gzip-compressed and POSTed to `https://http-intake.logs./api/v2/logs`. Like `aliyun_sls` (and unlike `otlp_http`), the Datadog API key is **never** part of this config: it would otherwise sit in plaintext on the kine path, and a Datadog API key grants broad org access. Instead the config carries a [`credential_ref`] pointer that the customer-side DP resolves to the real key locally (env / mounted secret); API7's control plane stores only the reference. This deliberately diverges from issue #688's `api_key: SecretRef` draft to stay consistent with SLS / object_store and avoid the #692 credential-encryption dependency.\n\n[`credential_ref`]: DatadogConfig::credential_ref", + "type": "object", + "required": [ + "credential_ref", + "kind", + "service", + "site" + ], + "properties": { + "content_max_bytes": { + "description": "Per-field byte cap for captured content under `content_mode = full`. The prompt and the response are each truncated to this many bytes (UTF-8-boundary safe), and the log carries a `content_truncated` marker when either was cut. Ignored under `metadata_only`. Defaults to 128 KiB.\n\nThis bounds each field *independently*: a single log carries BOTH the prompt and the response plus metadata, so the encoded log can reach ~2× this cap. Datadog rejects any single log over 1 MB and any request over 5 MB / 1000 logs; byte-aware per-log/per-request splitting to those limits is not yet enforced (tracked in api7/ai-gateway#556) — until it lands, a large cap on a busy `full` exporter risks Datadog rejecting an oversized batch (a `Permanent` delivery error surfaced via `last_error`). The 128 KiB default keeps a log well under the per-log limit.\n\n`0` is rejected — `full` with a zero cap captures nothing, which is a misconfiguration. The `range(min = 1, max = …)` keeps the generated JSON schema in step with the runtime validator so the CP and DP agree on the bounds.", + "default": 131072, + "type": "integer", + "format": "uint32", + "maximum": 1048576.0, + "minimum": 1.0 + }, + "content_mode": { + "description": "Whether captured request/response content is delivered to Datadog. `metadata_only` (default) ships only operational metadata — never a prompt or response. `full` additionally captures the request prompt and the assembled response, each truncated to [`content_max_bytes`]. Enabling `full` writes end-user prompt / response text into the customer's Datadog org, so the dashboard must surface the privacy implication when an operator turns it on. Reuses [`SlsContentMode`] — the codebase's existing `metadata_only | full` model — so the shared content-capture plumbing (`content_record` / `content_capture_cap`) stays single-sourced.\n\n[`content_max_bytes`]: DatadogConfig::content_max_bytes", + "default": "metadata_only", + "allOf": [ + { + "$ref": "#/definitions/SlsContentMode" + } + ] + }, + "credential_ref": { + "description": "Opaque pointer to the Datadog API key, resolved locally by the DP at delivery time. The plaintext key MUST NOT live in etcd/kine — the control plane stores only this reference, never the key itself. The DP reads `DD_CRED__API_KEY` from its own environment, where `` upper-cases the reference with non-alphanumerics folded to `_`.", + "type": "string" + }, + "ddsource": { + "description": "Datadog `ddsource` reserved attribute — the integration/source name. Defaults to `aisix-ai-gateway`.", + "default": "aisix-ai-gateway", + "type": "string" + }, + "kind": { + "type": "string", + "enum": [ + "datadog" + ] + }, + "service": { + "description": "Datadog `service` reserved attribute — the service name every log from this exporter is tagged with in Datadog's Log Explorer.", + "type": "string" + }, + "site": { + "description": "Datadog site, validated against the allow-list in the loader schema (`datadoghq.com`, `us3`/`us5`/`ap1`/`ap2` regions, `datadoghq.eu`, `ddog-gov.com`). The intake host is `http-intake.logs.`. A scheme-qualified loopback host (the e2e's `http://mock-datadog:*`) is admitted only for vetted local mocks, never to redirect real traffic.", + "type": "string" + }, + "tags": { + "description": "Operator-defined tags rendered into Datadog's comma-joined `ddtags` reserved attribute (e.g. `[\"team:platform\", \"tier:prod\"]` → `team:platform,tier:prod`). Empty by default.", + "default": [], + "type": "array", + "items": { + "type": "string" + } + } + } } ], "required": [ @@ -216,7 +276,7 @@ ] }, "SlsContentMode": { - "description": "Content-capture mode for an SLS exporter. Defaults to the privacy-preserving `metadata_only`.", + "description": "Content-capture mode for an SLS / Datadog exporter. Defaults to the privacy-preserving `metadata_only`. (`Hash` so a Datadog exporter's fingerprint can cover its content config; see `fingerprint_datadog`.)", "oneOf": [ { "description": "Operational metadata only — never the prompt or response.", diff --git a/tests/e2e/src/cases/datadog-exporter-e2e.test.ts b/tests/e2e/src/cases/datadog-exporter-e2e.test.ts new file mode 100644 index 00000000..1d4fba55 --- /dev/null +++ b/tests/e2e/src/cases/datadog-exporter-e2e.test.ts @@ -0,0 +1,391 @@ +import { createHash } from "node:crypto"; +import { createServer, type IncomingMessage, type Server } from "node:http"; +import { gunzipSync } from "node:zlib"; +import { afterAll, beforeAll, describe, expect, test } from "vitest"; +import { + AdminClient, + EtcdClient, + pickFreePort, + spawnApp, + startOpenAiUpstream, + waitConfigPropagation, + type OpenAiUpstream, + type SpawnedApp, +} from "../harness/index.js"; + +// L2 mock e2e (api7/ai-gateway#57, AISIX-Cloud#688): a dashboard-configured +// `datadog` exporter makes the real DP deliver request events to Datadog's +// native Logs HTTP intake over a gzip JSON-array POST. We stand up a mock +// Datadog intake (the SLS suite's in-test-receiver pattern), register the +// exporter, drive one chat, and assert the DP POSTed a correctly-shaped, +// API-key-bearing request to `/api/v2/logs`. +// +// Scope boundary (deliberate), mirroring the SLS L2 test: this pins the +// WIRING and the on-the-wire SHAPE the intake observes — the path, the +// `Content-Encoding: gzip` framing, that the `DD-API-KEY` header carries the +// key the DP resolved from its environment (proving the credential_ref → env +// path, not a key on the kine config), and that the gunzipped JSON-array body +// carries the Datadog reserved attributes + the OTel GenAI semconv token +// fields. The full field-mapping matrix is covered by the Rust round-trip unit +// tests (`sink::datadog::tests`); that a real Datadog site accepts the request +// is validated by the control-plane full-chain e2e (api7/AISIX-Cloud), not here. +// +// The harness binds the in-test mock to a free loopback port and points `site` +// at `127.0.0.1:`, exactly as the SLS / OTLP mock-edge tests point their +// endpoint at `http://127.0.0.1:`. The `datadog` `site` validator admits +// a loopback host with an optional `:port` via the same `(:[0-9]+)?` regex +// group as the SLS / OTLP bypasses (api7/ai-gateway#548, fixed in this PR), so +// the Admin API accepts it. + +const CALLER_PLAINTEXT = "sk-datadog-exporter-caller-PLAINTEXT"; +const CALLER_KEY_HASH = createHash("sha256").update(CALLER_PLAINTEXT).digest("hex"); +const PROVIDER_SECRET = "sk-mock-datadog-exporter"; + +// The exporter's credential_ref; the DP resolves it from the env var the +// harness injects (DD_CRED__API_KEY, ref upper-cased, non-alnum → `_`). +const CREDENTIAL_REF = "e2e"; +// The mock accepts any key — this is the test key the DP must surface in the +// DD-API-KEY header (and NOWHERE else). Not a real Datadog credential. +const DD_API_KEY = "dd-e2e-test-key-7f3a2b"; + +const DD_SERVICE = "aisix-e2e"; +const DD_TAGS = ["team:platform", "tier:e2e"]; +const INTAKE_PATH = "/api/v2/logs"; + +// Unique tokens planted in the request + the mock upstream response, so the +// content-capture assertion can prove which made it into the log body. The DP +// captures the prompt (request body, carrying PROMPT_TOKEN) and the assembled +// response (assistant content, carrying RESPONSE_TOKEN) only under +// `content_mode = full`. +const PROMPT_TOKEN = "dd-prompt-tok-9f3a2b"; +const RESPONSE_TOKEN = "dd-response-tok-7c1d8e"; + +interface CapturedLog { + method: string; + path: string; + headers: IncomingMessage["headers"]; + /** The gunzipped, JSON-parsed log objects (the intake body is a JSON array). */ + logs: unknown[]; + /** The raw decompressed body text, for substring (content-leak) assertions. */ + bodyText: string; +} + +interface MockDatadog { + /** `host:port` the exporter's `site` points at (the sink builds `http:///api/v2/logs`). */ + site: string; + requests: CapturedLog[]; + close(): Promise; +} + +/** Stand up a mock Datadog Logs intake on a free loopback port. */ +async function startMockDatadog(): Promise { + const requests: CapturedLog[] = []; + const server: Server = createServer((req, res) => { + const chunks: Buffer[] = []; + req.on("data", (c: Buffer) => chunks.push(c)); + req.on("end", () => { + const path = (req.url ?? "").split("?")[0]; + if (req.method === "POST" && path === INTAKE_PATH) { + const compressed = Buffer.concat(chunks); + // The intake advertises `Content-Encoding: gzip`; gunzip before parse. + // If the body isn't valid gzip JSON the capture stays empty and the + // poll below times out — a loud failure, never a false green. + let logs: unknown[] = []; + let bodyText = ""; + try { + bodyText = gunzipSync(compressed).toString("utf8"); + const parsed: unknown = JSON.parse(bodyText); + if (Array.isArray(parsed)) logs = parsed; + } catch { + // leave logs empty / bodyText as-is — the assertion side will fail + // visibly rather than silently pass. + } + requests.push({ + method: req.method ?? "", + path, + headers: req.headers, + logs, + bodyText, + }); + } + // Datadog's logs intake answers 202 Accepted on success. + res.statusCode = 202; + res.end(); + }); + }); + const port = await pickFreePort(); + await new Promise((resolve) => server.listen(port, "127.0.0.1", resolve)); + return { + site: `127.0.0.1:${port}`, + requests, + async close() { + await new Promise((resolve, reject) => { + server.close((err) => (err ? reject(err) : resolve())); + }); + }, + }; +} + +async function seedRouting(admin: AdminClient, upstream: OpenAiUpstream, model: string) { + const pk = await admin.createProviderKey({ + display_name: `${model}-pk`, + secret: PROVIDER_SECRET, + api_base: `${upstream.baseUrl}/v1`, + }); + await admin.createModel({ + display_name: model, + provider: "openai", + model_name: "gpt-4o-mini", + provider_key_id: pk.id, + }); + await admin.createApiKey({ + key_hash: CALLER_KEY_HASH, + allowed_models: [model], + }); +} + +async function chat(app: SpawnedApp, model: string, content: string): Promise { + return fetch(`${app.proxyUrl}/v1/chat/completions`, { + method: "POST", + headers: { + authorization: `Bearer ${CALLER_PLAINTEXT}`, + "content-type": "application/json", + }, + body: JSON.stringify({ + model, + messages: [{ role: "user", content }], + }), + }); +} + +async function waitForIntake( + dd: MockDatadog, + timeoutMs = 10_000, +): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + const hit = dd.requests.find((r) => r.path === INTAKE_PATH && r.logs.length > 0); + if (hit) return hit; + await new Promise((r) => setTimeout(r, 50)); + } + throw new Error(`no decodable POST to ${INTAKE_PATH} recorded within ${timeoutMs}ms`); +} + +function headerValue(headers: IncomingMessage["headers"], name: string): string { + const v = headers[name]; + return Array.isArray(v) ? (v[0] ?? "") : (v ?? ""); +} + +/** Narrow one captured log object to a plain record for field assertions. */ +function asRecord(log: unknown): Record { + expect(log, "log entry must be a JSON object").toBeTypeOf("object"); + return log as Record; +} + +describe("datadog exporter e2e (#688): DP delivers a gzip JSON intake to Datadog", () => { + let etcdReachable = false; + let upstream: OpenAiUpstream | undefined; + let dd: MockDatadog | undefined; + const apps: SpawnedApp[] = []; + + beforeAll(async () => { + etcdReachable = await new EtcdClient().ping(); + if (!etcdReachable) return; + // Plant the response token in the mock upstream's assistant content so the + // content-capture test can search for it in the `full` log body. + upstream = await startOpenAiUpstream({ + nonStreamBody: { + id: "mock-datadog-1", + object: "chat.completion", + created: 1_700_000_000, + model: "mock-model", + choices: [ + { + index: 0, + message: { role: "assistant", content: `sure, ${RESPONSE_TOKEN} noted` }, + finish_reason: "stop", + }, + ], + usage: { prompt_tokens: 5, completion_tokens: 3, total_tokens: 8 }, + }, + }); + dd = await startMockDatadog(); + }); + + afterAll(async () => { + await Promise.all(apps.map((a) => a.exit())); + await upstream?.close(); + await dd?.close(); + }); + + test( + "a configured datadog exporter posts a gzip JSON-array intake with the resolved DD-API-KEY", + async (ctx) => { + if (!etcdReachable || !upstream || !dd) { + ctx.skip(); + return; + } + const app = await spawnApp({ + // The API key rides the DP's own env, never the kine config. + extraEnv: { + [`DD_CRED_${CREDENTIAL_REF.toUpperCase()}_API_KEY`]: DD_API_KEY, + }, + }); + apps.push(app); + const admin = new AdminClient(app.adminUrl, app.adminKey); + await admin.createObservabilityExporter({ + name: "mock-datadog", + enabled: true, + kind: "datadog", + site: dd.site, + credential_ref: CREDENTIAL_REF, + service: DD_SERVICE, + tags: DD_TAGS, + // Default privacy posture: operational metadata only, never content. + content_mode: "metadata_only", + }); + await seedRouting(admin, upstream, "datadog-exporter-model"); + + await waitConfigPropagation(async () => { + try { + const r = await chat(app, "datadog-exporter-model", "hello datadog"); + await r.text(); + return r.status === 200; + } catch { + return false; + } + }); + + const res = await chat(app, "datadog-exporter-model", "hello datadog"); + expect(res.status).toBe(200); + await res.text(); + + const intake = await waitForIntake(dd); + + // Correct intake verb + path. + expect(intake.method).toBe("POST"); + expect(intake.path).toBe(INTAKE_PATH); + // Datadog Logs intake wire shape: gzip-compressed JSON. + expect(headerValue(intake.headers, "content-encoding")).toBe("gzip"); + expect(headerValue(intake.headers, "content-type")).toBe("application/json"); + // The DD-API-KEY header carries EXACTLY the key the DP resolved from its + // env — proves credential_ref → DD_CRED__API_KEY resolution wired + // into delivery, and that the key was not something else. + expect(headerValue(intake.headers, "dd-api-key")).toBe(DD_API_KEY); + + // Body is a JSON array of log objects (one per request event). + expect(Array.isArray(intake.logs)).toBe(true); + expect(intake.logs.length).toBeGreaterThan(0); + const log = asRecord(intake.logs[0]); + + // Datadog reserved attributes set by the sink. + expect(log.ddsource).toBeTypeOf("string"); + expect((log.ddsource as string).length).toBeGreaterThan(0); + expect(log.service).toBe(DD_SERVICE); + // ddtags is the configured tags, comma-joined. + expect(log.ddtags).toBe(DD_TAGS.join(",")); + + // OTel GenAI semconv token fields ride the log (same keys an OTLP span + // would carry). Counts come from the mock upstream's usage block. + expect(log["gen_ai.usage.input_tokens"]).toBe(5); + expect(log["gen_ai.usage.output_tokens"]).toBe(3); + // A model dimension is present under a GenAI / aisix key (the upstream + // echoes `mock-model`); assert the response model semconv field carries it. + expect(log["gen_ai.response.model"]).toBe("mock-model"); + + // The API key must appear ONLY in the header — never anywhere in the body. + expect(intake.bodyText.includes(DD_API_KEY)).toBe(false); + // Metadata-only posture: no captured prompt / response fields at all. + expect(log["gen_ai.prompt"]).toBeUndefined(); + expect(log["gen_ai.completion"]).toBeUndefined(); + }, + 60_000, + ); + + test( + "content_mode=full ships gen_ai.prompt/completion; metadata_only ships neither", + async (ctx) => { + if (!etcdReachable || !upstream || !dd) { + ctx.skip(); + return; + } + // A fresh mock so this test only sees its own request, and the wire-shape + // test's request can't bleed into the content assertions. + const ddFull = await startMockDatadog(); + const ddMeta = await startMockDatadog(); + const app = await spawnApp({ + extraEnv: { + [`DD_CRED_${CREDENTIAL_REF.toUpperCase()}_API_KEY`]: DD_API_KEY, + }, + }); + apps.push(app); + try { + const admin = new AdminClient(app.adminUrl, app.adminKey); + // Two exporters on the same DP: one captures content, one does not. + await admin.createObservabilityExporter({ + name: "datadog-full", + enabled: true, + kind: "datadog", + site: ddFull.site, + credential_ref: CREDENTIAL_REF, + service: DD_SERVICE, + content_mode: "full", + }); + await admin.createObservabilityExporter({ + name: "datadog-meta", + enabled: true, + kind: "datadog", + site: ddMeta.site, + credential_ref: CREDENTIAL_REF, + service: DD_SERVICE, + content_mode: "metadata_only", + }); + await seedRouting(admin, upstream, "datadog-content-model"); + + await waitConfigPropagation(async () => { + try { + const r = await chat( + app, + "datadog-content-model", + `please remember the token ${PROMPT_TOKEN}`, + ); + await r.text(); + return r.status === 200; + } catch { + return false; + } + }); + + const res = await chat( + app, + "datadog-content-model", + `please remember the token ${PROMPT_TOKEN}`, + ); + expect(res.status).toBe(200); + await res.text(); + + const fullIntake = await waitForIntake(ddFull); + const metaIntake = await waitForIntake(ddMeta); + + const fullLog = asRecord(fullIntake.logs[0]); + // Full capture carries both the prompt and the assembled response. + expect(fullLog["gen_ai.prompt"]).toBeTypeOf("string"); + expect(fullLog["gen_ai.completion"]).toBeTypeOf("string"); + expect(fullIntake.bodyText).toContain(PROMPT_TOKEN); + expect(fullIntake.bodyText).toContain(RESPONSE_TOKEN); + + // Metadata-only carries NEITHER the prompt nor the response — content + // gating holds end-to-end through the real binary. + const metaLog = asRecord(metaIntake.logs[0]); + expect(metaLog["gen_ai.prompt"]).toBeUndefined(); + expect(metaLog["gen_ai.completion"]).toBeUndefined(); + expect(metaIntake.bodyText).not.toContain(PROMPT_TOKEN); + expect(metaIntake.bodyText).not.toContain(RESPONSE_TOKEN); + } finally { + await ddFull.close(); + await ddMeta.close(); + } + }, + 60_000, + ); +});