feat: export promql service in server

src/cmd/src/standalone.rs

@@ -26,6 +26,7 @@ use frontend::mysql::MysqlOptions;
 use frontend::opentsdb::OpentsdbOptions;
 use frontend::postgres::PostgresOptions;
 use frontend::prometheus::PrometheusOptions;
+use frontend::promql::PromqlOptions;
 use frontend::Plugins;
 use serde::{Deserialize, Serialize};
 use servers::http::HttpOptions;
@@ -72,6 +73,7 @@ pub struct StandaloneOptions {
     pub opentsdb_options: Option<OpentsdbOptions>,
     pub influxdb_options: Option<InfluxdbOptions>,
     pub prometheus_options: Option<PrometheusOptions>,
+    pub promql_options: Option<PromqlOptions>,
     pub mode: Mode,
     pub wal: WalConfig,
     pub storage: ObjectStoreConfig,
@@ -88,6 +90,7 @@ impl Default for StandaloneOptions {
             opentsdb_options: Some(OpentsdbOptions::default()),
             influxdb_options: Some(InfluxdbOptions::default()),
             prometheus_options: Some(PrometheusOptions::default()),
+            promql_options: Some(PromqlOptions::default()),
             mode: Mode::Standalone,
             wal: WalConfig::default(),
             storage: ObjectStoreConfig::default(),
@@ -106,6 +109,7 @@ impl StandaloneOptions {
             opentsdb_options: self.opentsdb_options,
             influxdb_options: self.influxdb_options,
             prometheus_options: self.prometheus_options,
+            promql_options: self.promql_options,
             mode: self.mode,
             meta_client_opts: None,
         }

src/datanode/src/instance/sql.rs

@@ -14,13 +14,16 @@
 
 use async_trait::async_trait;
 use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
+use common_error::prelude::BoxedError;
 use common_query::Output;
 use common_recordbatch::RecordBatches;
 use common_telemetry::logging::info;
 use common_telemetry::timer;
 use query::parser::{QueryLanguageParser, QueryStatement};
+use servers::error as server_error;
+use servers::promql::PromqlHandler;
 use servers::query_handler::sql::SqlQueryHandler;
-use session::context::QueryContextRef;
+use session::context::{QueryContext, QueryContextRef};
 use snafu::prelude::*;
 use sql::ast::ObjectName;
 use sql::statements::statement::Statement;
@@ -225,6 +228,17 @@ impl SqlQueryHandler for Instance {
     }
 }
 
+#[async_trait]
+impl PromqlHandler for Instance {
+    async fn do_query(&self, query: &str) -> server_error::Result<Output> {
+        let _timer = timer!(metric::METRIC_HANDLE_PROMQL_ELAPSED);
+        self.execute_promql(query, QueryContext::arc())
+            .await
+            .map_err(BoxedError::new)
+            .with_context(|_| server_error::ExecuteQuerySnafu { query })
+    }
+}
+
 #[cfg(test)]
 mod test {
     use std::sync::Arc;

src/datanode/src/metric.rs

@@ -17,3 +17,4 @@
 pub const METRIC_HANDLE_SQL_ELAPSED: &str = "datanode.handle_sql_elapsed";
 pub const METRIC_HANDLE_SCRIPTS_ELAPSED: &str = "datanode.handle_scripts_elapsed";
 pub const METRIC_RUN_SCRIPT_ELAPSED: &str = "datanode.run_script_elapsed";
+pub const METRIC_HANDLE_PROMQL_ELAPSED: &str = "datanode.handle_promql_elapsed";

src/frontend/src/frontend.rs

@@ -28,6 +28,7 @@ use crate::mysql::MysqlOptions;
 use crate::opentsdb::OpentsdbOptions;
 use crate::postgres::PostgresOptions;
 use crate::prometheus::PrometheusOptions;
+use crate::promql::PromqlOptions;
 use crate::server::Services;
 use crate::Plugins;
 
@@ -41,6 +42,7 @@ pub struct FrontendOptions {
     pub opentsdb_options: Option<OpentsdbOptions>,
     pub influxdb_options: Option<InfluxdbOptions>,
     pub prometheus_options: Option<PrometheusOptions>,
+    pub promql_options: Option<PromqlOptions>,
     pub mode: Mode,
     pub meta_client_opts: Option<MetaClientOpts>,
 }
@@ -55,6 +57,7 @@ impl Default for FrontendOptions {
             opentsdb_options: Some(OpentsdbOptions::default()),
             influxdb_options: Some(InfluxdbOptions::default()),
             prometheus_options: Some(PrometheusOptions::default()),
+            promql_options: Some(PromqlOptions::default()),
             mode: Mode::Standalone,
             meta_client_opts: None,
         }

src/frontend/src/instance.rs

@@ -42,6 +42,7 @@ use meta_client::client::{MetaClient, MetaClientBuilder};
 use meta_client::MetaClientOpts;
 use servers::error as server_error;
 use servers::interceptor::{SqlQueryInterceptor, SqlQueryInterceptorRef};
+use servers::promql::{PromqlHandler, PromqlHandlerRef};
 use servers::query_handler::grpc::{GrpcQueryHandler, GrpcQueryHandlerRef};
 use servers::query_handler::sql::{SqlQueryHandler, SqlQueryHandlerRef};
 use servers::query_handler::{
@@ -71,6 +72,7 @@ pub trait FrontendInstance:
     + InfluxdbLineProtocolHandler
     + PrometheusProtocolHandler
     + ScriptHandler
+    + PromqlHandler
     + Send
     + Sync
     + 'static
@@ -88,6 +90,7 @@ pub struct Instance {
     script_handler: Option<ScriptHandlerRef>,
     sql_handler: SqlQueryHandlerRef<Error>,
     grpc_query_handler: GrpcQueryHandlerRef<Error>,
+    promql_handler: Option<PromqlHandlerRef>,
 
     create_expr_factory: CreateExprFactoryRef,
 
@@ -121,6 +124,7 @@ impl Instance {
             create_expr_factory: Arc::new(DefaultCreateExprFactory),
             sql_handler: dist_instance.clone(),
             grpc_query_handler: dist_instance,
+            promql_handler: None,
             plugins: Default::default(),
         })
     }
@@ -162,6 +166,7 @@ impl Instance {
             create_expr_factory: Arc::new(DefaultCreateExprFactory),
             sql_handler: StandaloneSqlQueryHandler::arc(dn_instance.clone()),
             grpc_query_handler: StandaloneGrpcQueryHandler::arc(dn_instance.clone()),
+            promql_handler: Some(dn_instance.clone()),
             plugins: Default::default(),
         }
     }
@@ -174,6 +179,7 @@ impl Instance {
             create_expr_factory: Arc::new(DefaultCreateExprFactory),
             sql_handler: dist_instance.clone(),
             grpc_query_handler: dist_instance,
+            promql_handler: None,
             plugins: Default::default(),
         }
     }
