plugins/fleet_detector: add schema and persisted_queries gauges

.changesets/feat_feat_fleet_detector_add_schema_metrics.md

@@ -0,0 +1,6 @@
+### Adds Fleet Awareness Schema Metrics
+
+Adds an additional metric to the `fleet_detector` plugin in the form of `apollo.router.schema` with 2 attributes:
+`schema_hash` and `launch_id`.
+
+By [@loshz](https://github.com/loshz) and [@nmoutschen](https://github.com/nmoutschen) in https://github.com/apollographql/router/pull/6283

apollo-router/src/plugin/mod.rs

@@ -77,6 +77,9 @@ pub struct PluginInit<T> {
     /// The parsed subgraph schemas from the query planner, keyed by subgraph name
     pub(crate) subgraph_schemas: Arc<SubgraphSchemas>,
 
+    /// Launch ID
+    pub(crate) launch_id: Option<Arc<String>>,
+
     pub(crate) notify: Notify<String, graphql::Response>,
 }
 
@@ -137,6 +140,7 @@ where
             .supergraph_schema_id(crate::spec::Schema::schema_id(&supergraph_sdl).into())
             .supergraph_sdl(supergraph_sdl)
             .supergraph_schema(supergraph_schema)
+            .launch_id(Arc::new("launch_id".to_string()))
             .notify(Notify::for_tests())
             .build()
     }
@@ -173,6 +177,7 @@ where
         supergraph_schema_id: Arc<String>,
         supergraph_schema: Arc<Valid<Schema>>,
         subgraph_schemas: Option<Arc<SubgraphSchemas>>,
+        launch_id: Option<Option<Arc<String>>>,
         notify: Notify<String, graphql::Response>,
     ) -> Self {
         PluginInit {
@@ -181,6 +186,7 @@ where
             supergraph_schema_id,
             supergraph_schema,
             subgraph_schemas: subgraph_schemas.unwrap_or_default(),
+            launch_id: launch_id.flatten(),
             notify,
         }
     }
@@ -196,6 +202,7 @@ where
         supergraph_schema_id: Arc<String>,
         supergraph_schema: Arc<Valid<Schema>>,
         subgraph_schemas: Option<Arc<SubgraphSchemas>>,
+        launch_id: Option<Arc<String>>,
         notify: Notify<String, graphql::Response>,
     ) -> Result<Self, BoxError> {
         let config: T = serde_json::from_value(config)?;
@@ -205,6 +212,7 @@ where
             supergraph_schema,
             supergraph_schema_id,
             subgraph_schemas: subgraph_schemas.unwrap_or_default(),
+            launch_id,
             notify,
         })
     }
@@ -217,6 +225,7 @@ where
         supergraph_schema_id: Option<Arc<String>>,
         supergraph_schema: Option<Arc<Valid<Schema>>>,
         subgraph_schemas: Option<Arc<SubgraphSchemas>>,
+        launch_id: Option<Arc<String>>,
         notify: Option<Notify<String, graphql::Response>>,
     ) -> Self {
         PluginInit {
@@ -226,6 +235,7 @@ where
             supergraph_schema: supergraph_schema
                 .unwrap_or_else(|| Arc::new(Valid::assume_valid(Schema::new()))),
             subgraph_schemas: subgraph_schemas.unwrap_or_default(),
+            launch_id,
             notify: notify.unwrap_or_else(Notify::for_tests),
         }
     }

apollo-router/src/plugins/fleet_detector.rs

@@ -45,14 +45,12 @@ impl SystemGetter {
     }
 
     fn get_system(&mut self) -> &System {
-        if self.start.elapsed() < REFRESH_INTERVAL {
-            &self.system
-        } else {
+        if self.start.elapsed() >= REFRESH_INTERVAL {
             self.start = Instant::now();
             self.system.refresh_cpu_all();
             self.system.refresh_memory();
-            &self.system
         }
+        &self.system
     }
 }
 
