feat: Implements OTEL Server and Output Capture

Author: andyquinterom

Cargo.lock

@@ -46,3 +46,4 @@ instant-acme = { version = "0.8.4", default-features = false, features = ["hyper
 reqwest = { version = "0.12", default-features = false, features = ["rustls-tls-manual-roots", "json"] }
 x509-parser = "0.18.0"
 http = "1.4.0"
+axum = "0.8.8"

Cargo.toml

@@ -1,14 +1,16 @@
 # Telemetry Configuration
 
-Dispenser includes a built-in, high-performance telemetry system powered by [Delta Lake](https://delta.io/). It allows you to automatically collect deployment events and container health status, writing them directly to data lakes (S3, GCS, Azure) or local filesystems in Parquet format.
+Dispenser includes a built-in, high-performance telemetry system powered by [Delta Lake](https://delta.io/). It allows you to automatically collect deployment events, container health status, application logs/traces, and raw container output, writing them directly to data lakes (S3, GCS, Azure) or local filesystems in Parquet format.
 
 ## Overview
 
 The telemetry system runs in a dedicated, isolated thread to ensure that heavy I/O operations never block the main orchestration loop. It provides:
 
 1.  **Deployment Tracking**: Every time a container is created, updated, or restarted, a detailed event is logged.
 2.  **Health Monitoring**: Periodically samples the status of all managed containers (CPU, memory, uptime, health checks).
-3.  **Delta Lake Integration**: Writes data using the Delta Lake protocol, enabling ACID transactions, scalable metadata handling, and direct compatibility with tools like Spark, Trino, Athena, and Databricks.
+3.  **Application Telemetry (OTLP)**: Ingests structured logs and traces from services using standard OpenTelemetry SDKs.
+4.  **Container Output**: Captures raw `stdout` and `stderr` streams from all managed containers with sequence-guaranteed ordering.
+5.  **Delta Lake Integration**: Writes data using the Delta Lake protocol, enabling ACID transactions, scalable metadata handling, and direct compatibility with tools like Spark, Trino, Athena, and Databricks.
 
 ## Configuration
 
@@ -24,6 +26,9 @@ enabled = true
 # Supported schemes: file://, s3://, gs://, az://, adls://
 table_uri_deployments = "s3://my-data-lake/dispenser/deployments"
 table_uri_status = "s3://my-data-lake/dispenser/status"
+table_uri_logs = "s3://my-data-lake/dispenser/logs"
+table_uri_traces = "s3://my-data-lake/dispenser/traces"
+table_uri_container_output = "s3://my-data-lake/dispenser/container-output"
 
 # Optional: How often to sample container status (default: 60 seconds)
 status_interval = 60
@@ -69,9 +74,25 @@ Dispenser supports several authentication methods via environment variables:
 *   Service Principal: `AZURE_CLIENT_ID`, `AZURE_CLIENT_SECRET`, `AZURE_TENANT_ID`.
 *   Managed Identity (if running on Azure VMs/AKS).
 
+## OpenTelemetry (OTLP) Ingestion
+
+Dispenser acts as a sidecar host for your services. When telemetry is enabled, Dispenser starts an **Ingestion Service** listening on.
+
+*   **Endpoint**: `http://0.0.0.0:4318`
+*   **Protocol**: OTLP/HTTP (JSON)
+
+### Automatic Environment Variables
+
+Dispenser automatically injects the following environment variables into all managed containers to simplify instrumentation:
+
+*   `OTEL_EXPORTER_OTLP_ENDPOINT="http://172.28.0.1:4318"`
+*   `OTEL_SERVICE_NAME="{service_name}"` (The name from your `service.toml`)
+
+Standard OTel SDKs will automatically detect these variables and begin shipping logs and traces to Dispenser without further configuration.
+
 ## Data Schemas
 
-Dispenser automatically manages two Delta tables. It will create them if they do not exist.
+Dispenser automatically manages several Delta tables. It will create them if they do not exist.
 
 ### Deployments Table (`dispenser-deployments`)
 
@@ -116,6 +137,56 @@ Records periodic snapshots of the runtime state of containers.
 | `failing_streak` | `INTEGER` | Consecutive healthcheck failures. |
 | `last_health_output` | `STRING` | Output of the last failed healthcheck (truncated). |
 
+### Logs Table (`dispenser-logs`)
+
+Stores structured logs emitted by applications via OTel.
+
+| Column | Type | Description |
+| :--- | :--- | :--- |
+| `date` | `DATE` | Partition column (UTC). |
+| `timestamp` | `TIMESTAMP` | Exact time of the log entry. |
+| `service` | `STRING` | Service name. |
+| `severity` | `STRING` | INFO, WARN, ERROR, etc. |
+| `body` | `STRING` | The log message. |
+| `trace_id` | `STRING` | Associated trace ID (hex). |
+| `span_id` | `STRING` | Associated span ID (hex). |
+| `attributes` | `MAP<STRING, STRING>` | Flattened log attributes. |
+| `resource` | `MAP<STRING, STRING>` | Resource attributes (pod, node, etc). |
+
+### Traces Table (`dispenser-traces`)
+
+Stores distributed tracing spans.
+
+| Column | Type | Description |
+| :--- | :--- | :--- |
+| `date` | `DATE` | Partition column. |
+| `trace_id` | `STRING` | Trace ID (32-char hex). |
+| `span_id` | `STRING` | Span ID (16-char hex). |
+| `parent_span_id` | `STRING` | Parent Span ID. |
+| `name` | `STRING` | Span name (e.g., "GET /api/users"). |
+| `kind` | `STRING` | SERVER, CLIENT, PRODUCER, etc. |
+| `start_time` | `TIMESTAMP` | Start time. |
+| `end_time` | `TIMESTAMP` | End time. |
+| `duration_ms` | `LONG` | Calculated duration. |
+| `status_code` | `STRING` | OK, ERROR. |
+| `status_message` | `STRING` | Error description. |
+| `service` | `STRING` | Service name. |
+| `attributes` | `MAP<STRING, STRING>` | Span attributes. |
+
+### Container Output Table (`dispenser-container-output`)
+
+Captures raw `stdout` and `stderr` streams.
+
+| Column | Type | Description |
+| :--- | :--- | :--- |
+| `date` | `DATE` | Partition column. |
+| `timestamp` | `TIMESTAMP` | Exact time of the log line. |
+| `service` | `STRING` | Service name. |
+| `container_id` | `STRING` | Full container ID. |
+| `stream` | `STRING` | `stdout` or `stderr`. |
+| `message` | `STRING` | The raw log line. |
+| `sequence` | `LONG` | Monotonically increasing counter for perfect ordering. |
+
 ## Performance Tuning
 
 ### Buffering & Latency
@@ -144,4 +215,4 @@ The telemetry service runs on a dedicated Tokio runtime spawned in a separate OS
 To prevent indefinite storage growth, Dispenser applies the following default retention policies during table creation:
 
 *   **Log Retention**: 30 days (Deployments), 7 days (Status). Delta log history is kept for time-travel queries.
-*   **Deleted Files**: 7 days (Deployments), 1 day (Status). Vacuum operations can reclaim space after this period.
+*   **Deleted Files**: 7 days (Deployments), 1 day (Status). Vacuum operations can reclaim space after this period.

TELEMETRY.md

@@ -4,6 +4,14 @@ delay = 60
 [proxy]
 enabled = true
 
+[telemetry]
+enabled = true
+table_uri_deployments = "telem/deployments"
+table_uri_status = "telem/status"
+table_uri_logs = "telem/logs"
+table_uri_container_output = "telem/container_out"
+table_uri_traces = "telem/traces"
+
 [[service]]
 path = "service1"
 

example/dispenser.toml

@@ -5,9 +5,11 @@ use crate::{
     cli::Commands,
     proxy::{acme, run_dummy_proxy, run_proxy, ProxySignals},
     service::{
+        file::TelemetryConfig,
         manager::{ServiceMangerConfig, ServicesManager},
         vars::ServiceConfigError,
     },
+    telemetry::{TelemetryClient, TelemetryService},
 };
 
 mod cli;
@@ -81,31 +83,7 @@ async fn main() -> ExitCode {
     };
 
     let telemetry_client =
-        if let Some(telemetry_config) = &service_manager_config.entrypoint_file.telemetry {
-            if telemetry_config.enabled {
-                let (tx, rx) = tokio::sync::mpsc::channel(10000);
-                let config = telemetry_config.clone();
-
-                std::thread::spawn(move || {
-                    let rt = tokio::runtime::Builder::new_multi_thread()
-                        .worker_threads(2)
-                        .enable_all()
-                        .build()
-                        .expect("Failed to build telemetry runtime");
-
-                    rt.block_on(async {
-                        let service = crate::telemetry::TelemetryService::new(config, rx);
-                        service.run().await;
-                    });
-                });
-
-                Some(crate::telemetry::TelemetryClient::new(tx))
-            } else {
-                None
-            }
-        } else {
-            None
-        };
+        init_telemetry(service_manager_config.entrypoint_file.telemetry.as_ref());
 
     let manager =
         match ServicesManager::from_config(service_manager_config, None, telemetry_client.clone())
@@ -226,3 +204,37 @@ async fn main() -> ExitCode {
         }
     }
 }
+
+fn init_telemetry(config: Option<&TelemetryConfig>) -> Option<TelemetryClient> {
+    let telemetry_config = config?;
+    if !telemetry_config.enabled {
+        return None;
+    }
+
+    let (tx, rx) = tokio::sync::mpsc::channel(10000);
+    let config = telemetry_config.clone();
+
+    // Run ingestion service on main tokio runtime
+    let tx_ingestion = tx.clone();
+    tokio::spawn(async move {
+        if let Err(e) = crate::telemetry::ingestion::start_ingestion_service(tx_ingestion).await {
+            log::error!("OTLP ingestion service failed: {}", e);
+        }
+    });
+
+    // Telemetry Service runs in its own thread/runtime for Delta Lake operations
+    std::thread::spawn(move || {
+        let rt = tokio::runtime::Builder::new_multi_thread()
+            .worker_threads(2)
+            .enable_all()
+            .build()
+            .expect("Failed to build telemetry runtime");
+
+        rt.block_on(async {
+            let service = TelemetryService::new(config, rx);
+            service.run().await;
+        });
+    });
+
+    Some(TelemetryClient::new(tx))
+}

src/main.rs

@@ -66,6 +66,9 @@ pub struct TelemetryConfig {
     pub enabled: bool,
     pub table_uri_deployments: String,
     pub table_uri_status: String,
+    pub table_uri_logs: String,
+    pub table_uri_traces: String,
+    pub table_uri_container_output: String,
     pub buffer_size: Option<usize>,
     #[serde(default = "default_status_interval")]
     pub status_interval: u64,