zstd configurable compr, retry fail early

src/cli.rs

@@ -440,6 +440,10 @@ pub struct Vector {
     /// Vector HTTP request timeout for a batch flush
     #[clap(env, long, default_value = "30s", value_parser = parse_duration)]
     pub log_vector_timeout: Duration,
+
+    /// ZSTD compression level to use when sending data
+    #[clap(env, long, default_value = "3")]
+    pub log_vector_zstd_level: usize,
 }
 
 #[derive(Args)]

src/metrics/vector.rs

@@ -217,6 +217,7 @@ impl Vector {
                 client: client.clone(),
                 url: cli.log_vector_url.clone().unwrap(),
                 auth: auth.clone(),
+                zstd_level: cli.log_vector_zstd_level,
                 token: token_flushers.child_token(),
                 timeout: cli.log_vector_timeout,
                 metrics: metrics.clone(),
@@ -367,6 +368,7 @@ struct Flusher {
     timeout: Duration,
     url: Url,
     auth: Option<HeaderValue>,
+    zstd_level: usize,
     token: CancellationToken,
     metrics: Metrics,
 }
@@ -414,15 +416,16 @@ impl Flusher {
     async fn flush(&self, batch: Bytes) -> Result<(), Error> {
         let raw_size = batch.len();
 
-        let batch = zstd::encode_all(batch.reader(), 3).context("unable to compress batch")?;
+        let batch = zstd::encode_all(batch.reader(), self.zstd_level as i32)
+            .context("unable to compress batch")?;
         let batch = Bytes::from(batch);
 
         // Retry
         // TODO make configurable?
         let mut interval = RETRY_INTERVAL;
         let mut retries = RETRY_COUNT;
 
-        while retries > 0 {
+        loop {
             let start = Instant::now();
             info!(
                 "{self}: sending batch (raw size {raw_size}, compressed {}, retry {})",
@@ -445,12 +448,15 @@ impl Flusher {
                 return Ok(());
             }
 
-            self.metrics.batch_flush_retries.inc();
-            sleep(interval).await;
-
             // Back off a bit
             retries -= 1;
             interval *= 2;
+            if retries == 0 {
+                break;
+            }
+
+            self.metrics.batch_flush_retries.inc();
+            sleep(interval).await;
         }
 
         self.metrics.batch_flushes.with_label_values(&["no"]).inc();
@@ -566,6 +572,7 @@ mod test {
             log_vector_interval: Duration::from_secs(100),
             log_vector_timeout: Duration::from_secs(10),
             log_vector_flushers: 4,
+            log_vector_zstd_level: 3,
         };
 
         let client = Arc::new(TestClient(AtomicU64::new(0), AtomicU64::new(0)));

src/routing/ic/transport.rs

@@ -120,7 +120,7 @@ impl ReqwestTransport {
         body: Option<Vec<u8>>,
     ) -> Result<(StatusCode, Vec<u8>), AgentError> {
         // Create the initial request with a fake URL which will be overridden later
-        let mut http_request = Request::new(method.clone(), Url::parse("http://foo").unwrap());
+        let mut http_request = Request::new(method.clone(), Url::parse("https://foo").unwrap());
 
         http_request
             .headers_mut()