diff --git a/README.md b/README.md index 56b54764..05b969f5 100644 --- a/README.md +++ b/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::115813213817:layer:dash0-extension-manual:`. -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 diff --git a/integration-tests/iac/lambdas/manual/tracing.js b/integration-tests/iac/lambdas/manual/tracing.js index 49b2d595..d89cfd7d 100644 --- a/integration-tests/iac/lambdas/manual/tracing.js +++ b/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}`, }, diff --git a/opt/java/opentelemetry-java-distro/custom/src/main/java/io/dash0/javaagent/Dash0Configurator.java b/opt/java/opentelemetry-java-distro/custom/src/main/java/io/dash0/javaagent/Dash0Configurator.java index 5d93ded1..9fe4df00 100644 --- a/opt/java/opentelemetry-java-distro/custom/src/main/java/io/dash0/javaagent/Dash0Configurator.java +++ b/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) { diff --git a/opt/node/distro/src/constants.ts b/opt/node/distro/src/constants.ts index 11d1c634..d1ff01fe 100644 --- a/opt/node/distro/src/constants.ts +++ b/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'; diff --git a/opt/python/distro/src/dash0_opentelemetry/__init__.py b/opt/python/distro/src/dash0_opentelemetry/__init__.py index a3e39205..090ae266 100644 --- a/opt/python/distro/src/dash0_opentelemetry/__init__.py +++ b/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")), ) ) diff --git a/src/config/endpoints.rs b/src/config/endpoints.rs index 8eef4425..e2f7088a 100644 --- a/src/config/endpoints.rs +++ b/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::().ok()) + .unwrap_or(crate::DEFAULT_OTLP_HTTP_PORT) +} + /// Latch in the API endpoints defined in ENV variables /// #[allow(dead_code)] diff --git a/src/extension/telemetry_receiver.rs b/src/extension/telemetry_receiver.rs index ccb59bf6..05151ff6 100644 --- a/src/extension/telemetry_receiver.rs +++ b/src/extension/telemetry_receiver.rs @@ -30,6 +30,8 @@ pub async fn telemetry(req: Request) -> Result, hype body_text ); + let mut saw_runtime_done = false; + if let Ok(mut logs) = serde_json::from_str::>(&body_text) { @@ -37,17 +39,13 @@ pub async fn telemetry(req: Request) -> Result, hype 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) -> Result, 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()) } diff --git a/src/main.rs b/src/main.rs index 8be3c786..d8eea6fb 100644 --- a/src/main.rs +++ b/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(listener: TcpListener, handler: F) -> tokio::task::JoinHandle<()> +where + F: Fn(hyper::Request) -> Fut + Copy + Send + 'static, + Fut: std::future::Future< + Output = Result, 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 - ); - } + }) } diff --git a/src/route.rs b/src/route.rs index ef3e05e9..c0c9a019 100644 --- a/src/route.rs +++ b/src/route.rs @@ -114,8 +114,20 @@ static ROUTES: Lazy = 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> = 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) -> Result, 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) -> Result, 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, ReqBody>> = Lazy::new(|| { let https = HttpsConnectorBuilder::new()