diff --git a/Cargo.lock b/Cargo.lock
index 0625b4de69e733..019863e468eeda 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -2799,6 +2799,37 @@ dependencies = [
  "syn 2.0.66",
 ]
 
+[[package]]
+name = "derive_builder"
+version = "0.20.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0350b5cb0331628a5916d6c5c0b72e97393b8b6b03b47a9284f4e7f5a405ffd7"
+dependencies = [
+ "derive_builder_macro",
+]
+
+[[package]]
+name = "derive_builder_core"
+version = "0.20.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d48cda787f839151732d396ac69e3473923d54312c070ee21e9effcaa8ca0b1d"
+dependencies = [
+ "darling 0.20.8",
+ "proc-macro2 1.0.86",
+ "quote 1.0.36",
+ "syn 2.0.66",
+]
+
+[[package]]
+name = "derive_builder_macro"
+version = "0.20.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "206868b8242f27cecce124c19fd88157fbd0dd334df2587f36417bafbc85097b"
+dependencies = [
+ "derive_builder_core",
+ "syn 2.0.66",
+]
+
 [[package]]
 name = "derive_more"
 version = "0.99.17"
@@ -3747,23 +3778,24 @@ dependencies = [
 [[package]]
 name = "greptime-proto"
 version = "0.1.0"
-source = "git+https://github.com/GreptimeTeam/greptime-proto.git?tag=v0.4.1#4306ab645ee55b3f7f2ad3fb7acc5820f967c1aa"
+source = "git+https://github.com/GreptimeTeam/greptime-proto.git?tag=v0.7.0#4bc0d17577dbea47396a064c1ccf229a4c9539fa"
 dependencies = [
  "prost 0.12.6",
  "serde",
  "serde_json",
  "strum 0.25.0",
  "strum_macros 0.25.3",
- "tonic 0.10.2",
- "tonic-build 0.10.2",
+ "tonic 0.11.0",
+ "tonic-build 0.11.0",
 ]
 
 [[package]]
-name = "greptimedb-client"
+name = "greptimedb-ingester"
 version = "0.1.0"
-source = "git+https://github.com/GreptimeTeam/greptimedb-ingester-rust.git?rev=d21dbcff680139ed2065b62100bac3123da7c789#d21dbcff680139ed2065b62100bac3123da7c789"
+source = "git+https://github.com/GreptimeTeam/greptimedb-ingester-rust?rev=2e6b0c5eb6a5e7549c3100e4d356b07d15cce66d#2e6b0c5eb6a5e7549c3100e4d356b07d15cce66d"
 dependencies = [
  "dashmap 5.5.3",
+ "derive_builder",
  "enum_dispatch",
  "futures 0.3.30",
  "futures-util",
@@ -3774,7 +3806,7 @@ dependencies = [
  "snafu 0.7.5",
  "tokio",
  "tokio-stream",
- "tonic 0.10.2",
+ "tonic 0.11.0",
  "tonic-build 0.9.2",
  "tower",
 ]
@@ -9576,6 +9608,7 @@ dependencies = [
  "axum",
  "base64 0.21.7",
  "bytes 1.6.0",
+ "flate2",
  "h2 0.3.26",
  "http 0.2.9",
  "http-body 0.4.5",
@@ -9584,12 +9617,17 @@ dependencies = [
  "percent-encoding",
  "pin-project",
  "prost 0.12.6",
+ "rustls-native-certs 0.7.0",
+ "rustls-pemfile 2.1.0",
+ "rustls-pki-types",
  "tokio",
+ "tokio-rustls 0.25.0",
  "tokio-stream",
  "tower",
  "tower-layer",
  "tower-service",
  "tracing 0.1.40",
+ "zstd 0.12.4",
 ]
 
 [[package]]
@@ -9618,6 +9656,19 @@ dependencies = [
  "syn 2.0.66",
 ]
 
+[[package]]
+name = "tonic-build"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "be4ef6dd70a610078cb4e338a0f79d06bc759ff1b22d2120c2ff02ae264ba9c2"
+dependencies = [
+ "prettyplease 0.2.15",
+ "proc-macro2 1.0.86",
+ "prost-build 0.12.6",
+ "quote 1.0.36",
+ "syn 2.0.66",
+]
+
 [[package]]
 name = "tower"
 version = "0.4.13"
@@ -10312,7 +10363,7 @@ dependencies = [
  "glob",
  "goauth",
  "governor",
- "greptimedb-client",
+ "greptimedb-ingester",
  "grok",
  "h2 0.4.5",
  "hash_hasher",
diff --git a/Cargo.toml b/Cargo.toml
index 7d4979cf02ebd8..73d5e48a3dd0e1 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -252,7 +252,7 @@ ratatui = { version = "0.26.3", optional = true, default-features = false, featu
 hex = { version = "0.4.3", default-features = false, optional = true }
 
 # GreptimeDB
-greptimedb-client = { git = "https://github.com/GreptimeTeam/greptimedb-ingester-rust.git", rev = "d21dbcff680139ed2065b62100bac3123da7c789", optional = true }
+greptimedb-ingester = { git = "https://github.com/GreptimeTeam/greptimedb-ingester-rust", rev = "2e6b0c5eb6a5e7549c3100e4d356b07d15cce66d", optional = true }
 
 # External libs
 arc-swap = { version = "1.7", default-features = false, optional = true }
@@ -674,6 +674,7 @@ sinks-logs = [
   "sinks-elasticsearch",
   "sinks-file",
   "sinks-gcp",
+  "sinks-greptimedb_logs",
   "sinks-honeycomb",
   "sinks-http",
   "sinks-humio",
@@ -737,7 +738,8 @@ sinks-elasticsearch = ["transforms-metric_to_log"]
 sinks-file = ["dep:async-compression"]
 sinks-gcp = ["sinks-gcp-chronicle", "dep:base64", "gcp"]
 sinks-gcp-chronicle =  ["gcp"]
-sinks-greptimedb = ["dep:greptimedb-client"]
+sinks-greptimedb = ["dep:greptimedb-ingester"]
+sinks-greptimedb_logs = []
 sinks-honeycomb = []
 sinks-http = []
 sinks-humio = ["sinks-splunk_hec", "transforms-metric_to_log"]
@@ -787,6 +789,7 @@ all-integration-tests = [
   "gcp-integration-tests",
   "gcp-pubsub-integration-tests",
   "greptimedb-integration-tests",
+  "greptimedb-logs-integration-tests",
   "http-client-integration-tests",
   "humio-integration-tests",
   "influxdb-integration-tests",
@@ -851,6 +854,7 @@ gcp-cloud-storage-integration-tests = ["sinks-gcp"]
 gcp-integration-tests = ["sinks-gcp"]
 gcp-pubsub-integration-tests = ["sinks-gcp", "sources-gcp_pubsub"]
 greptimedb-integration-tests = ["sinks-greptimedb"]
+greptimedb-logs-integration-tests = ["sinks-greptimedb_logs"]
 humio-integration-tests = ["sinks-humio"]
 http-client-integration-tests = ["sources-http_client"]
 influxdb-integration-tests = ["sinks-influxdb"]
diff --git a/src/sinks/greptimedb/logs/config.rs b/src/sinks/greptimedb/logs/config.rs
new file mode 100644
index 00000000000000..f8a7c9d48ea2ed
--- /dev/null
+++ b/src/sinks/greptimedb/logs/config.rs
@@ -0,0 +1,171 @@
+use std::collections::HashMap;
+
+use vector_lib::configurable::configurable_component;
+use vector_lib::sensitive_string::SensitiveString;
+
+use crate::http::{Auth, HttpClient};
+use crate::sinks::greptimedb::logs::http_request_builder::{
+    http_healthcheck, GreptimeDBHttpRetryLogic, GreptimeDBLogsHttpRequestBuilder, PartitionKey,
+};
+use crate::sinks::greptimedb::logs::sink::GreptimeDBLogsHttpSink;
+use crate::sinks::util::http::HttpService;
+use crate::sinks::{
+    greptimedb::{default_dbname_template, GreptimeDBDefaultBatchSettings},
+    prelude::*,
+};
+use vector_lib::codecs::{encoding::Framer, JsonSerializerConfig, NewlineDelimitedEncoderConfig};
+
+fn extra_params_examples() -> HashMap<String, String> {
+    HashMap::<_, _>::from_iter([("source".to_owned(), "vector".to_owned())])
+}
+
+/// Configuration for the `greptimedb_logs` sink.
+#[configurable_component(sink("greptimedb_logs", "Ingest logs data into GreptimeDB."))]
+#[derive(Clone, Debug, Default, Derivative)]
+#[serde(deny_unknown_fields)]
+pub struct GreptimeDBLogsConfig {
+    /// The endpoint of the GreptimeDB server.
+    #[serde(alias = "host")]
+    #[configurable(metadata(docs::examples = "http://localhost:4000"))]
+    pub endpoint: String,
+
+    /// The table that data is inserted into.
+    #[configurable(metadata(docs::examples = "mytable"))]
+    pub table: Template,
+
+    /// The GreptimeDB [database][database] name to connect.
+    ///
+    /// Default to `public`, the default database of GreptimeDB.
+    ///
+    /// Database can be created via `create database` statement on
+    /// GreptimeDB. If you are using GreptimeCloud, use `dbname` from the
+    /// connection information of your instance.
+    ///
+    /// [database]: https://docs.greptime.com/user-guide/concepts/key-concepts#database
+    #[configurable(metadata(docs::examples = "public"))]
+    #[derivative(Default(value = "default_dbname_template()"))]
+    #[serde(default = "default_dbname_template")]
+    pub dbname: Template,
+
+    /// pipeline name to be used for the logs
+    #[configurable(metadata(docs::examples = "pipeline_name"))]
+    pub pipeline_name: Template,
+
+    /// pipeline version to be used for the logs
+    #[configurable(metadata(docs::examples = "2024-06-07 06:46:23.858293"))]
+    pub pipeline_version: Option<Template>,
+
+    /// The username for your GreptimeDB instance.
+    ///
+    /// This is required if your instance has authentication enabled.
+    #[configurable(metadata(docs::examples = "username"))]
+    #[serde(default)]
+    pub username: Option<String>,
+    /// The password for your GreptimeDB instance.
+    ///
+    /// This is required if your instance has authentication enabled.
+    #[configurable(metadata(docs::examples = "password"))]
+    #[serde(default)]
+    pub password: Option<SensitiveString>,
+    /// Set http compression encoding for the request
+    /// Default to none, `gzip` or `zstd` is supported.
+    #[configurable(derived)]
+    #[serde(default = "Compression::gzip_default")]
+    pub compression: Compression,
+
+    #[configurable(derived)]
+    #[serde(default, skip_serializing_if = "crate::serde::is_default")]
+    pub encoding: Transformer,
+
+    /// Custom parameters to add to the query string for each HTTP request sent to GreptimeDB.
+    #[serde(default)]
+    #[configurable(metadata(docs::advanced))]
+    #[configurable(metadata(docs::additional_props_description = "A query string parameter."))]
+    #[configurable(metadata(docs::examples = "extra_params_examples()"))]
+    pub extra_params: Option<HashMap<String, String>>,
+
+    #[configurable(derived)]
+    #[serde(default)]
+    pub(crate) batch: BatchConfig<GreptimeDBDefaultBatchSettings>,
+
+    #[configurable(derived)]
+    #[serde(default)]
+    pub request: TowerRequestConfig,
+
+    #[configurable(derived)]
+    pub tls: Option<TlsConfig>,
+
+    #[configurable(derived)]
+    #[serde(
+        default,
+        deserialize_with = "crate::serde::bool_or_struct",
+        skip_serializing_if = "crate::serde::is_default"
+    )]
+    pub acknowledgements: AcknowledgementsConfig,
+}
+
+impl_generate_config_from_default!(GreptimeDBLogsConfig);
+
+#[async_trait::async_trait]
+#[typetag::serde(name = "greptimedb_logs")]
+impl SinkConfig for GreptimeDBLogsConfig {
+    async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
+        let tls_settings = TlsSettings::from_options(&self.tls)?;
+        let client = HttpClient::new(tls_settings, &cx.proxy)?;
+
+        let auth = match (self.username.clone(), self.password.clone()) {
+            (Some(username), Some(password)) => Some(Auth::Basic {
+                user: username,
+                password,
+            }),
+            _ => None,
+        };
+        let request_builder = GreptimeDBLogsHttpRequestBuilder {
+            endpoint: self.endpoint.clone(),
+            auth: auth.clone(),
+            encoder: (
+                self.encoding.clone(),
+                Encoder::<Framer>::new(
+                    NewlineDelimitedEncoderConfig.build().into(),
+                    JsonSerializerConfig::default().build().into(),
+                ),
+            ),
+            compression: self.compression,
+            extra_params: self.extra_params.clone(),
+        };
+
+        let service: HttpService<GreptimeDBLogsHttpRequestBuilder, PartitionKey> =
+            HttpService::new(client.clone(), request_builder.clone());
+
+        let request_limits = self.request.into_settings();
+
+        let service = ServiceBuilder::new()
+            .settings(request_limits, GreptimeDBHttpRetryLogic::default())
+            .service(service);
+
+        let sink = GreptimeDBLogsHttpSink::new(
+            self.batch.into_batcher_settings()?,
+            service,
+            self.dbname.clone(),
+            self.table.clone(),
+            self.pipeline_name.clone(),
+            self.pipeline_version.clone(),
+            request_builder,
+        );
+
+        let healthcheck = Box::pin(http_healthcheck(
+            client,
+            self.endpoint.clone(),
+            auth.clone(),
+        ));
+        Ok((VectorSink::from_event_streamsink(sink), healthcheck))
+    }
+
+    fn input(&self) -> Input {
+        Input::log()
+    }
+
+    fn acknowledgements(&self) -> &AcknowledgementsConfig {
+        &self.acknowledgements
+    }
+}
diff --git a/src/sinks/greptimedb/logs/http_request_builder.rs b/src/sinks/greptimedb/logs/http_request_builder.rs
new file mode 100644
index 00000000000000..d5248228018de5
--- /dev/null
+++ b/src/sinks/greptimedb/logs/http_request_builder.rs
@@ -0,0 +1,282 @@
+use std::collections::HashMap;
+
+use crate::codecs::{Encoder, Transformer};
+use crate::event::{Event, EventFinalizers, Finalizable};
+use crate::http::{Auth, HttpClient, HttpError};
+use crate::sinks::prelude::*;
+use crate::sinks::prelude::{
+    Compression, EncodeResult, Partitioner, RequestBuilder, RequestMetadata,
+    RequestMetadataBuilder, RetryLogic,
+};
+use crate::sinks::{HTTPRequestBuilderSnafu, HealthcheckError};
+use crate::Error;
+use bytes::Bytes;
+use http::header::{CONTENT_ENCODING, CONTENT_LENGTH, CONTENT_TYPE};
+use http::{Request, StatusCode};
+use hyper::Body;
+use snafu::ResultExt;
+
+use vector_lib::codecs::encoding::Framer;
+
+use crate::sinks::util::http::{
+    HttpRequest, HttpResponse, HttpRetryLogic, HttpServiceRequestBuilder,
+};
+
+/// Partition key for GreptimeDB logs sink.
+#[derive(Hash, Eq, PartialEq, Clone, Debug)]
+pub(super) struct PartitionKey {
+    pub dbname: String,
+    pub table: String,
+    pub pipeline_name: String,
+    pub pipeline_version: Option<String>,
+}
+
+/// KeyPartitioner that partitions events by (dbname, table, pipeline_name, pipeline_version) pair.
+pub(super) struct KeyPartitioner {
+    dbname: Template,
+    table: Template,
+    pipeline_name: Template,
+    pipeline_version: Option<Template>,
+}
+
+impl KeyPartitioner {
+    pub const fn new(
+        db: Template,
+        table: Template,
+        pipeline_name: Template,
+        pipeline_version: Option<Template>,
+    ) -> Self {
+        Self {
+            dbname: db,
+            table,
+            pipeline_name,
+            pipeline_version,
+        }
+    }
+
+    fn render(template: &Template, item: &Event, field: &'static str) -> Option<String> {
+        template
+            .render_string(item)
+            .map_err(|error| {
+                emit!(TemplateRenderingError {
+                    error,
+                    field: Some(field),
+                    drop_event: true,
+                });
+            })
+            .ok()
+    }
+}
+
+impl Partitioner for KeyPartitioner {
+    type Item = Event;
+    type Key = Option<PartitionKey>;
+
+    fn partition(&self, item: &Self::Item) -> Self::Key {
+        let dbname = Self::render(&self.dbname, item, "dbname_key")?;
+        let table = Self::render(&self.table, item, "table_key")?;
+        let pipeline_name = Self::render(&self.pipeline_name, item, "pipeline_name")?;
+        let pipeline_version = self
+            .pipeline_version
+            .as_ref()
+            .and_then(|template| Self::render(template, item, "pipeline_version"));
+        Some(PartitionKey {
+            dbname,
+            table,
+            pipeline_name,
+            pipeline_version,
+        })
+    }
+}
+
+/// GreptimeDB logs HTTP request builder.
+#[derive(Debug, Clone)]
+pub(super) struct GreptimeDBLogsHttpRequestBuilder {
+    pub(super) endpoint: String,
+    pub(super) auth: Option<Auth>,
+    pub(super) encoder: (Transformer, Encoder<Framer>),
+    pub(super) compression: Compression,
+    pub(super) extra_params: Option<HashMap<String, String>>,
+}
+
+fn prepare_log_ingester_url(
+    endpoint: &str,
+    db: &str,
+    table: &str,
+    metadata: &PartitionKey,
+    extra_params: &Option<HashMap<String, String>>,
+) -> String {
+    let path = format!("{}/v1/events/logs", endpoint);
+    let mut url = url::Url::parse(&path).unwrap();
+    let mut url_builder = url.query_pairs_mut();
+    url_builder
+        .append_pair("db", &db)
+        .append_pair("table", &table)
+        .append_pair("pipeline_name", &metadata.pipeline_name);
+
+    if let Some(pipeline_version) = metadata.pipeline_version.as_ref() {
+        url_builder.append_pair("pipeline_version", pipeline_version);
+    }
+
+    if let Some(extra_params) = extra_params.as_ref() {
+        for (key, value) in extra_params.iter() {
+            url_builder.append_pair(key, value);
+        }
+    }
+
+    url_builder.finish().to_string()
+}
+
+impl HttpServiceRequestBuilder<PartitionKey> for GreptimeDBLogsHttpRequestBuilder {
+    fn build(&self, mut request: HttpRequest<PartitionKey>) -> Result<Request<Bytes>, Error> {
+        let metadata = request.get_additional_metadata();
+        let table = metadata.table.clone();
+        let db = metadata.dbname.clone();
+
+        // prepare url
+        let url = prepare_log_ingester_url(
+            self.endpoint.as_str(),
+            &db,
+            &table,
+            &metadata,
+            &self.extra_params,
+        );
+
+        // prepare body
+        let payload = request.take_payload();
+
+        let mut builder = Request::post(&url)
+            .header(CONTENT_TYPE, "application/json")
+            .header(CONTENT_LENGTH, payload.len());
+
+        if let Some(ce) = self.compression.content_encoding() {
+            builder = builder.header(CONTENT_ENCODING, ce);
+        }
+
+        if let Some(auth) = self.auth.clone() {
+            builder = auth.apply_builder(builder);
+        }
+
+        builder
+            .body(payload)
+            .context(HTTPRequestBuilderSnafu)
+            .map_err(Into::into)
+    }
+}
+
+impl RequestBuilder<(PartitionKey, Vec<Event>)> for GreptimeDBLogsHttpRequestBuilder {
+    type Metadata = (PartitionKey, EventFinalizers);
+    type Events = Vec<Event>;
+    type Encoder = (Transformer, Encoder<Framer>);
+    type Payload = Bytes;
+    type Request = HttpRequest<PartitionKey>;
+    type Error = std::io::Error;
+
+    fn compression(&self) -> Compression {
+        self.compression
+    }
+
+    fn encoder(&self) -> &Self::Encoder {
+        &self.encoder
+    }
+
+    fn split_input(
+        &self,
+        input: (PartitionKey, Vec<Event>),
+    ) -> (Self::Metadata, RequestMetadataBuilder, Self::Events) {
+        let (key, mut events) = input;
+
+        let finalizers = events.take_finalizers();
+        let builder = RequestMetadataBuilder::from_events(&events);
+        ((key, finalizers), builder, events)
+    }
+
+    fn build_request(
+        &self,
+        metadata: Self::Metadata,
+        request_metadata: RequestMetadata,
+        payload: EncodeResult<Self::Payload>,
+    ) -> Self::Request {
+        let (key, finalizers) = metadata;
+        HttpRequest::new(
+            payload.into_payload(),
+            finalizers,
+            request_metadata,
+            PartitionKey {
+                dbname: key.dbname,
+                table: key.table,
+                pipeline_name: key.pipeline_name,
+                pipeline_version: key.pipeline_version,
+            },
+        )
+    }
+}
+
+pub(super) async fn http_healthcheck(
+    client: HttpClient,
+    endpoint: String,
+    auth: Option<Auth>,
+) -> crate::Result<()> {
+    let uri = format!("{endpoint}/health");
+    let mut request = Request::get(uri).body(Body::empty()).unwrap();
+
+    if let Some(auth) = auth {
+        auth.apply(&mut request);
+    }
+
+    let response = client.send(request).await?;
+
+    match response.status() {
+        StatusCode::OK => Ok(()),
+        status => Err(HealthcheckError::UnexpectedStatus { status }.into()),
+    }
+}
+
+/// GreptimeDB HTTP retry logic.
+#[derive(Clone, Default)]
+pub(super) struct GreptimeDBHttpRetryLogic {
+    inner: HttpRetryLogic,
+}
+
+impl RetryLogic for GreptimeDBHttpRetryLogic {
+    type Error = HttpError;
+    type Response = HttpResponse;
+
+    fn is_retriable_error(&self, error: &Self::Error) -> bool {
+        error.is_retriable()
+    }
+
+    fn should_retry_response(&self, response: &Self::Response) -> RetryAction {
+        self.inner.should_retry_response(&response.http_response)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_prepare_url() {
+        let endpoint = "http://localhost:8080";
+        let db = "test_db";
+        let table = "test_table";
+        let metadata = PartitionKey {
+            dbname: "test_db".to_string(),
+            table: "test_table".to_string(),
+            pipeline_name: "test_pipeline".to_string(),
+            pipeline_version: Some("test_version".to_string()),
+        };
+        let extra_params = Some(
+            vec![("param1", "value1"), ("param2", "value2")]
+                .into_iter()
+                .map(|(k, v)| (k.to_string(), v.to_string()))
+                .collect(),
+        );
+
+        let url = prepare_log_ingester_url(&endpoint, db, table, &metadata, &extra_params);
+        assert_eq!(
+            url,
+            "http://localhost:8080/v1/events/logs?db=test_db&table=test_table&pipeline_name=test_pipeline&pipeline_version=test_version&param1=value1&param2=value2"
+        );
+    }
+}
diff --git a/src/sinks/greptimedb/logs/integration_tests.rs b/src/sinks/greptimedb/logs/integration_tests.rs
new file mode 100644
index 00000000000000..fad6e0c8e24bf1
--- /dev/null
+++ b/src/sinks/greptimedb/logs/integration_tests.rs
@@ -0,0 +1,127 @@
+use chrono::{DateTime, Utc};
+use futures::stream;
+use vector_lib::event::{Event, LogEvent};
+
+use crate::sinks::util::test::load_sink;
+use crate::{
+    config::{SinkConfig, SinkContext},
+    test_util::{
+        components::{run_and_assert_sink_compliance, SINK_TAGS},
+        trace_init,
+    },
+};
+
+use crate::sinks::greptimedb::logs::config::GreptimeDBLogsConfig;
+
+#[tokio::test]
+async fn test_greptimedb_logs_sink() {
+    trace_init();
+    let greptimedb_http_endpoint =
+        std::env::var("GREPTIMEDB_HTTP").unwrap_or_else(|_| "http://localhost:4000".to_owned());
+    let cfg = format!(
+        r#"endpoint= "{}"
+table = "logs"
+dbname = "public"
+pipeline_name = "test"
+"#,
+        &greptimedb_http_endpoint
+    );
+
+    // This is a minimal config that is required to run the sink
+    let pipeline = r#"
+processors:
+  - date:
+      field: timestamp
+      formats:
+        - "%Y-%m-%dT%H:%M:%S%.9fZ"
+      ignore_missing: true
+
+transform:
+  - fields:
+      - name
+      - namespace
+      - kind
+      - message   
+    type: string
+  - field: timestamp
+    type: time
+    index: timestamp
+    "#;
+
+    let (config, _) = load_sink::<GreptimeDBLogsConfig>(&cfg).unwrap();
+    let (sink, _hc) = config.build(SinkContext::default()).await.unwrap();
+
+    let client = GreptimeClient::new(greptimedb_http_endpoint.clone());
+
+    // Create a pipeline
+    client.create_pipeline("test", pipeline).await;
+
+    // Drop the table and data inside
+    client.query("DROP TABLE logs").await;
+
+    let base_time = Utc::now();
+    let events: Vec<_> = (0..10).map(|idx| create_event(idx, base_time)).collect();
+
+    run_and_assert_sink_compliance(sink, stream::iter(events), &SINK_TAGS).await;
+
+    let query_response = client.query("SELECT * FROM logs").await;
+    let result: serde_json::Value =
+        serde_json::from_str(&query_response).expect("Invalid json returned from greptimedb query");
+    assert_eq!(
+        result
+            .pointer("/output/0/records/rows")
+            .and_then(|v| v.as_array())
+            .expect("Error getting greptimedb response array")
+            .len(),
+        10
+    )
+}
+
+struct GreptimeClient {
+    client: reqwest::Client,
+    endpoint: String,
+}
+
+impl GreptimeClient {
+    fn new(endpoint: String) -> Self {
+        GreptimeClient {
+            client: reqwest::Client::builder().build().unwrap(),
+            endpoint,
+        }
+    }
+
+    async fn create_pipeline(&self, pipeline_name: &str, pipeline_content: &str) {
+        self.client
+            .post(&format!(
+                "{}/v1/events/pipelines/{}",
+                self.endpoint, pipeline_name
+            ))
+            .header("Content-Type", "application/x-yaml")
+            .body(String::from(pipeline_content))
+            .send()
+            .await
+            .unwrap();
+    }
+
+    async fn query(&self, sql: &str) -> String {
+        self.client
+            .get(&format!("{}/v1/sql", self.endpoint))
+            .query(&[("sql", sql)])
+            .send()
+            .await
+            .unwrap()
+            .text()
+            .await
+            .expect("Fetch json from greptimedb failed")
+    }
+}
+
+fn create_event(i: i32, base_time: DateTime<Utc>) -> Event {
+    let mut event = LogEvent::default();
+    event.insert("message", format!("test message {}", i));
+    event.insert("timestamp", base_time);
+    event.insert("name", "test");
+    event.insert("namespace", "default");
+    event.insert("kind", "test");
+    Event::Log(event)
+}
diff --git a/src/sinks/greptimedb/logs/mod.rs b/src/sinks/greptimedb/logs/mod.rs
new file mode 100644
index 00000000000000..2e2c7e83a27178
--- /dev/null
+++ b/src/sinks/greptimedb/logs/mod.rs
@@ -0,0 +1,20 @@
+//! `GreptimeDB` log sink for vector.
+//!
+//! This sink writes Vector's metric data into
+//! [GreptimeDB](https://github.com/greptimeteam/greptimedb), a cloud-native
+//! time-series database. It uses GreptimeDB's logs http API
+//!
+//! This sink transforms metrics into GreptimeDB table using following rules:
+//!
+//! - Table name: `{namespace}_{metric_name}`. If the metric doesn't have a
+//! namespace, we will use metric_name for table name.
+//! - Timestamp: timestamp is stored as a column called `ts`.
+//! - Tags: log tags are stored as string columns with its name as column
+//! name
+//!
+
+mod config;
+mod http_request_builder;
+mod sink;
+#[cfg(all(test, feature = "greptimedb-logs-integration-tests"))]
+mod integration_tests;
diff --git a/src/sinks/greptimedb/logs/sink.rs b/src/sinks/greptimedb/logs/sink.rs
new file mode 100644
index 00000000000000..b1fa07e4af0350
--- /dev/null
+++ b/src/sinks/greptimedb/logs/sink.rs
@@ -0,0 +1,88 @@
+use crate::sinks::greptimedb::logs::http_request_builder::{
+    GreptimeDBLogsHttpRequestBuilder, KeyPartitioner, PartitionKey,
+};
+use crate::sinks::prelude::*;
+use crate::sinks::util::http::HttpRequest;
+
+/// A sink that ingests logs into GreptimeDB.
+pub struct GreptimeDBLogsHttpSink<S> {
+    batcher_settings: BatcherSettings,
+    service: S,
+    dbname: Template,
+    table: Template,
+    pipeline_name: Template,
+    pipeline_version: Option<Template>,
+    request_builder: GreptimeDBLogsHttpRequestBuilder,
+}
+
+impl<S> GreptimeDBLogsHttpSink<S>
+where
+    S: Service<HttpRequest<PartitionKey>> + Send + 'static,
+    S::Future: Send + 'static,
+    S::Response: DriverResponse + Send + 'static,
+    S::Error: std::fmt::Debug + Into<crate::Error> + Send,
+{
+    pub const fn new(
+        batcher_settings: BatcherSettings,
+        service: S,
+        dbname: Template,
+        table: Template,
+        pipeline_name: Template,
+        pipeline_version: Option<Template>,
+        request_builder: GreptimeDBLogsHttpRequestBuilder,
+    ) -> Self {
+        Self {
+            batcher_settings,
+            service,
+            dbname,
+            table,
+            pipeline_name,
+            pipeline_version,
+            request_builder,
+        }
+    }
+
+    async fn run_inner(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
+        let batcher_settings = self.batcher_settings;
+        input
+            .batched_partitioned(
+                KeyPartitioner::new(
+                    self.dbname,
+                    self.table,
+                    self.pipeline_name,
+                    self.pipeline_version,
+                ),
+                || batcher_settings.as_byte_size_config(),
+            )
+            .filter_map(|(key, batch)| async move { key.map(move |k| (k, batch)) })
+            .request_builder(
+                default_request_builder_concurrency_limit(),
+                self.request_builder,
+            )
+            .filter_map(|request| async {
+                match request {
+                    Err(error) => {
+                        emit!(SinkRequestBuildError { error });
+                        None
+                    }
+                    Ok(req) => Some(req),
+                }
+            })
+            .into_driver(self.service)
+            .run()
+            .await
+    }
+}
+
+#[async_trait::async_trait]
+impl<S> StreamSink<Event> for GreptimeDBLogsHttpSink<S>
+where
+    S: Service<HttpRequest<PartitionKey>> + Send + 'static,
+    S::Future: Send + 'static,
+    S::Response: DriverResponse + Send + 'static,
+    S::Error: std::fmt::Debug + Into<crate::Error> + Send,
+{
+    async fn run(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
+        self.run_inner(input).await
+    }
+}
diff --git a/src/sinks/greptimedb/batch.rs b/src/sinks/greptimedb/metrics/batch.rs
similarity index 92%
rename from src/sinks/greptimedb/batch.rs
rename to src/sinks/greptimedb/metrics/batch.rs
index 7bfd7bcce28630..60cd12bc6399a0 100644
--- a/src/sinks/greptimedb/batch.rs
+++ b/src/sinks/greptimedb/metrics/batch.rs
@@ -8,11 +8,12 @@ use super::request_builder::{
 const F64_BYTE_SIZE: usize = 8;
 const I64_BYTE_SIZE: usize = 8;
 
+/// GreptimeDBBatchSizer is a batch sizer for metrics.
 #[derive(Default)]
-pub(super) struct GreptimeDBBatchSizer;
+pub struct GreptimeDBBatchSizer;
 
 impl GreptimeDBBatchSizer {
-    pub(super) fn estimated_size_of(&self, item: &Metric) -> usize {
+    pub fn estimated_size_of(&self, item: &Metric) -> usize {
         // Metric name.
         item.series().name().name().len()
         // Metric namespace, with an additional 1 to account for the namespace separator.
diff --git a/src/sinks/greptimedb/metrics/config.rs b/src/sinks/greptimedb/metrics/config.rs
new file mode 100644
index 00000000000000..3d54e16d5e139a
--- /dev/null
+++ b/src/sinks/greptimedb/metrics/config.rs
@@ -0,0 +1,127 @@
+use vector_lib::configurable::configurable_component;
+use vector_lib::sensitive_string::SensitiveString;
+
+use super::request::GreptimeDBGrpcRetryLogic;
+
+use super::service::{healthcheck, GreptimeDBGrpcService};
+use super::sink;
+use crate::sinks::greptimedb::{default_dbname, GreptimeDBDefaultBatchSettings};
+use crate::sinks::prelude::*;
+
+/// Configuration items for GreptimeDB
+#[configurable_component(sink("greptimedb", "Ingest metrics data into GreptimeDB."))]
+#[derive(Clone, Debug, Derivative)]
+#[derivative(Default)]
+#[serde(deny_unknown_fields)]
+pub struct GreptimeDBConfig {
+    /// The GreptimeDB [database][database] name to connect.
+    ///
+    /// Default to `public`, the default database of GreptimeDB.
+    ///
+    /// Database can be created via `create database` statement on
+    /// GreptimeDB. If you are using GreptimeCloud, use `dbname` from the
+    /// connection information of your instance.
+    ///
+    /// [database]: https://docs.greptime.com/user-guide/concepts/key-concepts#database
+    #[configurable(metadata(docs::examples = "public"))]
+    #[derivative(Default(value = "default_dbname()"))]
+    #[serde(default = "default_dbname")]
+    pub dbname: String,
+    /// The host and port of GreptimeDB gRPC service.
+    ///
+    /// This sink uses GreptimeDB's gRPC interface for data ingestion. By
+    /// default, GreptimeDB listens to port 4001 for gRPC protocol.
+    ///
+    /// The address _must_ include a port.
+    #[configurable(metadata(docs::examples = "example.com:4001"))]
+    #[configurable(metadata(
+        docs::examples = "1nge17d2r3ns.ap-southeast-1.aws.greptime.cloud:4001"
+    ))]
+    pub endpoint: String,
+    /// The username for your GreptimeDB instance.
+    ///
+    /// This is required if your instance has authentication enabled.
+    #[configurable(metadata(docs::examples = "username"))]
+    #[serde(default)]
+    pub username: Option<String>,
+    /// The password for your GreptimeDB instance.
+    ///
+    /// This is required if your instance has authentication enabled.
+    #[configurable(metadata(docs::examples = "password"))]
+    #[serde(default)]
+    pub password: Option<SensitiveString>,
+    /// Set gRPC compression encoding for the request
+    /// Default to none, `gzip` or `zstd` is supported.
+    #[configurable(metadata(docs::examples = "grpc_compression"))]
+    #[serde(default)]
+    pub grpc_compression: Option<String>,
+
+    #[configurable(derived)]
+    #[serde(default)]
+    pub request: TowerRequestConfig,
+
+    #[configurable(derived)]
+    #[serde(default)]
+    pub(crate) batch: BatchConfig<GreptimeDBDefaultBatchSettings>,
+
+    #[configurable(derived)]
+    #[serde(
+        default,
+        deserialize_with = "crate::serde::bool_or_struct",
+        skip_serializing_if = "crate::serde::is_default"
+    )]
+    pub acknowledgements: AcknowledgementsConfig,
+
+    #[configurable(derived)]
+    pub tls: Option<TlsConfig>,
+}
+
+impl_generate_config_from_default!(GreptimeDBConfig);
+
+#[typetag::serde(name = "greptimedb")]
+#[async_trait::async_trait]
+impl SinkConfig for GreptimeDBConfig {
+    async fn build(&self, _cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
+        let request_settings = self.request.into_settings();
+        let service = ServiceBuilder::new()
+            .settings(request_settings, GreptimeDBGrpcRetryLogic)
+            .service(GreptimeDBGrpcService::try_new(self)?);
+        let sink = sink::GreptimeDBGrpcSink {
+            service,
+            batch_settings: self.batch.into_batcher_settings()?,
+        };
+
+        let healthcheck = healthcheck(self)?;
+        Ok((VectorSink::from_event_streamsink(sink), healthcheck))
+    }
+
+    fn input(&self) -> Input {
+        Input::metric()
+    }
+
+    fn acknowledgements(&self) -> &AcknowledgementsConfig {
+        &self.acknowledgements
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use indoc::indoc;
+
+    use super::*;
+
+    #[test]
+    fn generate_config() {
+        crate::test_util::test_generate_config::<GreptimeDBConfig>();
+    }
+
+    #[test]
+    fn test_config_with_username() {
+        let config = indoc! {r#"
+            endpoint = "foo-bar.ap-southeast-1.aws.greptime.cloud:4001"
+            dbname = "foo-bar"
+        "#};
+
+        toml::from_str::<GreptimeDBConfig>(config).unwrap();
+    }
+}
diff --git a/src/sinks/greptimedb/integration_tests.rs b/src/sinks/greptimedb/metrics/integration_tests.rs
similarity index 97%
rename from src/sinks/greptimedb/integration_tests.rs
rename to src/sinks/greptimedb/metrics/integration_tests.rs
index 6b342a0e455fa3..6f1aca07b2b462 100644
--- a/src/sinks/greptimedb/integration_tests.rs
+++ b/src/sinks/greptimedb/metrics/integration_tests.rs
@@ -12,7 +12,7 @@ use crate::{
     },
 };
 
-use super::GreptimeDBConfig;
+use crate::sinks::greptimedb::metrics::config::GreptimeDBConfig;
 
 #[tokio::test]
 async fn test_greptimedb_sink() {
diff --git a/src/sinks/greptimedb/metrics/mod.rs b/src/sinks/greptimedb/metrics/mod.rs
new file mode 100644
index 00000000000000..6c35341299126f
--- /dev/null
+++ b/src/sinks/greptimedb/metrics/mod.rs
@@ -0,0 +1,30 @@
+//! `GreptimeDB` grpc sink for vector.
+//!
+//! This sink writes Vector's metric data into
+//! [GreptimeDB](https://github.com/greptimeteam/greptimedb), a cloud-native
+//! time-series database. It uses GreptimeDB's [gRPC
+//! API](https://docs.greptime.com/user-guide/write-data/grpc) and GreptimeDB's
+//! [rust client](https://github.com/GreptimeTeam/greptimedb-ingester-rust).
+//!
+//! This sink transforms metrics into GreptimeDB table using following rules:
+//!
+//! - Table name: `{namespace}_{metric_name}`. If the metric doesn't have a
+//! namespace, we will use metric_name for table name.
+//! - Timestamp: timestamp is stored as a column called `ts`.
+//! - Tags: metric tags are stored as string columns with its name as column
+//! name
+//! - Counter and Gauge: the value of counter and gauge are stored in a column
+//! called `val`
+//! - Set: the number of set items is stored in a column called `val`.
+//! - Distribution, Histogram and Summary, Sketch: Statistical attributes like
+//! `sum`, `count`, "max", "min", quantiles and buckets are stored as columns.
+//!
+
+mod batch;
+mod config;
+#[cfg(all(test, feature = "greptimedb-integration-tests"))]
+mod integration_tests;
+mod request;
+mod request_builder;
+mod service;
+mod sink;
diff --git a/src/sinks/greptimedb/metrics/request.rs b/src/sinks/greptimedb/metrics/request.rs
new file mode 100644
index 00000000000000..68208d84133003
--- /dev/null
+++ b/src/sinks/greptimedb/metrics/request.rs
@@ -0,0 +1,97 @@
+use super::batch::GreptimeDBBatchSizer;
+use super::request_builder::metric_to_insert_request;
+use crate::sinks::prelude::*;
+use greptimedb_ingester::api::v1::*;
+use greptimedb_ingester::Error as GreptimeError;
+use std::num::NonZeroUsize;
+
+use vector_lib::event::Metric;
+
+/// GreptimeDBGrpcRequest is a wrapper around the RowInsertRequests
+/// that is used to send metrics to GreptimeDB.
+/// It also contains the finalizers and metadata that are used to
+#[derive(Clone)]
+pub(super) struct GreptimeDBGrpcRequest {
+    pub(super) items: RowInsertRequests,
+    pub(super) finalizers: EventFinalizers,
+    pub(super) metadata: RequestMetadata,
+}
+
+impl GreptimeDBGrpcRequest {
+    // convert metrics event to GreptimeDBGrpcRequest
+    pub(super) fn from_metrics(metrics: Vec<Metric>) -> Self {
+        let mut items = Vec::with_capacity(metrics.len());
+        let mut finalizers = EventFinalizers::default();
+        let mut request_metadata_builder = RequestMetadataBuilder::default();
+
+        let sizer = GreptimeDBBatchSizer;
+        let mut estimated_request_size = 0;
+        for mut metric in metrics.into_iter() {
+            finalizers.merge(metric.take_finalizers());
+            estimated_request_size += sizer.estimated_size_of(&metric);
+
+            request_metadata_builder.track_event(metric.clone());
+
+            items.push(metric_to_insert_request(metric));
+        }
+
+        let request_size =
+            NonZeroUsize::new(estimated_request_size).expect("request should never be zero length");
+
+        GreptimeDBGrpcRequest {
+            items: RowInsertRequests { inserts: items },
+            finalizers,
+            metadata: request_metadata_builder.with_request_size(request_size),
+        }
+    }
+}
+
+impl Finalizable for GreptimeDBGrpcRequest {
+    fn take_finalizers(&mut self) -> EventFinalizers {
+        std::mem::take(&mut self.finalizers)
+    }
+}
+
+impl MetaDescriptive for GreptimeDBGrpcRequest {
+    fn get_metadata(&self) -> &RequestMetadata {
+        &self.metadata
+    }
+
+    fn metadata_mut(&mut self) -> &mut RequestMetadata {
+        &mut self.metadata
+    }
+}
+
+/// GreptimeDBGrpcBatchOutput is the output of the [`GreptimeDBGrpcService`]
+#[derive(Debug)]
+pub struct GreptimeDBGrpcBatchOutput {
+    pub _item_count: u32,
+    pub metadata: RequestMetadata,
+}
+
+impl DriverResponse for GreptimeDBGrpcBatchOutput {
+    fn event_status(&self) -> EventStatus {
+        EventStatus::Delivered
+    }
+
+    fn events_sent(&self) -> &GroupedCountByteSize {
+        self.metadata.events_estimated_json_encoded_byte_size()
+    }
+
+    fn bytes_sent(&self) -> Option<usize> {
+        Some(self.metadata.request_encoded_size())
+    }
+}
+
+/// GreptimeDBGrpcRetryLogic is the retry logic for the [`GreptimeDBGrpcSink`]
+#[derive(Clone, Default)]
+pub struct GreptimeDBGrpcRetryLogic;
+
+impl RetryLogic for GreptimeDBGrpcRetryLogic {
+    type Response = GreptimeDBGrpcBatchOutput;
+    type Error = GreptimeError;
+
+    fn is_retriable_error(&self, error: &Self::Error) -> bool {
+        error.is_retriable()
+    }
+}
diff --git a/src/sinks/greptimedb/request_builder.rs b/src/sinks/greptimedb/metrics/request_builder.rs
similarity index 98%
rename from src/sinks/greptimedb/request_builder.rs
rename to src/sinks/greptimedb/metrics/request_builder.rs
index 7c7b6c8189b14f..39001680f7f004 100644
--- a/src/sinks/greptimedb/request_builder.rs
+++ b/src/sinks/greptimedb/metrics/request_builder.rs
@@ -1,6 +1,6 @@
 use chrono::Utc;
-use greptimedb_client::api::v1::*;
-use greptimedb_client::helpers::values::*;
+use greptimedb_ingester::api::v1::*;
+use greptimedb_ingester::helpers::values::*;
 use vector_lib::event::metric::{Bucket, MetricSketch, Quantile, Sample};
 use vector_lib::event::{Metric, MetricValue};
 use vector_lib::metrics::AgentDDSketch;
@@ -11,33 +11,6 @@ pub(super) const DISTRIBUTION_QUANTILES: [f64; 5] = [0.5, 0.75, 0.90, 0.95, 0.99
 pub(super) const DISTRIBUTION_STAT_FIELD_COUNT: usize = 5;
 pub(super) const SUMMARY_STAT_FIELD_COUNT: usize = 2;
 
-fn f64_column(name: &str) -> ColumnSchema {
-    ColumnSchema {
-        column_name: name.to_owned(),
-        semantic_type: SemanticType::Field as i32,
-        datatype: ColumnDataType::Float64 as i32,
-        ..Default::default()
-    }
-}
-
-fn ts_column(name: &str) -> ColumnSchema {
-    ColumnSchema {
-        column_name: name.to_owned(),
-        semantic_type: SemanticType::Timestamp as i32,
-        datatype: ColumnDataType::TimestampMillisecond as i32,
-        ..Default::default()
-    }
-}
-
-fn tag_column(name: &str) -> ColumnSchema {
-    ColumnSchema {
-        column_name: name.to_owned(),
-        semantic_type: SemanticType::Tag as i32,
-        datatype: ColumnDataType::String as i32,
-        ..Default::default()
-    }
-}
-
 fn encode_f64_value(
     name: &str,
     value: f64,
@@ -48,7 +21,7 @@ fn encode_f64_value(
     columns.push(f64_value(value));
 }
 
-pub(super) fn metric_to_insert_request(metric: Metric) -> RowInsertRequest {
+pub fn metric_to_insert_request(metric: Metric) -> RowInsertRequest {
     let ns = metric.namespace();
     let metric_name = metric.name();
     let table_name = if let Some(ns) = ns {
@@ -190,6 +163,33 @@ fn encode_sketch(sketch: &AgentDDSketch, schema: &mut Vec<ColumnSchema>, columns
     }
 }
 
+fn f64_column(name: &str) -> ColumnSchema {
+    ColumnSchema {
+        column_name: name.to_owned(),
+        semantic_type: SemanticType::Field as i32,
+        datatype: ColumnDataType::Float64 as i32,
+        ..Default::default()
+    }
+}
+
+fn ts_column(name: &str) -> ColumnSchema {
+    ColumnSchema {
+        column_name: name.to_owned(),
+        semantic_type: SemanticType::Timestamp as i32,
+        datatype: ColumnDataType::TimestampMillisecond as i32,
+        ..Default::default()
+    }
+}
+
+fn tag_column(name: &str) -> ColumnSchema {
+    ColumnSchema {
+        column_name: name.to_owned(),
+        semantic_type: SemanticType::Tag as i32,
+        datatype: ColumnDataType::String as i32,
+        ..Default::default()
+    }
+}
+
 #[cfg(test)]
 mod tests {
 
diff --git a/src/sinks/greptimedb/metrics/service.rs b/src/sinks/greptimedb/metrics/service.rs
new file mode 100644
index 00000000000000..8ea79fa9b83ecc
--- /dev/null
+++ b/src/sinks/greptimedb/metrics/service.rs
@@ -0,0 +1,139 @@
+use crate::sinks::prelude::*;
+use std::sync::Arc;
+
+use super::config::GreptimeDBConfig;
+use super::request::{GreptimeDBGrpcBatchOutput, GreptimeDBGrpcRequest};
+use greptimedb_ingester::api::v1::auth_header::AuthScheme;
+use greptimedb_ingester::api::v1::*;
+use greptimedb_ingester::{channel_manager::*, ClientBuilder, Compression};
+use greptimedb_ingester::{Client, Database, Error as GreptimeError};
+use std::task::Poll;
+use vector_lib::sensitive_string::SensitiveString;
+
+#[derive(Debug, Clone)]
+pub struct GreptimeDBGrpcService {
+    /// the client that connects to greptimedb
+    client: Arc<Database>,
+}
+
+fn new_client_from_config(config: &GreptimeDBGrpcServiceConfig) -> crate::Result<Client> {
+    let mut builder = ClientBuilder::default().peers(vec![&config.endpoint]);
+
+    if let Some(compression) = config.compression.as_ref() {
+        let compression = match compression.as_str() {
+            "gzip" => Compression::Gzip,
+            "zstd" => Compression::Zstd,
+            _ => {
+                warn!(message = "Unknown gRPC compression type: {compression}, disabled.");
+                Compression::None
+            }
+        };
+        builder = builder.compression(compression);
+    }
+
+    if let Some(tls_config) = &config.tls {
+        let channel_config = ChannelConfig {
+            client_tls: Some(try_from_tls_config(tls_config)?),
+            ..Default::default()
+        };
+
+        builder = builder
+            .channel_manager(ChannelManager::with_tls_config(channel_config).map_err(Box::new)?);
+    }
+
+    Ok(builder.build())
+}
+
+fn try_from_tls_config(tls_config: &TlsConfig) -> crate::Result<ClientTlsOption> {
+    if tls_config.key_pass.is_some()
+        || tls_config.alpn_protocols.is_some()
+        || tls_config.verify_certificate.is_some()
+        || tls_config.verify_hostname.is_some()
+    {
+        warn!(message = "TlsConfig: key_pass, alpn_protocols, verify_certificate and verify_hostname are not supported by greptimedb client at the moment.");
+    }
+
+    Ok(ClientTlsOption {
+        server_ca_cert_path: tls_config.ca_file.clone(),
+        client_cert_path: tls_config.crt_file.clone(),
+        client_key_path: tls_config.key_file.clone(),
+    })
+}
+
+impl GreptimeDBGrpcService {
+    pub fn try_new(config: impl Into<GreptimeDBGrpcServiceConfig>) -> crate::Result<Self> {
+        let config = config.into();
+
+        let grpc_client = new_client_from_config(&config)?;
+
+        let mut client = Database::new_with_dbname(&config.dbname, grpc_client);
+
+        if let (Some(username), Some(password)) = (&config.username, &config.password) {
+            client.set_auth(AuthScheme::Basic(Basic {
+                username: username.to_owned(),
+                password: password.clone().into(),
+            }))
+        };
+
+        Ok(GreptimeDBGrpcService {
+            client: Arc::new(client),
+        })
+    }
+}
+
+impl Service<GreptimeDBGrpcRequest> for GreptimeDBGrpcService {
+    type Response = GreptimeDBGrpcBatchOutput;
+    type Error = GreptimeError;
+    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
+
+    fn poll_ready(&mut self, _cx: &mut std::task::Context) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    // Convert vector metrics into GreptimeDB format and send them in batch
+    fn call(&mut self, req: GreptimeDBGrpcRequest) -> Self::Future {
+        let client = Arc::clone(&self.client);
+
+        Box::pin(async move {
+            let metadata = req.metadata;
+            let result = client.row_insert(req.items).await?;
+
+            Ok(GreptimeDBGrpcBatchOutput {
+                _item_count: result,
+                metadata,
+            })
+        })
+    }
+}
+
+/// Configuration for the GreptimeDB gRPC service
+pub(super) struct GreptimeDBGrpcServiceConfig {
+    endpoint: String,
+    dbname: String,
+    username: Option<String>,
+    password: Option<SensitiveString>,
+    compression: Option<String>,
+    tls: Option<TlsConfig>,
+}
+
+impl From<&GreptimeDBConfig> for GreptimeDBGrpcServiceConfig {
+    fn from(val: &GreptimeDBConfig) -> Self {
+        GreptimeDBGrpcServiceConfig {
+            endpoint: val.endpoint.clone(),
+            dbname: val.dbname.clone(),
+            username: val.username.clone(),
+            password: val.password.clone(),
+            compression: val.grpc_compression.clone(),
+            tls: val.tls.clone(),
+        }
+    }
+}
+
+pub(super) fn healthcheck(
+    config: impl Into<GreptimeDBGrpcServiceConfig>,
+) -> crate::Result<Healthcheck> {
+    let config = config.into();
+    let client = new_client_from_config(&config)?;
+
+    Ok(async move { client.health_check().await.map_err(|error| error.into()) }.boxed())
+}
diff --git a/src/sinks/greptimedb/sink.rs b/src/sinks/greptimedb/metrics/sink.rs
similarity index 73%
rename from src/sinks/greptimedb/sink.rs
rename to src/sinks/greptimedb/metrics/sink.rs
index 8268fd341c7c01..2fea39014228cf 100644
--- a/src/sinks/greptimedb/sink.rs
+++ b/src/sinks/greptimedb/metrics/sink.rs
@@ -4,12 +4,14 @@ use futures::StreamExt;
 use futures_util::stream::BoxStream;
 use vector_lib::event::{Metric, MetricValue};
 
+use super::request::GreptimeDBGrpcRetryLogic;
+use super::service::GreptimeDBGrpcService;
+use crate::sinks::greptimedb::metrics::request::GreptimeDBGrpcRequest;
 use crate::sinks::prelude::*;
 use crate::sinks::util::buffer::metrics::MetricNormalize;
 use crate::sinks::util::buffer::metrics::MetricSet;
 
 use super::batch::GreptimeDBBatchSizer;
-use super::service::{GreptimeDBRequest, GreptimeDBRetryLogic, GreptimeDBService};
 
 #[derive(Clone, Debug, Default)]
 pub struct GreptimeDBMetricNormalize;
@@ -25,12 +27,14 @@ impl MetricNormalize for GreptimeDBMetricNormalize {
     }
 }
 
-pub struct GreptimeDBSink {
-    pub(super) service: Svc<GreptimeDBService, GreptimeDBRetryLogic>,
+/// GreptimeDBGrpcSink is a sink that sends metrics to GreptimeDB via gRPC.
+/// It uses the `GreptimeDBGrpcService` to send the metrics.
+pub struct GreptimeDBGrpcSink {
+    pub(super) service: Svc<GreptimeDBGrpcService, GreptimeDBGrpcRetryLogic>,
     pub(super) batch_settings: BatcherSettings,
 }
 
-impl GreptimeDBSink {
+impl GreptimeDBGrpcSink {
     async fn run_inner(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
         input
             .map(|event| event.into_metric())
@@ -39,7 +43,7 @@ impl GreptimeDBSink {
                 self.batch_settings
                     .as_item_size_config(GreptimeDBBatchSizer),
             )
-            .map(GreptimeDBRequest::from_metrics)
+            .map(GreptimeDBGrpcRequest::from_metrics)
             .into_driver(self.service)
             .protocol("grpc")
             .run()
@@ -48,7 +52,7 @@ impl GreptimeDBSink {
 }
 
 #[async_trait]
-impl StreamSink<Event> for GreptimeDBSink {
+impl StreamSink<Event> for GreptimeDBGrpcSink {
     async fn run(self: Box<Self>, input: BoxStream<'_, Event>) -> Result<(), ()> {
         self.run_inner(input).await
     }
diff --git a/src/sinks/greptimedb/mod.rs b/src/sinks/greptimedb/mod.rs
index a8429b67c249d5..2a907aee4a40e2 100644
--- a/src/sinks/greptimedb/mod.rs
+++ b/src/sinks/greptimedb/mod.rs
@@ -1,165 +1,22 @@
-//! `GreptimeDB` sink for vector.
-//!
-//! This sink writes Vector's metric data into
-//! [GreptimeDB](https://github.com/greptimeteam/greptimedb), a cloud-native
-//! time-series database. It uses GreptimeDB's [gRPC
-//! API](https://docs.greptime.com/user-guide/write-data/grpc) and GreptimeDB's
-//! [rust client](https://github.com/GreptimeTeam/greptimedb-client-rust).
-//!
-//! This sink transforms metrics into GreptimeDB table using following rules:
-//!
-//! - Table name: `{namespace}_{metric_name}`. If the metric doesn't have a
-//! namespace, we will use metric_name for table name.
-//! - Timestamp: timestamp is stored as a column called `ts`.
-//! - Tags: metric tags are stored as string columns with its name as column
-//! name
-//! - Counter and Gauge: the value of counter and gauge are stored in a column
-//! called `val`
-//! - Set: the number of set items is stored in a column called `val`.
-//! - Distribution, Histogram and Summary, Sketch: Statistical attributes like
-//! `sum`, `count`, "max", "min", quantiles and buckets are stored as columns.
-//!
-use vector_lib::sensitive_string::SensitiveString;
-
 use crate::sinks::prelude::*;
 
-use self::service::GreptimeDBRetryLogic;
+// sub level implementations
+mod logs;
+mod metrics;
+
+fn default_dbname() -> String {
+    greptimedb_ingester::DEFAULT_SCHEMA_NAME.to_string()
+}
 
-mod batch;
-#[cfg(all(test, feature = "greptimedb-integration-tests"))]
-mod integration_tests;
-mod request_builder;
-mod service;
-mod sink;
+fn default_dbname_template() -> Template {
+    Template::try_from(default_dbname()).unwrap()
+}
 
 #[derive(Clone, Copy, Debug, Default)]
-pub struct GreptimeDBDefaultBatchSettings;
+struct GreptimeDBDefaultBatchSettings;
 
 impl SinkBatchSettings for GreptimeDBDefaultBatchSettings {
     const MAX_EVENTS: Option<usize> = Some(20);
     const MAX_BYTES: Option<usize> = None;
     const TIMEOUT_SECS: f64 = 1.0;
 }
-
-fn default_dbname() -> String {
-    greptimedb_client::DEFAULT_SCHEMA_NAME.to_string()
-}
-
-/// Configuration items for GreptimeDB
-#[configurable_component(sink("greptimedb", "Ingest metrics data into GreptimeDB."))]
-#[derive(Clone, Debug, Derivative)]
-#[derivative(Default)]
-#[serde(deny_unknown_fields)]
-pub struct GreptimeDBConfig {
-    /// The GreptimeDB [database][database] name to connect.
-    ///
-    /// Default to `public`, the default database of GreptimeDB.
-    ///
-    /// Database can be created via `create database` statement on
-    /// GreptimeDB. If you are using GreptimeCloud, use `dbname` from the
-    /// connection information of your instance.
-    ///
-    /// [database]: https://docs.greptime.com/user-guide/concepts/key-concepts#database
-    #[configurable(metadata(docs::examples = "public"))]
-    #[derivative(Default(value = "default_dbname()"))]
-    #[serde(default = "default_dbname")]
-    pub dbname: String,
-    /// The host and port of GreptimeDB gRPC service.
-    ///
-    /// This sink uses GreptimeDB's gRPC interface for data ingestion. By
-    /// default, GreptimeDB listens to port 4001 for gRPC protocol.
-    ///
-    /// The address _must_ include a port.
-    #[configurable(metadata(docs::examples = "example.com:4001"))]
-    #[configurable(metadata(
-        docs::examples = "1nge17d2r3ns.ap-southeast-1.aws.greptime.cloud:4001"
-    ))]
-    pub endpoint: String,
-    /// The username for your GreptimeDB instance.
-    ///
-    /// This is required if your instance has authentication enabled.
-    #[configurable(metadata(docs::examples = "username"))]
-    #[serde(default)]
-    pub username: Option<String>,
-    /// The password for your GreptimeDB instance.
-    ///
-    /// This is required if your instance has authentication enabled.
-    #[configurable(metadata(docs::examples = "password"))]
-    #[serde(default)]
-    pub password: Option<SensitiveString>,
-
-    #[configurable(derived)]
-    #[serde(default)]
-    pub request: TowerRequestConfig,
-
-    #[configurable(derived)]
-    #[serde(default)]
-    pub batch: BatchConfig<GreptimeDBDefaultBatchSettings>,
-
-    #[configurable(derived)]
-    #[serde(
-        default,
-        deserialize_with = "crate::serde::bool_or_struct",
-        skip_serializing_if = "crate::serde::is_default"
-    )]
-    pub acknowledgements: AcknowledgementsConfig,
-
-    #[configurable(derived)]
-    pub tls: Option<TlsConfig>,
-}
-
-impl_generate_config_from_default!(GreptimeDBConfig);
-
-#[typetag::serde(name = "greptimedb")]
-#[async_trait::async_trait]
-impl SinkConfig for GreptimeDBConfig {
-    async fn build(&self, _cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
-        let request_settings = self.request.into_settings();
-        let service = ServiceBuilder::new()
-            .settings(request_settings, GreptimeDBRetryLogic)
-            .service(service::GreptimeDBService::try_new(self)?);
-        let sink = sink::GreptimeDBSink {
-            service,
-            batch_settings: self.batch.into_batcher_settings()?,
-        };
-
-        let healthcheck = healthcheck(self)?;
-        Ok((VectorSink::from_event_streamsink(sink), healthcheck))
-    }
-
-    fn input(&self) -> Input {
-        Input::metric()
-    }
-
-    fn acknowledgements(&self) -> &AcknowledgementsConfig {
-        &self.acknowledgements
-    }
-}
-
-fn healthcheck(config: &GreptimeDBConfig) -> crate::Result<super::Healthcheck> {
-    let client = service::new_client_from_config(config)?;
-
-    Ok(async move { client.health_check().await.map_err(|error| error.into()) }.boxed())
-}
-
-#[cfg(test)]
-mod tests {
-    use indoc::indoc;
-
-    use super::*;
-
-    #[test]
-    fn generate_config() {
-        crate::test_util::test_generate_config::<GreptimeDBConfig>();
-    }
-
-    #[test]
-    fn test_config_with_username() {
-        let config = indoc! {r#"
-            endpoint = "foo-bar.ap-southeast-1.aws.greptime.cloud:4001"
-            dbname = "foo-bar"
-        "#};
-
-        toml::from_str::<GreptimeDBConfig>(config).unwrap();
-    }
-}
diff --git a/src/sinks/greptimedb/service.rs b/src/sinks/greptimedb/service.rs
deleted file mode 100644
index 809f1fc124d312..00000000000000
--- a/src/sinks/greptimedb/service.rs
+++ /dev/null
@@ -1,177 +0,0 @@
-use std::num::NonZeroUsize;
-use std::sync::Arc;
-use std::task::Poll;
-
-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 vector_lib::event::Metric;
-
-use crate::sinks::prelude::*;
-
-use super::batch::GreptimeDBBatchSizer;
-use super::request_builder::metric_to_insert_request;
-use super::GreptimeDBConfig;
-
-#[derive(Clone, Default)]
-pub(super) struct GreptimeDBRetryLogic;
-
-impl RetryLogic for GreptimeDBRetryLogic {
-    type Response = GreptimeDBBatchOutput;
-    type Error = GreptimeError;
-
-    fn is_retriable_error(&self, error: &Self::Error) -> bool {
-        error.is_retriable()
-    }
-}
-
-#[derive(Clone)]
-pub(super) struct GreptimeDBRequest {
-    items: RowInsertRequests,
-    finalizers: EventFinalizers,
-    metadata: RequestMetadata,
-}
-
-impl GreptimeDBRequest {
-    pub(super) fn from_metrics(metrics: Vec<Metric>) -> Self {
-        let mut items = Vec::with_capacity(metrics.len());
-        let mut finalizers = EventFinalizers::default();
-        let mut request_metadata_builder = RequestMetadataBuilder::default();
-
-        let sizer = GreptimeDBBatchSizer;
-        let mut estimated_request_size = 0;
-        for mut metric in metrics.into_iter() {
-            finalizers.merge(metric.take_finalizers());
-            estimated_request_size += sizer.estimated_size_of(&metric);
-
-            request_metadata_builder.track_event(metric.clone());
-
-            items.push(metric_to_insert_request(metric));
-        }
-
-        let request_size =
-            NonZeroUsize::new(estimated_request_size).expect("request should never be zero length");
-
-        GreptimeDBRequest {
-            items: RowInsertRequests { inserts: items },
-            finalizers,
-            metadata: request_metadata_builder.with_request_size(request_size),
-        }
-    }
-}
-
-impl Finalizable for GreptimeDBRequest {
-    fn take_finalizers(&mut self) -> EventFinalizers {
-        std::mem::take(&mut self.finalizers)
-    }
-}
-
-impl MetaDescriptive for GreptimeDBRequest {
-    fn get_metadata(&self) -> &RequestMetadata {
-        &self.metadata
-    }
-
-    fn metadata_mut(&mut self) -> &mut RequestMetadata {
-        &mut self.metadata
-    }
-}
-
-#[derive(Debug)]
-pub struct GreptimeDBBatchOutput {
-    pub metadata: RequestMetadata,
-}
-
-impl DriverResponse for GreptimeDBBatchOutput {
-    fn event_status(&self) -> EventStatus {
-        EventStatus::Delivered
-    }
-
-    fn events_sent(&self) -> &GroupedCountByteSize {
-        self.metadata.events_estimated_json_encoded_byte_size()
-    }
-
-    fn bytes_sent(&self) -> Option<usize> {
-        Some(self.metadata.request_encoded_size())
-    }
-}
-
-#[derive(Debug, Clone)]
-pub struct GreptimeDBService {
-    /// the client that connects to greptimedb
-    client: Arc<Database>,
-}
-
-pub(crate) fn new_client_from_config(config: &GreptimeDBConfig) -> crate::Result<Client> {
-    if let Some(tls_config) = &config.tls {
-        let channel_config = ChannelConfig {
-            client_tls: Some(try_from_tls_config(tls_config)?),
-            ..Default::default()
-        };
-        Ok(Client::with_manager_and_urls(
-            ChannelManager::with_tls_config(channel_config).map_err(Box::new)?,
-            vec![&config.endpoint],
-        ))
-    } else {
-        Ok(Client::with_urls(vec![&config.endpoint]))
-    }
-}
-
-fn try_from_tls_config(tls_config: &TlsConfig) -> crate::Result<ClientTlsOption> {
-    if tls_config.key_pass.is_some()
-        || tls_config.alpn_protocols.is_some()
-        || tls_config.verify_certificate.is_some()
-        || tls_config.verify_hostname.is_some()
-    {
-        warn!(
-                    message = "TlsConfig: key_pass, alpn_protocols, verify_certificate and verify_hostname are not supported by greptimedb client at the moment."
-                );
-    }
-
-    Ok(ClientTlsOption {
-        server_ca_cert_path: tls_config.ca_file.clone(),
-        client_cert_path: tls_config.crt_file.clone(),
-        client_key_path: tls_config.key_file.clone(),
-    })
-}
-
-impl GreptimeDBService {
-    pub fn try_new(config: &GreptimeDBConfig) -> crate::Result<Self> {
-        let grpc_client = new_client_from_config(config)?;
-
-        let mut client = Database::new_with_dbname(&config.dbname, grpc_client);
-
-        if let (Some(username), Some(password)) = (&config.username, &config.password) {
-            client.set_auth(AuthScheme::Basic(Basic {
-                username: username.to_owned(),
-                password: password.clone().into(),
-            }))
-        };
-
-        Ok(GreptimeDBService {
-            client: Arc::new(client),
-        })
-    }
-}
-
-impl Service<GreptimeDBRequest> for GreptimeDBService {
-    type Response = GreptimeDBBatchOutput;
-    type Error = GreptimeError;
-    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
-
-    fn poll_ready(&mut self, _cx: &mut std::task::Context) -> Poll<Result<(), Self::Error>> {
-        Poll::Ready(Ok(()))
-    }
-
-    // Convert vector metrics into GreptimeDB format and send them in batch
-    fn call(&mut self, req: GreptimeDBRequest) -> Self::Future {
-        let client = Arc::clone(&self.client);
-
-        Box::pin(async move {
-            let metadata = req.metadata;
-            client.row_insert(req.items).await?;
-
-            Ok(GreptimeDBBatchOutput { metadata })
-        })
-    }
-}