Rate-limit all tracing/logging to prevent log spam (#209)

Wraps all warn/error/info tracing calls that could fire repeatedly
in the `rate_limited!` macro (at most once per 60s per call site).

One-time lifecycle events (worker started/stopped, drain complete,
resolved bucket region, graceful_shutdown) are left un-rate-limited.

Also fixes the `rate_limited!` macro to use `std::time::Duration::MAX`
(fully qualified) so it works at call sites without `Duration` in scope.

Closes #139

Author: rcoh

dial9-tokio-telemetry/src/background_task/mod.rs

@@ -7,6 +7,7 @@ pub mod s3;
 pub(crate) mod sealed;
 
 use crate::metrics::{Operation, SegmentProcessMetrics, SegmentProcessMetricsGuard};
+use crate::rate_limit::rate_limited;
 use metrique::timers::Timer;
 use metrique_writer::BoxEntrySink;
 use pipeline_metrics::{MetriqueResult, PipelineMetrics, StageMetrics};
@@ -80,11 +81,13 @@ impl BackgroundTaskConfig {
         match stem {
             Some(s) if !s.is_empty() => s,
             _ => {
-                tracing::error!(
-                    target: "dial9_worker",
-                    path = %self.trace_path.display(),
-                    "trace_path has no file stem — pass a path like /tmp/traces/trace.bin, not a directory"
-                );
+                rate_limited!(Duration::from_secs(60), {
+                    tracing::error!(
+                        target: "dial9_worker",
+                        path = %self.trace_path.display(),
+                        "trace_path has no file stem — pass a path like /tmp/traces/trace.bin, not a directory"
+                    );
+                });
                 "trace"
             }
         }
@@ -378,7 +381,9 @@ impl SegmentProcessor for SymbolizeProcessor {
                     Ok(data)
                 }
                 Ok(Err(e)) => {
-                    tracing::warn!(target: "dial9_worker", error = %e, "symbolization failed, preserving original bytes");
+                    rate_limited!(Duration::from_secs(60), {
+                        tracing::warn!(target: "dial9_worker", error = %e, "symbolization failed, preserving original bytes");
+                    });
                     Err(ProcessError {
                         data,
                         kind: ProcessErrorKind::Io(e),
@@ -434,10 +439,12 @@ impl SegmentProcessor for WriteBackProcessor {
                                 let _ = std::fs::remove_file(&dest_path);
                             }
                             Err(e) => {
-                                tracing::warn!(
-                                    "failed to remove original segment {}: {e}",
-                                    original_path.display()
-                                );
+                                rate_limited!(Duration::from_secs(60), {
+                                    tracing::warn!(
+                                        "failed to remove original segment {}: {e}",
+                                        original_path.display()
+                                    );
+                                });
                             }
                         }
                     }
@@ -512,7 +519,9 @@ impl WorkerLoop {
         let segments = match sealed::find_sealed_segments(&self.dir, &self.stem) {
             Ok(s) => s,
             Err(e) => {
-                tracing::warn!(target: "dial9_worker", "failed to scan for sealed segments: {e}");
+                rate_limited!(Duration::from_secs(60), {
+                    tracing::warn!(target: "dial9_worker", "failed to scan for sealed segments: {e}");
+                });
                 return false;
             }
         };
@@ -540,7 +549,9 @@ impl WorkerLoop {
                     continue;
                 }
                 Err(e) => {
-                    tracing::warn!(target: "dial9_worker", error = %e, "failed to read segment");
+                    rate_limited!(Duration::from_secs(60), {
+                        tracing::warn!(target: "dial9_worker", error = %e, "failed to read segment");
+                    });
                     continue;
                 }
             };
@@ -592,9 +603,13 @@ impl WorkerLoop {
                             tracing::debug!(target: "dial9_worker", path = %segment.path.display(), "segment evicted during processing, skipping");
                         } else {
                             if let Err(remove_err) = std::fs::remove_file(&segment.path) {
-                                tracing::warn!(target: "dial9_worker", error = %remove_err, path = %segment.path.display(), "failed to remove corrupted segment");
+                                rate_limited!(Duration::from_secs(60), {
+                                    tracing::warn!(target: "dial9_worker", error = %remove_err, path = %segment.path.display(), "failed to remove corrupted segment");
+                                });
                             }
-                            tracing::warn!(target: "dial9_worker", error = %e.kind, cause = ?e.kind, path = %segment.path.display(), "processor failed, removing segment");
+                            rate_limited!(Duration::from_secs(60), {
+                                tracing::warn!(target: "dial9_worker", error = %e.kind, cause = ?e.kind, path = %segment.path.display(), "processor failed, removing segment");
+                            });
                         }
                         continue 'next_segment;
                     }
@@ -684,7 +699,9 @@ impl SegmentProcessor for S3PipelineUploader {
             {
                 Ok(key) => {
                     self.circuit_breaker.on_success();
-                    tracing::info!(target: "dial9_worker", "uploaded {key}");
+                    rate_limited!(Duration::from_secs(10), {
+                        tracing::info!(target: "dial9_worker", "uploaded {key}");
+                    });
                     Ok(data)
                 }
                 Err(kind) => {
@@ -693,7 +710,9 @@ impl SegmentProcessor for S3PipelineUploader {
                         tracing::debug!(target: "dial9_worker", path = %data.segment.path.display(), "segment already evicted, skipping");
                     } else {
                         self.circuit_breaker.on_failure();
-                        tracing::warn!(target: "dial9_worker", error = %kind, "upload failed");
+                        rate_limited!(Duration::from_secs(60), {
+                            tracing::warn!(target: "dial9_worker", error = %kind, "upload failed");
+                        });
                     }
                     Err(ProcessError { data, kind })
                 }

dial9-tokio-telemetry/src/background_task/sealed.rs

@@ -19,11 +19,13 @@ pub(crate) fn creation_epoch_secs(data: &[u8], path: &Path) -> (u64, bool) {
     match parse_segment_timestamp(data) {
         Ok(ts) => return (ts / 1_000_000_000, true),
         Err(e) => {
-            tracing::warn!(
-                path = %path.display(),
-                error = %e,
-                "failed to parse segment timestamp, falling back to mtime"
-            );
+            crate::rate_limit::rate_limited!(std::time::Duration::from_secs(60), {
+                tracing::warn!(
+                    path = %path.display(),
+                    error = %e,
+                    "failed to parse segment timestamp, falling back to mtime"
+                );
+            });
         }
     }
     let secs = std::fs::metadata(path)

dial9-tokio-telemetry/src/rate_limit.rs

@@ -22,7 +22,7 @@ macro_rules! rate_limited {
         if next <= time.as_secs() {
             let new_next = time
                 .checked_add(interval)
-                .unwrap_or(Duration::MAX)
+                .unwrap_or(std::time::Duration::MAX)
                 .as_secs();
             if NEXT_CALL
                 .compare_exchange(next, new_next, Ordering::Relaxed, Ordering::Relaxed)

dial9-tokio-telemetry/src/telemetry/buffer.rs

@@ -241,10 +241,12 @@ impl Drop for ThreadLocalBuffer {
             if let Some(collector) = self.collector.take() {
                 collector.accept_flush(self.flush());
             } else {
-                tracing::warn!(
-                    "dial9-tokio-telemetry: dropping {} unflushed events (no collector registered on this thread)",
-                    self.event_count
-                );
+                crate::rate_limit::rate_limited!(Duration::from_secs(60), {
+                    tracing::warn!(
+                        "dial9-tokio-telemetry: dropping {} unflushed events (no collector registered on this thread)",
+                        self.event_count
+                    );
+                });
             }
         }
     }

dial9-tokio-telemetry/src/telemetry/recorder/mod.rs

@@ -9,6 +9,7 @@ use event_writer::EventWriter;
 use runtime_context::{make_poll_end, make_poll_start, make_worker_park, make_worker_unpark};
 
 use crate::metrics::{FlushMetrics, Operation, TlDrainMetrics};
+use crate::rate_limit::rate_limited;
 use crate::telemetry::buffer;
 use crate::telemetry::events::RawEvent;
 use crate::telemetry::task_metadata::TaskId;
@@ -66,17 +67,21 @@ fn flush_once(
 
     let dropped = shared.collector.take_dropped_batches();
     if dropped > 0 {
-        tracing::warn!(
-            dropped_batches = dropped,
-            "telemetry flush fell behind, dropped batches"
-        );
+        rate_limited!(Duration::from_secs(60), {
+            tracing::warn!(
+                dropped_batches = dropped,
+                "telemetry flush fell behind, dropped batches"
+            );
+        });
     }
 
     while let Some(batch) = shared.collector.next() {
         if batch.event_count > 0
             && let Err(e) = event_writer.write_encoded_batch(&batch)
         {
-            tracing::warn!("failed to transcode batch: {e}");
+            rate_limited!(Duration::from_secs(60), {
+                tracing::warn!("failed to transcode batch: {e}");
+            });
             shared.enabled.store(false, Ordering::Relaxed);
             return FlushStats {
                 event_count: event_writer.events_written() - events_before,
@@ -86,7 +91,9 @@ fn flush_once(
         }
     }
     if let Err(e) = event_writer.flush() {
-        tracing::warn!("failed to flush trace data: {e}");
+        rate_limited!(Duration::from_secs(60), {
+            tracing::warn!("failed to flush trace data: {e}");
+        });
     }
     FlushStats {
         event_count: event_writer.events_written() - events_before,
@@ -222,9 +229,11 @@ fn attach_runtime(
     ctx.metrics_and_base
         .set((metrics, base))
         .unwrap_or_else(|_| {
-            tracing::warn!(
-                "metrics_and_base already set for runtime context; ignoring duplicate attach"
-            );
+            rate_limited!(Duration::from_secs(60), {
+                tracing::warn!(
+                    "metrics_and_base already set for runtime context; ignoring duplicate attach"
+                );
+            });
         });
 
     shared.contexts.lock().unwrap().push(ctx);
@@ -855,13 +864,17 @@ impl TelemetryCore {
             if let Some(ref config) = cpu_profiling {
                 match crate::telemetry::cpu_profile::CpuProfiler::start(config.clone()) {
                     Ok(sampler) => event_writer.cpu_profiler = Some(sampler),
-                    Err(e) => tracing::warn!("failed to start CPU profiler: {e}"),
+                    Err(e) => rate_limited!(Duration::from_secs(60), {
+                        tracing::warn!("failed to start CPU profiler: {e}");
+                    }),
                 }
             }
             if let Some(sched_cfg) = sched_events {
                 match crate::telemetry::cpu_profile::SchedProfiler::new(sched_cfg) {
                     Ok(sched) => *shared.sched_profiler.lock().unwrap() = Some(sched),
-                    Err(e) => tracing::warn!("failed to start scheduler event profiler: {e}"),
+                    Err(e) => rate_limited!(Duration::from_secs(60), {
+                        tracing::warn!("failed to start scheduler event profiler: {e}");
+                    }),
                 }
             }
         }
@@ -1059,13 +1072,17 @@ fn run_flush_loop(
             // Write final metadata before sealing so single-segment
             // traces contain runtime→worker mappings.
             if let Err(e) = event_writer.write_current_segment_metadata() {
-                tracing::warn!("failed to write final segment metadata: {e}");
+                rate_limited!(Duration::from_secs(60), {
+                    tracing::warn!("failed to write final segment metadata: {e}");
+                });
                 if let Some(g) = flush_guard.as_mut() {
                     g.write_metadata_failed = true;
                 }
             }
             if let Err(e) = event_writer.finalize() {
-                tracing::warn!("failed to finalize trace segment: {e}");
+                rate_limited!(Duration::from_secs(60), {
+                    tracing::warn!("failed to finalize trace segment: {e}");
+                });
                 if let Some(g) = flush_guard.as_mut() {
                     g.finalize_failed = true;
                 }

dial9-tokio-telemetry/src/telemetry/writer.rs

@@ -1,5 +1,6 @@
 use dial9_trace_format::encoder::{Encoder, RawEncoder};
 
+use crate::rate_limit::rate_limited;
 use crate::telemetry::collector::Batch;
 use crate::telemetry::format::SegmentMetadataEvent;
 use std::collections::VecDeque;
@@ -414,16 +415,23 @@ impl RotatingWriter {
                                     && name_str != file_name
                                     && let Err(e2) = fs::remove_file(entry.path())
                                 {
-                                    tracing::warn!(
-                                        "failed to evict old trace segment {}: {e2}",
-                                        entry.path().display()
-                                    );
+                                    rate_limited!(Duration::from_secs(60), {
+                                        tracing::warn!(
+                                            "failed to evict old trace segment {}: {e2}",
+                                            entry.path().display()
+                                        );
+                                    });
                                 }
                             }
                         }
                     }
                     Err(e) => {
-                        tracing::warn!("failed to evict old trace segment {}: {e}", path.display());
+                        rate_limited!(Duration::from_secs(60), {
+                            tracing::warn!(
+                                "failed to evict old trace segment {}: {e}",
+                                path.display()
+                            );
+                        });
                     }
                 }
             }
@@ -478,7 +486,9 @@ impl TraceWriter for RotatingWriter {
 
     fn finalize(&mut self) -> std::io::Result<()> {
         if matches!(self.state, WriterState::Finished) {
-            tracing::warn!("writer is already closed.")
+            rate_limited!(Duration::from_secs(60), {
+                tracing::warn!("writer is already closed.");
+            });
         }
         self.flush()?;
         // Rename .active → .bin for the current segment (if it has .active suffix)
@@ -536,11 +546,13 @@ impl TraceWriter for RotatingWriter {
 impl Drop for RotatingWriter {
     fn drop(&mut self) {
         if self.dropped_events > 0 {
-            tracing::info!(
-                target: "dial9_telemetry",
-                dropped_events = self.dropped_events,
-                "RotatingWriter dropped events after finalization"
-            );
+            rate_limited!(Duration::from_secs(60), {
+                tracing::info!(
+                    target: "dial9_telemetry",
+                    dropped_events = self.dropped_events,
+                    "RotatingWriter dropped events after finalization"
+                );
+            });
         }
     }
 }