diff --git a/apps/ingest/Cargo.lock b/apps/ingest/Cargo.lock index 615573810..a11dfc44a 100644 --- a/apps/ingest/Cargo.lock +++ b/apps/ingest/Cargo.lock @@ -1590,6 +1590,7 @@ dependencies = [ "opentelemetry_sdk", "prost", "reqwest", + "ring", "rustls", "serde", "serde_json", diff --git a/apps/ingest/Cargo.toml b/apps/ingest/Cargo.toml index a04dc0151..0157dd8e7 100644 --- a/apps/ingest/Cargo.toml +++ b/apps/ingest/Cargo.toml @@ -60,6 +60,9 @@ tokio-postgres = "0.7" deadpool-postgres = "0.14" tokio-postgres-rustls = "0.13" rustls = { version = "0.23", features = ["ring"] } +# RS256 signatures of the ID tokens Pub/Sub signs its GCP pushes with. Already +# in the tree through rustls. +ring = "0.17" webpki-roots = "0.26" # Opt-in profiler (https://hotpath.rs). Every macro is a no-op and no profiler # code is compiled unless `--features hotpath` is passed; the `tokio` and diff --git a/apps/ingest/alchemy.run.ts b/apps/ingest/alchemy.run.ts index d9741f2da..f26145834 100644 --- a/apps/ingest/alchemy.run.ts +++ b/apps/ingest/alchemy.run.ts @@ -616,6 +616,7 @@ export const createMapleIngest = ({ stage, region, domains, profile, dbLogin }: ...(yield* optionalPlain("INGEST_BATCH_MAX_WAIT_MS")), ...(yield* optionalPlain("INGEST_ORG_QUEUE_MAX_BYTES")), ...(yield* optionalPlain("INGEST_ORG_MAX_IN_FLIGHT")), + ...(yield* optionalPlain("INGEST_GCP_ORG_MAX_IN_FLIGHT")), ...(yield* optionalPlain("INGEST_REQUEST_TIMEOUT_SECS")), ...(yield* optionalPlain("INGEST_QUEUE_MAX_AGE_SECS")), ...(yield* optionalPlain("INGEST_MAX_REQUEST_BODY_BYTES")), diff --git a/apps/ingest/src/gcp_logging.rs b/apps/ingest/src/gcp_logging.rs new file mode 100644 index 000000000..d3883aae3 --- /dev/null +++ b/apps/ingest/src/gcp_logging.rs @@ -0,0 +1,1051 @@ +//! Cloud Logging `LogEntry` (delivered by an unwrapped Pub/Sub push, one entry +//! per request) to an OTLP logs request. + +use chrono::DateTime; +use opentelemetry_proto::tonic::collector::logs::v1::ExportLogsServiceRequest; +use opentelemetry_proto::tonic::common::v1::{ + any_value, AnyValue, ArrayValue, InstrumentationScope, KeyValue, +}; +use opentelemetry_proto::tonic::logs::v1::{LogRecord, ResourceLogs, ScopeLogs}; +use opentelemetry_proto::tonic::resource::v1::Resource; +use serde_json::{Map as JsonMap, Value as JsonValue}; + +pub const INGEST_SOURCE: &str = "gcp-logpush"; + +// What a connector's `last_error` holds. The customer reads it on the +// connection: one sentence of cause, one of what to do. + +/// A push body that is not a `LogEntry`. +pub const NOT_A_LOG_ENTRY: &str = "The Pub/Sub subscription wraps each entry in an envelope \ + Maple can't read. Run the setup script again: it resets the subscription. Entries sent \ + meanwhile are lost."; + +/// An entry Maple accepted but could not hand to the pipeline. +pub const ENTRY_NOT_STORED: &str = "Maple could not store an entry just now. Pub/Sub retries it \ + for up to a day. Nothing to do."; + +/// An entry refused because the organization is over its plan limit. +pub const OVER_PLAN_LIMIT: &str = "This Maple organization is over its plan limit, so Maple \ + refuses new logs. Pub/Sub retries for up to a day. See Settings, Billing."; + +/// The log a setup script names in the entry that reports what it applied. +const SETUP_REPORT_LOG: &str = "/logs/maple-setup"; + +/// What a run of the setup or cleanup script applied in Google Cloud. `None` +/// for a capability the reporting section did not touch. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct SetupReport { + pub logs: Option, + pub metrics: Option, +} + +type JsonObject = JsonMap; + +/// The connector a push authenticated as. +#[derive(Debug)] +pub struct Connector<'a> { + pub id: &'a str, + pub org_id: &'a str, +} + +/// A `LogEntry` field, the semantic-convention attribute it becomes, and that +/// attribute's value. `None` when the field does not fit the attribute's type. +type Rename = ( + &'static str, + &'static str, + fn(&JsonValue) -> Option, +); + +const HTTP_REQUEST_ATTRIBUTES: &[Rename] = &[ + ("requestMethod", "http.request.method", string_value), + ("requestUrl", "url.full", string_value), + ("status", "http.response.status_code", int_value), + ("requestSize", "http.request.size", int_value), + ("responseSize", "http.response.size", int_value), + ("userAgent", "user_agent.original", string_value), + ("remoteIp", "client.address", string_value), + ("serverIp", "server.address", string_value), + // Header attributes are arrays of the header's values. + ("referer", "http.request.header.referer", string_array_value), +]; + +const SOURCE_LOCATION_ATTRIBUTES: &[Rename] = &[ + ("file", "code.file.path", string_value), + ("line", "code.line.number", int_value), + ("function", "code.function.name", string_value), +]; + +/// `google.rpc.Code` names, indexed by code. +const RPC_CODES: [&str; 17] = [ + "OK", + "CANCELLED", + "UNKNOWN", + "INVALID_ARGUMENT", + "DEADLINE_EXCEEDED", + "NOT_FOUND", + "ALREADY_EXISTS", + "PERMISSION_DENIED", + "RESOURCE_EXHAUSTED", + "FAILED_PRECONDITION", + "ABORTED", + "OUT_OF_RANGE", + "UNIMPLEMENTED", + "INTERNAL", + "UNAVAILABLE", + "DATA_LOSS", + "UNAUTHENTICATED", +]; + +/// Every `LogEntry` carries `logName`. Anything without it (not JSON, or the +/// wrapped Pub/Sub envelope) is not one and will never become one on retry. +pub fn parse_log_entry(payload: &[u8]) -> Option { + let entry: JsonObject = serde_json::from_slice(payload).ok()?; + let is_log_entry = entry.get("logName").is_some_and(JsonValue::is_string); + is_log_entry.then_some(entry) +} + +/// A setup report is a `LogEntry` on the `maple-setup` log whose payload says +/// what the run applied. The script publishes it through the sink's topic, or +/// posts it here when there is no topic. +pub fn parse_setup_report(payload: &[u8]) -> Option { + // Asked of every pushed log line, so the bytes are searched before + // anything is parsed. + let marker = SETUP_REPORT_LOG.as_bytes(); + if !payload.windows(marker.len()).any(|window| window == marker) { + return None; + } + let entry = parse_log_entry(payload)?; + if !text(&entry, "logName")?.ends_with(SETUP_REPORT_LOG) { + return None; + } + // Cloud Logging stamps every entry it routes. The script's report never + // passes through it, so a stamped one was written to the log by anyone in + // the scope who may write logs, and is only a log line. + if entry.contains_key("receiveTimestamp") { + return None; + } + let applied = object(&entry, "jsonPayload")?; + let flag = |key: &str| applied.get(key).and_then(JsonValue::as_bool); + let report = SetupReport { + logs: flag("logs"), + metrics: flag("metrics"), + }; + (report.logs.is_some() || report.metrics.is_some()).then_some(report) +} + +pub fn build_logs_request( + entry: &JsonObject, + connector: &Connector<'_>, + now_unix_nano: u64, +) -> ExportLogsServiceRequest { + ExportLogsServiceRequest { + resource_logs: vec![ResourceLogs { + resource: Some(Resource { + attributes: resource_attributes(entry, connector), + ..Default::default() + }), + scope_logs: vec![ScopeLogs { + scope: Some(InstrumentationScope { + name: "gcp.logging".to_owned(), + ..Default::default() + }), + log_records: vec![log_record(entry, now_unix_nano)], + schema_url: String::new(), + }], + schema_url: String::new(), + }], + } +} + +fn resource_attributes(entry: &JsonObject, connector: &Connector<'_>) -> Vec { + let resource = object(entry, "resource"); + let resource_type = resource + .and_then(|resource| text(resource, "type")) + .unwrap_or("unknown"); + let labels = resource.and_then(|resource| object(resource, "labels")); + let label = |key: &str| labels.and_then(|labels| text(labels, key)); + let entry_labels = object(entry, "labels"); + let entry_label = |key: &str| entry_labels.and_then(|labels| text(labels, key)); + + let mut attributes = vec![ + attribute("maple_org_id", connector.org_id), + attribute("maple_ingest_source", INGEST_SOURCE), + attribute("maple_ingest_key_type", "connector"), + attribute("maple_gcp_connector_id", connector.id), + attribute("cloud.provider", "gcp"), + attribute("gcp.resource.type", resource_type), + ]; + // An aggregated sink delivers entries from many projects, so the project + // is the entry's own. Entries logged against an organization, folder or + // billing account have none. + let project = label("project_id").or_else(|| { + let (project, _) = text(entry, "logName")? + .strip_prefix("projects/")? + .split_once("/logs/")?; + Some(project) + }); + push(&mut attributes, "cloud.account.id", project); + push( + &mut attributes, + "cloud.platform", + cloud_platform(resource_type), + ); + + // `location` names a zone for zonal resources and a region otherwise. + let location = label("zone") + .or_else(|| label("location")) + .or_else(|| label("region")); + match location.and_then(|location| location.rsplit_once('-')) { + Some((region, suffix)) if matches!(suffix.as_bytes(), [b'a'..=b'z']) => { + push(&mut attributes, "cloud.region", Some(region)); + push(&mut attributes, "cloud.availability_zone", location); + } + _ => push(&mut attributes, "cloud.region", location), + } + + // Compute workloads take the bare workload name so their logs join the + // traced service of the same name; everything else is `gcp/`. + let workload = match resource_type { + "cloud_run_revision" => { + // A revision's own system events carry an empty `service_name`. A service's + // configuration has the service's name. + let service = label("service_name").or_else(|| label("configuration_name")); + push(&mut attributes, "faas.name", service); + push(&mut attributes, "faas.version", label("revision_name")); + push(&mut attributes, "faas.instance", entry_label("instanceId")); + service + } + "cloud_run_job" => { + push( + &mut attributes, + "gcp.cloud_run.job.execution", + entry_label("run.googleapis.com/execution_name"), + ); + attributes.extend( + entry_label("run.googleapis.com/task_index") + .and_then(|index| index.parse().ok()) + .map(|index| { + key_value( + "gcp.cloud_run.job.task_index", + any_value::Value::IntValue(index), + ) + }), + ); + label("job_name") + } + "cloud_function" => { + push(&mut attributes, "faas.name", label("function_name")); + label("function_name") + } + "k8s_container" => { + push(&mut attributes, "k8s.cluster.name", label("cluster_name")); + push( + &mut attributes, + "k8s.namespace.name", + label("namespace_name"), + ); + push(&mut attributes, "k8s.pod.name", label("pod_name")); + push( + &mut attributes, + "k8s.container.name", + label("container_name"), + ); + label("container_name") + } + "gce_instance" => { + let name = entry_label("compute.googleapis.com/resource_name"); + push(&mut attributes, "host.id", label("instance_id")); + push(&mut attributes, "host.name", name); + push(&mut attributes, "gcp.gce.instance.name", name); + name + } + "gae_app" => label("module_id"), + _ => None, + }; + let service_name = workload.map_or_else(|| format!("gcp/{resource_type}"), str::to_owned); + attributes.push(attribute("service.name", &service_name)); + + if let Some(labels) = labels { + push_fields(&mut attributes, labels, &[], "gcp.resource.labels."); + } + attributes +} + +fn cloud_platform(resource_type: &str) -> Option<&'static str> { + match resource_type { + "cloud_run_revision" | "cloud_run_job" => Some("gcp_cloud_run"), + "cloud_function" => Some("gcp_cloud_functions"), + "gce_instance" => Some("gcp_compute_engine"), + "gae_app" => Some("gcp_app_engine"), + other if other.starts_with("k8s_") => Some("gcp_kubernetes_engine"), + _ => None, + } +} + +fn log_record(entry: &JsonObject, now_unix_nano: u64) -> LogRecord { + let timestamp = |key: &str| text(entry, key).and_then(parse_timestamp); + let observed = timestamp("receiveTimestamp").unwrap_or(now_unix_nano); + let (severity_text, severity_number) = severity(text(entry, "severity").unwrap_or_default()); + + // Payload fields first, so the entry's own metadata wins a key collision. + let mut attributes = Vec::new(); + let body = body(entry, &mut attributes); + if let Some(request) = object(entry, "httpRequest") { + push_fields( + &mut attributes, + request, + HTTP_REQUEST_ATTRIBUTES, + "gcp.http_request.", + ); + } + if let Some(location) = object(entry, "sourceLocation") { + push_fields( + &mut attributes, + location, + SOURCE_LOCATION_ATTRIBUTES, + "gcp.source_location.", + ); + } + if let Some(labels) = object(entry, "labels") { + push_fields(&mut attributes, labels, &[], "gcp.labels."); + } + push(&mut attributes, "log.record.uid", text(entry, "insertId")); + push(&mut attributes, "gcp.log_name", text(entry, "logName")); + + LogRecord { + time_unix_nano: timestamp("timestamp").unwrap_or(observed), + observed_time_unix_nano: observed, + severity_number, + severity_text: severity_text.to_owned(), + body: Some(AnyValue { + value: Some(any_value::Value::StringValue(body)), + }), + attributes, + flags: u32::from(entry.get("traceSampled") == Some(&JsonValue::Bool(true))), + // `trace` is `projects//traces/<32 hex>`, or the bare id. + trace_id: text(entry, "trace").map_or_else(Vec::new, |trace| { + hex_bytes(trace.rsplit_once('/').map_or(trace, |(_, id)| id), 16) + }), + span_id: text(entry, "spanId").map_or_else(Vec::new, |span_id| hex_bytes(span_id, 8)), + ..Default::default() + } +} + +/// The body of whichever payload the entry carries; the payload's structured +/// fields are pushed onto `attributes`. +fn body(entry: &JsonObject, attributes: &mut Vec) -> String { + if let Some(text_payload) = text(entry, "textPayload") { + return text_payload.to_owned(); + } + + if let Some(payload) = object(entry, "jsonPayload") { + let message = ["message", "msg"] + .into_iter() + .find_map(|key| Some((key, text(payload, key)?))); + attributes.extend( + payload + .iter() + .filter(|(key, _)| { + message.is_none_or(|(message_key, _)| message_key != key.as_str()) + }) + .filter_map(|(key, value)| json_attribute(key, value)), + ); + return message.map_or_else(|| compact_json(payload), |(_, message)| message.to_owned()); + } + + // Audit logs. + if let Some(payload) = object(entry, "protoPayload") { + let method = text(payload, "methodName"); + let status = object(payload, "status"); + push( + attributes, + "gcp.audit.service.name", + text(payload, "serviceName"), + ); + push(attributes, "gcp.audit.method.name", method); + push( + attributes, + "gcp.audit.resource.name", + text(payload, "resourceName"), + ); + push( + attributes, + "user.email", + object(payload, "authenticationInfo").and_then(|info| text(info, "principalEmail")), + ); + if let Some(code) = status + .and_then(|status| status.get("code")) + .and_then(JsonValue::as_i64) + { + let key = "rpc.response.status_code"; + let name = usize::try_from(code) + .ok() + .and_then(|index| RPC_CODES.get(index)); + attributes.push(match name { + Some(name) => attribute(key, name), + None => attribute(key, &code.to_string()), + }); + } + return match (method, status.and_then(|status| text(status, "message"))) { + (Some(method), Some(message)) => format!("{method}: {message}"), + (Some(method), None) => method.to_owned(), + (None, _) => compact_json(payload), + }; + } + + // Request logs (Cloud Run, load balancers) carry only `httpRequest`. + object(entry, "httpRequest").map_or_else(String::new, |request| { + format!( + "{} {} -> {}", + text(request, "requestMethod").unwrap_or("UNKNOWN"), + text(request, "requestUrl").unwrap_or("-"), + request + .get("status") + .and_then(JsonValue::as_u64) + .unwrap_or(0), + ) + }) +} + +/// Cloud Logging severity to the OTel severity name and number. The numbers +/// are the OpenTelemetry Collector's Cloud Logging mapping; `DEFAULT` is +/// unspecified. +fn severity(name: &str) -> (&'static str, i32) { + match name { + "DEBUG" => ("DEBUG", 5), + "INFO" => ("INFO", 9), + "NOTICE" => ("INFO", 10), + "WARNING" => ("WARN", 13), + "ERROR" => ("ERROR", 17), + "CRITICAL" => ("FATAL", 21), + "ALERT" => ("FATAL", 22), + "EMERGENCY" => ("FATAL", 24), + _ => ("", 0), + } +} + +fn parse_timestamp(value: &str) -> Option { + let nanos = DateTime::parse_from_rfc3339(value) + .ok()? + .timestamp_nanos_opt()?; + u64::try_from(nanos).ok() +} + +/// `len` bytes from exactly `2 * len` hex digits, empty for anything else. +fn hex_bytes(hex: &str, len: usize) -> Vec { + if hex.len() != len * 2 || !hex.bytes().all(|byte| byte.is_ascii_hexdigit()) { + return Vec::new(); + } + (0..len) + .filter_map(|index| u8::from_str_radix(&hex[index * 2..index * 2 + 2], 16).ok()) + .collect() +} + +fn object<'a>(parent: &'a JsonObject, key: &str) -> Option<&'a JsonObject> { + parent.get(key)?.as_object() +} + +fn text<'a>(parent: &'a JsonObject, key: &str) -> Option<&'a str> { + parent.get(key)?.as_str().filter(|value| !value.is_empty()) +} + +fn compact_json(object: &JsonObject) -> String { + serde_json::to_string(object).unwrap_or_default() +} + +fn key_value(key: &str, value: any_value::Value) -> KeyValue { + KeyValue { + key: key.to_owned(), + key_strindex: 0, + value: Some(AnyValue { value: Some(value) }), + } +} + +fn attribute(key: &str, value: &str) -> KeyValue { + key_value(key, any_value::Value::StringValue(value.to_owned())) +} + +fn json_attribute(key: &str, value: &JsonValue) -> Option { + Some(key_value(key, string_value(value)?)) +} + +/// Scalars as strings, nested values as compact JSON; null is dropped. +fn string_value(value: &JsonValue) -> Option { + match value { + JsonValue::Null => None, + JsonValue::String(value) => Some(any_value::Value::StringValue(value.clone())), + other => Some(any_value::Value::StringValue(other.to_string())), + } +} + +/// Proto3 JSON sends 64-bit integers as strings, smaller ones as numbers. +fn int_value(value: &JsonValue) -> Option { + let int = value.as_i64().or_else(|| value.as_str()?.parse().ok())?; + Some(any_value::Value::IntValue(int)) +} + +fn string_array_value(value: &JsonValue) -> Option { + let value = any_value::Value::StringValue(value.as_str()?.to_owned()); + Some(any_value::Value::ArrayValue(ArrayValue { + values: vec![AnyValue { value: Some(value) }], + })) +} + +fn push(attributes: &mut Vec, key: &str, value: Option<&str>) { + if let Some(value) = value { + attributes.push(attribute(key, value)); + } +} + +/// Every field of `fields` as an attribute: under its `renames` key when it +/// has one and its value fits that attribute, under `prefix` + its own name +/// otherwise. +fn push_fields( + attributes: &mut Vec, + fields: &JsonObject, + renames: &[Rename], + prefix: &str, +) { + for (field, value) in fields { + let renamed = renames + .iter() + .find(|(name, ..)| *name == field.as_str()) + .and_then(|(_, key, convert)| Some(key_value(key, convert(value)?))); + attributes.extend(renamed.or_else(|| json_attribute(&format!("{prefix}{field}"), value))); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + use std::collections::HashMap; + + const NOW: u64 = 1_800_000_000_000_000_000; + const CONNECTOR: Connector<'static> = Connector { + id: "gcp_conn_1", + org_id: "org_1", + }; + + /// The resource's attributes and the record. + fn convert_typed(entry: JsonValue) -> (Vec, LogRecord) { + let entry = parse_log_entry(entry.to_string().as_bytes()).expect("a LogEntry"); + let mut request = build_logs_request(&entry, &CONNECTOR, NOW); + let mut resource_logs = request.resource_logs.remove(0); + let record = resource_logs.scope_logs.remove(0).log_records.remove(0); + (resource_logs.resource.unwrap().attributes, record) + } + + fn convert(entry: JsonValue) -> (HashMap, LogRecord) { + let (resource, record) = convert_typed(entry); + (strings(&resource), record) + } + + fn value_of<'a>(attributes: &'a [KeyValue], key: &str) -> Option<&'a any_value::Value> { + attributes + .iter() + .find(|attribute| attribute.key == key)? + .value + .as_ref()? + .value + .as_ref() + } + + fn strings(attributes: &[KeyValue]) -> HashMap { + attributes + .iter() + .filter_map( + |attribute| match attribute.value.as_ref()?.value.as_ref()? { + any_value::Value::StringValue(value) => { + Some((attribute.key.clone(), value.clone())) + } + _ => None, + }, + ) + .collect() + } + + fn body_of(record: &LogRecord) -> &str { + match record.body.as_ref().and_then(|body| body.value.as_ref()) { + Some(any_value::Value::StringValue(value)) => value, + _ => panic!("expected a string body"), + } + } + + #[test] + fn cloud_run_revision_event_without_a_service_name_is_named_by_its_configuration() { + let (resource, _) = convert(json!({ + "logName": "projects/my-project/logs/cloudaudit.googleapis.com%2Fsystem_event", + "resource": { + "type": "cloud_run_revision", + "labels": { + "project_id": "my-project", + "service_name": "", + "revision_name": "checkout-00042-abc", + "location": "europe-west4", + "configuration_name": "checkout" + } + } + })); + + assert_eq!(resource["service.name"], "checkout"); + assert_eq!(resource["faas.name"], "checkout"); + } + + #[test] + fn cloud_run_entry_maps_resource_trace_time_and_severity() { + let (resource, record) = convert(json!({ + "logName": "projects/my-project/logs/run.googleapis.com%2Fstdout", + "insertId": "insert-1", + "timestamp": "2025-03-07T12:34:56.123456789Z", + "receiveTimestamp": "2025-03-07T12:34:57Z", + "severity": "WARNING", + "textPayload": "cache miss", + "trace": "projects/my-project/traces/0af7651916cd43dd8448eb211c80319c", + "spanId": "b7ad6b7169203331", + "traceSampled": true, + "labels": { "instanceId": "abc" }, + "sourceLocation": { "file": "main.go", "line": "42", "function": "main.handle" }, + "resource": { + "type": "cloud_run_revision", + "labels": { + "project_id": "my-project", + "service_name": "checkout", + "revision_name": "checkout-00042-abc", + "location": "europe-west4", + "configuration_name": "checkout" + } + } + })); + + assert_eq!(resource["service.name"], "checkout"); + assert_eq!(resource["faas.name"], "checkout"); + assert_eq!(resource["faas.version"], "checkout-00042-abc"); + assert_eq!(resource["faas.instance"], "abc"); + assert_eq!(resource["cloud.provider"], "gcp"); + assert_eq!(resource["cloud.platform"], "gcp_cloud_run"); + assert_eq!(resource["cloud.account.id"], "my-project"); + assert_eq!(resource["cloud.region"], "europe-west4"); + assert!(!resource.contains_key("cloud.availability_zone")); + assert_eq!(resource["gcp.resource.type"], "cloud_run_revision"); + assert_eq!( + resource["gcp.resource.labels.configuration_name"], + "checkout" + ); + assert_eq!(resource["maple_org_id"], "org_1"); + assert_eq!(resource["maple_ingest_source"], INGEST_SOURCE); + assert_eq!(resource["maple_ingest_key_type"], "connector"); + assert_eq!(resource["maple_gcp_connector_id"], "gcp_conn_1"); + + assert_eq!(body_of(&record), "cache miss"); + assert_eq!(record.time_unix_nano, 1_741_350_896_123_456_789); + assert_eq!(record.observed_time_unix_nano, 1_741_350_897_000_000_000); + assert_eq!( + (record.severity_text.as_str(), record.severity_number), + ("WARN", 13) + ); + assert_eq!( + record.trace_id, + [ + 0x0a, 0xf7, 0x65, 0x19, 0x16, 0xcd, 0x43, 0xdd, 0x84, 0x48, 0xeb, 0x21, 0x1c, 0x80, + 0x31, 0x9c + ] + ); + assert_eq!( + record.span_id, + [0xb7, 0xad, 0x6b, 0x71, 0x69, 0x20, 0x33, 0x31] + ); + assert_eq!(record.flags, 1); + + let attributes = strings(&record.attributes); + assert_eq!(attributes["log.record.uid"], "insert-1"); + assert_eq!( + attributes["gcp.log_name"], + "projects/my-project/logs/run.googleapis.com%2Fstdout" + ); + assert_eq!(attributes["gcp.labels.instanceId"], "abc"); + assert_eq!(attributes["code.file.path"], "main.go"); + assert_eq!( + value_of(&record.attributes, "code.line.number"), + Some(&any_value::Value::IntValue(42)) + ); + assert_eq!(attributes["code.function.name"], "main.handle"); + } + + #[test] + fn cloud_run_job_entry_maps_execution_and_task_index() { + let (resource, _) = convert_typed(json!({ + "logName": "projects/my-project/logs/run.googleapis.com%2Fstdout", + "textPayload": "processing batch", + "labels": { + "instanceId": "abc", + "run.googleapis.com/execution_name": "nightly-export-x7k2p", + "run.googleapis.com/task_attempt": "0", + "run.googleapis.com/task_index": "3" + }, + "resource": { + "type": "cloud_run_job", + "labels": { + "project_id": "my-project", + "job_name": "nightly-export", + "location": "europe-west4" + } + } + })); + + let strings = strings(&resource); + assert_eq!(strings["service.name"], "nightly-export"); + assert_eq!(strings["cloud.platform"], "gcp_cloud_run"); + assert_eq!( + strings["gcp.cloud_run.job.execution"], + "nightly-export-x7k2p" + ); + assert_eq!( + value_of(&resource, "gcp.cloud_run.job.task_index"), + Some(&any_value::Value::IntValue(3)) + ); + assert!(!strings.contains_key("faas.instance")); + } + + #[test] + fn gke_container_json_payload_becomes_body_and_attributes() { + let (resource, record) = convert(json!({ + "logName": "projects/my-project/logs/stdout", + "severity": "ERROR", + "jsonPayload": { + "message": "payment failed", + "order_id": 1234, + "retry": false, + "customer": { "tier": "pro" }, + "ignored": null + }, + "resource": { + "type": "k8s_container", + "labels": { + "project_id": "my-project", + "location": "us-central1-a", + "cluster_name": "prod", + "namespace_name": "payments", + "pod_name": "api-7d9f-xk2", + "container_name": "api" + } + } + })); + + assert_eq!(resource["service.name"], "api"); + assert_eq!(resource["cloud.platform"], "gcp_kubernetes_engine"); + assert_eq!(resource["cloud.region"], "us-central1"); + assert_eq!(resource["cloud.availability_zone"], "us-central1-a"); + assert_eq!(resource["k8s.cluster.name"], "prod"); + assert_eq!(resource["k8s.namespace.name"], "payments"); + assert_eq!(resource["k8s.pod.name"], "api-7d9f-xk2"); + assert_eq!(resource["k8s.container.name"], "api"); + + assert_eq!(body_of(&record), "payment failed"); + assert_eq!( + (record.severity_text.as_str(), record.severity_number), + ("ERROR", 17) + ); + let attributes = strings(&record.attributes); + assert_eq!(attributes["order_id"], "1234"); + assert_eq!(attributes["retry"], "false"); + assert_eq!(attributes["customer"], r#"{"tier":"pro"}"#); + assert!(!attributes.contains_key("message")); + assert!(!attributes.contains_key("ignored")); + } + + #[test] + fn json_payload_without_a_message_is_the_body_as_compact_json() { + let (_, record) = convert(json!({ + "logName": "projects/my-project/logs/stdout", + "jsonPayload": { "event": "tick" } + })); + + assert_eq!(body_of(&record), r#"{"event":"tick"}"#); + assert_eq!(strings(&record.attributes)["event"], "tick"); + } + + #[test] + fn gce_instance_is_named_by_its_resource_name_label() { + let entry = |labels: JsonValue| { + json!({ + "logName": "projects/my-project/logs/syslog", + "textPayload": "started", + "labels": labels, + "resource": { + "type": "gce_instance", + "labels": { + "project_id": "my-project", + "instance_id": "8123456789", + "zone": "europe-west1-b" + } + } + }) + }; + + let (named, _) = convert(entry( + json!({ "compute.googleapis.com/resource_name": "web-1" }), + )); + assert_eq!(named["service.name"], "web-1"); + assert_eq!(named["host.name"], "web-1"); + assert_eq!(named["gcp.gce.instance.name"], "web-1"); + assert_eq!(named["host.id"], "8123456789"); + assert_eq!(named["cloud.platform"], "gcp_compute_engine"); + assert_eq!(named["cloud.region"], "europe-west1"); + assert_eq!(named["cloud.availability_zone"], "europe-west1-b"); + + let (unnamed, _) = convert(entry(json!({}))); + assert_eq!(unnamed["service.name"], "gcp/gce_instance"); + assert_eq!(unnamed["host.id"], "8123456789"); + assert!(!unnamed.contains_key("host.name")); + assert!(!unnamed.contains_key("gcp.gce.instance.name")); + } + + #[test] + fn audit_log_proto_payload_maps_method_status_and_principal() { + let (resource, record) = convert(json!({ + "logName": "projects/my-project/logs/cloudaudit.googleapis.com%2Factivity", + "severity": "NOTICE", + "protoPayload": { + "@type": "type.googleapis.com/google.cloud.audit.AuditLog", + "serviceName": "run.googleapis.com", + "methodName": "google.cloud.run.v1.Services.ReplaceService", + "resourceName": "namespaces/my-project/services/checkout", + "authenticationInfo": { "principalEmail": "deployer@example.com" }, + "status": { "code": 7, "message": "Permission denied" } + }, + "resource": { + "type": "audited_resource", + "labels": { "project_id": "my-project", "service": "run.googleapis.com" } + } + })); + + assert_eq!(resource["service.name"], "gcp/audited_resource"); + assert!(!resource.contains_key("cloud.platform")); + assert_eq!( + body_of(&record), + "google.cloud.run.v1.Services.ReplaceService: Permission denied" + ); + assert_eq!( + (record.severity_text.as_str(), record.severity_number), + ("INFO", 10) + ); + let attributes = strings(&record.attributes); + assert_eq!(attributes["gcp.audit.service.name"], "run.googleapis.com"); + assert_eq!( + attributes["gcp.audit.method.name"], + "google.cloud.run.v1.Services.ReplaceService" + ); + assert_eq!( + attributes["gcp.audit.resource.name"], + "namespaces/my-project/services/checkout" + ); + assert_eq!(attributes["user.email"], "deployer@example.com"); + assert_eq!(attributes["rpc.response.status_code"], "PERMISSION_DENIED"); + } + + #[test] + fn audit_status_code_is_the_google_rpc_code_name() { + let status_code = |code: JsonValue| { + let (_, record) = convert(json!({ + "logName": "projects/my-project/logs/cloudaudit.googleapis.com%2Factivity", + "protoPayload": { "methodName": "Get", "status": { "code": code } } + })); + strings(&record.attributes) + .get("rpc.response.status_code") + .cloned() + }; + + assert_eq!(status_code(json!(0)).as_deref(), Some("OK")); + assert_eq!(status_code(json!(4)).as_deref(), Some("DEADLINE_EXCEEDED")); + assert_eq!(status_code(json!(16)).as_deref(), Some("UNAUTHENTICATED")); + assert_eq!(status_code(json!(17)).as_deref(), Some("17")); + assert_eq!(status_code(json!(-1)).as_deref(), Some("-1")); + } + + #[test] + fn managed_service_request_log_gets_prefixed_service_and_http_attributes() { + let (resource, record) = convert(json!({ + "logName": "projects/my-project/logs/requests", + "severity": "INFO", + "httpRequest": { + "requestMethod": "GET", + "requestUrl": "https://example.com/health", + "status": 502, + "requestSize": "a lot", + "responseSize": "512", + "referer": "https://example.com/", + "userAgent": "curl/8.0", + "remoteIp": "203.0.113.7", + "latency": "0.250s" + }, + "resource": { + "type": "http_load_balancer", + "labels": { + "project_id": "my-project", + "forwarding_rule_name": "edge", + "zone": "global" + } + } + })); + + assert_eq!(resource["service.name"], "gcp/http_load_balancer"); + assert_eq!(resource["cloud.region"], "global"); + assert_eq!(resource["gcp.resource.labels.forwarding_rule_name"], "edge"); + + assert_eq!(body_of(&record), "GET https://example.com/health -> 502"); + let attributes = strings(&record.attributes); + assert_eq!(attributes["http.request.method"], "GET"); + assert_eq!(attributes["url.full"], "https://example.com/health"); + let typed = |key: &str| value_of(&record.attributes, key); + assert_eq!( + typed("http.response.status_code"), + Some(&any_value::Value::IntValue(502)) + ); + assert_eq!( + typed("http.response.size"), + Some(&any_value::Value::IntValue(512)) + ); + assert_eq!( + typed("http.request.header.referer"), + Some(&any_value::Value::ArrayValue(ArrayValue { + values: vec![AnyValue { + value: Some(any_value::Value::StringValue( + "https://example.com/".to_owned() + )), + }], + })) + ); + // Not an integer, so it keeps Cloud Logging's field name. + assert!(typed("http.request.size").is_none()); + assert_eq!(attributes["gcp.http_request.requestSize"], "a lot"); + assert_eq!(attributes["user_agent.original"], "curl/8.0"); + assert_eq!(attributes["client.address"], "203.0.113.7"); + assert_eq!(attributes["gcp.http_request.latency"], "0.250s"); + } + + #[test] + fn malformed_trace_and_timestamps_are_left_empty() { + let (resource, record) = convert(json!({ + "logName": "projects/my-project/logs/stdout", + "timestamp": "yesterday", + "severity": "DEFAULT", + "textPayload": "hello", + "trace": "projects/my-project/traces/not-a-trace-id", + "spanId": "+7ad6b7169203331", + "traceSampled": "yes" + })); + + assert_eq!(record.time_unix_nano, NOW); + assert_eq!(record.observed_time_unix_nano, NOW); + assert!(record.trace_id.is_empty()); + assert!(record.span_id.is_empty()); + assert_eq!(record.flags, 0); + assert_eq!( + (record.severity_text.as_str(), record.severity_number), + ("", 0) + ); + // No `resource` at all: the project comes from `logName`. + assert_eq!(resource["cloud.account.id"], "my-project"); + assert_eq!(resource["service.name"], "gcp/unknown"); + } + + #[test] + fn cloud_account_id_is_the_entry_project_when_it_has_one() { + // A child project's entry, delivered through an aggregated sink. + let (child, _) = convert(json!({ + "logName": "projects/child-project/logs/stdout", + "textPayload": "hello", + "resource": { + "type": "cloud_run_revision", + "labels": { "project_id": "child-project", "service_name": "api" } + } + })); + assert_eq!(child["cloud.account.id"], "child-project"); + + let (log_name_only, _) = convert(json!({ + "logName": "projects/other-project/logs/cloudaudit.googleapis.com%2Factivity", + "textPayload": "hello", + "resource": { "type": "global", "labels": {} } + })); + assert_eq!(log_name_only["cloud.account.id"], "other-project"); + + let (organization, record) = convert(json!({ + "logName": "organizations/123456789/logs/cloudaudit.googleapis.com%2Factivity", + "severity": "NOTICE", + "protoPayload": { + "@type": "type.googleapis.com/google.cloud.audit.AuditLog", + "serviceName": "cloudresourcemanager.googleapis.com", + "methodName": "SetIamPolicy" + }, + "resource": { + "type": "organization", + "labels": { "organization_id": "123456789" } + } + })); + assert!(!organization.contains_key("cloud.account.id")); + assert_eq!(organization["service.name"], "gcp/organization"); + assert_eq!( + organization["gcp.resource.labels.organization_id"], + "123456789" + ); + assert_eq!(body_of(&record), "SetIamPolicy"); + } + + #[test] + fn only_a_log_entry_parses() { + assert!(parse_log_entry(br#"{"logName":"projects/p/logs/l"}"#).is_some()); + assert!(parse_log_entry(b"not json").is_none()); + assert!(parse_log_entry(b"[]").is_none()); + // The wrapped Pub/Sub envelope, which this receiver does not accept. + assert!(parse_log_entry( + br#"{"message":{"data":"e30=","messageId":"1"},"subscription":"projects/p/subscriptions/s"}"# + ) + .is_none()); + } + + #[test] + fn only_an_entry_on_the_setup_log_with_a_flag_is_a_setup_report() { + let report = |payload: JsonValue| { + parse_setup_report( + json!({ "logName": "projects/p/logs/maple-setup", "jsonPayload": payload }) + .to_string() + .as_bytes(), + ) + }; + assert_eq!( + report(json!({ "logs": true })), + Some(SetupReport { + logs: Some(true), + metrics: None + }) + ); + assert_eq!( + report(json!({ "logs": false, "metrics": false })), + Some(SetupReport { + logs: Some(false), + metrics: Some(false) + }) + ); + // Nothing a script sends: no flag, or a flag that is not a boolean. + assert_eq!(report(json!({})), None); + assert_eq!(report(json!({ "logs": "true" })), None); + + // A customer's own entry that mentions the log, on another log. + let elsewhere = json!({ + "logName": "projects/p/logs/app", + "textPayload": "wrote projects/p/logs/maple-setup", + "jsonPayload": { "logs": true } + }); + assert_eq!(parse_setup_report(elsewhere.to_string().as_bytes()), None); + assert_eq!(parse_setup_report(b"not json /logs/maple-setup"), None); + + // Written with `gcloud logging write maple-setup` and routed by the sink. + let routed = json!({ + "logName": "projects/p/logs/maple-setup", + "receiveTimestamp": "2026-10-14T21:00:00.123Z", + "jsonPayload": { "logs": true } + }); + assert_eq!(parse_setup_report(routed.to_string().as_bytes()), None); + } +} diff --git a/apps/ingest/src/gcp_push_auth.rs b/apps/ingest/src/gcp_push_auth.rs new file mode 100644 index 000000000..f3383aa5a --- /dev/null +++ b/apps/ingest/src/gcp_push_auth.rs @@ -0,0 +1,437 @@ +//! Pub/Sub push authentication. A push subscription set up with a service +//! account sends `Authorization: Bearer `, signed by Google, naming +//! that account (`email`) and the audience the subscription was given (`aud`). +//! Only someone who may act as the account can make Pub/Sub sign as it, so the +//! token stands in for a secret in the push URL. + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use base64::engine::general_purpose::URL_SAFE_NO_PAD; +use base64::Engine; +use ring::signature::{RsaPublicKeyComponents, RSA_PKCS1_2048_8192_SHA256}; +use serde::de::DeserializeOwned; +use serde::Deserialize; +use tokio::sync::RwLock; + +use crate::telemetry::HttpClient; + +const GOOGLE_CERTS_URL: &str = "https://www.googleapis.com/oauth2/v3/certs"; +const GOOGLE_ISSUERS: [&str; 2] = ["https://accounts.google.com", "accounts.google.com"]; +/// Tolerated difference between Google's clock and this host's. +const CLOCK_SKEW_SECS: u64 = 60; +/// Google publishes a signing key well before it uses it, and answers with a +/// `max-age` of several hours. An hour keeps rotation far inside that. +const KEYS_TTL: Duration = Duration::from_hours(1); +/// A token naming a key not in hand fetches the keys again at most this often, +/// so made-up key ids can't turn every push into a fetch. +const REFETCH_INTERVAL: Duration = Duration::from_mins(1); + +/// An RSA public key, as the big-endian bytes of its modulus and exponent. +#[derive(Clone, Debug)] +pub struct RsaKey { + pub n: Vec, + pub e: Vec, +} + +/// Google's signing keys by key id. +pub type Keys = HashMap; + +/// What a valid token says. +#[derive(Debug, PartialEq, Eq)] +pub struct VerifiedPush { + /// The service account the subscription signs as. + pub email: String, + pub audience: String, +} + +#[derive(Debug, PartialEq, Eq)] +pub enum TokenError { + /// Signed with a key not among those given. + UnknownKey, + /// Never a token to accept: malformed, forged, expired, not Google's. + Invalid(&'static str), +} + +#[derive(Debug)] +pub enum PushAuthError { + /// The push is refused. + Invalid(&'static str), + /// Google's keys could not be fetched: the push is not to blame and is + /// worth delivering again. + KeysUnavailable(String), +} + +#[derive(Deserialize)] +struct Header { + alg: String, + kid: Option, +} + +#[derive(Deserialize)] +struct Claims { + iss: String, + aud: String, + exp: u64, + #[serde(default)] + iat: u64, + email: Option, + #[serde(default)] + email_verified: bool, +} + +#[derive(Deserialize)] +struct Jwk { + kid: String, + n: String, + e: String, +} + +#[derive(Deserialize)] +struct JwkSet { + keys: Vec, +} + +fn decode_json(segment: &str) -> Option { + serde_json::from_slice(&URL_SAFE_NO_PAD.decode(segment).ok()?).ok() +} + +/// Checks a token's RS256 signature against `keys`, and its claims at +/// `now_secs`. The algorithm is fixed: a token that names another is refused. +pub fn verify_with_keys( + token: &str, + keys: &Keys, + now_secs: u64, +) -> Result { + let invalid = TokenError::Invalid; + let mut segments = token.split('.'); + let (Some(header), Some(claims), Some(signature), None) = ( + segments.next(), + segments.next(), + segments.next(), + segments.next(), + ) else { + return Err(invalid("not a JWT")); + }; + let parsed: Header = decode_json(header).ok_or(invalid("malformed header"))?; + if parsed.alg != "RS256" { + return Err(invalid("unsupported algorithm")); + } + let key = parsed + .kid + .as_deref() + .and_then(|kid| keys.get(kid)) + .ok_or(TokenError::UnknownKey)?; + let signature = URL_SAFE_NO_PAD + .decode(signature) + .map_err(|_| invalid("malformed signature"))?; + let signed = &token[..header.len() + 1 + claims.len()]; + RsaPublicKeyComponents { + n: &key.n, + e: &key.e, + } + .verify(&RSA_PKCS1_2048_8192_SHA256, signed.as_bytes(), &signature) + .map_err(|_| invalid("bad signature"))?; + + let claims: Claims = decode_json(claims).ok_or(invalid("malformed claims"))?; + if !GOOGLE_ISSUERS.contains(&claims.iss.as_str()) { + return Err(invalid("not issued by Google")); + } + if claims.exp.saturating_add(CLOCK_SKEW_SECS) < now_secs { + return Err(invalid("expired")); + } + if claims.iat > now_secs.saturating_add(CLOCK_SKEW_SECS) { + return Err(invalid("issued in the future")); + } + let email = claims + .email + .filter(|_| claims.email_verified) + .ok_or(invalid("no verified email"))?; + Ok(VerifiedPush { + email, + audience: claims.aud, + }) +} + +fn parse_keys(body: &[u8]) -> Result { + let set: JwkSet = serde_json::from_slice(body).map_err(|error| error.to_string())?; + set.keys + .into_iter() + .map(|jwk| { + let n = URL_SAFE_NO_PAD + .decode(&jwk.n) + .map_err(|error| error.to_string())?; + let e = URL_SAFE_NO_PAD + .decode(&jwk.e) + .map_err(|error| error.to_string())?; + Ok((jwk.kid, RsaKey { n, e })) + }) + .collect() +} + +enum Source { + Google(HttpClient), + Fixed, +} + +struct Cached { + keys: Arc, + fetched_at: Option, +} + +/// Verifies push tokens against Google's published keys, fetched on first use +/// and kept for `KEYS_TTL`. +pub struct GooglePushAuth { + source: Source, + cached: RwLock, +} + +impl std::fmt::Debug for GooglePushAuth { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("GooglePushAuth") + .finish_non_exhaustive() + } +} + +impl GooglePushAuth { + pub fn new(client: impl Into) -> Self { + Self { + source: Source::Google(client.into()), + cached: RwLock::new(Cached { + keys: Arc::new(Keys::new()), + fetched_at: None, + }), + } + } + + /// Never fetches: for tests and hosts without network access. + pub fn with_keys(keys: Keys) -> Self { + Self { + source: Source::Fixed, + cached: RwLock::new(Cached { + keys: Arc::new(keys), + fetched_at: Some(Instant::now()), + }), + } + } + + pub async fn verify(&self, token: &str, now_secs: u64) -> Result { + let keys = self.keys(false).await?; + let verified = match verify_with_keys(token, &keys, now_secs) { + // Google may have started signing with a key published since. + Err(TokenError::UnknownKey) => { + verify_with_keys(token, &*self.keys(true).await?, now_secs) + } + other => other, + }; + verified.map_err(|error| match error { + TokenError::UnknownKey => PushAuthError::Invalid("unknown key"), + TokenError::Invalid(reason) => PushAuthError::Invalid(reason), + }) + } + + async fn keys(&self, unknown_key: bool) -> Result, PushAuthError> { + let due = |cached: &Cached| match cached.fetched_at { + None => true, + Some(at) => at.elapsed() > KEYS_TTL || (unknown_key && at.elapsed() > REFETCH_INTERVAL), + }; + { + let cached = self.cached.read().await; + if !due(&cached) { + return Ok(Arc::clone(&cached.keys)); + } + } + let Source::Google(client) = &self.source else { + return Ok(Arc::clone(&self.cached.read().await.keys)); + }; + let mut cached = self.cached.write().await; + // Another push may have fetched while this one waited for the lock. + if !due(&cached) { + return Ok(Arc::clone(&cached.keys)); + } + match fetch_keys(client).await { + Ok(keys) => { + cached.keys = Arc::new(keys); + cached.fetched_at = Some(Instant::now()); + Ok(Arc::clone(&cached.keys)) + } + // Keys still in hand keep verifying until a fetch succeeds. + Err(_) if !cached.keys.is_empty() => Ok(Arc::clone(&cached.keys)), + Err(error) => Err(PushAuthError::KeysUnavailable(error)), + } + } +} + +async fn fetch_keys(client: &HttpClient) -> Result { + let response = client + .get(GOOGLE_CERTS_URL) + .send() + .await + .map_err(|error| format!("fetching Google's signing keys failed: {error}"))?; + if !response.status().is_success() { + return Err(format!( + "fetching Google's signing keys failed: HTTP {}", + response.status() + )); + } + let body = response + .bytes() + .await + .map_err(|error| format!("reading Google's signing keys failed: {error}"))?; + parse_keys(&body) +} + +#[cfg(test)] +mod tests { + use super::*; + use ring::rand::SystemRandom; + use ring::signature::{RsaKeyPair, RSA_PKCS1_SHA256}; + use serde_json::json; + + const TEST_KEY: &[u8] = include_bytes!("testdata/gcp-push-test-key.pk8"); + const TEST_KEY_N: &str = include_str!("testdata/gcp-push-test-key.n"); + const NOW: u64 = 1_792_000_000; + + fn keys() -> Keys { + Keys::from([( + "key-1".to_owned(), + RsaKey { + n: URL_SAFE_NO_PAD.decode(TEST_KEY_N.trim()).unwrap(), + e: vec![1, 0, 1], + }, + )]) + } + + fn sign(header: &serde_json::Value, claims: &serde_json::Value) -> String { + let encode = |value: &serde_json::Value| URL_SAFE_NO_PAD.encode(value.to_string()); + let signed = format!("{}.{}", encode(header), encode(claims)); + let pair = RsaKeyPair::from_pkcs8(TEST_KEY).unwrap(); + let mut signature = vec![0; pair.public().modulus_len()]; + pair.sign( + &RSA_PKCS1_SHA256, + &SystemRandom::new(), + signed.as_bytes(), + &mut signature, + ) + .unwrap(); + format!("{signed}.{}", URL_SAFE_NO_PAD.encode(signature)) + } + + fn claims() -> serde_json::Value { + json!({ + "iss": "https://accounts.google.com", + "aud": "https://ingest.test/v1/logpush/gcp/c1", + "exp": NOW + 3600, + "iat": NOW, + "email": "maple-push-0a1b@acme.iam.gserviceaccount.com", + "email_verified": true, + "sub": "1234", + }) + } + + fn token(claims: &serde_json::Value) -> String { + sign( + &json!({ "alg": "RS256", "kid": "key-1", "typ": "JWT" }), + claims, + ) + } + + #[test] + fn accepts_a_token_google_signed_for_a_verified_account() { + assert_eq!( + verify_with_keys(&token(&claims()), &keys(), NOW), + Ok(VerifiedPush { + email: "maple-push-0a1b@acme.iam.gserviceaccount.com".to_owned(), + audience: "https://ingest.test/v1/logpush/gcp/c1".to_owned(), + }) + ); + } + + #[test] + fn refuses_tokens_that_are_not_google_signed_current_and_verified() { + let changed = |key: &str, value: serde_json::Value| { + let mut claims = claims(); + claims[key] = value; + verify_with_keys(&token(&claims), &keys(), NOW) + }; + assert_eq!( + changed("iss", json!("https://evil.test")), + Err(TokenError::Invalid("not issued by Google")) + ); + assert_eq!( + changed("exp", json!(NOW - 61)), + Err(TokenError::Invalid("expired")) + ); + assert_eq!( + changed("iat", json!(NOW + 61)), + Err(TokenError::Invalid("issued in the future")) + ); + assert_eq!( + changed("email_verified", json!(false)), + Err(TokenError::Invalid("no verified email")) + ); + // Within the tolerated clock difference. + assert!(changed("exp", json!(NOW - 59)).is_ok()); + + // Another account's claims under a genuine signature. + let valid = token(&claims()); + let segments: Vec<&str> = valid.split('.').collect(); + let mut forged = claims(); + forged["email"] = json!("someone-else@acme.iam.gserviceaccount.com"); + let swapped = format!( + "{}.{}.{}", + segments[0], + URL_SAFE_NO_PAD.encode(forged.to_string()), + segments[2] + ); + assert_eq!( + verify_with_keys(&swapped, &keys(), NOW), + Err(TokenError::Invalid("bad signature")) + ); + + let none = sign(&json!({ "alg": "none", "kid": "key-1" }), &claims()); + assert_eq!( + verify_with_keys(&none, &keys(), NOW), + Err(TokenError::Invalid("unsupported algorithm")) + ); + let other_key = sign(&json!({ "alg": "RS256", "kid": "key-2" }), &claims()); + assert_eq!( + verify_with_keys(&other_key, &keys(), NOW), + Err(TokenError::UnknownKey) + ); + assert_eq!( + verify_with_keys("a.b", &keys(), NOW), + Err(TokenError::Invalid("not a JWT")) + ); + } + + #[test] + fn reads_google_jwk_sets() { + let parsed = parse_keys( + json!({ "keys": [{ "kid": "abc", "n": TEST_KEY_N.trim(), "e": "AQAB", "kty": "RSA", "alg": "RS256", "use": "sig" }] }) + .to_string() + .as_bytes(), + ) + .unwrap(); + assert_eq!(parsed["abc"].e, vec![1, 0, 1]); + assert!(verify_with_keys( + &sign(&json!({ "alg": "RS256", "kid": "abc" }), &claims()), + &parsed, + NOW + ) + .is_ok()); + } + + #[tokio::test] + async fn fixed_keys_verify_without_fetching() { + let auth = GooglePushAuth::with_keys(keys()); + assert!(auth.verify(&token(&claims()), NOW).await.is_ok()); + let other_key = sign(&json!({ "alg": "RS256", "kid": "key-2" }), &claims()); + assert!(matches!( + auth.verify(&other_key, NOW).await, + Err(PushAuthError::Invalid("unknown key")) + )); + } +} diff --git a/apps/ingest/src/lib.rs b/apps/ingest/src/lib.rs index ceba8a7d6..22c1e5e57 100644 --- a/apps/ingest/src/lib.rs +++ b/apps/ingest/src/lib.rs @@ -1,6 +1,8 @@ pub mod ai_session; pub mod aws; pub mod clickhouse_insert_mappings; +pub mod gcp_logging; +pub mod gcp_push_auth; pub mod metrics; pub mod otel; pub mod otlp_json; diff --git a/apps/ingest/src/main.rs b/apps/ingest/src/main.rs index f153192c9..5c0a4b5fd 100644 --- a/apps/ingest/src/main.rs +++ b/apps/ingest/src/main.rs @@ -15,7 +15,7 @@ mod task_protection; use std::io::{Read, Write}; use std::path::PathBuf; -use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -46,6 +46,8 @@ use hmac::{Hmac, Mac}; use maple_ingest::ai_session; use maple_ingest::aws::CredentialsProvider as AwsCredentialsProvider; use maple_ingest::clickhouse_insert_mappings::SCHEMA_VERSION as CLICKHOUSE_SCHEMA_VERSION; +use maple_ingest::gcp_logging; +use maple_ingest::gcp_push_auth::{GooglePushAuth, PushAuthError}; use maple_ingest::metrics; use maple_ingest::otel::{ accept_internal_span, auth_internal_span, build_resource, decode_internal_span, @@ -159,6 +161,8 @@ struct AppConfig { tinybird: TinybirdConfig, max_request_body_bytes: usize, org_max_in_flight: u64, + /// Per-org cap on concurrent GCP Pub/Sub pushes, inside `org_max_in_flight`. + gcp_org_max_in_flight: u64, /// Wall-clock deadline for one HTTP request, or `Duration::ZERO` to run /// without one. See `request_timeout_middleware`. request_timeout: Duration, @@ -438,6 +442,18 @@ impl AppConfig { if org_max_in_flight == 0 { return Err("INGEST_ORG_MAX_IN_FLIGHT must be greater than 0".to_owned()); } + // Pub/Sub pushes one log entry per request, and every request is its own + // WAL fsync on the org's lane. This bounds how deep one org's pushes can + // queue on that lane, ahead of every other request sharing it. Pub/Sub + // slows down on the 429. + let gcp_org_max_in_flight = parse_u64( + "INGEST_GCP_ORG_MAX_IN_FLIGHT", + std::env::var("INGEST_GCP_ORG_MAX_IN_FLIGHT").ok(), + 32, + )?; + if gcp_org_max_in_flight == 0 { + return Err("INGEST_GCP_ORG_MAX_IN_FLIGHT must be greater than 0".to_owned()); + } // 30s. Comfortably above a healthy p99 (single-digit ms) and below the // load balancer idle timeout, so the gateway answers rather than having @@ -668,6 +684,7 @@ impl AppConfig { tinybird, max_request_body_bytes, org_max_in_flight, + gcp_org_max_in_flight, request_timeout, require_tls, key_store_backend, @@ -772,6 +789,39 @@ struct CloudflareConnectorResolver { routing: Arc, } +const CLOUDFLARE_CONNECTORS_TABLE: &str = "cloudflare_logpush_connectors"; +const GCP_CONNECTORS_TABLE: &str = "gcp_connectors"; + +/// How often one connector's health columns are rewritten per outcome. Pub/Sub +/// pushes one log entry per request, so an unthrottled write is one UPDATE per +/// log line. +const GCP_CONNECTOR_HEALTH_WRITE_INTERVAL: Duration = Duration::from_mins(1); + +/// Setup reports one connector may write per `GCP_SETUP_REPORT_WINDOW`. A run +/// sends at most two (logs, metrics); the rest is headroom for runs pasted +/// again and Pub/Sub redeliveries. Each report is a Postgres write on the pool +/// key resolution uses, so a connector that sends more is refused. +const GCP_SETUP_REPORTS_PER_WINDOW: u32 = 10; +const GCP_SETUP_REPORT_WINDOW: Duration = Duration::from_mins(1); + +struct GcpConnectorResolver { + store: Arc, + lookup_hmac_key: String, + /// Keyed on (connector id, secret hash). + cache: Cache<(String, String), GcpConnectorIdentity>, + // A deleted connector's subscription keeps pushing every log line, and + // Pub/Sub redelivers each 401; this keeps that off Postgres. + negative_cache: Cache<(String, String), ()>, + routing: Arc, + /// (connector id, succeeded) pairs already written within + /// `GCP_CONNECTOR_HEALTH_WRITE_INTERVAL`. + recent_health_writes: Cache<(String, bool), ()>, + /// Setup reports per connector in the current `GCP_SETUP_REPORT_WINDOW`. + recent_setup_reports: Cache>, + /// Verifies the ID tokens Pub/Sub signs pushes with. + push_auth: GooglePushAuth, +} + struct OrgRoutingResolver { store: Arc, cache: Cache, @@ -847,6 +897,18 @@ trait KeyStore: Send + Sync { secret_hash: &str, ) -> Result, String>; + async fn fetch_gcp_connector( + &self, + connector_id: &str, + secret_hash: &str, + ) -> Result, String>; + + /// For a push signed by Google: the caller checks the signing account. + async fn fetch_gcp_connector_by_id( + &self, + connector_id: &str, + ) -> Result, String>; + async fn fetch_sampling_policy( &self, org_id: &str, @@ -864,15 +926,29 @@ trait KeyStore: Send + Sync { async fn fetch_org_routing(&self, org_id: &str) -> Result, String>; - async fn record_connector_success(&self, connector_id: &str, now_ms: i64) - -> Result<(), String>; + /// `table` is one of the `*_CONNECTORS_TABLE` constants, never user input. + async fn record_connector_success( + &self, + table: &'static str, + connector_id: &str, + now_ms: i64, + ) -> Result<(), String>; async fn record_connector_failure( &self, + table: &'static str, connector_id: &str, error: &str, now_ms: i64, ) -> Result<(), String>; + + /// What a Google Cloud setup script run reported it applied. + async fn record_gcp_setup_report( + &self, + connector_id: &str, + report: gcp_logging::SetupReport, + now_ms: i64, + ) -> Result<(), String>; } #[derive(Clone, Debug)] @@ -892,6 +968,18 @@ struct ConnectorRow { clickhouse_ready: bool, } +#[derive(Clone, Debug)] +struct GcpConnectorRow { + org_id: String, + /// The host project: its push service account signs the connector's pushes. + project_id: String, + logs_enabled: bool, + /// What a setup run last reported; `None` before any report. + applied_logs_enabled: Option, + self_managed: bool, + clickhouse_ready: bool, +} + #[derive(Clone, Debug)] struct SamplingPolicyRow { trace_sample_ratio: f64, @@ -951,6 +1039,30 @@ impl CloudflareConnectorIdentity { } } +#[derive(Clone)] +struct GcpConnectorIdentity { + org_id: String, + secret_key_id: String, + logs_enabled: bool, + confirms_log_forwarding: bool, +} + +impl GcpConnectorIdentity { + fn into_resolved(self, routing: &OrgRouting) -> ResolvedGcpConnector { + ResolvedGcpConnector { + logs_enabled: self.logs_enabled, + confirms_log_forwarding: self.confirms_log_forwarding, + key: ResolvedIngestKey { + org_id: self.org_id, + key_type: IngestKeyType::Connector, + key_id: self.secret_key_id, + self_managed: routing.self_managed, + clickhouse_ready: routing.clickhouse_ready, + }, + } + } +} + #[derive(Clone, Debug, Default)] struct OrgRouting { self_managed: bool, @@ -1112,9 +1224,12 @@ struct AppState { key_store_ready: Arc, resolver: IngestKeyResolver, org_inflight_limiter: OrgInFlightLimiter, + /// GCP pushes only, taken before `org_inflight_limiter`. + gcp_inflight_limiter: OrgInFlightLimiter, sampling_resolver: SamplingPolicyResolver, attribute_mapping_resolver: AttributeMappingResolver, cloudflare_resolver: CloudflareConnectorResolver, + gcp_resolver: GcpConnectorResolver, autumn_tracker: Option, autumn_entitlements: Option, /// Per-org ingest volume, on its own delta-temporality provider. Recorded @@ -1160,6 +1275,28 @@ struct ResolvedCloudflareConnector { clickhouse_ready: bool, } +struct ResolvedGcpConnector { + /// False once the org turns log forwarding off; the sink may still push. + logs_enabled: bool, + /// A signed push to a connector no setup run has reported on: one set up + /// with Terraform, which sends no report. The push itself proves the sink, + /// topic and subscription are in place. + confirms_log_forwarding: bool, + key: ResolvedIngestKey, +} + +/// What an acked GCP push did. +enum GcpPushOutcome { + Accepted, + /// The body can never parse, so redelivering it is pointless. + NotALogEntry, + /// The connector has log forwarding turned off. + LogsDisabled, + /// A setup script run saying what it applied: recorded on the connector, + /// not a log. + SetupReport, +} + #[derive(Clone, Copy)] enum IngestKeyType { Public, @@ -1236,11 +1373,17 @@ impl Drop for InFlightGuard { struct OrgInFlightLimiter { max_per_org: u64, counts: Arc>>, + /// The `org_throttled` reason. + throttle_reason: &'static str, + /// Only the org-wide limiter owns the per-org in-flight gauge; a route's + /// limiter writing the same series would overwrite it. + records_gauge: bool, } struct OrgInFlightPermit { org_id: String, counter: Arc, + records_gauge: bool, } impl OrgInFlightLimiter { @@ -1248,6 +1391,17 @@ impl OrgInFlightLimiter { Self { max_per_org, counts: Arc::new(DashMap::new()), + throttle_reason: "in_flight", + records_gauge: true, + } + } + + /// A tighter cap for one route, taken on top of the org-wide one. + fn for_route(max_per_org: u64, throttle_reason: &'static str) -> Self { + Self { + throttle_reason, + records_gauge: false, + ..Self::new(max_per_org) } } @@ -1261,17 +1415,20 @@ impl OrgInFlightLimiter { loop { let current = counter.load(Ordering::Relaxed); if current >= self.max_per_org { - metrics::org_throttled(org_id, "in_flight"); + metrics::org_throttled(org_id, self.throttle_reason); return None; } if counter .compare_exchange(current, current + 1, Ordering::AcqRel, Ordering::Relaxed) .is_ok() { - metrics::org_requests_in_flight(org_id, current + 1); + if self.records_gauge { + metrics::org_requests_in_flight(org_id, current + 1); + } return Some(OrgInFlightPermit { org_id: org_id.to_owned(), counter, + records_gauge: self.records_gauge, }); } } @@ -1281,8 +1438,9 @@ impl OrgInFlightLimiter { impl Drop for OrgInFlightPermit { fn drop(&mut self) { let current = self.counter.fetch_sub(1, Ordering::AcqRel); - let next = current.saturating_sub(1); - metrics::org_requests_in_flight(&self.org_id, next); + if self.records_gauge { + metrics::org_requests_in_flight(&self.org_id, current.saturating_sub(1)); + } } } @@ -1664,9 +1822,9 @@ async fn request_timeout_middleware( ) -> Response { let method = request.method().clone(); let path = request.uri().path().to_owned(); - // The metric gets the matched route pattern, not the raw path: one route - // carries a path parameter (`/v1/logpush/cloudflare/http_requests/{connector_id}`), - // and labelling by raw path would add a metric series per connector. This + // The metric gets the matched route pattern, not the raw path: the logpush + // routes carry a path parameter (`/v1/logpush/gcp/{connector_id}`), and + // labelling by raw path would add a metric series per connector. This // layer sits on the `Router`, so routing has already run and the extension is // populated; "unknown" only appears if that ever stops being true. let matched_path = request @@ -2404,6 +2562,10 @@ async fn main() { routing: Arc::clone(&org_routing_resolver), }, org_inflight_limiter: OrgInFlightLimiter::new(config.org_max_in_flight), + gcp_inflight_limiter: OrgInFlightLimiter::for_route( + config.gcp_org_max_in_flight, + "gcp_in_flight", + ), sampling_resolver: SamplingPolicyResolver { store: Arc::clone(&store), cache: sampling_policy_cache, @@ -2416,8 +2578,15 @@ async fn main() { store: Arc::clone(&store), lookup_hmac_key: config.lookup_hmac_key.clone(), cache: cloudflare_connector_cache, - routing: org_routing_resolver, + routing: Arc::clone(&org_routing_resolver), }, + gcp_resolver: GcpConnectorResolver::new( + Arc::clone(&store), + config.lookup_hmac_key.clone(), + org_routing_resolver, + Duration::from_secs(config.ingest_key_cache_ttl_secs), + GooglePushAuth::new(http_client.clone()), + ), telemetry_pipeline, http_client, config: config.clone(), @@ -2474,6 +2643,7 @@ async fn main() { "/v1/logpush/cloudflare/http_requests/{connector_id}", post(handle_cloudflare_logpush_http_requests), ) + .route("/v1/logpush/gcp/{connector_id}", post(handle_gcp_logpush)) .layer(cors) .layer(DefaultBodyLimit::max(config.max_request_body_bytes)) .with_state(state); @@ -4068,14 +4238,14 @@ struct AcceptedBody { } #[derive(Deserialize)] -struct CloudflareLogpushQuery { +struct LogpushQuery { secret: Option, } async fn handle_cloudflare_logpush_http_requests( State(state): State>, Path(connector_id): Path, - Query(query): Query, + Query(query): Query, headers: HeaderMap, body: Bytes, ) -> Response { @@ -4674,6 +4844,345 @@ async fn handle_cloudflare_logpush_inner( } } +/// Receives a Cloud Logging sink's Pub/Sub push subscription: one `LogEntry` +/// per request, unwrapped. Pub/Sub acks on 102/200/201/202/204 and redelivers +/// on anything else until the message's retention expires. +async fn handle_gcp_logpush( + State(state): State>, + Path(connector_id): Path, + Query(query): Query, + headers: HeaderMap, + body: Bytes, +) -> Response { + let start = Instant::now(); + + metrics::request_started(); + let _guard = InFlightGuard; + + let span = tracing::info_span!( + "gcp_logpush", + otel.name = "POST /v1/logpush/gcp/{connector_id}", + otel.kind = "server", + otel.status_code = tracing::field::Empty, + otel.status_description = tracing::field::Empty, + "maple.ingest.reject_reason" = tracing::field::Empty, + "http.request.method" = "POST", + "http.route" = "/v1/logpush/gcp/{connector_id}", + "http.request.body.size" = body.len(), + "http.response.status_code" = tracing::field::Empty, + "error.type" = tracing::field::Empty, + "maple.signal" = "logs", + "maple.org_id" = tracing::field::Empty, + "maple.gcp.connector_id" = tracing::field::Empty, + "maple.gcp.setup_report" = tracing::field::Empty, + "maple.gcp.push_auth" = tracing::field::Empty, + "maple.ingest.self_managed" = tracing::field::Empty, + "maple.ingest.clickhouse_ready" = tracing::field::Empty, + "maple.ingest.destination" = tracing::field::Empty, + "maple.ingest.item_count" = tracing::field::Empty, + ); + let span_handle = span.clone(); + + // Set up by the current script or Terraform: a token Pub/Sub signed. + // Earlier subscriptions carry the secret in the URL. + let token = headers + .get(AUTHORIZATION) + .and_then(|value| value.to_str().ok()) + .and_then(|value| value.strip_prefix("Bearer ")); + let credential = match (token, query.secret.as_deref()) { + (Some(token), _) => Some(GcpPushCredential::Token(token)), + (None, Some(secret)) => Some(GcpPushCredential::Secret(secret)), + (None, None) => None, + }; + let result = handle_gcp_logpush_inner(&state, &connector_id, credential, body) + .instrument(span) + .await; + let duration = start.elapsed().as_secs_f64(); + + match result { + Ok((response, outcome)) => { + span_handle.record("http.response.status_code", response.status().as_u16()); + match outcome { + GcpPushOutcome::Accepted => { + span_handle.record("otel.status_code", "Ok"); + span_handle.record("maple.ingest.item_count", 1); + metrics::request_completed("logs", "ok", "none", duration); + metrics::gcp_entry(); + } + GcpPushOutcome::NotALogEntry => { + // Acked, but the entry is gone: an `Error` like every other + // rejection that loses the sender's data. + span_handle.record("error.type", "decode"); + record_rejection_reason( + &span_handle, + 200, + "decode", + "Push payload is not a LogEntry", + ); + metrics::request_completed("logs", "error", "decode", duration); + metrics::gcp_parse_failure(); + } + GcpPushOutcome::LogsDisabled => { + // The org asked for this, so it is neither an error nor + // data loss: `Ok`, told apart by the reject reason. + record_rejection_reason( + &span_handle, + 200, + "logs_disabled", + "Log forwarding is turned off for this connector", + ); + metrics::request_completed("logs", "ok", "none", duration); + metrics::gcp_disabled_drop(); + } + GcpPushOutcome::SetupReport => { + span_handle.record("otel.status_code", "Ok"); + span_handle.record("maple.gcp.setup_report", true); + metrics::request_completed("logs", "ok", "none", duration); + } + } + response + } + Err((error, error_kind)) => { + let status = error.status.as_u16(); + span_handle.record("http.response.status_code", status); + span_handle.record("error.type", error_kind); + record_rejection_reason(&span_handle, status, error_kind, error.reason().as_str()); + metrics::request_completed("logs", "error", error_kind, duration); + if error_kind == "auth" { + metrics::gcp_auth_failure(); + } + error.into_response() + } + } +} + +enum GcpPushCredential<'a> { + Token(&'a str), + Secret(&'a str), +} + +/// Returns Ok((ack response, what the ack means)) or +/// Err((ApiError, error_kind_label)). +async fn handle_gcp_logpush_inner( + state: &AppState, + connector_id: &str, + credential: Option>, + body: Bytes, +) -> Result<(Response, GcpPushOutcome), (ApiError, &'static str)> { + let unauthorized = || { + ( + ApiError::unauthorized("Invalid connector credentials"), + "auth", + ) + }; + let credential = credential.filter(|credential| match credential { + GcpPushCredential::Token(value) | GcpPushCredential::Secret(value) => { + !value.trim().is_empty() + } + }); + let (Some(credential), true) = (credential, is_connector_id(connector_id)) else { + warn!("Missing GCP connector credentials or malformed connector id"); + return Err(unauthorized()); + }; + Span::current().record("maple.gcp.connector_id", connector_id); + + let lookup = match credential { + GcpPushCredential::Token(token) => { + Span::current().record("maple.gcp.push_auth", "oidc"); + state + .gcp_resolver + .resolve_connector_by_token(connector_id, token.trim()) + .await + } + GcpPushCredential::Secret(secret) => { + Span::current().record("maple.gcp.push_auth", "secret"); + state + .gcp_resolver + .resolve_connector(connector_id, secret.trim()) + .await + } + }; + let resolved = lookup + .map_err(|error| { + error!(error = %error, connector_id, "GCP connector resolution failed"); + ( + ApiError::service_unavailable("Connector authentication unavailable"), + "auth", + ) + })? + .ok_or_else(|| { + warn!(connector_id, "Invalid GCP connector credentials"); + unauthorized() + })?; + let org_id = resolved.key.org_id.as_str(); + + Span::current().record("maple.org_id", org_id); + if resolved.confirms_log_forwarding { + state + .gcp_resolver + .confirm_log_forwarding(connector_id) + .await; + } + Span::current().record("maple.ingest.self_managed", resolved.key.self_managed); + Span::current().record( + "maple.ingest.clickhouse_ready", + resolved.key.clickhouse_ready, + ); + + let _gcp_inflight_permit = state + .gcp_inflight_limiter + .try_acquire(org_id) + .ok_or_else(|| { + // Not `warn!`: a busy sink meets this cap by design, once per + // refused log line. `org_throttled` counts it. + debug!( + org_id, + connector_id, "Per-org in-flight GCP push limit exceeded" + ); + ( + ApiError::too_many_requests("Per-org ingest limit exceeded"), + "throttle", + ) + })?; + + // A setup script run reporting what it applied. Ahead of everything a log + // goes through: it is told to a connector whose log forwarding is off as + // well, and it is neither stored nor metered. A report that cannot be + // written is refused, so Pub/Sub delivers it again. + if let Some(report) = gcp_logging::parse_setup_report(&body) { + if !state + .gcp_resolver + .take_setup_report_slot(connector_id) + .await + { + warn!(org_id, connector_id, "GCP setup report rate limit exceeded"); + return Err(( + ApiError::too_many_requests("Too many setup reports"), + "throttle", + )); + } + state + .gcp_resolver + .store + .record_gcp_setup_report(connector_id, report, current_time_millis()) + .await + .map_err(|error| { + error!(error = %error, connector_id, "Failed to record GCP setup report"); + ( + ApiError::service_unavailable("Setup report could not be recorded"), + "setup_report", + ) + })?; + return Ok((StatusCode::OK.into_response(), GcpPushOutcome::SetupReport)); + } + + // Acked, not rejected: the sink keeps pushing until the customer removes + // it, and Pub/Sub would redeliver every refused entry for its retention. + // Ahead of the entitlement check and metering, so nothing is billed. + if !resolved.logs_enabled { + debug!( + org_id, + connector_id, "GCP log forwarding is off; entry acked and dropped" + ); + return Ok((StatusCode::OK.into_response(), GcpPushOutcome::LogsDisabled)); + } + + let _org_inflight_permit = state + .org_inflight_limiter + .try_acquire(org_id) + .ok_or_else(|| { + warn!( + org_id, + connector_id, "Per-org in-flight ingest limit exceeded" + ); + ( + ApiError::too_many_requests("Per-org ingest limit exceeded"), + "throttle", + ) + })?; + + let Some(entry) = gcp_logging::parse_log_entry(&body) else { + // Not `warn!`: a wrapped subscription fails this way on every log line. + debug!( + org_id, + connector_id, + body_bytes = body.len(), + "GCP push payload is not a LogEntry; acked and dropped" + ); + metrics::org_data_loss(org_id, Signal::Logs.path(), "decode"); + state + .gcp_resolver + .record_health(connector_id, Some(gcp_logging::NOT_A_LOG_ENTRY)) + .await; + return Ok((StatusCode::OK.into_response(), GcpPushOutcome::NotALogEntry)); + }; + + if let Some(error) = entitlement_rejection(state, org_id, Signal::Logs.path()).await { + state + .gcp_resolver + .record_health(connector_id, Some(gcp_logging::OVER_PLAN_LIMIT)) + .await; + return Err((error, "billing_limit")); + } + let request = gcp_logging::build_logs_request( + &entry, + &gcp_logging::Connector { + id: connector_id, + org_id, + }, + current_time_unix_nano(), + ); + let response = match process_decoded_payload( + state, + Signal::Logs, + PayloadFormat::Protobuf, + None, + &DecodedPayload::Logs(request), + &resolved.key, + ) + .await + { + Ok(response) => response, + Err(error) => { + state + .gcp_resolver + .record_health(connector_id, Some(gcp_logging::ENTRY_NOT_STORED)) + .await; + return Err((error, "forward")); + } + }; + state.gcp_resolver.record_health(connector_id, None).await; + + if let Some(usage) = &state.usage_metrics { + usage.record(org_id, Signal::Logs.path(), body.len() as u64, 1); + } + if let Some(tracker) = &state.autumn_tracker { + tracker.track(org_id, Signal::Logs.path(), billable_gb(body.len() as u64)); + } + + Ok((response, GcpPushOutcome::Accepted)) +} + +/// Connector ids are generated by the API; anything outside this alphabet is +/// refused before it reaches this handler's span, its logs, or Postgres. +/// The push service account a connector's setup creates in its host project, +/// named as `gcpConnectorResourceNames` in `@maple/domain/gcp` names it. +fn gcp_push_service_account(connector_id: &str, project_id: &str) -> String { + let hex: String = connector_id + .chars() + .filter(|c| *c != '-') + .take(19) + .collect(); + format!("maple-push-{hex}@{project_id}.iam.gserviceaccount.com") +} + +fn is_connector_id(value: &str) -> bool { + (1..=128).contains(&value.len()) + && value + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || byte == b'_' || byte == b'-') +} + enum ParsedCloudflarePayload { Validation, Records(Vec>), @@ -5659,7 +6168,11 @@ impl CloudflareConnectorResolver { async fn record_success(&self, connector_id: &str) { if let Err(error) = self .store - .record_connector_success(connector_id, current_time_millis()) + .record_connector_success( + CLOUDFLARE_CONNECTORS_TABLE, + connector_id, + current_time_millis(), + ) .await { debug!(connector_id, error, "Failed to record connector success"); @@ -5669,7 +6182,12 @@ impl CloudflareConnectorResolver { async fn record_failure(&self, connector_id: &str, error_message: &str) { if let Err(error) = self .store - .record_connector_failure(connector_id, error_message, current_time_millis()) + .record_connector_failure( + CLOUDFLARE_CONNECTORS_TABLE, + connector_id, + error_message, + current_time_millis(), + ) .await { debug!(connector_id, error, "Failed to record connector failure"); @@ -5677,28 +6195,245 @@ impl CloudflareConnectorResolver { } } -impl SamplingPolicyResolver { - #[hotpath::measure] - async fn resolve_policy(&self, org_id: &str) -> SamplingPolicy { - if let Some(policy) = self.cache.get(org_id).await { - return policy; +impl GcpConnectorResolver { + fn new( + store: Arc, + lookup_hmac_key: String, + routing: Arc, + cache_ttl: Duration, + push_auth: GooglePushAuth, + ) -> Self { + Self { + store, + lookup_hmac_key, + cache: Cache::builder() + .time_to_live(cache_ttl) + .max_capacity(10_000) + .build(), + negative_cache: Cache::builder() + .time_to_live(Duration::from_secs(30)) + .max_capacity(10_000) + .build(), + routing, + recent_health_writes: Cache::builder() + .time_to_live(GCP_CONNECTOR_HEALTH_WRITE_INTERVAL) + .max_capacity(20_000) + .build(), + // The TTL runs from the first report, so each entry is one window. + recent_setup_reports: Cache::builder() + .time_to_live(GCP_SETUP_REPORT_WINDOW) + .max_capacity(10_000) + .build(), + push_auth, } + } - let policy = match self.store.fetch_sampling_policy(org_id).await { - Ok(Some(row)) => SamplingPolicy { - trace_sample_ratio: row.trace_sample_ratio, - always_keep_error_spans: row.always_keep_error_spans, - always_keep_slow_spans_ms: row.always_keep_slow_spans_ms, - }, - Ok(None) => SamplingPolicy::default(), - Err(error) => { - warn!( - org_id, - error = %error, - "Sampling policy lookup failed; using unsampled default" - ); - SamplingPolicy::default() - } + /// Whether this connector may write another setup report in this window. + async fn take_setup_report_slot(&self, connector_id: &str) -> bool { + let count = self + .recent_setup_reports + .get_with_by_ref(connector_id, async { Arc::new(AtomicU32::new(0)) }) + .await; + count.fetch_add(1, Ordering::Relaxed) < GCP_SETUP_REPORTS_PER_WINDOW + } + + async fn resolve_connector( + &self, + connector_id: &str, + raw_secret: &str, + ) -> Result, String> { + let secret_hash = hash_ingest_key(raw_secret, &self.lookup_hmac_key)?; + let cache_key = (connector_id.to_owned(), secret_hash); + if let Some(identity) = self.cache.get(&cache_key).await { + let routing = self.routing.resolve_org_routing(&identity.org_id).await?; + return Ok(Some(identity.into_resolved(&routing))); + } + if self.negative_cache.contains_key(&cache_key) { + return Ok(None); + } + + let Some(row) = self + .store + .fetch_gcp_connector(connector_id, &cache_key.1) + .await? + else { + self.negative_cache.insert(cache_key, ()).await; + return Ok(None); + }; + + let routing = OrgRouting { + self_managed: row.self_managed, + clickhouse_ready: row.clickhouse_ready, + }; + let identity = GcpConnectorIdentity { + org_id: row.org_id, + secret_key_id: cache_key.1.chars().take(16).collect(), + logs_enabled: row.logs_enabled, + confirms_log_forwarding: false, + }; + + self.cache.insert(cache_key, identity.clone()).await; + self.routing + .remember_org_routing(&identity.org_id, routing.clone()) + .await; + + Ok(Some(identity.into_resolved(&routing))) + } + + /// A push Pub/Sub signed as the connector's push service account, for an + /// audience naming this connector. `Err` only when Google's keys can't be + /// fetched, which a redelivery may get past. + async fn resolve_connector_by_token( + &self, + connector_id: &str, + token: &str, + ) -> Result, String> { + let now_secs = u64::try_from(current_time_millis() / 1000).unwrap_or_default(); + let push = match self.push_auth.verify(token, now_secs).await { + Ok(push) => push, + Err(PushAuthError::Invalid(reason)) => { + debug!(connector_id, reason, "GCP push token refused"); + return Ok(None); + } + Err(PushAuthError::KeysUnavailable(error)) => return Err(error), + }; + // The audience is the receiver URL; its host is whatever the + // subscription was given, the path names the connector. + let audience_path = url::Url::parse(&push.audience) + .map(|url| url.path().to_owned()) + .unwrap_or_default(); + if audience_path != format!("/v1/logpush/gcp/{connector_id}") { + debug!(connector_id, "GCP push token is for another audience"); + return Ok(None); + } + + let cache_key = (connector_id.to_owned(), format!("oidc:{}", push.email)); + if let Some(identity) = self.cache.get(&cache_key).await { + let routing = self.routing.resolve_org_routing(&identity.org_id).await?; + return Ok(Some(identity.into_resolved(&routing))); + } + if self.negative_cache.contains_key(&cache_key) { + return Ok(None); + } + let row = self + .store + .fetch_gcp_connector_by_id(connector_id) + .await? + .filter(|row| push.email == gcp_push_service_account(connector_id, &row.project_id)); + let Some(row) = row else { + self.negative_cache.insert(cache_key, ()).await; + return Ok(None); + }; + + let routing = OrgRouting { + self_managed: row.self_managed, + clickhouse_ready: row.clickhouse_ready, + }; + let identity = GcpConnectorIdentity { + org_id: row.org_id, + // Names the signing account, as a secret's hash prefix names the secret. + secret_key_id: push.email.chars().take(16).collect(), + logs_enabled: row.logs_enabled, + confirms_log_forwarding: row.applied_logs_enabled.is_none(), + }; + self.cache.insert(cache_key, identity.clone()).await; + self.routing + .remember_org_routing(&identity.org_id, routing.clone()) + .await; + Ok(Some(identity.into_resolved(&routing))) + } + + /// Records log forwarding as set up, once. The cached identities stop + /// asking first, so a busy sink writes once rather than once per entry + /// until the write lands; a failed write lets a later push try again. + async fn confirm_log_forwarding(&self, connector_id: &str) { + // Collected first: the cache's iterator is not held across an await. + let pending: Vec<_> = self + .cache + .iter() + .filter(|(key, identity)| key.0 == connector_id && identity.confirms_log_forwarding) + .collect(); + for (key, identity) in &pending { + let confirmed = GcpConnectorIdentity { + confirms_log_forwarding: false, + ..identity.clone() + }; + self.cache.insert((**key).clone(), confirmed).await; + } + let report = gcp_logging::SetupReport { + logs: Some(true), + metrics: None, + }; + if let Err(error) = self + .store + .record_gcp_setup_report(connector_id, report, current_time_millis()) + .await + { + warn!(error = %error, connector_id, "Failed to record GCP log forwarding as set up"); + for (key, _) in &pending { + self.cache.invalidate(&**key).await; + } + } + } + + /// Writes `last_received_at` (`error == None`) or `last_error`, at most + /// once per connector and outcome per interval. Best-effort: the request's + /// outcome is already decided, so a failed write is only logged. + /// + /// A written failure lets the next success through, so the error does not + /// outlive the entry Pub/Sub delivers again: at most one failure and two + /// successes are written per interval. + async fn record_health(&self, connector_id: &str, error: Option<&str>) { + let throttle_key = (connector_id.to_owned(), error.is_none()); + let throttle = self.recent_health_writes.entry(throttle_key).or_insert(()); + if !throttle.await.is_fresh() { + return; + } + + let now_ms = current_time_millis(); + let written = match error { + None => { + self.store + .record_connector_success(GCP_CONNECTORS_TABLE, connector_id, now_ms) + .await + } + Some(error) => { + self.recent_health_writes + .invalidate(&(connector_id.to_owned(), true)) + .await; + self.store + .record_connector_failure(GCP_CONNECTORS_TABLE, connector_id, error, now_ms) + .await + } + }; + if let Err(error) = written { + debug!(connector_id, error, "Failed to record GCP connector health"); + } + } +} + +impl SamplingPolicyResolver { + #[hotpath::measure] + async fn resolve_policy(&self, org_id: &str) -> SamplingPolicy { + if let Some(policy) = self.cache.get(org_id).await { + return policy; + } + + let policy = match self.store.fetch_sampling_policy(org_id).await { + Ok(Some(row)) => SamplingPolicy { + trace_sample_ratio: row.trace_sample_ratio, + always_keep_error_spans: row.always_keep_error_spans, + always_keep_slow_spans_ms: row.always_keep_slow_spans_ms, + }, + Ok(None) => SamplingPolicy::default(), + Err(error) => { + warn!( + org_id, + error = %error, + "Sampling policy lookup failed; using unsampled default" + ); + SamplingPolicy::default() + } }; self.cache.insert(org_id.to_owned(), policy.clone()).await; policy @@ -5810,6 +6545,27 @@ fn schema_revision_at_least(stamped: &str, needed: i32) -> bool { const SCHEMA_REVISION_COMPATIBLE_SQL: &str = "(CASE WHEN s.schema_version ~ '^[0-9]+$' THEN s.schema_version::int ELSE -1 END >= $1)"; +/// A `GcpConnectorRow`'s columns over `gcp_connectors c` joined to +/// `org_clickhouse_settings s`, with the schema revision as `$1`. +fn gcp_connector_columns() -> String { + format!( + "c.org_id, c.project_id, c.logs_enabled, c.applied_logs_enabled, \ + COALESCE(s.sync_status = 'connected', false) AS self_managed, \ + COALESCE(s.sync_status = 'connected' AND {SCHEMA_REVISION_COMPATIBLE_SQL}, false) AS clickhouse_ready" + ) +} + +fn gcp_connector_row(row: &tokio_postgres::Row) -> GcpConnectorRow { + GcpConnectorRow { + org_id: row.get("org_id"), + project_id: row.get("project_id"), + logs_enabled: row.get("logs_enabled"), + applied_logs_enabled: row.get("applied_logs_enabled"), + self_managed: row.get("self_managed"), + clickhouse_ready: row.get("clickhouse_ready"), + } +} + #[async_trait::async_trait] impl ClickHouseTargetProvider for ClickHouseTargetResolver { async fn resolve_clickhouse_target( @@ -6106,6 +6862,60 @@ impl KeyStore for PostgresKeyStore { })) } + async fn fetch_gcp_connector( + &self, + connector_id: &str, + secret_hash: &str, + ) -> Result, String> { + let revision = required_schema_revision(); + let client = self.client().await?; + // No `enabled` column: deleting the row is how a connector is turned off. + let columns = gcp_connector_columns(); + let sql = format!( + "SELECT {columns} \ + FROM gcp_connectors c \ + LEFT JOIN org_clickhouse_settings s ON s.org_id = c.org_id \ + WHERE c.id = $2 AND c.secret_hash = $3 LIMIT 1" + ); + let rows = client + .query(&sql, &[&revision, &connector_id, &secret_hash]) + .instrument(postgres_client_span( + "fetch_gcp_connector", + "SELECT", + GCP_CONNECTORS_TABLE, + &self.target, + )) + .await + .map_err(|error| format!("postgres fetch_gcp_connector failed: {error}"))?; + Ok(rows.first().map(gcp_connector_row)) + } + + async fn fetch_gcp_connector_by_id( + &self, + connector_id: &str, + ) -> Result, String> { + let revision = required_schema_revision(); + let client = self.client().await?; + let columns = gcp_connector_columns(); + let sql = format!( + "SELECT {columns} \ + FROM gcp_connectors c \ + LEFT JOIN org_clickhouse_settings s ON s.org_id = c.org_id \ + WHERE c.id = $2 LIMIT 1" + ); + let rows = client + .query(&sql, &[&revision, &connector_id]) + .instrument(postgres_client_span( + "fetch_gcp_connector_by_id", + "SELECT", + GCP_CONNECTORS_TABLE, + &self.target, + )) + .await + .map_err(|error| format!("postgres fetch_gcp_connector_by_id failed: {error}"))?; + Ok(rows.first().map(gcp_connector_row)) + } + async fn fetch_sampling_policy( &self, org_id: &str, @@ -6233,23 +7043,26 @@ impl KeyStore for PostgresKeyStore { async fn record_connector_success( &self, + table: &'static str, connector_id: &str, now_ms: i64, ) -> Result<(), String> { let client = self.client().await?; client .execute( - "UPDATE cloudflare_logpush_connectors \ - SET last_received_at = to_timestamp($1::bigint / 1000.0), \ - last_error = NULL, \ - updated_at = to_timestamp($1::bigint / 1000.0) \ - WHERE id = $2", + &format!( + "UPDATE {table} \ + SET last_received_at = to_timestamp($1::bigint / 1000.0), \ + last_error = NULL, \ + updated_at = to_timestamp($1::bigint / 1000.0) \ + WHERE id = $2" + ), &[&now_ms, &connector_id], ) .instrument(postgres_client_span( "record_connector_success", "UPDATE", - "cloudflare_logpush_connectors", + table, &self.target, )) .await @@ -6259,6 +7072,7 @@ impl KeyStore for PostgresKeyStore { async fn record_connector_failure( &self, + table: &'static str, connector_id: &str, error: &str, now_ms: i64, @@ -6266,26 +7080,59 @@ impl KeyStore for PostgresKeyStore { let client = self.client().await?; client .execute( - "UPDATE cloudflare_logpush_connectors \ - SET last_error = $1, updated_at = to_timestamp($2::bigint / 1000.0) \ - WHERE id = $3", + &format!( + "UPDATE {table} \ + SET last_error = $1, updated_at = to_timestamp($2::bigint / 1000.0) \ + WHERE id = $3" + ), &[&error, &now_ms, &connector_id], ) .instrument(postgres_client_span( "record_connector_failure", "UPDATE", - "cloudflare_logpush_connectors", + table, &self.target, )) .await .map(|_| ()) .map_err(|err| format!("postgres record_connector_failure failed: {err}")) } + + async fn record_gcp_setup_report( + &self, + connector_id: &str, + report: gcp_logging::SetupReport, + now_ms: i64, + ) -> Result<(), String> { + let client = self.client().await?; + // A capability the run did not report keeps what an earlier run said. + // Log forwarding set up again answers whatever the last push failed on. + client + .execute( + "UPDATE gcp_connectors \ + SET applied_logs_enabled = COALESCE($1::boolean, applied_logs_enabled), \ + applied_metrics_enabled = COALESCE($2::boolean, applied_metrics_enabled), \ + setup_reported_at = to_timestamp($3::bigint / 1000.0), \ + last_error = CASE WHEN $1::boolean THEN NULL ELSE last_error END, \ + updated_at = to_timestamp($3::bigint / 1000.0) \ + WHERE id = $4", + &[&report.logs, &report.metrics, &now_ms, &connector_id], + ) + .instrument(postgres_client_span( + "record_gcp_setup_report", + "UPDATE", + GCP_CONNECTORS_TABLE, + &self.target, + )) + .await + .map(|_| ()) + .map_err(|error| format!("postgres record_gcp_setup_report failed: {error}")) + } } // Local-dev / single-tenant KeyStore: every well-formed ingest key resolves to // the configured org. No DB, no network. Connector flows are no-ops since -// Cloudflare Logpush is a production-only integration. +// the logpush receivers need connector rows, which only Postgres holds. struct StaticKeyStore { org_id: String, } @@ -6312,6 +7159,21 @@ impl KeyStore for StaticKeyStore { Ok(None) } + async fn fetch_gcp_connector( + &self, + _connector_id: &str, + _secret_hash: &str, + ) -> Result, String> { + Ok(None) + } + + async fn fetch_gcp_connector_by_id( + &self, + _connector_id: &str, + ) -> Result, String> { + Ok(None) + } + async fn fetch_sampling_policy( &self, _org_id: &str, @@ -6339,6 +7201,7 @@ impl KeyStore for StaticKeyStore { async fn record_connector_success( &self, + _table: &'static str, _connector_id: &str, _now_ms: i64, ) -> Result<(), String> { @@ -6347,12 +7210,22 @@ impl KeyStore for StaticKeyStore { async fn record_connector_failure( &self, + _table: &'static str, _connector_id: &str, _error: &str, _now_ms: i64, ) -> Result<(), String> { Ok(()) } + + async fn record_gcp_setup_report( + &self, + _connector_id: &str, + _report: gcp_logging::SetupReport, + _now_ms: i64, + ) -> Result<(), String> { + Ok(()) + } } fn infer_ingest_key_type(raw_key: &str) -> Option { @@ -7441,6 +8314,13 @@ mod tests { struct FakeKeyStore { keys: std::sync::Mutex>, connectors: std::sync::Mutex>, + gcp_connectors: + std::sync::Mutex>, + connector_successes: AtomicU64, + /// (table, error) of every `record_connector_failure` call. + connector_failures: std::sync::Mutex>, + /// (connector id, report) of every `record_gcp_setup_report` call. + gcp_setup_reports: std::sync::Mutex>, routings: std::sync::Mutex>, targets: std::sync::Mutex>, ingest_key_fetches: AtomicU64, @@ -7481,6 +8361,21 @@ mod tests { .insert((connector_id.to_owned(), hash), row); } + fn insert_gcp_connector(&self, connector_id: &str, raw_secret: &str, row: GcpConnectorRow) { + let hash = hash_ingest_key(raw_secret, "test-hmac-key").unwrap(); + self.set_org_routing( + &row.org_id, + OrgRouting { + self_managed: row.self_managed, + clickhouse_ready: row.clickhouse_ready, + }, + ); + self.gcp_connectors + .lock() + .unwrap() + .insert((connector_id.to_owned(), hash), row); + } + fn set_org_routing(&self, org_id: &str, routing: OrgRouting) { self.routings .lock() @@ -7521,6 +8416,30 @@ mod tests { .get(&(connector_id.to_owned(), secret_hash.to_owned())) .cloned()) } + async fn fetch_gcp_connector( + &self, + connector_id: &str, + secret_hash: &str, + ) -> Result, String> { + Ok(self + .gcp_connectors + .lock() + .unwrap() + .get(&(connector_id.to_owned(), secret_hash.to_owned())) + .cloned()) + } + async fn fetch_gcp_connector_by_id( + &self, + connector_id: &str, + ) -> Result, String> { + Ok(self + .gcp_connectors + .lock() + .unwrap() + .iter() + .find(|((id, _), _)| id == connector_id) + .map(|(_, row)| row.clone())) + } async fn fetch_sampling_policy( &self, _org_id: &str, @@ -7549,17 +8468,36 @@ mod tests { } async fn record_connector_success( &self, + _table: &'static str, _connector_id: &str, _now_ms: i64, ) -> Result<(), String> { + self.connector_successes.fetch_add(1, Ordering::Relaxed); Ok(()) } async fn record_connector_failure( &self, + table: &'static str, _connector_id: &str, - _error: &str, + error: &str, + _now_ms: i64, + ) -> Result<(), String> { + self.connector_failures + .lock() + .unwrap() + .push((table, error.to_owned())); + Ok(()) + } + async fn record_gcp_setup_report( + &self, + connector_id: &str, + report: gcp_logging::SetupReport, _now_ms: i64, ) -> Result<(), String> { + self.gcp_setup_reports + .lock() + .unwrap() + .push((connector_id.to_owned(), report)); Ok(()) } } @@ -7791,6 +8729,7 @@ mod tests { tinybird, max_request_body_bytes: 1024 * 1024, org_max_in_flight: 100, + gcp_org_max_in_flight: 100, request_timeout: Duration::from_secs(30), require_tls: false, key_store_backend: KeyStoreBackend::Static { @@ -7833,6 +8772,7 @@ mod tests { routing: Arc::clone(&routing), }, org_inflight_limiter: OrgInFlightLimiter::new(100), + gcp_inflight_limiter: OrgInFlightLimiter::for_route(100, "gcp_in_flight"), sampling_resolver: SamplingPolicyResolver { store: Arc::clone(&key_store), cache: Cache::builder() @@ -7848,14 +8788,21 @@ mod tests { .build(), }, cloudflare_resolver: CloudflareConnectorResolver { - store: key_store, + store: Arc::clone(&key_store), lookup_hmac_key: "test-hmac-key".to_owned(), cache: Cache::builder() .time_to_live(Duration::from_mins(1)) .max_capacity(16) .build(), - routing, + routing: Arc::clone(&routing), }, + gcp_resolver: GcpConnectorResolver::new( + key_store, + "test-hmac-key".to_owned(), + routing, + Duration::from_mins(1), + GooglePushAuth::with_keys(gcp_push_test_keys()), + ), autumn_tracker: None, autumn_entitlements: None, usage_metrics: None, @@ -8857,6 +9804,449 @@ mod tests { assert_eq!(store.routing_fetches.load(Ordering::Relaxed), 1); } + const GCP_LOG_ENTRY: &str = r#"{ + "logName": "projects/my-project/logs/run.googleapis.com%2Fstdout", + "insertId": "insert-1", + "timestamp": "2025-03-07T12:34:56.123456789Z", + "severity": "WARNING", + "textPayload": "cache miss", + "resource": { + "type": "cloud_run_revision", + "labels": { + "project_id": "my-project", + "service_name": "checkout", + "revision_name": "checkout-00042-abc", + "location": "europe-west4" + } + } + }"#; + + async fn spawn_fake_clickhouse() -> ( + String, + tokio::sync::mpsc::UnboundedReceiver, + ) { + let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)) + .await + .unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let app = Router::new() + .route("/", post(fake_clickhouse_import)) + .with_state(tx); + tokio::spawn(async move { + axum::serve(listener, app).await.unwrap(); + }); + (url, rx) + } + + /// One org with three connectors: `gcp_conn_1`, `gcp_conn_off` whose log + /// forwarding is turned off, both set up by an earlier script that put the + /// secret in the push URL, and `GCP_SIGNED_CONNECTOR`, set up with + /// Terraform and never reported on, whose pushes Pub/Sub signs. + async fn gcp_logpush_state( + name: &str, + clickhouse_url: &str, + ) -> (Arc, Arc, PathBuf) { + let queue_dir = unique_main_test_dir(name); + let store = Arc::new(FakeKeyStore::default()); + for (connector_id, secret, logs_enabled, applied_logs_enabled) in [ + ("gcp_conn_1", "gcp-secret", true, Some(true)), + ("gcp_conn_off", "gcp-secret-off", false, Some(true)), + (GCP_SIGNED_CONNECTOR, "gcp-secret-signed", true, None), + ] { + store.insert_gcp_connector( + connector_id, + secret, + GcpConnectorRow { + org_id: "org_gcp".to_owned(), + project_id: "acme-host".to_owned(), + logs_enabled, + applied_logs_enabled, + self_managed: true, + clickhouse_ready: true, + }, + ); + } + store.insert_clickhouse_target( + "org_gcp", + ClickHouseTargetRow { + ch_url: clickhouse_url.to_owned(), + ch_user: "ingest".to_owned(), + ch_password_ciphertext: None, + ch_password_iv: None, + ch_password_tag: None, + ch_database: "maple".to_owned(), + schema_version: CLICKHOUSE_SCHEMA_VERSION.to_owned(), + }, + ); + let state = test_app_state( + Arc::clone(&store), + queue_dir.clone(), + "http://127.0.0.1:1".to_owned(), + Duration::from_mins(1), + ) + .await; + (Arc::new(state), store, queue_dir) + } + + async fn gcp_push( + state: &Arc, + connector_id: &str, + secret: Option<&str>, + body: &'static str, + ) -> Response { + handle_gcp_logpush( + State(Arc::clone(state)), + Path(connector_id.to_owned()), + Query(LogpushQuery { + secret: secret.map(str::to_owned), + }), + HeaderMap::new(), + Bytes::from_static(body.as_bytes()), + ) + .await + } + + const GCP_SIGNED_CONNECTOR: &str = "0a1b2c3d-4e5f-4071-8293-a4b5c6d7e8f9"; + const GCP_SIGNED_ACCOUNT: &str = + "maple-push-0a1b2c3d4e5f4071829@acme-host.iam.gserviceaccount.com"; + const GCP_PUSH_TEST_KEY: &[u8] = include_bytes!("testdata/gcp-push-test-key.pk8"); + const GCP_PUSH_TEST_KEY_N: &str = include_str!("testdata/gcp-push-test-key.n"); + + fn gcp_push_test_keys() -> maple_ingest::gcp_push_auth::Keys { + maple_ingest::gcp_push_auth::Keys::from([( + "key-1".to_owned(), + maple_ingest::gcp_push_auth::RsaKey { + n: URL_SAFE_NO_PAD.decode(GCP_PUSH_TEST_KEY_N.trim()).unwrap(), + e: vec![1, 0, 1], + }, + )]) + } + + /// An ID token as Pub/Sub signs it for a push subscription's account. + fn gcp_push_token(email: &str, audience: &str) -> String { + use ring::signature::{RsaKeyPair, RSA_PKCS1_SHA256}; + let now = current_time_millis() / 1000; + let encode = |value: serde_json::Value| URL_SAFE_NO_PAD.encode(value.to_string()); + let signed = format!( + "{}.{}", + encode(serde_json::json!({ "alg": "RS256", "kid": "key-1", "typ": "JWT" })), + encode(serde_json::json!({ + "iss": "https://accounts.google.com", + "aud": audience, + "iat": now, + "exp": now + 3600, + "email": email, + "email_verified": true, + })), + ); + let pair = RsaKeyPair::from_pkcs8(GCP_PUSH_TEST_KEY).unwrap(); + let mut signature = vec![0; pair.public().modulus_len()]; + pair.sign( + &RSA_PKCS1_SHA256, + &ring::rand::SystemRandom::new(), + signed.as_bytes(), + &mut signature, + ) + .unwrap(); + format!("{signed}.{}", URL_SAFE_NO_PAD.encode(signature)) + } + + async fn gcp_signed_push( + state: &Arc, + connector_id: &str, + token: &str, + body: &'static str, + ) -> Response { + let mut headers = HeaderMap::new(); + headers.insert( + AUTHORIZATION, + HeaderValue::from_str(&format!("Bearer {token}")).unwrap(), + ); + handle_gcp_logpush( + State(Arc::clone(state)), + Path(connector_id.to_owned()), + Query(LogpushQuery { secret: None }), + headers, + Bytes::from_static(body.as_bytes()), + ) + .await + } + + #[tokio::test] + async fn gcp_logpush_accepts_a_push_signed_as_the_connectors_push_account() { + let (ch_url, mut ch_rx) = spawn_fake_clickhouse().await; + let (state, store, queue_dir) = gcp_logpush_state("gcp-logpush-signed", &ch_url).await; + let token = gcp_push_token( + GCP_SIGNED_ACCOUNT, + &format!("https://ingest.test/v1/logpush/gcp/{GCP_SIGNED_CONNECTOR}"), + ); + + for _ in 0..2 { + let response = + gcp_signed_push(&state, GCP_SIGNED_CONNECTOR, &token, GCP_LOG_ENTRY).await; + assert_eq!(response.status(), StatusCode::OK); + tokio::time::timeout(Duration::from_secs(2), ch_rx.recv()) + .await + .expect("the entry is stored") + .expect("fake ClickHouse is running"); + } + // Nothing reported on it before: the first push proves log forwarding + // is set up, and only the first records it. + assert_eq!( + *store.gcp_setup_reports.lock().unwrap(), + [( + GCP_SIGNED_CONNECTOR.to_owned(), + gcp_logging::SetupReport { + logs: Some(true), + metrics: None + } + )] + ); + + drop(std::fs::remove_dir_all(queue_dir)); + } + + #[tokio::test] + async fn gcp_logpush_refuses_a_token_for_another_account_or_connector() { + let (state, store, queue_dir) = + gcp_logpush_state("gcp-logpush-signed-refused", "http://127.0.0.1:1").await; + let audience = format!("https://ingest.test/v1/logpush/gcp/{GCP_SIGNED_CONNECTOR}"); + + for (case, connector_id, token) in [ + ( + "the same name in another project", + GCP_SIGNED_CONNECTOR, + gcp_push_token( + "maple-push-0a1b2c3d4e5f4071829@attacker.iam.gserviceaccount.com", + &audience, + ), + ), + ( + "a token for another connector's endpoint", + GCP_SIGNED_CONNECTOR, + gcp_push_token( + GCP_SIGNED_ACCOUNT, + "https://ingest.test/v1/logpush/gcp/gcp_conn_1", + ), + ), + ( + "another connector's endpoint with this token", + "gcp_conn_1", + gcp_push_token(GCP_SIGNED_ACCOUNT, &audience), + ), + ( + "not a token", + GCP_SIGNED_CONNECTOR, + "gcp-secret-signed".to_owned(), + ), + ] { + let response = gcp_signed_push(&state, connector_id, &token, GCP_LOG_ENTRY).await; + assert_eq!(response.status(), StatusCode::UNAUTHORIZED, "{case}"); + } + assert!(store.gcp_setup_reports.lock().unwrap().is_empty()); + + drop(std::fs::remove_dir_all(queue_dir)); + } + + #[tokio::test] + async fn gcp_logpush_rejects_bad_credentials() { + let (state, _store, queue_dir) = + gcp_logpush_state("gcp-logpush-auth", "http://127.0.0.1:1").await; + + for (case, connector_id, secret) in [ + ("missing secret", "gcp_conn_1", None), + ("wrong secret", "gcp_conn_1", Some("wrong-secret")), + ("unknown connector", "gcp_conn_unknown", Some("gcp-secret")), + ("malformed id", "not/a connector id", Some("gcp-secret")), + ] { + let response = gcp_push(&state, connector_id, secret, GCP_LOG_ENTRY).await; + assert_eq!(response.status(), StatusCode::UNAUTHORIZED, "{case}"); + } + + drop(std::fs::remove_dir_all(queue_dir)); + } + + #[tokio::test] + async fn gcp_logpush_stores_a_log_entry_and_acks_a_poison_payload() { + let (ch_url, mut ch_rx) = spawn_fake_clickhouse().await; + let (state, store, queue_dir) = gcp_logpush_state("gcp-logpush-accept", &ch_url).await; + + let accepted = gcp_push(&state, "gcp_conn_1", Some("gcp-secret"), GCP_LOG_ENTRY).await; + assert_eq!(accepted.status(), StatusCode::OK); + let stored = tokio::time::timeout(Duration::from_secs(2), ch_rx.recv()) + .await + .expect("the entry should reach the org's ClickHouse") + .expect("ClickHouse channel should stay open"); + assert!(stored.query.starts_with("INSERT INTO logs")); + assert!(stored.body.contains("cache miss"), "{}", stored.body); + assert!( + stored.body.contains(r#""service_name":"checkout""#), + "{}", + stored.body + ); + assert_eq!(store.connector_successes.load(Ordering::Relaxed), 1); + assert!(store.connector_failures.lock().unwrap().is_empty()); + + // The wrapped Pub/Sub envelope can never parse, and Pub/Sub redelivers + // anything but an ack, so it is acked. Sent twice: the connector's + // `last_error` write is throttled to one. + for _ in 0..2 { + let poison = gcp_push( + &state, + "gcp_conn_1", + Some("gcp-secret"), + r#"{"message":{"data":"e30=","messageId":"1"},"subscription":"s"}"#, + ) + .await; + assert_eq!(poison.status(), StatusCode::OK); + } + assert_eq!( + *store.connector_failures.lock().unwrap(), + [( + GCP_CONNECTORS_TABLE, + gcp_logging::NOT_A_LOG_ENTRY.to_owned() + )] + ); + assert!( + tokio::time::timeout(Duration::from_millis(50), ch_rx.recv()) + .await + .is_err(), + "a poison payload must not be stored" + ); + + // The next accepted entry is recorded at once and clears the error, + // although a success was already written within the interval. + let again = gcp_push(&state, "gcp_conn_1", Some("gcp-secret"), GCP_LOG_ENTRY).await; + assert_eq!(again.status(), StatusCode::OK); + assert_eq!(store.connector_successes.load(Ordering::Relaxed), 2); + + drop(std::fs::remove_dir_all(queue_dir)); + } + + #[tokio::test] + async fn gcp_logpush_acks_and_drops_while_log_forwarding_is_off() { + let (ch_url, mut ch_rx) = spawn_fake_clickhouse().await; + let (state, store, queue_dir) = gcp_logpush_state("gcp-logpush-off", &ch_url).await; + + // Acked: a non-2xx would have Pub/Sub redeliver the entry for the + // subscription's whole retention. + let dropped = gcp_push( + &state, + "gcp_conn_off", + Some("gcp-secret-off"), + GCP_LOG_ENTRY, + ) + .await; + assert_eq!(dropped.status(), StatusCode::OK); + assert!( + tokio::time::timeout(Duration::from_millis(50), ch_rx.recv()) + .await + .is_err(), + "nothing may be stored while log forwarding is off" + ); + assert_eq!(store.connector_successes.load(Ordering::Relaxed), 0); + assert!(store.connector_failures.lock().unwrap().is_empty()); + + drop(std::fs::remove_dir_all(queue_dir)); + } + + #[tokio::test] + async fn gcp_logpush_records_a_setup_report_and_stores_nothing() { + const LOGS_ON: &str = r#"{"logName":"projects/p/logs/maple-setup","timestamp":"2026-10-09T10:00:00Z","jsonPayload":{"logs":true}}"#; + const LOGS_OFF: &str = r#"{"logName":"projects/p/logs/maple-setup","timestamp":"2026-10-09T10:00:00Z","jsonPayload":{"logs":false,"metrics":true}}"#; + let (ch_url, mut ch_rx) = spawn_fake_clickhouse().await; + let (state, store, queue_dir) = gcp_logpush_state("gcp-logpush-report", &ch_url).await; + + // Through the topic of a connector that forwards logs, and straight + // from the script of one whose log forwarding is off. + for (connector_id, secret, body) in [ + ("gcp_conn_1", "gcp-secret", LOGS_ON), + ("gcp_conn_off", "gcp-secret-off", LOGS_OFF), + ] { + let response = gcp_push(&state, connector_id, Some(secret), body).await; + assert_eq!(response.status(), StatusCode::OK, "{connector_id}"); + } + // Another connector's secret reports for nobody. + let forged = gcp_push(&state, "gcp_conn_off", Some("gcp-secret"), LOGS_ON).await; + assert_eq!(forged.status(), StatusCode::UNAUTHORIZED); + + assert_eq!( + *store.gcp_setup_reports.lock().unwrap(), + [ + ( + "gcp_conn_1".to_owned(), + gcp_logging::SetupReport { + logs: Some(true), + metrics: None + } + ), + ( + "gcp_conn_off".to_owned(), + gcp_logging::SetupReport { + logs: Some(false), + metrics: Some(true) + } + ), + ] + ); + assert!( + tokio::time::timeout(Duration::from_millis(50), ch_rx.recv()) + .await + .is_err(), + "a setup report must not be stored as a log" + ); + assert_eq!(store.connector_successes.load(Ordering::Relaxed), 0); + assert!(store.connector_failures.lock().unwrap().is_empty()); + + drop(std::fs::remove_dir_all(queue_dir)); + } + + #[tokio::test] + async fn gcp_logpush_refuses_pushes_over_the_org_gcp_in_flight_cap() { + let (ch_url, _ch_rx) = spawn_fake_clickhouse().await; + let (state, _store, queue_dir) = gcp_logpush_state("gcp-logpush-cap", &ch_url).await; + + // Every GCP slot of the org taken by pushes still in progress. + let held: Vec<_> = (0..state.config.gcp_org_max_in_flight) + .map(|_| state.gcp_inflight_limiter.try_acquire("org_gcp").unwrap()) + .collect(); + let refused = gcp_push(&state, "gcp_conn_1", Some("gcp-secret"), GCP_LOG_ENTRY).await; + assert_eq!(refused.status(), StatusCode::TOO_MANY_REQUESTS); + // The org's other ingest keeps its own slots. + assert!(state.org_inflight_limiter.try_acquire("org_gcp").is_some()); + + drop(held); + let accepted = gcp_push(&state, "gcp_conn_1", Some("gcp-secret"), GCP_LOG_ENTRY).await; + assert_eq!(accepted.status(), StatusCode::OK); + + drop(std::fs::remove_dir_all(queue_dir)); + } + + #[tokio::test] + async fn gcp_logpush_refuses_setup_reports_past_the_window_limit() { + const REPORT: &str = r#"{"logName":"projects/p/logs/maple-setup","timestamp":"2026-10-09T10:00:00Z","jsonPayload":{"logs":true}}"#; + let (state, store, queue_dir) = + gcp_logpush_state("gcp-logpush-report-limit", "http://127.0.0.1:1").await; + + for _ in 0..GCP_SETUP_REPORTS_PER_WINDOW { + let response = gcp_push(&state, "gcp_conn_1", Some("gcp-secret"), REPORT).await; + assert_eq!(response.status(), StatusCode::OK); + } + let refused = gcp_push(&state, "gcp_conn_1", Some("gcp-secret"), REPORT).await; + assert_eq!(refused.status(), StatusCode::TOO_MANY_REQUESTS); + // The limit is per connector. + let other = gcp_push(&state, "gcp_conn_off", Some("gcp-secret-off"), REPORT).await; + assert_eq!(other.status(), StatusCode::OK); + + assert_eq!( + store.gcp_setup_reports.lock().unwrap().len(), + GCP_SETUP_REPORTS_PER_WINDOW as usize + 1 + ); + + drop(std::fs::remove_dir_all(queue_dir)); + } + /// Regression for the gateway going silent in its own traces: a global /// `warn` filter (the #581 default) discards every `info_span!` before the /// OTel layer sees it. The span filter must admit info spans; the log diff --git a/apps/ingest/src/metrics.rs b/apps/ingest/src/metrics.rs index 355dcb648..807547eaa 100644 --- a/apps/ingest/src/metrics.rs +++ b/apps/ingest/src/metrics.rs @@ -111,6 +111,34 @@ static CLOUDFLARE_RECORDS_TOTAL: LazyLock> = LazyLock::new(|| { .build() }); +static GCP_ENTRIES_TOTAL: LazyLock> = LazyLock::new(|| { + METER + .u64_counter("ingest_gcp_entries_total") + .with_description("Cloud Logging entries accepted from GCP Pub/Sub pushes") + .build() +}); + +static GCP_AUTH_FAILURES_TOTAL: LazyLock> = LazyLock::new(|| { + METER + .u64_counter("ingest_gcp_auth_failures_total") + .with_description("GCP Pub/Sub pushes rejected for bad auth") + .build() +}); + +static GCP_PARSE_FAILURES_TOTAL: LazyLock> = LazyLock::new(|| { + METER + .u64_counter("ingest_gcp_parse_failures_total") + .with_description("GCP Pub/Sub pushes acked and dropped: body was not a LogEntry") + .build() +}); + +static GCP_DISABLED_DROPS_TOTAL: LazyLock> = LazyLock::new(|| { + METER + .u64_counter("ingest_gcp_disabled_drops_total") + .with_description("GCP Pub/Sub pushes acked and dropped: log forwarding is off") + .build() +}); + static SENTINEL_TOTAL: LazyLock> = LazyLock::new(|| { METER .u64_counter("ingest_sentinel_total") @@ -597,6 +625,27 @@ pub fn cloudflare_records(dataset: &str, count: u64) { CLOUDFLARE_RECORDS_TOTAL.add(count, &[KeyValue::new("dataset", dataset.to_owned())]); } +/// A Cloud Logging entry pushed by GCP Pub/Sub was accepted. +pub fn gcp_entry() { + GCP_ENTRIES_TOTAL.add(1, &[]); +} + +/// A GCP Pub/Sub push failed authentication. +pub fn gcp_auth_failure() { + GCP_AUTH_FAILURES_TOTAL.add(1, &[]); +} + +/// A GCP Pub/Sub push was acked and dropped because its body was not a LogEntry. +pub fn gcp_parse_failure() { + GCP_PARSE_FAILURES_TOTAL.add(1, &[]); +} + +/// A GCP Pub/Sub push was acked and dropped because the connector has log +/// forwarding turned off. +pub fn gcp_disabled_drop() { + GCP_DISABLED_DROPS_TOTAL.add(1, &[]); +} + /// A WAL append was rejected because the lane file is full. `shard` is the real /// shard; `destination` distinguishes the per-destination lane within it. pub fn wal_shard_full(shard: usize, destination: &str) { diff --git a/apps/ingest/src/testdata/gcp-push-test-key.n b/apps/ingest/src/testdata/gcp-push-test-key.n new file mode 100644 index 000000000..423ab8909 --- /dev/null +++ b/apps/ingest/src/testdata/gcp-push-test-key.n @@ -0,0 +1 @@ +xw7kNmTxqS6r6IoeOEQW1iQ1oWnhL99UYQgthpdP8KgzW-FVhltnHG-X8tu8t-GIaC9j_Uhf9YH3268I0hpFbakiGNwy-WnctTy4UG5kY-S1RlTaTEgbIuI-N6BUoCCCbLQCBg3IP9e7O-c2fAZQ1pn7EAIV3qDtTGE4b4aBRrD3f6pJ-h8UgnhWn4u-1QzzhdPgRe22E5lceZyuMiFtZhzyxvak82FirJ5xGgJx-zVYpGDjEioczBLM-RR6LTFYo9Zk7wZugaGl0dgQfqwpQp3fMf0POjU0QDa5Dct76HvFf6CWAEUdwZpYlfdRJOOgZFe49bhmKhB_yTBOQpVocw \ No newline at end of file diff --git a/apps/ingest/src/testdata/gcp-push-test-key.pk8 b/apps/ingest/src/testdata/gcp-push-test-key.pk8 new file mode 100644 index 000000000..d39622efe Binary files /dev/null and b/apps/ingest/src/testdata/gcp-push-test-key.pk8 differ diff --git a/apps/web/perf/check-bundle-budget.ts b/apps/web/perf/check-bundle-budget.ts index aa2d704fb..78668ea33 100644 --- a/apps/web/perf/check-bundle-budget.ts +++ b/apps/web/perf/check-bundle-budget.ts @@ -105,6 +105,7 @@ const gzipBytes = chunks.reduce((total, chunk) => total + chunk.gzipBytes, 0) // #1502) — the connector group's endpoints and schemas on the v2 contract // every page's client carries. Each Google Cloud step below keeps ~1 KB of // headroom, since CI measures the PR merged with main. +// 699.5 KB with main's 1 KB from #1494, #1500, #1502 merged up the stack. const maxGzipBytes = 699.5 * 1024 const budgetLabel = `${(maxGzipBytes / 1024).toFixed(1)} KB`