chore: Add/update doc for aggregators (#764)

Add / update doc for trace / log / otlp / proxy aggregator.

Author: lym953

bottlecap/src/logs/aggregator.rs

@@ -3,6 +3,7 @@ use tracing::warn;
 
 use crate::logs::constants;
 
+/// Takes in log payloads and aggregates them into larger batches to be flushed to Datadog.
 #[derive(Debug, Clone)]
 pub struct Aggregator {
     messages: VecDeque<String>,
@@ -41,12 +42,14 @@ impl Aggregator {
         }
     }
 
+    /// Takes in a batch of log payloads.
     pub fn add_batch(&mut self, logs: Vec<String>) {
         for log in logs {
             self.messages.push_back(log);
         }
     }
 
+    /// Returns a batch of log payloads, subject to the max content size.
     pub fn get_batch(&mut self) -> Vec<u8> {
         self.buffer.extend(b"[");
 

bottlecap/src/otlp/agent.rs

@@ -176,18 +176,18 @@ impl Agent {
 
         match trace_tx.send(send_data_builder).await {
             Ok(()) => {
-                debug!("OTLP | Successfully buffered traces to be flushed.");
+                debug!("OTLP | Successfully buffered traces to be aggregated.");
                 (
                     StatusCode::OK,
                     json!({"rate_by_service":{"service:,env:":1}}).to_string(),
                 )
                     .into_response()
             }
             Err(err) => {
-                error!("OTLP | Error sending traces to the trace flusher: {err}");
+                error!("OTLP | Error sending traces to the trace aggregator: {err}");
                 (
                     StatusCode::INTERNAL_SERVER_ERROR,
-                    json!({ "message": format!("Error sending traces to the trace flusher: {err}") }).to_string()
+                    json!({ "message": format!("Error sending traces to the trace aggregator: {err}") }).to_string()
                 ).into_response()
             }
         }

bottlecap/src/traces/proxy_aggregator.rs

@@ -7,6 +7,7 @@ pub struct ProxyRequest {
     pub target_url: String,
 }
 
+/// Takes in individual proxy requests and aggregates them into batches to be flushed to Datadog.
 pub struct Aggregator {
     queue: Vec<ProxyRequest>,
 }
@@ -20,10 +21,12 @@ impl Default for Aggregator {
 }
 
 impl Aggregator {
+    /// Takes in an individual proxy request.
     pub fn add(&mut self, request: ProxyRequest) {
         self.queue.push(request);
     }
 
+    /// Returns a batch of proxy requests.
     pub fn get_batch(&mut self) -> Vec<ProxyRequest> {
         std::mem::take(&mut self.queue)
     }

bottlecap/src/traces/stats_aggregator.rs

@@ -34,6 +34,7 @@ impl Default for StatsAggregator {
     }
 }
 
+/// Takes in individual trace stats payloads and aggregates them into batches to be flushed to Datadog.
 impl StatsAggregator {
     #[allow(dead_code)]
     #[allow(clippy::must_use_candidate)]
@@ -45,10 +46,12 @@ impl StatsAggregator {
         }
     }
 
+    /// Takes in an individual trace stats payload.
     pub fn add(&mut self, payload: ClientStatsPayload) {
         self.queue.push_back(payload);
     }
 
+    /// Returns a batch of trace stats payloads, subject to the max content size.
     pub fn get_batch(&mut self) -> Vec<ClientStatsPayload> {
         let mut batch_size = 0;
 

bottlecap/src/traces/stats_processor.rs

@@ -93,18 +93,18 @@ impl StatsProcessor for ServerlessStatsProcessor {
             return Ok((StatusCode::INTERNAL_SERVER_ERROR, error_msg).into_response());
         };
 
-        // send trace payload to our trace flusher
+        // send trace stats payload to our stats aggregator
         match tx.send(stats).await {
             Ok(()) => {
-                debug!("Successfully buffered stats to be flushed.");
+                debug!("Successfully buffered stats to be aggregated.");
                 Ok((
                     StatusCode::ACCEPTED,
-                    "Successfully buffered stats to be flushed.",
+                    "Successfully buffered stats to be aggregated.",
                 )
                     .into_response())
             }
             Err(err) => {
-                let error_msg = format!("Error sending stats to the stats flusher: {err}");
+                let error_msg = format!("Error sending stats to the stats aggregator: {err}");
                 error!("{}", error_msg);
                 Ok((StatusCode::INTERNAL_SERVER_ERROR, error_msg).into_response())
             }

bottlecap/src/traces/trace_agent.rs

@@ -119,15 +119,13 @@ impl TraceAgent {
         invocation_processor: Arc<Mutex<InvocationProcessor>>,
         tags_provider: Arc<provider::Provider>,
     ) -> TraceAgent {
-        // setup a channel to send processed traces to our flusher. tx is passed through each
+        // Set up a channel to send processed traces to our trace aggregator. tx is passed through each
         // endpoint_handler to the trace processor, which uses it to send de-serialized
-        // processed trace payloads to our trace flusher.
+        // processed trace payloads to our trace aggregator.
         let (trace_tx, mut trace_rx): (Sender<SendDataBuilderInfo>, Receiver<SendDataBuilderInfo>) =
             mpsc::channel(TRACER_PAYLOAD_CHANNEL_BUFFER_SIZE);
 
-        // start our trace flusher. receives trace payloads and handles buffering + deciding when to
-        // flush to backend.
-
+        // Start the trace aggregator, which receives and buffers trace payloads to be consumed by the trace flusher.
         tokio::spawn(async move {
             while let Some(tracer_payload_info) = trace_rx.recv().await {
                 let mut aggregator = trace_aggregator.lock().await;
@@ -151,13 +149,13 @@ impl TraceAgent {
     pub async fn start(&self) -> Result<(), Box<dyn std::error::Error>> {
         let now = Instant::now();
 
-        // channels to send processed stats to our stats flusher.
+        // Set up a channel to send processed stats to our stats aggregator.
         let (stats_tx, mut stats_rx): (
             Sender<pb::ClientStatsPayload>,
             Receiver<pb::ClientStatsPayload>,
         ) = mpsc::channel(STATS_PAYLOAD_CHANNEL_BUFFER_SIZE);
 
-        // Receive stats payload and send it to the aggregator
+        // Start the stats aggregator, which receives and buffers stats payloads to be consumed by the stats flusher.
         let stats_aggregator = self.stats_aggregator.clone();
         tokio::spawn(async move {
             while let Some(stats_payload) = stats_rx.recv().await {
@@ -500,12 +498,12 @@ impl TraceAgent {
             )
             .await;
 
-        // send trace payload to our trace flusher
+        // send trace payload to our trace aggregator
         match trace_tx.send(send_data).await {
-            Ok(()) => success_response("Successfully buffered traces to be flushed."),
+            Ok(()) => success_response("Successfully buffered traces to be aggregated."),
             Err(err) => error_response(
                 StatusCode::INTERNAL_SERVER_ERROR,
-                format!("Error sending traces to the trace flusher: {err}"),
+                format!("Error sending traces to the trace aggregator: {err}"),
             ),
         }
     }

bottlecap/src/traces/trace_aggregator.rs

@@ -20,6 +20,7 @@ impl SendDataBuilderInfo {
     }
 }
 
+/// Takes in individual trace payloads and aggregates them into batches to be flushed to Datadog.
 #[allow(clippy::module_name_repetitions)]
 pub struct TraceAggregator {
     queue: VecDeque<SendDataBuilderInfo>,
@@ -48,10 +49,12 @@ impl TraceAggregator {
         }
     }
 
+    /// Takes in an individual trace payload.
     pub fn add(&mut self, payload_info: SendDataBuilderInfo) {
         self.queue.push_back(payload_info);
     }
 
+    /// Returns a batch of trace payloads, subject to the max content size.
     pub fn get_batch(&mut self) -> Vec<SendDataBuilder> {
         let mut batch_size = 0;