From 25551f6a66895115170a02abd21b50903dd5a77f Mon Sep 17 00:00:00 2001
From: Igor Novgorodov <igor.novgorodov@dfinity.org>
Date: Fri, 20 Sep 2024 12:15:19 +0200
Subject: [PATCH 1/6] remove domain from http metrics

---
 src/metrics/mod.rs | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/src/metrics/mod.rs b/src/metrics/mod.rs
index d19e631..c8a4db2 100644
--- a/src/metrics/mod.rs
+++ b/src/metrics/mod.rs
@@ -105,7 +105,6 @@ impl HttpMetrics {
             "tls",
             "method",
             "http",
-            "domain",
             "status",
             "error",
             "cache_status",
@@ -313,7 +312,6 @@ pub async fn middleware(
             tls_version,
             method,
             http_version,
-            &domain,
             &status.to_string(),
             &error_cause,
             cache_status_str,
@@ -407,7 +405,7 @@ pub async fn middleware(
                 http_version,
                 request_type: request_type.clone(),
                 status,
-                domain: domain.clone(),
+                domain,
                 host: host.into(),
                 path: path.into(),
                 canister_id: canister_id.clone(),

From b21ddc8663efb0bcd80f26d16e87aa8f6da3703c Mon Sep 17 00:00:00 2001
From: Igor Novgorodov <igor.novgorodov@dfinity.org>
Date: Fri, 20 Sep 2024 13:58:10 +0200
Subject: [PATCH 2/6] enable verification, add metrics, vector logging debug

---
 src/metrics/vector.rs                   |  17 +--
 src/tls/cert/providers/issuer/mod.rs    | 169 ++++++++++++------------
 src/tls/cert/providers/issuer/verify.rs |   2 -
 src/tls/mod.rs                          |   1 +
 4 files changed, 94 insertions(+), 95 deletions(-)

diff --git a/src/metrics/vector.rs b/src/metrics/vector.rs
index d252619..9ab0474 100644
--- a/src/metrics/vector.rs
+++ b/src/metrics/vector.rs
@@ -26,7 +26,7 @@ use tokio_util::{
     sync::CancellationToken,
     task::TaskTracker,
 };
-use tracing::{info, warn};
+use tracing::{debug, warn};
 use url::Url;
 use vector_lib::{codecs::encoding::NativeSerializer, config::LogNamespace, event::Event};
 
@@ -310,10 +310,10 @@ impl Batcher {
         let batch = self.batch.clone().freeze();
 
         let start = Instant::now();
-        info!("{self}: queueing batch (len {})", batch.len());
+        debug!("{self}: queueing batch (len {})", batch.len());
         // In our case the Batcher is dropped before the Flusher, so no error can occur
         let _ = self.tx.send(batch).await;
-        info!("{self}: batch queued in {}s", start.elapsed().as_secs_f64());
+        debug!("{self}: batch queued in {}s", start.elapsed().as_secs_f64());
         self.metrics.buffer_batch_size.inc();
         self.batch.clear();
     }
@@ -336,7 +336,6 @@ impl Batcher {
     }
 
     async fn run(mut self) {
-        warn!("{self}: started");
         loop {
             select! {
                 biased;
@@ -427,7 +426,7 @@ impl Flusher {
 
         loop {
             let start = Instant::now();
-            info!(
+            debug!(
                 "{self}: sending batch (raw size {raw_size}, compressed {}, retry {})",
                 batch.len(),
                 RETRY_COUNT - retries + 1
@@ -444,7 +443,7 @@ impl Flusher {
                 self.metrics.sent_compressed.inc_by(batch.len() as u64);
                 self.metrics.batch_flushes.with_label_values(&["yes"]).inc();
 
-                info!("{self}: batch sent in {}s", start.elapsed().as_secs_f64());
+                debug!("{self}: batch sent in {}s", start.elapsed().as_secs_f64());
                 return Ok(());
             }
 
@@ -476,8 +475,6 @@ impl Flusher {
     }
 
     async fn run(self) {
-        warn!("{self} started");
-
         loop {
             select! {
                 biased;
@@ -495,13 +492,13 @@ impl Flusher {
 
                 Ok(batch) = self.rx.recv() => {
                     self.metrics.buffer_batch_size.dec();
-                    info!("{self}: received batch (len {})", batch.len());
+                    debug!("{self}: received batch (len {})", batch.len());
 
                     if let Err(e) = self.flush(batch).await {
                         warn!("{self}: unable to flush: {e:#}");
                     };
 
-                    info!("{self}: received batch flushed");
+                    debug!("{self}: received batch flushed");
                 }
             }
         }
diff --git a/src/tls/cert/providers/issuer/mod.rs b/src/tls/cert/providers/issuer/mod.rs
index ab0f4d5..38a3b2e 100644
--- a/src/tls/cert/providers/issuer/mod.rs
+++ b/src/tls/cert/providers/issuer/mod.rs
@@ -13,6 +13,10 @@ use candid::Principal;
 use fqdn::FQDN;
 use ic_bn_lib::{http, tasks::Run};
 use mockall::automock;
+use prometheus::{
+    register_histogram_vec_with_registry, register_int_gauge_vec_with_registry, HistogramVec,
+    IntGaugeVec, Registry,
+};
 use reqwest::{Method, Request, StatusCode, Url};
 use serde::Deserialize;
 use tokio::select;
@@ -20,7 +24,7 @@ use tokio_util::sync::CancellationToken;
 use tracing::{info, warn};
 
 use crate::routing::domain::{CustomDomain, ProvidesCustomDomains};
-use verify::{Verify, VerifyError, WithVerify};
+use verify::{Parser, Verifier, Verify, VerifyError};
 
 use super::{Pem, ProvidesCertificates};
 
@@ -33,6 +37,44 @@ pub enum Error {
     VerificationError(#[from] VerifyError),
 }
 
+#[derive(Debug, Clone)]
+pub struct Metrics {
+    packages: IntGaugeVec,
+    errors: IntGaugeVec,
+    duration: HistogramVec,
+}
+
+impl Metrics {
+    pub fn new(registry: &Registry) -> Self {
+        Self {
+            packages: register_int_gauge_vec_with_registry!(
+                format!("issuer_packages_loaded"),
+                format!("Number of packages in the current snapshot"),
+                &["issuer"],
+                registry
+            )
+            .unwrap(),
+
+            errors: register_int_gauge_vec_with_registry!(
+                format!("issuer_packages_errors"),
+                format!("Number of packages with errors"),
+                &["issuer"],
+                registry
+            )
+            .unwrap(),
+
+            duration: register_histogram_vec_with_registry!(
+                format!("issuer_fetch_duration_sec"),
+                format!("Time it takes to fetch a package in seconds"),
+                &["issuer"],
+                vec![0.2, 0.4, 0.8, 1.6, 3.2, 6.4, 12.8, 25.6],
+                registry
+            )
+            .unwrap(),
+        }
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
 pub struct Pair(
     pub Vec<u8>, // Private Key
@@ -46,6 +88,12 @@ pub struct Package {
     pub pair: Pair,
 }
 
+impl std::fmt::Display for Package {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "{} ({})", self.name, self.canister)
+    }
+}
+
 #[automock]
 #[async_trait]
 pub trait Import: Sync + Send {
@@ -57,6 +105,8 @@ pub struct CertificatesImporter {
     exporter_url: Url,
     poll_interval: Duration,
     snapshot: ArcSwapOption<Vec<Package>>,
+    verifier: Verifier<Parser>,
+    metrics: Metrics,
 }
 
 impl std::fmt::Debug for CertificatesImporter {
@@ -70,6 +120,7 @@ impl CertificatesImporter {
         http_client: Arc<dyn http::Client>,
         mut exporter_url: Url,
         poll_interval: Duration,
+        registry: &Registry,
     ) -> Self {
         exporter_url.set_path("");
         let exporter_url = exporter_url.join("/certificates").unwrap();
@@ -79,18 +130,49 @@ impl CertificatesImporter {
             exporter_url,
             poll_interval,
             snapshot: ArcSwapOption::empty(),
+            verifier: Verifier(Parser),
+            metrics: Metrics::new(registry),
         }
     }
 
     async fn refresh(&self) -> Result<(), Error> {
         let start = Instant::now();
         let packages = self.import().await.context("unable to fetch packages")?;
+        let len_full = packages.len();
+
+        let packages = packages
+            .into_iter()
+            .filter(|x| match self.verifier.verify(x) {
+                Ok(_) => true,
+                Err(e) => {
+                    warn!("{self:?}: package '{x}' verification failed, skipping: {e:#}");
+                    false
+                }
+            })
+            .collect::<Vec<_>>();
+
         info!(
-            "{self:?}: {} certs loaded in {}s",
+            "{self:?}: {} certs loaded ({} skipped due to errors) in {}s",
             packages.len(),
+            len_full - packages.len(),
             start.elapsed().as_secs_f64()
         );
 
+        let id = format!("{self:?}");
+        let labels = &[id.as_str()];
+        self.metrics
+            .packages
+            .with_label_values(labels)
+            .set(packages.len() as i64);
+        self.metrics
+            .errors
+            .with_label_values(labels)
+            .set((len_full - packages.len()) as i64);
+        self.metrics
+            .duration
+            .with_label_values(labels)
+            .observe(start.elapsed().as_secs_f64());
+
         self.snapshot.store(Some(Arc::new(packages)));
         Ok(())
     }
@@ -194,23 +276,6 @@ impl Run for CertificatesImporter {
     }
 }
 
-// Wraps an importer with a verifier
-// The importer imports a set of packages as usual, but then passes the packages to the verifier.
-// The verifier parses out the public certificate and compares the common name to the name in the package to make sure they match.
-// This should help eliminate risk of the replica returning a malicious package.
-#[async_trait]
-impl<T: Import, V: Verify> Import for WithVerify<T, V> {
-    async fn import(&self) -> Result<Vec<Package>, Error> {
-        let pkgs = self.0.import().await?;
-
-        for pkg in &pkgs {
-            self.1.verify(pkg)?;
-        }
-
-        Ok(pkgs)
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -222,8 +287,6 @@ mod tests {
     use reqwest::Body;
     use std::{str::FromStr, sync::Arc};
 
-    use crate::tls::cert::providers::issuer::verify::MockVerify;
-
     #[tokio::test]
     async fn import_ok() -> Result<(), AnyhowError> {
         let mut http_client = MockClient::new();
@@ -256,6 +319,7 @@ mod tests {
             Arc::new(http_client),
             Url::from_str("http://foo")?,
             Duration::ZERO,
+            &Registry::new(),
         );
 
         let out = importer.import().await?;
@@ -273,30 +337,7 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn import_verify_multiple() {
-        let mut verifier = MockVerify::new();
-        verifier
-            .expect_verify()
-            .times(3)
-            .with(predicate::in_iter(vec![
-                Package {
-                    name: "name-1".into(),
-                    canister: Principal::from_text("aaaaa-aa").unwrap(),
-                    pair: Pair(vec![], vec![]),
-                },
-                Package {
-                    name: "name-2".into(),
-                    canister: Principal::from_text("aaaaa-aa").unwrap(),
-                    pair: Pair(vec![], vec![]),
-                },
-                Package {
-                    name: "name-3".into(),
-                    canister: Principal::from_text("aaaaa-aa").unwrap(),
-                    pair: Pair(vec![], vec![]),
-                },
-            ]))
-            .returning(|_| Ok(()));
-
+    async fn import_multiple() {
         let mut importer = MockImport::new();
         importer.expect_import().times(1).returning(|| {
             Ok(vec![
@@ -318,47 +359,9 @@ mod tests {
             ])
         });
 
-        let importer = WithVerify(importer, verifier);
-
         match importer.import().await {
             Ok(_) => {}
             other => panic!("expected Ok but got {other:?}"),
         }
     }
-
-    #[tokio::test]
-    async fn import_verify_mismatch() {
-        let mut verifier = MockVerify::new();
-        verifier
-            .expect_verify()
-            .times(1)
-            .with(predicate::eq(Package {
-                name: "name-1".into(),
-                canister: Principal::from_text("aaaaa-aa").unwrap(),
-                pair: Pair(vec![], vec![]),
-            }))
-            .returning(|_| {
-                // Mock an error
-                Err(VerifyError::CommonNameMismatch(
-                    "name-1".into(),
-                    "name-2".into(),
-                ))
-            });
-
-        let mut importer = MockImport::new();
-        importer.expect_import().times(1).returning(|| {
-            Ok(vec![Package {
-                name: "name-1".into(),
-                canister: Principal::from_text("aaaaa-aa").unwrap(),
-                pair: Pair(vec![], vec![]),
-            }])
-        });
-
-        let importer = WithVerify(importer, verifier);
-
-        match importer.import().await {
-            Err(Error::VerificationError(_)) => {}
-            other => panic!("expected VerificationError but got {other:?}"),
-        }
-    }
 }
diff --git a/src/tls/cert/providers/issuer/verify.rs b/src/tls/cert/providers/issuer/verify.rs
index 007d860..f38637d 100644
--- a/src/tls/cert/providers/issuer/verify.rs
+++ b/src/tls/cert/providers/issuer/verify.rs
@@ -90,8 +90,6 @@ impl<P: Parse> Verify for Verifier<P> {
     }
 }
 
-pub struct WithVerify<T, V>(pub T, pub V);
-
 #[cfg(test)]
 mod tests {
     use candid::Principal;
diff --git a/src/tls/mod.rs b/src/tls/mod.rs
index 86f27d4..5c7f593 100644
--- a/src/tls/mod.rs
+++ b/src/tls/mod.rs
@@ -159,6 +159,7 @@ pub async fn setup(
             http_client.clone(),
             v.clone(),
             cli.cert.cert_provider_issuer_poll_interval,
+            registry,
         ));
 
         cert_providers.push(issuer.clone());

From 9b89723cb692a3b16d32759c6d43f0ada42a1fce Mon Sep 17 00:00:00 2001
From: Igor Novgorodov <igor.novgorodov@dfinity.org>
Date: Fri, 20 Sep 2024 14:00:52 +0200
Subject: [PATCH 3/6] fix metrics

---
 src/tls/cert/providers/issuer/mod.rs | 4 ++--
 src/tls/mod.rs                       | 3 ++-
 2 files changed, 4 insertions(+), 3 deletions(-)

diff --git a/src/tls/cert/providers/issuer/mod.rs b/src/tls/cert/providers/issuer/mod.rs
index 38a3b2e..4a1b07b 100644
--- a/src/tls/cert/providers/issuer/mod.rs
+++ b/src/tls/cert/providers/issuer/mod.rs
@@ -120,7 +120,7 @@ impl CertificatesImporter {
         http_client: Arc<dyn http::Client>,
         mut exporter_url: Url,
         poll_interval: Duration,
-        registry: &Registry,
+        metrics: Metrics,
     ) -> Self {
         exporter_url.set_path("");
         let exporter_url = exporter_url.join("/certificates").unwrap();
@@ -131,7 +131,7 @@ impl CertificatesImporter {
             poll_interval,
             snapshot: ArcSwapOption::empty(),
             verifier: Verifier(Parser),
-            metrics: Metrics::new(registry),
+            metrics,
         }
     }
 
diff --git a/src/tls/mod.rs b/src/tls/mod.rs
index 5c7f593..74b1b31 100644
--- a/src/tls/mod.rs
+++ b/src/tls/mod.rs
@@ -154,12 +154,13 @@ pub async fn setup(
 
     // Create CertIssuer providers
     // It's a custom domain & cert provider at the same time.
+    let issuer_metrics = providers::issuer::Metrics::new(registry);
     for v in &cli.cert.cert_provider_issuer_url {
         let issuer = Arc::new(providers::Issuer::new(
             http_client.clone(),
             v.clone(),
             cli.cert.cert_provider_issuer_poll_interval,
-            registry,
+            issuer_metrics.clone(),
         ));
 
         cert_providers.push(issuer.clone());

From 70d1b1009dad717cfc507a8859092da00b45eaf2 Mon Sep 17 00:00:00 2001
From: Igor Novgorodov <igor.novgorodov@dfinity.org>
Date: Fri, 20 Sep 2024 14:02:36 +0200
Subject: [PATCH 4/6] fix test

---
 src/tls/cert/providers/issuer/mod.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/tls/cert/providers/issuer/mod.rs b/src/tls/cert/providers/issuer/mod.rs
index 4a1b07b..004c5a3 100644
--- a/src/tls/cert/providers/issuer/mod.rs
+++ b/src/tls/cert/providers/issuer/mod.rs
@@ -319,7 +319,7 @@ mod tests {
             Arc::new(http_client),
             Url::from_str("http://foo")?,
             Duration::ZERO,
-            &Registry::new(),
+            Metrics::new(&Registry::new()),
         );
 
         let out = importer.import().await?;

From 2e9596b516c10411f0053bb4edaef08a1a43bc7b Mon Sep 17 00:00:00 2001
From: Igor Novgorodov <igor.novgorodov@dfinity.org>
Date: Fri, 20 Sep 2024 23:07:28 +0200
Subject: [PATCH 5/6] make vector retries & drain correct

---
 src/cli.rs            | 12 ++++++
 src/metrics/vector.rs | 90 ++++++++++++++++++++++++++++++++++---------
 2 files changed, 83 insertions(+), 19 deletions(-)

diff --git a/src/cli.rs b/src/cli.rs
index 637daa5..a06ebe6 100644
--- a/src/cli.rs
+++ b/src/cli.rs
@@ -433,6 +433,12 @@ pub struct Vector {
     #[clap(env, long, default_value = "16MB", value_parser = parse_size_usize)]
     pub log_vector_batch: usize,
 
+    /// Number of batches to store in the queue for the Flushers to consume.
+    /// So the maximum memory occupied by batch queue will be roughly
+    /// log_vector_batch*log_vector_batch_count, plus anything in the event queue (log_vector_buffer).
+    #[clap(env, long, default_value = "32")]
+    pub log_vector_batch_count: usize,
+
     /// Vector batch flush interval
     #[clap(env, long, default_value = "5s", value_parser = parse_duration)]
     pub log_vector_interval: Duration,
@@ -452,6 +458,12 @@ pub struct Vector {
     #[clap(env, long, default_value = "30s", value_parser = parse_duration)]
     pub log_vector_timeout: Duration,
 
+    /// Vector HTTP request retry interval
+    /// With each retry it will be linearly increased until it reaches 5x
+    /// E.g. for 2s the retry intervals will be 2s/4s/6s/8s/10s/10s/10s...
+    #[clap(env, long, default_value = "2s", value_parser = parse_duration)]
+    pub log_vector_retry_interval: Duration,
+
     /// ZSTD compression level to use when sending data
     #[clap(env, long, default_value = "3")]
     pub log_vector_zstd_level: usize,
diff --git a/src/metrics/vector.rs b/src/metrics/vector.rs
index 9ab0474..cd9bc86 100644
--- a/src/metrics/vector.rs
+++ b/src/metrics/vector.rs
@@ -32,8 +32,6 @@ use vector_lib::{codecs::encoding::NativeSerializer, config::LogNamespace, event
 
 use crate::cli;
 
-const RETRY_INTERVAL: Duration = Duration::from_millis(200);
-const RETRY_COUNT: usize = 5;
 const CONTENT_ENCODING_ZSTD: HeaderValue = HeaderValue::from_static("zstd");
 
 #[derive(Clone)]
@@ -159,6 +157,7 @@ impl EventEncoder {
 pub struct Vector {
     token_batcher: CancellationToken,
     token_flushers: CancellationToken,
+    token_flushers_drain: CancellationToken,
     tracker_batcher: TaskTracker,
     tracker_flushers: TaskTracker,
     tx: mpsc::Sender<Event>,
@@ -170,7 +169,7 @@ impl Vector {
         let cli = cli.clone();
 
         let (tx_event, rx_event) = mpsc::channel(cli.log_vector_buffer);
-        let (tx_batch, rx_batch) = bounded(64);
+        let (tx_batch, rx_batch) = bounded(cli.log_vector_batch_count);
 
         let metrics = Metrics::new(registry);
 
@@ -202,6 +201,7 @@ impl Vector {
 
         // Start flushers
         let token_flushers = CancellationToken::new();
+        let token_flushers_drain = CancellationToken::new();
         let tracker_flushers = TaskTracker::new();
 
         // Prepare auth header
@@ -219,6 +219,8 @@ impl Vector {
                 auth: auth.clone(),
                 zstd_level: cli.log_vector_zstd_level,
                 token: token_flushers.child_token(),
+                token_drain: token_flushers_drain.child_token(),
+                retry_interval: cli.log_vector_retry_interval,
                 timeout: cli.log_vector_timeout,
                 metrics: metrics.clone(),
             };
@@ -231,6 +233,7 @@ impl Vector {
         Self {
             token_batcher,
             token_flushers,
+            token_flushers_drain,
             tracker_batcher,
             tracker_flushers,
             tx: tx_event,
@@ -251,6 +254,9 @@ impl Vector {
     }
 
     pub async fn stop(&self) {
+        // Signal the flushers to limit the retries first
+        self.token_flushers_drain.cancel();
+
         warn!("Vector: shutting down batcher");
         self.token_batcher.cancel();
         self.tracker_batcher.close();
@@ -364,11 +370,13 @@ struct Flusher {
     id: usize,
     rx: Receiver<Bytes>,
     client: Arc<dyn http::Client>,
+    retry_interval: Duration,
     timeout: Duration,
     url: Url,
     auth: Option<HeaderValue>,
     zstd_level: usize,
     token: CancellationToken,
+    token_drain: CancellationToken,
     metrics: Metrics,
 }
 
@@ -420,23 +428,24 @@ impl Flusher {
         let batch = Bytes::from(batch);
 
         // Retry
-        // TODO make configurable?
-        let mut interval = RETRY_INTERVAL;
-        let mut retries = RETRY_COUNT;
+        let mut interval = self.retry_interval;
+        let mut retries = 1;
 
         loop {
             let start = Instant::now();
             debug!(
                 "{self}: sending batch (raw size {raw_size}, compressed {}, retry {})",
                 batch.len(),
-                RETRY_COUNT - retries + 1
+                retries
             );
 
             // Bytes is cheap to clone
             if let Err(e) = self.send(batch.clone()).await {
+                self.metrics.batch_flushes.with_label_values(&["no"]).inc();
                 warn!(
-                    "{self}: unable to send (try {}): {e:#}",
-                    RETRY_COUNT - retries + 1
+                    "{self}: unable to send (try {}, retry interval {}s): {e:#}",
+                    retries,
+                    interval.as_secs_f64()
                 );
             } else {
                 self.metrics.sent.inc_by(raw_size as u64);
@@ -447,18 +456,21 @@ impl Flusher {
                 return Ok(());
             }
 
-            // Back off a bit
-            retries -= 1;
-            interval *= 2;
-            if retries == 0 {
+            // Back off a bit until some limit
+            interval = (interval + self.retry_interval).min(self.retry_interval * 5);
+
+            self.metrics.batch_flush_retries.inc();
+            retries += 1;
+
+            // Limit the retry count if we're draining.
+            // Otherwise we wouldn't be able to stop with dead endpoint.
+            if self.token_drain.is_cancelled() && retries > 5 {
                 break;
             }
 
-            self.metrics.batch_flush_retries.inc();
             sleep(interval).await;
         }
 
-        self.metrics.batch_flushes.with_label_values(&["no"]).inc();
         Err(anyhow!("unable to flush batch: retries exhausted"))
     }
 
@@ -539,6 +551,21 @@ mod test {
         }
     }
 
+    #[derive(Debug)]
+    struct TestClientDead;
+
+    #[async_trait]
+    impl http::Client for TestClientDead {
+        async fn execute(
+            &self,
+            _req: reqwest::Request,
+        ) -> Result<reqwest::Response, reqwest::Error> {
+            let mut resp = ::http::Response::new(vec![]);
+            *resp.status_mut() = ::http::StatusCode::SERVICE_UNAVAILABLE;
+            Ok(resp.into())
+        }
+    }
+
     #[test]
     fn test_encoder() {
         let mut encoder = EventEncoder::new();
@@ -558,9 +585,8 @@ mod test {
         );
     }
 
-    #[tokio::test]
-    async fn test_vector() {
-        let cli = cli::Vector {
+    fn make_cli() -> cli::Vector {
+        cli::Vector {
             log_vector_url: Some(Url::parse("http://127.0.0.1:1234").unwrap()),
             log_vector_user: None,
             log_vector_pass: None,
@@ -570,7 +596,14 @@ mod test {
             log_vector_timeout: Duration::from_secs(10),
             log_vector_flushers: 4,
             log_vector_zstd_level: 3,
-        };
+            log_vector_batch_count: 32,
+            log_vector_retry_interval: Duration::from_millis(1),
+        }
+    }
+
+    #[tokio::test]
+    async fn test_vector() {
+        let cli = make_cli();
 
         let client = Arc::new(TestClient(AtomicU64::new(0), AtomicU64::new(0)));
         let vector = Vector::new(&cli, client.clone(), &Registry::new());
@@ -589,4 +622,23 @@ mod test {
         assert_eq!(client.0.load(Ordering::SeqCst), 131);
         assert_eq!(client.1.load(Ordering::SeqCst), 197780); // Uncompressed size
     }
+
+    /// Make sure we can drain when the endpoint is down
+    #[tokio::test]
+    async fn test_vector_drain() {
+        let cli = make_cli();
+
+        let client = Arc::new(TestClientDead);
+        let vector = Vector::new(&cli, client, &Registry::new());
+
+        for i in 0..6000 {
+            let event = json!({
+                format!("foo{i}"): format!("bar{i}"),
+            });
+
+            vector.send(event.clone());
+        }
+
+        vector.stop().await;
+    }
 }

From 56441a6e3758ee27d86ef60a75c3ea57e52adbd7 Mon Sep 17 00:00:00 2001
From: Igor Novgorodov <igor.novgorodov@dfinity.org>
Date: Fri, 20 Sep 2024 23:08:07 +0200
Subject: [PATCH 6/6] deps

---
 Cargo.lock | 110 +++++++++++++++++++++++++++--------------------------
 1 file changed, 56 insertions(+), 54 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 2818db7..6ab02a6 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -428,7 +428,7 @@ dependencies = [
  "log",
  "rustls-pki-types",
  "thiserror",
- "webpki-roots 0.26.5",
+ "webpki-roots 0.26.6",
 ]
 
 [[package]]
@@ -457,9 +457,9 @@ dependencies = [
 
 [[package]]
 name = "aws-lc-sys"
-version = "0.21.1"
+version = "0.21.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "234314bd569802ec87011d653d6815c6d7b9ffb969e9fee5b8b20ef860e8dce9"
+checksum = "b3ddc4a5b231dd6958b140ff3151b6412b3f4321fab354f399eec8f14b06df62"
 dependencies = [
  "bindgen",
  "cc",
@@ -500,12 +500,12 @@ dependencies = [
 
 [[package]]
 name = "axum"
-version = "0.7.5"
+version = "0.7.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3a6c9af12842a67734c9a2e355436e5d03b22383ed60cf13cd0c18fbfe3dcbcf"
+checksum = "8f43644eed690f5374f1af436ecd6aea01cd201f6fbdf0178adaf6907afb2cec"
 dependencies = [
  "async-trait",
- "axum-core 0.4.3",
+ "axum-core 0.4.4",
  "bytes",
  "futures-util",
  "http 1.1.0",
@@ -526,7 +526,7 @@ dependencies = [
  "serde_urlencoded",
  "sync_wrapper 1.0.1",
  "tokio",
- "tower 0.4.13",
+ "tower 0.5.1",
  "tower-layer",
  "tower-service",
  "tracing",
@@ -551,9 +551,9 @@ dependencies = [
 
 [[package]]
 name = "axum-core"
-version = "0.4.3"
+version = "0.4.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a15c63fd72d41492dc4f497196f5da1fb04fb7529e631d73630d1b491e47a2e3"
+checksum = "5e6b8ba012a258d63c9adfa28b9ddcf66149da6f986c5b5452e629d5ee64bf00"
 dependencies = [
  "async-trait",
  "bytes",
@@ -564,7 +564,7 @@ dependencies = [
  "mime",
  "pin-project-lite",
  "rustversion",
- "sync_wrapper 0.1.2",
+ "sync_wrapper 1.0.1",
  "tower-layer",
  "tower-service",
  "tracing",
@@ -572,12 +572,12 @@ dependencies = [
 
 [[package]]
 name = "axum-extra"
-version = "0.9.3"
+version = "0.9.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0be6ea09c9b96cb5076af0de2e383bd2bc0c18f827cf1967bdd353e0b910d733"
+checksum = "73c3220b188aea709cf1b6c5f9b01c3bd936bb08bd2b5184a12b35ac8131b1f9"
 dependencies = [
- "axum 0.7.5",
- "axum-core 0.4.3",
+ "axum 0.7.6",
+ "axum-core 0.4.4",
  "bytes",
  "futures-util",
  "http 1.1.0",
@@ -586,7 +586,7 @@ dependencies = [
  "mime",
  "pin-project-lite",
  "serde",
- "tower 0.4.13",
+ "tower 0.5.1",
  "tower-layer",
  "tower-service",
  "tracing",
@@ -909,9 +909,9 @@ checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b"
 
 [[package]]
 name = "bytes"
-version = "1.7.1"
+version = "1.7.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8318a53db07bb3f8dca91a600466bdb3f2eaadeedfdbcf02e1accbad9271ba50"
+checksum = "428d9aa8fbc0670b7b8d6030a7fadd0f86151cae55e4dbbece15f3780a3dfaf3"
 dependencies = [
  "serde",
 ]
@@ -1014,9 +1014,9 @@ dependencies = [
 
 [[package]]
 name = "cc"
-version = "1.1.20"
+version = "1.1.21"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "45bcde016d64c21da4be18b655631e5ab6d3107607e71a73a9f53eb48aae23fb"
+checksum = "07b1695e2c7e8fc85310cde85aeaab7e3097f593c91d209d3f9df76c928100f0"
 dependencies = [
  "jobserver",
  "libc",
@@ -1224,9 +1224,9 @@ dependencies = [
 
 [[package]]
 name = "clap"
-version = "4.5.17"
+version = "4.5.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3e5a21b8495e732f1b3c364c9949b201ca7bae518c502c80256c96ad79eaf6ac"
+checksum = "b0956a43b323ac1afaffc053ed5c4b7c1f1800bacd1683c353aabbb752515dd3"
 dependencies = [
  "clap_builder",
  "clap_derive",
@@ -1234,9 +1234,9 @@ dependencies = [
 
 [[package]]
 name = "clap_builder"
-version = "4.5.17"
+version = "4.5.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8cf2dd12af7a047ad9d6da2b6b249759a22a7abc0f474c1dae1777afa4b21a73"
+checksum = "4d72166dd41634086d5803a47eb71ae740e61d84709c36f3c34110173db3961b"
 dependencies = [
  "anstream",
  "anstyle",
@@ -1246,9 +1246,9 @@ dependencies = [
 
 [[package]]
 name = "clap_derive"
-version = "4.5.13"
+version = "4.5.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "501d359d5f3dcaf6ecdeee48833ae73ec6e42723a1e52419c79abf9507eec0a0"
+checksum = "4ac6a0c7b1a9e9a5186361f67dfa1b88213572f427fb9ab038efb2bd8c582dab"
 dependencies = [
  "heck 0.5.0",
  "proc-macro2",
@@ -2922,7 +2922,7 @@ dependencies = [
  "tokio",
  "tokio-rustls 0.26.0",
  "tower-service",
- "webpki-roots 0.26.5",
+ "webpki-roots 0.26.6",
 ]
 
 [[package]]
@@ -3047,7 +3047,7 @@ dependencies = [
  "anyhow",
  "arc-swap",
  "async-trait",
- "axum 0.7.5",
+ "axum 0.7.6",
  "backoff",
  "base64 0.22.1",
  "bytes",
@@ -3145,7 +3145,7 @@ dependencies = [
  "arc-swap",
  "async-channel",
  "async-trait",
- "axum 0.7.5",
+ "axum 0.7.6",
  "axum-extra",
  "backoff",
  "base64 0.22.1",
@@ -3214,7 +3214,7 @@ dependencies = [
  "url",
  "uuid",
  "vector-lib",
- "webpki-roots 0.26.5",
+ "webpki-roots 0.26.6",
  "x509-parser",
  "zstd",
 ]
@@ -3627,9 +3627,9 @@ checksum = "17ab85f84ca42c5ec520e6f3c9966ba1fd62909ce260f8837e248857d2560509"
 
 [[package]]
 name = "k256"
-version = "0.13.3"
+version = "0.13.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "956ff9b67e26e1a6a866cb758f12c6f8746208489e3e4a4b5580802f2f0a587b"
+checksum = "f6e3919bbaa2945715f0bb6d3934a173d1e9a59ac23767fbaaef277265a7411b"
 dependencies = [
  "cfg-if",
  "ecdsa",
@@ -4517,9 +4517,9 @@ checksum = "e3148f5046208a5d56bcfc03053e3ca6334e51da8dfb19b6cdc8b306fae3283e"
 
 [[package]]
 name = "pest"
-version = "2.7.12"
+version = "2.7.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9c73c26c01b8c87956cea613c907c9d6ecffd8d18a2a5908e5de0adfaa185cea"
+checksum = "fdbef9d1d47087a895abd220ed25eb4ad973a5e26f6a4367b038c25e28dfc2d9"
 dependencies = [
  "memchr",
  "thiserror",
@@ -4528,9 +4528,9 @@ dependencies = [
 
 [[package]]
 name = "pest_derive"
-version = "2.7.12"
+version = "2.7.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "664d22978e2815783adbdd2c588b455b1bd625299ce36b2a99881ac9627e6d8d"
+checksum = "4d3a6e3394ec80feb3b6393c725571754c6188490265c61aaf260810d6b95aa0"
 dependencies = [
  "pest",
  "pest_generator",
@@ -4538,9 +4538,9 @@ dependencies = [
 
 [[package]]
 name = "pest_generator"
-version = "2.7.12"
+version = "2.7.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a2d5487022d5d33f4c30d91c22afa240ce2a644e87fe08caad974d4eab6badbe"
+checksum = "94429506bde1ca69d1b5601962c73f4172ab4726571a59ea95931218cb0e930e"
 dependencies = [
  "pest",
  "pest_meta",
@@ -4551,9 +4551,9 @@ dependencies = [
 
 [[package]]
 name = "pest_meta"
-version = "2.7.12"
+version = "2.7.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0091754bbd0ea592c4deb3a122ce8ecbb0753b738aa82bc055fcc2eccc8d8174"
+checksum = "ac8a071862e93690b6e34e9a5fb8e33ff3734473ac0245b27232222c4906a33f"
 dependencies = [
  "once_cell",
  "pest",
@@ -4732,9 +4732,9 @@ dependencies = [
 
 [[package]]
 name = "portable-atomic"
-version = "1.7.0"
+version = "1.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "da544ee218f0d287a911e9c99a39a8c9bc8fcad3cb8db5959940044ecfc67265"
+checksum = "d30538d42559de6b034bc76fd6dd4c38961b1ee5c6c56e3808c50128fdbc22ce"
 
 [[package]]
 name = "powerfmt"
@@ -5516,7 +5516,7 @@ dependencies = [
  "wasm-bindgen-futures",
  "wasm-streams",
  "web-sys",
- "webpki-roots 0.26.5",
+ "webpki-roots 0.26.6",
  "windows-registry",
 ]
 
@@ -5728,7 +5728,7 @@ dependencies = [
  "serde",
  "serde_json",
  "thiserror",
- "webpki-roots 0.26.5",
+ "webpki-roots 0.26.6",
  "x509-parser",
 ]
 
@@ -5800,7 +5800,7 @@ dependencies = [
  "rustls-webpki 0.102.8",
  "security-framework",
  "security-framework-sys",
- "webpki-roots 0.26.5",
+ "webpki-roots 0.26.6",
  "winapi",
 ]
 
@@ -5954,9 +5954,9 @@ dependencies = [
 
 [[package]]
 name = "security-framework-sys"
-version = "2.11.1"
+version = "2.12.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "75da29fe9b9b08fe9d6b22b5b4bcbc75d8db3aa31e639aa56bb62e9d46bfceaf"
+checksum = "ea4a292869320c0272d7bc55a5a6aafaff59b4f63404a003887b679a2e05b4b6"
 dependencies = [
  "core-foundation-sys",
  "libc",
@@ -6957,7 +6957,7 @@ checksum = "c6f6ba989e4b2c58ae83d862d3a3e27690b6e3ae630d0deb59f3697f32aa88ad"
 dependencies = [
  "async-stream",
  "async-trait",
- "axum 0.7.5",
+ "axum 0.7.6",
  "base64 0.22.1",
  "bytes",
  "h2 0.4.6",
@@ -7009,8 +7009,10 @@ dependencies = [
  "futures-util",
  "pin-project-lite",
  "sync_wrapper 0.1.2",
+ "tokio",
  "tower-layer",
  "tower-service",
+ "tracing",
 ]
 
 [[package]]
@@ -7051,7 +7053,7 @@ version = "0.4.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "313fa625fea5790ed56360a30ea980e41229cf482b4835801a67ef1922bf63b9"
 dependencies = [
- "axum 0.7.5",
+ "axum 0.7.6",
  "forwarded-header-value",
  "governor",
  "http 1.1.0",
@@ -7317,15 +7319,15 @@ checksum = "f6ccf251212114b54433ec949fd6a7841275f9ada20dddd2f29e9ceea4501493"
 
 [[package]]
 name = "unicode-width"
-version = "0.1.13"
+version = "0.1.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0336d538f7abc86d282a4189614dfaa90810dfc2c6f6427eaf88e16311dd225d"
+checksum = "7dd6e30e90baa6f72411720665d41d89b9a3d039dc45b8faea1ddd07f617f6af"
 
 [[package]]
 name = "unicode-xid"
-version = "0.2.5"
+version = "0.2.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "229730647fbc343e3a80e463c1db7f78f3855d3f3739bee0dda773c9a037c90a"
+checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853"
 
 [[package]]
 name = "universal-hash"
@@ -7885,9 +7887,9 @@ checksum = "5f20c57d8d7db6d3b86154206ae5d8fba62dd39573114de97c2cb0578251f8e1"
 
 [[package]]
 name = "webpki-roots"
-version = "0.26.5"
+version = "0.26.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0bd24728e5af82c6c4ec1b66ac4844bdf8156257fccda846ec58b42cd0cdbe6a"
+checksum = "841c67bff177718f1d4dfefde8d8f0e78f9b6589319ba88312f567fc5841a958"
 dependencies = [
  "rustls-pki-types",
 ]