fix(tracing): Flush spans on interval instead of idle timeout

Our trace span exporter uses batching logic to avoid making excessive requests to the trace collector. It does this by either waiting for the batch buffer to hit a maximum number of spans (currently 1000), or waiting for no spans to be collected in some timeout window (currently 10s).

If a workload emits spans at a slow but consistent rate, say 1 span/s, it would cause the timeout window to never be hit while the buffer slowly fills. At this rate, it would cause spans to not be emitted for a long time.

This replaces the timeout window with an interval timer that attempts to send the currently buffered spans on a consistent interval, regardless of if the collector has sat idle. This interval is set to 10s to match the current idle timeout, but this can be tuned in the future along with the buffer size if we feel the need.

Signed-off-by: Scott Fleener <[email protected]>

Author: sfleen

linkerd/opentelemetry/src/lib.rs

@@ -28,7 +28,10 @@ use opentelemetry_proto::{
 };
 use opentelemetry_sdk::trace::{SpanData, SpanLinks};
 pub use opentelemetry_sdk::{self as sdk};
-use tokio::{sync::mpsc, time};
+use tokio::{
+    sync::mpsc,
+    time::{self, Instant, MissedTickBehavior},
+};
 use tonic::{self as grpc, body::Body as TonicBody, client::GrpcService};
 use tracing::{debug, info, trace};
 
@@ -72,7 +75,7 @@ where
     S: Stream<Item = ExportSpan> + Unpin,
 {
     const MAX_BATCH_SIZE: usize = 1000;
-    const MAX_BATCH_IDLE: time::Duration = time::Duration::from_secs(10);
+    const BATCH_INTERVAL: time::Duration = time::Duration::from_secs(10);
 
     fn new(client: T, spans: S, resource: ResourceAttributesWithSchema, metrics: Registry) -> Self {
         Self {
@@ -191,6 +194,10 @@ where
     ) -> Result<(), SpanRxClosed> {
         let mut input_accum: Vec<SpanData> = vec![];
 
+        let mut interval =
+            time::interval_at(Instant::now() + Self::BATCH_INTERVAL, Self::BATCH_INTERVAL);
+        interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
+
         let res = loop {
             if input_accum.len() == Self::MAX_BATCH_SIZE {
                 trace!(capacity = Self::MAX_BATCH_SIZE, "Batch capacity reached");
@@ -216,11 +223,11 @@ where
                     None => break Err(SpanRxClosed),
                 },
 
-                // Don't hold spans indefinitely. Return if we hit an idle
-                // timeout and spans have been collected.
-                _ = time::sleep(Self::MAX_BATCH_IDLE) => {
+                // Don't hold spans indefinitely. Return if we hit an interval tick and spans have
+                // been collected.
+                _ = interval.tick() => {
                     if !input_accum.is_empty() {
-                        trace!(spans = input_accum.len(), "Flushing spans due to inactivitiy");
+                        trace!(spans = input_accum.len(), "Flushing spans due to interval tick");
                         break Ok(());
                     }
                 }