otlp: implement exporting signals' data

Signed-off-by: inge4pres <[email protected]>

Author: inge4pres

src/api/metrics/instrument.zig

@@ -745,7 +745,7 @@ fn testCounterAddingOne(counter: *Counter(u64)) !void {
 fn testCounterCollect(counter: *Counter(u64)) !void {
     // FIXME flaky test can result in failure, so we added a sleep but we should find a more robust solution.
     while (!counter.lock.tryLock()) {
-        std.time.sleep(10);
+        std.time.sleep(44);
     }
     counter.lock.unlock();
 
@@ -790,7 +790,7 @@ fn testHistogramRecordOne(histogram: *Histogram(u64)) !void {
 fn testHistogramCollect(histogram: *Histogram(u64)) !void {
     // FIXME flaky test can result in failure, so we added a sleep but we should find a more robust solution.
     while (!histogram.lock.tryLock()) {
-        std.time.sleep(10);
+        std.time.sleep(44);
     }
     histogram.lock.unlock();
 

src/otlp.zig

@@ -20,6 +20,11 @@ pub const ConfigError = error{
     InvalidProtocol,
 };
 
+/// Error set for the OTLP Export operation.
+pub const ExportError = error{
+    UnimplementedTransportProtocol,
+};
+
 /// The combination of underlying transport protocol and format used to send the data.
 pub const Protocol = enum {
     // In order of precedence: SDK MUST support http/protobuf and SHOULD support grpc and http/json.
@@ -75,15 +80,60 @@ pub const Signal = enum {
     // TODO add other signals when implemented
     // profiles,
 
-    fn defaulttHttpPath(self: Signal) []const u8 {
+    const Self = @This();
+
+    fn defaulttHttpPath(self: Self) []const u8 {
         switch (self) {
             .metrics => return "/v1/metrics",
             .logs => return "/v1/logs",
             .traces => return "/v1/traces",
         }
     }
+
+    /// Actual signal data as protobuf messages.
+    pub const Data = union(Self) {
+        metrics: pbmetrics.MetricsData,
+        logs: pblogs.LogsData,
+        traces: pbtrace.TracesData,
+        // TODO add other signals when implemented
+        // profiles: profiles.ProfilesData,
+
+        fn toOwnedSlice(self: Data, allocator: std.mem.Allocator, protocol: Protocol) ![]const u8 {
+            return switch (protocol) {
+                .http_json => {
+                    switch (self) {
+                        // All protobuf-generated structs have a json_encode method.
+                        inline else => |data| return data.json_encode(.{}, allocator),
+                    }
+                },
+                .http_protobuf, .grpc => {
+                    switch (self) {
+                        // All protobuf-generated structs have a encode method.
+                        inline else => |data| return data.encode(allocator),
+                    }
+                },
+            };
+        }
+
+        fn signal(self: Data) Signal {
+            return std.meta.activeTag(self);
+        }
+    };
 };
 
+test "otlp Signal.Data get payload bytes" {
+    const allocator = std.testing.allocator;
+    var data = Signal.Data{
+        .metrics = pbmetrics.MetricsData{
+            .resource_metrics = std.ArrayList(pbmetrics.ResourceMetrics).init(allocator),
+        },
+    };
+    const payload = try data.toOwnedSlice(allocator, Protocol.http_protobuf);
+    defer allocator.free(payload);
+
+    try std.testing.expectEqual(payload.len, 0);
+}
+
 /// Configuration options for the OTLP transport.
 pub const ConfigOptions = struct {
     allocator: std.mem.Allocator,
@@ -293,6 +343,8 @@ const HTTPClient = struct {
     config: ConfigOptions,
     // Default HTTP Client
     client: http.Client,
+    // Retries are processed using a separate thread.
+    // A priority queue is maintained in the ExpBackoffRetry struct.
     retry: *ExpBackoffRetry,
 
     pub fn init(allocator: std.mem.Allocator, config: ConfigOptions) !*Self {
@@ -339,9 +391,10 @@ const HTTPClient = struct {
         return request_options;
     }
 
-    /// For the Signal type, send the data to the OTLP endpoint using the client's configuration.
-    /// Data passed as argument should either be protobuf or JSON encoded, as specified in the config.
-    pub fn send(self: *Self, signal: Signal, data: []u8) !void {
+    // Send the OTLP data to the url using the client's configuration.
+    // Data passed as argument should either be protobuf or JSON encoded, as specified in the config.
+    // Data will be compressed here.
+    fn send(self: *Self, url: []const u8, data: []u8) !void {
         var resp_body = std.ArrayList(u8).init(self.allocator);
         defer resp_body.deinit();
 
@@ -363,9 +416,6 @@ const HTTPClient = struct {
         };
         defer self.allocator.free(req_body);
 
-        const url = try self.config.httpUrlForSignal(signal, self.allocator);
-        defer self.allocator.free(url);
-
         const req_opts = try requestOptions(self.config);
 
         const fetch_request = http.Client.FetchOptions{
@@ -654,7 +704,41 @@ test "otlp HTTPClient send fails for missing server" {
     const client = try HTTPClient.init(allocator, config.*);
     defer client.deinit();
 
+    const url = try config.httpUrlForSignal(.metrics, allocator);
+    defer allocator.free(url);
+
     var payload = [_]u8{0} ** 1024;
-    const result = client.send(Signal.metrics, &payload);
+    const result = client.send(url, &payload);
     try std.testing.expectError(std.posix.ConnectError.ConnectionRefused, result);
 }
+
+const pbmetrics = @import("opentelemetry/proto/metrics/v1.pb.zig");
+const pblogs = @import("opentelemetry/proto/logs/v1.pb.zig");
+const pbtrace = @import("opentelemetry/proto/trace/v1.pb.zig");
+
+/// Export the data to the OTLP endpoint using the options configured with ConfigOptions.
+pub fn Export(
+    allocator: std.mem.Allocator,
+    config: ConfigOptions,
+    otlp_payload: Signal.Data,
+) !void {
+    // Determine the type of client to be used, currently only HTTP is supported.
+    const client = switch (config.protocol) {
+        .http_json, .http_protobuf => try HTTPClient.init(allocator, config),
+        .grpc => return ExportError.UnimplementedTransportProtocol,
+    };
+
+    const payload = otlp_payload.toOwnedSlice(allocator, config.protocol) catch |err| {
+        std.debug.print("OTLP transport: failed to encode payload via {s}: {?}\n", .{ @tagName(config.protocol), err });
+        return err;
+    };
+    defer allocator.free(payload);
+
+    const url = try config.httpUrlForSignal(otlp_payload.signal(), allocator);
+    defer allocator.free(url);
+
+    client.send(url, payload) catch |err| {
+        std.debug.print("OTLP transport: failed to send payload: {?}\n", .{err});
+        return err;
+    };
+}

src/sdk/metrics/exporters/otlp.zig

@@ -27,7 +27,7 @@ const MetricReadError = @import("../reader.zig").MetricReadError;
 const otlp = @import("../../../otlp.zig");
 
 /// Exports metrics via the OpenTelemetry Protocol (OTLP).
-/// OTLP is a binary protocol used for transmitting telemetry data, encoding them with protobuf.
+/// OTLP is a binary protocol used for transmitting telemetry data, encoding them with protobuf or JSON.
 /// See https://opentelemetry.io/docs/specs/otlp/
 pub const OTLPExporter = struct {
     const Self = @This();
@@ -56,8 +56,7 @@ pub const OTLPExporter = struct {
             }
             self.allocator.free(data);
         }
-        var output = try self.allocator.create(pbmetrics.ResourceMetrics);
-        defer output.deinit(self.allocator);
+        var resource_metrics = try self.allocator.alloc(pbmetrics.ResourceMetrics, 1);
 
         var scope_metrics = try self.allocator.alloc(pbmetrics.ScopeMetrics, data.len);
         for (data, 0..) |measurement, i| {
@@ -68,14 +67,27 @@ pub const OTLPExporter = struct {
                     // .schema_url = if (measurement.meterSchemaUrl) |s| ManagedString.managed(s) else .Empty,
                     .attributes = try attributesToProtobufKeyValueList(self.allocator, measurement.meterAttributes),
                 },
+                .schema_url = if (measurement.meterSchemaUrl) |s| ManagedString.managed(s) else .Empty,
                 .metrics = try toProtobufMetric(self.allocator, measurement, self.temporailty(measurement.instrumentKind)),
             };
         }
-        output.* = pbmetrics.ResourceMetrics{
+        resource_metrics[0] = pbmetrics.ResourceMetrics{
             .resource = null, //FIXME support resource attributes
             .scope_metrics = scope_metrics,
             .schema_url = .Empty,
         };
+
+        const metrics_data = pbmetrics.MetricsData{
+            .resource_metrics = std.ArrayList(pbmetrics.ResourceMetrics).fromOwnedSlice(self.allocator, resource_metrics),
+        };
+        defer metrics_data.deinit(self.allocator);
+
+        // TODO: offload the data the the OTLP transport.
+        // Problem is: the OTLP transport should be the one in charge of detecting the encoding via the configuration.
+        otlp.Export(self.config, otlp.Signal{ .metrics = metrics_data }) catch |err| {
+            std.debug.print("OTLP export failed: {s}", .{@tagName(err)});
+            return MetricReadError.ExportFailed;
+        };
     }
 };