feat(elasticsearch sink): Add HTTP Basic authorization

.metadata.toml

@@ -1059,6 +1059,23 @@ null = false
 templateable = true
 description = "Index name to write events to."
 
+[sinks.elasticsearch.options.basic_auth]
+type = "table"
+null = true
+description = "Options for basic authentication."
+
+[sinks.elasticsearch.options.basic_auth.options.password]
+type = "string"
+examples = ["password"]
+null = false
+description = "The basic authentication password."
+
+[sinks.elasticsearch.options.basic_auth.options.user]
+type = "string"
+examples = ["username"]
+null = false
+description = "The basic authentication user name."
+
 # ------------------------------------------------------------------------------
 # sinks.http
 # ------------------------------------------------------------------------------

Cargo.toml

@@ -116,6 +116,7 @@ grok = "1.0.0"
 nom = "5.0.0"
 uuid = { version = "0.7", features = ["serde", "v4"] }
 exitcode = "1.1.2"
+base64 = "0.10.1"
 
 [build-dependencies]
 prost-build = "0.4.0"

config/vector.spec.toml

@@ -1448,6 +1448,23 @@ end
   # * unit: seconds
   retry_backoff_secs = 5
 
+  #
+  # Basic auth
+  #
+
+  [sinks.elasticsearch.basic_auth]
+    # The basic authentication password.
+    # 
+    # * required
+    # * no default
+    password = "password"
+
+    # The basic authentication user name.
+    # 
+    # * required
+    # * no default
+    user = "username"
+
   #
   # Buffer
   #

docs/usage/configuration/sinks/elasticsearch.md

