feat: introduce proxy module (#732)

* feat: introduce proxy module (#715)

* impl route service with proxy

* impl write service with proxy

* remove forward module in proxy

* refactor code

* add tests in write

* feat: impl query with proxy (#717)

* refactor: refactor proxy module (#726)

* refactor: refactor proxy module

* cargo fmt

* refactor by CR

* Feat proxy prom query (#727)

* feat: impl prom query with proxy

* refactor code

* feat: impl stream write with proxy (#737)

* feat: impl stream query with proxy (#742)

* feat: impl stream query with proxy

* refactor by CR

* feat: introduce proxy module

* refactor code

* add header in storage service

* feat: impl storage service with proxy

* make CI happy

* refactor code

* refactor code

* refactor by CR

* refactor by CR

Author: chunshao90

server/src/config.rs

@@ -15,7 +15,7 @@ use router::{
 use serde::{Deserialize, Serialize};
 use table_engine::ANALYTIC_ENGINE_TYPE;
 
-use crate::{grpc::forward, http::DEFAULT_MAX_BODY_SIZE};
+use crate::{http::DEFAULT_MAX_BODY_SIZE, proxy::forward};
 
 #[derive(Clone, Debug, Default, Deserialize, Serialize)]
 #[serde(default)]

server/src/grpc/metrics.rs

@@ -11,9 +11,10 @@ make_auto_flush_static_metric! {
     pub label_enum GrpcTypeKind {
         handle_route,
         handle_write,
-        handle_query,
+        handle_sql_query,
+        handle_prom_query,
         handle_stream_write,
-        handle_stream_query,
+        handle_stream_sql_query,
     }
 
     pub struct GrpcHandlerDurationHistogramVec: LocalHistogram {

server/src/grpc/mod.rs

@@ -4,7 +4,6 @@
 
 use std::{
     net::{AddrParseError, SocketAddr},
-    str::FromStr,
     stringify,
     sync::Arc,
     time::Duration,
@@ -20,35 +19,38 @@ use cluster::ClusterRef;
 use common_types::column_schema;
 use common_util::{
     define_result,
-    error::GenericError,
+    error::{BoxError, GenericError},
     runtime::{JoinHandle, Runtime},
 };
 use futures::FutureExt;
 use log::{info, warn};
 use query_engine::executor::Executor as QueryExecutor;
-use router::{endpoint::Endpoint, RouterRef};
+use router::RouterRef;
 use snafu::{Backtrace, OptionExt, ResultExt, Snafu};
 use table_engine::engine::EngineRuntimes;
 use tokio::sync::oneshot::{self, Sender};
 use tonic::transport::Server;
 
 use crate::{
     grpc::{
-        forward::Forwarder, meta_event_service::MetaServiceImpl,
-        remote_engine_service::RemoteEngineServiceImpl, storage_service::StorageServiceImpl,
+        meta_event_service::MetaServiceImpl, remote_engine_service::RemoteEngineServiceImpl,
+        storage_service::StorageServiceImpl,
     },
     instance::InstanceRef,
+    proxy::{forward, Proxy},
     schema_config_provider::{self, SchemaConfigProviderRef},
 };
 
-pub mod forward;
 mod meta_event_service;
 mod metrics;
 mod remote_engine_service;
 pub(crate) mod storage_service;
 
 #[derive(Debug, Snafu)]
 pub enum Error {
+    #[snafu(display("Internal error, message:{}, cause:{}", msg, source))]
+    Internal { msg: String, source: GenericError },
+
     #[snafu(display(
         "Failed to keep grpc service, err:{}.\nBacktrace:\n{}",
         source,
@@ -325,28 +327,24 @@ impl<Q: QueryExecutor + 'static> Builder<Q> {
         };
 
         let forward_config = self.forward_config.unwrap_or_default();
-        let forwarder = if forward_config.enable {
-            let local_endpoint =
-                Endpoint::from_str(&self.local_endpoint.context(MissingLocalEndpoint)?)
-                    .context(InvalidLocalEndpoint)?;
-            let forwarder = Arc::new(
-                Forwarder::try_new(forward_config, router.clone(), local_endpoint)
-                    .context(BuildForwarder)?,
-            );
-            Some(forwarder)
-        } else {
-            None
-        };
         let bg_runtime = runtimes.bg_runtime.clone();
-        let storage_service = StorageServiceImpl {
+        let proxy = Proxy::try_new(
             router,
             instance,
-            runtimes,
+            forward_config,
+            self.local_endpoint.context(MissingLocalEndpoint)?,
+            self.resp_compress_min_length,
+            self.auto_create_table,
             schema_config_provider,
-            forwarder,
+        )
+        .box_err()
+        .context(Internal {
+            msg: "fail to init proxy",
+        })?;
+        let storage_service = StorageServiceImpl {
+            proxy: Arc::new(proxy),
+            runtimes,
             timeout: self.timeout,
-            resp_compress_min_length: self.resp_compress_min_length,
-            auto_create_table: self.auto_create_table,
         };
         let rpc_server = StorageServiceServer::new(storage_service);
 

server/src/grpc/storage_service/error.rs

@@ -1,81 +1,9 @@
-// Copyright 2022 CeresDB Project Authors. Licensed under Apache-2.0.
+// Copyright 2023 CeresDB Project Authors. Licensed under Apache-2.0.
 
 //! Error definitions for storage service.
 
 use ceresdbproto::common::ResponseHeader;
-use common_util::{define_result, error::GenericError};
-use http::StatusCode;
-use snafu::Snafu;
 
-use crate::error_util;
-
-define_result!(Error);
-
-#[derive(Snafu, Debug)]
-#[snafu(visibility(pub))]
-pub enum Error {
-    #[snafu(display("Rpc error, code:{:?}, message:{}", code, msg))]
-    ErrNoCause { code: StatusCode, msg: String },
-
-    #[snafu(display("Rpc error, code:{:?}, message:{}, cause:{}", code, msg, source))]
-    ErrWithCause {
-        code: StatusCode,
-        msg: String,
-        source: GenericError,
-    },
-}
-
-impl Error {
-    pub fn code(&self) -> StatusCode {
-        match *self {
-            Error::ErrNoCause { code, .. } => code,
-            Error::ErrWithCause { code, .. } => code,
-        }
-    }
-
-    /// Get the error message returned to the user.
-    pub fn error_message(&self) -> String {
-        match self {
-            Error::ErrNoCause { msg, .. } => msg.clone(),
-
-            Error::ErrWithCause { msg, source, .. } => {
-                let err_string = source.to_string();
-                let first_line = error_util::remove_backtrace_from_err(&err_string);
-                format!("{msg}. Caused by: {first_line}")
-            }
-        }
-    }
-}
-
-pub fn build_err_header(err: Error) -> ResponseHeader {
-    ResponseHeader {
-        code: err.code().as_u16() as u32,
-        error: err.error_message(),
-    }
-}
-
-pub fn build_ok_header() -> ResponseHeader {
-    ResponseHeader {
-        code: StatusCode::OK.as_u16() as u32,
-        ..Default::default()
-    }
-}
-
-impl From<router::Error> for Error {
-    fn from(route_err: router::Error) -> Self {
-        match &route_err {
-            router::Error::RouteNotFound { .. } | router::Error::ShardNotFound { .. } => {
-                Error::ErrNoCause {
-                    code: StatusCode::NOT_FOUND,
-                    msg: route_err.to_string(),
-                }
-            }
-            router::Error::ParseEndpoint { .. }
-            | router::Error::OtherWithCause { .. }
-            | router::Error::OtherNoCause { .. } => Error::ErrNoCause {
-                code: StatusCode::INTERNAL_SERVER_ERROR,
-                msg: route_err.to_string(),
-            },
-        }
-    }
+pub fn build_err_header(code: u32, msg: String) -> ResponseHeader {
+    ResponseHeader { code, error: msg }
 }

server/src/grpc/storage_service/header.rs

@@ -0,0 +1,44 @@
+// Copyright 2023 CeresDB Project Authors. Licensed under Apache-2.0.
+
+use std::collections::HashMap;
+
+use log::warn;
+use tonic::metadata::{KeyAndValueRef, MetadataMap};
+
+/// Rpc request header
+/// Tenant/token will be saved in header in future
+#[allow(dead_code)]
+#[derive(Debug, Default)]
+pub struct RequestHeader {
+    metas: HashMap<String, Vec<u8>>,
+}
+
+impl From<&MetadataMap> for RequestHeader {
+    fn from(meta: &MetadataMap) -> Self {
+        let metas = meta
+            .iter()
+            .filter_map(|kv| match kv {
+                KeyAndValueRef::Ascii(key, val) => {
+                    // TODO: The value may be encoded in base64, which is not expected.
+                    Some((key.to_string(), val.as_encoded_bytes().to_vec()))
+                }
+                KeyAndValueRef::Binary(key, val) => {
+                    warn!(
+                        "Binary header is not supported yet and will be omit, key:{:?}, val:{:?}",
+                        key, val
+                    );
+                    None
+                }
+            })
+            .collect();
+
+        Self { metas }
+    }
+}
+
+impl RequestHeader {
+    #[allow(dead_code)]
+    pub fn get(&self, key: &str) -> Option<&[u8]> {
+        self.metas.get(key).map(|v| v.as_slice())
+    }
+}