Collector plugin do not send error to consumer

Summary:
Below has a collector plugin framework to collect extra data async. Currently if the collector fails (timeout, incorrect config, dependency missing etc), an error is returned to the collecting thread, and also stored as a shared value later retrieved by the consumer (main) thread. When the consumer sees the error, the whole sample fails.

This is undesirable because we already log the error in the collecting thread. Below follows the model that most failures result in a None field of the sample instead of failing the whole sample.

This diff updates to log the error in the collecting thread, and report None to the consumer for any error.

Differential Revision: D76748582

fbshipit-source-id: 96b095e215cedb8fbb176e6801a25751f047783b

Author: lnyng

below/model/src/collector.rs

@@ -275,17 +275,8 @@ fn collect_sample(
             if let Some(gpu_stats_receiver) = &options.gpu_stats_receiver {
                 // It is possible to receive no sample if the
                 // collector has not updated since the previous take
-                // or the collector encountered a recoverable error
-                // (e.g. timeout). The behavior for now is to store an
-                // empty map. Alternatively we could store the latest
-                // sample and read that, but then we have to decide how
-                // stale the data can be.
-                Some(
-                    gpu_stats_receiver
-                        .try_take()
-                        .context("GPU stats collector had an error")?
-                        .unwrap_or_default(),
-                )
+                // or the collector encountered errors (e.g. timeout).
+                gpu_stats_receiver.take()
             } else {
                 None
             }
@@ -320,12 +311,7 @@ fn collect_sample(
             }
         },
         tc: if let Some(tc_stats_receiver) = &options.tc_stats_receiver {
-            Some(
-                tc_stats_receiver
-                    .try_take()
-                    .context("TC stats collector had an error")?
-                    .unwrap_or_default(),
-            )
+            tc_stats_receiver.take()
         } else {
             None
         },
