feat: [Trace Stats] Handle hostname and env (#862)

## This PR
1. For `hostname` field: pass in `tags_provider` and use it to get
function arn. Refer to existing code about:
    1. how to get function arn:

https://github.com/DataDog/datadog-lambda-extension/blob/3f90d2bb09b4213e3b683dc84333e48faeddba3a/bottlecap/src/logs/lambda/processor.rs#L65
    2. use function arn as hostname:

https://github.com/DataDog/datadog-lambda-extension/blob/3f90d2bb09b4213e3b683dc84333e48faeddba3a/bottlecap/src/logs/lambda/processor.rs#L259
2. Change the default value of `env` to `unknown-env`, following this
RFC:
https://docs.google.com/document/d/1NmGVxNyd9o8UQ7AvFKjOVq5tBDN6GN8jeQmvyBUnkrQ

## Notes
Jira: https://datadoghq.atlassian.net/browse/SVLS-7593

Author: lym953

bottlecap/src/bin/bottlecap/main.rs

@@ -1003,7 +1003,7 @@ fn start_trace_agent(
 ) {
     // Stats
     let (stats_concentrator_service, stats_concentrator_handle) =
-        StatsConcentratorService::new(Arc::clone(config));
+        StatsConcentratorService::new(Arc::clone(config), tags_provider.clone());
     tokio::spawn(stats_concentrator_service.run());
     let stats_aggregator: Arc<TokioMutex<StatsAggregator>> = Arc::new(TokioMutex::new(
         StatsAggregator::new_with_concentrator(stats_concentrator_handle.clone()),

bottlecap/src/traces/stats_aggregator.rs

@@ -89,14 +89,22 @@ impl StatsAggregator {
 #[allow(clippy::unwrap_used)]
 mod tests {
     use super::*;
+    use crate::LAMBDA_RUNTIME_SLUG;
     use crate::config::Config;
+    use crate::tags::provider::Provider as TagProvider;
     use crate::traces::stats_concentrator_service::StatsConcentratorService;
+    use std::collections::HashMap;
     use std::sync::Arc;
 
     #[test]
     fn test_add() {
         let config = Arc::new(Config::default());
-        let (_, concentrator) = StatsConcentratorService::new(config);
+        let tags_provider = Arc::new(TagProvider::new(
+            config.clone(),
+            LAMBDA_RUNTIME_SLUG.to_string(),
+            &HashMap::new(),
+        ));
+        let (_, concentrator) = StatsConcentratorService::new(config, tags_provider);
         let mut aggregator = StatsAggregator::new_with_concentrator(concentrator);
         let payload = ClientStatsPayload {
             hostname: "hostname".to_string(),
@@ -123,7 +131,12 @@ mod tests {
     #[tokio::test]
     async fn test_get_batch() {
         let config = Arc::new(Config::default());
-        let (_, concentrator) = StatsConcentratorService::new(config);
+        let tags_provider = Arc::new(TagProvider::new(
+            config.clone(),
+            LAMBDA_RUNTIME_SLUG.to_string(),
+            &HashMap::new(),
+        ));
+        let (_, concentrator) = StatsConcentratorService::new(config, tags_provider);
         let mut aggregator = StatsAggregator::new_with_concentrator(concentrator);
         let payload = ClientStatsPayload {
             hostname: "hostname".to_string(),
@@ -150,7 +163,12 @@ mod tests {
     #[tokio::test]
     async fn test_get_batch_full_entries() {
         let config = Arc::new(Config::default());
-        let (_, concentrator) = StatsConcentratorService::new(config);
+        let tags_provider = Arc::new(TagProvider::new(
+            config.clone(),
+            LAMBDA_RUNTIME_SLUG.to_string(),
+            &HashMap::new(),
+        ));
+        let (_, concentrator) = StatsConcentratorService::new(config, tags_provider);
         let mut aggregator = StatsAggregator::new(640, concentrator);
         // Payload below is 115 bytes
         let payload = ClientStatsPayload {

bottlecap/src/traces/stats_concentrator.rs

@@ -1,4 +1,5 @@
 use crate::config::Config;
+use crate::tags::provider::Provider as TagProvider;
 use datadog_trace_protobuf::pb;
 use std::time::{SystemTime, UNIX_EPOCH};
 use std::{collections::HashMap, sync::Arc};
@@ -54,6 +55,7 @@ pub struct StatsConcentrator {
     config: Arc<Config>,
     tracer_metadata: TracerMetadata,
     buckets: HashMap<u64, Bucket>,
+    hostname: String,
 }
 
 // The number of latest buckets to not flush when force_flush is false.
@@ -71,11 +73,13 @@ const BUCKET_DURATION_NS: u64 = 10 * S_TO_NS; // 10 seconds
 // Aggregates stats into buckets, which are then pulled by the stats aggregator.
 impl StatsConcentrator {
     #[must_use]
-    pub fn new(config: Arc<Config>) -> Self {
+    pub fn new(config: Arc<Config>, tags_provider: Arc<TagProvider>) -> Self {
+        let hostname = tags_provider.get_canonical_id().unwrap_or_default();
         Self {
             config,
             buckets: HashMap::new(),
             tracer_metadata: TracerMetadata::default(), // to be set when a trace is processed
+            hostname,
         }
     }
 
@@ -122,6 +126,7 @@ impl StatsConcentrator {
                         aggregation_key,
                         *stats,
                         &self.tracer_metadata,
+                        &self.hostname,
                     ));
                 }
                 false
@@ -146,10 +151,10 @@ impl StatsConcentrator {
         aggregation_key: &AggregationKey,
         stats: Stats,
         tracer_metadata: &TracerMetadata,
+        hostname: &str,
     ) -> pb::ClientStatsPayload {
         pb::ClientStatsPayload {
-            // TODO: handle this
-            hostname: String::new(),
+            hostname: hostname.to_string(),
             env: aggregation_key.env.clone(),
             // Version is not in the trace payload. Need to read it from config.
             version: config.version.clone().unwrap_or_default(),

bottlecap/src/traces/stats_concentrator_service.rs

@@ -1,6 +1,7 @@
 use tokio::sync::{mpsc, oneshot};
 
 use crate::config::Config;
+use crate::tags::provider::Provider as TagProvider;
 use crate::traces::stats_concentrator::StatsConcentrator;
 use crate::traces::stats_concentrator::StatsEvent;
 use crate::traces::stats_concentrator::TracerMetadata;
@@ -99,10 +100,13 @@ pub struct StatsConcentratorService {
 // to avoid using mutex, which may cause lock contention.
 impl StatsConcentratorService {
     #[must_use]
-    pub fn new(config: Arc<Config>) -> (Self, StatsConcentratorHandle) {
+    pub fn new(
+        config: Arc<Config>,
+        tags_provider: Arc<TagProvider>,
+    ) -> (Self, StatsConcentratorHandle) {
         let (tx, rx) = mpsc::unbounded_channel();
         let handle = StatsConcentratorHandle::new(tx);
-        let concentrator = StatsConcentrator::new(config);
+        let concentrator = StatsConcentrator::new(config, tags_provider);
         let service: StatsConcentratorService = Self { concentrator, rx };
         (service, handle)
     }

bottlecap/src/traces/stats_generator.rs

@@ -39,7 +39,11 @@ impl StatsGenerator {
                         let stats = StatsEvent {
                             time: span.start.try_into().unwrap_or_default(),
                             aggregation_key: AggregationKey {
-                                env: span.meta.get("env").cloned().unwrap_or_default(),
+                                env: span
+                                    .meta
+                                    .get("env")
+                                    .cloned()
+                                    .unwrap_or("unknown-env".to_string()),
                                 service: span.service.clone(),
                                 name: span.name.clone(),
                                 resource: span.resource.clone(),