chore(metrics): add v3 endpoint telemetry (#1888)

## Summary
<!-- Please provide a brief summary about what this PR does.
This should help the reviewers give feedback faster and with higher
quality. -->

  Adds ADP telemetry needed for Metrics V3 rollout dashboards.

  New/remapped telemetry:

  - `serializer.v3_column_size{column,compressed}`
  - `serializer.v3_values_count{type}`
  - `serializer.v3_payload_split_reason{reason}`
  - `sketch_series.sketch_too_big`
  - transaction bytes by endpoint/domain/status

## Change Type
- [ ] Bug fix
- [ ] New feature
- [x] Non-functional (chore, refactoring, docs)
- [ ] Performance


## How did you test this PR?
<!-- Please how you tested these changes here -->
Deployed custom image to staging
## References

<!-- Please list any issues closed by this PR. -->

<!--
- Closes: <issue link>
-->

<!-- Any other issues or PRs relevant to this PR? Feel free to list them
here. -->

Author: rayz

bin/agent-data-plane/src/state/metrics/rules/mod.rs

@@ -3,6 +3,7 @@ use saluki_core::observability::metrics::RemapperRule;
 mod aggregation;
 mod compat;
 mod dogstatsd;
+mod serializer;
 mod transaction;
 
 /// Returns the list of remapper rules relevant to metrics we send to the Datadog Agent via Remote Agent Registry (RAR).
@@ -14,6 +15,7 @@ pub fn get_datadog_agent_remappings() -> Vec<RemapperRule> {
     rules.extend(self::dogstatsd::get_dogstatsd_remappings());
     rules.extend(self::aggregation::get_aggregation_remappings());
     rules.extend(self::transaction::get_transaction_remappings());
+    rules.extend(self::serializer::get_serializer_remappings());
     rules
 }
 

bin/agent-data-plane/src/state/metrics/rules/serializer.rs

@@ -0,0 +1,20 @@
+use super::RemapperRule;
+
+pub fn get_serializer_remappings() -> Vec<RemapperRule> {
+    vec![
+        RemapperRule::by_name("adp.serializer.v3_column_size", "serializer.v3_column_size")
+            .with_original_tags(["column", "compressed"])
+            .with_help_text("Number of bytes occupied by each column"),
+        RemapperRule::by_name("adp.serializer.v3_values_count", "serializer.v3_values_count")
+            .with_original_tags(["type"])
+            .with_help_text("Number of values encoded using a specific encoding type"),
+        RemapperRule::by_name(
+            "adp.serializer.v3_payload_split_reason",
+            "serializer.v3_payload_split_reason",
+        )
+        .with_original_tags(["reason"])
+        .with_help_text("Why payload was split"),
+        RemapperRule::by_name("adp.serializer.v3_item_too_big", "sketch_series.sketch_too_big")
+            .with_help_text("Number of payloads dropped because they were too big for the stream compressor"),
+    ]
+}

bin/agent-data-plane/src/state/metrics/rules/transaction.rs

@@ -3,6 +3,15 @@ use super::RemapperRule;
 pub fn get_transaction_remappings() -> Vec<RemapperRule> {
     vec![
         // Transaction metrics.
+        RemapperRule::by_name(
+            "adp.network_http_requests_input_bytes_total",
+            "transactions.input_bytes",
+        )
+        .with_original_tags(["domain", "endpoint"])
+        .with_help_text("Incoming transaction sizes in bytes"),
+        RemapperRule::by_name("adp.network_http_requests_input_total", "transactions.input_count")
+            .with_original_tags(["domain", "endpoint"])
+            .with_help_text("Incoming transaction count"),
         RemapperRule::by_name("adp.network_http_requests_failed_total", "transactions.dropped")
             .with_original_tags(["domain", "endpoint"])
             .with_help_text("Transaction drop count"),

lib/saluki-components/src/common/datadog/io.rs

@@ -43,6 +43,8 @@ use super::{
     METRIC_INTAKE_PATHS,
 };
 
+type EndpointNameFn = dyn Fn(&Uri) -> Option<MetaString> + Send + Sync;
+
 /// Size of buffer chunks for request builder buffers.
 ///
 /// Used to influence the size of chunks in `ChunkedBytesBuffer`.
@@ -99,6 +101,7 @@ pub struct TransactionForwarder<B> {
     telemetry: ComponentTelemetry,
     metrics_builder: MetricsBuilder,
     client: HttpClient,
+    endpoint_name: Arc<EndpointNameFn>,
     endpoints: Vec<RoutableEndpoint>,
     _marker: std::marker::PhantomData<B>,
 }
@@ -160,13 +163,18 @@ where
         F: Fn(&Uri) -> Option<MetaString> + Send + Sync + 'static,
     {
         let endpoints = config.build_routable_endpoints(live_config.clone())?;
+        let endpoint_name: Arc<EndpointNameFn> = Arc::new(endpoint_name);
+        let endpoint_name_for_client = Arc::clone(&endpoint_name);
         let mut client_builder = HttpClient::builder()
             .with_request_timeout(config.request_timeout())
             .with_max_idle_conns_per_host(config.max_idle_connections_per_host())
             .with_min_tls_version(config.min_tls_version())
             .with_http_protocol(config.http_protocol())
             .with_bytes_sent_counter(telemetry.bytes_sent().clone())
-            .with_endpoint_telemetry(metrics_builder.clone(), Some(endpoint_name));
+            .with_endpoint_telemetry(
+                metrics_builder.clone(),
+                Some(move |uri: &Uri| endpoint_name_for_client(uri)),
+            );
         if let Some(path) = config.ssl_key_log_file_path() {
             client_builder = client_builder.with_tls_config(|builder| builder.with_key_log_file(path));
         }
@@ -189,6 +197,7 @@ where
             telemetry,
             metrics_builder,
             client,
+            endpoint_name,
             endpoints,
             _marker: std::marker::PhantomData,
         })
@@ -210,6 +219,7 @@ where
             telemetry,
             metrics_builder,
             client,
+            endpoint_name,
             endpoints,
             _marker,
         } = self;
@@ -225,6 +235,7 @@ where
                 client,
                 telemetry,
                 metrics_builder,
+                endpoint_name,
                 endpoints,
             ),
         );
