feat: Add message parameter to otel_debug macro

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

@@ -334,7 +334,6 @@ impl<PData> PipelineCtrlMsgManager<PData> {
                                 otel_debug!(
                                     "pipeline.draining.ignored_start_timer",
                                     node_id = node_id,
-                                    "Ignoring StartTimer during shutdown draining"
                                 );
                             } else {
                                 self.tick_timers.start(node_id, duration);
@@ -350,8 +349,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/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)
@@ -141,7 +141,7 @@ impl RecordsetKqlProcessor {
                     "processor.failure",
                     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

@@ -631,7 +631,7 @@ pub(crate) async fn connect_tcp_stream_with_proxy_config(
                 );
                 ProxyError::ProxyConnectionFailed(e)
             })?;
-        otel_debug!("proxy.connected");
+        otel_debug!("proxy.connected", 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
                     },
                 }

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

@@ -844,7 +844,7 @@ pub async fn serve(
                             tokio::select! {
                                 res = &mut conn => {
                                     if let Err(err) = res {
-                                        otap_df_telemetry::otel_debug!("HttpConnectionError", error = err.to_string());
+                                        otap_df_telemetry::otel_debug!("http.connection_error", error = err.to_string());
                                     }
                                 },
                                 _ = shutdown.cancelled() => {

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

@@ -364,7 +364,12 @@ async fn add_system_trust_anchors_if_enabled(
     let mut store = RootCertStore::empty();
     // Best-effort: accept that some system certs might not parse.
     let (added, ignored) = store.add_parsable_certificates(roots);
-    otel_debug!("Loaded system CA certificates", added, ignored,);
+    otel_debug!(
+        "loaded.system.ca.certificates",
+        "Loaded system CA certificates",
+        added,
+        ignored,
+    );
 
     Ok(tls.trust_anchors(store.roots))
 }
@@ -667,7 +672,7 @@ impl CaWatcherState {
 
     /// Process a file system event, potentially triggering a reload.
     fn process_event(&self, event: Event) {
-        otel_debug!("File watcher event", event = ?event);
+        otel_debug!("tls.file_watcher.event", "File watcher event", event = ?event);
 
         // Filter out irrelevant event types early (before expensive path checks)
         if matches!(event.kind, notify::EventKind::Access(_)) {
@@ -678,7 +683,10 @@ impl CaWatcherState {
             return;
         }
 
-        otel_debug!("Event matches our CA file, proceeding with reload check");
+        otel_debug!(
+            "tls.file_watcher.match",
+            "Event matches our CA file, proceeding with reload check"
+        );
 
         // Small delay to allow filesystem operations to complete (e.g., atomic renames).
         // This blocks the notify thread briefly, but is acceptable because:
@@ -708,6 +716,7 @@ impl CaWatcherState {
 
         if !is_match {
             otel_debug!(
+                "tls.file_watcher.no_match",
                 "Event not for our file",
                 event_paths = ?event.paths,
                 watched_path = ?self.watched_path,
@@ -723,23 +732,34 @@ impl CaWatcherState {
         let current_identity = match get_file_identity(&self.reload_path) {
             Ok(id) => id,
             Err(e) => {
-                otel_debug!("Failed to get file identity, skipping reload", error = ?e);
+                otel_debug!("tls.file_watcher.identity_error", "Failed to get file identity, skipping reload", error = ?e);
                 return false;
             }
         };
 
         let prev_identity = self.last_identity.load(Ordering::Relaxed);
         if current_identity == prev_identity {
-            otel_debug!("File identity unchanged, skipping reload");
+            otel_debug!(
+                "tls.file_watcher.identity_unchanged",
+                "File identity unchanged, skipping reload"
+            );
             return false;
         }
-        otel_debug!("File identity changed", prev_identity, current_identity);
+        otel_debug!(
+            "tls.file_watcher.identity_changed",
+            "File identity changed",
+            prev_identity,
+            current_identity
+        );
 
         // Check debounce window
         let now = current_timestamp();
         let last = self.last_reload.load(Ordering::Relaxed);
         if now.saturating_sub(last) < CA_RELOAD_DEBOUNCE_SECS {
-            otel_debug!("Debouncing CA file change event");
+            otel_debug!(
+                "tls.file_watcher.debounce",
+                "Debouncing CA file change event"
+            );
             return false;
         }
 
@@ -749,7 +769,10 @@ impl CaWatcherState {
             .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
             .is_err()
         {
-            otel_debug!("CA reload already in progress, skipping");
+            otel_debug!(
+                "tls.file_watcher.reload_in_progress",
+                "CA reload already in progress, skipping"
+            );
             return false;
         }
 
@@ -889,7 +912,11 @@ impl ReloadableClientCaVerifier {
     ) -> Result<Arc<Self>, io::Error> {
         // Initial load
         let ca_pem = read_file_with_limit_sync(&ca_file_path)?;
-        otel_debug!("Initial CA PEM size", size_bytes = ca_pem.len());
+        otel_debug!(
+            "tls.ca.initial_load",
+            "Initial CA PEM size",
+            size_bytes = ca_pem.len()
+        );
         let verifier = build_webpki_verifier(&ca_pem, include_system_cas)?;
 
         let inner = Arc::new(ArcSwap::from_pointee(verifier));
@@ -1191,7 +1218,11 @@ fn build_webpki_verifier(
         ));
     }
 
-    otel_debug!("Built verifier with CA certificates", count);
+    otel_debug!(
+        "tls.ca.verifier_built",
+        "Built verifier with CA certificates",
+        count
+    );
 
     WebPkiClientVerifier::builder(roots.into())
         .build()
@@ -1204,7 +1235,11 @@ fn reload_ca_verifier(
     include_system_cas: bool,
 ) -> Result<Arc<dyn ClientCertVerifier>, io::Error> {
     let ca_pem = read_file_with_limit_sync(ca_path)?;
-    otel_debug!("Reloaded CA PEM", size_bytes = ca_pem.len());
+    otel_debug!(
+        "tls.ca.reloaded",
+        "Reloaded CA PEM",
+        size_bytes = ca_pem.len()
+    );
     build_webpki_verifier(&ca_pem, include_system_cas)
 }
 
@@ -1390,7 +1425,10 @@ async fn build_client_auth(
 
     if !has_ca_file && !has_ca_pem && !include_system_cas {
         // No client auth configured
-        otel_debug!("No client CA configured, disabling client authentication");
+        otel_debug!(
+            "tls.mtls.disabled",
+            "No client CA configured, disabling client authentication"
+        );
         return Ok(builder.with_no_client_auth());
     }
 

rust/otap-dataflow/crates/otap/tests/mtls_tests.rs

@@ -107,7 +107,7 @@ async fn test_mtls_client_cert_verification() {
         match tls_acceptor.accept(stream).await {
             Ok(_tls_stream) => true,
             Err(e) => {
-                otel_debug!("Server handshake failed", error = ?e);
+                otel_debug!("handshake.failed","Server handshake failed", error = ?e);
                 false
             }
         }

rust/otap-dataflow/crates/otap/tests/otlp_exporter_proxy_tls.rs

@@ -193,7 +193,7 @@ async fn start_tls_logs_server() -> (
 ) {
     if let Err(err) = rustls::crypto::ring::default_provider().install_default() {
         // It's fine if the provider is already installed (e.g. by another test)
-        otel_debug!("rustls default provider installation failed in test", error = ?err);
+        otel_debug!("provider.installation.failed","rustls default provider installation failed in test", error = ?err);
     }
 
     let (ca, ca_issuer) = new_ca();

rust/otap-dataflow/crates/telemetry/src/internal_events.rs

@@ -69,10 +69,17 @@ macro_rules! otel_warn {
 /// # Example:
 /// ```ignore
 /// use otap_df_telemetry::otel_debug;
-/// otel_debug!("processing.batch", batch_size = 100);
+/// otel_debug!("processing.batch", "Processing batch of items", batch_size = 100);
+/// otel_debug!("processing.done", count = 5);  // no message, just fields
+/// otel_debug!("processing.start");            // event name only
 /// ```
 #[macro_export]
 macro_rules! otel_debug {
+    // With message
+    ($name:expr, $message:literal $(, $($fields:tt)*)?) => {
+        $crate::_private::debug!(name: $name, target: env!("CARGO_PKG_NAME"), { $($($fields)*)? }, $message);
+    };
+    // Without message (uses empty string as message)
     ($name:expr $(, $($fields:tt)*)?) => {
         $crate::_private::debug!(name: $name, target: env!("CARGO_PKG_NAME"), { $($($fields)*)? }, "");
     };
@@ -155,4 +162,28 @@ mod tests {
         raw_error!("raw error message", error = ?err);
         raw_error!("simple error message");
     }
+
+    #[test]
+    fn test_otel_debug() {
+        otel_debug!("debug.event");
+    }
+
+    #[test]
+    fn test_otel_debug_with_attributes() {
+        otel_debug!("debug.event.with_attributes", value = 42);
+    }
+
+    #[test]
+    fn test_otel_debug_with_message() {
+        otel_debug!("debug.event.with_message", "This is a debug message");
+    }
+
+    #[test]
+    fn test_otel_debug_with_message_and_attributes() {
+        otel_debug!(
+            "debug.event.with_message_and_attributes",
+            "This is a debug message",
+            value = 42
+        );
+    }
 }

rust/otap-dataflow/crates/telemetry/src/tracing_init.rs

@@ -206,7 +206,7 @@ mod tests {
         for level in ALL_LEVELS {
             let setup = test_setup(ProviderSetup::Noop, level);
             setup.with_subscriber(|| {
-                otel_debug!("debug");
+                otel_debug!("debug", "debug message");
                 otel_info!("info");
                 otel_warn!("warn");
                 otel_error!("error");
@@ -227,7 +227,7 @@ mod tests {
         for level in ALL_LEVELS {
             let setup = test_setup(ProviderSetup::ConsoleDirect, level);
             setup.with_subscriber(|| {
-                otel_debug!("debug");
+                otel_debug!("debug", "debug message");
                 otel_info!("info");
                 otel_warn!("warn");
                 otel_error!("error");
@@ -258,7 +258,7 @@ mod tests {
             let (reporter, receiver) = test_reporter();
             let setup = test_setup(ProviderSetup::InternalAsync { reporter }, level);
             setup.with_subscriber(|| {
-                otel_debug!("debug");
+                otel_debug!("debug", "debug message");
                 otel_info!("info");
                 otel_warn!("warn");
                 otel_error!("error");
@@ -283,7 +283,7 @@ mod tests {
         let setup = test_setup(ProviderSetup::InternalAsync { reporter }, LogLevel::Info);
 
         setup.with_subscriber(|| {
-            otel_debug!("filtered");
+            otel_debug!("filtered", "debug message filtered out");
         });
 
         assert!(
@@ -298,7 +298,7 @@ mod tests {
         let setup = test_setup(ProviderSetup::InternalAsync { reporter }, LogLevel::Warn);
 
         setup.with_subscriber(|| {
-            otel_debug!("filtered");
+            otel_debug!("filtered", "debug message filtered out");
             otel_info!("filtered");
             otel_warn!("not_filtered");
         });
@@ -315,7 +315,7 @@ mod tests {
         let setup = test_setup(ProviderSetup::InternalAsync { reporter }, LogLevel::Error);
 
         setup.with_subscriber(|| {
-            otel_debug!("filtered");
+            otel_debug!("filtered", "debug message filtered out");
             otel_info!("filtered");
             otel_warn!("filtered");
             otel_error!("not_filtered");
@@ -332,7 +332,7 @@ mod tests {
         let setup = test_setup(ProviderSetup::InternalAsync { reporter }, LogLevel::Off);
 
         setup.with_subscriber(|| {
-            otel_debug!("filtered");
+            otel_debug!("filtered", "debug message filtered out");
             otel_info!("filtered");
             otel_warn!("filtered");
             otel_error!("filtered");
@@ -347,7 +347,7 @@ mod tests {
         let setup = test_setup(ProviderSetup::InternalAsync { reporter }, LogLevel::Debug);
 
         setup.with_subscriber(|| {
-            otel_debug!("d");
+            otel_debug!("d", "debug message");
             otel_info!("i");
             otel_warn!("w");
             otel_error!("e");
@@ -403,7 +403,7 @@ mod tests {
         let setup = test_setup(ProviderSetup::InternalAsync { reporter }, LogLevel::Warn);
 
         setup.with_subscriber(|| {
-            otel_debug!("filtered");
+            otel_debug!("filtered", "debug message filtered out");
             otel_info!("filtered");
             otel_warn!("not_filtered");
             otel_error!("not_filtered");