Add retry mechanism (#102)

Author: toelo3

dsh_sdk/CHANGELOG.md

@@ -5,6 +5,10 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.4.11] -2024-09-30
+### Fixed
+- Retry mechanism for when PKI endpoint is not yet avaialble during rolling restart DSH ([#101](https://github.com/kpn-dsh/dsh-sdk-platform-rs/issues/101))
+
 ## [0.4.10] -2024-09-30
 ### Added
 - Add new with client methods to REST and MQTTT token fetcher

dsh_sdk/Cargo.toml

@@ -9,7 +9,7 @@ license.workspace = true
 name = "dsh_sdk"
 readme = 'README.md'
 repository.workspace = true
-version = "0.4.10"
+version = "0.4.11"
 
 [package.metadata.docs.rs]
 all-features = true

dsh_sdk/src/dsh/bootstrap.rs

@@ -24,7 +24,7 @@ pub(crate) fn bootstrap(
 ) -> Result<Cert, DshError> {
     let dsh_config = DshConfig::new(config_host, tenant_name, task_id)?;
     let client = reqwest_ca_client(dsh_config.dsh_ca_certificate.as_bytes())?;
-    let dn = DshBootstapCall::Dn(&dsh_config).perform_call(&client)?;
+    let dn = DshBootstapCall::Dn(&dsh_config).retryable_call(&client)?;
     let dn = Dn::parse_string(&dn)?;
     let certificates = Cert::get_signed_client_cert(dn, &dsh_config, &client)?;
     info!("Successfully connected to DSH");
@@ -115,7 +115,7 @@ impl DshBootstapCall<'_> {
         }
     }
 
-    pub(crate) fn perform_call(&self, client: &Client) -> Result<String, DshError> {
+    fn perform_call(&self, client: &Client) -> Result<String, DshError> {
         let response = self.request_builder(client).send()?;
         if !response.status().is_success() {
             return Err(DshError::DshCallError {
@@ -126,6 +126,28 @@ impl DshBootstapCall<'_> {
         }
         Ok(response.text()?)
     }
+
+    pub(crate) fn retryable_call(&self, client: &Client) -> Result<String, DshError> {
+        let mut retries = 0;
+        loop {
+            match self.perform_call(client) {
+                Ok(response) => return Ok(response),
+                Err(err) => {
+                    if retries >= 30 {
+                        return Err(err);
+                    }
+                    retries += 1;
+                    // sleep exponentially
+                    let sleep: u64 = std::cmp::min(2u64.pow(retries), 60);
+                    log::warn!(
+                        "Retrying call to DSH in {sleep} seconds due to error: {}",
+                        crate::error::report(&err)
+                    );
+                    std::thread::sleep(std::time::Duration::from_secs(sleep));
+                }
+            }
+        }
+    }
 }
 
 /// Struct to parse DN string into separate fields.

dsh_sdk/src/dsh/certificates.rs

@@ -71,7 +71,7 @@ impl Cert {
             config: dsh_config,
             csr: &csr.pem()?,
         }
-        .perform_call(client)?;
+        .retryable_call(client)?;
         let ca_cert = pem::parse_many(dsh_config.dsh_ca_certificate())?;
         let client_cert = pem::parse_many(client_certificate)?;
         Ok(Self::new(

dsh_sdk/src/dsh/properties.rs

@@ -129,24 +129,21 @@ impl Properties {
             }
         };
         let task_id = utils::get_env_var(VAR_TASK_ID).unwrap_or("local_task_id".to_string());
-        let config_host = utils::get_env_var(VAR_KAFKA_CONFIG_HOST)
-            .map(|host| format!("https://{}", host))
-            .unwrap_or_else(|_| {
-                warn!(
-                    "{} is not set, using default value {}",
-                    VAR_KAFKA_CONFIG_HOST, DEFAULT_CONFIG_HOST
-                );
-                DEFAULT_CONFIG_HOST.to_string()
-            });
+        let config_host =
+            utils::get_env_var(VAR_KAFKA_CONFIG_HOST).map(|host| format!("https://{}", host));
         let certificates = if let Ok(cert) = pki_config_dir::get_pki_cert() {
             Some(cert)
-        } else {
-            bootstrap(&config_host, &tenant_name, &task_id)
+        } else if let Ok(config_host) = &config_host {
+            bootstrap(config_host, &tenant_name, &task_id)
                 .inspect_err(|e| {
                     warn!("Could not bootstrap to DSH, due to: {}", e);
                 })
                 .ok()
+        } else {
+            warn!("Could not bootstrap to DSH, as it does not seem to be running on DSH due to missing enivironment variables");
+            None
         };
+        let config_host = config_host.unwrap_or(DEFAULT_CONFIG_HOST.to_string()); // Default is for running on local machine with VPN
         let fetched_datastreams = certificates.as_ref().and_then(|cert| {
             cert.reqwest_blocking_client_config()
                 .ok()

dsh_sdk/src/error.rs

@@ -1,5 +1,3 @@
-//! Error types for the DSH SDK
-
 use thiserror::Error;
 
 #[derive(Error, Debug)]
@@ -55,6 +53,15 @@ pub enum DshError {
     HyperError(#[from] hyper::http::Error),
 }
 
+pub(crate) fn report(mut err: &dyn std::error::Error) -> String {
+    let mut s = format!("{}", err);
+    while let Some(src) = err.source() {
+        s.push_str(&format!("\n\nCaused by: {}", src));
+        err = src;
+    }
+    s
+}
+
 #[cfg(feature = "rest-token-fetcher")]
 #[derive(Error, Debug)]
 #[non_exhaustive]