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 <marshawcoco@users.noreply.github.com>

.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

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/<timestamp>/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<FileReader>;
+    async fn open_write(&self, request: WriteStreamRequest) -> Result<FileWriter>;
+    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.

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);
+    }
 }

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
+                let mut client = self.get_client().await.map_err(|err| {
-                    .get_client()
+                    global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL);
-                    .await
+                    Error::other(format!("can not get client, err: {err}"))
-                    .map_err(|err| 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
+                let mut client = self.get_client().await.map_err(|err| {
-                    .get_client()
+                    global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL);
-                    .await
+                    Error::other(format!("can not get client, err: {err}"))
-                    .map_err(|err| 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(),

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<Duration>,
         stall_timer: Option<Pin<Box<Sleep>>>,
         #[pin]
@@ -188,6 +196,7 @@ impl HttpReader {
         stall_timeout: Option<Duration>,
     ) -> io::Result<Self> {
         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 {
+            record_internode_error(track_internode_metrics, internode_operation);
-                global_internode_metrics().record_error();
-            }
             Error::other(format!("HttpReader HTTP request error: {e}"))
         })?;
 
         if resp.status().is_success().not() {
-            if track_internode_metrics {
+            record_internode_error(track_internode_metrics, internode_operation);
-                global_internode_metrics().record_error();
-            }
             return Err(Error::other(format!(
                 "HttpReader HTTP request failed with non-200 status {}",
                 resp.status()
             )));
         }
 
-        if track_internode_metrics {
+        record_internode_outgoing_request(track_internode_metrics, internode_operation);
-            global_internode_metrics().record_outgoing_request();
-        }
 
         let stream = resp.bytes_stream().map_err(move |e| {
-            if track_internode_metrics {
+            record_internode_error(track_internode_metrics, internode_operation);
-                global_internode_metrics().record_error();
-            }
             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 {
+                if bytes_read > 0 {
-                    global_internode_metrics().record_recv_bytes(bytes_read);
+                    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 {
+                    record_internode_error(*this.track_internode_metrics, *this.internode_operation);
-                        global_internode_metrics().record_error();
-                    }
                     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<Option<Bytes>>,
     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 {
+                record_internode_sent_bytes(self.track_internode_metrics, self.internode_operation, bytes.len());
-                    global_internode_metrics().record_sent_bytes(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::<Option<Bytes>>(HTTP_WRITER_CHANNEL_CAPACITY);
         let (err_tx, err_rx) = tokio::sync::oneshot::channel::<io::Error>();
@@ -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 {
+                        record_internode_error(track_internode_metrics, internode_operation);
-                            global_internode_metrics().record_error();
-                        }
                         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 {
+                    record_internode_error(track_internode_metrics, internode_operation);
-                        global_internode_metrics().record_error();
-                    }
                     // 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 {
+        record_internode_outgoing_request(track_internode_metrics, internode_operation);
-            global_internode_metrics().record_outgoing_request();
-        }
         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<T>(err: tokio_util::sync::PollSendError<T>, 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();

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<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);
         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 {
+                Err(err) => {
-                    success: false,
+                    global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL);
-                    error: Some(err.into()),
+                    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 {
+                Ok(data) => {
-                    success: true,
+                    global_internode_metrics().record_sent_bytes_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL, data.len());
-                    data,
+                    Ok(Response::new(ReadAllResponse {
-                    error: None,
+                        success: true,
-                })),
+                        data,
-                Err(err) => Ok(Response::new(ReadAllResponse {
+                        error: None,
-                    success: false,
+                    }))
-                    data: Bytes::new(),
+                }
-                    error: Some(err.into()),
+                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(),

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<Incoming>) -> Response<Body> {
+    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<Incoming>) -> Response<Body> {
     };
 
     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<Incoming>) -> Response<Body> {
         Err(e) => return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("read file err {e}")),
     };
 
-    global_internode_metrics().record_incoming_request();
+    global_internode_metrics().record_incoming_request_for_operation(INTERNODE_OPERATION_READ_FILE_STREAM);
-    let stream = read_file_body_stream(file, query.length);
+    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<Incoming>) -> Response<Body> {
         .expect("failed to build read file stream response")
 }
 
-fn read_file_body_stream<R>(reader: R, length: usize) -> Pin<Box<dyn futures::Stream<Item = io::Result<Bytes>> + Send + Sync>>
+fn read_file_body_stream<R>(
+    reader: R,
+    length: usize,
+    operation: &'static str,
+) -> Pin<Box<dyn futures::Stream<Item = io::Result<Bytes>> + 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<Incoming>) -> Response<Body> {
         }
     });
 
-    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<Incoming>) -> Response<Body> {
         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_incoming_request_for_operation(INTERNODE_OPERATION_PUT_FILE_STREAM);
-    global_internode_metrics().record_recv_bytes(copied as usize);
+    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"));