gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-27 15:37:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

scanner status command(2)

Browse Source 
Signed-off-by: mujunxiang <1948535941@qq.com>
This commit is contained in:
mujunxiang
2024-12-03 09:46:09 +08:00
parent f87b2bee95
commit f707639392
1 changed files with 11 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers.rs
+11 -13
View File
@@ -18,7 +18,7 @@ use ecstore::utils::time::parse_duration;
use ecstore::utils::xml; use ecstore::utils::xml;
use ecstore::GLOBAL_Endpoints; use ecstore::GLOBAL_Endpoints;
use ecstore::{new_object_layer_fn, store_api::BackendInfo}; use ecstore::{new_object_layer_fn, store_api::BackendInfo};
use futures::Stream; use futures::{Stream, StreamExt};
use http::Uri; use http::Uri;
use hyper::StatusCode; use hyper::StatusCode;
use madmin::metrics::RealtimeMetrics; use madmin::metrics::RealtimeMetrics;
@@ -43,6 +43,7 @@ use tokio::spawn;
use tokio::sync::mpsc::error::TryRecvError; use tokio::sync::mpsc::error::TryRecvError;
use tokio::sync::mpsc::{self, Receiver}; use tokio::sync::mpsc::{self, Receiver};
use tokio::time::interval; use tokio::time::interval;
use tokio_stream::wrappers::ReceiverStream;
use tracing::{error, info, warn}; use tracing::{error, info, warn};
pub mod service_account; pub mod service_account;
@@ -332,23 +333,16 @@ fn extract_metrics_init_params(uri: &Uri) -> MetricsParams {
} }
struct MetricsStream { struct MetricsStream {
inner: Receiver<Result<Bytes, StdError>>, inner: ReceiverStream<Result<Bytes, StdError>>,
} }
impl Stream for MetricsStream { impl Stream for MetricsStream {
type Item = Result<Bytes, StdError>; type Item = Result<Bytes, StdError>;
fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
info!("MetricsStream poll_next");
let this = Pin::into_inner(self); let this = Pin::into_inner(self);
match this.inner.try_recv() { this.inner.poll_next_unpin(cx)
Ok(re) => Poll::Ready(Some(re)),
Err(err) => {
if err == TryRecvError::Empty {
return Poll::Pending;
}
Poll::Ready(None)
}
}
} }
} }
@@ -415,10 +409,13 @@ impl Operation for MetricsHandler {
dep_id: d_id, dep_id: d_id,
}; };
let (tx, rx) = mpsc::channel(10); let (tx, rx) = mpsc::channel(10);
let in_stream: DynByteStream = Box::pin(MetricsStream { inner: rx }); let in_stream: DynByteStream = Box::pin(MetricsStream {
inner: ReceiverStream::new(rx),
});
let body = Body::from(in_stream); let body = Body::from(in_stream);
tokio::spawn(async move { tokio::spawn(async move {
while n > 0 { while n > 0 {
info!("loop, n: {n}");
let mut m = RealtimeMetrics::default(); let mut m = RealtimeMetrics::default();
let m_local = collect_local_metrics(types, &opts).await; let m_local = collect_local_metrics(types, &opts).await;
m.merge(m_local); m.merge(m_local);
@@ -435,6 +432,7 @@ impl Operation for MetricsHandler {
// todo write resp // todo write resp
match serde_json::to_vec(&m) { match serde_json::to_vec(&m) {
Ok(re) => { Ok(re) => {
info!("got metrics, send it to client, m: {m:?}, re: {re:?}");
let _ = tx.send(Ok(Bytes::from(re))).await; let _ = tx.send(Ok(Bytes::from(re))).await;
} }
Err(e) => { Err(e) => {