feat: support otlp logs in the proxy (#33)

Author: jaronoff97

.github/workflows/release.yaml

@@ -17,6 +17,7 @@ on:
         type: choice
         options:
           - datadog
+          - otlp
 
 env:
   REGISTRY: ghcr.io
@@ -35,7 +36,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        distribution: ${{ github.event_name == 'workflow_dispatch' && fromJson(format('["{0}"]', inputs.distribution)) || fromJson('["datadog"]') }}
+        distribution: ${{ github.event_name == 'workflow_dispatch' && fromJson(format('["{0}"]', inputs.distribution)) || fromJson('["datadog", "otlp"]') }}
         arch:
           - platform: linux/amd64
             runner: ubuntu-latest
@@ -116,7 +117,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        distribution: ${{ github.event_name == 'workflow_dispatch' && fromJson(format('["{0}"]', inputs.distribution)) || fromJson('["datadog"]') }}
+        distribution: ${{ github.event_name == 'workflow_dispatch' && fromJson(format('["{0}"]', inputs.distribution)) || fromJson('["datadog", "otlp"]') }}
     steps:
       - name: Prepare
         run: |

Dockerfile

@@ -1,6 +1,6 @@
 # Multi-stage Dockerfile for Tero Edge distributions
 # Build specific distributions with: docker build --build-arg DISTRIBUTION=<name> -t tero-edge-<name> .
-# Available distributions: datadog, otlp (future), full
+# Available distributions: datadog, otlp
 #
 # =============================================================================
 # Build stage

README.md

@@ -1,17 +1,22 @@
 # Tero Edge
 
-Tero Edge is a lightweight, high-performance telemetry proxy that enables efficient telemetry processing via a set of policies. The repository is structured as modular packages that can be composed to enable multiple use cases and scenarios.
+Tero Edge is a lightweight, high-performance telemetry proxy that enables efficient telemetry processing via a set of policies. The repository is structured as modular packages that can be composed to enable multiple use cases and scenarios. The proxy demonstrates how to implement the policy spec described in [this OTEP](https://github.com/open-telemetry/opentelemetry-specification/pull/4738). This project is not meant to replace the opentelemetry collector but rather work in tandem with it, providing a lightweight alternative for solely applying policies. It's expected that a follow up to this project will be a collector processor Policy implementation.
 
 ## Current Configurations
 
 1. **Edge data proxy** - Receives data in a gateway or sidecar configuration, processes the data through its policy engine, and forwards it to the next destination.
 
+2. **OTLP (OpenTelemetry) proxy** - Receives OpenTelemetry logs via the `/v1/logs` endpoint, applies policy-based filtering (DROP/KEEP), and forwards to OTLP-compatible backends.
+
+2. **Datadog proxy** - Receives OpenTelemetry logs via the `/api/v2/logs` endpoint, applies policy-based filtering (DROP/KEEP), and forwards to Datadog.
+
 ## Repository Structure
 
 ```
 src/
 ├── main.zig              # Default distribution entry point
 ├── datadog_main.zig      # Datadog-focused distribution entry point
+├── otlp_main.zig         # OTLP-focused distribution entry point
 ├── root.zig              # Library root (public API exports)
 │
 ├── policy/               # Policy management package
@@ -105,6 +110,7 @@ Protocol-specific request processing modules that plug into the proxy infrastruc
 - `ProxyModule` - Vtable interface for request processing
 - `ModuleRegistration` - Module configuration (routes, upstream, etc.)
 - `DatadogModule` - Datadog log ingestion with policy-based filtering
+- `OtlpModule` - OpenTelemetry log ingestion with policy-based filtering
 - `PassthroughModule` - No-op passthrough for unhandled routes
 
 **Integration:**
@@ -135,7 +141,24 @@ const registrations = [_]modules.ModuleRegistration{
 };
 ```
 
-**Dependencies:** `policy/` (for DatadogModule), `observability/`
+**Dependencies:** `policy/` (for DatadogModule, OtlpModule), `observability/`
+
+#### OtlpModule
+
+The OTLP module handles OpenTelemetry log ingestion with policy-based filtering. It processes the standard OTLP JSON format (`ExportLogsServiceRequest`).
+
+**Supported filter match cases:**
+- `log_body` - Match against log message body
+- `log_severity_text` - Match against severity text (INFO, DEBUG, ERROR, etc.)
+- `log_severity_number` - Match against severity number (1-24)
+- `log_attribute` - Match against log record attributes
+- `resource_attribute` - Match against resource attributes (e.g., `service.name`)
+- `scope_name` - Match against instrumentation scope name
+- `scope_version` - Match against instrumentation scope version
+- `scope_attribute` - Match against scope attributes
+
+**Routes:**
+- `POST /v1/logs` - OTLP log ingestion endpoint
 
 ---
 
@@ -255,6 +278,19 @@ Focused distribution for Datadog log ingestion with:
 
 Build: `zig build` (default) or `zig build datadog`
 
+### OTLP (`otlp_main.zig`)
+Focused distribution for OpenTelemetry Protocol (OTLP) log ingestion with:
+- Handles `/v1/logs` endpoint (OTLP JSON format)
+- Policy-based log filtering (DROP/KEEP) on log body, severity, attributes
+- Support for resource attributes, scope attributes, and log attributes
+- Fail-open behavior (errors pass logs through unchanged)
+- Lock-free policy updates via atomic snapshots
+- Graceful shutdown with signal handling
+- Compatible with any OTLP-receiving backend (Datadog, Jaeger, etc.)
+
+Build: `zig build otlp`
+Run: `zig build run-otlp` or `./zig-out/bin/edge-otlp [config-file]`
+
 ## Building
 
 ```bash
@@ -266,16 +302,53 @@ zig build test
 
 # Build specific distribution
 zig build datadog
+zig build otlp
+
+# Run specific distribution
+zig build run-datadog
+zig build run-otlp
 ```
 
 ## Configuration
 
-See `config.json` for example configuration. Key settings:
+See `config.json` for the Datadog distribution or `config-otlp.json` for the OTLP distribution.
+
+### Key settings:
 - `listen_address` / `listen_port` - Server bind address
 - `upstream_url` - Default upstream destination
+- `workspace_id` - Workspace identifier for policy sync
+- `log_level` - Logging level (trace, debug, info, warn, err)
 - `policy_providers` - List of policy sources (file/http)
 - `max_body_size` - Request/response body limits
 
+### Example OTLP Configuration (`config-otlp.json`):
+
+```json
+{
+  "listen_address": "127.0.0.1",
+  "listen_port": 8080,
+  "upstream_url": "https://otlp.us5.datadoghq.com",
+  "workspace_id": "your-workspace-id",
+  "log_level": "info",
+  "max_body_size": 1048576,
+  "policy_providers": [
+    {
+      "id": "file",
+      "type": "file",
+      "path": "policies.json"
+    },
+    {
+      "id": "http",
+      "type": "http",
+      "url": "http://localhost:9090/v1/policy/sync"
+    }
+  ]
+}
+```
+
+### Environment Variables:
+- `TERO_LOG_LEVEL` - Override log level (trace, debug, info, warn, err)
+
 ## Design Principles
 
 1. **Data-Oriented Design** - Optimize for cache coherency and memory access patterns

Taskfile.yml

@@ -226,7 +226,8 @@ tasks:
   update:protos:
     desc: "Update proto files"
     cmds:
-      - buf export buf.build/tero/policy -o proto/policy
+      - buf export buf.build/tero/policy -o proto
+      - buf export buf.build/opentelemetry/opentelemetry -o proto
       - task: build:protos
 
   # =============================================================================
@@ -268,6 +269,8 @@ tasks:
     cmds:
       - task: build:container
         vars: { DIST: "datadog" }
+      - task: build:container
+        vars: { DIST: "otlp" }
 
   push:container:
     desc: "🚀 Push container to GCP Artifact Registry (DIST=datadog|otlp|full)"
@@ -301,6 +304,8 @@ tasks:
     cmds:
       - task: push:container
         vars: { DIST: "datadog" }
+      - task: push:container
+        vars: { DIST: "otlp" }
 
   release:
     desc: "🎯 Build and push all containers (for CI)"

build.zig

@@ -119,36 +119,48 @@ pub fn build(b: *std.Build) void {
     b.installArtifact(exe);
 
     // ==========================================================================
-    // Datadog Distribution
+    // Distribution Builds
     // ==========================================================================
-    // A focused distribution for Datadog log ingestion with filtering.
-    // Build with: zig build datadog
-    // Run with: zig build run-datadog
-    const datadog_exe = b.addExecutable(.{
-        .name = "edge-datadog",
-        .root_module = b.createModule(.{
-            .root_source_file = b.path("src/datadog_main.zig"),
-            .target = target,
-            .optimize = optimize,
-        }),
-    });
-    datadog_exe.root_module.addImport("httpz", httpz.module("httpz"));
-    datadog_exe.root_module.addImport("protobuf", protobuf_dep.module("protobuf"));
-    datadog_exe.root_module.addImport("proto", proto_mod);
-    datadog_exe.root_module.link_libc = true;
-    datadog_exe.root_module.linkSystemLibrary("z", .{});
-    datadog_exe.root_module.linkSystemLibrary("zstd", .{});
-    datadog_exe.root_module.linkSystemLibrary("hs", .{});
-
-    const datadog_step = b.step("datadog", "Build the Datadog distribution");
-    datadog_step.dependOn(&b.addInstallArtifact(datadog_exe, .{}).step);
-
-    const run_datadog_step = b.step("run-datadog", "Run the Datadog distribution");
-    const run_datadog_cmd = b.addRunArtifact(datadog_exe);
-    run_datadog_step.dependOn(&run_datadog_cmd.step);
-    run_datadog_cmd.step.dependOn(b.getInstallStep());
-    if (b.args) |args| {
-        run_datadog_cmd.addArgs(args);
+    // Each distribution is a focused edge proxy for a specific backend.
+    // Build with: zig build <distribution>
+    // Run with: zig build run-<distribution>
+
+    const distributions = .{
+        .{ "datadog", "src/datadog_main.zig", "Datadog log ingestion" },
+        .{ "otlp", "src/otlp_main.zig", "OpenTelemetry (OTLP) log ingestion" },
+    };
+
+    inline for (distributions) |dist| {
+        const name = dist[0];
+        const source = dist[1];
+        const desc = dist[2];
+
+        const dist_exe = b.addExecutable(.{
+            .name = "edge-" ++ name,
+            .root_module = b.createModule(.{
+                .root_source_file = b.path(source),
+                .target = target,
+                .optimize = optimize,
+            }),
+        });
+        dist_exe.root_module.addImport("httpz", httpz.module("httpz"));
+        dist_exe.root_module.addImport("protobuf", protobuf_dep.module("protobuf"));
+        dist_exe.root_module.addImport("proto", proto_mod);
+        dist_exe.root_module.link_libc = true;
+        dist_exe.root_module.linkSystemLibrary("z", .{});
+        dist_exe.root_module.linkSystemLibrary("zstd", .{});
+        dist_exe.root_module.linkSystemLibrary("hs", .{});
+
+        const dist_step = b.step(name, "Build the " ++ name ++ " distribution (" ++ desc ++ ")");
+        dist_step.dependOn(&b.addInstallArtifact(dist_exe, .{}).step);
+
+        const run_dist_step = b.step("run-" ++ name, "Run the " ++ name ++ " distribution");
+        const run_dist_cmd = b.addRunArtifact(dist_exe);
+        run_dist_step.dependOn(&run_dist_cmd.step);
+        run_dist_cmd.step.dependOn(b.getInstallStep());
+        if (b.args) |args| {
+            run_dist_cmd.addArgs(args);
+        }
     }
 
     // This creates a top level step. Top level steps have a name and can be
@@ -201,10 +213,11 @@ pub fn build(b: *std.Build) void {
         const protoc_step = protobuf.RunProtocStep.create(protobuf_dep.builder, target, .{
             .destination_directory = b.path("src/proto"),
             .source_files = &.{
-                "proto/policy/opentelemetry/proto/policy/v1/policy.proto",
+                "proto/opentelemetry/proto/policy/v1/policy.proto",
+                "proto/opentelemetry/proto/logs/v1/logs.proto",
             },
             .include_directories = &.{
-                "proto/policy",
+                "proto",
             },
         });
         gen_proto.dependOn(&protoc_step.step);

config-otlp.json

@@ -0,0 +1,20 @@
+{
+  "listen_address": "127.0.0.1",
+  "listen_port": 8080,
+  "upstream_url": "https://otlp.us5.datadoghq.com",
+  "workspace_id": "90A6EFC2-27B8-41BC-9343-43BFB1DF0732",
+  "log_level": "info",
+  "max_body_size": 1048576,
+  "policy_providers": [
+    {
+      "id": "file",
+      "type": "file",
+      "path": "policies.json"
+    },
+    {
+      "id": "http",
+      "type": "http",
+      "url": "http://localhost:9090/v1/policy/sync"
+    }
+  ]
+}

proto/policy/google/api/annotations.proto proto/google/api/annotations.protoproto/policy/google/api/annotations.proto renamed to proto/google/api/annotations.proto

@@ -0,0 +1,77 @@
+// Copyright 2020, OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+syntax = "proto3";
+
+package opentelemetry.proto.collector.logs.v1;
+
+import "opentelemetry/proto/logs/v1/logs.proto";
+
+option csharp_namespace = "OpenTelemetry.Proto.Collector.Logs.V1";
+option java_multiple_files = true;
+option java_package = "io.opentelemetry.proto.collector.logs.v1";
+option java_outer_classname = "LogsServiceProto";
+option go_package = "go.opentelemetry.io/proto/otlp/collector/logs/v1";
+
+// Service that can be used to push logs between one Application instrumented with
+// OpenTelemetry and an collector, or between an collector and a central collector (in this
+// case logs are sent/received to/from multiple Applications).
+service LogsService {
+  rpc Export(ExportLogsServiceRequest) returns (ExportLogsServiceResponse) {}
+}
+
+message ExportLogsServiceRequest {
+  // An array of ResourceLogs.
+  // For data coming from a single resource this array will typically contain one
+  // element. Intermediary nodes (such as OpenTelemetry Collector) that receive
+  // data from multiple origins typically batch the data before forwarding further and
+  // in that case this array will contain multiple elements.
+  repeated opentelemetry.proto.logs.v1.ResourceLogs resource_logs = 1;
+}
+
+message ExportLogsServiceResponse {
+  // The details of a partially successful export request.
+  //
+  // If the request is only partially accepted
+  // (i.e. when the server accepts only parts of the data and rejects the rest)
+  // the server MUST initialize the `partial_success` field and MUST
+  // set the `rejected_<signal>` with the number of items it rejected.
+  //
+  // Servers MAY also make use of the `partial_success` field to convey
+  // warnings/suggestions to senders even when the request was fully accepted.
+  // In such cases, the `rejected_<signal>` MUST have a value of `0` and
+  // the `error_message` MUST be non-empty.
+  //
+  // A `partial_success` message with an empty value (rejected_<signal> = 0 and
+  // `error_message` = "") is equivalent to it not being set/present. Senders
+  // SHOULD interpret it the same way as in the full success case.
+  ExportLogsPartialSuccess partial_success = 1;
+}
+
+message ExportLogsPartialSuccess {
+  // The number of rejected log records.
+  //
+  // A `rejected_<signal>` field holding a `0` value indicates that the
+  // request was fully accepted.
+  int64 rejected_log_records = 1;
+
+  // A developer-facing human-readable message in English. It should be used
+  // either to explain why the server rejected parts of the data during a partial
+  // success or to convey warnings/suggestions during a full success. The message
+  // should offer guidance on how users can address such issues.
+  //
+  // error_message is an optional field. An error_message with an empty value
+  // is equivalent to it not being set.
+  string error_message = 2;
+}