From f695870626383f2f638230abfe64b9eb2e72fbfb Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Tue, 19 May 2026 15:16:05 +0800 Subject: [PATCH] feat(internode): add transport observability (#3007) * docs: add internode data transport RFC * feat: add internode operation metrics * fix feedback * fix(ci): fallback protoc token to github.token --------- Co-authored-by: Henry Guo --- .github/actions/setup/action.yml | 2 +- INTERNODE_DATA_TRANSPORT_RFC.md | 390 +++++++++++++++++++++++++ crates/common/src/internode_metrics.rs | 66 +++++ crates/ecstore/src/rpc/remote_disk.rs | 44 ++- crates/rio/src/http_reader.rs | 130 +++++++-- rustfs/src/storage/rpc/disk.rs | 46 ++- rustfs/src/storage/rpc/http_service.rs | 61 +++- 7 files changed, 671 insertions(+), 68 deletions(-) create mode 100644 INTERNODE_DATA_TRANSPORT_RFC.md diff --git a/.github/actions/setup/action.yml b/.github/actions/setup/action.yml index 8a3ae8fc4..581ec2d86 100644 --- a/.github/actions/setup/action.yml +++ b/.github/actions/setup/action.yml @@ -59,7 +59,7 @@ runs: uses: arduino/setup-protoc@v3 with: version: "34.1" - repo-token: ${{ inputs.github-token }} + repo-token: ${{ inputs.github-token || github.token }} - name: Install flatc uses: Nugine/setup-flatc@v1 diff --git a/INTERNODE_DATA_TRANSPORT_RFC.md b/INTERNODE_DATA_TRANSPORT_RFC.md new file mode 100644 index 000000000..117ff23be --- /dev/null +++ b/INTERNODE_DATA_TRANSPORT_RFC.md @@ -0,0 +1,390 @@ +# RFC: Pluggable Internode Data Transport + +> Status: draft +> Last updated: 2026-05-19 +> Scope: internode data-path analysis, benchmark baseline, and transport boundary + +## Summary + +RustFS does not currently include RDMA, RoCE, InfiniBand, DPU, BlueField/DOCA, +DPDK, SPDK, or SmartNIC offload support. The current distributed internode +paths use TCP-based HTTP/gRPC transports: + +- `tonic` gRPC `NodeService` for most control, metadata, lock, health, and + peer operations. +- HTTP streaming routes under `/rustfs/rpc/` for remote disk file streams. + +RDMA/RoCE is still a plausible future optimization for large internode disk +data transfers, but it should not replace the whole internode RPC surface. +The correct first step is to isolate the data plane, establish a TCP baseline, +and introduce a pluggable transport boundary only around high-volume streams. + +## Goals + +- Document the current internode control plane and data plane. +- Identify the existing transfer paths that could benefit from a future + high-throughput backend. +- Define the minimum benchmark baseline required before transport changes. +- Sketch a pluggable transport boundary that preserves the current TCP/HTTP + behavior as the default backend. +- Reserve explicit boundaries for future RDMA/RoCE/InfiniBand work without + committing RustFS to a specific vendor stack. + +## Non-Goals + +- Implement RDMA, RoCE, InfiniBand, DPU, DOCA, DPDK, SPDK, or SmartNIC support. +- Replace `tonic` gRPC for control-plane RPCs. +- Redesign erasure coding, quorum handling, disk health tracking, or object + correctness semantics. +- Require RDMA-capable hardware for default development, CI, or ordinary + RustFS deployments. + +## Current Internode Architecture + +### Server-side entry points + +The main HTTP server builds a hybrid service per connection: + +- `rustfs/src/server/http.rs` wires a `NodeServiceServer` for gRPC. +- `rustfs/src/storage/rpc/InternodeRpcService` intercepts HTTP paths under + `/rustfs/rpc/`. +- Other HTTP/S3 traffic continues through the normal S3 service. + +Compression logic already treats `/rustfs/rpc/` and `/rustfs/peer/` as internode +RPC paths and skips normal response compression for them. + +### gRPC channel management + +`crates/protos/src/lib.rs` creates internode gRPC channels with `tonic` +`Endpoint`: + +- connect timeout +- TCP keepalive +- HTTP/2 keepalive interval and timeout +- request timeout +- optional TLS configuration +- global channel caching and failed-connection eviction + +This confirms the current gRPC transport is TCP/HTTP2-based. + +### NodeService layout + +`crates/protos/src/node.proto` defines one `NodeService` that mixes several +classes of RPCs: + +- meta service: bucket and metadata operations +- disk service: local/remote disk operations +- lock service: distributed lock operations +- peer rest service: node health, metrics, IAM/policy reload, rebalance, + profiling, events, and admin-style operations + +The service layout is practical today, but it is too broad to become an RDMA +surface. A future high-throughput transport should target only disk data +streams and keep this gRPC service as the control plane. + +## Control Plane vs Data Plane + +### Control plane + +These paths carry coordination, metadata, health, and administrative state. +They should remain on gRPC/TCP: + +| Area | Client/server code | Examples | Notes | +| --- | --- | --- | --- | +| Bucket peer ops | `crates/ecstore/src/rpc/peer_s3_client.rs`, `rustfs/src/storage/rpc/bucket.rs` | `MakeBucket`, `ListBucket`, `DeleteBucket`, `GetBucketInfo`, `HealBucket` | Small metadata/control payloads. | +| Locking | `crates/ecstore/src/rpc/remote_locker.rs`, `rustfs/src/storage/rpc/lock.rs` | `Lock`, `UnLock`, `Refresh`, batch lock/unlock | Latency-sensitive but not bulk data; correctness and timeout semantics matter more than transport bandwidth. | +| Peer/admin state | `crates/ecstore/src/rpc/peer_rest_client.rs`, `rustfs/src/storage/rpc/health.rs`, `metrics.rs`, `event.rs` | `LocalStorageInfo`, `ServerInfo`, `GetMetrics`, `GetLiveEvents`, reload APIs, rebalance APIs | Operational control plane. | +| Disk metadata/control | `crates/ecstore/src/rpc/remote_disk.rs`, `rustfs/src/storage/rpc/disk.rs` | `DiskInfo`, `ReadXL`, `ReadVersion`, `ReadMetadata`, `WriteMetadata`, `RenameFile`, `RenamePart`, `Delete*`, `VerifyFile`, `CheckParts` | Usually metadata, integrity checks, or namespace mutations. | +| Connection health | `RemoteDisk`, `RemotePeerS3Client`, `PeerRestClient` | TCP connectivity probes and fault/recovery state | Must remain available even if an optional data backend is unavailable. | + +### Data plane candidates + +These paths move object shard bytes or stream potentially large disk data and +are the only reasonable first candidates for a pluggable transport. + +| Priority | Path | Current client | Current server | Current transport | Why it matters | +| --- | --- | --- | --- | --- | --- | +| P0 | `read_file_stream` | `RemoteDisk::read_file_stream` | `handle_read_file` in `http_service.rs` | HTTP `GET /rustfs/rpc/read_file_stream` with a streaming response body | Main remote disk read stream used by bitrot readers and erasure reads. | +| P0 | `put_file_stream` | `RemoteDisk::create_file` and `RemoteDisk::append_file` | `handle_put_file` in `http_service.rs` | HTTP `PUT /rustfs/rpc/put_file_stream` with a streaming request body | Main remote disk write stream used by bitrot writers and erasure writes. | +| P1 | `walk_dir` | `RemoteDisk::walk_dir` | `handle_walk_dir` in `http_service.rs` | HTTP `GET /rustfs/rpc/walk_dir` with a streamed metadata listing | Can be high-volume during scans/healing, but it is metadata-oriented rather than object byte data. | +| P1 | `ReadAll` / `WriteAll` | `RemoteDisk::read_all` / `write_all` | gRPC unary disk handlers | gRPC unary `bytes` payload | Moves bytes today, but should be measured before treating it as a high-throughput data path. | +| P2 | proto `WriteStream` / `ReadAt` | currently not used | currently returns unimplemented | gRPC streaming definitions exist but are not implemented | Possible future API shape, not a current production path. | + +## Current Object Write Path + +For object PUTs in distributed erasure mode, the relevant flow is: + +1. Upper storage layers prepare object data and erasure metadata. +2. `SetDisks` selects local and remote disks. +3. `create_bitrot_writer` calls `disk.create_file(...)` for each shard writer. +4. For a remote disk, `RemoteDisk::create_file` returns an `HttpWriter`. +5. `HttpWriter` sends an HTTP `PUT` to `/rustfs/rpc/put_file_stream`. +6. The remote node's `handle_put_file` opens the local file writer and copies + incoming body chunks into it. +7. `Erasure::encode` writes shards through `MultiWriter` to all selected + writers while enforcing write quorum. + +This is the primary write data-plane candidate. + +## Current Object Read Path + +For object GETs and repair reads in distributed erasure mode, the relevant flow is: + +1. `SetDisks` prepares shard readers for the selected disks. +2. `create_bitrot_reader` uses local zero-copy only when `disk.is_local()`. +3. For a remote disk, it calls `disk.read_file_stream(...)`. +4. `RemoteDisk::read_file_stream` returns an `HttpReader`. +5. `HttpReader` sends an HTTP `GET` to `/rustfs/rpc/read_file_stream`. +6. The remote node's `handle_read_file` opens the local disk stream and returns + it as an HTTP streaming body. +7. The erasure decoder reads from the shard streams and reconstructs the object. + +This is the primary read data-plane candidate. + +## Existing Metrics and Benchmark Surface + +RustFS already has coarse internode metrics in `crates/common/src/internode_metrics.rs`: + +- sent bytes +- received bytes +- outgoing requests +- incoming requests +- errors +- dial errors +- average dial time + +These metrics are useful as a starting point, but they are not enough for a +transport RFC. A transport benchmark needs route-level and operation-level +measurements for at least: + +- `read_file_stream` +- `put_file_stream` +- `walk_dir` +- gRPC `ReadAll` / `WriteAll` +- gRPC control-plane request volume + +Existing benchmark assets: + +- `scripts/run_object_batch_bench.sh` +- `scripts/run_object_batch_bench_enhanced.sh` +- `scripts/run_object_batch_bench_abc.sh` +- `scripts/run_four_node_cluster_failover_bench.sh` +- Criterion benches under `crates/ecstore/benches/` + +These mostly cover S3/object workload or erasure coding performance. They do +not yet isolate internode transport cost. + +## Required TCP Baseline + +Before adding a transport abstraction or any RDMA backend, collect a baseline +for the current TCP/HTTP/gRPC implementation. + +### Topology + +Minimum: + +- 1-node local erasure deployment, to measure local disk and erasure overhead. +- 4-node distributed erasure deployment, to measure internode overhead. + +Preferred: + +- Same host count and disk layout for every run. +- Dedicated network interface or isolated VLAN. +- Fixed CPU governor and no unrelated background load. +- Recorded kernel version, NIC model, MTU, RustFS commit, Rust toolchain, and + benchmark tool versions. + +### Workloads + +| Workload | Sizes | Concurrency | Main signal | +| --- | --- | --- | --- | +| S3 PUT | 4 KiB, 1 MiB, 16 MiB, 128 MiB, 1 GiB | 1, 16, 64, 128 | End-to-end write throughput and tail latency. | +| S3 GET | 4 KiB, 1 MiB, 16 MiB, 128 MiB, 1 GiB | 1, 16, 64, 128 | End-to-end read throughput and tail latency. | +| Remote disk stream read | shard-sized ranges from `read_file_stream` | 1, 16, 64 | Isolated internode read path. | +| Remote disk stream write | shard-sized writes through `put_file_stream` | 1, 16, 64 | Isolated internode write path. | +| Healing / repair | missing disk or missing shard scenario | controlled | Rebuild throughput and read/write amplification. | +| Scanner walk | large bucket/object namespace | controlled | Metadata streaming pressure, not primary RDMA target. | + +### Measurements + +Collect: + +- throughput in bytes/s and objects/s +- p50, p95, p99, and max latency +- CPU utilization per process and per core +- memory RSS and allocation pressure where available +- `rustfs_system_network_internode_*` metrics +- TCP retransmits, socket errors, and NIC throughput +- disk throughput and utilization +- failure/retry/fallback counts + +The baseline should produce a machine-readable artifact, for example +`target/bench/internode-transport//summary.csv`, plus the exact +commands and configuration used. + +## Transport Abstraction Proposal + +### Design principle + +Keep `NodeService` as the control plane. Introduce a separate data transport +only below `RemoteDisk`, where remote disk byte streams are opened today. + +The first implementation should be a no-behavior-change TCP/HTTP backend that +wraps the current `HttpReader`, `HttpWriter`, and `/rustfs/rpc/*` handlers. +Only after that wrapper is benchmarked should an experimental RDMA/RoCE backend +be considered. + +### Candidate boundary + +The narrowest useful boundary is remote disk stream transfer: + +```rust +#[async_trait::async_trait] +pub trait InternodeDataTransport: Send + Sync + std::fmt::Debug { + async fn open_read(&self, request: ReadStreamRequest) -> Result; + async fn open_write(&self, request: WriteStreamRequest) -> Result; + async fn walk_dir(&self, request: WalkDirStreamRequest, writer: &mut dyn AsyncWrite) -> Result<()>; + fn capabilities(&self) -> InternodeTransportCapabilities; +} +``` + +Initial request fields should mirror the current HTTP query parameters: + +- peer endpoint +- disk reference +- volume +- path +- offset +- length +- append/create mode +- expected size +- auth or transfer token material + +The initial TCP backend can keep the current signed HTTP URLs internally. + +### Integration point + +`RemoteDisk` should delegate only these methods to the data transport: + +- `read_file_stream` +- `read_file_zero_copy` as a wrapper over `read_file_stream` unless the backend + supports a stronger zero-copy API +- `append_file` +- `create_file` +- optionally `walk_dir` + +All other `RemoteDisk` methods should continue using the current gRPC client +until measurements prove otherwise. + +### Capability model + +Avoid hard-coding RDMA assumptions into the generic interface. Use capabilities: + +- stream read +- stream write +- bounded range read +- bidirectional streaming +- registered memory support +- scatter/gather support +- zero-copy receive into caller-owned buffers +- authenticated out-of-band transfer +- transport fallback support + +The first TCP backend should report only capabilities that it actually provides. + +## TCP Fallback Requirements + +TCP/HTTP/gRPC must remain the default and required backend. + +Fallback rules: + +- If no explicit data transport is configured, use the current TCP/HTTP + implementation. +- If an experimental backend fails initialization, either fail fast with a clear + error or fall back to TCP only when the configured policy allows fallback. +- Runtime fallback must preserve object correctness and quorum semantics. +- Fallback events must be logged and counted in metrics. +- CI and local development must not require RDMA-capable hardware. + +Suggested future configuration shape: + +```text +RUSTFS_INTERNODE_DATA_TRANSPORT=tcp +RUSTFS_INTERNODE_DATA_TRANSPORT_FALLBACK=tcp +``` + +Do not add these settings until there is an implementation PR that uses them. + +## Future RDMA/RoCE/InfiniBand Boundary + +A future RDMA backend should be experimental and feature-gated. It should be +designed as an optional data-plane backend, not as a replacement for the gRPC +control plane. + +Required design areas: + +- peer capability discovery over the existing gRPC control plane +- connection management and health mapping into existing disk fault handling +- memory registration lifecycle and registration cache +- buffer ownership, pinning, alignment, and lifetime rules +- scatter/gather behavior for erasure shards +- authentication and authorization for out-of-band data transfers +- encryption/TLS-equivalent story or a documented deployment boundary +- timeout, cancellation, retry, and fallback behavior +- metrics for registration cost, transfer latency, bytes, queue depth, retries, + fallback, and errors +- hardware and kernel compatibility matrix + +The first RDMA prototype should target `read_file_stream` and `put_file_stream` +only. `walk_dir`, metadata RPCs, locks, admin RPCs, and bucket coordination +should remain on gRPC unless a later benchmark identifies a specific bottleneck. + +## DPU, DOCA, DPDK, SPDK, and SmartNIC Notes + +These technologies should not drive the first abstraction: + +- DPU/BlueField/DOCA may become relevant for TLS, checksum, compression, or + storage/network offload, but they are vendor- and deployment-specific. +- DPDK is a poor first fit because RustFS is currently an HTTP/S3 object store + and does not have a custom packet data plane. +- SPDK may be relevant only if RustFS adds a raw block or NVMe-oriented local + storage backend. The current disk model is filesystem-based. +- SmartNIC offload should be discussed only after the data-plane boundary and + baseline metrics show where CPU is spent. + +## Suggested PR Sequence + +1. Add this RFC and the current-path classification. +2. Add route-level internode metrics for `/rustfs/rpc/read_file_stream`, + `/rustfs/rpc/put_file_stream`, `/rustfs/rpc/walk_dir`, and gRPC disk byte + calls. +3. Add an internode transport benchmark harness that can run against a local + multi-node cluster and produce repeatable artifacts. +4. Introduce an `InternodeDataTransport` wrapper with a TCP/HTTP backend that + preserves current behavior. +5. Move `RemoteDisk` stream methods to the transport wrapper without changing + default behavior. +6. Add an experimental feature-gated RDMA/RoCE backend only after the baseline + proves that internode byte transfer is a limiting factor. + +## Open Questions + +- Which production workload is the primary target: large-object throughput, + small-object tail latency, healing throughput, or rebalance throughput? +- Should `ReadAll` and `WriteAll` stay as gRPC unary calls, or should large + payloads be redirected to the data transport? +- Is `walk_dir` a metadata control stream or a secondary data-plane stream for + scanner/healing workloads? +- What is the acceptable fallback policy for an explicitly configured + experimental backend? +- How should an RDMA backend preserve authentication and encryption guarantees + currently provided by signed HTTP requests and TLS-capable gRPC/HTTP clients? +- What hardware matrix is required before accepting a non-default RDMA backend? + +## Immediate Next Steps + +- Create a focused issue from this RFC. +- Add route-level internode metrics before changing transport code. +- Extend existing benchmark scripts or add a new script to isolate remote disk + stream read/write throughput. +- Keep the first code PR behavior-preserving and TCP-only. diff --git a/crates/common/src/internode_metrics.rs b/crates/common/src/internode_metrics.rs index a089e3067..228a9e7a2 100644 --- a/crates/common/src/internode_metrics.rs +++ b/crates/common/src/internode_metrics.rs @@ -19,6 +19,19 @@ use std::sync::{ }; use std::time::{Duration, SystemTime, UNIX_EPOCH}; +pub const INTERNODE_OPERATION_READ_FILE_STREAM: &str = "read_file_stream"; +pub const INTERNODE_OPERATION_PUT_FILE_STREAM: &str = "put_file_stream"; +pub const INTERNODE_OPERATION_WALK_DIR: &str = "walk_dir"; +pub const INTERNODE_OPERATION_GRPC_READ_ALL: &str = "grpc_read_all"; +pub const INTERNODE_OPERATION_GRPC_WRITE_ALL: &str = "grpc_write_all"; + +const OPERATION_LABEL: &str = "operation"; +const INTERNODE_OPERATION_SENT_BYTES_TOTAL: &str = "rustfs_system_network_internode_operation_sent_bytes_total"; +const INTERNODE_OPERATION_RECV_BYTES_TOTAL: &str = "rustfs_system_network_internode_operation_recv_bytes_total"; +const INTERNODE_OPERATION_REQUESTS_OUTGOING_TOTAL: &str = "rustfs_system_network_internode_operation_requests_outgoing_total"; +const INTERNODE_OPERATION_REQUESTS_INCOMING_TOTAL: &str = "rustfs_system_network_internode_operation_requests_incoming_total"; +const INTERNODE_OPERATION_ERRORS_TOTAL: &str = "rustfs_system_network_internode_operation_errors_total"; + #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct InternodeMetricsSnapshot { pub sent_bytes_total: u64, @@ -54,6 +67,16 @@ impl InternodeMetrics { counter!("rustfs_system_network_internode_sent_bytes_total").increment(bytes); } + pub fn record_sent_bytes_for_operation(&self, operation: &'static str, bytes: usize) { + self.record_sent_bytes(bytes); + + let bytes = bytes as u64; + if bytes == 0 { + return; + } + counter!(INTERNODE_OPERATION_SENT_BYTES_TOTAL, OPERATION_LABEL => operation).increment(bytes); + } + pub fn record_recv_bytes(&self, bytes: usize) { let bytes = bytes as u64; if bytes == 0 { @@ -63,21 +86,46 @@ impl InternodeMetrics { counter!("rustfs_system_network_internode_recv_bytes_total").increment(bytes); } + pub fn record_recv_bytes_for_operation(&self, operation: &'static str, bytes: usize) { + self.record_recv_bytes(bytes); + + let bytes = bytes as u64; + if bytes == 0 { + return; + } + counter!(INTERNODE_OPERATION_RECV_BYTES_TOTAL, OPERATION_LABEL => operation).increment(bytes); + } + pub fn record_outgoing_request(&self) { self.outgoing_requests_total.fetch_add(1, Ordering::Relaxed); counter!("rustfs_system_network_internode_requests_outgoing_total").increment(1); } + pub fn record_outgoing_request_for_operation(&self, operation: &'static str) { + self.record_outgoing_request(); + counter!(INTERNODE_OPERATION_REQUESTS_OUTGOING_TOTAL, OPERATION_LABEL => operation).increment(1); + } + pub fn record_incoming_request(&self) { self.incoming_requests_total.fetch_add(1, Ordering::Relaxed); counter!("rustfs_system_network_internode_requests_incoming_total").increment(1); } + pub fn record_incoming_request_for_operation(&self, operation: &'static str) { + self.record_incoming_request(); + counter!(INTERNODE_OPERATION_REQUESTS_INCOMING_TOTAL, OPERATION_LABEL => operation).increment(1); + } + pub fn record_error(&self) { self.errors_total.fetch_add(1, Ordering::Relaxed); counter!("rustfs_system_network_internode_errors_total").increment(1); } + pub fn record_error_for_operation(&self, operation: &'static str) { + self.record_error(); + counter!(INTERNODE_OPERATION_ERRORS_TOTAL, OPERATION_LABEL => operation).increment(1); + } + pub fn record_dial_result(&self, duration: Duration, success: bool) { let elapsed_nanos = duration.as_nanos().min(u128::from(u64::MAX)) as u64; self.dial_total_time_nanos.fetch_add(elapsed_nanos, Ordering::Relaxed); @@ -163,4 +211,22 @@ mod tests { metrics.reset_for_test(); } + + #[test] + fn operation_metrics_also_update_aggregate_snapshot() { + let metrics = InternodeMetrics::default(); + + metrics.record_sent_bytes_for_operation(INTERNODE_OPERATION_READ_FILE_STREAM, 128); + metrics.record_recv_bytes_for_operation(INTERNODE_OPERATION_PUT_FILE_STREAM, 256); + metrics.record_outgoing_request_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL); + metrics.record_incoming_request_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); + metrics.record_error_for_operation(INTERNODE_OPERATION_WALK_DIR); + + let snapshot = metrics.snapshot(); + assert_eq!(snapshot.sent_bytes_total, 128); + assert_eq!(snapshot.recv_bytes_total, 256); + assert_eq!(snapshot.outgoing_requests_total, 1); + assert_eq!(snapshot.incoming_requests_total, 1); + assert_eq!(snapshot.errors_total, 1); + } } diff --git a/crates/ecstore/src/rpc/remote_disk.rs b/crates/ecstore/src/rpc/remote_disk.rs index 3591e6353..de5c18f0f 100644 --- a/crates/ecstore/src/rpc/remote_disk.rs +++ b/crates/ecstore/src/rpc/remote_disk.rs @@ -36,6 +36,9 @@ use bytes::Bytes; use futures::lock::Mutex; use http::{HeaderMap, HeaderValue, Method, header::CONTENT_TYPE}; use metrics::counter; +use rustfs_common::internode_metrics::{ + INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL, global_internode_metrics, +}; use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo}; use rustfs_protos::evict_failed_connection; use rustfs_protos::proto_gen::node_service::RenamePartRequest; @@ -1576,11 +1579,12 @@ impl DiskAPI for RemoteDisk { self.execute_with_timeout( || async { + let data_len = data.len(); let disk = self.disk_ref().await; - let mut client = self - .get_client() - .await - .map_err(|err| Error::other(format!("can not get client, err: {err}")))?; + let mut client = self.get_client().await.map_err(|err| { + global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL); + Error::other(format!("can not get client, err: {err}")) + })?; let request = Request::new(WriteAllRequest { disk, volume: volume.to_string(), @@ -1588,9 +1592,19 @@ impl DiskAPI for RemoteDisk { data, }); - let response = client.write_all(request).await?.into_inner(); + global_internode_metrics().record_outgoing_request_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL); + let response = match client.write_all(request).await { + Ok(response) => response.into_inner(), + Err(err) => { + global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL); + return Err(err.into()); + } + }; + + global_internode_metrics().record_sent_bytes_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL, data_len); if !response.success { + global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL); return Err(response.error.unwrap_or_default().into()); } @@ -1608,22 +1622,32 @@ impl DiskAPI for RemoteDisk { self.execute_with_timeout( || async { let disk = self.disk_ref().await; - let mut client = self - .get_client() - .await - .map_err(|err| Error::other(format!("can not get client, err: {err}")))?; + let mut client = self.get_client().await.map_err(|err| { + global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); + Error::other(format!("can not get client, err: {err}")) + })?; let request = Request::new(ReadAllRequest { disk, volume: volume.to_string(), path: path.to_string(), }); - let response = client.read_all(request).await?.into_inner(); + global_internode_metrics().record_outgoing_request_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); + let response = match client.read_all(request).await { + Ok(response) => response.into_inner(), + Err(err) => { + global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); + return Err(err.into()); + } + }; if !response.success { + global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); return Err(response.error.unwrap_or_default().into()); } + global_internode_metrics() + .record_recv_bytes_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL, response.data.len()); Ok(response.data) }, get_max_timeout_duration(), diff --git a/crates/rio/src/http_reader.rs b/crates/rio/src/http_reader.rs index bd770f24b..149a8b44a 100644 --- a/crates/rio/src/http_reader.rs +++ b/crates/rio/src/http_reader.rs @@ -18,7 +18,10 @@ use futures::{Stream, TryStreamExt as _}; use http::HeaderMap; use pin_project_lite::pin_project; use reqwest::{Certificate, Client, Identity, Method, RequestBuilder}; -use rustfs_common::internode_metrics::global_internode_metrics; +use rustfs_common::internode_metrics::{ + INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, INTERNODE_OPERATION_WALK_DIR, + global_internode_metrics, +}; use rustfs_utils::get_env_opt_str; use std::io::IoSlice; use std::io::{self, Error}; @@ -35,6 +38,10 @@ use tokio_util::io::StreamReader; use tokio_util::sync::PollSender; use tracing::error; +const READ_FILE_STREAM_PATH: &str = "/rustfs/rpc/read_file_stream"; +const PUT_FILE_STREAM_PATH: &str = "/rustfs/rpc/put_file_stream"; +const WALK_DIR_PATH: &str = "/rustfs/rpc/walk_dir"; + /// Get the TLS path from the RUSTFS_TLS_PATH environment variable. /// If the variable is not set, return None. fn tls_path() -> Option<&'static std::path::PathBuf> { @@ -145,6 +152,7 @@ pin_project! { method: Method, headers: HeaderMap, track_internode_metrics: bool, + internode_operation: Option<&'static str>, stall_timeout: Option, stall_timer: Option>>, #[pin] @@ -188,6 +196,7 @@ impl HttpReader { stall_timeout: Option, ) -> io::Result { let track_internode_metrics = is_internode_rpc_url(&url); + let internode_operation = internode_rpc_operation(&url); let client = get_http_client(&url); let mut request: RequestBuilder = client.request(method.clone(), url.clone()).headers(headers.clone()); if let Some(body) = body { @@ -195,30 +204,22 @@ impl HttpReader { } let resp = request.send().await.map_err(|e| { - if track_internode_metrics { - global_internode_metrics().record_error(); - } + record_internode_error(track_internode_metrics, internode_operation); Error::other(format!("HttpReader HTTP request error: {e}")) })?; if resp.status().is_success().not() { - if track_internode_metrics { - global_internode_metrics().record_error(); - } + record_internode_error(track_internode_metrics, internode_operation); return Err(Error::other(format!( "HttpReader HTTP request failed with non-200 status {}", resp.status() ))); } - if track_internode_metrics { - global_internode_metrics().record_outgoing_request(); - } + record_internode_outgoing_request(track_internode_metrics, internode_operation); let stream = resp.bytes_stream().map_err(move |e| { - if track_internode_metrics { - global_internode_metrics().record_error(); - } + record_internode_error(track_internode_metrics, internode_operation); Error::other(format!("HttpReader stream error: {e}")) }); @@ -228,6 +229,7 @@ impl HttpReader { method, headers, track_internode_metrics, + internode_operation, stall_timer: stall_timeout.map(|timeout| Box::pin(time::sleep(timeout))), stall_timeout, }) @@ -251,8 +253,8 @@ impl AsyncRead for HttpReader { match this.inner.as_mut().poll_read(cx, buf) { Poll::Ready(Ok(())) => { let bytes_read = buf.filled().len().saturating_sub(filled_before); - if *this.track_internode_metrics && bytes_read > 0 { - global_internode_metrics().record_recv_bytes(bytes_read); + if bytes_read > 0 { + record_internode_recv_bytes(*this.track_internode_metrics, *this.internode_operation, bytes_read); } if bytes_read > 0 { if let Some(stall_timeout) = *this.stall_timeout { @@ -267,9 +269,7 @@ impl AsyncRead for HttpReader { if let Some(timer) = this.stall_timer.as_mut() && timer.as_mut().poll(cx).is_ready() { - if *this.track_internode_metrics { - global_internode_metrics().record_error(); - } + record_internode_error(*this.track_internode_metrics, *this.internode_operation); Poll::Ready(Err(Error::new( io::ErrorKind::TimedOut, "HttpReader stall timeout: no data received before deadline", @@ -305,6 +305,7 @@ impl HashReaderDetector for HttpReader { struct ReceiverStream { receiver: mpsc::Receiver>, track_internode_metrics: bool, + internode_operation: Option<&'static str>, } impl Stream for ReceiverStream { @@ -327,9 +328,7 @@ impl Stream for ReceiverStream { // } match poll { Poll::Ready(Some(Some(bytes))) => { - if self.track_internode_metrics { - global_internode_metrics().record_sent_bytes(bytes.len()); - } + record_internode_sent_bytes(self.track_internode_metrics, self.internode_operation, bytes.len()); Poll::Ready(Some(Ok(bytes))) } Poll::Ready(Some(None)) => Poll::Ready(None), // Sender shutdown @@ -364,6 +363,7 @@ impl HttpWriter { let method_clone = method.clone(); let headers_clone = headers.clone(); let track_internode_metrics = is_internode_rpc_url(&url); + let internode_operation = internode_rpc_operation(&url); let (sender, receiver) = tokio::sync::mpsc::channel::>(HTTP_WRITER_CHANNEL_CAPACITY); let (err_tx, err_rx) = tokio::sync::oneshot::channel::(); @@ -372,6 +372,7 @@ impl HttpWriter { let stream = ReceiverStream { receiver, track_internode_metrics, + internode_operation, }; let body = reqwest::Body::wrap_stream(stream); // http_log!( @@ -391,9 +392,7 @@ impl HttpWriter { Ok(resp) => { // http_log!("[HttpWriter::spawn] got response: status={}", resp.status()); if !resp.status().is_success() { - if track_internode_metrics { - global_internode_metrics().record_error(); - } + record_internode_error(track_internode_metrics, internode_operation); let _ = err_tx.send(Error::other(format!( "HttpWriter HTTP request failed with non-200 status {}", resp.status() @@ -402,9 +401,7 @@ impl HttpWriter { } } Err(e) => { - if track_internode_metrics { - global_internode_metrics().record_error(); - } + record_internode_error(track_internode_metrics, internode_operation); // http_log!("[HttpWriter::spawn] HTTP request error: {e}"); let _ = err_tx.send(Error::other(format!("HTTP request failed: {e}"))); return Err(Error::other(format!("HTTP request failed: {e}"))); @@ -416,9 +413,7 @@ impl HttpWriter { }); // http_log!("[HttpWriter::new] connection established successfully"); - if track_internode_metrics { - global_internode_metrics().record_outgoing_request(); - } + record_internode_outgoing_request(track_internode_metrics, internode_operation); Ok(Self { url, method, @@ -448,6 +443,60 @@ fn is_internode_rpc_url(url: &str) -> bool { url.contains("/rustfs/rpc/") } +fn internode_rpc_operation(url: &str) -> Option<&'static str> { + let url = reqwest::Url::parse(url).ok()?; + match url.path() { + READ_FILE_STREAM_PATH => Some(INTERNODE_OPERATION_READ_FILE_STREAM), + PUT_FILE_STREAM_PATH => Some(INTERNODE_OPERATION_PUT_FILE_STREAM), + WALK_DIR_PATH => Some(INTERNODE_OPERATION_WALK_DIR), + _ => None, + } +} + +fn record_internode_outgoing_request(track: bool, operation: Option<&'static str>) { + if !track { + return; + } + + match operation { + Some(operation) => global_internode_metrics().record_outgoing_request_for_operation(operation), + None => global_internode_metrics().record_outgoing_request(), + } +} + +fn record_internode_sent_bytes(track: bool, operation: Option<&'static str>, bytes: usize) { + if !track { + return; + } + + match operation { + Some(operation) => global_internode_metrics().record_sent_bytes_for_operation(operation, bytes), + None => global_internode_metrics().record_sent_bytes(bytes), + } +} + +fn record_internode_recv_bytes(track: bool, operation: Option<&'static str>, bytes: usize) { + if !track { + return; + } + + match operation { + Some(operation) => global_internode_metrics().record_recv_bytes_for_operation(operation, bytes), + None => global_internode_metrics().record_recv_bytes(bytes), + } +} + +fn record_internode_error(track: bool, operation: Option<&'static str>) { + if !track { + return; + } + + match operation { + Some(operation) => global_internode_metrics().record_error_for_operation(operation), + None => global_internode_metrics().record_error(), + } +} + fn poll_send_error_to_io(err: tokio_util::sync::PollSendError, context: &str) -> io::Error { Error::other(format!("{context}: {err}")) } @@ -681,6 +730,27 @@ mod tests { (format!("http://{addr}/stream"), handle) } + #[test] + fn internode_rpc_operation_maps_known_routes() { + assert_eq!( + internode_rpc_operation(&format!("http://node:9000{READ_FILE_STREAM_PATH}?disk=d")), + Some(INTERNODE_OPERATION_READ_FILE_STREAM) + ); + assert_eq!( + internode_rpc_operation(&format!("http://node:9000{PUT_FILE_STREAM_PATH}?disk=d")), + Some(INTERNODE_OPERATION_PUT_FILE_STREAM) + ); + assert_eq!( + internode_rpc_operation(&format!("http://node:9000{WALK_DIR_PATH}?disk=d")), + Some(INTERNODE_OPERATION_WALK_DIR) + ); + assert_eq!(internode_rpc_operation("http://node:9000/rustfs/rpc/unknown"), None); + assert_eq!( + internode_rpc_operation("http://node:9000/rustfs/rpc/unknown?next=/rustfs/rpc/read_file_stream"), + None + ); + } + #[tokio::test] async fn http_reader_does_not_send_preflight_head() { let state = TestState::default(); diff --git a/rustfs/src/storage/rpc/disk.rs b/rustfs/src/storage/rpc/disk.rs index efc107a37..ec6cf4ef4 100644 --- a/rustfs/src/storage/rpc/disk.rs +++ b/rustfs/src/storage/rpc/disk.rs @@ -13,6 +13,9 @@ // limitations under the License. use super::*; +use rustfs_common::internode_metrics::{ + INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL, global_internode_metrics, +}; use serde::de::DeserializeOwned; use std::io::Cursor; @@ -929,18 +932,25 @@ impl NodeService { pub(super) async fn handle_write_all(&self, request: Request) -> Result, 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); 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 { success: true, error: None, })), - Err(err) => Ok(Response::new(WriteAllResponse { - success: false, - error: Some(err.into()), - })), + Err(err) => { + global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL); + Ok(Response::new(WriteAllResponse { + success: false, + error: Some(err.into()), + })) + } } } else { + global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL); Ok(Response::new(WriteAllResponse { success: false, error: Some(DiskError::other("can not find disk".to_string()).into()), @@ -952,20 +962,28 @@ impl NodeService { debug!("read all"); let request = request.into_inner(); + global_internode_metrics().record_incoming_request_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); if let Some(disk) = self.find_disk(&request.disk).await { match disk.read_all(&request.volume, &request.path).await { - Ok(data) => Ok(Response::new(ReadAllResponse { - success: true, - data, - error: None, - })), - Err(err) => Ok(Response::new(ReadAllResponse { - success: false, - data: Bytes::new(), - error: Some(err.into()), - })), + Ok(data) => { + global_internode_metrics().record_sent_bytes_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL, data.len()); + Ok(Response::new(ReadAllResponse { + success: true, + data, + error: None, + })) + } + Err(err) => { + global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); + Ok(Response::new(ReadAllResponse { + success: false, + data: Bytes::new(), + error: Some(err.into()), + })) + } } } else { + global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); Ok(Response::new(ReadAllResponse { success: false, data: Bytes::new(), diff --git a/rustfs/src/storage/rpc/http_service.rs b/rustfs/src/storage/rpc/http_service.rs index 48e3713a1..e28c256fc 100644 --- a/rustfs/src/storage/rpc/http_service.rs +++ b/rustfs/src/storage/rpc/http_service.rs @@ -18,7 +18,10 @@ use futures_util::TryStreamExt; use http::{HeaderMap, Method, Request, Response, StatusCode, Uri}; use http_body_util::{BodyExt, Limited}; use hyper::body::Incoming; -use rustfs_common::internode_metrics::global_internode_metrics; +use rustfs_common::internode_metrics::{ + INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, INTERNODE_OPERATION_WALK_DIR, + global_internode_metrics, +}; use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE; use rustfs_ecstore::disk::{DiskAPI, WalkDirOptions}; use rustfs_ecstore::rpc::verify_rpc_signature; @@ -106,8 +109,9 @@ fn is_internode_rpc_path(path: &str) -> bool { } async fn handle_internode_rpc(req: Request) -> Response { + let operation = internode_http_operation(req.uri().path()); if let Err(response) = verify_internode_rpc_signature(req.uri(), req.method(), req.headers()) { - global_internode_metrics().record_error(); + record_internode_rpc_error(operation); return *response; } @@ -122,12 +126,28 @@ async fn handle_internode_rpc(req: Request) -> Response { }; if !response.status().is_success() { - global_internode_metrics().record_error(); + record_internode_rpc_error(operation); } response } +fn internode_http_operation(path: &str) -> Option<&'static str> { + match path { + READ_FILE_STREAM_PATH => Some(INTERNODE_OPERATION_READ_FILE_STREAM), + PUT_FILE_STREAM_PATH => Some(INTERNODE_OPERATION_PUT_FILE_STREAM), + WALK_DIR_PATH => Some(INTERNODE_OPERATION_WALK_DIR), + _ => None, + } +} + +fn record_internode_rpc_error(operation: Option<&'static str>) { + match operation { + Some(operation) => global_internode_metrics().record_error_for_operation(operation), + None => global_internode_metrics().record_error(), + } +} + fn verify_internode_rpc_signature(uri: &Uri, method: &Method, headers: &HeaderMap) -> Result<(), RpcErrorResponse> { if method == Method::HEAD { return Ok(()); @@ -163,8 +183,8 @@ async fn handle_read_file(req: Request) -> Response { Err(e) => return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("read file err {e}")), }; - global_internode_metrics().record_incoming_request(); - let stream = read_file_body_stream(file, query.length); + global_internode_metrics().record_incoming_request_for_operation(INTERNODE_OPERATION_READ_FILE_STREAM); + let stream = read_file_body_stream(file, query.length, INTERNODE_OPERATION_READ_FILE_STREAM); Response::builder() .status(StatusCode::OK) @@ -172,13 +192,17 @@ async fn handle_read_file(req: Request) -> Response { .expect("failed to build read file stream response") } -fn read_file_body_stream(reader: R, length: usize) -> Pin> + Send + Sync>> +fn read_file_body_stream( + reader: R, + length: usize, + operation: &'static str, +) -> Pin> + Send + Sync>> where R: tokio::io::AsyncRead + Unpin + Send + Sync + 'static, { let metrics = global_internode_metrics().clone(); let stream = ReaderStream::with_capacity(reader, DEFAULT_READ_BUFFER_SIZE).map_ok(move |bytes| { - metrics.record_sent_bytes(bytes.len()); + metrics.record_sent_bytes_for_operation(operation, bytes.len()); bytes }); @@ -220,10 +244,10 @@ async fn handle_walk_dir(req: Request) -> Response { } }); - global_internode_metrics().record_incoming_request(); + global_internode_metrics().record_incoming_request_for_operation(INTERNODE_OPERATION_WALK_DIR); let metrics = global_internode_metrics().clone(); let stream = ReaderStream::with_capacity(rd, DEFAULT_READ_BUFFER_SIZE).map_ok(move |bytes| { - metrics.record_sent_bytes(bytes.len()); + metrics.record_sent_bytes_for_operation(INTERNODE_OPERATION_WALK_DIR, bytes.len()); bytes }); @@ -260,8 +284,8 @@ async fn handle_put_file(req: Request) -> Response { Err(e) => return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("write file err {e}")), }; - global_internode_metrics().record_incoming_request(); - global_internode_metrics().record_recv_bytes(copied as usize); + 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); if let Err(e) = file.flush().await { return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("write file err {e}")); @@ -337,6 +361,17 @@ mod tests { assert!(!is_internode_rpc_path("/rustfs/admin/v3/info")); } + #[test] + fn internode_http_operation_maps_only_known_routes() { + assert_eq!( + internode_http_operation(READ_FILE_STREAM_PATH), + Some(INTERNODE_OPERATION_READ_FILE_STREAM) + ); + assert_eq!(internode_http_operation(PUT_FILE_STREAM_PATH), Some(INTERNODE_OPERATION_PUT_FILE_STREAM)); + assert_eq!(internode_http_operation(WALK_DIR_PATH), Some(INTERNODE_OPERATION_WALK_DIR)); + assert_eq!(internode_http_operation("/rustfs/rpc/unknown"), None); + } + #[test] fn rpc_head_signature_verification_is_skipped() { let uri: Uri = READ_FILE_STREAM_PATH.parse().expect("uri"); @@ -377,7 +412,7 @@ mod tests { writer.write_all(b"hello world").await.expect("write succeeds"); }); - let mut stream = read_file_body_stream(reader, 0); + let mut stream = read_file_body_stream(reader, 0, INTERNODE_OPERATION_READ_FILE_STREAM); let mut out = Vec::new(); while let Some(chunk) = stream.next().await { out.extend_from_slice(&chunk.expect("chunk succeeds")); @@ -393,7 +428,7 @@ mod tests { writer.write_all(b"hello world").await.expect("write succeeds"); }); - let mut stream = read_file_body_stream(reader, 5); + let mut stream = read_file_body_stream(reader, 5, INTERNODE_OPERATION_READ_FILE_STREAM); let mut out = Vec::new(); while let Some(chunk) = stream.next().await { out.extend_from_slice(&chunk.expect("chunk succeeds"));