enhancement(elasticsearch sink): add support for data streams

docs/reference/components/sinks/elasticsearch.cue

@@ -76,7 +76,11 @@ components: sinks: elasticsearch: {
 			"x86_64-unknown-linux-musl":  true
 		}
 
-		requirements: []
+		requirements: [
+			#"""
+				Elasticsearch's Data streams feature requires Vector to be configured with the `create` `bulk_action`. *This is not enabled by default.*
+				"""#,
+		]
 		warnings: []
 		notices: []
 	}
@@ -151,6 +155,16 @@ components: sinks: elasticsearch: {
 				}
 			}
 		}
+		bulk_action: {
+			common:      false
+			description: "Action to use when making requests to the [Elasticsearch Bulk API](elasticsearch_bulk). Supports `index` and `create`."
+			required:    false
+			warnings: []
+			type: string: {
+				default: "index"
+				examples: ["index", "create"]
+			}
+		}
 		doc_type: {
 			common:      false
 			description: "The `doc_type` for your index data. This is only relevant for Elasticsearch <= 6.X. If you are using >= 7.0 you do not need to set this option since Elasticsearch has removed it."
@@ -236,9 +250,19 @@ components: sinks: elasticsearch: {
 			title: "Conflicts"
 			body: """
 				Vector [batches](#buffers--batches) data flushes it to Elasticsearch's
-				[`_bulk` API endpoint][urls.elasticsearch_bulk]. All events are inserted
-				via the `index` action. In the case of an conflict, such as a document with the
-				same `id`, Vector will add or _replace_ the document as necessary.
+				[`_bulk` API endpoint][urls.elasticsearch_bulk]. By default, all events are
+				inserted via the `index` action which will update documents if an existing
+				one has the same `id`. If `bulk_action` is configured with `create`, Elasticsearch
+				will _not_ replace an existing document and instead return a conflict error.
+				"""
+		}
+
+		data_streams: {
+			title: "Data streams"
+			body: """
+				By default, Vector will use the `index` action with Elasticsearch's Bulk API.
+				To use [Data streams][urls.elasticsearch_data_streams], `bulk_action` must be configured
+				with the `create` option.
 				"""
 		}
 

docs/reference/urls.cue

@@ -122,6 +122,7 @@ urls: {
 	cidr:                                                     "https://en.wikipedia.org/wiki/Classless_Inter-Domain_Routing"
 	elasticsearch:                                            "https://www.elastic.co/products/elasticsearch"
 	elasticsearch_bulk:                                       "https://www.elastic.co/guide/en/elasticsearch/reference/current/docs-bulk.html"
+	elasticsearch_data_streams:                               "https://www.elastic.co/guide/en/elasticsearch/reference/current/data-streams.html"
 	elasticsearch_id_field:                                   "https://www.elastic.co/guide/en/elasticsearch/reference/current/mapping-id-field.html"
 	elasticsearch_id_performance:                             "https://www.elastic.co/guide/en/elasticsearch/reference/master/tune-for-indexing-speed.html#_use_auto_generated_ids"
 	elasticsearch_ignore_malformed:                           "https://www.elastic.co/guide/en/elasticsearch/reference/current/ignore-malformed.html"

src/sinks/elasticsearch.rs

@@ -61,6 +61,8 @@ pub struct ElasticSearchConfig {
 
     pub aws: Option<RegionOrEndpoint>,
     pub tls: Option<TlsOptions>,
+    #[serde(default)]
+    pub bulk_action: BulkAction,
 }
 
 lazy_static! {
@@ -84,6 +86,31 @@ pub enum ElasticSearchAuth {
     Aws { assume_role: Option<String> },
 }
 
+#[derive(Derivative, Deserialize, Serialize, Clone, Debug)]
+#[serde(deny_unknown_fields, rename_all = "snake_case")]
+#[derivative(Default)]
+pub enum BulkAction {
+    #[derivative(Default)]
+    Index,
+    Create,
+}
+
+impl BulkAction {
+    pub fn as_str(&self) -> &'static str {
+        match *self {
+            BulkAction::Index => "index",
+            BulkAction::Create => "create",
+        }
+    }
+
+    pub fn as_json_pointer(&self) -> &'static str {
+        match *self {
+            BulkAction::Index => "/index",
+            BulkAction::Create => "/create",
+        }
+    }
+}
+
 inventory::submit! {
     SinkDescription::new::<ElasticSearchConfig>("elasticsearch")
 }
@@ -146,6 +173,7 @@ pub struct ElasticSearchCommon {
     compression: Compression,
     region: Region,
     query_params: HashMap<String, String>,
+    bulk_action: BulkAction,
 }
 
 #[derive(Debug, Snafu)]
@@ -177,14 +205,16 @@ impl HttpSink for ElasticSearchCommon {
             .ok()?;
 
         let mut action = json!({
-            "index": {
+            self.bulk_action.as_str(): {
                 "_index": index,
                 "_type": self.doc_type,
             }
         });
         maybe_set_id(
             self.config.id_key.as_ref(),
-            action.pointer_mut("/index").unwrap(),
+            action
+                .pointer_mut(self.bulk_action.as_json_pointer())
+                .unwrap(),
             &mut event,
         );
 
@@ -372,6 +402,7 @@ impl ElasticSearchCommon {
         let index = Template::try_from(index).context(IndexTemplate)?;
 
         let doc_type = config.doc_type.clone().unwrap_or_else(|| "_doc".into());
+        let bulk_action = config.bulk_action.clone();
 
         let request = config.request.unwrap_with(&REQUEST_DEFAULTS);
 
@@ -404,6 +435,7 @@ impl ElasticSearchCommon {
             compression,
             region,
             query_params,
+            bulk_action,
         })
     }
 
@@ -527,6 +559,31 @@ mod tests {
         assert_eq!(json!({}), action);
     }
 
+    #[test]
+    fn sets_create_action_when_configured() {
+        use crate::config::log_schema;
+        use chrono::{TimeZone, Utc};
+
+        let config = ElasticSearchConfig {
+            bulk_action: BulkAction::Create,
+            index: Some(String::from("vector")),
+            endpoint: String::from("https://example.com"),
+            ..Default::default()
+        };
+        let es = ElasticSearchCommon::parse_config(&config).unwrap();
+
+        let mut event = Event::from("hello there");
+        event.as_mut_log().insert(
+            log_schema().timestamp_key(),
+            Utc.ymd(2020, 12, 1).and_hms(1, 2, 3),
+        );
+        let encoded = es.encode_event(event).unwrap();
+        let expected = r#"{"create":{"_index":"vector","_type":"_doc"}}
+{"message":"hello there","timestamp":"2020-12-01T01:02:03Z"}
+"#;
+        assert_eq!(std::str::from_utf8(&encoded).unwrap(), &expected[..]);
+    }
+
     #[test]
     fn handles_error_response() {
         let json = "{\"took\":185,\"errors\":true,\"items\":[{\"index\":{\"_index\":\"test-hgw28jv10u\",\"_type\":\"log_lines\",\"_id\":\"3GhQLXEBE62DvOOUKdFH\",\"status\":400,\"error\":{\"type\":\"illegal_argument_exception\",\"reason\":\"mapper [message] of different type, current_type [long], merged_type [text]\"}}}]}";