enhancement(elasticsearch sink): add pipeline config (#2734)

Author: francescop

.meta/sinks/elasticsearch.toml.erb

@@ -188,6 +188,12 @@ examples = [
 templateable = true
 description = "Index name to write events to."
 
+[sinks.elasticsearch.options.pipeline]
+type = "string"
+common = true
+examples = ["pipeline-name"]
+description = "Name of the pipeline to apply."
+
 [sinks.elasticsearch.options.query]
 type = "table"
 description = "Custom parameters to Elasticsearch query string."

config/vector.spec.toml

@@ -4741,6 +4741,13 @@ require('custom_module')
   # * type: [string]
   inputs = ["my-source-or-transform-id"]
 
+  # Name of the pipeline to apply.
+  #
+  # * optional
+  # * no default
+  # * type: string
+  pipeline = "pipeline-name"
+
   # The component type. This is a required field that tells Vector which
   # component to use. The value _must_ be `#{name}`.
   #

src/sinks/elasticsearch.rs

@@ -41,6 +41,8 @@ pub struct ElasticSearchConfig {
     pub index: Option<String>,
     pub doc_type: Option<String>,
     pub id_key: Option<String>,
+    pub pipeline: Option<String>,
+
     #[serde(default)]
     pub compression: Compression,
     #[serde(
@@ -353,6 +355,10 @@ impl ElasticSearchCommon {
         let mut query_params = config.query.clone().unwrap_or_default();
         query_params.insert("timeout".into(), format!("{}s", request.timeout.as_secs()));
 
+        if let Some(pipeline) = &config.pipeline {
+            query_params.insert("pipeline".into(), pipeline.into());
+        }
+
         let mut query = url::form_urlencoded::Serializer::new(String::new());
         for (p, v) in &query_params {
             query.append_pair(&p[..], &v[..]);
@@ -523,6 +529,22 @@ mod integration_tests {
     use std::fs::File;
     use std::io::Read;
 
+    #[test]
+    fn ensure_pipeline_in_params() {
+        let index = gen_index();
+        let pipeline = String::from("test-pipeline");
+
+        let config = ElasticSearchConfig {
+            host: "http://localhost:9200".into(),
+            index: Some(index.clone()),
+            pipeline: Some(pipeline.clone()),
+            ..config()
+        };
+        let common = ElasticSearchCommon::parse_config(&config).expect("Config error");
+
+        assert_eq!(common.query_params["pipeline"], pipeline);
+    }
+
     #[test]
     fn structures_events_correctly() {
         let mut rt = runtime();

website/docs/reference/sinks/elasticsearch.md

@@ -1,5 +1,5 @@
 ---
-last_modified_on: "2020-06-02"
+last_modified_on: "2020-06-03"
 delivery_guarantee: "at_least_once"
 component_title: "Elasticsearch"
 description: "The Vector `elasticsearch` sink batches `log` events to Elasticsearch via the `_bulk` API endpoint."
@@ -58,6 +58,7 @@ endpoint][urls.elasticsearch_bulk].
   healthcheck = true # optional, default
   host = "http://10.24.32.122:9000" # optional, no default
   index = "vector-%F" # optional, default
+  pipeline = "pipeline-name" # optional, no default
 ```
 
 </TabItem>
@@ -74,6 +75,7 @@ endpoint][urls.elasticsearch_bulk].
   host = "http://10.24.32.122:9000" # optional, no default
   id_key = "id" # optional, no default
   index = "vector-%F" # optional, default
+  pipeline = "pipeline-name" # optional, no default
 
   # Auth
   auth.password = "${ELASTICSEARCH_PASSWORD}" # required, required when strategy = "basic"
@@ -747,6 +749,29 @@ Index name to write events to.
  See [Document Conflicts](#document-conflicts) and [Template Syntax](#template-syntax) for more info.
 
 
+</Field>
+<Field
+  common={true}
+  defaultValue={null}
+  enumValues={null}
+  examples={["pipeline-name"]}
+  groups={[]}
+  name={"pipeline"}
+  path={null}
+  relevantWhen={null}
+  required={false}
+  templateable={false}
+  type={"string"}
+  unit={null}
+  warnings={[]}
+  >
+
+### pipeline
+
+Name of the pipeline to apply.
+
+
+
 </Field>
 <Field
   common={false}