@@ -506,6 +512,20 @@ impl ScriptHandler for Instance {
     }
 }
 
+#[async_trait]
+impl PromqlHandler for Instance {
+    async fn do_query(&self, query: &str) -> server_error::Result<Output> {
+        if let Some(promql_handler) = &self.promql_handler {
+            promql_handler.do_query(query).await
+        } else {
+            server_error::NotSupportedSnafu {
+                feat: "PromQL query in Frontend",
+            }
+            .fail()
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::borrow::Cow;

src/frontend/src/lib.rs

@@ -29,6 +29,7 @@ pub mod opentsdb;
 pub mod partitioning;
 pub mod postgres;
 pub mod prometheus;
+pub mod promql;
 mod server;
 pub mod spliter;
 mod sql;

src/frontend/src/promql.rs

@@ -0,0 +1,42 @@
+// Copyright 2023 Greptime Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct PromqlOptions {
+    pub enable: bool,
+    pub addr: String,
+}
+
+impl Default for PromqlOptions {
+    fn default() -> Self {
+        Self {
+            enable: true,
+            addr: "127.0.0.1:4004".to_string(),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::PromqlOptions;
+
+    #[test]
+    fn test_prometheus_options() {
+        let default = PromqlOptions::default();
+        assert!(default.enable);
+        assert_eq!(default.addr, "127.0.0.1:4004".to_string());
+    }
+}

src/frontend/src/server.rs

@@ -24,6 +24,7 @@ use servers::http::HttpServer;
 use servers::mysql::server::{MysqlServer, MysqlSpawnConfig, MysqlSpawnRef};
 use servers::opentsdb::OpentsdbServer;
 use servers::postgres::PostgresServer;
+use servers::promql::PromqlServer;
 use servers::query_handler::grpc::ServerGrpcQueryHandlerAdaptor;
 use servers::query_handler::sql::ServerSqlQueryHandlerAdaptor;
 use servers::server::Server;
@@ -154,7 +155,7 @@ impl Services {
                 ServerSqlQueryHandlerAdaptor::arc(instance.clone()),
                 http_options.clone(),
             );
-            if let Some(user_provider) = user_provider {
+            if let Some(user_provider) = user_provider.clone() {
                 http_server.set_user_provider(user_provider);
             }
 
@@ -181,12 +182,26 @@ impl Services {
             None
         };
 
+        let promql_server_and_addr = if let Some(promql_options) = &opts.promql_options {
+            let promql_addr = parse_addr(&promql_options.addr)?;
+
+            let mut promql_server = PromqlServer::create_server(instance.clone());
+            if let Some(user_provider) = user_provider {
+                promql_server.set_user_provider(user_provider);
+            }
+
+            Some((promql_server as _, promql_addr))
+        } else {
+            None
+        };
+
         try_join!(
             start_server(http_server_and_addr),
             start_server(grpc_server_and_addr),
             start_server(mysql_server_and_addr),
             start_server(postgres_server_and_addr),
-            start_server(opentsdb_server_and_addr)
+            start_server(opentsdb_server_and_addr),
+            start_server(promql_server_and_addr),
         )
         .context(error::StartServerSnafu)?;
         Ok(())

src/servers/src/grpc.rs

@@ -81,7 +81,7 @@ impl Server for GrpcServer {
                 .await
                 .context(TcpBindSnafu { addr })?;
             let addr = listener.local_addr().context(TcpBindSnafu { addr })?;
-            info!("GRPC server is bound to {}", addr);
+            info!("gRPC server is bound to {}", addr);
 
             *shutdown_tx = Some(tx);
 

src/servers/src/http.rs

@@ -406,8 +406,8 @@ impl HttpServer {
     pub fn make_app(&self) -> Router {
         let mut api = OpenApi {
             info: Info {
-                title: "Greptime DB HTTP API".to_string(),
-                description: Some("HTTP APIs to interact with Greptime DB".to_string()),
+                title: "GreptimeDB HTTP API".to_string(),
+                description: Some("HTTP APIs to interact with GreptimeDB".to_string()),
                 version: HTTP_API_VERSION.to_string(),
                 ..Info::default()
             },

src/servers/src/lib.rs

@@ -28,11 +28,11 @@ pub mod mysql;
 pub mod opentsdb;
 pub mod postgres;
 pub mod prometheus;
+pub mod promql;
 pub mod query_handler;
 pub mod server;
-pub mod tls;
-
 mod shutdown;
+pub mod tls;
 
 #[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
 #[serde(rename_all = "lowercase")]