Something went wrong. Try again.
jetstream v2 in zig stream.waow.tech
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687//! OpenTelemetry bootstrap matching the pinned Jetstream V2 runtime contract.
const std = @import("std");const otel_api = @import("otel-api");const otel_sdk = @import("otel-sdk");const otel_exporters = @import("otel-exporters");
const Io = std.Io;
pub const InitConfig = struct { io: Io = std.Options.debug_io, environment: []const []const u8, service_name: []const u8, service_version: []const u8, shutdown_timeout_ms: u64 = 30_000,};
const SamplerKind = enum { always_on, always_off, trace_id_ratio, parent_always_on, parent_always_off, parent_trace_id_ratio,};
const Parsed = struct { allocator: std.mem.Allocator, endpoint: ?[]u8 = null, append_signal_path: bool = true, compression: otel_exporters.otlp.CompressionMethod = .none, timeout_ms: u64 = 10_000, export_timeout_ms: u64 = 30_000, ca_file: ?[]u8 = null, client_cert_file: ?[]u8 = null, client_key_file: ?[]u8 = null, headers: []std.http.Header = &.{}, batch: otel_sdk.trace.BatchSpanProcessor.PipelineStep.ContextType = .{}, sampler_kind: SamplerKind = .parent_always_on, sampler_ratio: f64 = 1.0,
fn init(allocator: std.mem.Allocator, entries: []const []const u8) !Parsed { var parsed: Parsed = .{ .allocator = allocator }; errdefer parsed.deinit();
const generic_raw = envValue(entries, "OTEL_EXPORTER_OTLP_ENDPOINT"); const traces_raw = envValue(entries, "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"); const configured = (if (generic_raw) |value| value.len > 0 else false) or (if (traces_raw) |value| value.len > 0 else false); if (!configured) return parsed;
const traces_endpoint = envTrimmed(entries, "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"); const generic_endpoint = envTrimmed(entries, "OTEL_EXPORTER_OTLP_ENDPOINT"); var endpoint: []const u8 = "https://localhost:4318"; if (generic_endpoint) |candidate| { if (validUri(candidate)) endpoint = candidate; } if (traces_endpoint) |candidate| { if (validUri(candidate)) { endpoint = candidate; parsed.append_signal_path = false; } } parsed.endpoint = try allocator.dupe(u8, endpoint); try applyInsecureOverride(allocator, &parsed.endpoint.?, insecureOverride(entries));
if (signalValue(entries, "OTEL_EXPORTER_OTLP_COMPRESSION", "OTEL_EXPORTER_OTLP_TRACES_COMPRESSION")) |compression| { // Go's OTLP HTTP exporter selects gzip only for the exact value // "gzip"; every other non-empty value means no compression. parsed.compression = if (std.mem.eql(u8, compression, "gzip")) .gzip else .none; } parsed.timeout_ms = parsePositiveU64Or( signalValue(entries, "OTEL_EXPORTER_OTLP_TIMEOUT", "OTEL_EXPORTER_OTLP_TRACES_TIMEOUT"), 10_000, );
const raw_headers = signalValue(entries, "OTEL_EXPORTER_OTLP_HEADERS", "OTEL_EXPORTER_OTLP_TRACES_HEADERS"); parsed.headers = try parseHeaders(allocator, raw_headers orelse "");
parsed.batch = .{ .export_interval_ms = parsePositiveU32Or(envTrimmed(entries, "OTEL_BSP_SCHEDULE_DELAY"), 5_000), .max_queue_size = parsePositiveUsizeOr(envTrimmed(entries, "OTEL_BSP_MAX_QUEUE_SIZE"), 2_048), .max_export_batch_size = parsePositiveUsizeOr(envTrimmed(entries, "OTEL_BSP_MAX_EXPORT_BATCH_SIZE"), 512), }; parsed.export_timeout_ms = parsePositiveU64Or(envTrimmed(entries, "OTEL_BSP_EXPORT_TIMEOUT"), 30_000);
if (signalValue(entries, "OTEL_EXPORTER_OTLP_CERTIFICATE", "OTEL_EXPORTER_OTLP_TRACES_CERTIFICATE")) |path| parsed.ca_file = try allocator.dupe(u8, path); const client_cert = signalValue(entries, "OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE", "OTEL_EXPORTER_OTLP_TRACES_CLIENT_CERTIFICATE"); const client_key = signalValue(entries, "OTEL_EXPORTER_OTLP_CLIENT_KEY", "OTEL_EXPORTER_OTLP_TRACES_CLIENT_KEY"); if (client_cert != null and client_key != null) { parsed.client_cert_file = try allocator.dupe(u8, client_cert.?); parsed.client_key_file = try allocator.dupe(u8, client_key.?); }
const sampler_name = envTrimmed(entries, "OTEL_TRACES_SAMPLER") orelse "parentbased_always_on"; const sampler_arg = envTrimmed(entries, "OTEL_TRACES_SAMPLER_ARG"); if (std.ascii.eqlIgnoreCase(sampler_name, "always_on")) { parsed.sampler_kind = .always_on; } else if (std.ascii.eqlIgnoreCase(sampler_name, "always_off")) { parsed.sampler_kind = .always_off; } else if (std.ascii.eqlIgnoreCase(sampler_name, "traceidratio")) { parsed.sampler_kind = .trace_id_ratio; parsed.sampler_ratio = parseRatioOrOne(sampler_arg); } else if (std.ascii.eqlIgnoreCase(sampler_name, "parentbased_always_on")) { parsed.sampler_kind = .parent_always_on; } else if (std.ascii.eqlIgnoreCase(sampler_name, "parentbased_always_off")) { parsed.sampler_kind = .parent_always_off; } else if (std.ascii.eqlIgnoreCase(sampler_name, "parentbased_traceidratio")) { parsed.sampler_kind = .parent_trace_id_ratio; parsed.sampler_ratio = parseRatioOrOne(sampler_arg); } else { // The Go SDK reports unsupported samplers and retains its default. parsed.sampler_kind = .parent_always_on; }
return parsed; }
fn exporterConfig(self: *const Parsed, io: Io) otel_exporters.otlp.OtlpExporterConfig { return .{ .endpoint = self.endpoint.?, .io = io, // Pinned upstream constructs otlptracehttp directly. Its protocol // environment variables therefore do not switch this exporter. .transport = .http_protobuf, .headers = self.headers, .compression = self.compression, .timeout_millis = self.timeout_ms, .export_timeout_millis = self.export_timeout_ms, .tls_config = if (self.ca_file != null or (self.client_cert_file != null and self.client_key_file != null)) .{ .ca_file = self.ca_file, .cert_file = self.client_cert_file, .key_file = self.client_key_file, } else null, .append_signal_path = self.append_signal_path, }; }
fn deinit(self: *Parsed) void { if (self.endpoint) |endpoint| self.allocator.free(endpoint); if (self.ca_file) |path| self.allocator.free(path); if (self.client_cert_file) |path| self.allocator.free(path); if (self.client_key_file) |path| self.allocator.free(path); for (self.headers) |header| { self.allocator.free(header.name); self.allocator.free(header.value); } if (self.headers.len > 0) self.allocator.free(self.headers); self.* = .{ .allocator = self.allocator }; }};
pub const Runtime = struct { allocator: std.mem.Allocator, provider: ?*otel_sdk.trace.TracerProvider = null, parsed: ?Parsed = null, ratio_sampler: ?*otel_sdk.trace.samplers.TraceIdRatioBasedSampler = null, parent_sampler: ?*otel_sdk.trace.samplers.ParentBasedSampler = null, shutdown_called: bool = false, shutdown_timeout_ms: u64 = 30_000,
pub fn init(allocator: std.mem.Allocator, config: InitConfig) !Runtime { var parsed = try Parsed.init(allocator, config.environment); try otel_api.setGlobalTextMapPropagator(otel_api.context.createW3cBaggagePropagator()); errdefer otel_api.setGlobalTextMapPropagator(null) catch {};
if (parsed.endpoint == null) { // The registry's built-in provider is a real no-op tracer provider. try otel_api.provider_registry.setGlobalTracerProvider(null); parsed.deinit(); return .{ .allocator = allocator, .shutdown_timeout_ms = config.shutdown_timeout_ms }; }
var runtime: Runtime = .{ .allocator = allocator, .parsed = parsed, .shutdown_timeout_ms = config.shutdown_timeout_ms, }; errdefer runtime.deinit();
const sampler = try runtime.makeSampler(parsed.sampler_kind, parsed.sampler_ratio); var detected = try otel_sdk.resource.detectResource(allocator); defer detected.deinitOwned(allocator); const service_attributes = [_]otel_api.AttributeKeyValue{ .{ .key = "service.name", .value = .{ .string = config.service_name } }, .{ .key = "service.version", .value = .{ .string = config.service_version } }, }; var service = try otel_sdk.resource.Resource.initOwned(allocator, .{ .attributes = &service_attributes }); defer service.deinitOwned(allocator); const resource = try otel_sdk.resource.Resource.initOwnedMerge(allocator, detected, service);
const provider = allocator.create(otel_sdk.trace.TracerProvider) catch |err| { resource.deinitOwned(allocator); return err; }; errdefer allocator.destroy(provider); provider.* = otel_sdk.trace.TracerProvider.init( allocator, resource, otel_sdk.trace.createDefaultIdGenerator(), sampler, ); errdefer provider.deinit();
var builder = provider.pipelineBuilder(); builder = builder.with( otel_sdk.trace.BatchSpanProcessor.PipelineStep.init(parsed.batch) .flowTo(otel_exporters.otlp.OtlpTraceExporter.PipelineStep.init(parsed.exporterConfig(config.io))), ); try builder.done(); try otel_api.provider_registry.setGlobalTracerProvider(provider.tracerProvider()); runtime.provider = provider; return runtime; }
pub fn enabled(self: *const Runtime) bool { return self.provider != null; }
pub fn shutdown(self: *Runtime, timeout_ms: ?u64) otel_api.common.ProcessResult { if (self.shutdown_called) return .success; self.shutdown_called = true; const result = if (self.provider) |provider| provider.shutdown(timeout_ms) else otel_api.common.ProcessResult.success; otel_api.provider_registry.setGlobalTracerProvider(null) catch {}; otel_api.setGlobalTextMapPropagator(null) catch {}; return result; }
pub fn deinit(self: *Runtime) void { _ = self.shutdown(self.shutdown_timeout_ms); if (self.provider) |provider| { provider.deinit(); provider.destroy(); } if (self.parent_sampler) |sampler| self.allocator.destroy(sampler); if (self.ratio_sampler) |sampler| self.allocator.destroy(sampler); if (self.parsed) |*parsed| parsed.deinit(); self.provider = null; self.parent_sampler = null; self.ratio_sampler = null; self.parsed = null; }
fn makeSampler(self: *Runtime, kind: SamplerKind, ratio: f64) !otel_api.trace.Sampler { const samplers = otel_sdk.trace.samplers; const root: otel_api.trace.Sampler = switch (kind) { .always_on, .parent_always_on => samplers.always_on, .always_off, .parent_always_off => samplers.always_off, .trace_id_ratio, .parent_trace_id_ratio => root: { const concrete = try self.allocator.create(samplers.TraceIdRatioBasedSampler); concrete.* = samplers.TraceIdRatioBasedSampler.init(ratio, 14); self.ratio_sampler = concrete; break :root .{ .bridge = otel_api.trace.Sampler.Bridge.init(concrete) }; }, }; return switch (kind) { .always_on, .always_off, .trace_id_ratio => root, .parent_always_on, .parent_always_off, .parent_trace_id_ratio => parent: { const concrete = try self.allocator.create(samplers.ParentBasedSampler); concrete.* = samplers.ParentBasedSampler.init(root); self.parent_sampler = concrete; break :parent .{ .bridge = otel_api.trace.Sampler.Bridge.init(concrete) }; }, }; }};
pub fn tracer(scope: []const u8) !otel_api.trace.Tracer { var name_buffer: [256]u8 = undefined; const name = try std.fmt.bufPrint(&name_buffer, "jetstream/{s}", .{scope}); return otel_api.getGlobalTracerProvider().getTracerWithScope(.{ .name = name });}
pub const Context = []const otel_api.ContextKeyValue;pub const root_context: Context = &.{};pub const Attribute = otel_api.AttributeKeyValue;
/// One explicitly propagated operation span. Stream passes `context()` into/// child operations rather than relying on thread-local state: bootstrap and/// steady work cross worker threads, and pinned upstream propagates the Go/// context across those same boundaries.pub const Span = struct { allocator: std.mem.Allocator, value: otel_api.trace.Span = .{ .noop = .invalid }, parent: Context = root_context, owned_context: ?[]otel_api.ContextKeyValue = null, ended: bool = false,
pub fn start( allocator: std.mem.Allocator, scope: []const u8, name: []const u8, parent: Context, attributes: []const otel_api.AttributeKeyValue, ) Span { var result: Span = .{ .allocator = allocator, .parent = parent }; var span_tracer = tracer(scope) catch return result; result.value = span_tracer.startSpan(name, .{ .attributes = attributes }, parent) catch return result; const span_context = result.value.getSpanContext(); if (span_context.isValid()) { result.owned_context = otel_api.trace.trace_context.withActiveSpanContext( allocator, parent, span_context, ) catch null; } return result; }
pub fn context(self: *const Span) Context { return if (self.owned_context) |owned| owned else self.parent; }
pub fn setAttribute(self: *Span, attribute: Attribute) void { if (!self.ended) self.value.setAttribute(attribute); }
pub fn succeed(self: *Span) void { if (self.ended) return; self.value.setStatus(.{ .code = .ok }); self.value.end(null); self.ended = true; }
pub fn fail(self: *Span, err: anyerror) void { if (self.ended) return; self.value.recordException(err, null, null) catch {}; self.value.setStatus(.{ .code = .@"error", .description = @errorName(err) }); self.value.end(null); self.ended = true; }
/// Fails the span with `result`'s error, if it carries one, and returns /// `result` unchanged for the caller to `try` or `return`. pub fn recordError(self: *Span, result: anytype) @TypeOf(result) { if (result) |_| {} else |err| self.fail(err); return result; }
pub fn deinit(self: *Span) void { if (!self.ended) self.succeed(); if (self.owned_context) |owned| otel_api.ContextKeyValue.deinitOwnedSlice(self.allocator, owned); self.value.deinit(); self.owned_context = null; }};
fn envValue(entries: []const []const u8, key: []const u8) ?[]const u8 { for (entries) |entry| { const split = std.mem.indexOfScalar(u8, entry, '=') orelse entry.len; if (std.mem.eql(u8, entry[0..split], key)) return if (split < entry.len) entry[split + 1 ..] else ""; } return null;}
fn validUri(raw: []const u8) bool { _ = std.Uri.parse(raw) catch return false; return true;}
fn envTrimmed(entries: []const []const u8, key: []const u8) ?[]const u8 { const raw = envValue(entries, key) orelse return null; const trimmed = std.mem.trim(u8, raw, " \t\r\n"); return if (trimmed.len == 0) null else trimmed;}
fn signalValue(entries: []const []const u8, generic: []const u8, specific: []const u8) ?[]const u8 { return envTrimmed(entries, specific) orelse envTrimmed(entries, generic);}
fn insecureOverride(entries: []const []const u8) ?bool { const raw = signalValue(entries, "OTEL_EXPORTER_OTLP_INSECURE", "OTEL_EXPORTER_OTLP_TRACES_INSECURE") orelse return null; return std.ascii.eqlIgnoreCase(raw, "true");}
fn applyInsecureOverride(allocator: std.mem.Allocator, endpoint: *[]u8, insecure: ?bool) !void { const use_http = insecure orelse return; const separator = std.mem.indexOf(u8, endpoint.*, "://") orelse return; const replacement = if (use_http) "http" else "https"; const rewritten = try std.fmt.allocPrint(allocator, "{s}{s}", .{ replacement, endpoint.*[separator..] }); allocator.free(endpoint.*); endpoint.* = rewritten;}
fn parsePositiveU64Or(raw: ?[]const u8, default: u64) u64 { const value = std.fmt.parseInt(i64, raw orelse return default, 10) catch return default; return if (value > 0) @intCast(value) else default;}
fn parsePositiveU32Or(raw: ?[]const u8, default: u32) u32 { const value = std.fmt.parseInt(i64, raw orelse return default, 10) catch return default; return if (value > 0 and value <= std.math.maxInt(u32)) @intCast(value) else default;}
fn parsePositiveUsizeOr(raw: ?[]const u8, default: usize) usize { const value = std.fmt.parseInt(i128, raw orelse return default, 10) catch return default; return if (value > 0 and value <= std.math.maxInt(usize)) @intCast(value) else default;}
fn parseRatioOrOne(raw: ?[]const u8) f64 { const ratio = std.fmt.parseFloat(f64, raw orelse return 1.0) catch return 1.0; return if (std.math.isFinite(ratio) and ratio >= 0 and ratio <= 1) ratio else 1.0;}
fn validHeaderName(name: []const u8) bool { if (name.len == 0) return false; for (name) |c| if (!(std.ascii.isAlphanumeric(c) or switch (c) { '!', '#', '$', '%', '&', '\'', '*', '+', '-', '.', '^', '_', '`', '|', '~' => true, else => false, })) return false; return true;}
fn validPercentEncoding(value: []const u8) bool { var i: usize = 0; while (i < value.len) : (i += 1) { if (value[i] != '%') continue; if (i + 2 >= value.len or !std.ascii.isHex(value[i + 1]) or !std.ascii.isHex(value[i + 2])) return false; i += 2; } return true;}
fn parseHeaders(allocator: std.mem.Allocator, raw: []const u8) ![]std.http.Header { var headers: std.ArrayList(std.http.Header) = .empty; errdefer { for (headers.items) |header| { allocator.free(header.name); allocator.free(header.value); } headers.deinit(allocator); } var pairs = std.mem.splitScalar(u8, raw, ','); while (pairs.next()) |pair| { const equals = std.mem.indexOfScalar(u8, pair, '=') orelse continue; const name = std.mem.trim(u8, pair[0..equals], " \t\r\n"); const encoded = pair[equals + 1 ..]; if (!validHeaderName(name) or !validPercentEncoding(encoded)) continue; const owned_name = try allocator.dupe(u8, name); errdefer allocator.free(owned_name); const owned_encoded = try allocator.dupe(u8, encoded); const decoded = std.Uri.percentDecodeInPlace(owned_encoded); const trimmed = std.mem.trim(u8, decoded, " \t\r\n"); const owned_value = try allocator.dupe(u8, trimmed); allocator.free(owned_encoded); errdefer allocator.free(owned_value);
var replaced = false; for (headers.items) |*header| { if (!std.mem.eql(u8, header.name, name)) continue; allocator.free(header.name); allocator.free(header.value); header.* = .{ .name = owned_name, .value = owned_value }; replaced = true; break; } if (!replaced) try headers.append(allocator, .{ .name = owned_name, .value = owned_value }); } return headers.toOwnedSlice(allocator);}
test "OTLP activation and per-signal precedence match upstream" { const testing = std.testing; var disabled = try Parsed.init(testing.allocator, &.{"OTEL_EXPORTER_OTLP_ENDPOINT="}); defer disabled.deinit(); try testing.expect(disabled.endpoint == null);
var whitespace = try Parsed.init(testing.allocator, &.{"OTEL_EXPORTER_OTLP_ENDPOINT= "}); defer whitespace.deinit(); try testing.expectEqualStrings("https://localhost:4318", whitespace.endpoint.?);
var invalid_specific = try Parsed.init(testing.allocator, &.{ "OTEL_EXPORTER_OTLP_ENDPOINT=https://generic.example/base", "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=://invalid", }); defer invalid_specific.deinit(); try testing.expectEqualStrings("https://generic.example/base", invalid_specific.endpoint.?); try testing.expect(invalid_specific.append_signal_path);
var parsed = try Parsed.init(testing.allocator, &.{ "OTEL_EXPORTER_OTLP_ENDPOINT=https://generic.example/base", "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT=http://trace.example/custom", "OTEL_EXPORTER_OTLP_HEADERS=generic=one", "OTEL_EXPORTER_OTLP_TRACES_HEADERS=trace=two%20words", "OTEL_EXPORTER_OTLP_TIMEOUT=99", "OTEL_EXPORTER_OTLP_TRACES_TIMEOUT=42", "OTEL_EXPORTER_OTLP_COMPRESSION=gzip", // Pinned upstream's explicit HTTP/protobuf constructor ignores this. "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL=http/json", }); defer parsed.deinit(); try testing.expectEqualStrings("http://trace.example/custom", parsed.endpoint.?); try testing.expect(!parsed.append_signal_path); try testing.expectEqual(otel_exporters.otlp.CompressionMethod.gzip, parsed.compression); try testing.expectEqual(@as(u64, 42), parsed.timeout_ms); try testing.expectEqual(@as(usize, 1), parsed.headers.len); try testing.expectEqualStrings("trace", parsed.headers[0].name); try testing.expectEqualStrings("two words", parsed.headers[0].value);}
test "OTLP insecure and batch settings follow standard environment semantics" { const testing = std.testing; var parsed = try Parsed.init(testing.allocator, &.{ "OTEL_EXPORTER_OTLP_ENDPOINT=https://collector.example/base", "OTEL_EXPORTER_OTLP_INSECURE=true", "OTEL_BSP_SCHEDULE_DELAY=17", "OTEL_BSP_MAX_QUEUE_SIZE=31", "OTEL_BSP_MAX_EXPORT_BATCH_SIZE=7", "OTEL_BSP_EXPORT_TIMEOUT=23", "OTEL_EXPORTER_OTLP_CERTIFICATE=/generic/ca.pem", "OTEL_EXPORTER_OTLP_TRACES_CERTIFICATE=/trace/ca.pem", "OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE=/generic/client.pem", "OTEL_EXPORTER_OTLP_CLIENT_KEY=/generic/client.key", "OTEL_EXPORTER_OTLP_TRACES_CLIENT_CERTIFICATE=/trace/client.pem", "OTEL_EXPORTER_OTLP_TRACES_CLIENT_KEY=/trace/client.key", "OTEL_TRACES_SAMPLER=parentbased_traceidratio", "OTEL_TRACES_SAMPLER_ARG=0.25", }); defer parsed.deinit(); try testing.expectEqualStrings("http://collector.example/base", parsed.endpoint.?); try testing.expectEqual(@as(u32, 17), parsed.batch.export_interval_ms.?); try testing.expectEqual(@as(usize, 31), parsed.batch.max_queue_size.?); try testing.expectEqual(@as(usize, 7), parsed.batch.max_export_batch_size.?); try testing.expectEqual(@as(u64, 23), parsed.export_timeout_ms); try testing.expectEqualStrings("/trace/ca.pem", parsed.ca_file.?); try testing.expectEqualStrings("/trace/client.pem", parsed.client_cert_file.?); try testing.expectEqualStrings("/trace/client.key", parsed.client_key_file.?); try testing.expectEqual(SamplerKind.parent_trace_id_ratio, parsed.sampler_kind); try testing.expectEqual(@as(f64, 0.25), parsed.sampler_ratio);}
test "an incomplete OTLP client credential pair is ignored like upstream" { var parsed = try Parsed.init(std.testing.allocator, &.{ "OTEL_EXPORTER_OTLP_ENDPOINT=https://collector.example", "OTEL_EXPORTER_OTLP_CLIENT_CERTIFICATE=/client.pem", }); defer parsed.deinit(); try std.testing.expect(parsed.client_cert_file == null); try std.testing.expect(parsed.client_key_file == null);}
test "invalid OTLP headers and sampler values use upstream fallbacks" { const testing = std.testing; var parsed = try Parsed.init(testing.allocator, &.{ "OTEL_EXPORTER_OTLP_ENDPOINT=http://collector.example", "OTEL_EXPORTER_OTLP_HEADERS=bad header=x,ok=first,missing,escape=%zz,ok=last", "OTEL_TRACES_SAMPLER=traceidratio", "OTEL_TRACES_SAMPLER_ARG=2", "OTEL_BSP_MAX_QUEUE_SIZE=not-a-number", }); defer parsed.deinit(); try testing.expectEqual(@as(usize, 1), parsed.headers.len); try testing.expectEqualStrings("ok", parsed.headers[0].name); try testing.expectEqualStrings("last", parsed.headers[0].value); try testing.expectEqual(@as(f64, 1), parsed.sampler_ratio); try testing.expectEqual(@as(usize, 2_048), parsed.batch.max_queue_size.?);}
const CollectorCapture = struct { target: [256]u8 = undefined, target_len: usize = 0, body: [64 * 1024]u8 = undefined, body_len: usize = 0, content_type: [64]u8 = undefined, content_type_len: usize = 0, api_key: [64]u8 = undefined, api_key_len: usize = 0,
fn targetSlice(self: *const CollectorCapture) []const u8 { return self.target[0..self.target_len]; }
fn bodySlice(self: *const CollectorCapture) []const u8 { return self.body[0..self.body_len]; }};
const Collector = struct { fn copy(destination: []u8, length: *usize, value: []const u8) !void { if (value.len > destination.len) return error.CaptureOverflow; @memcpy(destination[0..value.len], value); length.* = value.len; }
fn receive(listener: *Io.net.Server, io: Io, capture: *CollectorCapture) !void { var stream = try listener.accept(io); defer stream.close(io); var read_buffer: [4096]u8 = undefined; var write_buffer: [4096]u8 = undefined; var reader = stream.reader(io, &read_buffer); var writer = stream.writer(io, &write_buffer); var server = std.http.Server.init(&reader.interface, &writer.interface); var request = try server.receiveHead();
try copy(&capture.target, &capture.target_len, request.head.target); var headers = request.iterateHeaders(); while (headers.next()) |header| { if (std.ascii.eqlIgnoreCase(header.name, "content-type")) { try copy(&capture.content_type, &capture.content_type_len, header.value); } else if (std.ascii.eqlIgnoreCase(header.name, "x-api-key")) { try copy(&capture.api_key, &capture.api_key_len, header.value); } } var body_buffer: [4096]u8 = undefined; const body_reader = request.readerExpectNone(&body_buffer); var body_writer = std.Io.Writer.fixed(&capture.body); _ = try body_reader.streamRemaining(&body_writer); capture.body_len = body_writer.buffered().len; try request.respond("", .{ .status = .ok, .keep_alive = false }); }};
test "runtime exports a real protobuf span and resource during bounded shutdown" { const testing = std.testing; var threaded: Io.Threaded = .init(testing.allocator, .{}); defer threaded.deinit(); const io = threaded.io(); var listener = (Io.net.IpAddress{ .ip4 = .loopback(0) }).listen(io, .{ .reuse_address = true }) catch unreachable; defer listener.deinit(io); var capture: CollectorCapture = .{}; var collector = try io.concurrent(Collector.receive, .{ &listener, io, &capture }); defer _ = collector.cancel(io) catch {};
var endpoint_buffer: [160]u8 = undefined; const endpoint_entry = try std.fmt.bufPrint( &endpoint_buffer, "OTEL_EXPORTER_OTLP_ENDPOINT=http://127.0.0.1:{d}/collector", .{listener.socket.address.getPort()}, ); var runtime = try Runtime.init(testing.allocator, .{ .io = io, .environment = &.{ endpoint_entry, "OTEL_EXPORTER_OTLP_HEADERS=x-api-key=offline-secret", "OTEL_EXPORTER_OTLP_PROTOCOL=http/json", "OTEL_BSP_SCHEDULE_DELAY=60000", }, .service_name = "stream-offline-test", .service_version = "v-test", }); defer runtime.deinit(); try testing.expect(runtime.enabled());
var parent = Span.start(testing.allocator, "offline-contract", "provider-shutdown-span", root_context, &.{}); const parent_context = otel_api.trace.trace_context.getActiveSpanContext(parent.context()).?; var child = Span.start(testing.allocator, "offline-contract", "child-error-span", parent.context(), &.{}); const child_context = otel_api.trace.trace_context.getActiveSpanContext(child.context()).?; try testing.expectEqualSlices(u8, &parent_context.trace_id.bytes, &child_context.trace_id.bytes); try testing.expect(!std.mem.eql(u8, &parent_context.span_id.bytes, &child_context.span_id.bytes)); child.fail(error.OfflineContractFailure); child.deinit(); parent.succeed(); parent.deinit(); try testing.expectEqual(otel_api.common.ProcessResult.success, runtime.shutdown(2_000)); try collector.await(io);
try testing.expectEqualStrings("/collector/v1/traces", capture.targetSlice()); try testing.expectEqualStrings("application/x-protobuf", capture.content_type[0..capture.content_type_len]); try testing.expectEqualStrings("offline-secret", capture.api_key[0..capture.api_key_len]); try testing.expect(std.mem.indexOf(u8, capture.bodySlice(), "provider-shutdown-span") != null); try testing.expect(std.mem.indexOf(u8, capture.bodySlice(), "child-error-span") != null); try testing.expect(std.mem.indexOf(u8, capture.bodySlice(), "OfflineContractFailure") != null); try testing.expect(std.mem.indexOf(u8, capture.bodySlice(), "jetstream/offline-contract") != null); try testing.expect(std.mem.indexOf(u8, capture.bodySlice(), "stream-offline-test") != null); try testing.expect(std.mem.indexOf(u8, capture.bodySlice(), "v-test") != null);}
test "no endpoint installs a callable no-op tracer and shuts down promptly" { var runtime = try Runtime.init(std.testing.allocator, .{ .environment = &.{}, .service_name = "test", .service_version = "test", }); defer runtime.deinit(); try std.testing.expect(!runtime.enabled()); const propagation_fields = try otel_api.getGlobalTextMapPropagator().fields(std.testing.allocator); defer std.testing.allocator.free(propagation_fields); try std.testing.expectEqualSlices([]const u8, &.{ "traceparent", "tracestate", "baggage" }, propagation_fields); var span_tracer = try tracer("test"); var span = try span_tracer.startSpan("noop-span", .{}, &.{}); span.end(null); span.deinit(); try std.testing.expectEqual(otel_api.common.ProcessResult.success, runtime.shutdown(10));}