Added retry mechanism for file uploads

Prevents zip files from being lost if the upload failed for whichever
reason.

Signed-off-by: Nathan Monfils <nathan.monfils@destiny.eu>

Author: Nathan Monfils

core-dump-agent/src/main.rs

@@ -94,7 +94,7 @@ async fn main() -> Result<(), anyhow::Error> {
             process_file(p, &bucket).await;
         } else {
             info!("Uploading all content in {}", core_dir_command);
-            run_polling_agent().await;
+            run_polling_agent(false).await;
         }
         process::exit(0);
     }
@@ -149,7 +149,7 @@ async fn main() -> Result<(), anyhow::Error> {
             std::thread::sleep(Duration::from_millis(1000));
         }
     } else {
-        run_polling_agent().await;
+        run_polling_agent(use_inotify == "true").await;
     }
 
     if !interval.is_empty() && !schedule.is_empty() {
@@ -190,7 +190,7 @@ async fn main() -> Result<(), anyhow::Error> {
                 match next_tick {
                     Ok(Some(ts)) => {
                         info!("Next scheduled run {:?}", ts);
-                        run_polling_agent().await;
+                        run_polling_agent(false).await;
                     }
                     _ => warn!("Could not get next tick for job"),
                 }
@@ -255,22 +255,14 @@ async fn main() -> Result<(), anyhow::Error> {
                         if event.mask.contains(EventMask::ISDIR) {
                             warn!("Unknown Directory created: {:?}", event.name);
                         } else {
-                            let bucket = match get_bucket() {
-                                Ok(v) => v,
-                                Err(e) => {
-                                    error!("Bucket creation failed in event: {}", e);
-                                    continue;
-                                }
-                            };
                             match event.name {
                                 Some(s) => {
                                     let file = format!(
                                         "{}/{}",
                                         core_dir_command,
                                         s.to_str().unwrap_or_default()
                                     );
-                                    let p = Path::new(&file);
-                                    process_file(p, &bucket).await
+                                    tokio::spawn(process_file_or_retry(PathBuf::from(file), 0));
                                 }
                                 None => {
                                     continue;
@@ -287,7 +279,30 @@ async fn main() -> Result<(), anyhow::Error> {
     Ok(())
 }
 
-async fn process_file(zip_path: &Path, bucket: &Bucket) {
+async fn process_file_or_retry(file: PathBuf, iteration: usize) {
+    let bucket = match get_bucket() {
+        Ok(v) => v,
+        Err(e) => {
+            error!("Bucket creation failed in event: {}", e);
+            return;
+        }
+    };
+
+    match process_file(&file, &bucket).await {
+        Ok(()) => (),
+        Err(e) => {
+            let backoff = Duration::from_secs(60).mul_f32((iteration as f32 + 1.0).powf(1.5));
+
+            error!("Core dump file processing failed: {e}. Retrying in {} s.", backoff.as_secs());
+            tokio::time::sleep(backoff).await;
+
+            Box::pin(process_file_or_retry(file, iteration + 1)).await;
+        }
+    }
+
+}
+
+async fn process_file(zip_path: &Path, bucket: &Bucket) -> Result<(), String> {
     info!("Uploading: {}", zip_path.display());
 
     let f = File::open(zip_path).expect("no file found");
@@ -395,7 +410,7 @@ fn get_bucket() -> Result<Box<Bucket>, anyhow::Error> {
     Ok(Bucket::new(&s3.bucket, s3.region, s3.credentials)?.with_path_style())
 }
 
-async fn run_polling_agent() {
+async fn run_polling_agent(retry: bool) {
     let core_location = env::var("CORE_DIR").unwrap_or_else(|_| DEFAULT_CORE_DIR.to_string());
     info!("Executing Agent with location : {}", core_location);
 
@@ -418,7 +433,16 @@ async fn run_polling_agent() {
 
     info!("Dir Content {:?}", paths);
     for zip_path in paths {
-        process_file(&zip_path, &bucket).await;
+        if retry {
+            process_file_or_retry(zip_path, 0).await;
+        } else {
+            match process_file(&zip_path, &bucket).await {
+                Ok(()) => (),
+                Err(e) => {
+                    error!("File processing failed: {e}");
+                }
+            };
+        }
     }
 }