batch: add metrics for total batches and timeout batches (#97)

Allow evaluating the efficiency of the batch flush settings.

Signed-off-by: Matt Klein <[email protected]>

Author: mattklein123

Cargo.lock

@@ -16,15 +16,15 @@ ahash                  = "0.8"
 anyhow                 = "1"
 assert_matches         = "1.5.0"
 async-trait            = "0.1"
-aws-config             = "1.8.6"
-aws-credential-types   = "1.2.6"
-aws-sdk-sqs            = "1.83.0"
-aws-sigv4              = "1.3.4"
-aws-smithy-async       = "1.2.5"
-aws-smithy-http        = "0.62.3"
-aws-smithy-runtime-api = { version = "1.9.0", features = ["test-util"] }
-aws-smithy-types       = "1.3.2"
-axum                   = "0.8.4"
+aws-config             = "1.8.8"
+aws-credential-types   = "1.2.8"
+aws-sdk-sqs            = "1.86.0"
+aws-sigv4              = "1.3.5"
+aws-smithy-async       = "1.2.6"
+aws-smithy-http        = "0.62.4"
+aws-smithy-runtime-api = { version = "1.9.1", features = ["test-util"] }
+aws-smithy-types       = "1.3.3"
+axum                   = "0.8.6"
 backoff                = { version = "0.4.0", features = ["tokio"] }
 base64ct               = "1.8.0"
 bd-grpc                = { git = "https://github.com/bitdriftlabs/shared-core.git" }
@@ -42,12 +42,12 @@ bd-test-helpers        = { git = "https://github.com/bitdriftlabs/shared-core.gi
 bd-time                = { git = "https://github.com/bitdriftlabs/shared-core.git" }
 built                  = { version = "0.8", features = ["git2"] }
 bytes                  = "1"
-cc                     = "1.2.37"
-clap                   = { version = "4.5.47", features = ["derive", "env"] }
+cc                     = "1.2.41"
+clap                   = { version = "4.5.50", features = ["derive", "env"] }
 comfy-table            = "7.2.1"
 console-subscriber     = "0.4.1"
 criterion              = { version = "0.7", features = ["html_reports"] }
-ctor                   = "0.5.0"
+ctor                   = "0.6.0"
 cuckoofilter           = "0.5.0"
 dashmap                = { version = "6", features = ["raw-api"] }
 deadpool               = { version = "0.12", features = ["managed", "rt_tokio_1"] }
@@ -69,13 +69,13 @@ hyper-rustls = { version = "0.27.7", default-features = false, features = [
   "aws-lc-rs",
 ] }
 
-hyper-util            = { version = "0.1.16", features = ["client", "client-legacy"] }
+hyper-util            = { version = "0.1.17", features = ["client", "client-legacy"] }
 hyperloglogplus       = "0.4.1"
 intrusive-collections = "0.9.7"
 itertools             = "0.14.0"
 k8s-openapi           = { version = "0.26.0", features = ["v1_30"] }
 
-kube = { version = "2.0.0", features = [
+kube = { version = "2.0.1", features = [
   "runtime",
   "derive",
   "rustls-tls",
@@ -106,24 +106,24 @@ rand              = { version = "0.9", features = ["small_rng"] }
 rand_xoshiro      = "0.7"
 regex             = "1"
 
-reqwest = { version = "0.12.23", default-features = false, features = [
+reqwest = { version = "0.12.24", default-features = false, features = [
   "rustls-tls-webpki-roots",
   "json",
 ] }
 
 reusable-fmt               = "0.2.0"
-rustls                     = "0.23.31"
+rustls                     = "0.23.34"
 serde                      = { version = "1", features = ["derive"] }
 serde_json                 = "1"
 serde_yaml                 = "0.9.34"
 snap                       = "1"
-socket2                    = "0.6.0"
-tempfile                   = "3.22"
+socket2                    = "0.6.1"
+tempfile                   = "3.23"
 thiserror                  = "2"
-tikv-jemalloc-ctl          = "0.6.0"
+tikv-jemalloc-ctl          = "0.6.1"
 tikv-jemallocator.features = ["profiling", "stats", "unprefixed_malloc_on_supported_platforms"]
-tikv-jemallocator.version  = "0.6.0"
-time                       = { version = "0.3.43", features = ["formatting"] }
+tikv-jemallocator.version  = "0.6.1"
+time                       = { version = "0.3.44", features = ["formatting"] }
 tokio                      = { version = "1", features = ["full", "parking_lot", "tracing", "test-util"] }
 tokio-stream               = "0.1.17"
 tokio-test                 = "0.4"

Cargo.toml

@@ -1,5 +1,6 @@
 [graph]
 all-features        = false
+exclude             = ["bd-workspace-hack"]
 no-default-features = false
 targets             = []
 

deny.toml

@@ -60,6 +60,8 @@ struct PerBatchLockedData<T> {
 struct Stats {
   dropped_bytes: IntCounter,
   queued_bytes: IntGauge,
+  total_batches: IntCounter,
+  timeout_batches: IntCounter,
 }
 
 //
@@ -99,6 +101,8 @@ impl<I: Send + Sync + 'static, B: Batch<I> + Send + Sync + 'static> BatchBuilder
     let stats = Stats {
       dropped_bytes: scope.counter("dropped_bytes"),
       queued_bytes: scope.gauge("queued_bytes"),
+      total_batches: scope.counter("total_batches"),
+      timeout_batches: scope.counter("timeout_batches"),
     };
 
     let batch_builder = Arc::new(Self {
@@ -151,6 +155,7 @@ impl<I: Send + Sync + 'static, B: Batch<I> + Send + Sync + 'static> BatchBuilder
           &cloned_batch_builder.notify_on_data,
           batch,
           size,
+          true,
         );
       }
 
@@ -255,6 +260,7 @@ impl<I: Send + Sync + 'static, B: Batch<I> + Send + Sync + 'static> BatchBuilder
           &self.notify_on_data,
           pending_batch,
           finished_size,
+          false,
         );
       } else if locked_pending_data.batch_fill_wait_task.is_none() {
         // If there is no fill task, start one.
@@ -286,6 +292,7 @@ impl<I: Send + Sync + 'static, B: Batch<I> + Send + Sync + 'static> BatchBuilder
             &cloned_self.notify_on_data,
             pending_batch,
             size,
+            true,
           );
         }));
       }
@@ -299,8 +306,13 @@ impl<I: Send + Sync + 'static, B: Batch<I> + Send + Sync + 'static> BatchBuilder
     notify_on_data: &Notify,
     pending_batch: B,
     size: usize,
+    timeout_batch: bool,
   ) {
     Self::inc_total_size(locked_data, stats, size);
+    stats.total_batches.inc();
+    if timeout_batch {
+      stats.timeout_batches.inc();
+    }
 
     // In order to avoid spurious wakeups, we keep track of whether there are any waiters.
     // TODO(mattklein123): Due to the use of multiple batch builders in the Lyft specific config

pulse-metrics/src/batch.rs

@@ -164,6 +164,6 @@ fn parsed_metric_roundtrip_metric(metric: Metric) -> anyhow::Result<()> {
   if m.eq(&m2) {
     Ok(())
   } else {
-    Err(anyhow::anyhow!("m = {:?}, m2 = {:?}", m, m2))
+    Err(anyhow::anyhow!("m = {m:?}, m2 = {m2:?}"))
   }
 }

pulse-metrics/src/pipeline/processor/internode/convert_test.rs

@@ -129,22 +129,14 @@ fn metrics_roundtrip_write_request(input: Vec<ParsedMetric>) -> anyhow::Result<(
   );
   assert!(errors.is_empty());
   if input.len() != output.len() {
-    bail!(
-      "mismatched lengths: input {:?}, output: {:?}",
-      input,
-      output
-    );
+    bail!("mismatched lengths: input {input:?}, output: {output:?}");
   }
   for input_metric in input.clone() {
     let matching = output
       .iter()
       .find(|output_metric| &input_metric.metric == *output_metric);
     if Some(&input_metric.metric.clone()) != matching {
-      bail!(
-        "input missing from output: input {:?}, output: {:?}",
-        input_metric,
-        output
-      );
+      bail!("input missing from output: input {input_metric:?}, output: {output:?}");
     }
   }
   for output_metric in output {
@@ -153,11 +145,7 @@ fn metrics_roundtrip_write_request(input: Vec<ParsedMetric>) -> anyhow::Result<(
       .find(|input_metric| input_metric.metric == output_metric)
       .map(|metric| &metric.metric);
     if Some(&output_metric.clone()) != matching {
-      bail!(
-        "output missing from input: output {:?}, input: {:?}",
-        output_metric,
-        input
-      );
+      bail!("output missing from input: output {output_metric:?}, input: {input:?}");
     }
   }
   Ok(())

pulse-metrics/src/protos/metric_test.rs

@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.1
+// .proto file is parsed by protoc 33.0
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702

pulse-protobuf/src/protos/opentelemetry/common.rs

@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.1
+// .proto file is parsed by protoc 33.0
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702

pulse-protobuf/src/protos/opentelemetry/metrics.rs

@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.1
+// .proto file is parsed by protoc 33.0
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702

pulse-protobuf/src/protos/opentelemetry/metrics_service.rs

@@ -6,7 +6,7 @@
 // https://polyformproject.org/wp-content/uploads/2020/06/PolyForm-Shield-1.0.0.txt
 
 // This file is generated by rust-protobuf 4.0.0-alpha.0. Do not edit
-// .proto file is parsed by protoc 32.1
+// .proto file is parsed by protoc 33.0
 // @generated
 
 // https://github.com/rust-lang/rust-clippy/issues/702