record metrics in control thread only

Author: lherman-cs

pulsebeam/src/rtp/sync.rs

@@ -35,12 +35,10 @@ pub struct Synchronizer {
     /// The server Instant that corresponds to a specific NTP time, representing
     /// the minimum observed propagation delay.
     ntp_anchor: Option<ClockReference>,
-    ppm_hist: metrics::Histogram,
 }
 
 impl Synchronizer {
     pub fn new(clock_rate: Frequency) -> Self {
-        let ppm_hist = metrics::histogram!("rtp_sync_clock_drift_ppm");
         Self {
             clock_rate,
             first_sr: None,
@@ -50,7 +48,6 @@ impl Synchronizer {
             base_server_time: None,
             estimated_clock_drift_ppm: 0.0,
             ntp_anchor: None,
-            ppm_hist,
         }
     }
 
@@ -175,7 +172,6 @@ impl Synchronizer {
 
         if let (Some(first), Some(latest)) = (self.first_sr, self.latest_sr) {
             self.estimated_clock_drift_ppm = Self::compute_clock_drift(&first, &latest);
-            self.ppm_hist.record(self.estimated_clock_drift_ppm);
         }
 
         // Update the NTP anchor with a minimum envelope filter

pulsebeam/src/shard/worker.rs

@@ -152,11 +152,6 @@ pub struct ShardWorker {
     cross_shard_event_rx: mailbox::Receiver<CrossShardEvent>,
     router: ShardRouter,
     metrics: Arc<ShardMetrics>,
-
-    delay_ingress: metrics::Histogram,
-    delay_compute: metrics::Histogram,
-    delay_egress: metrics::Histogram,
-    delay_tick: metrics::Histogram,
 }
 
 impl ShardWorker {
@@ -177,15 +172,6 @@ impl ShardWorker {
             cross_shard_event_txs,
         };
 
-        metrics::describe_histogram!(
-            "shard_tick_delay_us",
-            Unit::Microseconds,
-            "shard tick delay distribution"
-        );
-        let delay_tick = metrics::histogram!("shard_tick_delay_us",);
-        let delay_ingress = metrics::histogram!("shard_ingress_delay_us",);
-        let delay_compute = metrics::histogram!("shard_compute_delay_us",);
-        let delay_egress = metrics::histogram!("shard_egress_delay_us",);
         Self {
             core,
             recv_batch: Vec::with_capacity(net::BATCH_SIZE),
@@ -196,10 +182,6 @@ impl ShardWorker {
             cross_shard_event_rx,
             router,
             metrics,
-            delay_tick,
-            delay_ingress,
-            delay_compute,
-            delay_egress,
         }
     }
 
@@ -225,7 +207,6 @@ impl ShardWorker {
             loop_start = busy_end;
             let busy_duration = busy_end.duration_since(busy_start);
             self.metrics.record_busy(busy_duration);
-            self.delay_tick.record(busy_duration.as_micros() as f64);
         }
     }
 
@@ -279,26 +260,19 @@ impl ShardWorker {
         for batch in self.recv_batch.drain(..) {
             self.core.on_udp_batch(batch, &self.router);
         }
-        let ingress_delay = now.elapsed();
-        self.delay_ingress.record(ingress_delay.as_micros() as f64);
 
         // phase 2: compute
         let now = Instant::now();
         self.core.poll_input(now);
         self.core.flush_rtp_events(&self.router);
         self.core.poll_fanout(now);
         self.core.flush_participant_events(&self.router);
-        let compute_delay = now.elapsed();
-        self.delay_compute.record(compute_delay.as_micros() as f64);
 
         // phase 3: output
-        let now = Instant::now();
         self.core
             .flush_egress(&self.udp_socket, &mut self.tcp_socket);
-        let egress_delay = now.elapsed();
         self.core
             .flush_close_peers(&mut self.udp_socket, &mut self.tcp_socket);
-        self.delay_egress.record(egress_delay.as_micros() as f64);
     }
 
     async fn flush_shard_events(&mut self) -> Result<(), ShardError> {