@@ -52,6 +52,11 @@ The `elasticsearch` sink [batches](#buffers-and-batches) [`log`][docs.log_event]
   retry_attempts = 5 # default
   retry_backoff_secs = 5 # default, seconds
 
+  # OPTIONAL - Basic auth
+  [sinks.my_sink_id.basic_auth]
+    password = "password"
+    user = "username"
+
   # OPTIONAL - Buffer
   [sinks.my_sink_id.buffer]
     type = "memory" # default, enum: "memory" or "disk"
@@ -85,6 +90,11 @@ The `elasticsearch` sink [batches](#buffers-and-batches) [`log`][docs.log_event]
   retry_attempts = <int>
   retry_backoff_secs = <int>
 
+  # OPTIONAL - Basic auth
+  [sinks.<sink-id>.basic_auth]
+    password = "<string>"
+    user = "<string>"
+
   # OPTIONAL - Buffer
   [sinks.<sink-id>.buffer]
     type = {"memory" | "disk"}
@@ -204,6 +214,23 @@ The `elasticsearch` sink [batches](#buffers-and-batches) [`log`][docs.log_event]
   # * unit: seconds
   retry_backoff_secs = 5
 
+  #
+  # Basic auth
+  #
+
+  [sinks.elasticsearch_sink.basic_auth]
+    # The basic authentication password.
+    # 
+    # * required
+    # * no default
+    password = "password"
+
+    # The basic authentication user name.
+    # 
+    # * required
+    # * no default
+    user = "username"
+
   #
   # Buffer
   #
@@ -265,6 +292,9 @@ The `elasticsearch` sink [batches](#buffers-and-batches) [`log`][docs.log_event]
 | `request_timeout_secs` | `int` | The maximum time a request can take before being aborted. See [Timeouts](#timeouts) for more info.<br />`default: 60` `unit: seconds` |
 | `retry_attempts` | `int` | The maximum number of retries to make for failed requests. See [Retry Policy](#retry-policy) for more info.<br />`default: 5` |
 | `retry_backoff_secs` | `int` | The amount of time to wait before attempting a failed request again. See [Retry Policy](#retry-policy) for more info.<br />`default: 5` `unit: seconds` |
+| **OPTIONAL** - Basic auth | | |
+| `basic_auth.password` | `string` | The basic authentication password.<br />`required` `example: "password"` |
+| `basic_auth.user` | `string` | The basic authentication user name.<br />`required` `example: "username"` |
 | **OPTIONAL** - Buffer | | |
 | `buffer.type` | `string` | The buffer's type / location. `disk` buffers are persistent and will be retained between restarts.<br />`default: "memory"` `enum: "memory" or "disk"` |
 | `buffer.when_full` | `string` | The behavior when the buffer becomes full.<br />`default: "block"` `enum: "block" or "drop_newest"` |

docs/usage/configuration/specification.md

@@ -1468,6 +1468,23 @@ end
   # * unit: seconds
   retry_backoff_secs = 5
 
+  #
+  # Basic auth
+  #
+
+  [sinks.elasticsearch.basic_auth]
+    # The basic authentication password.
+    # 
+    # * required
+    # * no default
+    password = "password"
+
+    # The basic authentication user name.
+    # 
+    # * required
+    # * no default
+    user = "username"
+
   #
   # Buffer
   #

src/sinks/elasticsearch.rs

@@ -36,13 +36,22 @@ pub struct ElasticSearchConfig {
     pub request_rate_limit_num: Option<u64>,
     pub request_retry_attempts: Option<usize>,
     pub request_retry_backoff_secs: Option<u64>,
+
+    pub basic_auth: Option<ElasticSearchBasicAuthConfig>,
+}
+
+#[derive(Deserialize, Serialize, Debug, Clone, Default)]
+#[serde(deny_unknown_fields)]
+pub struct ElasticSearchBasicAuthConfig {
+    pub password: String,
+    pub user: String,
 }
 
 #[typetag::serde(name = "elasticsearch")]
 impl SinkConfig for ElasticSearchConfig {
     fn build(&self, acker: Acker) -> Result<(super::RouterSink, super::Healthcheck), String> {
-        let sink = es(self.clone(), acker);
-        let healthcheck = healthcheck(self.host.clone());
+        let sink = es(self, acker);
+        let healthcheck = healthcheck(&self.host);
 
         Ok((sink, healthcheck))
     }
@@ -52,7 +61,7 @@ impl SinkConfig for ElasticSearchConfig {
     }
 }
 
-fn es(config: ElasticSearchConfig, acker: Acker) -> super::RouterSink {
+fn es(config: &ElasticSearchConfig, acker: Acker) -> super::RouterSink {
     let host = config.host.clone();
     let id_key = config.id_key.clone();
     let gzip = match config.compression.unwrap_or(Compression::Gzip) {
@@ -75,14 +84,19 @@ fn es(config: ElasticSearchConfig, acker: Acker) -> super::RouterSink {
     } else {
         Template::from("vector-%Y.%m.%d")
     };
-    let doc_type = config.clone().doc_type.unwrap_or("_doc".into());
+    let doc_type = config.doc_type.clone().unwrap_or("_doc".into());
 
     let policy = FixedRetryPolicy::new(
         retry_attempts,
         Duration::from_secs(retry_backoff_secs),
         HttpRetryLogic,
     );
 
+    let authorization = config.basic_auth.clone().map(|auth| {
+        let token = format!("{}:{}", auth.user, auth.password);
+        format!("Basic {}", base64::encode(token.as_bytes()))
+    });
+
     let http_service = HttpService::new(move |body: Vec<u8>| {
         let uri = format!("{}/_bulk", host);
         let uri: Uri = uri.parse().unwrap();
@@ -92,6 +106,9 @@ fn es(config: ElasticSearchConfig, acker: Acker) -> super::RouterSink {
         builder.uri(uri);
 
         builder.header("Content-Type", "application/x-ndjson");
+        if let Some(ref auth) = authorization {
+            builder.header("Authorization", &auth[..]);
+        }
 
         if gzip {
             builder.header("Content-Encoding", "gzip");
@@ -154,7 +171,7 @@ fn encode_event(
     Some(body)
 }
 
-fn healthcheck(host: String) -> super::Healthcheck {
+fn healthcheck(host: &str) -> super::Healthcheck {
     let uri = format!("{}/_cluster/health", host);
     let request = Request::get(uri).body(Body::empty()).unwrap();