@@ -396,13 +382,13 @@ fn collect_cgroup_sample(
         None
     };
     Ok(CgroupSample {
-        cpu_stat: wrap(reader.read_cpu_stat())?.map(Into::into),
+        cpu_stat: wrap(reader.read_cpu_stat())?,
         io_stat,
         tids_current: wrap(reader.read_pids_current())?,
         tids_max: wrap(reader.read_pids_max())?,
         memory_current: wrap(reader.read_memory_current().map(|v| v as i64))?,
-        memory_stat: wrap(reader.read_memory_stat())?.map(Into::into),
-        pressure: pressure_wrap(reader.read_pressure())?.map(Into::into),
+        memory_stat: wrap(reader.read_memory_stat())?,
+        pressure: pressure_wrap(reader.read_pressure())?,
         // We transpose at the end here to convert the
         // Option<Result<BTreeMap... into Result<Option<BTreeMap and
         // then bail any errors with `?` - leaving us with the
@@ -448,17 +434,17 @@ fn collect_cgroup_sample(
         memory_swap_max: wrap(reader.read_memory_swap_max())?,
         memory_zswap_max: wrap(reader.read_memory_zswap_max())?,
         memory_zswap_writeback: wrap(reader.read_memory_zswap_writeback())?,
-        memory_events: wrap(reader.read_memory_events())?.map(Into::into),
-        memory_events_local: wrap(reader.read_memory_events_local())?.map(Into::into),
+        memory_events: wrap(reader.read_memory_events())?,
+        memory_events_local: wrap(reader.read_memory_events_local())?,
         inode_number: match reader.read_inode_number() {
             Ok(st_ino) => Some(st_ino as i64),
             Err(e) => {
                 error!(logger, "Fail to collect inode number: {:#}", e);
                 None
             }
         },
-        cgroup_stat: wrap(reader.read_cgroup_stat())?.map(Into::into),
-        memory_numa_stat: wrap(reader.read_memory_numa_stat())?.map(Into::into),
+        cgroup_stat: wrap(reader.read_cgroup_stat())?,
+        memory_numa_stat: wrap(reader.read_memory_numa_stat())?,
         cpuset_cpus: wrap(reader.read_cpuset_cpus())?,
         cpuset_cpus_effective: wrap(reader.read_cpuset_cpus_effective())?,
         cpuset_mems: wrap(reader.read_cpuset_mems())?,

below/model/src/collector_plugin.rs

@@ -28,15 +28,14 @@ pub trait AsyncCollectorPlugin {
     //
     // On success, this should return `Ok(Some(sample))`.
     //
-    // On a recoverable error, this should return `Ok(None)`. The
+    // On a error handled by the collector, this should return `Ok(None)`. The
     // function itself should consume the error (e.g. log the error)
-    // so that it does not get sent to a consumer thread
     //
-    // On unrecoverable error, this should return `Err(e)`.
+    // On unhandled errors, this should return `Err(e)`.
     async fn try_collect(&mut self) -> Result<Option<Self::T>>;
 }
 
-type SharedVal<T> = Arc<Mutex<Option<Result<T>>>>;
+type SharedVal<T> = Arc<Mutex<Option<T>>>;
 
 // A wrapper around an `AsyncCollectorPlugin` that allows samples to
 // be sent to a `Consumer`.
@@ -50,35 +49,27 @@ impl<T, Plugin: AsyncCollectorPlugin<T = T>> AsyncCollector<T, Plugin> {
         Self { shared, plugin }
     }
 
-    fn update(&self, value: Result<T>) {
-        *self.shared.lock().unwrap() = Some(value);
+    fn update(&self, value: Option<T>) {
+        *self.shared.lock().unwrap() = value;
     }
 
     // Collect sample and update value shared with consumer. Replaces
     // any existing sample that consumer has not consumed yet.
     //
     // Returns true if data was collected and sent. Returns false if
     // there was a recoverable error. Returns an error if there was an
-    // unrecoverable error.
+    // unrecoverable error. Errors are never sent to the consumer.
     pub async fn collect_and_update(&mut self) -> Result<bool> {
-        let collect_result = self
-            .plugin
-            .try_collect()
-            .await
-            .context("Collector failed to read");
-
-        match collect_result {
-            Ok(Some(sample)) => {
-                self.update(Ok(sample));
-                Ok(true)
-            }
-            Ok(None) => Ok(false),
+        let maybe_sample = match self.plugin.try_collect().await {
+            Ok(maybe_sample) => maybe_sample,
             Err(e) => {
-                let error_msg = format!("{:#}", e);
-                self.update(Err(e));
-                Err(anyhow!(error_msg))
+                self.update(None);
+                return Err(e).context("Collector failed to collect");
             }
-        }
+        };
+        let collected = maybe_sample.is_some();
+        self.update(maybe_sample);
+        Ok(collected)
     }
 }
 
@@ -93,13 +84,8 @@ impl<T> Consumer<T> {
     }
 
     // Try to get latest sample of data if it exists.
-    // Returns the error if the collector sent an error.
-    pub fn try_take(&self) -> Result<Option<T>> {
-        match self.shared.lock().unwrap().take() {
-            Some(Ok(v)) => Ok(Some(v)),
-            Some(Err(e)) => Err(e),
-            None => Ok(None),
-        }
+    pub fn take(&self) -> Option<T> {
+        self.shared.lock().unwrap().take()
     }
 }
 
@@ -169,15 +155,15 @@ mod test {
         });
         // Collector overwriting sample
         barrier.wait(); // <-- 1
-        assert_eq!(Some(2), consumer.try_take().unwrap());
+        assert_eq!(Some(2), consumer.take());
         barrier.wait(); // <-- 2
         // Collector sending None
         barrier.wait(); // <-- 3
-        assert_eq!(None, consumer.try_take().unwrap());
+        assert!(consumer.take().is_none());
         barrier.wait(); // <-- 4
         // Collector sending error
         barrier.wait(); // <-- 5
-        assert!(consumer.try_take().is_err());
+        assert!(consumer.take().is_none());
 
         handle.join().unwrap();
     }