feat: introduce admin-api

Signed-off-by: bestgopher <84328409@qq.com>

rustfs/Cargo.toml

@@ -49,6 +49,8 @@ tracing-error.workspace = true
 tracing-subscriber.workspace = true
 transform-stream.workspace = true
 uuid = "1.10.0"
+admin = { path = "../admin" }
+axum.workspace = true
 
 [build-dependencies]
 prost-build.workspace = true
@@ -67,7 +69,6 @@ hyper-util = { version = "0.1.9", features = [
     "server-auto",
     "server-graceful",
 ] }
-mime = "0.3.17"
 transform-stream = "0.3.0"
 netif = "0.1.6"
 # pin-utils = "0.1.0"

rustfs/src/main.rs

@@ -147,55 +147,59 @@ async fn run(opt: config::Opt) -> Result<()> {
 
     let rpc_service = NodeServiceServer::with_interceptor(make_server(), check_auth);
 
-    tokio::spawn(async move {
-        let hyper_service = service.into_shared();
-
-        let hybrid_service = TowerToHyperService::new(hybrid(hyper_service, rpc_service));
-
-        let http_server = ConnBuilder::new(TokioExecutor::new());
-        let mut ctrl_c = std::pin::pin!(tokio::signal::ctrl_c());
-        let graceful = hyper_util::server::graceful::GracefulShutdown::new();
-        info!("server is running at http://{local_addr}");
-
-        loop {
-            let (socket, _) = tokio::select! {
-                res =  listener.accept() => {
-                    match res {
-                        Ok(conn) => conn,
-                        Err(err) => {
-                            tracing::error!("error accepting connection: {err}");
-                            continue;
-                        }
-                    }
-                }
-                _ = ctrl_c.as_mut() => {
-                    break;
-                }
-            };
-
-            let conn = http_server.serve_connection(TokioIo::new(socket), hybrid_service.clone());
-            let conn = graceful.watch(conn.into_owned());
-            tokio::spawn(async move {
-                let _ = conn.await;
-            });
-        }
-
-        tokio::select! {
-            () = graceful.shutdown() => {
-                 tracing::debug!("Gracefully shutdown!");
-            },
-            () = tokio::time::sleep(std::time::Duration::from_secs(10)) => {
-                 tracing::debug!("Waited 10 seconds for graceful shutdown, aborting...");
-            }
-        }
-    });
-
     // init store
     let store = ECStore::new(opt.address.clone(), endpoint_pools.clone())
         .await
         .map_err(|err| Error::from_string(err.to_string()))?;
     info!(" init store success!");
 
+    tokio::spawn({
+        let store = store.clone();
+        async move {
+            let hyper_service = service.into_shared();
+            let adm_service = admin::register_admin_router(Some(store));
+
+            let hybrid_service = TowerToHyperService::new(hybrid(hyper_service, rpc_service, adm_service));
+
+            let http_server = ConnBuilder::new(TokioExecutor::new());
+            let mut ctrl_c = std::pin::pin!(tokio::signal::ctrl_c());
+            let graceful = hyper_util::server::graceful::GracefulShutdown::new();
+            info!("server is running at http://{local_addr}");
+
+            loop {
+                let (socket, _) = tokio::select! {
+                    res =  listener.accept() => {
+                        match res {
+                            Ok(conn) => conn,
+                            Err(err) => {
+                                tracing::error!("error accepting connection: {err}");
+                                continue;
+                            }
+                        }
+                    }
+                    _ = ctrl_c.as_mut() => {
+                        break;
+                    }
+                };
+
+                let conn = http_server.serve_connection(TokioIo::new(socket), hybrid_service.clone());
+                let conn = graceful.watch(conn.into_owned());
+                tokio::spawn(async move {
+                    let _ = conn.await;
+                });
+            }
+
+            tokio::select! {
+                () = graceful.shutdown() => {
+                     tracing::debug!("Gracefully shutdown!");
+                },
+                () = tokio::time::sleep(std::time::Duration::from_secs(10)) => {
+                     tracing::debug!("Waited 10 seconds for graceful shutdown, aborting...");
+                }
+            }
+        }
+    });
+
     let buckets_list = store
         .list_bucket(&BucketOptions {
             no_metadata: true,

rustfs/src/service.rs

@@ -1,6 +1,7 @@
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
+use axum::body::Body;
 use futures::Future;
 use http_body::Frame;
 use hyper::body::Incoming;
@@ -11,35 +12,52 @@ use tower::Service;
 type BoxError = Box<dyn std::error::Error + Send + Sync + 'static>;
 
 /// Generate a [`HybridService`]
-pub(crate) fn hybrid<MakeRest, Grpc>(make_rest: MakeRest, grpc: Grpc) -> HybridService<MakeRest, Grpc> {
-    HybridService { rest: make_rest, grpc }
+pub(crate) fn hybrid<MakeRest, Grpc, Admin>(
+    make_rest: MakeRest,
+    grpc: Grpc,
+    admin: Admin,
+) -> HybridService<MakeRest, Grpc, Admin> {
+    HybridService {
+        rest: make_rest,
+        grpc,
+        admin,
+    }
 }
 
 /// The service that can serve both gRPC and REST HTTP Requests
 #[derive(Clone)]
-pub struct HybridService<Rest, Grpc> {
+pub struct HybridService<Rest, Grpc, Admin> {
     rest: Rest,
     grpc: Grpc,
+    admin: Admin,
 }
 
-impl<Rest, Grpc, RestBody, GrpcBody> Service<Request<Incoming>> for HybridService<Rest, Grpc>
+impl<Rest, Grpc, Admin, RestBody, GrpcBody> Service<Request<Incoming>> for HybridService<Rest, Grpc, Admin>
 where
     Rest: Service<Request<Incoming>, Response = Response<RestBody>>,
     Grpc: Service<Request<Incoming>, Response = Response<GrpcBody>>,
+    Admin: Service<Request<Body>, Response = Response<Body>>,
     Rest::Error: Into<BoxError>,
     Grpc::Error: Into<BoxError>,
+    Admin::Error: Into<BoxError>,
 {
-    type Response = Response<HybridBody<RestBody, GrpcBody>>;
+    type Response = Response<HybridBody<RestBody, GrpcBody, Body>>;
     type Error = BoxError;
-    type Future = HybridFuture<Rest::Future, Grpc::Future>;
+    type Future = HybridFuture<Rest::Future, Grpc::Future, Admin::Future>;
 
     fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         match self.rest.poll_ready(cx) {
             Poll::Ready(Ok(())) => match self.grpc.poll_ready(cx) {
-                Poll::Ready(Ok(())) => Poll::Ready(Ok(())),
+                Poll::Ready(Ok(())) => match self.admin.poll_ready(cx) {
+                    Poll::Ready(Ok(())) => Poll::Ready(Ok(())),
+                    Poll::Ready(Err(e)) => Poll::Ready(Err(e.into())),
+                    Poll::Pending => Poll::Pending,
+                },
+
                 Poll::Ready(Err(e)) => Poll::Ready(Err(e.into())),
                 Poll::Pending => Poll::Pending,
             },
+
             Poll::Ready(Err(e)) => Poll::Ready(Err(e.into())),
             Poll::Pending => Poll::Pending,
         }
@@ -53,6 +71,14 @@ where
             (hyper::Version::HTTP_2, Some(hv)) if hv.as_bytes().starts_with(b"application/grpc") => HybridFuture::Grpc {
                 grpc_future: self.grpc.call(req),
             },
+
+            _ if req.uri().path().starts_with("/admin/v3") => HybridFuture::Admin {
+                admin_future: self.admin.call({
+                    let (parts, body) = req.into_parts();
+                    Request::from_parts(parts, Body::new(body).into())
+                }),
+            },
+
             _ => HybridFuture::Rest {
                 rest_future: self.rest.call(req),
             },
@@ -64,7 +90,7 @@ pin_project! {
     /// A hybrid HTTP body that will be used in the response type for the
     /// [`HybridFuture`], i.e., the output of the [`HybridService`]
     #[project = HybridBodyProj]
-    pub enum HybridBody<RestBody, GrpcBody> {
+    pub enum HybridBody<RestBody, GrpcBody, AdminBody> {
         Rest {
             #[pin]
             rest_body: RestBody
@@ -73,15 +99,21 @@ pin_project! {
             #[pin]
             grpc_body: GrpcBody
         },
+        Admin {
+            #[pin]
+            admin_body: AdminBody
+        },
     }
 }
 
-impl<RestBody, GrpcBody> http_body::Body for HybridBody<RestBody, GrpcBody>
+impl<RestBody, GrpcBody, AdminBody> http_body::Body for HybridBody<RestBody, GrpcBody, AdminBody>
 where
     RestBody: http_body::Body + Send + Unpin,
     GrpcBody: http_body::Body<Data = RestBody::Data> + Send + Unpin,
+    AdminBody: http_body::Body<Data = RestBody::Data> + Send + Unpin,
     RestBody::Error: Into<BoxError>,
     GrpcBody::Error: Into<BoxError>,
+    AdminBody::Error: Into<BoxError>,
 {
     type Data = RestBody::Data;
     type Error = BoxError;
@@ -90,6 +122,7 @@ where
         match self {
             Self::Rest { rest_body } => rest_body.is_end_stream(),
             Self::Grpc { grpc_body } => grpc_body.is_end_stream(),
+            Self::Admin { admin_body } => admin_body.is_end_stream(),
         }
     }
 
@@ -97,6 +130,7 @@ where
         match self.project() {
             HybridBodyProj::Rest { rest_body } => rest_body.poll_frame(cx).map_err(Into::into),
             HybridBodyProj::Grpc { grpc_body } => grpc_body.poll_frame(cx).map_err(Into::into),
+            HybridBodyProj::Admin { admin_body } => admin_body.poll_frame(cx).map_err(Into::into),
         }
     }
 
@@ -104,6 +138,7 @@ where
         match self {
             Self::Rest { rest_body } => rest_body.size_hint(),
             Self::Grpc { grpc_body } => grpc_body.size_hint(),
+            Self::Admin { admin_body } => admin_body.size_hint(),
         }
     }
 }
@@ -112,7 +147,7 @@ pin_project! {
     /// A future that accepts an HTTP request as input and returns an HTTP
     /// response as output for the [`HybridService`]
     #[project = HybridFutureProj]
-    pub enum HybridFuture<RestFuture, GrpcFuture> {
+    pub enum HybridFuture<RestFuture, GrpcFuture, AdminFuture> {
         Rest {
             #[pin]
             rest_future: RestFuture,
@@ -120,18 +155,26 @@ pin_project! {
         Grpc {
             #[pin]
             grpc_future: GrpcFuture,
+        },
+
+        Admin {
+            #[pin]
+            admin_future: AdminFuture,
         }
     }
 }
 
-impl<RestFuture, GrpcFuture, RestBody, GrpcBody, RestError, GrpcError> Future for HybridFuture<RestFuture, GrpcFuture>
+impl<RestFuture, GrpcFuture, AdminFuture, RestBody, GrpcBody, AdminBody, RestError, GrpcError, AdminError> Future
+    for HybridFuture<RestFuture, GrpcFuture, AdminFuture>
 where
     RestFuture: Future<Output = Result<Response<RestBody>, RestError>>,
     GrpcFuture: Future<Output = Result<Response<GrpcBody>, GrpcError>>,
+    AdminFuture: Future<Output = Result<Response<AdminBody>, AdminError>>,
     RestError: Into<BoxError>,
     GrpcError: Into<BoxError>,
+    AdminError: Into<BoxError>,
 {
-    type Output = Result<Response<HybridBody<RestBody, GrpcBody>>, BoxError>;
+    type Output = Result<Response<HybridBody<RestBody, GrpcBody, AdminBody>>, BoxError>;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         match self.project() {
@@ -145,6 +188,11 @@ where
                 Poll::Ready(Err(err)) => Poll::Ready(Err(err.into())),
                 Poll::Pending => Poll::Pending,
             },
+            HybridFutureProj::Admin { admin_future } => match admin_future.poll(cx) {
+                Poll::Ready(Ok(res)) => Poll::Ready(Ok(res.map(|admin_body| HybridBody::Admin { admin_body }))),
+                Poll::Ready(Err(err)) => Poll::Ready(Err(err.into())),
+                Poll::Pending => Poll::Pending,
+            },
         }
     }
 }