feat: Add message parameter to otel_debug macro

rust/otap-dataflow/crates/engine/src/lib.rs

@@ -998,6 +998,7 @@ impl<PData: 'static + Clone + Debug> PipelineFactory<PData> {
 
         otel_debug!(
             "receiver.create.start",
+            "Creating receiver node...",
             pipeline_group_id = pipeline_group_id.as_ref(),
             pipeline_id = pipeline_id.as_ref(),
             core_id = core_id,
@@ -1039,6 +1040,7 @@ impl<PData: 'static + Clone + Debug> PipelineFactory<PData> {
 
         otel_debug!(
             "receiver.create.complete",
+            "Receiver node created successfully.",
             pipeline_group_id = pipeline_group_id.as_ref(),
             pipeline_id = pipeline_id.as_ref(),
             core_id = core_id,
@@ -1062,6 +1064,7 @@ impl<PData: 'static + Clone + Debug> PipelineFactory<PData> {
 
         otel_debug!(
             "processor.create.start",
+            "Creating processor node...",
             pipeline_group_id = pipeline_group_id.as_ref(),
             pipeline_id = pipeline_id.as_ref(),
             core_id = core_id,
@@ -1103,6 +1106,7 @@ impl<PData: 'static + Clone + Debug> PipelineFactory<PData> {
 
         otel_debug!(
             "processor.create.complete",
+            "Processor node created successfully.",
             pipeline_group_id = pipeline_group_id.as_ref(),
             pipeline_id = pipeline_id.as_ref(),
             core_id = core_id,
@@ -1126,6 +1130,7 @@ impl<PData: 'static + Clone + Debug> PipelineFactory<PData> {
 
         otel_debug!(
             "exporter.create.start",
+            "Creating exporter node...",
             pipeline_group_id = pipeline_group_id.as_ref(),
             pipeline_id = pipeline_id.as_ref(),
             core_id = core_id,
@@ -1159,7 +1164,7 @@ impl<PData: 'static + Clone + Debug> PipelineFactory<PData> {
 
         otel_debug!(
             "exporter.create.complete",
-            pipeline_group_id = pipeline_group_id.as_ref(),
+            "Exporter node created successfully.",
             pipeline_id = pipeline_id.as_ref(),
             core_id = core_id,
             node_id = name.as_ref(),
@@ -1356,6 +1361,7 @@ where
                 })?;
             otel_debug!(
                 "pdata.sender.set",
+                "Setting pdata sender for source port...",
                 pipeline_group_id = pipeline_group_id.as_ref(),
                 pipeline_id = pipeline_id.as_ref(),
                 core_id = core_id,
@@ -1372,6 +1378,7 @@ where
                 })?;
             otel_debug!(
                 "pdata.receiver.set",
+                "Setting pdata receiver for destination node...",
                 pipeline_group_id = pipeline_group_id.as_ref(),
                 pipeline_id = pipeline_id.as_ref(),
                 core_id = core_id,

rust/otap-dataflow/crates/engine/src/pipeline_ctrl.rs

@@ -333,8 +333,8 @@ impl<PData> PipelineCtrlMsgManager<PData> {
                             if is_draining {
                                 otel_debug!(
                                     "pipeline.draining.ignored_start_timer",
+                                    "Ignoring StartTimer during shutdown draining",
                                     node_id = node_id,
-                                    "Ignoring StartTimer during shutdown draining"
                                 );
                             } else {
                                 self.tick_timers.start(node_id, duration);
@@ -350,8 +350,8 @@ impl<PData> PipelineCtrlMsgManager<PData> {
                             if is_draining {
                                 otel_debug!(
                                     "pipeline.draining.ignored_start_telemetry_timer",
-                                    node_id = node_id,
-                                    "Ignoring StartTelemetryTimer during shutdown draining"
+                                    "Ignoring StartTelemetryTimer during shutdown draining",
+                                    node_id = node_id
                                 );
                             } else {
                                 self.telemetry_timers.start(node_id, duration);

rust/otap-dataflow/crates/otap/src/durable_buffer_processor/mod.rs

@@ -790,6 +790,7 @@ impl DurableBuffer {
             if Instant::now() >= deadline {
                 otel_debug!(
                     "durable_buffer.drain.budget_exhausted",
+                    "time budget for draining bundles exhausted",
                     bundles_processed = bundles_processed,
                     budget_ms = drain_budget.as_millis()
                 );
@@ -800,6 +801,7 @@ impl DurableBuffer {
             if !self.can_send_more() {
                 otel_debug!(
                     "durable_buffer.drain.at_capacity",
+                    "max in-flight bundles reached",
                     bundles_processed = bundles_processed,
                     in_flight = self.pending_bundles.len(),
                     max_in_flight = self.config.max_in_flight
@@ -832,6 +834,7 @@ impl DurableBuffer {
                                 // are blocked. Break to avoid busy-looping.
                                 otel_debug!(
                                     "durable_buffer.drain.all_blocked",
+                                    "all available bundles are in-flight or scheduled for retry",
                                     bundles_processed = bundles_processed,
                                     in_flight = self.pending_bundles.len(),
                                     retry_scheduled = self.retry_scheduled.len()
@@ -847,6 +850,7 @@ impl DurableBuffer {
                             // processor handle other messages (including incoming data)
                             otel_debug!(
                                 "durable_buffer.drain.backpressure",
+                                "downstream channel is full, deferring remaining bundles",
                                 bundles_processed = bundles_processed
                             );
                             break;
@@ -972,6 +976,7 @@ impl DurableBuffer {
 
                         otel_debug!(
                             "durable_buffer.bundle.forwarded",
+                            "bundle forwarded downstream",
                             segment_seq = bundle_ref.segment_seq.raw(),
                             bundle_index = bundle_ref.bundle_index.raw(),
                             retry_count = retry_count
@@ -1056,6 +1061,7 @@ impl DurableBuffer {
                 .set(self.pending_bundles.len() as u64);
             otel_debug!(
                 "durable_buffer.bundle.acked",
+                "bundle ACK'd by downstream",
                 segment_seq = bundle_ref.segment_seq.raw(),
                 bundle_index = bundle_ref.bundle_index.raw()
             );
@@ -1102,6 +1108,7 @@ impl DurableBuffer {
 
             otel_debug!(
                 "durable_buffer.bundle.nacked",
+                "bundle NACK'd by downstream, scheduling retry",
                 segment_seq = bundle_ref.segment_seq.raw(),
                 bundle_index = bundle_ref.bundle_index.raw(),
                 retry_count = retry_count,
@@ -1164,6 +1171,7 @@ impl DurableBuffer {
             // Bundle stays in retry_scheduled (wasn't removed yet).
             otel_debug!(
                 "durable_buffer.retry.deferred",
+                "backpressure still present, deferring retry",
                 segment_seq = bundle_ref.segment_seq.raw(),
                 bundle_index = bundle_ref.bundle_index.raw(),
                 in_flight = self.pending_bundles.len(),
@@ -1209,6 +1217,7 @@ impl DurableBuffer {
                     ProcessBundleResult::Sent => {
                         otel_debug!(
                             "durable_buffer.retry.sent",
+                            "retry successful, bundle forwarded downstream",
                             segment_seq = bundle_ref.segment_seq.raw(),
                             bundle_index = bundle_ref.bundle_index.raw(),
                             retry_count = retry_count
@@ -1227,6 +1236,7 @@ impl DurableBuffer {
                         // Re-schedule retry with a short delay.
                         otel_debug!(
                             "durable_buffer.retry.backpressure",
+                            "backpressure on retry, deferring again",
                             segment_seq = bundle_ref.segment_seq.raw(),
                             bundle_index = bundle_ref.bundle_index.raw()
                         );
@@ -1251,6 +1261,7 @@ impl DurableBuffer {
                 // Claim failed - bundle may have been resolved or segment dropped
                 otel_debug!(
                     "durable_buffer.retry.claim_failed",
+                    "failed to claim bundle for retry, it may have been resolved or expired",
                     segment_seq = bundle_ref.segment_seq.raw(),
                     bundle_index = bundle_ref.bundle_index.raw(),
                     error = %e
@@ -1389,6 +1400,7 @@ impl otap_df_engine::local::processor::Processor<OtapPdata> for DurableBuffer {
             self.timer_started = true;
             otel_debug!(
                 "durable_buffer.timer.started",
+                "started periodic timer for polling Quiver",
                 poll_interval = ?self.config.poll_interval
             );
         }
@@ -1433,7 +1445,7 @@ impl otap_df_engine::local::processor::Processor<OtapPdata> for DurableBuffer {
                         })
                 }
                 NodeControlMsg::Config { config } => {
-                    otel_debug!("durable_buffer.config.update", config = ?config);
+                    otel_debug!("durable_buffer.config.update", "Durable buffer config update", config = ?config);
                     Ok(())
                 }
                 NodeControlMsg::DelayedData { data, .. } => {

rust/otap-dataflow/crates/otap/src/experimental/recordset_kql_processor/processor.rs

@@ -101,8 +101,8 @@ impl RecordsetKqlProcessor {
             OtlpProtoBytes::ExportLogsRequest(bytes) => {
                 otap_df_telemetry::otel_debug!(
                     "processor.processing_logs",
+                    "Processing KQL query",
                     processor = "recordset_kql",
-                    message = "Processing KQL query",
                     input_items = input_items,
                 );
                 self.process_logs(bytes, signal)
@@ -125,6 +125,7 @@ impl RecordsetKqlProcessor {
 
                 otap_df_telemetry::otel_debug!(
                     "processor.success",
+                    "KQL processing succeeded",
                     processor = "recordset_kql",
                     input_items = input_items,
                     output_items = output_items,
@@ -139,9 +140,10 @@ impl RecordsetKqlProcessor {
                 let message = e.to_string();
                 otap_df_telemetry::otel_debug!(
                     "processor.failure",
+                    "KQL processing failed",
                     processor = "recordset_kql",
                     input_items = input_items,
-                    message = message,
+                    error = message,
                 );
 
                 effect_handler

rust/otap-dataflow/crates/otap/src/fake_data_generator.rs

@@ -353,16 +353,16 @@ impl local::Receiver<OtapPdata> for FakeGeneratorReceiver {
                                 if remaining_time.as_secs_f64() > 0.0 {
                                     otel_debug!(
                                         "rate_limit.sleep",
+                                        "Sleeping to maintain configured signal rate",
                                         sleep_duration_ms = remaining_time.as_millis() as u64,
-                                        message = "Sleeping to maintain configured signal rate"
                                     );
                                     sleep(remaining_time).await;
                                 }
                                 // ToDo: Handle negative time, not able to keep up with specified rate limit
                             } else {
                                 otel_debug!(
                                     "rate_limit.uncapped",
-                                    message = "Rate limiting disabled, continuing immediately"
+                                    "Rate limiting disabled, continuing immediately"
                                 );
                             }
                         }

rust/otap-dataflow/crates/otap/src/otap_grpc/proxy.rs

@@ -235,7 +235,12 @@ impl ProxyConfig {
 
         // Check if host should bypass proxy
         if self.should_bypass(host, port) {
-            otel_debug!("proxy.bypass", host = host, port = port);
+            otel_debug!(
+                "proxy.bypass",
+                "bypassing proxy for target",
+                host = host,
+                port = port
+            );
             return None;
         }
 
@@ -248,11 +253,16 @@ impl ProxyConfig {
         if let Some(url) = proxy {
             otel_debug!(
                 "proxy.using",
+                "using proxy for target",
                 proxy = url.to_string(),
                 target = uri.to_string(),
             );
         } else {
-            otel_debug!("proxy.none", target = uri.to_string());
+            otel_debug!(
+                "proxy.none",
+                "no proxy for target",
+                target = uri.to_string()
+            );
         }
 
         proxy.map(|u| u.expose())
@@ -485,6 +495,7 @@ async fn http_connect_tunnel_on_stream(
     // Avoid logging the raw request to reduce the risk of leaking headers or internal targets.
     otel_debug!(
         "proxy.connect_request",
+        "sending HTTP CONNECT request to proxy",
         target = format!("{formatted_target}:{target_port}"),
         has_auth = proxy_auth.is_some()
     );
@@ -501,7 +512,11 @@ async fn http_connect_tunnel_on_stream(
         ));
     }
 
-    otel_debug!("proxy.connect_response", status_line = status_line.trim());
+    otel_debug!(
+        "proxy.connect_response",
+        "received HTTP CONNECT response from proxy",
+        status_line = status_line.trim()
+    );
 
     // Parse "HTTP/1.1 200 Connection established".
     // Be robust to multiple ASCII spaces/tabs between tokens.
@@ -618,7 +633,12 @@ pub(crate) async fn connect_tcp_stream_with_proxy_config(
     if let Some(proxy_url) = proxy_config.get_proxy_for_uri(target_uri) {
         let (proxy_host, proxy_port, proxy_auth) = parse_proxy_url(proxy_url)?;
 
-        otel_debug!("proxy.connecting", host = proxy_host, port = proxy_port);
+        otel_debug!(
+            "proxy.connecting",
+            "connecting to proxy",
+            host = proxy_host,
+            port = proxy_port
+        );
         let stream = TcpStream::connect((proxy_host.as_str(), proxy_port))
             .await
             .map_err(|e| {
@@ -631,7 +651,12 @@ pub(crate) async fn connect_tcp_stream_with_proxy_config(
                 );
                 ProxyError::ProxyConnectionFailed(e)
             })?;
-        otel_debug!("proxy.connected");
+        otel_debug!(
+            "proxy.connected",
+            "connected to proxy",
+            host = proxy_host,
+            port = proxy_port
+        );
 
         // Apply socket options to the proxy connection
         let stream = apply_socket_options(

rust/otap-dataflow/crates/otap/src/otlp_exporter.rs

@@ -195,10 +195,7 @@ impl Exporter<OtapPdata> for OTLPExporter {
                 msg
             } else if inflight_exports.is_empty() {
                 let msg = msg_chan.recv().await?;
-                otel_debug!(
-                    "exporter.receive",
-                    message = "Received message from pipeline"
-                );
+                otel_debug!("exporter.receive", "Received message from pipeline");
                 msg
             } else {
                 let completion_fut = inflight_exports.next_completion().fuse();
@@ -220,7 +217,7 @@ impl Exporter<OtapPdata> for OTLPExporter {
                     }
                     msg = recv_fut => {
                         let msg = msg?;
-                        otel_debug!("exporter.receive", message = "Received message from pipeline");
+                        otel_debug!("exporter.receive", "Received message from pipeline");
                         msg
                     },
                 }