@@ -241,7 +252,7 @@ async fn run_io_loop<B>(
     mut transactions_rx: mpsc::Receiver<Transaction<B>>, io_shutdown_tx: oneshot::Sender<()>,
     context: ComponentContext, config: ForwarderConfiguration, live_config: Option<GenericConfiguration>,
     service: HttpClient, telemetry: ComponentTelemetry, metrics_builder: MetricsBuilder,
-    resolved_endpoints: Vec<RoutableEndpoint>,
+    endpoint_name: Arc<EndpointNameFn>, resolved_endpoints: Vec<RoutableEndpoint>,
 ) where
     B: Body + Buf + Clone + Send + Sync + 'static,
     B::Data: Send,
@@ -278,6 +289,7 @@ async fn run_io_loop<B>(
                 service.clone(),
                 telemetry.clone(),
                 txnq_telemetry,
+                Arc::clone(&endpoint_name),
                 resolved_endpoint,
             ),
         );
@@ -357,7 +369,8 @@ fn should_route_to_endpoint(is_metrics_request: bool, has_metrics_primary: bool,
 async fn run_endpoint_io_loop<B>(
     mut txns_rx: mpsc::Receiver<Transaction<B>>, task_barrier: Arc<Barrier>, context: ComponentContext,
     config: ForwarderConfiguration, live_config: Option<GenericConfiguration>, service: HttpClient,
-    telemetry: ComponentTelemetry, txnq_telemetry: TransactionQueueTelemetry, endpoint: ResolvedEndpoint,
+    telemetry: ComponentTelemetry, txnq_telemetry: TransactionQueueTelemetry, endpoint_name: Arc<EndpointNameFn>,
+    endpoint: ResolvedEndpoint,
 ) where
     B: Body + Buf + Clone + Send + Sync + 'static,
     B::Data: Send,
@@ -454,6 +467,14 @@ async fn run_endpoint_io_loop<B>(
                     } else {
                         strip_metrics_validation_headers(txn)
                     };
+                    let transaction_size = txn.size_bytes();
+                    let transaction_endpoint_name = endpoint_name(txn.request_uri())
+                        .unwrap_or_else(|| MetaString::from(txn.request_uri().path()));
+                    telemetry.track_transaction_input(
+                        &endpoint_domain,
+                        &transaction_endpoint_name,
+                        transaction_size,
+                    );
 
                     match pending_txns.push_high_priority(txn).await {
                         Ok(push_result) => track_queue_drops(&telemetry, &endpoint_domain, push_result),

lib/saluki-components/src/common/datadog/telemetry.rs

@@ -12,6 +12,11 @@ const NETWORK_HTTP_REQUESTS_ERRORS_TOTAL: &str = "network_http_requests_errors_t
 const ERROR_TYPE_SENT_REQUEST: &str = "sent_request_error";
 const ERROR_SCOPE_TRANSACTION: &str = "transaction";
 
+struct TransactionInputTelemetry {
+    count: Counter,
+    bytes: Counter,
+}
+
 /// Component-specific telemetry.
 ///
 /// This type covers high-level component telemetry, such as events/bytes sent, tailored to the Datadog destinations and
@@ -22,6 +27,7 @@ pub struct ComponentTelemetry {
     events_sent: Counter,
     events_sent_batch_size: Histogram,
     bytes_sent: Counter,
+    transaction_input_by_endpoint: Arc<Mutex<HashMap<(String, String), TransactionInputTelemetry>>>,
     data_points_sent_by_domain: Arc<Mutex<HashMap<String, Gauge>>>,
     data_points_dropped_by_domain: Arc<Mutex<HashMap<String, Gauge>>>,
     events_dropped_http: Counter,
@@ -41,6 +47,7 @@ impl ComponentTelemetry {
             events_sent: builder.register_counter("component_events_sent_total"),
             events_sent_batch_size: builder.register_trace_histogram("component_events_sent_batch_size"),
             bytes_sent: builder.register_counter("component_bytes_sent_total"),
+            transaction_input_by_endpoint: Arc::new(Mutex::new(HashMap::new())),
             data_points_sent_by_domain: Arc::new(Mutex::new(HashMap::new())),
             data_points_dropped_by_domain: Arc::new(Mutex::new(HashMap::new())),
             events_dropped_http: builder.register_counter_with_tags(
@@ -73,6 +80,29 @@ impl ComponentTelemetry {
         &self.bytes_sent
     }
 
+    /// Tracks a transaction entering the forwarder queue.
+    pub fn track_transaction_input(&self, domain: &str, endpoint: &str, bytes: u64) {
+        let mut telemetry = self.transaction_input_by_endpoint.lock().unwrap();
+        let per_endpoint = telemetry
+            // TODO: Avoid allocating key strings on every transaction if we can make the telemetry registry lookup
+            // support borrowed endpoint/domain keys.
+            .entry((domain.to_string(), endpoint.to_string()))
+            .or_insert_with(|| {
+                let tags = [("domain", domain.to_string()), ("endpoint", endpoint.to_string())];
+                TransactionInputTelemetry {
+                    count: self
+                        .builder
+                        .register_counter_with_tags("network_http_requests_input_total", tags.clone()),
+                    bytes: self
+                        .builder
+                        .register_counter_with_tags("network_http_requests_input_bytes_total", tags),
+                }
+            });
+
+        per_endpoint.count.increment(1);
+        per_endpoint.bytes.increment(bytes);
+    }
+
     /// Returns a reference to the "events dropped (encoder)" counter.
     pub fn events_dropped_encoder(&self) -> &Counter {
         &self.events_dropped_encoder