gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-12 08:06:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

init admin_route

Browse Source
This commit is contained in:
weisd
2024-11-05 17:41:35 +08:00
parent 0ae379fd7a
commit c410784cc9
12 changed files with 234 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/Cargo.toml
+2
View File
@@ -53,6 +53,8 @@ transform-stream.workspace = true
uuid = "1.11.0"
admin = { path = "../api/admin" }
axum.workspace = true
router = { version = "0.0.1", path = "../router" }
matchit = "0.8.4"
[build-dependencies]
prost-build.workspace = true
rustfs/src/main.rs
+4 -2
View File
@@ -1,5 +1,6 @@
mod config;
mod grpc;
mod route;
mod service;
mod storage;
@@ -113,6 +114,8 @@ async fn run(opt: config::Opt) -> Result<()> {
b.set_access(store.clone());
b.set_route(route::make_admin_route()?);
// // Enable parsing virtual-hosted-style requests
// if let Some(dm) = opt.domain_name {
// info!("virtual-hosted-style requests are enabled use domain_name {}", &dm);
@@ -135,9 +138,8 @@ async fn run(opt: config::Opt) -> Result<()> {
tokio::spawn(async move {
let hyper_service = service.into_shared();
let adm_service = admin::register_admin_router();
let hybrid_service = TowerToHyperService::new(hybrid(hyper_service, rpc_service, adm_service));
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());
rustfs/src/route.rs
+68
View File
@@ -0,0 +1,68 @@
use common::error::{Error, Result};
use matchit::Params;
use router::{operation::Operation, router::S3Router};
use s3s::route::S3Route;
use s3s::{Body, S3Request, S3Response, S3Result};
use tracing::warn;
use hyper::Method;
use hyper::StatusCode;
pub fn make_admin_route() -> Result<impl S3Route> {
let mut r = S3Router::new();
r.insert(AdminOperation(&InfoHandler {}))?;
r.insert(AdminOperation(&ListPoolHandler {}))?;
Ok(r)
}
struct AdminOperation(&'static dyn Operation);
#[async_trait::async_trait]
impl Operation for AdminOperation {
fn method(&self) -> Method {
self.0.method()
}
fn uri(&self) -> &'static str {
self.0.uri()
}
async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
self.0.call(req, params).await
}
}
struct InfoHandler {}
#[async_trait::async_trait]
impl Operation for InfoHandler {
fn method(&self) -> Method {
Method::GET
}
fn uri(&self) -> &'static str {
"/v3/info"
}
async fn call(&self, _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
warn!("handle info");
unimplemented!()
}
}
struct ListPoolHandler {}
#[async_trait::async_trait]
impl Operation for ListPoolHandler {
fn method(&self) -> Method {
Method::GET
}
fn uri(&self) -> &'static str {
"/v3/pool/list"
}
async fn call(&self, _req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
warn!("handle info");
unimplemented!()
}
}
rustfs/src/service.rs
+14 -61
View File
@@ -1,59 +1,41 @@
use std::pin::Pin;
use std::task::{Context, Poll};
use axum::body::Body;
use futures::Future;
use http_body::Frame;
use hyper::body::Incoming;
use hyper::{Request, Response};
use pin_project_lite::pin_project;
use std::pin::Pin;
use std::task::{Context, Poll};
use tower::Service;
type BoxError = Box<dyn std::error::Error + Send + Sync + 'static>;
/// Generate a [`HybridService`]
pub(crate) fn hybrid<MakeRest, Grpc, Admin>(
make_rest: MakeRest,
grpc: Grpc,
admin: Admin,
) -> HybridService<MakeRest, Grpc, Admin> {
HybridService {
rest: make_rest,
grpc,
admin,
}
pub(crate) fn hybrid<MakeRest, Grpc>(make_rest: MakeRest, grpc: Grpc) -> HybridService<MakeRest, Grpc> {
HybridService { rest: make_rest, grpc }
}
/// The service that can serve both gRPC and REST HTTP Requests
#[derive(Clone)]
pub struct HybridService<Rest, Grpc, Admin> {
pub struct HybridService<Rest, Grpc> {
rest: Rest,
grpc: Grpc,
admin: Admin,
}
impl<Rest, Grpc, Admin, RestBody, GrpcBody> Service<Request<Incoming>> for HybridService<Rest, Grpc, Admin>
impl<Rest, Grpc, RestBody, GrpcBody> Service<Request<Incoming>> for HybridService<Rest, Grpc>
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, Body>>;
type Response = Response<HybridBody<RestBody, GrpcBody>>;
type Error = BoxError;
type Future = HybridFuture<Rest::Future, Grpc::Future, Admin::Future>;
type Future = HybridFuture<Rest::Future, Grpc::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(())) => 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(Ok(())) => Poll::Ready(Ok(())),
Poll::Ready(Err(e)) => Poll::Ready(Err(e.into())),
Poll::Pending => Poll::Pending,
},
@@ -72,13 +54,6 @@ where
grpc_future: self.grpc.call(req),
},
_ if req.uri().path().starts_with("/rustfs") => 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),
},
@@ -90,7 +65,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, AdminBody> {
pub enum HybridBody<RestBody, GrpcBody> {
Rest {
#[pin]
rest_body: RestBody
@@ -99,21 +74,15 @@ pin_project! {
#[pin]
grpc_body: GrpcBody
},
Admin {
#[pin]
admin_body: AdminBody
},
}
}
impl<RestBody, GrpcBody, AdminBody> http_body::Body for HybridBody<RestBody, GrpcBody, AdminBody>
impl<RestBody, GrpcBody> http_body::Body for HybridBody<RestBody, GrpcBody>
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;
@@ -122,7 +91,6 @@ 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(),
}
}
@@ -130,7 +98,6 @@ 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),
}
}
@@ -138,7 +105,6 @@ 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(),
}
}
}
@@ -147,7 +113,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, AdminFuture> {
pub enum HybridFuture<RestFuture, GrpcFuture> {
Rest {
#[pin]
rest_future: RestFuture,
@@ -156,25 +122,17 @@ pin_project! {
#[pin]
grpc_future: GrpcFuture,
},
Admin {
#[pin]
admin_future: AdminFuture,
}
}
}
impl<RestFuture, GrpcFuture, AdminFuture, RestBody, GrpcBody, AdminBody, RestError, GrpcError, AdminError> Future
for HybridFuture<RestFuture, GrpcFuture, AdminFuture>
impl<RestFuture, GrpcFuture, RestBody, GrpcBody, RestError, GrpcError> Future for HybridFuture<RestFuture, GrpcFuture>
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, AdminBody>>, BoxError>;
type Output = Result<Response<HybridBody<RestBody, GrpcBody>>, BoxError>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
match self.project() {
@@ -188,11 +146,6 @@ 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,
},
}
}
}