chore add metrics for meta service

src/cmd/src/metasrv.rs

@@ -80,6 +80,8 @@ struct StartCommand {
     selector: Option<String>,
     #[clap(long)]
     use_memory_store: bool,
+    #[clap(long)]
+    metrics_addr: Option<String>,
 }
 
 impl StartCommand {
@@ -128,6 +130,10 @@ impl TryFrom<StartCommand> for MetaSrvOptions {
             opts.use_memory_store = true;
         }
 
+        if let Some(metrics_addr) = cmd.metrics_addr {
+            opts.metrics_addr = metrics_addr;
+        }
+
         Ok(opts)
     }
 }
@@ -150,6 +156,7 @@ mod tests {
             config_file: None,
             selector: Some("LoadBased".to_string()),
             use_memory_store: false,
+            metrics_addr: None,
         };
         let options: MetaSrvOptions = cmd.try_into().unwrap();
         assert_eq!("127.0.0.1:3002".to_string(), options.bind_addr);
@@ -178,6 +185,7 @@ mod tests {
             selector: None,
             config_file: Some(file.path().to_str().unwrap().to_string()),
             use_memory_store: false,
+            metrics_addr: None,
         };
         let options: MetaSrvOptions = cmd.try_into().unwrap();
         assert_eq!("127.0.0.1:3002".to_string(), options.bind_addr);

src/datanode/src/server.rs

@@ -17,8 +17,8 @@ use std::net::SocketAddr;
 use std::sync::Arc;
 
 use common_runtime::Builder as RuntimeBuilder;
-use servers::datanode_metrics_server::DatanodeMetricsServer;
 use servers::grpc::GrpcServer;
+use servers::metrics_server::MetricsServer;
 use servers::query_handler::grpc::ServerGrpcQueryHandlerAdaptor;
 use servers::server::Server;
 use servers::Mode;
@@ -36,7 +36,7 @@ pub mod grpc;
 /// All rpc services.
 pub struct Services {
     grpc_server: GrpcServer,
-    metrics_server: DatanodeMetricsServer,
+    metrics_server: MetricsServer,
 }
 
 impl Services {
@@ -54,7 +54,7 @@ impl Services {
                 None,
                 grpc_runtime,
             ),
-            metrics_server: DatanodeMetricsServer::new(),
+            metrics_server: MetricsServer::new(),
         })
     }
 
