feat: remote write metric task

config/standalone.example.toml

@@ -177,3 +177,15 @@ parallel_scan_channel_size = 32
 # otlp_endpoint = "localhost:4317"
 # The percentage of tracing will be sampled and exported. Valid range `[0, 1]`, 1 means all traces are sampled, 0 means all traces are not sampled, the default value is 1. ratio > 1 are treated as 1. Fractions < 0 are treated as 0
 # tracing_sample_ratio = 1.0
+
+# standalone/frontend/datanode/metasrv export the metrics generated by itself
+# send metrics to Prometheus remote-write compatible receiver (e.g. `greptimedb`)
+# [remote_write]
+# whether enable export remote_write, default is false
+# enable = false
+# The url of remote write endpoint.
+# Taking greptimedb as an example, for `standalone` deployed under the default configuration.
+# The user can create a database called `system` in the db and export the metric to `http://127.0.0.1:4000/v1/prometheus/write?db=system`
+# endpoint = "http://127.0.0.1:4000/v1/prometheus/write?db=system"
+# The interval of export metric, 
+# write_interval = "30s"

src/cmd/src/error.rs

@@ -140,6 +140,12 @@ pub enum Error {
         source: common_recordbatch::error::Error,
     },
 
+    #[snafu(display("Failed to init remote write metric task"))]
+    InitRemoteWriteMetricTask {
+        location: Location,
+        source: servers::error::Error,
+    },
+
     #[snafu(display("Failed to pretty print Recordbatches"))]
     PrettyPrintRecordBatches {
         location: Location,
@@ -260,6 +266,7 @@ impl ErrorExt for Error {
             | Error::NotDataFromOutput { .. }
             | Error::CreateDir { .. }
             | Error::EmptyResult { .. }
+            | Error::InitRemoteWriteMetricTask { .. }
             | Error::InvalidDatabaseName { .. } => StatusCode::InvalidArguments,
 
             Error::StartProcedureManager { source, .. }

src/cmd/src/frontend.rs

@@ -249,6 +249,10 @@ impl StartCommand {
         .await
         .context(StartFrontendSnafu)?;
 
+        instance
+            .build_remote_write_metric_task(&opts.remote_write)
+            .context(StartFrontendSnafu)?;
+
         instance
             .build_servers(opts)
             .await

src/cmd/src/options.rs

@@ -21,6 +21,7 @@ use frontend::error::{Result as FeResult, TomlFormatSnafu};
 use frontend::frontend::{FrontendOptions, TomlSerializable};
 use meta_srv::metasrv::MetaSrvOptions;
 use serde::{Deserialize, Serialize};
+use servers::remote_writer::RemoteWriteOptions;
 use snafu::ResultExt;
 
 use crate::error::{LoadLayeredConfigSnafu, Result, SerdeJsonSnafu};
@@ -37,6 +38,7 @@ pub struct MixOptions {
     pub frontend: FrontendOptions,
     pub datanode: DatanodeOptions,
     pub logging: LoggingOptions,
+    pub remote_write: RemoteWriteOptions,
 }
 
 impl From<MixOptions> for FrontendOptions {

src/cmd/src/standalone.rs

@@ -43,14 +43,15 @@ use frontend::service_config::{
 use mito2::config::MitoConfig;
 use serde::{Deserialize, Serialize};
 use servers::http::HttpOptions;
+use servers::remote_writer::{RemoteWriteMetricTask, RemoteWriteOptions};
 use servers::tls::{TlsMode, TlsOption};
 use servers::Mode;
 use snafu::ResultExt;
 
 use crate::error::{
-    CreateDirSnafu, IllegalConfigSnafu, InitDdlManagerSnafu, InitMetadataSnafu, Result,
-    ShutdownDatanodeSnafu, ShutdownFrontendSnafu, StartDatanodeSnafu, StartFrontendSnafu,
-    StartProcedureManagerSnafu, StopProcedureManagerSnafu,
+    CreateDirSnafu, IllegalConfigSnafu, InitDdlManagerSnafu, InitMetadataSnafu,
+    InitRemoteWriteMetricTaskSnafu, Result, ShutdownDatanodeSnafu, ShutdownFrontendSnafu,
+    StartDatanodeSnafu, StartFrontendSnafu, StartProcedureManagerSnafu, StopProcedureManagerSnafu,
 };
 use crate::options::{CliOptions, MixOptions, Options};
 use crate::App;
@@ -110,6 +111,7 @@ pub struct StandaloneOptions {
     pub user_provider: Option<String>,
     /// Options for different store engines.
     pub region_engine: Vec<RegionEngineConfig>,
+    pub remote_write: RemoteWriteOptions,
 }
 
 impl Default for StandaloneOptions {
@@ -129,6 +131,7 @@ impl Default for StandaloneOptions {
             metadata_store: KvBackendConfig::default(),
             procedure: ProcedureConfig::default(),
             logging: LoggingOptions::default(),
+            remote_write: RemoteWriteOptions::default(),
             user_provider: None,
             region_engine: vec![
                 RegionEngineConfig::Mito(MitoConfig::default()),
@@ -173,6 +176,7 @@ pub struct Instance {
     datanode: Datanode,
     frontend: FeInstance,
     procedure_manager: ProcedureManagerRef,
+    remote_write_metric_task: Option<RemoteWriteMetricTask>,
 }
 
 #[async_trait]
@@ -189,6 +193,10 @@ impl App for Instance {
             .await
             .context(StartProcedureManagerSnafu)?;
 
+        if let Some(t) = self.remote_write_metric_task.as_ref() {
+            t.start()
+        }
+
         self.frontend.start().await.context(StartFrontendSnafu)?;
         Ok(())
     }
@@ -318,6 +326,7 @@ impl StartCommand {
         let procedure = opts.procedure.clone();
         let frontend = opts.clone().frontend_options();
         let logging = opts.logging.clone();
+        let remote_write = opts.remote_write.clone();
         let datanode = opts.datanode_options();
 
         Ok(Options::Standalone(Box::new(MixOptions {
@@ -327,6 +336,7 @@ impl StartCommand {
             frontend,
             datanode,
             logging,
+            remote_write,
         })))
     }
 
@@ -382,6 +392,10 @@ impl StartCommand {
         )
         .await?;
 
+        let remote_write_metric_task =
+            RemoteWriteMetricTask::try_new(&opts.remote_write, Some(&fe_plugins))
+                .context(InitRemoteWriteMetricTaskSnafu)?;
+
         let mut frontend = FrontendBuilder::new(kv_backend, datanode_manager, ddl_task_executor)
             .with_plugin(fe_plugins)
             .try_build()
@@ -397,6 +411,7 @@ impl StartCommand {
             datanode,
             frontend,
             procedure_manager,
+            remote_write_metric_task,
         })
     }
 

src/common/telemetry/Cargo.toml

@@ -12,6 +12,7 @@ deadlock_detection = ["parking_lot/deadlock_detection"]
 backtrace = "0.3"
 common-error.workspace = true
 console-subscriber = { version = "0.1", optional = true }
+greptime-proto.workspace = true
 lazy_static.workspace = true
 once_cell.workspace = true
 opentelemetry = { version = "0.21.0", default-features = false, features = [