diff --git a/apps/ingest/Dockerfile.prebuilt b/apps/ingest/Dockerfile.prebuilt index ac45b59c2d..e6392ea8e4 100644 --- a/apps/ingest/Dockerfile.prebuilt +++ b/apps/ingest/Dockerfile.prebuilt @@ -19,7 +19,7 @@ # The compile MUST happen against bookworm's glibc (2.36) — building on the # ubuntu-24.04 runner (2.39) instead produces a binary that dies at startup with # `version 'GLIBC_2.39' not found`. The workflow enforces this by compiling -# inside rust:1.88-bookworm rather than on the host. +# inside rust:1.94-bookworm rather than on the host. # # `Dockerfile` (the self-contained source build) stays the default and is what a # local `alchemy deploy` uses; alchemy.run.ts picks this one only when diff --git a/apps/ingest/alchemy.run.ts b/apps/ingest/alchemy.run.ts index b7e8b91245..57db62b136 100644 --- a/apps/ingest/alchemy.run.ts +++ b/apps/ingest/alchemy.run.ts @@ -699,7 +699,6 @@ export const createMapleIngest = ({ stage, region, domains, profile, dbLogin }: // keys again after normalizing the header, so a respelled one is still blocked. ...(yield* Config.map(blockedIngestKeys, blockedIngestKeysEnv)), ...(yield* optionalPlain("INGEST_EXPORT_MAX_ATTEMPTS")), - ...(yield* optionalPlain("INGEST_TINYBIRD_CONCURRENCY_PER_SHARD")), ...(yield* optionalPlain("INGEST_REPLAY_MAX_SESSION_BYTES")), // The org Maple's own telemetry is filed under. Required, no fallback. MAPLE_INTERNAL_ORG_ID: yield* requiredPlain("MAPLE_INTERNAL_ORG_ID"), diff --git a/apps/ingest/benches/ingest_bench.rs b/apps/ingest/benches/ingest_bench.rs index b6b665705e..5ec9323b40 100644 --- a/apps/ingest/benches/ingest_bench.rs +++ b/apps/ingest/benches/ingest_bench.rs @@ -122,7 +122,6 @@ impl BenchFixture { batch_max_bytes: 4 * 1024 * 1024, queue_max_age: Duration::ZERO, batch_max_wait: Duration::from_millis(10), - export_concurrency_per_shard: 1, export_max_attempts: 20, clickhouse_export_timeout: Duration::from_secs(5), clickhouse_breaker: ClickHouseBreakerConfig::default(), diff --git a/apps/ingest/src/main.rs b/apps/ingest/src/main.rs index 0ab806f4ae..94765134ee 100644 --- a/apps/ingest/src/main.rs +++ b/apps/ingest/src/main.rs @@ -276,6 +276,16 @@ fn uses_forward_path_for(write_mode: WriteMode, destination: ExportDestination) write_mode.uses_forward() && destination != ExportDestination::ClickHouse } +/// Whether a request left rows behind: the collector took it, or the native +/// pipeline had already accepted it before the forward (dual mode). +fn batch_was_stored( + forwarded_ok: bool, + write_mode: WriteMode, + destination: ExportDestination, +) -> bool { + forwarded_ok || uses_native_pipeline_for(write_mode, destination) +} + #[derive(Clone)] enum KeyStoreBackend { // No-DB local backend: every well-formed ingest key resolves to a single @@ -411,11 +421,6 @@ impl AppConfig { std::env::var("INGEST_BATCH_MAX_WAIT_MS").ok(), 100, )?), - export_concurrency_per_shard: parse_usize( - "INGEST_TINYBIRD_CONCURRENCY_PER_SHARD", - std::env::var("INGEST_TINYBIRD_CONCURRENCY_PER_SHARD").ok(), - 1, - )?, export_max_attempts: parse_u32( "INGEST_EXPORT_MAX_ATTEMPTS", std::env::var("INGEST_EXPORT_MAX_ATTEMPTS").ok(), @@ -3295,8 +3300,9 @@ async fn main() { // Even a clean drain runs this: it marks the owner retired, so a // successor claims anything left instead of waiting out the staleness // window. With a backlog it also seals and ships the tail, which is the - // difference between "replays if this task's storage survives" (it does - // not — Fargate ephemeral storage dies with the task) and "replays". + // difference between "replays if this task's storage survives" (the + // instance store under `/mnt/wal` is wiped when the host is stopped or + // replaced) and "replays". let shipped = pipeline.flush_wal_to_object_store().await; if shipped > 0 { info!( @@ -3868,6 +3874,16 @@ async fn accept_grpc_decoded( .await; match result { + // A collector refusal that left nothing stored is neither key use nor billed. + Ok(response) + if !batch_was_stored( + response.status().is_success(), + state.config.write_mode, + native_destination_for(resolved), + ) => + { + Ok(()) + } Ok(_) => { state.resolver.note_source_use(resolved, claims.kept_claim); state @@ -5664,7 +5680,13 @@ async fn handle_signal_inner( let kind = error.error_kind(); (error, if kind == INGEST_RULES_SCAN_BUDGET.error_kind { kind } else { "forward" }) })?; - if response.status().is_success() { + // A collector refusal that left nothing stored is not key use, and 0 bytes to meter. + let stored = batch_was_stored( + response.status().is_success(), + state.config.write_mode, + native_destination_for(&resolved_key), + ); + if stored { state.resolver.note_source_use(&resolved_key, kept_claim); state .resolver @@ -5675,7 +5697,7 @@ async fn handle_signal_inner( response, item_count, resolved_key.org_id.clone(), - decoded_bytes, + if stored { decoded_bytes } else { 0 }, )) } @@ -11217,6 +11239,26 @@ mod tests { )); } + #[test] + fn a_refused_relay_still_counts_when_the_native_pipeline_took_the_batch() { + let refused = false; + assert!(!batch_was_stored( + refused, + WriteMode::Forward, + ExportDestination::Tinybird + )); + assert!(batch_was_stored( + refused, + WriteMode::Dual, + ExportDestination::Tinybird + )); + assert!(batch_was_stored( + true, + WriteMode::Forward, + ExportDestination::Tinybird + )); + } + /// In-memory KeyStore used to exercise the resolver's behavior (caching, /// key-type inference, ResolvedIngestKey construction) without a database. /// Keyed on the same `(hash, column)` shape the real key store sees. @@ -11714,7 +11756,6 @@ mod tests { batch_max_bytes: 1024 * 1024, queue_max_age: Duration::ZERO, batch_max_wait: Duration::from_millis(1), - export_concurrency_per_shard: 1, export_max_attempts: 1, clickhouse_export_timeout: Duration::from_secs(5), clickhouse_breaker: ClickHouseBreakerConfig::default(), @@ -16190,11 +16231,34 @@ mod tests { .await; let body = Bytes::from(test_log_request("refused").encode_to_vec()); - let (response, ..) = + let (response, _items, _org_id, metered_bytes) = handle_signal_inner(&state, &test_headers(raw_key), body, Signal::Logs) .await .expect("a collector 4xx is relayed, not raised"); assert_eq!(response.status(), StatusCode::BAD_REQUEST); + assert_eq!(metered_bytes, 0, "a refused batch is not billed"); + assert!(recorded_key_uses(&store).await.is_empty()); + + // Over gRPC the refusal is answered OK, and is not key use either. + let resolved = state + .resolver + .resolve_ingest_key(raw_key) + .await + .expect("resolution should not fail") + .expect("the test key resolves"); + let payload = test_log_request("refused over grpc"); + let decoded_bytes = payload.encoded_len(); + accept_grpc_decoded( + &state, + Signal::Logs, + DecodedPayload::Logs(payload), + &resolved, + KeyAdmission::Ingest, + decoded_bytes, + &ClaimOutcome::default(), + ) + .await + .expect("a collector refusal over gRPC is answered OK"); assert!(recorded_key_uses(&store).await.is_empty()); drop(std::fs::remove_dir_all(&queue_dir)); diff --git a/apps/ingest/src/task_protection.rs b/apps/ingest/src/task_protection.rs index e5e347d530..28932bfc89 100644 --- a/apps/ingest/src/task_protection.rs +++ b/apps/ingest/src/task_protection.rs @@ -1,11 +1,12 @@ //! ECS task scale-in protection driven by WAL backlog. //! -//! The WAL lives on Fargate ephemeral storage, which is destroyed with the -//! task — so a scale-in that lands on a task holding unexported frames loses -//! them. Backlog is largest exactly when the autoscaler wants to scale in -//! (after a burst or a warehouse outage), so this loop marks the task -//! protected via the ECS agent's task-protection endpoint while the primary -//! lanes hold meaningful backlog, and releases it once they drain. +//! The WAL lives on the EC2 host's instance store (`/mnt/wal`), which is wiped +//! when the instance stops or is replaced, so a scale-in that lands on a task +//! holding unexported frames puts them at risk. Backlog is largest exactly +//! when the autoscaler wants to scale in (after a burst or a warehouse +//! outage), so this loop marks the task protected via the ECS agent's +//! task-protection endpoint while the primary lanes hold meaningful backlog, +//! and releases it once they drain. //! //! The endpoint is signed by the agent itself; the task role only needs //! `ecs:UpdateTaskProtection` (granted in `apps/ingest/alchemy.run.ts`). diff --git a/apps/ingest/src/telemetry.rs b/apps/ingest/src/telemetry.rs index af2fe4b0a5..70eda0d3e2 100644 --- a/apps/ingest/src/telemetry.rs +++ b/apps/ingest/src/telemetry.rs @@ -58,6 +58,9 @@ const WAL_MAGIC: &[u8; 4] = b"MTW1"; const WAL_V1_HEADER_LEN: usize = 20; const WAL_V2_HEADER_LEN: usize = 22; const WAL_V3_HEADER_LEN: usize = 23; +/// Most a frame read reserves before any payload byte arrives. The declared +/// length is unverified until the checksum, so a corrupt one must not size it. +const WAL_PAYLOAD_PREALLOC_MAX: usize = 1024 * 1024; #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)] pub enum ExportDestination { @@ -628,7 +631,6 @@ pub struct TinybirdConfig { pub batch_max_rows: usize, pub batch_max_bytes: usize, pub batch_max_wait: Duration, - pub export_concurrency_per_shard: usize, pub export_max_attempts: u32, /// Per-attempt timeout for a direct ClickHouse export POST. Kept short and /// separate from `forward_timeout` so a hanging BYO-ClickHouse target fails @@ -698,8 +700,9 @@ impl TinybirdConfig { if self.org_queue_max_bytes == 0 { return Err("INGEST_ORG_QUEUE_MAX_BYTES must be greater than 0".to_owned()); } - if self.export_concurrency_per_shard == 0 { - return Err("INGEST_TINYBIRD_CONCURRENCY_PER_SHARD must be greater than 0".to_owned()); + // `mpsc::channel(0)` panics, so a zero capacity is refused at boot instead. + if self.queue_channel_capacity == 0 { + return Err("INGEST_QUEUE_CHANNEL_CAPACITY must be greater than 0".to_owned()); } if self.export_max_attempts == 0 { return Err("INGEST_EXPORT_MAX_ATTEMPTS must be greater than 0".to_owned()); @@ -1166,23 +1169,6 @@ impl TelemetryPipeline { /// the session-replay ingest path, whose rows are built directly by the /// gateway (not derived from OTLP). Routes by org so a session's metadata /// and chunk-index rows land on the same shard in order. - pub async fn accept_rows( - &self, - org_id: &str, - datasource: String, - rows: Vec>, - signal: TelemetrySignal, - ) -> Result { - self.accept_rows_to( - org_id, - datasource, - rows, - signal, - ExportDestination::Tinybird, - ) - .await - } - #[hotpath::measure] pub async fn accept_rows_to( &self, @@ -1456,9 +1442,9 @@ struct WalLane { /// Bytes held by every segment on disk, the exported prefix included. Drives /// the lane cap and the `wal_shard_bytes` gauge. live_bytes: AtomicU64, - /// Cumulative bytes appended and marked exported since this lane was opened. - /// Their difference is the backlog, and it stays exact across rotation and - /// deletion — which nothing derived from file sizes does. + /// Cumulative bytes handed to export and marked exported since this lane + /// was opened. Their difference is the backlog, and it stays exact across + /// rotation and deletion, which nothing derived from file sizes does. committed_bytes: AtomicU64, exported_bytes: AtomicU64, append: Mutex, @@ -1492,13 +1478,16 @@ impl Drop for WaitingCaller { } /// `deliver`, run only while the returned guard is alive. Skipped, the inner -/// delivery is dropped, which releases its org reservation. -fn while_caller_waits(deliver: Delivery) -> (WaitingCaller, Delivery) { +/// delivery is dropped, which releases its org reservation, and the frame's +/// bytes leave `lane`'s backlog: nothing in this process will export them. +fn while_caller_waits(deliver: Delivery, lane: Arc) -> (WaitingCaller, Delivery) { let waiting = WaitingCaller(Arc::new(AtomicBool::new(true))); let still_waiting = Arc::clone(&waiting.0); let guarded: Delivery = Box::new(move |segment, start, end| { if still_waiting.load(Ordering::Acquire) { deliver(segment, start, end); + } else { + lane.forget_committed(end.saturating_sub(start)); } }); (waiting, guarded) @@ -2041,6 +2030,12 @@ impl WalLane { .load(Ordering::Relaxed) .saturating_sub(self.exported_bytes.load(Ordering::Relaxed)) } + + /// Take a synced frame back out of the backlog. Its sync already counted + /// it, but it never reaches export, so it would never be marked exported. + fn forget_committed(&self, bytes: u64) { + self.committed_bytes.fetch_sub(bytes, Ordering::Relaxed); + } } /// What WAL orphan recovery did with one segment's source object. @@ -2573,7 +2568,7 @@ impl ShardedWal { let lane_ref = self.lane(lane)?; // A caller dropped mid-append (a request timeout) has already answered // with a retryable error, so its frame must not reach export as well. - let (waiting, deliver) = while_caller_waits(deliver); + let (waiting, deliver) = while_caller_waits(deliver, Arc::clone(&lane_ref)); let appended = tokio::task::spawn_blocking(move || { lane_ref.append_blocking(&encoded, enforce_cap, deliver) }) @@ -3022,9 +3017,17 @@ fn read_wal_frame(file: &mut impl Read, start: u64) -> Result Delivery { let ran = Arc::clone(&ran); @@ -8023,13 +8113,13 @@ mod tests { }) }; - let (gone, deliver) = while_caller_waits(counting()); + let (gone, deliver) = while_caller_waits(counting(), Arc::clone(&lane)); drop(gone); lane.append_blocking(&encoded_test_frame(64), true, deliver) .expect("the frame is still made durable"); assert_eq!(ran.load(Ordering::SeqCst), 0, "but never reaches export"); - let (waiting, deliver) = while_caller_waits(counting()); + let (waiting, deliver) = while_caller_waits(counting(), Arc::clone(&lane)); lane.append_blocking(&encoded_test_frame(64), true, deliver) .expect("append"); drop(waiting); @@ -8041,6 +8131,35 @@ mod tests { drop(std::fs::remove_dir_all(queue_dir)); } + #[test] + fn a_frame_whose_caller_gave_up_is_not_backlog() { + let (lane, _probe, queue_dir) = slow_sync_lane("wal-caller-gone-backlog", Duration::ZERO); + let lane = Arc::new(lane); + let nothing = || -> Delivery { Box::new(|_, _, _| {}) }; + + let (gone, deliver) = while_caller_waits(nothing(), Arc::clone(&lane)); + drop(gone); + lane.append_blocking(&encoded_test_frame(64), true, deliver) + .expect("append"); + assert_eq!( + lane.backlog_bytes(), + 0, + "nothing will export a skipped frame, so it must not hold the drain open" + ); + + let delivered = encoded_test_frame(64); + let (waiting, deliver) = while_caller_waits(nothing(), Arc::clone(&lane)); + lane.append_blocking(&delivered, true, deliver) + .expect("append"); + drop(waiting); + assert_eq!( + lane.backlog_bytes(), + delivered.len() as u64, + "a delivered frame is backlog until it exports" + ); + drop(std::fs::remove_dir_all(queue_dir)); + } + #[test] fn a_failed_sync_refuses_its_batch_and_the_lane_recovers() { let (lane, probe, queue_dir) = slow_sync_lane("wal-group-commit-fail", Duration::ZERO); diff --git a/apps/ingest/src/wal_store.rs b/apps/ingest/src/wal_store.rs index f379737caf..e17de5f9f0 100644 --- a/apps/ingest/src/wal_store.rs +++ b/apps/ingest/src/wal_store.rs @@ -1,9 +1,10 @@ //! The WAL's durability tier: sealed segments shipped to S3, and the claiming //! protocol that lets a booting task recover a dead one's. //! -//! The local WAL lives on Fargate ephemeral storage, which is destroyed with the -//! task. A drain on SIGTERM covers a graceful stop; this covers everything else -//! — a crash, a spot reclaim, an AZ event — by keeping every sealed, unexported +//! The local WAL lives on the EC2 host's instance store (`/mnt/wal`), which +//! survives a reboot but is wiped when the instance stops or is replaced. A +//! drain on SIGTERM covers a graceful stop; this covers everything else (a +//! crash, a spot reclaim, an AZ event) by keeping every sealed, unexported //! segment in an object store too. //! //! Cost is small because segments are transient: a segment is deleted from the diff --git a/packages/otel-collector-maple-exporter/README.md b/packages/otel-collector-maple-exporter/README.md index 6fb44f118e..e327a28fee 100644 --- a/packages/otel-collector-maple-exporter/README.md +++ b/packages/otel-collector-maple-exporter/README.md @@ -26,9 +26,9 @@ exporters: database: default username: maple password: ${env:MAPLE_CLICKHOUSE_PASSWORD} - org_id: org_3AuiNCIuD1XCbbzcjkzE3s5HoQj # static fallback; per-record - # `maple_org_id` resource attr - # wins over this + org_id: org_3AuiNCIuD1XCbbzcjkzE3s5HoQj # stamped on every record; a + # resource attribute overrides it only when + # `org_id_from_resource_attribute` is set timeout: 30s retry_on_failure: enabled: true diff --git a/packages/otel-collector-maple-exporter/exporter_traces.go b/packages/otel-collector-maple-exporter/exporter_traces.go index 538c4bf2a5..f89a7c7405 100644 --- a/packages/otel-collector-maple-exporter/exporter_traces.go +++ b/packages/otel-collector-maple-exporter/exporter_traces.go @@ -4,7 +4,6 @@ import ( "context" "go.opentelemetry.io/collector/component" - "go.opentelemetry.io/collector/pdata/pcommon" "go.opentelemetry.io/collector/pdata/ptrace" "go.uber.org/zap" @@ -181,6 +180,3 @@ func encodeTraceRow( "LinksAttributes": linksAttributes, } } - -// silence unused import warning during incremental build -var _ pcommon.Map diff --git a/packages/otel-collector-maple-exporter/internal/encoding.go b/packages/otel-collector-maple-exporter/internal/encoding.go index 9c9ac6f877..cfa7dde4a0 100644 --- a/packages/otel-collector-maple-exporter/internal/encoding.go +++ b/packages/otel-collector-maple-exporter/internal/encoding.go @@ -3,7 +3,7 @@ package internal import ( "encoding/json" "fmt" - "strconv" + "math" "time" "go.opentelemetry.io/collector/pdata/pcommon" @@ -72,7 +72,9 @@ func ServiceName(resource pcommon.Map) string { // the layout — but ClickHouse parses the no-zone form and stores at the // configured precision, so we omit the zone to match its native rendering. func FormatTimestampNano(unixNano uint64) string { - if unixNano == 0 { + // DateTime64(9) is an Int64 of nanoseconds: a value past MaxInt64 would wrap + // negative below and fail the whole insert, so it maps to the epoch like 0. + if unixNano == 0 || unixNano > math.MaxInt64 { return "1970-01-01 00:00:00.000000000" } t := time.Unix(0, int64(unixNano)).UTC() @@ -82,7 +84,7 @@ func FormatTimestampNano(unixNano uint64) string { // FormatDateTime formats a UnixNano integer as the second-precision DateTime // form used by Maple's `Timestamp DateTime` and `Hour DateTime` columns. func FormatDateTime(unixNano uint64) string { - if unixNano == 0 { + if unixNano == 0 || unixNano > math.MaxInt64 { return "1970-01-01 00:00:00" } return time.Unix(0, int64(unixNano)).UTC().Format("2006-01-02 15:04:05") @@ -123,8 +125,8 @@ func StatusCodeString(code int32) string { } // SeverityNumberToText returns the canonical OTel severity text for a -// numeric severity (1-24). Falls back to the numeric form for unknown -// values. +// numeric severity (1-24). Unknown values are the empty string, as in the +// gateway's and Maple Local's encoders, so all three write the same rows. func SeverityNumberToText(n int32) string { switch { case n >= 1 && n <= 4: @@ -140,7 +142,7 @@ func SeverityNumberToText(n int32) string { case n >= 21 && n <= 24: return "FATAL" default: - return strconv.Itoa(int(n)) + return "" } } diff --git a/packages/otel-collector-maple-exporter/internal/encoding_test.go b/packages/otel-collector-maple-exporter/internal/encoding_test.go index af0468f1e0..7e1a353cd6 100644 --- a/packages/otel-collector-maple-exporter/internal/encoding_test.go +++ b/packages/otel-collector-maple-exporter/internal/encoding_test.go @@ -1,6 +1,7 @@ package internal import ( + "math" "testing" "go.opentelemetry.io/collector/pdata/pcommon" @@ -16,6 +17,10 @@ func TestFormatTimestampNano(t *testing.T) { {"epoch_one_ns", 1, "1970-01-01 00:00:00.000000001"}, {"epoch_one_sec", 1_000_000_000, "1970-01-01 00:00:01.000000000"}, {"specific_moment", 1_704_067_200_123_456_789, "2024-01-01 00:00:00.123456789"}, + {"int64_max", math.MaxInt64, "2262-04-11 23:47:16.854775807"}, + // Past Int64 the value would wrap negative and fail the whole insert. + {"past_int64_max", math.MaxInt64 + 1, "1970-01-01 00:00:00.000000000"}, + {"uint64_max", math.MaxUint64, "1970-01-01 00:00:00.000000000"}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -39,6 +44,7 @@ func TestFormatDateTime(t *testing.T) { {"zero", 0, "1970-01-01 00:00:00"}, {"end_of_year", 1_704_067_199_999_999_999, "2023-12-31 23:59:59"}, {"start_of_year", 1_704_067_200_000_000_000, "2024-01-01 00:00:00"}, + {"past_int64_max", math.MaxInt64 + 1, "1970-01-01 00:00:00"}, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { @@ -108,8 +114,8 @@ func TestSeverityNumberToText(t *testing.T) { {20, "ERROR"}, {21, "FATAL"}, {24, "FATAL"}, - {0, "0"}, // out-of-band → numeric form - {25, "25"}, + {0, ""}, // out-of-band is empty, as in the gateway's encoder + {25, ""}, } for _, tc := range cases { got := SeverityNumberToText(tc.n)