Introduce data_points_transmitted / second in OTel metrics HTTP, GRPC (#1357)

### What does this PR do?

This commit implements a new piece of telemetry: data points transmitted per
    second. This is the equivalent of bytes per second but, when applicable, in
    terms of the metric (or other) "data points" transmitted per second. This
    approach should be immediately applicable to DogStatsD. This further opens the
    door to other block-based metadata.

### Motivation

REF SMPTNG-659

Author: blt

CHANGELOG.md

@@ -13,6 +13,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Block cache generation now logs its number of rejected, success blocks once
   the cache is created.
 - Added configurable metric kind weights in OpenTelemetry metrics payload generator.
+- Added new telemetry to OpenTelemetry metrics generation: data points per
+  second. This opens the door for doing the same for other payloads.
 
 ## [0.25.9]
 ## Added

lading/src/generator/grpc.rs

@@ -8,6 +8,7 @@
 //! `bytes_written`: Total bytes written
 //! `response_bytes`: Total bytes received
 //! `bytes_per_second`: Configured rate to send data
+//! `data_points_transmitted`: Total data points transmitted (for OpenTelemetry metrics)
 //!
 //! Additional metrics may be emitted by this generator's [throttle].
 //!
@@ -290,6 +291,9 @@ impl Grpc {
                     match res {
                         Ok(res) => {
                             counter!("bytes_written", &self.metric_labels).increment(block_length as u64);
+                            if let Some(data_points) = blk.metadata.data_points {
+                                counter!("data_points_transmitted", &self.metric_labels).increment(data_points);
+                            }
                             counter!("request_ok", &self.metric_labels).increment(1);
                             counter!("response_bytes", &self.metric_labels).increment(res.into_inner() as u64);
                         }

lading/src/generator/http.rs

@@ -7,6 +7,7 @@
 //! `request_failure`: Failed requests
 //! `bytes_written`: Total bytes written
 //! `bytes_per_second`: Configured rate to send data
+//! `data_points_transmitted`: Total data points transmitted (for OpenTelemetry metrics)
 //!
 //! Additional metrics may be emitted by this generator's [throttle].
 //!
@@ -233,12 +234,18 @@ impl Http {
                             let client = client.clone();
                             let labels = labels.clone();
 
+                            let data_points = blk.metadata.data_points;
                             let permit = CONNECTION_SEMAPHORE.get().expect("Connection Semaphore is being initialized or cell is empty").acquire().await.expect("Connection Semaphore has already closed");
                             tokio::spawn(async move {
                                 counter!("requests_sent", &labels).increment(1);
                                 match client.request(request).await {
                                     Ok(response) => {
                                         counter!("bytes_written", &labels).increment(block_length as u64);
+
+                                        if let Some(dp) = data_points {
+                                            counter!("data_points_transmitted", &labels).increment(dp);
+                                        }
+
                                         let status = response.status();
                                         let mut status_labels = labels.clone();
                                         status_labels

lading_payload/src/block.rs

@@ -90,6 +90,15 @@ pub struct Block {
     pub total_bytes: NonZeroU32,
     /// The bytes of this block.
     pub bytes: Bytes,
+    /// Optional metadata for the block
+    pub metadata: BlockMetadata,
+}
+
+/// Metadata associated with a Block
+#[derive(Debug, Clone, Default, Copy)]
+pub struct BlockMetadata {
+    /// Number of data points in this block
+    pub data_points: Option<u64>,
 }
 
 /// Errors for the construction of the block cache
@@ -108,6 +117,7 @@ impl<'a> arbitrary::Arbitrary<'a> for Block {
         Ok(Self {
             total_bytes: NonZeroU32::new(total_bytes).expect("total_bytes must be non-zero"),
             bytes,
+            metadata: BlockMetadata::default(),
         })
     }
 }
@@ -332,6 +342,7 @@ impl Cache {
                 let mut pyld = crate::OpentelemetryMetrics::new(*config, &mut rng)?;
                 let span = span!(Level::INFO, "fixed", payload = "otel-metrics");
                 let _guard = span.enter();
+
                 construct_block_cache_inner(rng, &mut pyld, maximum_block_bytes, total_bytes.get())?
             }
         };
@@ -585,9 +596,21 @@ where
         let min_actual_block_str = Byte::from_u64(min_actual_block_size.into())
             .get_appropriate_unit(byte_unit::UnitType::Binary)
             .to_string();
-        info!(
-            "Filled {filled_sum_str} of requested {capacity_sum_str}. Discovered minimum block size of {min_actual_block_str}, maximum: {max_actual_block_str}. Total success blocks: {success_block_sizes}. Total rejected blocks: {rejected_block_sizes}."
-        );
+
+        let total_data_points: u64 = block_cache
+            .iter()
+            .filter_map(|b| b.metadata.data_points)
+            .sum();
+
+        if total_data_points > 0 {
+            info!(
+                "Filled {filled_sum_str} of requested {capacity_sum_str}. Discovered minimum block size of {min_actual_block_str}, maximum: {max_actual_block_str}. Total success blocks: {success_block_sizes}. Total rejected blocks: {rejected_block_sizes}. Total data points: {total_data_points}."
+            );
+        } else {
+            info!(
+                "Filled {filled_sum_str} of requested {capacity_sum_str}. Discovered minimum block size of {min_actual_block_str}, maximum: {max_actual_block_str}. Total success blocks: {success_block_sizes}. Total rejected blocks: {rejected_block_sizes}."
+            );
+        }
 
         Ok(block_cache)
     }
@@ -626,6 +649,16 @@ where
                 .expect("failed to get length of bytes"),
         )
         .ok_or(SpinError::Zero)?;
-        Ok(Block { total_bytes, bytes })
+
+        let mut metadata = BlockMetadata::default();
+        if let Some(data_points) = serializer.data_points_generated() {
+            metadata.data_points = Some(data_points);
+        }
+
+        Ok(Block {
+            total_bytes,
+            bytes,
+            metadata,
+        })
     }
 }

lading_payload/src/lib.rs

@@ -104,6 +104,18 @@ pub trait Serialize {
     where
         R: Rng + Sized,
         W: Write;
+
+    /// Reports data points count for the most recently generated content.
+    ///
+    /// IMPORTANT: This method should be called immediately after `to_bytes` to
+    /// get accurate counts for the most recently generated block. The
+    /// information WILL be overwritten by subsequent calls to `to_bytes`.
+    ///
+    /// If this function returns None the serialize does not support tracking
+    /// data points.
+    fn data_points_generated(&self) -> Option<u64> {
+        None
+    }
 }
 
 /// Sub-configuration for `TraceAgent` format
@@ -201,6 +213,14 @@ impl Serialize for Payload {
             Payload::TraceAgent(ser) => ser.to_bytes(rng, max_bytes, writer),
         }
     }
+
+    fn data_points_generated(&self) -> Option<u64> {
+        match self {
+            Payload::OtelMetrics(ser) => ser.data_points_generated(),
+            // Other implementations use the default None
+            _ => None,
+        }
+    }
 }
 
 /// Generate instances of `Self::Output` from source of randomness.

