Merge pull request #6469 from PsiACE/refactor-httpshutdown

refactor: migrate to common-base http shutdown

common/base/src/base/http_shutdown_handlers.rs

@@ -47,14 +47,14 @@ impl HttpShutdownHandler {
 
     pub async fn start_service(
         &mut self,
-        listening: String,
+        listening: SocketAddr,
         tls_config: Option<RustlsConfig>,
         ep: impl Endpoint + 'static,
     ) -> Result<SocketAddr> {
         assert!(self.join_handle.is_none());
         assert!(self.abort_handle.is_none());
 
-        let mut acceptor = TcpListener::bind(listening.clone())
+        let mut acceptor = TcpListener::bind(listening)
             .into_acceptor()
             .await
             .map_err(|err| ErrorCode::CannotListenerPort(format!("{}:{}", err, listening)))?
@@ -144,4 +144,23 @@ impl HttpShutdownHandler {
         }
         Ok(())
     }
+
+    pub async fn shutdown(&mut self, graceful: bool) {
+        if graceful {
+            if let Some(abort_handle) = self.abort_handle.take() {
+                let _ = abort_handle.send(());
+            }
+            if let Some(join_handle) = self.join_handle.take() {
+                if let Err(error) = join_handle.await {
+                    tracing::error!(
+                        "Unexpected error during shutdown Http Server {}. cause {}",
+                        self.service_name,
+                        error
+                    );
+                }
+            }
+        } else if let Some(join_handle) = self.join_handle.take() {
+            join_handle.abort();
+        }
+    }
 }

metasrv/src/api/http_service.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::net::SocketAddr;
 use std::sync::Arc;
 
 use common_base::base::tokio::sync::broadcast;
@@ -94,7 +95,7 @@ impl HttpService {
         Ok(cfg)
     }
 
-    async fn start_with_tls(&mut self, listening: String) -> Result<()> {
+    async fn start_with_tls(&mut self, listening: SocketAddr) -> Result<()> {
         tracing::info!("Http API TLS enabled");
 
         let tls_config = Self::build_tls(&self.cfg.clone())?;
@@ -104,7 +105,7 @@ impl HttpService {
         Ok(())
     }
 
-    async fn start_without_tls(&mut self, listening: String) -> Result<()> {
+    async fn start_without_tls(&mut self, listening: SocketAddr) -> Result<()> {
         tracing::warn!("Http API TLS not set");
 
         self.shutdown_handler
@@ -118,9 +119,10 @@ impl HttpService {
 impl Stoppable for HttpService {
     async fn start(&mut self) -> Result<()> {
         let conf = self.cfg.clone();
+        let listening = conf.admin_api_address.parse::<SocketAddr>()?;
         match conf.admin_tls_server_key.is_empty() || conf.admin_tls_server_cert.is_empty() {
-            true => self.start_without_tls(conf.admin_api_address).await,
-            false => self.start_with_tls(conf.admin_api_address).await,
+            true => self.start_without_tls(listening).await,
+            false => self.start_with_tls(listening).await,
         }
     }
 

query/src/api/http_service.rs

@@ -16,6 +16,7 @@ use std::net::SocketAddr;
 use std::path::Path;
 use std::sync::Arc;
 
+use common_base::base::HttpShutdownHandler;
 use common_exception::Result;
 use common_tracing::tracing;
 use poem::get;
@@ -25,7 +26,6 @@ use poem::Endpoint;
 use poem::EndpointExt;
 use poem::Route;
 
-use crate::common::service::HttpShutdownHandler;
 use crate::servers::Server;
 use crate::sessions::SessionManager;
 use crate::Config;

query/src/common/mod.rs

@@ -15,7 +15,6 @@
 pub mod context_function;
 mod evaluator;
 mod expression_evaluator;
-pub mod service;
 
 pub use evaluator::*;
 pub use expression_evaluator::ExpressionEvaluator;

query/src/metrics/metric_service.rs

@@ -15,14 +15,14 @@
 use std::net::SocketAddr;
 use std::sync::Arc;
 
+use common_base::base::HttpShutdownHandler;
 use common_exception::ErrorCode;
 use common_exception::Result;
 use common_metrics::PrometheusHandle;
 use poem::web::Data;
 use poem::EndpointExt;
 use poem::IntoResponse;
 
-use crate::common::service::HttpShutdownHandler;
 use crate::servers::Server;
 use crate::sessions::SessionManager;
 

query/src/servers/http/http_services.rs

@@ -16,6 +16,7 @@ use std::net::SocketAddr;
 use std::path::Path;
 use std::sync::Arc;
 
+use common_base::base::HttpShutdownHandler;
 use common_exception::Result;
 use common_tracing::tracing;
 use poem::get;
@@ -30,7 +31,6 @@ use poem::EndpointExt;
 use poem::Route;
 
 use super::v1::upload_to_stage;
-use crate::common::service::HttpShutdownHandler;
 use crate::servers::http::middleware::HTTPSessionMiddleware;
 use crate::servers::http::v1::clickhouse_router;
 use crate::servers::http::v1::query_route;