feat(new sink): Adding greptimedb metrics sink

Cargo.toml

@@ -226,7 +226,7 @@ hex = { version = "0.4.3", default-features = false, optional = true }
 sha2 = { version = "0.10.7", default-features = false, optional = true }
 
 # GreptimeDB
-greptimedb-client = { git = "https://github.com/GreptimeTeam/greptimedb-client-rust.git", rev = "0841134b004cfcf228af038b2b338dccaf743644", optional = true }
+greptimedb-client = { git = "https://github.com/GreptimeTeam/greptimedb-client-rust.git", rev = "bc32362adf0df17a41a95bae4221d6d8f1775656", optional = true }
 
 # VRL Lang
 vrl = { package = "vrl", version = "0.4.0", features = ["cli", "test"] }

src/sinks/greptimedb/mod.rs

@@ -25,10 +25,9 @@ use tower::ServiceBuilder;
 use vector_common::sensitive_string::SensitiveString;
 use vector_config::configurable_component;
 use vector_core::config::{AcknowledgementsConfig, Input};
+use vector_core::tls::TlsConfig;
 
-use crate::config::{SinkConfig, SinkContext};
-use crate::sinks::util::{BatchConfig, SinkBatchSettings};
-use crate::sinks::{Healthcheck, VectorSink};
+use crate::sinks::prelude::*;
 
 use self::service::GreptimeDBRetryLogic;
 
@@ -111,6 +110,9 @@ pub struct GreptimeDBConfig {
         skip_serializing_if = "crate::serde::skip_serializing_if_default"
     )]
     pub acknowledgements: AcknowledgementsConfig,
+
+    #[configurable(derived)]
+    pub tls: Option<TlsConfig>,
 }
 
 impl_generate_config_from_default!(GreptimeDBConfig);
