fix(dogstatsd): properly pack multiple metrics into each payload up to the configured limit (#586)

Author: tobz

Cargo.lock

@@ -0,0 +1,8 @@
+# Seeds for failure cases proptest has generated in the past. It is
+# automatically read and these particular cases re-run before any
+# novel cases are generated.
+#
+# It is recommended to check this file in to source control so that
+# everyone who runs the test benefits from these saved cases.
+cc ecafaadc4a51995b06f20080248f4e17c818026dc20e29e2a2dd233393af6773 # shrinks to payload_limit = 27, inputs = [Histogram(Key { name: KeyName("aAAaAa0a"), labels: [Label("aaaaa", "")], hashed: true, hash: 4563833112871895604 }, [-0.0]), Histogram(Key { name: KeyName("0AA0AaaA"), labels: [], hashed: true, hash: 13699395814728153738 }, [-0.0])]
+cc 8d5fca863d5e68150df55ff75e9718f6491d27dfd598e53b29fd0583a5e05a66 # shrinks to payload_limit = 0, inputs = [Counter(Key { name: KeyName("aaAAAAaa"), labels: [], hashed: true, hash: 15261619708481900111 }, 0, None)]

metrics-exporter-dogstatsd/proptest-regressions/forwarder/writer.txt

@@ -347,7 +347,6 @@ impl DogStatsDBuilder {
             histogram_sampling: self.histogram_sampling,
             histogram_reservoir_size: self.histogram_reservoir_size,
             histograms_as_distributions: self.histograms_as_distributions,
-            global_labels: self.global_labels,
             global_prefix: self.global_prefix,
         };
 
@@ -367,6 +366,7 @@ impl DogStatsDBuilder {
             max_payload_len,
             flush_interval,
             write_timeout: self.write_timeout,
+            global_labels: self.global_labels,
         };
 
         if self.synchronous {

metrics-exporter-dogstatsd/proptest-regressions/writer.txt

@@ -6,6 +6,8 @@ use std::{
     time::Duration,
 };
 
+use metrics::Label;
+
 pub mod sync;
 
 #[derive(Clone, Debug, Eq, PartialEq)]
@@ -105,9 +107,20 @@ fn unknown_scheme_error_str(scheme: &str) -> String {
 #[derive(Clone)]
 pub(crate) struct ForwarderConfiguration {
     pub remote_addr: RemoteAddr,
+
+    /// Maximum size, in bytes, for an individual payload.
+    ///
+    /// Payloads may contain multiple metrics.
     pub max_payload_len: usize,
+
+    /// Duration to wait between flushing metrics.
     pub flush_interval: Duration,
+
+    /// Timeout for writing to the socket.
     pub write_timeout: Duration,
+
+    /// Global labels to attach to all metrics.
+    pub global_labels: Vec<Label>,
 }
 
 impl ForwarderConfiguration {

metrics-exporter-dogstatsd/src/builder.rs

@@ -146,7 +146,8 @@ impl Forwarder {
     pub fn run(mut self) {
         let mut flush_state = FlushState::default();
         let mut writer =
-            PayloadWriter::new(self.config.max_payload_len, self.config.is_length_prefixed());
+            PayloadWriter::new(self.config.max_payload_len, self.config.is_length_prefixed())
+                .with_global_labels(&self.config.global_labels);
         let mut telemetry_update = TelemetryUpdate::default();
 
         let mut next_flush = Instant::now() + self.config.flush_interval;

metrics-exporter-dogstatsd/src/forwarder/mod.rs

@@ -1,6 +1,6 @@
 use std::{collections::HashSet, time::SystemTime};
 
-use metrics::{Key, Label};
+use metrics::Key;
 use metrics_util::registry::Registry;
 use tracing::error;
 
@@ -28,9 +28,6 @@ pub struct StateConfiguration {
     /// Whether or not to emit histograms as distributions.
     pub histograms_as_distributions: bool,
 
-    /// Global labels to add to all metrics
-    pub global_labels: Vec<Label>,
-
     /// Global prefix/namespace to use for all metrics
     pub global_prefix: Option<String>,
 }
@@ -108,13 +105,8 @@ impl State {
                 self.config.global_prefix.as_deref()
             };
 
-            let result = writer.write_counter(
-                &key,
-                value,
-                self.get_aggregation_timestamp(),
-                prefix,
-                &self.config.global_labels,
-            );
+            let result =
+                writer.write_counter(&key, value, self.get_aggregation_timestamp(), prefix);
             if result.any_failures() {
                 let points_dropped = result.points_dropped();
                 error!(
@@ -140,13 +132,7 @@ impl State {
             } else {
                 self.config.global_prefix.as_deref()
             };
-            let result = writer.write_gauge(
-                &key,
-                value,
-                self.get_aggregation_timestamp(),
-                prefix,
-                &self.config.global_labels,
-            );
+            let result = writer.write_gauge(&key, value, self.get_aggregation_timestamp(), prefix);
             if result.any_failures() {
                 let points_dropped = result.points_dropped();
                 error!(metric_name = key.name(), points_dropped, "Failed to build gauge payload.");
@@ -175,21 +161,9 @@ impl State {
             histogram.flush(|maybe_sample_rate, values| {
                 let points_len = values.len();
                 let result = if self.config.histograms_as_distributions {
-                    writer.write_distribution(
-                        &key,
-                        values,
-                        maybe_sample_rate,
-                        prefix,
-                        &self.config.global_labels,
-                    )
+                    writer.write_distribution(&key, values, maybe_sample_rate, prefix)
                 } else {
-                    writer.write_histogram(
-                        &key,
-                        values,
-                        maybe_sample_rate,
-                        prefix,
-                        &self.config.global_labels,
-                    )
+                    writer.write_histogram(&key, values, maybe_sample_rate, prefix)
                 };
 
                 // Scale the points flushed/dropped values by the sample rate to determine the true number of points flushed/dropped.