lading_payload/src/opentelemetry_metric.rs

@@ -260,16 +260,19 @@ impl Config {
 #[derive(Debug, Clone)]
 /// OTLP metric payload
 pub struct OpentelemetryMetrics {
-    /// Template pool for metric generation
     pool: Pool,
-    /// Scratch buffer for serialization
     scratch: RefCell<BytesMut>,
     /// Current tick count for monotonic timing (starts at 0)
     tick: u64,
     /// Accumulating sum increment, floating point
     incr_f: f64,
     /// Accumulating sum increment, integer
     incr_i: i64,
+    /// Number of data points in the most recent `ResourceMetrics` (set by
+    /// `generate`).
+    data_points_per_resource: u64,
+    /// Number of data points in the most recent block (set by `to_bytes`).
+    data_points_per_block: u64,
 }
 
 impl OpentelemetryMetrics {
@@ -293,6 +296,8 @@ impl OpentelemetryMetrics {
             tick: 0,
             incr_f: 0.0,
             incr_i: 0,
+            data_points_per_resource: 0,
+            data_points_per_block: 0,
         })
     }
 }
@@ -303,8 +308,10 @@ impl<'a> SizedGenerator<'a> for OpentelemetryMetrics {
 
     /// Generate OTLP metrics with the following enhancements:
     ///
-    /// * Monotonic sums are truly monotonic, incrementing by a random amount each tick
-    /// * Timestamps advance monotonically based on internal tick counter starting at epoch
+    /// * Monotonic sums are truly monotonic, incrementing by a random amount
+    ///   each tick
+    /// * Timestamps advance monotonically based on internal tick counter
+    ///   starting at epoch
     /// * Each call advances the tick counter by a random amount (1-60)
     fn generate<R>(&'a mut self, rng: &mut R, budget: &mut usize) -> Result<Self::Output, Error>
     where
@@ -326,11 +333,14 @@ impl<'a> SizedGenerator<'a> for OpentelemetryMetrics {
         // Update data points in each metric. For gauges we use random values,
         // for accumulating sums we increment by a fixed amount per tick.
         // All timestamps are updated based on the current tick.
+        let mut data_points_count = 0;
+
         for scope_metrics in &mut tpl.scope_metrics {
             for metric in &mut scope_metrics.metrics {
                 if let Some(data) = &mut metric.data {
                     match data {
                         Data::Gauge(gauge) => {
+                            data_points_count += gauge.data_points.len() as u64;
                             for point in &mut gauge.data_points {
                                 point.time_unix_nano = self.tick * TIME_INCREMENT_NANOS;
                                 if let Some(value) = &mut point.value {
@@ -346,6 +356,7 @@ impl<'a> SizedGenerator<'a> for OpentelemetryMetrics {
                             }
                         }
                         Data::Sum(sum) => {
+                            data_points_count += sum.data_points.len() as u64;
                             let is_accumulating = sum.is_monotonic;
                             for point in &mut sum.data_points {
                                 point.time_unix_nano = self.tick * TIME_INCREMENT_NANOS;
@@ -385,6 +396,8 @@ impl<'a> SizedGenerator<'a> for OpentelemetryMetrics {
             }
         }
 
+        self.data_points_per_resource = data_points_count;
+
         Ok(tpl)
     }
 }
@@ -403,10 +416,14 @@ impl crate::Serialize for OpentelemetryMetrics {
             resource_metrics: Vec::with_capacity(8),
         };
 
+        let mut total_data_points = 0;
         let loop_id: u32 = rng.random();
+
         while bytes_remaining >= SMALLEST_PROTOBUF {
             if let Ok(rm) = self.generate(&mut rng, &mut bytes_remaining) {
+                total_data_points += self.data_points_per_resource;
                 request.resource_metrics.push(rm);
+
                 let required_bytes = request.encoded_len();
                 debug!(
                     ?loop_id,
@@ -417,6 +434,7 @@ impl crate::Serialize for OpentelemetryMetrics {
                     "to_bytes inner loop"
                 );
                 if required_bytes > max_bytes {
+                    total_data_points -= self.data_points_per_resource;
                     drop(request.resource_metrics.pop());
                     break;
                 }
@@ -443,8 +461,14 @@ impl crate::Serialize for OpentelemetryMetrics {
             writer.write_all(&buf)?;
         }
 
+        self.data_points_per_block = total_data_points;
+
         Ok(())
     }
+
+    fn data_points_generated(&self) -> Option<u64> {
+        Some(self.data_points_per_block)
+    }
 }
 
 #[cfg(test)]