ref(vector): restructure code (#219)

- [x] remove debug logs
- [x] move send_batch out of class
- [x] use a generic error for vector errors

---------

Co-authored-by: getsantry[bot] <66042841+getsantry[bot]@users.noreply.github.com>

Author: JoshFerge

src/producer.rs

@@ -16,10 +16,8 @@ pub enum ExtractCodeError {
     Producer(#[from] ProducerError),
     #[error(transparent)]
     Schema(#[from] SchemaError),
-    #[error(transparent)]
-    VectorRequestError(#[from] reqwest::Error),
-    #[error("Vector request failed with status: {0}")]
-    VectorRequestStatusError(reqwest::StatusCode),
+    #[error("Vector request failed")]
+    VectorError,
 }
 
 pub trait ResultsProducer: Send + Sync {

src/producer/vector_producer.rs

@@ -25,78 +25,31 @@ impl VectorResultsProducer {
         (Self { schema, sender }, worker)
     }
 
-    async fn send_batch(
-        batch: Vec<Vec<u8>>,
-        client: Client,
-        endpoint: String,
-    ) -> Result<(), ExtractCodeError> {
-        if batch.is_empty() {
-            return Ok(());
-        }
-
-        let body: String = batch
-            .iter()
-            .filter_map(|json| String::from_utf8(json.clone()).ok())
-            .map(|s| s + "\n")
-            .collect();
-
-        // Log the exact payload for debugging
-        tracing::debug!(%body, "payload.sending");
-        tracing::debug!(size = body.len(), "request.sending_to_vector");
-
-        let response = client
-            .post(&endpoint)
-            .header("Content-Type", "application/json")
-            .body(body)
-            .send()
-            .await;
-
-        match response {
-            Ok(_) => {
-                tracing::debug!("request.sent_successfully");
-                Ok(())
-            }
-            Err(e) => {
-                tracing::error!(error = ?e, "request.failed_to_vector");
-                Err(ExtractCodeError::VectorRequestStatusError(
-                    reqwest::StatusCode::INTERNAL_SERVER_ERROR,
-                ))
-            }
-        }
-    }
-
     fn spawn_worker(
         vector_batch_size: usize,
         client: Client,
         endpoint: String,
         mut receiver: UnboundedReceiver<Vec<u8>>,
     ) -> JoinHandle<()> {
         tokio::spawn(async move {
-            tracing::debug!(%endpoint, "worker.started");
             let mut batch = Vec::with_capacity(vector_batch_size);
 
             while let Some(json) = receiver.recv().await {
-                tracing::debug!(size = json.len(), "event.received");
                 batch.push(json);
-                tracing::debug!(size = batch.len(), "batch.size.updated");
 
                 if batch.len() < vector_batch_size {
                     continue;
                 }
 
-                tracing::debug!(size = batch.len(), "batch.sending");
                 let batch_to_send =
                     std::mem::replace(&mut batch, Vec::with_capacity(vector_batch_size));
-                if let Err(e) =
-                    Self::send_batch(batch_to_send, client.clone(), endpoint.clone()).await
-                {
-                    tracing::error!(error = ?e, "batch.send_failed");
+                if let Err(e) = send_batch(batch_to_send, client.clone(), endpoint.clone()).await {
+                    tracing::error!(error = ?e, "vector_batch.send_failed");
                 }
             }
 
             if !batch.is_empty() {
-                tracing::debug!(size = batch.len(), "final_batch.sending");
-                if let Err(e) = Self::send_batch(batch, client, endpoint).await {
+                if let Err(e) = send_batch(batch, client, endpoint).await {
                     tracing::error!(error = ?e, "final_batch.send_failed");
                 }
             }
@@ -114,15 +67,44 @@ impl ResultsProducer for VectorResultsProducer {
         // Send the serialized result to the worker task
         if self.sender.send(json).is_err() {
             tracing::error!("event.send_failed_channel_closed");
-            return Err(ExtractCodeError::VectorRequestStatusError(
-                reqwest::StatusCode::INTERNAL_SERVER_ERROR,
-            ));
+            return Err(ExtractCodeError::VectorError);
         }
 
         Ok(())
     }
 }
 
+async fn send_batch(
+    batch: Vec<Vec<u8>>,
+    client: Client,
+    endpoint: String,
+) -> Result<(), ExtractCodeError> {
+    if batch.is_empty() {
+        return Ok(());
+    }
+
+    let body: String = batch
+        .iter()
+        .filter_map(|json| String::from_utf8(json.clone()).ok())
+        .map(|s| s + "\n")
+        .collect();
+
+    let response = client
+        .post(&endpoint)
+        .header("Content-Type", "application/json")
+        .body(body)
+        .send()
+        .await;
+
+    match response {
+        Ok(_) => Ok(()),
+        Err(e) => {
+            tracing::error!(error = ?e, "request.failed_to_vector");
+            Err(ExtractCodeError::VectorError)
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::VectorResultsProducer;
@@ -170,7 +152,6 @@ mod tests {
     #[tokio::test]
     async fn test_single_event() {
         let mock_server = MockServer::start();
-        tracing::debug!("Mock server started at {}", mock_server.url("/"));
         let test_result = create_test_result();
 
         let mock = mock_server.mock(|when, then| {
@@ -179,16 +160,11 @@ mod tests {
                 .header("Content-Type", "application/json")
                 .matches(|req| {
                     if let Some(body) = &req.body {
-                        tracing::debug!(
-                            "Received request body: {:?}",
-                            String::from_utf8_lossy(body)
-                        );
                         let lines: Vec<_> = body
                             .split(|&b| b == b'\n')
                             .filter(|l| !l.is_empty())
                             .collect();
                         if lines.len() != 1 {
-                            tracing::debug!("Expected 1 line, got {}", lines.len());
                             return false;
                         }
                         if let Ok(value) = serde_json::from_slice::<serde_json::Value>(lines[0]) {
@@ -221,16 +197,14 @@ mod tests {
     #[tokio::test]
     async fn test_batch_events() {
         let mock_server = MockServer::start();
-        tracing::debug!("Mock server started at {}", mock_server.url("/"));
         let (producer, worker) = VectorResultsProducer::new(
             "uptime-results",
             mock_server.url("/").to_string(),
             TEST_BATCH_SIZE,
         );
 
         // Create and send BATCH_SIZE + 2 events
-        for i in 0..(10 + 2) {
-            tracing::debug!("Sending event {}", i + 1);
+        for _ in 0..(TEST_BATCH_SIZE + 2) {
             let test_result = create_test_result();
             let result = producer.produce_checker_result(&test_result);
             assert!(result.is_ok());
@@ -243,17 +217,12 @@ mod tests {
                 .header("Content-Type", "application/json")
                 .matches(|req| {
                     if let Some(body) = &req.body {
-                        tracing::debug!(
-                            "Received request body: {:?}",
-                            String::from_utf8_lossy(body)
-                        );
                         let lines: Vec<_> = body
                             .split(|&b| b == b'\n')
                             .filter(|l| !l.is_empty())
                             .collect();
                         let len = lines.len();
                         if len != 10 && len != 2 {
-                            tracing::debug!("Expected {} or 2 lines, got {}", 10, len);
                             return false;
                         }
                         // Verify each line is valid JSON with expected fields
@@ -289,7 +258,6 @@ mod tests {
     #[tokio::test]
     async fn test_server_error() {
         let mock_server = MockServer::start();
-        tracing::debug!("Mock server started at {}", mock_server.url("/"));
         let test_result = create_test_result();
 
         let error_mock = mock_server.mock(|when, then| {
@@ -298,16 +266,11 @@ mod tests {
                 .header("Content-Type", "application/json")
                 .matches(|req| {
                     if let Some(body) = &req.body {
-                        tracing::debug!(
-                            "Received request body: {:?}",
-                            String::from_utf8_lossy(body)
-                        );
                         let lines: Vec<_> = body
                             .split(|&b| b == b'\n')
                             .filter(|l| !l.is_empty())
                             .collect();
                         if lines.len() != 1 {
-                            tracing::debug!("Expected 1 line, got {}", lines.len());
                             return false;
                         }
                         if let Ok(value) = serde_json::from_slice::<serde_json::Value>(lines[0]) {
@@ -343,25 +306,19 @@ mod tests {
     #[tokio::test]
     async fn test_flush_on_shutdown() {
         let mock_server = MockServer::start();
-        tracing::debug!("Mock server started at {}", mock_server.url("/"));
         // Create a mock that expects a single request with less than BATCH_SIZE events
         let mock = mock_server.mock(|when, then| {
             when.method(Method::POST)
                 .path("/")
                 .header("Content-Type", "application/json")
                 .matches(|req| {
                     if let Some(body) = &req.body {
-                        tracing::debug!(
-                            "Received request body: {:?}",
-                            String::from_utf8_lossy(body)
-                        );
                         let lines: Vec<_> = body
                             .split(|&b| b == b'\n')
                             .filter(|l| !l.is_empty())
                             .collect();
                         // We expect only 1 event, which is less than BATCH_SIZE
                         if lines.len() != 1 {
-                            tracing::debug!("Expected 1 line, got {}", lines.len());
                             return false;
                         }
                         if let Ok(value) = serde_json::from_slice::<serde_json::Value>(lines[0]) {