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

feat(internode): label transport operation metrics (#3045)

Browse Source 
Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>
Co-authored-by: houseme <housemecn@gmail.com>
This commit is contained in:
Henry Guo
2026-05-21 23:20:39 +08:00
committed by GitHub
parent e69e32285b
commit 0985f0b37b
7 changed files with 261 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/storage/rpc/disk.rs
+32 -9
View File
@@ -14,7 +14,8 @@
use super::*;
use rustfs_io_metrics::internode_metrics::{
INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL, global_internode_metrics,
INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC,
global_internode_metrics,
};
use serde::de::DeserializeOwned;
use std::io::Cursor;
@@ -933,8 +934,15 @@ impl NodeService {
pub(super) async fn handle_write_all(&self, request: Request<WriteAllRequest>) -> Result<Response<WriteAllResponse>, Status> {
let request = request.into_inner();
let data_len = request.data.len();
global_internode_metrics().record_incoming_request_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL);
global_internode_metrics().record_recv_bytes_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL, data_len);
global_internode_metrics().record_incoming_request_for_operation_and_backend(
INTERNODE_OPERATION_GRPC_WRITE_ALL,
INTERNODE_TRANSPORT_BACKEND_GRPC,
);
global_internode_metrics().record_recv_bytes_for_operation_and_backend(
INTERNODE_OPERATION_GRPC_WRITE_ALL,
INTERNODE_TRANSPORT_BACKEND_GRPC,
data_len,
);
if let Some(disk) = self.find_disk(&request.disk).await {
match disk.write_all(&request.volume, &request.path, request.data).await {
Ok(_) => Ok(Response::new(WriteAllResponse {
@@ -942,7 +950,10 @@ impl NodeService {
error: None,
})),
Err(err) => {
global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL);
global_internode_metrics().record_error_for_operation_and_backend(
INTERNODE_OPERATION_GRPC_WRITE_ALL,
INTERNODE_TRANSPORT_BACKEND_GRPC,
);
Ok(Response::new(WriteAllResponse {
success: false,
error: Some(err.into()),
@@ -950,7 +961,8 @@ impl NodeService {
}
}
} else {
global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL);
global_internode_metrics()
.record_error_for_operation_and_backend(INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC);
Ok(Response::new(WriteAllResponse {
success: false,
error: Some(DiskError::other("can not find disk".to_string()).into()),
@@ -962,11 +974,18 @@ impl NodeService {
debug!("read all");
let request = request.into_inner();
global_internode_metrics().record_incoming_request_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL);
global_internode_metrics().record_incoming_request_for_operation_and_backend(
INTERNODE_OPERATION_GRPC_READ_ALL,
INTERNODE_TRANSPORT_BACKEND_GRPC,
);
if let Some(disk) = self.find_disk(&request.disk).await {
match disk.read_all(&request.volume, &request.path).await {
Ok(data) => {
global_internode_metrics().record_sent_bytes_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL, data.len());
global_internode_metrics().record_sent_bytes_for_operation_and_backend(
INTERNODE_OPERATION_GRPC_READ_ALL,
INTERNODE_TRANSPORT_BACKEND_GRPC,
data.len(),
);
Ok(Response::new(ReadAllResponse {
success: true,
data,
@@ -974,7 +993,10 @@ impl NodeService {
}))
}
Err(err) => {
global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL);
global_internode_metrics().record_error_for_operation_and_backend(
INTERNODE_OPERATION_GRPC_READ_ALL,
INTERNODE_TRANSPORT_BACKEND_GRPC,
);
Ok(Response::new(ReadAllResponse {
success: false,
data: Bytes::new(),
@@ -983,7 +1005,8 @@ impl NodeService {
}
}
} else {
global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL);
global_internode_metrics()
.record_error_for_operation_and_backend(INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC);
Ok(Response::new(ReadAllResponse {
success: false,
data: Bytes::new(),
rustfs/src/storage/rpc/http_service.rs
+25 -8
View File
@@ -25,7 +25,7 @@ use rustfs_ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE;
use rustfs_ecstore::store::find_local_disk_by_ref;
use rustfs_io_metrics::internode_metrics::{
INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, INTERNODE_OPERATION_WALK_DIR,
global_internode_metrics,
INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, global_internode_metrics,
};
use rustfs_utils::net::bytes_stream;
use s3s::Body;
@@ -143,7 +143,9 @@ fn internode_http_operation(path: &str) -> Option<&'static str> {
fn record_internode_rpc_error(operation: Option<&'static str>) {
match operation {
Some(operation) => global_internode_metrics().record_error_for_operation(operation),
Some(operation) => {
global_internode_metrics().record_error_for_operation_and_backend(operation, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP)
}
None => global_internode_metrics().record_error(),
}
}
@@ -183,7 +185,10 @@ async fn handle_read_file(req: Request<Incoming>) -> Response<Body> {
Err(e) => return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("read file err {e}")),
};
global_internode_metrics().record_incoming_request_for_operation(INTERNODE_OPERATION_READ_FILE_STREAM);
global_internode_metrics().record_incoming_request_for_operation_and_backend(
INTERNODE_OPERATION_READ_FILE_STREAM,
INTERNODE_TRANSPORT_BACKEND_TCP_HTTP,
);
let stream = read_file_body_stream(file, query.length, INTERNODE_OPERATION_READ_FILE_STREAM);
Response::builder()
@@ -202,7 +207,7 @@ where
{
let metrics = global_internode_metrics().clone();
let stream = ReaderStream::with_capacity(reader, DEFAULT_READ_BUFFER_SIZE).map_ok(move |bytes| {
metrics.record_sent_bytes_for_operation(operation, bytes.len());
metrics.record_sent_bytes_for_operation_and_backend(operation, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, bytes.len());
bytes
});
@@ -244,10 +249,15 @@ async fn handle_walk_dir(req: Request<Incoming>) -> Response<Body> {
}
});
global_internode_metrics().record_incoming_request_for_operation(INTERNODE_OPERATION_WALK_DIR);
global_internode_metrics()
.record_incoming_request_for_operation_and_backend(INTERNODE_OPERATION_WALK_DIR, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP);
let metrics = global_internode_metrics().clone();
let stream = ReaderStream::with_capacity(rd, DEFAULT_READ_BUFFER_SIZE).map_ok(move |bytes| {
metrics.record_sent_bytes_for_operation(INTERNODE_OPERATION_WALK_DIR, bytes.len());
metrics.record_sent_bytes_for_operation_and_backend(
INTERNODE_OPERATION_WALK_DIR,
INTERNODE_TRANSPORT_BACKEND_TCP_HTTP,
bytes.len(),
);
bytes
});
@@ -284,8 +294,15 @@ async fn handle_put_file(req: Request<Incoming>) -> Response<Body> {
Err(e) => return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("write file err {e}")),
};
global_internode_metrics().record_incoming_request_for_operation(INTERNODE_OPERATION_PUT_FILE_STREAM);
global_internode_metrics().record_recv_bytes_for_operation(INTERNODE_OPERATION_PUT_FILE_STREAM, copied as usize);
global_internode_metrics().record_incoming_request_for_operation_and_backend(
INTERNODE_OPERATION_PUT_FILE_STREAM,
INTERNODE_TRANSPORT_BACKEND_TCP_HTTP,
);
global_internode_metrics().record_recv_bytes_for_operation_and_backend(
INTERNODE_OPERATION_PUT_FILE_STREAM,
INTERNODE_TRANSPORT_BACKEND_TCP_HTTP,
copied as usize,
);
if let Err(e) = file.flush().await {
return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("write file err {e}"));