diff --git a/src/config/types.zig b/src/config/types.zig index 1ebb70c..1761876 100644 --- a/src/config/types.zig +++ b/src/config/types.zig @@ -55,6 +55,30 @@ pub const S3DumpConfig = struct { max_sealed_bytes: usize = 32 << 20, max_attempts: usize = 1, targets: []const S3TargetConfig = &.{}, + + /// Batch/backlog caps that must be nonzero no matter how flush is driven — + /// a zero cap makes the handler treat every record as oversized and drop it. + /// Used by every distribution (the Lambda flush is event-driven, so it only + /// needs this, not the flush-interval check below). + pub fn validateBatching(self: S3DumpConfig) !void { + if (!self.enabled) return; + if (self.max_batch_bytes == 0 or self.max_sealed_bytes == 0 or self.max_batch_records == 0) { + log.warn("s3_dump batch caps (max_batch_bytes/max_sealed_bytes/max_batch_records) must be > 0", .{}); + return error.InvalidS3DumpConfig; + } + } + + /// Full validation for distributions that run the wall-clock flush task + /// (the server distro): the batch caps plus a nonzero flush interval, since + /// a zero interval spins the flush loop tight. Lambda calls + /// `validateBatching` directly instead — `flush_interval_ms` is unused there. + pub fn validate(self: S3DumpConfig) !void { + try self.validateBatching(); + if (self.enabled and self.flush_interval_ms == 0) { + log.warn("s3_dump.flush_interval_ms must be > 0", .{}); + return error.InvalidS3DumpConfig; + } + } }; /// Main proxy configuration - loadable via zonfig @@ -106,21 +130,9 @@ pub const ProxyConfig = struct { /// com.usetero/s3-dump extension (Datadog logs → S3). Disabled by default. s3_dump: S3DumpConfig = .{}, - /// Post-load validation hook (called by zonfig). Rejects s3-dump knobs that - /// would silently break the runtime: a zero flush interval spins the flush - /// task in a tight loop, and a zero batch/backlog cap makes the handler drop - /// every record. Only checked when the extension is enabled. + /// Post-load validation hook (called by zonfig). pub fn validate(self: *ProxyConfig) !void { - const s = self.s3_dump; - if (!s.enabled) return; - if (s.flush_interval_ms == 0) { - log.warn("s3_dump.flush_interval_ms must be > 0", .{}); - return error.InvalidS3DumpConfig; - } - if (s.max_batch_bytes == 0 or s.max_sealed_bytes == 0 or s.max_batch_records == 0) { - log.warn("s3_dump batch caps (max_batch_bytes/max_sealed_bytes/max_batch_records) must be > 0", .{}); - return error.InvalidS3DumpConfig; - } + try self.s3_dump.validate(); } }; @@ -141,3 +153,15 @@ test "ProxyConfig.validate rejects zero s3_dump knobs when enabled" { var ok: ProxyConfig = .{ .s3_dump = .{ .enabled = true } }; try ok.validate(); } + +test "S3DumpConfig.validateBatching (Lambda path) ignores flush_interval_ms" { + // Lambda flush is event-driven, so a zero interval is fine there... + const zero_interval: S3DumpConfig = .{ .enabled = true, .flush_interval_ms = 0 }; + try zero_interval.validateBatching(); + // ...but the full validate (server distro) still rejects it. + try std.testing.expectError(error.InvalidS3DumpConfig, zero_interval.validate()); + + // Zero batch caps are rejected on both paths. + const zero_cap: S3DumpConfig = .{ .enabled = true, .max_batch_bytes = 0 }; + try std.testing.expectError(error.InvalidS3DumpConfig, zero_cap.validateBatching()); +} diff --git a/src/lambda_main.zig b/src/lambda_main.zig index 4f3e82f..0f29dda 100644 --- a/src/lambda_main.zig +++ b/src/lambda_main.zig @@ -19,6 +19,8 @@ const build_options = @import("build_options"); const edge = @import("root.zig"); const runtime_metrics_mod = @import("runtime/runtime_metrics.zig"); const app = @import("runtime/app.zig"); +const ext_rt = @import("runtime/extensions.zig"); +const config_types = @import("config/types.zig"); const policy = edge.policy; const zonfig = edge.zonfig; @@ -91,6 +93,11 @@ pub const LambdaConfig = struct { api_key: ?[]const u8 = null, } = .{}, + /// com.usetero/s3-dump extension (Datadog logs → S3). Disabled by default. + /// In Lambda the flush is event-driven (per-invoke + shutdown), so + /// `flush_interval_ms` is unused here — see the event loop in `main`. + s3_dump: config_types.S3DumpConfig = .{}, + /// zonfig validation hook, run after env/JSON overrides. Clamps env knobs /// into ranges the runtime requires so a bad override degrades instead of /// aborting (or silently wedging) the extension at startup. @@ -118,6 +125,10 @@ pub const LambdaConfig = struct { std.log.warn("TERO_THREAD_POOL_COUNT=0 starts no handler threads; using httpz default", .{}); self.thread_pool_count = null; }; + + // Lambda flush is event-driven (per-invoke + shutdown), so + // `flush_interval_ms` is unused here — validate only the batch caps. + try self.s3_dump.validateBatching(); } }; @@ -227,6 +238,18 @@ pub fn main(init: std.process.Init) !void { var registry = policy.Registry.init(allocator, bus); defer registry.deinit(); + // s3-dump extension: wired before policies compile so extension bindings + // resolve at snapshot build. Inert unless enabled with targets. + var exts = ext_rt.Extensions.init(allocator, .{ .log = ext_rt.datadogLogEncode }); + defer exts.deinit(); + const s3_dump_active = config.s3_dump.enabled and config.s3_dump.targets.len > 0; + var extension_sink: ?policy.ExtensionSink = null; + if (s3_dump_active) { + try ext_rt.configure(&exts, allocator, io, config.s3_dump, init.environ_map); + exts.register(®istry); + extension_sink = exts.sink(); + } + var runtime_metrics = try RuntimeMetrics.init(allocator, io, .lambda); defer runtime_metrics.deinit(); runtime_metrics.setBuildInfo(build_options.version, build_options.commit); @@ -305,6 +328,7 @@ pub fn main(init: std.process.Init) !void { .upstream_url = config.upstream_url, .logs_url = config.logs_url, .metrics_url = config.metrics_url, + .extension_sink = extension_sink, }); defer engine.destroy(); @@ -342,14 +366,22 @@ pub fn main(init: std.process.Init) !void { bus.debug(LambdaInvokeReceived{ .request_id = invoke_event.request_id, }); - // Proxy server handles requests in background - // Nothing to do here - just continue polling + // Lambda freezes the environment between invocations, so the + // server distro's wall-clock flush timer is unreliable here. + // Flush at each invoke boundary instead: receiving this event + // means the PRIOR invocation's function has completed, so its + // telemetry is fully batched. The final invocation's tail is + // drained by the SHUTDOWN force-flush below. + if (s3_dump_active) runtime_metrics.recordS3DumpFlush(exts.flush(io, .{})); }, .shutdown => |shutdown_event| { bus.info(LambdaExtensionShutdown{ .reason = @tagName(shutdown_event.reason), }); + // Final drain within Lambda's shutdown deadline (io still live). + if (s3_dump_active) runtime_metrics.recordS3DumpFlush(exts.flush(io, .{ .force = true })); + // Stop proxy server gracefully shutdown_requested.store(true, .release); engine.requestShutdown(); diff --git a/src/runtime/extensions.zig b/src/runtime/extensions.zig index 569996e..c2d47a2 100644 --- a/src/runtime/extensions.zig +++ b/src/runtime/extensions.zig @@ -43,9 +43,33 @@ pub fn datadogLogEncode(record: *const anyopaque, writer: *std.Io.Writer) anyerr /// Read S3 credentials from the environment. Returns null when either half is /// absent — the extension then counts-and-drops deliveries for its targets /// (fail-open), so a missing credential never stalls telemetry. +/// +/// The `TERO_S3_*` pair is tried first, then the standard `AWS_*` pair — each +/// resolved as a UNIT (both halves from the same set), never mixed. A partial +/// `TERO_S3_*` (id without secret) means the override is intentionally absent, +/// so we fall through to `AWS_*` rather than pairing keys across credential +/// sets (which would sign every request with an invalid keypair). +/// +/// This matters in Lambda: `AWS_ACCESS_KEY_ID`/`AWS_SECRET_ACCESS_KEY` are +/// *reserved* — the runtime overwrites them with the execution role's TEMPORARY +/// creds (which z3 can't sign, since it emits no `x-amz-security-token`). So to +/// use static IAM-user keys there, set them under the non-reserved `TERO_S3_*` +/// names, which the runtime never touches. Outside Lambda the `AWS_*` fallback +/// works as usual. pub fn credentialsFromEnv(env: *const std.process.Environ.Map) ?ext.S3Dump.Credentials { - const access = env.get("AWS_ACCESS_KEY_ID") orelse return null; - const secret = env.get("AWS_SECRET_ACCESS_KEY") orelse return null; + return pairFromEnv(env, "TERO_S3_ACCESS_KEY_ID", "TERO_S3_SECRET_ACCESS_KEY") orelse + pairFromEnv(env, "AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"); +} + +/// Resolve a same-set access-key/secret pair, or null if either half is +/// missing or empty (so the caller can cleanly fall back to the next set). +fn pairFromEnv( + env: *const std.process.Environ.Map, + access_name: []const u8, + secret_name: []const u8, +) ?ext.S3Dump.Credentials { + const access = env.get(access_name) orelse return null; + const secret = env.get(secret_name) orelse return null; if (access.len == 0 or secret.len == 0) return null; return .{ .access_key_id = access, .secret_access_key = secret }; } @@ -107,6 +131,30 @@ pub fn flushLoop( } } +test "credentialsFromEnv prefers TERO_S3_* over the reserved AWS_* names" { + var env = std.process.Environ.Map.init(std.testing.allocator); + defer env.deinit(); + + // Neither set → null (fail-open: deliveries drop, nothing stalls). + try std.testing.expect(credentialsFromEnv(&env) == null); + + // Only AWS_* (in Lambda these are the reserved role creds) → fallback. + try env.put("AWS_ACCESS_KEY_ID", "AKIA-fallback"); + try env.put("AWS_SECRET_ACCESS_KEY", "fallback-secret"); + try std.testing.expectEqualStrings("AKIA-fallback", credentialsFromEnv(&env).?.access_key_id); + + // A PARTIAL TERO_S3_* (id only) must NOT pair with AWS_*'s secret — that + // would sign with a mixed keypair. Fall back to the whole AWS_* pair. + try env.put("TERO_S3_ACCESS_KEY_ID", "AKIA-user"); + try std.testing.expectEqualStrings("AKIA-fallback", credentialsFromEnv(&env).?.access_key_id); + + // Complete TERO_S3_* pair → wins (the Lambda static-key path). + try env.put("TERO_S3_SECRET_ACCESS_KEY", "user-secret"); + const creds = credentialsFromEnv(&env).?; + try std.testing.expectEqualStrings("AKIA-user", creds.access_key_id); + try std.testing.expectEqualStrings("user-secret", creds.secret_access_key); +} + test "datadogLogEncode renders the delivered log context as JSON" { const allocator = std.testing.allocator; var ddlog = try datadog_log.DatadogLog.parseRaw(allocator, "{\"status\":\"info\",\"message\":\"hi\"}");