feat(msd): metrics

rs/ic-observability/config-writer-common/src/config_writer.rs

@@ -22,11 +22,7 @@ pub struct ConfigWriter {
 }
 
 impl ConfigWriter {
-    pub fn new<P: AsRef<Path>>(
-        write_path: P,
-        filters: Arc<dyn TargetGroupFilter>,
-        log: Logger,
-    ) -> Self {
+    pub fn new<P: AsRef<Path>>(write_path: P, filters: Arc<dyn TargetGroupFilter>, log: Logger) -> Self {
         ConfigWriter {
             base_directory: PathBuf::from(write_path.as_ref()),
             last_targets: Default::default(),
@@ -48,16 +44,10 @@ impl ConfigWriter {
     ) -> std::io::Result<()> {
         let last_job_targets = self.last_targets.entry(job.to_string()).or_default();
         if last_job_targets == &target_groups {
-            debug!(
-                self.log,
-                "Targets didn't change, skipped regenerating config"
-            );
+            debug!(self.log, "Targets didn't change, skipped regenerating config");
             return Ok(());
         }
-        debug!(
-            self.log,
-            "Targets changed, proceeding with regenerating config"
-        );
+        debug!(self.log, "Targets changed, proceeding with regenerating config");
         let target_path = self.base_directory.join(format!("{}.json", job));
 
         let filtered_target_groups: BTreeSet<TargetGroup> = target_groups
@@ -69,12 +59,8 @@ impl ConfigWriter {
         let vector_config = vector_config_builder.build(filtered_target_groups, job);
 
         ic_utils::fs::write_atomically(target_path.as_path(), |f| {
-            serde_json::to_writer_pretty(f, &vector_config).map_err(|e| {
-                std::io::Error::new(
-                    std::io::ErrorKind::Other,
-                    format!("Serialization error: {:?}", e),
-                )
-            })
+            serde_json::to_writer_pretty(f, &vector_config)
+                .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, format!("Serialization error: {:?}", e)))
         })?;
         self.last_targets.insert(job.to_string(), target_groups);
         Ok(())
@@ -84,25 +70,15 @@ impl ConfigWriter {
 impl ConfigUpdater for ConfigWriter {
     fn update(&self, config: &dyn Config) -> Result<(), Box<dyn Error>> {
         if !config.updated() {
-            debug!(
-                self.log,
-                "Targets didn't change, skipped regenerating config"
-            );
+            debug!(self.log, "Targets didn't change, skipped regenerating config");
             return Ok(());
         }
-        debug!(
-            self.log,
-            "Targets changed, proceeding with regenerating config"
-        );
+        debug!(self.log, "Targets changed, proceeding with regenerating config");
         let target_path = self.base_directory.join(format!("{}.json", config.name()));
 
         ic_utils::fs::write_atomically(target_path.as_path(), |f| {
-            serde_json::to_writer_pretty(f, &config).map_err(|e| {
-                std::io::Error::new(
-                    std::io::ErrorKind::Other,
-                    format!("Serialization error: {:?}", e),
-                )
-            })
+            serde_json::to_writer_pretty(f, &config)
+                .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, format!("Serialization error: {:?}", e)))
         })?;
         Ok(())
     }

rs/ic-observability/config-writer-common/src/config_writer_loop.rs

@@ -26,8 +26,7 @@ pub fn config_writer_loop(
     metrics: Metrics,
 ) -> impl FnMut() {
     move || {
-        let mut config_writer =
-            ConfigWriter::new(vector_config_dir.clone(), filters.clone(), log.clone());
+        let mut config_writer = ConfigWriter::new(vector_config_dir.clone(), filters.clone(), log.clone());
         loop {
             for job in &jobs {
                 let targets = match discovery.get_target_groups(*job, log.clone()) {
@@ -42,10 +41,7 @@ pub fn config_writer_loop(
                     .with_label_values(&[job.to_string().as_str()])
                     .set(targets.len().try_into().unwrap());
                 if let Err(e) = config_writer.write_config(*job, targets, &vector_config_builder) {
-                    warn!(
-                        log,
-                        "Failed to write config for targets for job {}: {:?}", job, e
-                    );
+                    warn!(log, "Failed to write config for targets for job {}: {:?}", job, e);
                 };
             }
             select! {

rs/ic-observability/config-writer-common/src/filters.rs

@@ -62,9 +62,7 @@ mod tests {
 
     fn create_dummy_target_group(ipv6: &str) -> TargetGroup {
         let mut targets = BTreeSet::new();
-        targets.insert(std::net::SocketAddr::V6(
-            SocketAddrV6::from_str(ipv6).unwrap(),
-        ));
+        targets.insert(std::net::SocketAddr::V6(SocketAddrV6::from_str(ipv6).unwrap()));
         TargetGroup {
             node_id: NodeId::from(PrincipalId::new_anonymous()),
             ic_name: "mercury".into(),
@@ -82,10 +80,7 @@ mod tests {
 
         let accepted_tg = TargetGroup {
             node_id: NodeId::from(
-                PrincipalId::from_str(
-                    "iylgr-zpxwq-kqgmf-4srtx-o4eey-d6bln-smmq6-we7px-ibdea-nondy-eae",
-                )
-                .unwrap(),
+                PrincipalId::from_str("iylgr-zpxwq-kqgmf-4srtx-o4eey-d6bln-smmq6-we7px-ibdea-nondy-eae").unwrap(),
             ),
             ic_name: "mercury".into(),
             targets: BTreeSet::new(),
@@ -98,10 +93,7 @@ mod tests {
 
         let rejected_tg = TargetGroup {
             node_id: NodeId::from(
-                PrincipalId::from_str(
-                    "x33ed-h457x-bsgyx-oqxqf-6pzwv-wkhzr-rm2j3-npodi-purzm-n66cg-gae",
-                )
-                .unwrap(),
+                PrincipalId::from_str("x33ed-h457x-bsgyx-oqxqf-6pzwv-wkhzr-rm2j3-npodi-purzm-n66cg-gae").unwrap(),
             ),
             ic_name: "mercury".into(),
             targets: BTreeSet::new(),
@@ -121,10 +113,7 @@ mod tests {
 
         let accepted_tg = TargetGroup {
             node_id: NodeId::from(
-                PrincipalId::from_str(
-                    "iylgr-zpxwq-kqgmf-4srtx-o4eey-d6bln-smmq6-we7px-ibdea-nondy-eae",
-                )
-                .unwrap(),
+                PrincipalId::from_str("iylgr-zpxwq-kqgmf-4srtx-o4eey-d6bln-smmq6-we7px-ibdea-nondy-eae").unwrap(),
             ),
             ic_name: "mercury".into(),
             targets: BTreeSet::new(),
@@ -137,10 +126,7 @@ mod tests {
 
         let rejected_tg_1 = TargetGroup {
             node_id: NodeId::from(
-                PrincipalId::from_str(
-                    "x33ed-h457x-bsgyx-oqxqf-6pzwv-wkhzr-rm2j3-npodi-purzm-n66cg-gae",
-                )
-                .unwrap(),
+                PrincipalId::from_str("x33ed-h457x-bsgyx-oqxqf-6pzwv-wkhzr-rm2j3-npodi-purzm-n66cg-gae").unwrap(),
             ),
             ic_name: "mercury".into(),
             targets: BTreeSet::new(),

rs/ic-observability/multiservice-discovery-downloader/src/main.rs

@@ -21,11 +21,7 @@ fn main() {
 
     info!(logger, "Starting downloader loop"; "cli_args" => ?cli_args);
 
-    let downloader_handle = rt.spawn(run_downloader_loop(
-        logger.clone(),
-        cli_args,
-        stop_signal_rcv,
-    ));
+    let downloader_handle = rt.spawn(run_downloader_loop(logger.clone(), cli_args, stop_signal_rcv));
 
     rt.block_on(shutdown_signal);
     info!(logger, "Received shutdown signal, shutting down ...");
@@ -127,11 +123,7 @@ pub mod log_subtype {
     use super::*;
     #[derive(Parser, Clone, Debug)]
     pub struct LogSubtype {
-        #[clap(
-            long = "port",
-            help = "Custom port for standard nodes",
-            default_value = "19531"
-        )]
+        #[clap(long = "port", help = "Custom port for standard nodes", default_value = "19531")]
         pub port: u64,
         #[clap(
             long = "boundary-nodes-port",
@@ -158,10 +150,7 @@ pub mod log_subtype {
             #[clap(long = "journals-folder", help = "Path to the root journals folder")]
             journals_folder: String,
 
-            #[clap(
-                long = "worker-cursor-folder",
-                help = "Path for the root worker cursors folder"
-            )]
+            #[clap(long = "worker-cursor-folder", help = "Path for the root worker cursors folder")]
             worker_cursor_folder: String,
 
             #[clap(long = "data-folder", help = "Path for the data folder")]

rs/ic-observability/multiservice-discovery-shared/src/filters/mod.rs

@@ -31,10 +31,7 @@ impl TargetGroupFilter for TargetGroupFilterList {
         if self.filters.is_empty() {
             true
         } else {
-            self.filters
-                .iter()
-                .map(|f| f.filter(target_group))
-                .all(|status| status)
+            self.filters.iter().map(|f| f.filter(target_group)).all(|status| status)
         }
     }
 }

rs/ic-observability/multiservice-discovery/Cargo.toml

@@ -31,6 +31,8 @@ tokio = { workspace = true }
 url = { workspace = true }
 futures.workspace = true
 axum = "0.7.4"
+axum-otel-metrics = "0.8.0"
+opentelemetry = { version = "0.21.0", features = ["metrics"] }
 retry = { workspace = true }
 
 [dev-dependencies]