elasticsearch retries

Author: leoyvens

graph/src/log/elastic.rs

@@ -16,6 +16,8 @@ use serde_json::json;
 use slog::*;
 use slog_async;
 
+use crate::util::futures::retry;
+
 /// General configuration parameters for Elasticsearch logging.
 #[derive(Clone, Debug)]
 pub struct ElasticLoggingConfig {
@@ -140,6 +142,8 @@ pub struct ElasticDrainConfig {
     pub custom_id_value: String,
     /// The batching interval.
     pub flush_interval: Duration,
+    /// Maximum retries in case of error.
+    pub max_retries: usize,
 }
 
 /// An slog `Drain` for logging to Elasticsearch.
@@ -187,6 +191,7 @@ impl ElasticDrain {
         let logs = self.logs.clone();
         let config = self.config.clone();
         let mut interval = tokio::time::interval(self.config.flush_interval);
+        let max_retries = self.config.max_retries;
 
         crate::task_spawn::spawn(async move {
             loop {
@@ -261,7 +266,6 @@ impl ElasticDrain {
 
                 // Send batch of logs to Elasticsearch
                 let client = Client::new();
-                let logger_for_err = flush_logger.clone();
 
                 let header = match config.general.username {
                     Some(username) => client
@@ -272,19 +276,24 @@ impl ElasticDrain {
                         .post(batch_url)
                         .header(CONTENT_TYPE, "application/json"),
                 };
-                header
-                    .body(batch_body)
-                    .send()
-                    .and_then(|response| async { response.error_for_status() })
-                    .map_ok(|_| ())
-                    .unwrap_or_else(move |e| {
+
+                retry("send logs to elasticsearch", &flush_logger)
+                    .limit(max_retries)
+                    .timeout_secs(30)
+                    .run(move || {
+                        header
+                            .try_clone()
+                            .unwrap() // Unwrap: Request body not yet set
+                            .body(batch_body.clone())
+                            .send()
+                            .and_then(|response| async { response.error_for_status() })
+                            .map_ok(|_| ())
+                    })
+                    .await
+                    .unwrap_or_else(|e| {
                         // Log if there was a problem sending the logs
-                        error!(
-                            logger_for_err,
-                            "Failed to send logs to Elasticsearch: {}", e
-                        );
+                        error!(flush_logger, "Failed to send logs to Elasticsearch: {}", e);
                     })
-                    .await;
             }
         });
     }

graph/src/log/factory.rs

@@ -11,6 +11,12 @@ lazy_static::lazy_static! {
         .unwrap_or("5".into())
         .parse::<u64>()
         .expect("invalid GRAPH_ELASTIC_SEARCH_FLUSH_INTERVAL_SECS"));
+
+    static ref ES_MAX_RETRIES: usize =
+        std::env::var("GRAPH_ELASTIC_SEARCH_MAX_RETRIES")
+        .unwrap_or("5".into())
+        .parse::<usize>()
+        .expect("invalid GRAPH_ELASTIC_SEARCH_MAX_RETRIES");
 }
 
 /// Configuration for component-specific logging to Elasticsearch.
@@ -73,6 +79,7 @@ impl LoggerFactory {
                                     custom_id_key: String::from("componentId"),
                                     custom_id_value: component.to_string(),
                                     flush_interval: *ES_FLUSH_INTERVAL,
+                                    max_retries: *ES_MAX_RETRIES,
                                 },
                                 term_logger.clone(),
                             ),
@@ -102,6 +109,7 @@ impl LoggerFactory {
                             custom_id_key: String::from("subgraphId"),
                             custom_id_value: loc.hash.to_string(),
                             flush_interval: *ES_FLUSH_INTERVAL,
+                            max_retries: *ES_MAX_RETRIES,
                         },
                         term_logger.clone(),
                     ),