feat: option to use basic auth for elastic

Author: d-perl

src/config.rs

@@ -47,13 +47,33 @@ pub struct RedisConfig {
 #[derive(Clone, Debug, Deserialize)]
 pub struct ElasticConfig {
     pub url: UrlPort,
-    pub api_key: String,
+    pub api_key: Option<String>,
+    pub username: Option<String>,
+    pub password: Option<String>,
     #[serde(default = "default_chunk_size")]
     pub chunk_size: u16,
     #[serde(default = "default_index")]
     pub index: String,
 }
 
+impl ElasticConfig {
+    pub fn credentials(&self) -> Result<elasticsearch::auth::Credentials, &str> {
+        if let Some(api_key) = &self.api_key {
+            Ok(elasticsearch::auth::Credentials::EncodedApiKey(
+                api_key.into(),
+            ))
+        } else if let Some(username) = &self.username
+            && let Some(password) = &self.password
+        {
+            Ok(elasticsearch::auth::Credentials::Basic(
+                username.to_owned(),
+                password.to_owned(),
+            ))
+        } else {
+            Err("No credentials in config!")
+        }
+    }
+}
 #[derive(Clone, Debug, Deserialize)]
 pub struct IngestorConfig {
     pub redis: RedisConfig,
@@ -113,7 +133,7 @@ port = 9876
         let config: IngestorConfig = toml::from_str(&test_str).unwrap();
         assert_eq!(config.redis.url.full_url(), "http://127.0.0.1:12345");
         assert_eq!(config.elastic.url.full_url(), "http://127.0.0.1:9876");
-        assert_eq!(config.elastic.api_key, "abcdefgh==");
+        assert_eq!(config.elastic.api_key, Some("abcdefgh==".into()));
     }
 
     #[test]
@@ -138,7 +158,7 @@ api_key = \"testkey\"
 ";
         let elastic: ElasticConfig = toml::from_str(&test_str).unwrap();
         assert_eq!(elastic.chunk_size, 100);
-        assert_eq!(elastic.api_key, "testkey");
+        assert_eq!(elastic.api_key, Some("testkey".into()));
         assert_eq!(elastic.url.full_url(), "http://localhost:9200");
     }
 

src/elastic_push.rs

@@ -1,14 +1,14 @@
 use elasticsearch::{Elasticsearch, http::request::JsonBody};
 use tokio::sync::mpsc;
 
-use std::iter::once;
+use std::{error::Error, iter::once};
 
 use crate::{config::ElasticConfig, redis_logs::LogRecord};
 
-fn elastic_client(url: &str, api_key: &str) -> Result<Elasticsearch, elasticsearch::Error> {
-    let url = elasticsearch::http::Url::parse(url)?;
+fn elastic_client(config: &ElasticConfig) -> Result<Elasticsearch, Box<dyn Error>> {
+    let url = elasticsearch::http::Url::parse(&config.url.full_url())?;
     let conn_pool = elasticsearch::http::transport::SingleNodeConnectionPool::new(url);
-    let credentials = elasticsearch::auth::Credentials::EncodedApiKey(api_key.into());
+    let credentials = config.credentials()?;
     let transport = elasticsearch::http::transport::TransportBuilder::new(conn_pool)
         .auth(credentials)
         .build()?;
@@ -41,8 +41,7 @@ fn make_json_body(
 }
 
 pub async fn consumer_loop(rx: &mut mpsc::UnboundedReceiver<LogRecord>, config: ElasticConfig) {
-    let elastic_client = elastic_client(&config.url.full_url(), &config.api_key)
-        .expect("Failed to connect to Elastic!");
+    let elastic_client = elastic_client(&config).expect("Failed to connect to Elastic!");
 
     let mut buffer: Vec<LogRecord> = Vec::with_capacity(config.chunk_size.into());
 
@@ -69,6 +68,8 @@ pub async fn consumer_loop(rx: &mut mpsc::UnboundedReceiver<LogRecord>, config:
 
 #[cfg(test)]
 mod tests {
+    use crate::config::UrlPort;
+
     use super::*;
     use serde::{Deserialize, Serialize};
 
@@ -156,7 +157,17 @@ mod tests {
 
     #[test]
     fn test_elastic_client_invalid_url() {
-        let result = elastic_client("not a url", "apikey");
+        let result = elastic_client(&ElasticConfig {
+            url: UrlPort {
+                url: "not an url".into(),
+                port: 9876,
+            },
+            api_key: Some("key".into()),
+            username: None,
+            password: None,
+            chunk_size: 8,
+            index: "".into(),
+        });
         assert!(result.is_err());
     }
 }