Add dedicated OTLP/HTTP listener on port 4318

README.md

@@ -125,10 +125,12 @@ The following environment variables allow fine-grained control over secret maski
 If you prefer to set up OpenTelemetry instrumentation yourself instead of relying on the extension's auto-instrumentation, you can use the manual layer and point your OTLP exporters to the extension's local endpoint. The extension will receive the telemetry, enrich it, and forward it to Dash0.
 
 1. Add the manual layer to your Lambda function: `arn:aws:lambda:<region>:115813213817:layer:dash0-extension-manual:<version>`.
-2. Configure your OTLP trace exporter to send to `http://127.0.0.1:9009/v1/traces`.
-3. If exporting metrics, configure your OTLP metric exporter to send to `http://127.0.0.1:9009/v1/metrics`.
+2. Configure your OTLP trace exporter to send to `http://127.0.0.1:4318/v1/traces` (OTLP/HTTP).
+3. If exporting metrics, configure your OTLP metric exporter to send to `http://127.0.0.1:4318/v1/metrics` (OTLP/HTTP).
 4. Make sure to flush all telemetry before the Lambda invocation completes (e.g., in a response hook or before returning the response).
 
+The extension accepts OTLP over HTTP on the default OpenTelemetry HTTP port `4318` (overridable via the `DASH0_OTLP_HTTP_PORT` environment variable). OTLP/gRPC is not supported. For backwards compatibility, OTLP/HTTP is also still accepted on the extension's proxy port (`9009`).
+
 
 ## Enrichment Attributes
 

integration-tests/iac/lambdas/manual/tracing.js

@@ -16,8 +16,9 @@ const resource = resourceFromAttributes({
   [ATTR_SERVICE_NAME]: process.env.AWS_LAMBDA_FUNCTION_NAME || 'lambda-metrics',
 });
 