@@ -65,7 +63,7 @@ enum GaugeStore {
 }
 
 impl GaugeStore {
-    fn active() -> GaugeStore {
+    fn active(opts: &GaugeOptions) -> GaugeStore {
         let system_getter = Arc::new(Mutex::new(SystemGetter::new()));
         let meter = meter_provider().meter("apollo/router");
 
@@ -112,14 +110,14 @@ impl GaugeStore {
                         "The CPU frequency of the underlying instance the router is deployed to",
                     )
                     .with_unit(Unit::new("Mhz"))
-                    .with_callback(move |i| {
+                    .with_callback(move |gauge| {
                         let local_system_getter = system_getter.clone();
                         let mut system_getter = local_system_getter.lock().unwrap();
                         let system = system_getter.get_system();
                         let cpus = system.cpus();
                         let cpu_freq =
                             cpus.iter().map(|cpu| cpu.frequency()).sum::<u64>() / cpus.len() as u64;
-                        i.observe(cpu_freq, &[])
+                        gauge.observe(cpu_freq, &[])
                     })
                     .init(),
             );
@@ -133,12 +131,12 @@ impl GaugeStore {
                     .with_description(
                         "The number of CPUs reported by the instance the router is running on",
                     )
-                    .with_callback(move |i| {
+                    .with_callback(move |gauge| {
                         let local_system_getter = system_getter.clone();
                         let mut system_getter = local_system_getter.lock().unwrap();
                         let system = system_getter.get_system();
                         let cpu_count = detect_cpu_count(system);
-                        i.observe(cpu_count, &[KeyValue::new("host.arch", get_otel_arch())])
+                        gauge.observe(cpu_count, &[KeyValue::new("host.arch", get_otel_arch())])
                     })
                     .init(),
             );
@@ -152,11 +150,11 @@ impl GaugeStore {
                     .with_description(
                         "The amount of memory reported by the instance the router is running on",
                     )
-                    .with_callback(move |i| {
+                    .with_callback(move |gauge| {
                         let local_system_getter = system_getter.clone();
                         let mut system_getter = local_system_getter.lock().unwrap();
                         let system = system_getter.get_system();
-                        i.observe(
+                        gauge.observe(
                             system.total_memory(),
                             &[KeyValue::new("host.arch", get_otel_arch())],
                         )
@@ -165,19 +163,50 @@ impl GaugeStore {
                     .init(),
             );
         }
+        {
+            let opts = opts.clone();
+            gauges.push(
+                meter
+                    .u64_observable_gauge("apollo.router.instance.schema")
+                    .with_description("Details about the current in-use schema")
+                    .with_callback(move |gauge| {
+                        // NOTE: this is a fixed gauge. We only care about observing the included
+                        // attributes.
+                        let mut attributes: Vec<KeyValue> = vec![KeyValue::new(
+                            "schema_hash",
+                            opts.supergraph_schema_hash.clone(),
+                        )];
+                        if let Some(launch_id) = opts.launch_id.as_ref() {
+                            attributes.push(KeyValue::new("launch_id", launch_id.to_string()));
+                        }
+                        gauge.observe(1, attributes.as_slice())
+                    })
+                    .init(),
+            )
+        }
         GaugeStore::Active(gauges)
     }
 }
 
+#[derive(Clone, Default)]
+struct GaugeOptions {
+    supergraph_schema_hash: String,
+    launch_id: Option<String>,
+}
+
 #[derive(Default)]
 struct FleetDetector {
     gauge_store: Mutex<GaugeStore>,
+
+    // Options passed to the gauge_store during activation.
+    gauge_options: GaugeOptions,
 }
+
 #[async_trait::async_trait]
 impl PluginPrivate for FleetDetector {
     type Config = Conf;
 
-    async fn new(_: PluginInit<Self::Config>) -> Result<Self, BoxError> {
+    async fn new(plugin: PluginInit<Self::Config>) -> Result<Self, BoxError> {
         debug!("initialising fleet detection plugin");
         if let Ok(val) = env::var(APOLLO_TELEMETRY_DISABLED) {
             if val == "true" {
@@ -186,15 +215,21 @@ impl PluginPrivate for FleetDetector {
             }
         }
 
+        let gauge_options = GaugeOptions {
+            supergraph_schema_hash: plugin.supergraph_schema_id.to_string(),
+            launch_id: plugin.launch_id.map(|s| s.to_string()),
+        };
+
         Ok(FleetDetector {
             gauge_store: Mutex::new(GaugeStore::Pending),
+            gauge_options,
         })
     }
 
     fn activate(&self) {
         let mut store = self.gauge_store.lock().expect("lock poisoned");
         if matches!(*store, GaugeStore::Pending) {
-            *store = GaugeStore::active();
+            *store = GaugeStore::active(&self.gauge_options);
         }
     }
 }

apollo-router/src/router/event/mod.rs

@@ -22,6 +22,7 @@ use self::Event::UpdateConfiguration;
 use self::Event::UpdateLicense;
 use self::Event::UpdateSchema;
 use crate::uplink::license_enforcement::LicenseState;
+use crate::uplink::schema::SchemaState;
 use crate::Configuration;
 
 /// Messages that are broadcast across the app.
@@ -33,7 +34,7 @@ pub(crate) enum Event {
     NoMoreConfiguration,
 
     /// The schema was updated.
-    UpdateSchema(String),
+    UpdateSchema(SchemaState),
 
     /// There are no more updates to the schema
     NoMoreSchema,

apollo-router/src/router/event/schema.rs

@@ -11,6 +11,7 @@ use url::Url;
 use crate::router::Event;
 use crate::router::Event::NoMoreSchema;
 use crate::router::Event::UpdateSchema;
+use crate::uplink::schema::SchemaState;
 use crate::uplink::schema_stream::SupergraphSdlQuery;
 use crate::uplink::stream_from_uplink;
 use crate::uplink::UplinkConfig;
@@ -74,9 +75,20 @@ impl SchemaSource {
     pub(crate) fn into_stream(self) -> impl Stream<Item = Event> {
         match self {
             SchemaSource::Static { schema_sdl: schema } => {
-                stream::once(future::ready(UpdateSchema(schema))).boxed()
+                let update_schema = UpdateSchema(SchemaState {
+                    sdl: schema,
+                    launch_id: None,
+                });
+                stream::once(future::ready(update_schema)).boxed()
             }
-            SchemaSource::Stream(stream) => stream.map(UpdateSchema).boxed(),
+            SchemaSource::Stream(stream) => stream
+                .map(|sdl| {
+                    UpdateSchema(SchemaState {
+                        sdl,
+                        launch_id: None,
+                    })
+                })
+                .boxed(),
             #[allow(deprecated)]
             SchemaSource::File {
                 path,
@@ -100,7 +112,13 @@ impl SchemaSource {
                                         let path = path.clone();
                                         async move {
                                             match tokio::fs::read_to_string(&path).await {
-                                                Ok(schema) => Some(UpdateSchema(schema)),
+                                                Ok(schema) => {
+                                                    let update_schema = UpdateSchema(SchemaState {
+                                                        sdl: schema,
+                                                        launch_id: None,
+                                                    });
+                                                    Some(update_schema)
+                                                }
                                                 Err(err) => {
                                                     tracing::error!(reason = %err, "failed to read supergraph schema");
                                                     None
@@ -110,7 +128,11 @@ impl SchemaSource {
                                     })
                                     .boxed()
                             } else {
-                                stream::once(future::ready(UpdateSchema(schema))).boxed()
+                                let update_schema = UpdateSchema(SchemaState {
+                                    sdl: schema,
+                                    launch_id: None,
+                                });
+                                stream::once(future::ready(update_schema)).boxed()
                             }
                         }
                         Err(err) => {
@@ -121,10 +143,13 @@ impl SchemaSource {
                 }
             }
             SchemaSource::Registry(uplink_config) => {
-                stream_from_uplink::<SupergraphSdlQuery, String>(uplink_config)
+                stream_from_uplink::<SupergraphSdlQuery, SchemaState>(uplink_config)
                     .filter_map(|res| {
                         future::ready(match res {
-                            Ok(schema) => Some(UpdateSchema(schema)),
+                            Ok(schema) => {
+                                let update_schema = UpdateSchema(schema);
+                                Some(update_schema)
+                            }
                             Err(e) => {
                                 tracing::error!("{}", e);
                                 None
@@ -222,7 +247,13 @@ impl Fetcher {
                 .await
             {
                 Ok(res) if res.status().is_success() => match res.text().await {
-                    Ok(schema) => return Some(UpdateSchema(schema)),
+                    Ok(schema) => {
+                        let update_schema = UpdateSchema(SchemaState {
+                            sdl: schema,
+                            launch_id: None,
+                        });
+                        return Some(update_schema);
+                    }
                     Err(err) => {
                         tracing::warn!(
                             url.full = %url,
@@ -346,10 +377,10 @@ mod tests {
             .into_stream();
 
             assert!(
-                matches!(stream.next().await.unwrap(), UpdateSchema(schema) if schema == SCHEMA_1)
+                matches!(stream.next().await.unwrap(), UpdateSchema(schema) if schema.sdl == SCHEMA_1)
             );
             assert!(
-                matches!(stream.next().await.unwrap(), UpdateSchema(schema) if schema == SCHEMA_1)
+                matches!(stream.next().await.unwrap(), UpdateSchema(schema) if schema.sdl == SCHEMA_1)
             );
         }
         .with_subscriber(assert_snapshot_subscriber!())
@@ -382,10 +413,10 @@ mod tests {
             .into_stream();
 
             assert!(
-                matches!(stream.next().await.unwrap(), UpdateSchema(schema) if schema == SCHEMA_2)
+                matches!(stream.next().await.unwrap(), UpdateSchema(schema) if schema.sdl == SCHEMA_2)
             );
             assert!(
-                matches!(stream.next().await.unwrap(), UpdateSchema(schema) if schema == SCHEMA_2)
+                matches!(stream.next().await.unwrap(), UpdateSchema(schema) if schema.sdl == SCHEMA_2)
             );
         }
         .with_subscriber(assert_snapshot_subscriber!({
@@ -448,7 +479,7 @@ mod tests {
                 .await;
 
             assert!(
-                matches!(stream.next().await.unwrap(), UpdateSchema(schema) if schema == SCHEMA_1)
+                matches!(stream.next().await.unwrap(), UpdateSchema(schema) if schema.sdl == SCHEMA_1)
             );
 
             drop(success);
@@ -468,7 +499,7 @@ mod tests {
                 .await;
 
             assert!(
-                matches!(stream.next().await.unwrap(), UpdateSchema(schema) if schema == SCHEMA_1)
+                matches!(stream.next().await.unwrap(), UpdateSchema(schema) if schema.sdl == SCHEMA_1)
             );
         }
         .with_subscriber(assert_snapshot_subscriber!({
@@ -497,7 +528,7 @@ mod tests {
             .into_stream();
 
             assert!(
-                matches!(stream.next().await.unwrap(), UpdateSchema(schema) if schema == SCHEMA_1)
+                matches!(stream.next().await.unwrap(), UpdateSchema(schema) if schema.sdl == SCHEMA_1)
             );
             assert!(matches!(stream.next().await.unwrap(), NoMoreSchema));
         }