exporters(otlp): wire in OTLP transport

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

Author: inge4pres

src/otlp.zig

@@ -340,14 +340,14 @@ const HTTPClient = struct {
     const Self = @This();
 
     allocator: std.mem.Allocator,
-    config: ConfigOptions,
+    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 {
+    pub fn init(allocator: std.mem.Allocator, config: *ConfigOptions) !*Self {
         try config.validate();
 
         const s = try allocator.create(Self);
@@ -370,7 +370,7 @@ const HTTPClient = struct {
         self.allocator.destroy(self);
     }
 
-    fn requestOptions(config: ConfigOptions) !http.Client.RequestOptions {
+    fn requestOptions(config: *ConfigOptions) !http.Client.RequestOptions {
         const headers: http.Client.Request.Headers = .{
             .accept_encoding = if (config.compression.encodingHeaderValue()) |v| .{ .override = v } else .default,
             .content_type = .{ .override = switch (config.protocol) {
@@ -394,7 +394,7 @@ const HTTPClient = struct {
     // 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 {
+    fn send(self: *Self, url: []const u8, data: []const u8) !void {
         var resp_body = std.ArrayList(u8).init(self.allocator);
         defer resp_body.deinit();
 
@@ -701,7 +701,7 @@ test "otlp HTTPClient send fails for missing server" {
     var config = try ConfigOptions.init(allocator);
     defer config.deinit();
 
-    const client = try HTTPClient.init(allocator, config.*);
+    const client = try HTTPClient.init(allocator, config);
     defer client.deinit();
 
     const url = try config.httpUrlForSignal(.metrics, allocator);
@@ -719,7 +719,7 @@ 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,
+    config: *ConfigOptions,
     otlp_payload: Signal.Data,
 ) !void {
     // Determine the type of client to be used, currently only HTTP is supported.

src/sdk/metrics/exporter.zig

@@ -13,6 +13,9 @@ const Attributes = @import("../../attributes.zig").Attributes;
 const InMemoryExporter = @import("./exporters/in_memory.zig").InMemoryExporter;
 const StdoutExporter = @import("./exporters/stdout.zig").StdoutExporter;
 
+const otlp = @import("exporters/otlp.zig");
+const OTLPExporter = @import("./exporters/otlp.zig").OTLPExporter;
+
 const view = @import("view.zig");
 
 pub const ExportResult = enum {
@@ -90,6 +93,23 @@ pub const MetricExporter = struct {
         return .{ .exporter = exporter, .stdout = stdout };
     }
 
+    pub fn OTLP(
+        allocator: std.mem.Allocator,
+        temporality: ?view.TemporalitySelector,
+        aggregation: ?view.AggregationSelector,
+        options: *otlp.ConfigOptions,
+    ) !struct { exporter: *MetricExporter, otlp: *OTLPExporter } {
+        const temporality_ = temporality orelse view.DefaultTemporality;
+
+        const otlp_exporter = try OTLPExporter.init(allocator, options, temporality_);
+        const exporter = try MetricExporter.new(allocator, &otlp_exporter.exporter);
+        // Default configuration
+        exporter.temporality = temporality_;
+        exporter.aggregation = aggregation orelse view.DefaultAggregation;
+
+        return .{ .exporter = exporter, .otlp = otlp_exporter };
+    }
+
     /// ExportBatch exports a batch of metrics data by calling the exporter implementation.
     /// The passed metrics data will be owned by the exporter implementation.
     //TODO exportBatch MUST have a timeout

src/sdk/metrics/exporters/otlp.zig

@@ -35,57 +35,72 @@ pub const OTLPExporter = struct {
     allocator: std.mem.Allocator,
     exporter: ExporterImpl,
 
-    temporailty: view.TemporalitySelector,
+    temporality: view.TemporalitySelector,
+    config: *otlp.ConfigOptions,
+
+    pub fn init(allocator: std.mem.Allocator, config: *otlp.ConfigOptions, temporality: view.TemporalitySelector) !*Self {
+        const s = try allocator.create(Self);
+        s.* = Self{
+            .allocator = allocator,
+            .exporter = ExporterImpl{
+                .exportFn = exportBatch,
+            },
+            .temporality = temporality,
+            .config = config,
+        };
+        return s;
+    }
 
-    config: otlp.ConfigOptions,
+    pub fn deinit(self: *Self) void {
+        self.allocator.destroy(self);
+    }
 
     pub fn exportBatch(iface: *ExporterImpl, data: []Measurements) MetricReadError!void {
         // Get a pointer to the instance of the struct that implements the interface.
         const self: *Self = @fieldParentPtr("exporter", iface);
-
-        // TODO: implement the OTLP exporter.
-        // Processing pipeline:
-        // 1. convert the measurements to protobuf format (clear up the datapoints after reading them).
-        // 2. create an HTTP or gRPC client.
-        // 3. send the data to the endpoint.
-        // 4. handle the response and potential retries (exp backoff).
-
+        // Cleanup the data after use, it is mandatory for all exporters as they own the data argument.
         defer {
             for (data) |*m| {
                 m.deinit(self.allocator);
             }
             self.allocator.free(data);
         }
-        var resource_metrics = try self.allocator.alloc(pbmetrics.ResourceMetrics, 1);
+        var resource_metrics = self.allocator.alloc(pbmetrics.ResourceMetrics, 1) catch |err| {
+            std.debug.print("OTLP export failed to allocate memory for resource metrics: {s}\n", .{@errorName(err)});
+            return MetricReadError.OutOfMemory;
+        };
 
         var scope_metrics = try self.allocator.alloc(pbmetrics.ScopeMetrics, data.len);
         for (data, 0..) |measurement, i| {
+            const metrics = std.ArrayList(pbmetrics.Metric).initCapacity(self.allocator, 1) catch |err| {
+                std.debug.print("OTLP export failed to allocate memory for metrics: {s}\n", .{@errorName(err)});
+                return MetricReadError.OutOfMemory;
+            };
+            metrics.items[0] = try toProtobufMetric(self.allocator, measurement, self.temporality);
+            const attributes = try attributesToProtobufKeyValueList(self.allocator, measurement.meterAttributes);
             scope_metrics[i] = pbmetrics.ScopeMetrics{
                 .scope = pbcommon.InstrumentationScope{
                     .name = ManagedString.managed(measurement.meterName),
                     .version = if (measurement.meterVersion) |version| ManagedString.managed(version) else .Empty,
-                    // .schema_url = if (measurement.meterSchemaUrl) |s| ManagedString.managed(s) else .Empty,
-                    .attributes = try attributesToProtobufKeyValueList(self.allocator, measurement.meterAttributes),
+                    .attributes = attributes.values,
                 },
                 .schema_url = if (measurement.meterSchemaUrl) |s| ManagedString.managed(s) else .Empty,
-                .metrics = try toProtobufMetric(self.allocator, measurement, self.temporailty(measurement.instrumentKind)),
+                .metrics = metrics,
             };
         }
         resource_metrics[0] = pbmetrics.ResourceMetrics{
             .resource = null, //FIXME support resource attributes
-            .scope_metrics = scope_metrics,
+            .scope_metrics = std.ArrayList(pbmetrics.ScopeMetrics).fromOwnedSlice(self.allocator, 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);
+        defer metrics_data.deinit();
 
-        // 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)});
+        otlp.Export(self.allocator, self.config, otlp.Signal.Data{ .metrics = metrics_data }) catch |err| {
+            std.debug.print("OTLP export failed: {s}", .{@errorName(err)});
             return MetricReadError.ExportFailed;
         };
     }
@@ -298,3 +313,12 @@ test "exporters/otlp conversion for HistogramDataPoint" {
         metric.data.?.histogram.data_points.items[0].attributes.items[0].value.?.value.?.string_value,
     });
 }
+
+test "exporters/otlp init/deinit" {
+    const allocator = std.testing.allocator;
+    const config = try otlp.ConfigOptions.init(allocator);
+    defer config.deinit();
+
+    var exporter = try OTLPExporter.init(allocator, config, view.DefaultTemporality);
+    defer exporter.deinit();
+}

src/sdk/metrics/reader.zig

@@ -34,6 +34,7 @@ pub const MetricReadError = error{
     ExportFailed,
     ForceFlushTimedOut,
     ConcurrentCollectNotAllowed,
+    OutOfMemory,
 };
 
 /// MetricReader reads metrics' data from a MeterProvider.