Merge branch 'dev' into tninesling/schema-upgrader-tests

Author: tninesling

.changesets/fix_caroline_compute_pool_metrics.md

@@ -0,0 +1,19 @@
+### Add compute pool metrics ([PR #7184](https://github.com/apollographql/router/pull/7184))
+
+The compute job pool is used within the router for compute intensive jobs that should not block the Tokio worker threads.
+When this pool becomes saturated it is difficult for users to see why so that they can take action.
+This change adds new metrics to help users understand how long jobs are waiting to be processed.  
+
+New metrics:
+- `apollo.router.compute_jobs.queue_is_full` - A counter of requests rejected because the queue was full.
+- `apollo.router.compute_jobs.duration` - A histogram of time spent in the compute pipeline by the job, including the queue and query planning.
+  - `job.type`: (`QueryPlanning`, `QueryParsing`, `Introspection`)
+  - `job.outcome`: (`ExecutedOk`, `ExecutedError`, `ChannelError`, `RejectedQueueFull`, `Abandoned`)
+- `apollo.router.compute_jobs.queue.wait.duration` - A histogram of time spent in the compute queue by the job.
+  - `job.type`: (`QueryPlanning`, `QueryParsing`, `Introspection`)
+- `apollo.router.compute_jobs.execution.duration` - A histogram of time spent to execute job (excludes time spent in the queue).
+  - `job.type`: (`QueryPlanning`, `QueryParsing`, `Introspection`)
+- `apollo.router.compute_jobs.active_jobs` - A gauge of the number of compute jobs being processed in parallel.
+  - `job.type`: (`QueryPlanning`, `QueryParsing`, `Introspection`)
+
+By [@carodewig](https://github.com/carodewig) in https://github.com/apollographql/router/pull/7184

Cargo.lock

@@ -1351,7 +1351,7 @@ dependencies = [
  "bitflags 2.6.0",
  "cexpr",
  "clang-sys",
- "itertools 0.10.5",
+ "itertools 0.11.0",
  "lazy_static",
  "lazycell",
  "log",
@@ -1946,9 +1946,9 @@ dependencies = [
 
 [[package]]
 name = "crossbeam-channel"
-version = "0.5.14"
+version = "0.5.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "06ba6d68e24814cb8de6bb986db8222d3a027d15872cabc0d18817bc3c0e4471"
+checksum = "82b8f8f868b36967f9606790d1903570de9ceaf870a7bf9fbbd3016d636a2cb2"
 dependencies = [
  "crossbeam-utils",
 ]
@@ -2382,7 +2382,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "33d852cb9b869c2a9b3df2f71a3074817f01e1844f839a144f5fcef059a4eb5d"
 dependencies = [
  "libc",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
@@ -3993,7 +3993,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "fc2f4eb4bc735547cfed7c0a4922cbd04a4655978c09b54f1f7b228750664c34"
 dependencies = [
  "cfg-if",
- "windows-targets 0.48.5",
+ "windows-targets 0.52.6",
 ]
 
 [[package]]
@@ -5149,7 +5149,7 @@ version = "0.13.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d0f3e5beed80eb580c68e2c600937ac2c4eedabdfd5ef1e5b7ea4f3fba84497b"
 dependencies = [
- "heck 0.4.1",
+ "heck 0.5.0",
  "itertools 0.13.0",
  "log",
  "multimap 0.10.0",
@@ -5170,7 +5170,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "81bddcdb20abf9501610992b6759a4c888aef7d1a7247ef75e2404275ac24af1"
 dependencies = [
  "anyhow",
- "itertools 0.10.5",
+ "itertools 0.11.0",
  "proc-macro2",
  "quote",
  "syn 2.0.90",
@@ -5291,7 +5291,7 @@ dependencies = [
  "once_cell",
  "socket2",
  "tracing",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
@@ -5912,7 +5912,7 @@ dependencies = [
  "errno",
  "libc",
  "linux-raw-sys 0.4.14",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
@@ -5925,7 +5925,7 @@ dependencies = [
  "errno",
  "libc",
  "linux-raw-sys 0.9.3",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
@@ -6474,7 +6474,7 @@ dependencies = [
  "cfg-if",
  "libc",
  "psm",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
@@ -6666,7 +6666,7 @@ dependencies = [
  "getrandom 0.3.1",
  "once_cell",
  "rustix 1.0.3",
- "windows-sys 0.52.0",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]
@@ -7739,7 +7739,7 @@ version = "0.1.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cf221c93e13a30d793f7645a0e7762c55d169dbb0a49671918a2319d289b10bb"
 dependencies = [
- "windows-sys 0.48.0",
+ "windows-sys 0.59.0",
 ]
 
 [[package]]

apollo-router/Cargo.toml

@@ -81,7 +81,7 @@ clap = { version = "4.5.8", default-features = false, features = [
     "help",
 ] }
 cookie = { version = "0.18.0", default-features = false }
-crossbeam-channel = "0.5.14"
+crossbeam-channel = "0.5.15"
 ci_info = { version = "0.14.14", features = ["serde-1"] }
 dashmap = { version = "5.5.3", features = ["serde"] }
 derivative = "2.2.0"

apollo-router/src/compute_job/metrics.rs

@@ -0,0 +1,172 @@
+use std::time::Duration;
+use std::time::Instant;
+
+use crate::compute_job::ComputeJobType;
+
+#[derive(Copy, Clone, strum_macros::IntoStaticStr)]
+pub(super) enum Outcome {
+    ExecutedOk,
+    ExecutedError,
+    ChannelError,
+    RejectedQueueFull,
+    Abandoned,
+}
+
+impl From<Outcome> for opentelemetry::Value {
+    fn from(outcome: Outcome) -> Self {
+        let s: &'static str = outcome.into();
+        s.into()
+    }
+}
+
+pub(super) struct JobWatcher {
+    queue_start: Instant,
+    compute_job_type: ComputeJobType,
+    pub(super) outcome: Outcome,
+}
+
+impl JobWatcher {
+    pub(super) fn new(compute_job_type: ComputeJobType) -> Self {
+        Self {
+            queue_start: Instant::now(),
+            outcome: Outcome::Abandoned,
+            compute_job_type,
+        }
+    }
+}
+
+impl Drop for JobWatcher {
+    fn drop(&mut self) {
+        let full_duration = self.queue_start.elapsed();
+        f64_histogram_with_unit!(
+            "apollo.router.compute_jobs.duration",
+            "Total job processing time",
+            "s",
+            full_duration.as_secs_f64(),
+            "job.type" = self.compute_job_type,
+            "job.outcome" = self.outcome
+        );
+    }
+}
+
+pub(super) struct ActiveComputeMetric {
+    compute_job_type: ComputeJobType,
+}
+
+impl ActiveComputeMetric {
+    // create metric (auto-increments and decrements)
+    pub(super) fn register(compute_job_type: ComputeJobType) -> Self {
+        let s = Self { compute_job_type };
+        s.incr(1);
+        s
+    }
+
+    fn incr(&self, value: i64) {
+        i64_up_down_counter_with_unit!(
+            "apollo.router.compute_jobs.active_jobs",
+            "Number of computation jobs in progress",
+            "{job}",
+            value,
+            job.type = self.compute_job_type
+        );
+    }
+}
+
+impl Drop for ActiveComputeMetric {
+    fn drop(&mut self) {
+        self.incr(-1);
+    }
+}
+
+pub(super) fn observe_queue_wait_duration(
+    compute_job_type: ComputeJobType,
+    queue_duration: Duration,
+) {
+    f64_histogram_with_unit!(
+        "apollo.router.compute_jobs.queue.wait.duration",
+        "Time spent by the job in the compute queue",
+        "s",
+        queue_duration.as_secs_f64(),
+        "job.type" = compute_job_type
+    );
+}
+
+pub(super) fn observe_compute_duration(compute_job_type: ComputeJobType, job_duration: Duration) {
+    f64_histogram_with_unit!(
+        "apollo.router.compute_jobs.execution.duration",
+        "Time to execute the job, after it has been pulled from the queue",
+        "s",
+        job_duration.as_secs_f64(),
+        "job.type" = compute_job_type
+    );
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::compute_job::ComputeJobType;
+    use crate::compute_job::metrics::ActiveComputeMetric;
+    use crate::compute_job::metrics::JobWatcher;
+    use crate::compute_job::metrics::Outcome;
+
+    #[test]
+    fn test_job_watcher() {
+        let check_histogram_count =
+            |count: u64, job_type: &'static str, job_outcome: &'static str| {
+                assert_histogram_count!(
+                    "apollo.router.compute_jobs.duration",
+                    count,
+                    "job.type" = job_type,
+                    "job.outcome" = job_outcome
+                );
+            };
+
+        {
+            let _job_watcher = JobWatcher::new(ComputeJobType::Introspection);
+        }
+        check_histogram_count(1, "Introspection", Outcome::Abandoned.into());
+
+        {
+            let mut job_watcher = JobWatcher::new(ComputeJobType::QueryParsing);
+            job_watcher.outcome = Outcome::ExecutedOk;
+        }
+        check_histogram_count(1, "QueryParsing", Outcome::ExecutedOk.into());
+
+        for count in 1..5 {
+            {
+                let mut job_watcher = JobWatcher::new(ComputeJobType::QueryPlanning);
+                job_watcher.outcome = Outcome::RejectedQueueFull;
+            }
+            check_histogram_count(count, "QueryPlanning", Outcome::RejectedQueueFull.into());
+        }
+    }
+
+    #[test]
+    fn test_active_compute_metric() {
+        let check_count = |count: i64, job_type: &'static str| {
+            assert_up_down_counter!(
+                "apollo.router.compute_jobs.active_jobs",
+                count,
+                "job.type" = job_type
+            );
+        };
+
+        {
+            let _introspection_1 = ActiveComputeMetric::register(ComputeJobType::Introspection);
+            let _introspection_2 = ActiveComputeMetric::register(ComputeJobType::Introspection);
+            let introspection_3 = ActiveComputeMetric::register(ComputeJobType::Introspection);
+            check_count(3, "Introspection");
+
+            let _planning_1 = ActiveComputeMetric::register(ComputeJobType::QueryPlanning);
+            check_count(3, "Introspection");
+            check_count(1, "QueryPlanning");
+
+            drop(introspection_3);
+            check_count(2, "Introspection");
+            check_count(1, "QueryPlanning");
+        }
+
+        // block ended, so should have no ongoing computation
+        check_count(0, "Introspection");
+        check_count(0, "QueryPlanning");
+    }
+}