+// The extension accepts OTLP/HTTP on the default OpenTelemetry HTTP port (4318).
 const exporter = new OTLPTraceExporter({
-  url: `http://127.0.0.1:9009/v1/traces`,
+  url: `http://127.0.0.1:4318/v1/traces`,
 });
 
 const provider = new NodeTracerProvider({
@@ -29,7 +30,7 @@ provider.register();
 
 // Metrics setup
 const metricExporter = new OTLPMetricExporter({
-  url: `http://127.0.0.1:9009/v1/metrics`,
+  url: `http://127.0.0.1:4318/v1/metrics`,
   headers: {
     Authorization: `Bearer ${process.env.DASH0_TOKEN}`,
   },

opt/java/opentelemetry-java-distro/custom/src/main/java/io/dash0/javaagent/Dash0Configurator.java

@@ -47,7 +47,7 @@ public class Dash0Configurator implements AutoConfigurationCustomizerProvider {
   public static final Logger LOGGER = Logger.getLogger(Dash0Configurator.class.getName());
 
   public static final String DASH0_EXTENSION_ENDPOINT_URL =
-      "http://127.0.0.1:9009";
+      "http://127.0.0.1:4318";
 
   @Override
   public void customize(AutoConfigurationCustomizer autoConfiguration) {

opt/node/distro/src/constants.ts

@@ -1,4 +1,4 @@
 export const DASH0_LOGGING_NAMESPACE = '@dash0/opentelemetry';
 
 export const DEFAULT_DASH0_EXTENSION_ENDPOINT =
-  'http://127.0.0.1:9009/v1/traces';
+  'http://127.0.0.1:4318/v1/traces';

opt/python/distro/src/dash0_opentelemetry/__init__.py

@@ -102,7 +102,7 @@ def init() -> Dict[str, Any]:
     from opentelemetry.sdk.trace.export import BatchSpanProcessor
     from opentelemetry.sdk.trace import SpanLimits, TracerProvider
 
-    DASH0_EXTENSION_ENDPOINT = "http://127.0.0.1:9009/v1/traces"
+    DASH0_EXTENSION_ENDPOINT = "http://127.0.0.1:4318/v1/traces"
 
     traces_endpoint = os.getenv("DASH0_EXTENSION_ENDPOINT", DASH0_EXTENSION_ENDPOINT)
     dash0_token = os.getenv("DASH0_TOKEN")
@@ -124,6 +124,11 @@ def init() -> Dict[str, Any]:
                 endpoint=traces_endpoint,
                 headers={"Authorization": f"Bearer {dash0_token}"},
             ),
+            # The default 5s schedule delay races against Lambda killing the
+            # runtime on timeout, losing spans that are still queued. Export
+            # aggressively: the endpoint is the extension on loopback, and the
+            # Java distro batches equally tightly (otel.bsp.schedule.delay=10ms).
+            schedule_delay_millis=int(os.getenv("OTEL_BSP_SCHEDULE_DELAY", "10")),
         )
     )
 

src/config/endpoints.rs

@@ -16,6 +16,17 @@ pub fn extension_port() -> u16 {
         .unwrap_or(crate::DEFAULT_PROXY_PORT)
 }
 
+/// Get the OTLP/HTTP receiver port from environment variable or use the
+/// default OpenTelemetry HTTP port (4318)
+pub fn otlp_http_port() -> u16 {
+    use std::env::var;
+
+    var("DASH0_OTLP_HTTP_PORT")
+        .ok()
+        .and_then(|v| v.parse::<u16>().ok())
+        .unwrap_or(crate::DEFAULT_OTLP_HTTP_PORT)
+}
+
 /// Latch in the API endpoints defined in ENV variables
 ///
 #[allow(dead_code)]

src/extension/telemetry_receiver.rs

@@ -30,24 +30,22 @@ pub async fn telemetry(req: Request<Incoming>) -> Result<Response<ResBody>, hype
         body_text
     );
 
+    let mut saw_runtime_done = false;
+
     if let Ok(mut logs) =
         serde_json::from_str::<Vec<crate::state::invocation_data::TelemetryLog>>(&body_text)
     {
         crate::util::log_processing::process_telemetry_logs(&mut logs);
 
         for log in &logs {
             if log.r#type == "platform.runtimeDone" {
+                saw_runtime_done = true;
                 if let Some(id) = &log.invocation_id {
                     let is_error = error_invocation_ids.iter().any(|(eid, _)| eid == id);
                     if !is_error {
                         create_supplementary_spans(id, true);
                     }
                 }
-                if let Some(notifier) = crate::state::invocation_data::take_runtime_done_notifier()
-                {
-                    tracing::info!("[{}] Signaled platform.runtimeDone", crate::log_prefix());
-                    let _ = notifier.send(());
-                }
             }
 
             if log.r#type == "platform.report" {
@@ -106,5 +104,16 @@ pub async fn telemetry(req: Request<Incoming>) -> Result<Response<ResBody>, hype
         flush_telemetry_logs(None).await;
     }
 
+    // Signal platform.runtimeDone only after error traces have been built and
+    // sent. Signaling earlier lets the send-on-invocation-end flush task drain
+    // the stored traces concurrently, so build_synthetic_trace can no longer
+    // recover the invocation's real trace id and correlation is lost.
+    if saw_runtime_done {
+        if let Some(notifier) = crate::state::invocation_data::take_runtime_done_notifier() {
+            tracing::info!("[{}] Signaled platform.runtimeDone", crate::log_prefix());
+            let _ = notifier.send(());
+        }
+    }
+
     Ok(Response::builder().status(200).body(empty_body()).unwrap())
 }

src/main.rs

@@ -28,6 +28,9 @@ pub const EXTENSION_NAME: &str = "dash0";
 /// Default port to listen on, overriden by DASH0_LISTENER_PORT environment variable
 pub const DEFAULT_PROXY_PORT: u16 = 9009;
 
+/// Default OpenTelemetry OTLP/HTTP port, overriden by DASH0_OTLP_HTTP_PORT environment variable
+pub const DEFAULT_OTLP_HTTP_PORT: u16 = 4318;
+
 pub static LAMBDA_RUNTIME_API_VERSION: &str = "2018-06-01";
 
 /// Returns the log prefix for standard log messages
@@ -90,20 +93,67 @@ async fn main() {
             );
         }
     };
+
+    // Runtime API proxy, plus OTLP/HTTP for backwards compatibility.
+    let proxy_listener = bind_listener(addr).await;
+    let server_join_handle = spawn_http1_server(proxy_listener, route::dispatch);
+
+    // Dedicated OTLP/HTTP receiver on the default OpenTelemetry HTTP port.
+    let otlp_http_addr = SocketAddr::from(([0, 0, 0, 0], config::endpoints::otlp_http_port()));
+    let otlp_http_listener = bind_listener(otlp_http_addr).await;
+    spawn_http1_server(otlp_http_listener, route::dispatch_otlp);
+
+    // Initialize the extension and continually get next extension event.
+    tokio::task::spawn(async {
+        extension::register::register().await;
+        extension::register::register_telemetry().await;
+        // Lambda Application runtime will start once our extension is registered
+        stats::app_start();
+
+        loop {
+            // Lambda Extension API requires we wait for next extension event
+            extension::events::get_next().await;
+        }
+    });
+
+    if let Err(e) = server_join_handle.await {
+        tracing::error!(
+            "[{}] Failed to join server task: {}",
+            crate::log_prefix(),
+            e
+        );
+    }
+}
+
+async fn bind_listener(addr: SocketAddr) -> TcpListener {
     tracing::info!("[{}] listening on {}", crate::log_prefix(), addr);
 
-    let listener = match TcpListener::bind(addr).await {
+    match TcpListener::bind(addr).await {
         Ok(listener) => listener,
         Err(e) => {
-            tracing::error!("[{}] Failed to bind listener: {}", crate::log_prefix(), e);
+            tracing::error!(
+                "[{}] Failed to bind listener on {}: {}",
+                crate::log_prefix(),
+                addr,
+                e
+            );
             panic!(
                 "[{}] Cannot start without bound listener",
                 crate::log_prefix()
             );
         }
-    };
+    }
+}
 
