metrics: Report database dump size and duration

Successful tar.gz and zip uploads now record archive size and upload duration through WorkerMetrics with the existing format attribute. The legacy direct Datadog submission remains active when configured during the migration.

Author: Turbo87

src/metrics/consts.rs

@@ -19,6 +19,15 @@ pub const DB_CLIENT_CONNECTION_POOL_NAME: &str = "db.client.connection.pool.name
 /// The database connection state attribute key.
 pub const DB_CLIENT_CONNECTION_STATE: &str = "db.client.connection.state";
 
+/// The database dump size instrument name.
+pub const DB_DUMP_SIZE_BYTES: &str = "crates_io.db_dump_size_bytes";
+
+/// The database dump upload duration instrument name.
+pub const DB_DUMP_UPLOAD_DURATION_NS: &str = "crates_io.db_dump_upload_duration_ns";
+
+/// The archive format attribute key.
+pub const FORMAT: &str = "format";
+
 /// The HTTP request method attribute key.
 pub const HTTP_REQUEST_METHOD: &str = "http.request.method";
 

src/metrics/worker.rs

@@ -1,8 +1,12 @@
 use super::SharedMetrics;
-use super::consts::{BACKGROUND_JOBS, CRATES_TOTAL, JOB, PRIORITY, VERSIONS_TOTAL};
+use super::consts::{
+    BACKGROUND_JOBS, CRATES_TOTAL, DB_DUMP_SIZE_BYTES, DB_DUMP_UPLOAD_DURATION_NS, FORMAT, JOB,
+    PRIORITY, VERSIONS_TOTAL,
+};
 use super::otel::kv;
 use derive_more::Deref;
 use opentelemetry::metrics::{Gauge, Meter};
+use std::time::Duration;
 
 /// OpenTelemetry instruments recorded by the background worker.
 #[derive(Clone, Debug, Deref)]
@@ -13,6 +17,8 @@ pub struct WorkerMetrics {
     background_jobs: Gauge<i64>,
     crates_total: Gauge<i64>,
     versions_total: Gauge<i64>,
+    db_dump_size: Gauge<u64>,
+    db_dump_upload_duration: Gauge<u64>,
 }
 
 impl WorkerMetrics {
@@ -22,12 +28,16 @@ impl WorkerMetrics {
         let background_jobs = meter.i64_gauge(BACKGROUND_JOBS).build();
         let crates_total = meter.i64_gauge(CRATES_TOTAL).build();
         let versions_total = meter.i64_gauge(VERSIONS_TOTAL).build();
+        let db_dump_size = meter.u64_gauge(DB_DUMP_SIZE_BYTES).build();
+        let db_dump_upload_duration = meter.u64_gauge(DB_DUMP_UPLOAD_DURATION_NS).build();
 
         Self {
             shared,
             background_jobs,
             crates_total,
             versions_total,
+            db_dump_size,
+            db_dump_upload_duration,
         }
     }
 
@@ -43,4 +53,13 @@ impl WorkerMetrics {
         let attributes = [kv(PRIORITY, priority.clone()), kv(JOB, job.clone())];
         self.background_jobs.record(count, &attributes);
     }
+
+    /// Records the size and upload duration of a database dump archive.
+    pub fn record_db_dump(&self, format: &'static str, size: u64, upload_duration: Duration) {
+        let attributes = [kv(FORMAT, format)];
+        let upload_ns = u64::try_from(upload_duration.as_nanos()).unwrap_or(u64::MAX);
+
+        self.db_dump_size.record(size, &attributes);
+        self.db_dump_upload_duration.record(upload_ns, &attributes);
+    }
 }

src/tests/dump_db.rs

@@ -1,18 +1,31 @@
 use crate::builders::CrateBuilder;
 use crate::util::TestApp;
 use bytes::Buf;