@@ -80,6 +80,10 @@ impl Services {
 
     pub async fn shutdown(&self) -> Result<()> {
         self.grpc_server
+            .shutdown()
+            .await
+            .context(ShutdownServerSnafu)?;
+        self.metrics_server
             .shutdown()
             .await
             .context(ShutdownServerSnafu)

src/meta-srv/Cargo.toml

@@ -39,6 +39,7 @@ tokio-stream = { version = "0.1", features = ["net"] }
 tonic.workspace = true
 tower = "0.4"
 url = "2.3"
+servers = { path = "../servers" }
 
 [dev-dependencies]
 tracing = "0.1"

src/meta-srv/src/bootstrap.rs

@@ -20,6 +20,8 @@ use api::v1::meta::lock_server::LockServer;
 use api::v1::meta::router_server::RouterServer;
 use api::v1::meta::store_server::StoreServer;
 use etcd_client::Client;
+use servers::metrics_server::MetricsServer;
+use servers::server::Server;
 use snafu::ResultExt;
 use tokio::net::TcpListener;
 use tokio::sync::mpsc::{self, Receiver, Sender};
@@ -44,6 +46,8 @@ use crate::{error, Result};
 pub struct MetaSrvInstance {
     meta_srv: MetaSrv,
 
+    metrics_srv: Arc<MetricsServer>,
+
     opts: MetaSrvOptions,
 
     signal_sender: Option<Sender<()>>,
@@ -52,9 +56,10 @@ pub struct MetaSrvInstance {
 impl MetaSrvInstance {
     pub async fn new(opts: MetaSrvOptions) -> Result<MetaSrvInstance> {
         let meta_srv = build_meta_srv(&opts).await?;
-
+        let metrics_srv = Arc::new(MetricsServer::new());
         Ok(MetaSrvInstance {
             meta_srv,
+            metrics_srv,
             opts,
             signal_sender: None,
         })
@@ -72,6 +77,17 @@ impl MetaSrvInstance {
             &mut rx,
         )
         .await?;
+        let addr = self
+            .opts
+            .metrics_addr
+            .parse()
+            .context(error::ParseAddrSnafu {
+                addr: &self.opts.metrics_addr,
+            })?;
+        self.metrics_srv
+            .start(addr)
+            .await
+            .context(error::StartMetricsExportSnafu)?;
 
         Ok(())
     }
@@ -85,7 +101,10 @@ impl MetaSrvInstance {
         }
 
         self.meta_srv.shutdown();
-
+        self.metrics_srv
+            .shutdown()
+            .await
+            .context(error::ShutdownServerSnafu)?;
         Ok(())
     }
 }

src/meta-srv/src/error.rs

@@ -25,6 +25,12 @@ pub enum Error {
     #[snafu(display("Failed to send shutdown signal"))]
     SendShutdownSignal { source: SendError<()> },
 
+    #[snafu(display("Failed to shutdown server, source: {}", source))]
+    ShutdownServer {
+        #[snafu(backtrace)]
+        source: servers::error::Error,
+    },
+
     #[snafu(display("Error stream request next is None"))]
     StreamNone { backtrace: Backtrace },
 
@@ -55,7 +61,16 @@ pub enum Error {
         source: tonic::transport::Error,
         backtrace: Backtrace,
     },
-
+    #[snafu(display("Failed to start gRPC server, source: {}", source))]
+    StartMetricsExport {
+        #[snafu(backtrace)]
+        source: servers::error::Error,
+    },
+    #[snafu(display("Failed to parse address {}, source: {}", addr, source))]
+    ParseAddr {
+        addr: String,
+        source: std::net::AddrParseError,
+    },
     #[snafu(display("Empty table name"))]
     EmptyTableName { backtrace: Backtrace },
 
@@ -317,6 +332,9 @@ impl ErrorExt for Error {
             | Error::LockNotConfig { .. }
             | Error::ExceededRetryLimit { .. }
             | Error::SendShutdownSignal { .. }
+            | Error::ShutdownServer { .. }
+            | Error::StartMetricsExport { .. }
+            | Error::ParseAddr { .. }
             | Error::StartGrpc { .. } => StatusCode::Internal,
             Error::EmptyKey { .. }
             | Error::MissingRequiredParameter { .. }

src/meta-srv/src/metasrv.rs

@@ -40,6 +40,7 @@ pub struct MetaSrvOptions {
     pub datanode_lease_secs: i64,
     pub selector: SelectorType,
     pub use_memory_store: bool,
+    pub metrics_addr: String,
 }
 
 impl Default for MetaSrvOptions {
@@ -51,6 +52,7 @@ impl Default for MetaSrvOptions {
             datanode_lease_secs: 15,
             selector: SelectorType::default(),
             use_memory_store: false,
+            metrics_addr: "127.0.0.1:5000".to_string(),
         }
     }
 }

src/servers/src/lib.rs

@@ -19,13 +19,13 @@ use common_catalog::consts::DEFAULT_CATALOG_NAME;
 use serde::{Deserialize, Serialize};
 
 pub mod auth;
-pub mod datanode_metrics_server;
 pub mod error;
 pub mod grpc;
 pub mod http;
 pub mod influxdb;
 pub mod interceptor;
 pub mod line_writer;
+pub mod metrics_server;
 pub mod mysql;
 pub mod opentsdb;
 pub mod postgres;

src/servers/src/datanode_metrics_server.rs → src/servers/src/metrics_server.rs

@@ -33,12 +33,12 @@ pub const METRIC_SERVER: &str = "METRIC_SERVER";
 /// a server that serves metrics
 /// only start when datanode starts in distributed mode
 #[derive(Default)]
-pub struct DatanodeMetricsServer {
+pub struct MetricsServer {
     shutdown_tx: Mutex<Option<Sender<()>>>,
 }
 
 #[async_trait]
-impl ServerTrait for DatanodeMetricsServer {
+impl ServerTrait for MetricsServer {
     async fn start(&self, listening: SocketAddr) -> Result<SocketAddr> {
         self.start(listening).await
     }
@@ -59,7 +59,7 @@ impl ServerTrait for DatanodeMetricsServer {
     }
 }
 
-impl DatanodeMetricsServer {
+impl MetricsServer {
     pub fn new() -> Self {
         Self {
             shutdown_tx: Mutex::new(None),