-    let server_join_handle = tokio::spawn(async move {
+fn spawn_http1_server<F, Fut>(listener: TcpListener, handler: F) -> tokio::task::JoinHandle<()>
+where
+    F: Fn(hyper::Request<hyper::body::Incoming>) -> Fut + Copy + Send + 'static,
+    Fut: std::future::Future<
+            Output = Result<hyper::Response<route::ResBody>, std::convert::Infallible>,
+        > + Send
+        + 'static,
+{
+    tokio::spawn(async move {
         loop {
             let (stream, _peer) = match listener.accept().await {
                 Ok(pair) => pair,
@@ -115,33 +165,12 @@ async fn main() {
             let io = TokioIo::new(stream);
             tokio::spawn(async move {
                 if let Err(e) = http1::Builder::new()
-                    .serve_connection(io, service_fn(route::dispatch))
+                    .serve_connection(io, service_fn(handler))
                     .await
                 {
                     tracing::debug!("[{}] connection error: {}", crate::log_prefix(), e);
                 }
             });
         }
-    });
-
-    // Initialize the extension and continually get next extension event.
-    tokio::task::spawn(async {
-        extension::register::register().await;
-        extension::register::register_telemetry().await;
-        // Lambda Application runtime will start once our extension is registered
-        stats::app_start();
-
-        loop {
-            // Lambda Extension API requires we wait for next extension event
-            extension::events::get_next().await;
-        }
-    });
-
-    if let Err(e) = server_join_handle.await {
-        tracing::error!(
-            "[{}] Failed to join server task: {}",
-            crate::log_prefix(),
-            e
-        );
-    }
+    })
 }

src/route.rs

@@ -114,8 +114,20 @@ static ROUTES: Lazy<Routes> = Lazy::new(|| {
     Routes { get, post }
 });
 
+/// Routes for the dedicated OTLP/HTTP listener on the default OpenTelemetry
+/// HTTP port. Only OTLP ingest paths are exposed there — the Lambda Runtime
+/// API proxy routes stay on the proxy listener.
+static OTLP_ROUTES: Lazy<MatchitRouter<Handler>> = Lazy::new(|| {
+    let mut post = MatchitRouter::new();
+    post.insert("/v1/traces", h_traces as Handler).unwrap();
+    post.insert("/v1/logs", h_logs as Handler).unwrap();
+    post.insert("/v1/metrics", h_metrics as Handler).unwrap();
+    post
+});
+
 pub fn init() {
     Lazy::force(&ROUTES);
+    Lazy::force(&OTLP_ROUTES);
     Lazy::force(&HTTPS_CLIENT);
     invocation_entry::force_init();
 }
@@ -143,6 +155,31 @@ pub async fn dispatch(req: Request<Incoming>) -> Result<Response<ResBody>, Infal
     }
 }
 
+/// Dispatch for the dedicated OTLP/HTTP listener: only OTLP ingest routes,
+/// anything else is a 404 (no runtime proxy passthrough).
+pub async fn dispatch_otlp(req: Request<Incoming>) -> Result<Response<ResBody>, Infallible> {
+    let handler = if *req.method() == Method::POST {
+        OTLP_ROUTES.at(req.uri().path()).ok().map(|m| *m.value)
+    } else {
+        None
+    };
+
+    let Some(handler) = handler else {
+        return Ok(Response::builder().status(404).body(empty_body()).unwrap());
+    };
+
+    match handler(req).await {
+        Ok(resp) => Ok(resp),
+        Err(e) => {
+            tracing::error!("[{}] Handler error: {}", crate::log_prefix(), e);
+            Ok(Response::builder()
+                .status(500)
+                .body(full_body(Bytes::from_static(b"500 - Internal Error")))
+                .unwrap())
+        }
+    }
+}
+
 pub(crate) static HTTPS_CLIENT: Lazy<Client<hyper_rustls::HttpsConnector<HttpConnector>, ReqBody>> =
     Lazy::new(|| {
         let https = HttpsConnectorBuilder::new()