+use claims::assert_gt;
+use crates_io::metrics::consts::{
+    DB_DUMP_SIZE_BYTES, DB_DUMP_UPLOAD_DURATION_NS, FORMAT, METER_NAME,
+};
 use crates_io::worker::jobs::DumpDb;
 use crates_io_worker::BackgroundJob;
 use flate2::read::GzDecoder;
 use insta::{assert_debug_snapshot, assert_snapshot};
 use object_store::ObjectStoreExt;
+use opentelemetry::KeyValue;
+use opentelemetry::metrics::MeterProvider;
+use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData};
+use opentelemetry_sdk::metrics::{InMemoryMetricExporter, PeriodicReader, SdkMeterProvider};
 use regex::regex;
 use std::io::{Cursor, Read};
 use tar::Archive;
 
 #[tokio::test(flavor = "multi_thread")]
 async fn test_dump_db_job() -> anyhow::Result<()> {
-    let (app, _, _, token) = TestApp::full().with_token().await;
+    let exporter = InMemoryMetricExporter::default();
+    let reader = PeriodicReader::builder(exporter.clone()).build();
+    let provider = SdkMeterProvider::builder().with_reader(reader).build();
+    let meter = provider.meter(METER_NAME);
+
+    let (app, _, _, token) = TestApp::full().with_meter(meter).with_token().await;
     let mut conn = app.db_conn().await;
 
     CrateBuilder::new("test-crate", token.as_model().user_id)
@@ -22,6 +35,13 @@ async fn test_dump_db_job() -> anyhow::Result<()> {
     DumpDb::for_schema(app.db_schema()).enqueue(&conn).await?;
 
     app.run_pending_background_jobs().await;
+    provider.force_flush()?;
+
+    for metric in [DB_DUMP_SIZE_BYTES, DB_DUMP_UPLOAD_DURATION_NS] {
+        for format in ["tar.gz", "zip"] {
+            assert_gt!(metric_value(&exporter, metric, format), 0);
+        }
+    }
 
     assert_snapshot!(app.stored_files().await.join("\n"), @r"
     db-dump.tar.gz
@@ -108,6 +128,26 @@ async fn test_dump_db_job() -> anyhow::Result<()> {
     Ok(())
 }
 
+fn metric_value(exporter: &InMemoryMetricExporter, name: &str, format: &str) -> u64 {
+    let batches = exporter.get_finished_metrics().unwrap();
+    let format = KeyValue::new(FORMAT, format.to_owned());
+    let metric = batches
+        .iter()
+        .flat_map(|batch| batch.scope_metrics())
+        .flat_map(|scope| scope.metrics())
+        .find(|metric| metric.name() == name)
+        .unwrap();
+    let AggregatedMetrics::U64(MetricData::Gauge(gauge)) = metric.data() else {
+        panic!("database dump metrics should be u64 gauges");
+    };
+
+    gauge
+        .data_points()
+        .find(|point| point.attributes().any(|attribute| attribute == &format))
+        .unwrap()
+        .value()
+}
+
 fn tar_paths<R: Read>(archive: &mut Archive<R>) -> Vec<String> {
     let path_date_re = regex!(r"^\d{4}-\d{2}-\d{2}-\d{6}");
 

src/worker/jobs/dump_db.rs

@@ -68,6 +68,8 @@ impl BackgroundJob for DumpDb {
         let upload_start = Instant::now();
         ctx.storage.upload_stream(&tar_key, tar_file).await?;
         let upload_duration = upload_start.elapsed();
+        ctx.metrics.record_db_dump("tar.gz", size, upload_duration);
+
         if let Some(datadog) = &ctx.datadog {
             let domain = &ctx.config.domain_name;
             let result =
@@ -93,6 +95,8 @@ impl BackgroundJob for DumpDb {
         let upload_start = Instant::now();
         ctx.storage.upload_stream(&zip_key, zip_file).await?;
         let upload_duration = upload_start.elapsed();
+        ctx.metrics.record_db_dump("zip", size, upload_duration);
+
         if let Some(datadog) = &ctx.datadog {
             let domain = &ctx.config.domain_name;
             let result = report_dump_metrics(datadog, domain, "zip", size, upload_duration).await;