chore(sinks): migrated elasticsearch sink

txpipe · Jun 26, 2023 · d01a0db · d01a0db
1 parent 4a16ada
commit d01a0db
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 3 deletions.
diff --git a/examples/elasticsearch/daemon.toml b/examples/elasticsearch/daemon.toml
@@ -17,4 +17,5 @@ type = "ParseCbor"
 [sink]
 type = "ElasticSearch"
 url = "http://localhost:9200"
-index = "oura.sink.v2"
+index = "oura.sink.v6"
+idempotency = true
diff --git a/src/sinks/elasticsearch.rs b/src/sinks/elasticsearch.rs
@@ -66,11 +66,15 @@ impl gasket::framework::Worker<Stage> for Worker {
             return Ok(());
         }
 
+        let slot = point.slot_or_default().to_string();
+        let mut parts = IndexParts::Index(&stage.config.index);
+        if stage.config.idempotency {
+            parts = IndexParts::IndexId(&stage.config.index, &slot);
+        }
+
         let timestamp = stage.genesis.slot_to_wallclock(point.slot_or_default());
         let payload = ESRecord::new(record.unwrap(), timestamp);
 
-        let parts = IndexParts::Index(&stage.config.index);
-
         self.client
             .index(parts)
             .body(payload)
@@ -124,6 +128,8 @@ pub struct Config {
     pub url: String,
     pub index: String,
     pub credentials: Option<CredentialsConfig>,
+    #[serde(default)]
+    pub idempotency: bool,
 }
 
 impl Config {