@@ -122,7 +124,7 @@ impl SinkConfig for GreptimeDBConfig {
         let request_settings = self.request.unwrap_with(&TowerRequestConfig::default());
         let service = ServiceBuilder::new()
             .settings(request_settings, GreptimeDBRetryLogic)
-            .service(service::GreptimeDBService::new(self));
+            .service(service::GreptimeDBService::try_new(self)?);
         let sink = sink::GreptimeDBSink {
             service,
             batch_settings: self.batch.into_batcher_settings()?,

src/sinks/greptimedb/service.rs

@@ -3,19 +3,14 @@ use std::sync::Arc;
 use std::task::Poll;
 
 use futures_util::future::BoxFuture;
-
 use greptimedb_client::api::v1::auth_header::AuthScheme;
 use greptimedb_client::api::v1::*;
+use greptimedb_client::channel_manager::*;
 use greptimedb_client::{Client, Database, Error as GreptimeError};
 use tower::Service;
-use vector_common::finalization::{EventFinalizers, EventStatus, Finalizable};
-use vector_common::internal_event::CountByteSize;
-use vector_common::request_metadata::{MetaDescriptive, RequestMetadata};
 use vector_core::event::Metric;
-use vector_core::stream::DriverResponse;
 
-use crate::sinks::prelude::RetryLogic;
-use crate::sinks::util::metadata::RequestMetadataBuilder;
+use crate::sinks::prelude::*;
 
 use super::batch::GreptimeDBBatchSizer;
 use super::request_builder::metric_to_insert_request;
@@ -108,8 +103,20 @@ pub struct GreptimeDBService {
 }
 
 impl GreptimeDBService {
-    pub fn new(config: &GreptimeDBConfig) -> Self {
-        let grpc_client = Client::with_urls(vec![&config.endpoint]);
+    pub fn try_new(config: &GreptimeDBConfig) -> crate::Result<Self> {
+        let grpc_client = if let Some(tls_config) = &config.tls {
+            let channel_config = ChannelConfig {
+                client_tls: Self::try_from_tls_config(tls_config),
+                ..Default::default()
+            };
+            Client::with_manager_and_urls(
+                ChannelManager::with_tls_config(channel_config).map_err(Box::new)?,
+                vec![&config.endpoint],
+            )
+        } else {
+            Client::with_urls(vec![&config.endpoint])
+        };
+
         let mut client = Database::new_with_dbname(&config.dbname, grpc_client);
 
         if let (Some(username), Some(password)) = (&config.username, &config.password) {
@@ -119,10 +126,33 @@ impl GreptimeDBService {
             }))
         };
 
-        // TODO: tls configuration
-
-        GreptimeDBService {
+        Ok(GreptimeDBService {
             client: Arc::new(client),
+        })
+    }
+
+    fn try_from_tls_config(tls_config: &TlsConfig) -> Option<ClientTlsOption> {
+        if let Some(ca_path) = tls_config.ca_file.as_ref() {
+            let cert_path = tls_config.crt_file.as_ref().expect(
+                "Client cert file is required for greptimedb sink when custom CA specified",
+            );
+            let key_path = tls_config
+                .key_file
+                .as_ref()
+                .expect("Client key file is required for greptimedb sink when custom CA specified");
+            if tls_config.key_pass.is_some() {
+                warn!(
+                    message = "TLS key file with password is not supported by greptimedb client at the moment."
+                );
+            }
+
+            Some(ClientTlsOption {
+                server_ca_cert_path: ca_path.clone(),
+                client_key_path: key_path.clone(),
+                client_cert_path: cert_path.clone(),
+            })
+        } else {
+            None
         }
     }
 }

src/sinks/greptimedb/sink.rs

@@ -2,14 +2,11 @@ use async_trait::async_trait;
 
 use futures::StreamExt;
 use futures_util::stream::BoxStream;
-use vector_core::event::{Event, Metric, MetricValue};
-use vector_core::sink::StreamSink;
-use vector_core::stream::BatcherSettings;
+use vector_core::event::{Metric, MetricValue};
 
+use crate::sinks::prelude::*;
 use crate::sinks::util::buffer::metrics::MetricNormalize;
 use crate::sinks::util::buffer::metrics::MetricSet;
-use crate::sinks::util::service::Svc;
-use crate::sinks::util::SinkBuilderExt;
 
 use super::batch::GreptimeDBBatchSizer;
 use super::service::{GreptimeDBRequest, GreptimeDBRetryLogic, GreptimeDBService};

website/content/en/docs/reference/configuration/sinks/greptimedb.md

@@ -0,0 +1,14 @@
+---
+title: GreptimeDB
+description: Writes metric data to [GreptimeDB](https://github.com/greptimeteam/greptimedb)
+kind: sink
+layout: component
+tags: ["greptimedb", "component", "sink", "storage", "time-series", "metrics"]
+---
+
+{{/*
+This doc is generated using:
+
+1. The template in layouts/docs/component.html
+2. The relevant CUE data in cue/reference/components/...
+*/}}

website/cue/reference/components/sinks/base/greptimedb.cue

@@ -243,6 +243,91 @@ base: components: sinks: greptimedb: configuration: {
 			}
 		}
 	}
+	tls: {
+		description: "TLS configuration."
+		required:    false
+		type: object: options: {
+			alpn_protocols: {
+				description: """
+					Sets the list of supported ALPN protocols.
+
+					Declare the supported ALPN protocols, which are used during negotiation with peer. They are prioritized in the order
+					that they are defined.
+					"""
+				required: false
+				type: array: items: type: string: examples: ["h2"]
+			}
+			ca_file: {
+				description: """
+					Absolute path to an additional CA certificate file.
+
+					The certificate must be in the DER or PEM (X.509) format. Additionally, the certificate can be provided as an inline string in PEM format.
+					"""
+				required: false
+				type: string: examples: ["/path/to/certificate_authority.crt"]
+			}
+			crt_file: {
+				description: """
+					Absolute path to a certificate file used to identify this server.
+
+					The certificate must be in DER, PEM (X.509), or PKCS#12 format. Additionally, the certificate can be provided as
+					an inline string in PEM format.
+
+					If this is set, and is not a PKCS#12 archive, `key_file` must also be set.
+					"""
+				required: false
+				type: string: examples: ["/path/to/host_certificate.crt"]
+			}
+			key_file: {
+				description: """
+					Absolute path to a private key file used to identify this server.
+
+					The key must be in DER or PEM (PKCS#8) format. Additionally, the key can be provided as an inline string in PEM format.
+					"""
+				required: false
+				type: string: examples: ["/path/to/host_certificate.key"]
+			}
+			key_pass: {
+				description: """
+					Passphrase used to unlock the encrypted key file.
+
+					This has no effect unless `key_file` is set.
+					"""
+				required: false
+				type: string: examples: ["${KEY_PASS_ENV_VAR}", "PassWord1"]
+			}
+			verify_certificate: {
+				description: """
+					Enables certificate verification.
+
+					If enabled, certificates must not be expired and must be issued by a trusted
+					issuer. This verification operates in a hierarchical manner, checking that the leaf certificate (the
+					certificate presented by the client/server) is not only valid, but that the issuer of that certificate is also valid, and
+					so on until the verification process reaches a root certificate.
+
+					Relevant for both incoming and outgoing connections.
+
+					Do NOT set this to `false` unless you understand the risks of not verifying the validity of certificates.
+					"""
+				required: false
+				type: bool: {}
+			}
+			verify_hostname: {
+				description: """
+					Enables hostname verification.
+
+					If enabled, the hostname used to connect to the remote host must be present in the TLS certificate presented by
+					the remote host, either as the Common Name or as an entry in the Subject Alternative Name extension.
+
+					Only relevant for outgoing connections.
+
+					Do NOT set this to `false` unless you understand the risks of not verifying the remote hostname.
+					"""
+				required: false
+				type: bool: {}
+			}
+		}
+	}
 	username: {
 		description: """
 			The username for your GreptimeDB instance.