From 6f613317f6b9f9e5330574b72d93d8e75408eb6a Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 7 Jul 2026 05:18:31 +0800 Subject: [PATCH] feat(internode): optimize gRPC transport (#4337) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(internode): P0 gRPC transport tuning, message limits, payload metrics Land the P0 subtask from docs/grpc-optimization: close the client-vs-server transport gaps and add instrumentation to size which unary RPCs need channel isolation in P1. Transport tuning (G3): the client `Endpoint` now disables Nagle and raises the HTTP/2 stream/connection flow-control windows to mirror the server socket, so small lock/health RPCs are not batched and larger metadata responses are not throttled by the 64KiB default window. All env-overridable, 0 opts out. Message-size limits (G1): both `NodeServiceClient` and `NodeServiceServer` set max decode/encode size (default 100MiB) instead of tonic's silent 4MiB cap, so a large multi-version xl.meta or aggregated ReadMultiple no longer fails out_of_range. The server limit is set on `NodeServiceServer` before wrapping in the auth `InterceptedService` (the interceptor type does not expose it). Payload instrumentation (P1 prep): ReadAll/ReadMultiple record a payload-size histogram plus a large-payload counter when a response crosses the configured threshold (default 8MiB), feeding alerting on paths that contend with latency-sensitive control-plane traffic on the shared channel. Threshold-only counter, no per-call hot-path log. Verification: cargo check/test on config, io-metrics, ecstore, rustfs; clippy clean on touched files; make pre-commit green. Co-Authored-By: heihutu * feat(internode): P1 control/bulk gRPC channel isolation (opt-in) Land the P1 subtask from docs/grpc-optimization: physically separate large bytes-carrying unary RPCs from latency-sensitive control-plane RPCs so a big transfer can no longer head-of-line block a lock/health RPC on the shared HTTP/2 connection (G2/G5). Introduce ChannelClass { Control, Bulk } and get_channel_for_class in protos. Control RPCs keep the per-peer connection keyed by the bare address; Bulk RPCs (ReadAll/WriteAll/ReadMultiple/BatchReadVersion, via a new get_bulk_client) are round-robined across a small per-peer bulk pool. Rather than restructuring the global GLOBAL_CONN_MAP (and every consumer), bulk channels are cached under a composite key (addr\0bulk\0idx). The NUL separator cannot appear in a URL, so bulk keys never collide with the control key. This keeps the blast radius small on a consistency-sensitive path. create_new_channel is refactored into build_channel(dial_addr, cache_key) so several physically distinct channels to one peer cache independently while dialing/TLS still use the real address. Gated by RUSTFS_INTERNODE_CHANNEL_ISOLATION (default OFF) so the default build is byte-for-byte the pre-P1 behavior: bulk resolves to the control channel and the switch is a single-env rollback. RUSTFS_INTERNODE_BULK_CHANNELS (default 2, clamped >=1) sizes the pool. On failure, evict_failed_connection drops the whole bulk pool for the peer (round-robin hides which index was used), avoiding half-dead cached channels. Lock RPCs (remote_locker) already use the default Control path, so lock semantics and retry behavior are unchanged. Verification: cargo check/test on config, protos, ecstore, rustfs; new protos tests for bulk key routing and isolation-off passthrough; clippy clean on touched files; make pre-commit green. Co-Authored-By: heihutu * feat(internode): P2 msgpack/JSON codec observability + encode buffer presizing Land the safe, wire-compatible slice of P2 from docs/grpc-optimization: the observability prerequisite for retiring the redundant JSON fields, plus a codec micro-optimization. No proto/wire-format change; JSON is still dual-written. Internode RPCs today dual-encode each metadata value as both msgpack (`*_bin`) and a JSON compatibility string, and decoders prefer `_bin` with a JSON fallback. Before the JSON fields can ever be dropped (a cross-version change), that fallback must be proven unused in production. Add rustfs_system_network_internode_msgpack_json_fallback_total{direction, message}: incremented whenever a decode falls back to the JSON field because the msgpack payload was absent. Wired into both directions — the client decoding peer responses (remote_disk.rs, incl. the list-level read_multiple/batch fallbacks) and the server decoding peer requests (node_service/disk.rs). This counter must read zero across a release window before send paths stop writing JSON and the proto text fields are reserved/removed (the deferred P2-1 steps). Also pre-size the msgpack encode buffers (Vec::with_capacity(512)) on both sides, eliminating the repeated growth reallocations for typical FileInfo payloads with zero added copy. Full thread_local buffer pooling is deferred: it needs either an extra copy (unclear net win) or a send-path buffer-return lifecycle, to be justified by a codec microbenchmark first. Verification: cargo check/test on io-metrics, ecstore, rustfs; new fallback counter smoke test; existing codec decode tests green; clippy clean on touched files; make pre-commit green. Co-Authored-By: heihutu * docs(internode): add msgpack/JSON convergence observation runbook Runbook driving the observation-gated retirement of the redundant JSON compatibility fields on internode gRPC metadata RPCs (grpc-optimization P2-1). Documents the shipped fallback counter (rustfs_system_network_internode_msgpack_json_fallback_total{direction,message}), the PromQL to confirm it reads zero across a release window, a standing alert, and the staged flip/rollback procedure (env-gated msgpack-only send, then proto field removal in N+1). Includes the verified field -> peer-decoder audit: only fields whose peer decodes _bin first may be converged. Notes DeleteVersion.opts (DeleteOptions) is NOT convergence-ready — its server handler is not _bin-first and must gain a decode_msgpack_or_json path first. This gates the send-side change so it cannot empty a JSON field an old peer still needs. Co-Authored-By: heihutu * feat(internode): env-gated msgpack-only send + DeleteVersion _bin support (P2-1) Implements the send-side lever for retiring the redundant JSON compatibility fields on internode gRPC metadata RPCs, plus the missing `_bin` support on the delete path that it depends on (grpc-optimization P2-1). Default-off: the base build is byte-for-byte the prior dual-write behavior. Gated msgpack-only send (RUSTFS_INTERNODE_RPC_MSGPACK_ONLY, default false): - New rustfs_protos::internode_rpc_msgpack_only() reads the flag. - Client (remote_disk.rs) compat_json() and server (node_service/disk.rs) compat_response_json() emit an empty JSON string when the flag is on, so only the msgpack _bin payload is sent. The _bin field is always sent; decoders keep the JSON read fallback. Applied only to fields with a confirmed _bin-first peer decoder (WriteMetadata/UpdateMetadata/RenameData file_info, UpdateMetadata opts, ReadOptions, ReadMultipleReq, BatchReadVersionReq; ReadVersion/ReadXL/RenameData responses and the ReadMultiple/BatchReadVersion response lists). - Only enable after the P2 fallback counter has read zero across a release window (see docs/operations/internode-msgpack-json-convergence-runbook.md). Single-env rollback; no wire-format break. DeleteVersion(s) _bin support (prerequisite): - The DeleteVersion/DeleteVersions protos had NO _bin fields. Add additive (backward-compatible) bytes file_info_bin/opts_bin (DeleteVersion) and repeated bytes versions_bin + bytes opts_bin (DeleteVersions); regenerate the checked-in prost struct. - Client dual-writes them; server decodes them _bin-first with JSON fallback. - These delete fields are kept OUT of the msgpack-only set (always dual-write) until their own fallback counter reads zero across a window with the new decoders fully deployed. DeleteVersion.raw_file_info stays JSON-only (no _bin field yet). Verification: cargo check/test on protos, config, ecstore, rustfs (incl. the six delete request handler tests and a compat_json default-path test); clippy clean on touched files; make pre-commit green. Co-Authored-By: heihutu * feat(internode): P3 cluster peer online/offline health metric Land the safe observability core of P3 (grpc-optimization G6/G8): track each internode peer's reachability and expose the offline count, for parity with MinIO's minio_cluster_servers_offline_total. Pure instrumentation — peer selection and quorum are unchanged. - io-metrics: per-peer PeerHealthState { online, consecutive_failures } registry plus record_peer_reachable/record_peer_unreachable. A peer flips offline after N consecutive failures (dial failures or RPC-triggered evictions) and back online on the next successful dial; the count of offline peers is published to the rustfs_cluster_servers_offline_total gauge. - config: RUSTFS_INTERNODE_OFFLINE_FAILURE_THRESHOLD (default 3, clamped >= 1). - protos: build_channel marks the peer reachable on a successful dial and unreachable on a dial failure; evict_failed_connection feeds the failure signal too. Keyed by the real peer address, so control and bulk channels to one peer share health state. Deferred (documented in docs/grpc-optimization P3): startup prewarm (no clean topology-ready hook yet), the offline fast-bypass in peer routing (consistency- sensitive; must not change quorum), and idempotent-read-only retry. This commit is observability only. Verification: cargo check/test on io-metrics, config, protos (new peer-health state-machine and threshold-clamp tests); clippy clean on touched files; make pre-commit green. Co-Authored-By: heihutu * feat(internode): P3 control-channel prewarm + self-healing offline bypass Add the remaining P3 connection-lifecycle levers (grpc-optimization G6/G8), both env-gated and default-off so the base build is unchanged. Prewarm (RUSTFS_INTERNODE_PREWARM, default off): RemoteDisk::new spawns a best-effort background dial of the peer's control channel, deduped per peer address, moving the connect cost off the first RPC. Failures fall through to the existing lazy connect + recovery monitor. Offline bypass (RUSTFS_INTERNODE_OFFLINE_BYPASS, default off): remote_disk get_client/get_bulk_client fast-fail a peer already marked offline instead of paying the connect timeout, so the erasure layer proceeds on quorum sooner. This does NOT change quorum. It is self-healing: cluster_peer_should_bypass lets one request per RUSTFS_INTERNODE_OFFLINE_REPROBE_SECS (default 5s) through to recover the peer even with no background monitor, and the recovery monitor's own probe path calls the client directly so it is never bypassed. io-metrics gains cluster_peer_is_offline / cluster_peer_should_bypass (with a per-peer re-probe timestamp). Scope: data path only — remote_locker (lock RPCs, most consistency-sensitive) is left dual-writing/unbypassed as a follow-up. Verification: cargo check/test on io-metrics, config, ecstore (new self-healing bypass tests; all 105 rpc tests green); clippy clean on touched files; make pre-commit green. Co-Authored-By: heihutu * docs(internode): add A/B benchmark runbook for gRPC optimization stages Reproducible before/after collection procedure for grpc-optimization P0–P3. Since every stage is env-gated, before/after is the same binary with different env — no rebuild. Documents, per stage: the exact env toggles (baseline vs enabled column), which existing bench script to run (run_internode_transport_baseline.sh / run_four_node_cluster_failover_bench.sh), the Prometheus metrics to capture, and the acceptance gates from the design docs (e.g. lock p99 down >= 20% for P1, msgpack fallback counter = 0 before enabling P2, correct rustfs_cluster_servers_offline_total for P3). Live runs require a multi-node cluster + load tool + Prometheus scrape and cannot be produced in a single-process sandbox; artifacts land under target/bench (gitignored) and attach to the PR. Co-Authored-By: heihutu * feat(internode): P3-2 lock-path offline bypass + P3-3 idempotent read retry Extend the offline bypass to the lock path and add opt-in retries for idempotent reads (grpc-optimization P3-2/P3-3). Both env-gated and default-off/zero. Offline bypass (lock path): factor the bypass decision into a shared pub(crate) internode_offline_bypass_reason(addr) and call it from remote_locker::get_client too, so lock RPCs to an offline peer fast-fail (letting dsync reach quorum sooner) instead of paying the connect timeout. Does not change quorum; the self-healing re-probe keeps peers recoverable. Gated by RUSTFS_INTERNODE_OFFLINE_BYPASS (default off). Idempotent read retry (P3-3): add execute_read_with_retry — a bounded, exponential-backoff retry for read-only/reentrant RPCs on transient network errors — and route disk_info through it. RUSTFS_INTERNODE_IDEMPOTENT_READ_RETRIES defaults to 0 (disabled). Write/lock RPCs are never retried (quorum/idempotency safety, per CLAUDE.md); the wrapper requires an Fn closure so only reads that rebuild their request from borrowed inputs qualify. Deferred: grpc.health.v1 (optional ecosystem-compat only; needs a new tonic-health dep and 3-way hybrid-service wiring — internal needs are met by the existing Ping RPC). Verification: cargo check/test on config, ecstore (105 rpc tests green incl. disk_info now via the retry wrapper); clippy clean on touched files; make pre-commit green. Co-Authored-By: heihutu * feat(scripts): one-click internode gRPC A/B benchmark driver Wrap the per-stage env matrix from the benchmark runbook into scripts/run_internode_grpc_ab_bench.sh: given --stage and --phase , it emits the stage/phase RUSTFS_INTERNODE_* server env to /server-env.sh and runs the right underlying bench (run_internode_transport_baseline.sh for p0/p1/p2, run_four_node_cluster_failover_bench.sh for p3) into a labeled target/bench/internode-transport/-/. Passthrough args after `--` reach the underlying bench; --dry-run previews the env + command. The script is explicit that RUSTFS_INTERNODE_* are server env, so for the load-driven stages the operator must restart rustfs with the emitted env before the run; the docker four-node (p3) path exports them for a forwarding compose. shellcheck-clean. Runbook updated with a "One-click driver" section. Co-Authored-By: heihutu * chore(compose): forward RUSTFS_INTERNODE_* into the four-node cluster The four-node local-build compose only forwarded a fixed whitelist of env, so the internode gRPC knobs (grpc-optimization P0-P3) never reached the containers and the A/B bench driver's "after" phase was a no-op. Forward the full RUSTFS_INTERNODE_* set with defaults matching the binary defaults, so leaving them unset is a no-op and the A/B driver can toggle a stage per phase. Co-Authored-By: heihutu * fix(internode): address Copilot review — retry health action + poison-safe peer health Two review nits on #4337: - P3-3 idempotent read retry (remote_disk.rs): execute_read_with_retry ran every attempt through execute_with_timeout_for_op, which hardcodes FailureHealthAction::MarkFailure. So the first transient error could flip the disk faulty and short-circuit the remaining retries, and each attempt over-counted the failure. Route all but the final attempt through execute_with_timeout_for_op_and_health_action with IgnoreFailure; only the last attempt marks faulty/evicts. No default impact (retries default 0). - Peer-health helpers (io-metrics): record_peer_reachable/record_peer_unreachable, cluster_peer_is_offline and cluster_peer_should_bypass early-returned on a poisoned mutex, permanently stalling the offline gauge and bypass state after a single panic. Recover via PoisonError::into_inner(). Verification: cargo check/test on io-metrics + ecstore (105 rpc tests green); clippy clean on touched files; make pre-commit green. Co-Authored-By: heihutu --------- Co-authored-by: heihutu --- .../docker-compose.cluster.local-build.yml | 56 ++++ crates/config/src/constants/internode.rs | 199 +++++++++++ crates/ecstore/src/cluster/rpc/client.rs | 45 ++- crates/ecstore/src/cluster/rpc/remote_disk.rs | 312 ++++++++++++++++-- .../ecstore/src/cluster/rpc/remote_locker.rs | 6 + .../src/cluster/rpc/runtime_sources.rs | 42 ++- crates/io-metrics/src/internode_metrics.rs | 247 +++++++++++++- .../src/generated/proto_gen/node_service.rs | 8 + crates/protos/src/lib.rs | 225 ++++++++++++- crates/protos/src/node.proto | 8 + crates/protos/src/runtime_sources.rs | 11 + .../internode-grpc-benchmark-runbook.md | 97 ++++++ ...ernode-msgpack-json-convergence-runbook.md | 145 ++++++++ rustfs/src/server/http.rs | 15 +- rustfs/src/storage/rpc/node_service.rs | 6 + rustfs/src/storage/rpc/node_service/disk.rs | 51 ++- scripts/run_internode_grpc_ab_bench.sh | 164 +++++++++ 17 files changed, 1571 insertions(+), 66 deletions(-) create mode 100644 docs/operations/internode-grpc-benchmark-runbook.md create mode 100644 docs/operations/internode-msgpack-json-convergence-runbook.md create mode 100755 scripts/run_internode_grpc_ab_bench.sh diff --git a/.docker/compose/docker-compose.cluster.local-build.yml b/.docker/compose/docker-compose.cluster.local-build.yml index 121b5f671..615219162 100644 --- a/.docker/compose/docker-compose.cluster.local-build.yml +++ b/.docker/compose/docker-compose.cluster.local-build.yml @@ -33,6 +33,20 @@ services: - RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT=${RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT:-5} - RUSTFS_LOCK_ACQUIRE_TIMEOUT=${RUSTFS_LOCK_ACQUIRE_TIMEOUT:-5} - RUSTFS_UNSAFE_BYPASS_DISK_CHECK=${RUSTFS_UNSAFE_BYPASS_DISK_CHECK:-true} + # Internode gRPC optimization knobs (grpc-optimization P0-P3). Defaults match the binary + # defaults, so leaving these unset is a no-op; the A/B driver exports them to toggle a stage. + - RUSTFS_INTERNODE_RPC_TCP_NODELAY=${RUSTFS_INTERNODE_RPC_TCP_NODELAY:-true} + - RUSTFS_INTERNODE_RPC_HTTP2_STREAM_WINDOW_SIZE=${RUSTFS_INTERNODE_RPC_HTTP2_STREAM_WINDOW_SIZE:-1048576} + - RUSTFS_INTERNODE_RPC_HTTP2_CONN_WINDOW_SIZE=${RUSTFS_INTERNODE_RPC_HTTP2_CONN_WINDOW_SIZE:-2097152} + - RUSTFS_INTERNODE_RPC_MAX_MESSAGE_SIZE=${RUSTFS_INTERNODE_RPC_MAX_MESSAGE_SIZE:-104857600} + - RUSTFS_INTERNODE_CHANNEL_ISOLATION=${RUSTFS_INTERNODE_CHANNEL_ISOLATION:-false} + - RUSTFS_INTERNODE_BULK_CHANNELS=${RUSTFS_INTERNODE_BULK_CHANNELS:-2} + - RUSTFS_INTERNODE_RPC_MSGPACK_ONLY=${RUSTFS_INTERNODE_RPC_MSGPACK_ONLY:-false} + - RUSTFS_INTERNODE_PREWARM=${RUSTFS_INTERNODE_PREWARM:-false} + - RUSTFS_INTERNODE_OFFLINE_BYPASS=${RUSTFS_INTERNODE_OFFLINE_BYPASS:-false} + - RUSTFS_INTERNODE_OFFLINE_REPROBE_SECS=${RUSTFS_INTERNODE_OFFLINE_REPROBE_SECS:-5} + - RUSTFS_INTERNODE_OFFLINE_FAILURE_THRESHOLD=${RUSTFS_INTERNODE_OFFLINE_FAILURE_THRESHOLD:-3} + - RUSTFS_INTERNODE_IDEMPOTENT_READ_RETRIES=${RUSTFS_INTERNODE_IDEMPOTENT_READ_RETRIES:-0} extra_hosts: - "host.docker.internal:host-gateway" volumes: @@ -65,6 +79,20 @@ services: - RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT=${RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT:-5} - RUSTFS_LOCK_ACQUIRE_TIMEOUT=${RUSTFS_LOCK_ACQUIRE_TIMEOUT:-5} - RUSTFS_UNSAFE_BYPASS_DISK_CHECK=${RUSTFS_UNSAFE_BYPASS_DISK_CHECK:-true} + # Internode gRPC optimization knobs (grpc-optimization P0-P3). Defaults match the binary + # defaults, so leaving these unset is a no-op; the A/B driver exports them to toggle a stage. + - RUSTFS_INTERNODE_RPC_TCP_NODELAY=${RUSTFS_INTERNODE_RPC_TCP_NODELAY:-true} + - RUSTFS_INTERNODE_RPC_HTTP2_STREAM_WINDOW_SIZE=${RUSTFS_INTERNODE_RPC_HTTP2_STREAM_WINDOW_SIZE:-1048576} + - RUSTFS_INTERNODE_RPC_HTTP2_CONN_WINDOW_SIZE=${RUSTFS_INTERNODE_RPC_HTTP2_CONN_WINDOW_SIZE:-2097152} + - RUSTFS_INTERNODE_RPC_MAX_MESSAGE_SIZE=${RUSTFS_INTERNODE_RPC_MAX_MESSAGE_SIZE:-104857600} + - RUSTFS_INTERNODE_CHANNEL_ISOLATION=${RUSTFS_INTERNODE_CHANNEL_ISOLATION:-false} + - RUSTFS_INTERNODE_BULK_CHANNELS=${RUSTFS_INTERNODE_BULK_CHANNELS:-2} + - RUSTFS_INTERNODE_RPC_MSGPACK_ONLY=${RUSTFS_INTERNODE_RPC_MSGPACK_ONLY:-false} + - RUSTFS_INTERNODE_PREWARM=${RUSTFS_INTERNODE_PREWARM:-false} + - RUSTFS_INTERNODE_OFFLINE_BYPASS=${RUSTFS_INTERNODE_OFFLINE_BYPASS:-false} + - RUSTFS_INTERNODE_OFFLINE_REPROBE_SECS=${RUSTFS_INTERNODE_OFFLINE_REPROBE_SECS:-5} + - RUSTFS_INTERNODE_OFFLINE_FAILURE_THRESHOLD=${RUSTFS_INTERNODE_OFFLINE_FAILURE_THRESHOLD:-3} + - RUSTFS_INTERNODE_IDEMPOTENT_READ_RETRIES=${RUSTFS_INTERNODE_IDEMPOTENT_READ_RETRIES:-0} extra_hosts: - "host.docker.internal:host-gateway" volumes: @@ -97,6 +125,20 @@ services: - RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT=${RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT:-5} - RUSTFS_LOCK_ACQUIRE_TIMEOUT=${RUSTFS_LOCK_ACQUIRE_TIMEOUT:-5} - RUSTFS_UNSAFE_BYPASS_DISK_CHECK=${RUSTFS_UNSAFE_BYPASS_DISK_CHECK:-true} + # Internode gRPC optimization knobs (grpc-optimization P0-P3). Defaults match the binary + # defaults, so leaving these unset is a no-op; the A/B driver exports them to toggle a stage. + - RUSTFS_INTERNODE_RPC_TCP_NODELAY=${RUSTFS_INTERNODE_RPC_TCP_NODELAY:-true} + - RUSTFS_INTERNODE_RPC_HTTP2_STREAM_WINDOW_SIZE=${RUSTFS_INTERNODE_RPC_HTTP2_STREAM_WINDOW_SIZE:-1048576} + - RUSTFS_INTERNODE_RPC_HTTP2_CONN_WINDOW_SIZE=${RUSTFS_INTERNODE_RPC_HTTP2_CONN_WINDOW_SIZE:-2097152} + - RUSTFS_INTERNODE_RPC_MAX_MESSAGE_SIZE=${RUSTFS_INTERNODE_RPC_MAX_MESSAGE_SIZE:-104857600} + - RUSTFS_INTERNODE_CHANNEL_ISOLATION=${RUSTFS_INTERNODE_CHANNEL_ISOLATION:-false} + - RUSTFS_INTERNODE_BULK_CHANNELS=${RUSTFS_INTERNODE_BULK_CHANNELS:-2} + - RUSTFS_INTERNODE_RPC_MSGPACK_ONLY=${RUSTFS_INTERNODE_RPC_MSGPACK_ONLY:-false} + - RUSTFS_INTERNODE_PREWARM=${RUSTFS_INTERNODE_PREWARM:-false} + - RUSTFS_INTERNODE_OFFLINE_BYPASS=${RUSTFS_INTERNODE_OFFLINE_BYPASS:-false} + - RUSTFS_INTERNODE_OFFLINE_REPROBE_SECS=${RUSTFS_INTERNODE_OFFLINE_REPROBE_SECS:-5} + - RUSTFS_INTERNODE_OFFLINE_FAILURE_THRESHOLD=${RUSTFS_INTERNODE_OFFLINE_FAILURE_THRESHOLD:-3} + - RUSTFS_INTERNODE_IDEMPOTENT_READ_RETRIES=${RUSTFS_INTERNODE_IDEMPOTENT_READ_RETRIES:-0} extra_hosts: - "host.docker.internal:host-gateway" volumes: @@ -129,6 +171,20 @@ services: - RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT=${RUSTFS_OBJECT_LOCK_ACQUIRE_TIMEOUT:-5} - RUSTFS_LOCK_ACQUIRE_TIMEOUT=${RUSTFS_LOCK_ACQUIRE_TIMEOUT:-5} - RUSTFS_UNSAFE_BYPASS_DISK_CHECK=${RUSTFS_UNSAFE_BYPASS_DISK_CHECK:-true} + # Internode gRPC optimization knobs (grpc-optimization P0-P3). Defaults match the binary + # defaults, so leaving these unset is a no-op; the A/B driver exports them to toggle a stage. + - RUSTFS_INTERNODE_RPC_TCP_NODELAY=${RUSTFS_INTERNODE_RPC_TCP_NODELAY:-true} + - RUSTFS_INTERNODE_RPC_HTTP2_STREAM_WINDOW_SIZE=${RUSTFS_INTERNODE_RPC_HTTP2_STREAM_WINDOW_SIZE:-1048576} + - RUSTFS_INTERNODE_RPC_HTTP2_CONN_WINDOW_SIZE=${RUSTFS_INTERNODE_RPC_HTTP2_CONN_WINDOW_SIZE:-2097152} + - RUSTFS_INTERNODE_RPC_MAX_MESSAGE_SIZE=${RUSTFS_INTERNODE_RPC_MAX_MESSAGE_SIZE:-104857600} + - RUSTFS_INTERNODE_CHANNEL_ISOLATION=${RUSTFS_INTERNODE_CHANNEL_ISOLATION:-false} + - RUSTFS_INTERNODE_BULK_CHANNELS=${RUSTFS_INTERNODE_BULK_CHANNELS:-2} + - RUSTFS_INTERNODE_RPC_MSGPACK_ONLY=${RUSTFS_INTERNODE_RPC_MSGPACK_ONLY:-false} + - RUSTFS_INTERNODE_PREWARM=${RUSTFS_INTERNODE_PREWARM:-false} + - RUSTFS_INTERNODE_OFFLINE_BYPASS=${RUSTFS_INTERNODE_OFFLINE_BYPASS:-false} + - RUSTFS_INTERNODE_OFFLINE_REPROBE_SECS=${RUSTFS_INTERNODE_OFFLINE_REPROBE_SECS:-5} + - RUSTFS_INTERNODE_OFFLINE_FAILURE_THRESHOLD=${RUSTFS_INTERNODE_OFFLINE_FAILURE_THRESHOLD:-3} + - RUSTFS_INTERNODE_IDEMPOTENT_READ_RETRIES=${RUSTFS_INTERNODE_IDEMPOTENT_READ_RETRIES:-0} extra_hosts: - "host.docker.internal:host-gateway" volumes: diff --git a/crates/config/src/constants/internode.rs b/crates/config/src/constants/internode.rs index 14ca76c79..8db3e0bb1 100644 --- a/crates/config/src/constants/internode.rs +++ b/crates/config/src/constants/internode.rs @@ -39,6 +39,150 @@ pub const DEFAULT_INTERNODE_HTTP2_KEEPALIVE_TIMEOUT_SECS: u64 = 20; pub const ENV_INTERNODE_RPC_TIMEOUT_SECS: &str = "RUSTFS_INTERNODE_RPC_TIMEOUT_SECS"; pub const DEFAULT_INTERNODE_RPC_TIMEOUT_SECS: u64 = 30; +// ── Client-side internode gRPC channel tuning (P0) ── +// These mirror the server-side HTTP/2 transport tuning in `rustfs/src/server/http.rs` +// on the *client* `tonic` `Endpoint` used for internode control-plane RPCs. Prior to +// this, the client channel only set timeouts/keepalive, leaving Nagle enabled and the +// default 64KiB HTTP/2 window in place — hurting small lock-RPC latency and large +// metadata-response throughput respectively. + +/// Disable Nagle's algorithm on internode gRPC client sockets. +/// +/// Latency-sensitive control-plane RPCs (locks, health, small metadata) send tiny +/// frames; Nagle batching adds avoidable delay. Defaults to `true` (nodelay on), +/// matching the server socket configuration. +pub const ENV_INTERNODE_RPC_TCP_NODELAY: &str = "RUSTFS_INTERNODE_RPC_TCP_NODELAY"; +pub const DEFAULT_INTERNODE_RPC_TCP_NODELAY: bool = true; + +// Compile-time invariant: nodelay defaults on so latency-sensitive control-plane RPCs are not +// batched by Nagle, matching the server socket configuration. +const _: () = assert!(DEFAULT_INTERNODE_RPC_TCP_NODELAY); + +/// HTTP/2 initial stream window size (bytes) for internode gRPC client channels. +/// +/// The library default (64KiB) throttles larger unary responses (e.g. `ReadMultiple`, +/// `BatchReadVersion`) by the bandwidth-delay product. Set to 0 to fall back to the +/// `tonic`/`hyper` default. Defaults to 1MiB. +pub const ENV_INTERNODE_RPC_HTTP2_STREAM_WINDOW_SIZE: &str = "RUSTFS_INTERNODE_RPC_HTTP2_STREAM_WINDOW_SIZE"; +pub const DEFAULT_INTERNODE_RPC_HTTP2_STREAM_WINDOW_SIZE: u32 = 1024 * 1024; + +/// HTTP/2 initial connection window size (bytes) for internode gRPC client channels. +/// +/// Should be >= the stream window so multiple concurrent streams are not starved by the +/// connection-level flow-control window. Set to 0 to fall back to the library default. +/// Defaults to 2MiB. +pub const ENV_INTERNODE_RPC_HTTP2_CONN_WINDOW_SIZE: &str = "RUSTFS_INTERNODE_RPC_HTTP2_CONN_WINDOW_SIZE"; +pub const DEFAULT_INTERNODE_RPC_HTTP2_CONN_WINDOW_SIZE: u32 = 2 * 1024 * 1024; + +// Compile-time invariant: the connection-level window must be >= the per-stream window, or +// concurrent streams would be starved by the connection-level flow-control budget. +const _: () = assert!(DEFAULT_INTERNODE_RPC_HTTP2_CONN_WINDOW_SIZE >= DEFAULT_INTERNODE_RPC_HTTP2_STREAM_WINDOW_SIZE); + +/// Maximum encoded message size (bytes) for internode gRPC, applied to both the client +/// `NodeServiceClient` and the server `NodeServiceServer`. +/// +/// Without this, `tonic`'s default 4MiB decode limit silently caps `bytes`-carrying +/// unary RPCs (`ReadAll`/`WriteAll`/`ReadMultiple`/`BatchReadVersion`); a large multi-version +/// `xl.meta` or an aggregated response then fails with `out_of_range`. The default comes +/// from `rustfs_protos::DEFAULT_GRPC_SERVER_MESSAGE_LEN` (100MiB) at the call sites. +pub const ENV_INTERNODE_RPC_MAX_MESSAGE_SIZE: &str = "RUSTFS_INTERNODE_RPC_MAX_MESSAGE_SIZE"; + +/// Payload-size threshold (bytes) above which a unary internode gRPC response counts as a +/// "large payload" for alerting. +/// +/// Large `ReadAll`/`ReadMultiple` responses share the control-plane channel with +/// latency-sensitive lock/health RPCs and can head-of-line block them (see grpc-optimization +/// G2). This threshold drives the `rustfs_system_network_internode_operation_large_payloads_total` +/// counter so operators can size which paths need channel isolation in P1. Defaults to 8MiB. +pub const ENV_INTERNODE_RPC_LARGE_PAYLOAD_WARN_BYTES: &str = "RUSTFS_INTERNODE_RPC_LARGE_PAYLOAD_WARN_BYTES"; +pub const DEFAULT_INTERNODE_RPC_LARGE_PAYLOAD_WARN_BYTES: usize = 8 * 1024 * 1024; + +/// Stop dual-writing the JSON compatibility strings on internode metadata RPCs and send only the +/// msgpack `_bin` payloads (grpc-optimization P2-1). +/// +/// Defaults to `false` (dual-write, byte-for-byte legacy behavior). This is a rollout lever, not a +/// wire-format change: it may only be enabled **after** the JSON-fallback counter +/// (`rustfs_system_network_internode_msgpack_json_fallback_total`) has read zero across a release +/// window fleet-wide, confirming every peer decodes `_bin` first. Single-env rollback. See +/// `docs/operations/internode-msgpack-json-convergence-runbook.md`. +pub const ENV_INTERNODE_RPC_MSGPACK_ONLY: &str = "RUSTFS_INTERNODE_RPC_MSGPACK_ONLY"; +pub const DEFAULT_INTERNODE_RPC_MSGPACK_ONLY: bool = false; + +// Compile-time invariant: dual-write by default so the base build is byte-for-byte legacy behavior. +const _: () = assert!(!DEFAULT_INTERNODE_RPC_MSGPACK_ONLY); + +/// Consecutive-failure threshold after which an internode peer is marked offline (grpc-optimization +/// P3 observability). +/// +/// A peer accrues a failure on each dial failure or RPC-triggered connection eviction, and flips +/// back online on the next successful dial. Drives the `rustfs_cluster_servers_offline_total` gauge +/// (parity with MinIO's `minio_cluster_servers_offline_total`). Clamped to at least 1. Defaults to 3. +pub const ENV_INTERNODE_OFFLINE_FAILURE_THRESHOLD: &str = "RUSTFS_INTERNODE_OFFLINE_FAILURE_THRESHOLD"; +pub const DEFAULT_INTERNODE_OFFLINE_FAILURE_THRESHOLD: u32 = 3; + +/// Prewarm internode control channels in the background at remote-disk construction, moving the +/// connect cost off the first RPC (grpc-optimization P3-1). +/// +/// Best-effort and deduped per peer; failures fall through to the existing lazy connect + recovery +/// monitor. Defaults to `false` (opt-in): enabling it dials every peer at startup, and it is not yet +/// validated against a cold-start baseline. +pub const ENV_INTERNODE_PREWARM: &str = "RUSTFS_INTERNODE_PREWARM"; +pub const DEFAULT_INTERNODE_PREWARM: bool = false; + +/// Fast-fail (bypass) internode RPCs to a peer already marked offline, instead of paying the connect +/// timeout, letting quorum proceed sooner (grpc-optimization P3-2). +/// +/// Defaults to `false` (opt-in): this touches peer routing (consistency-sensitive) and must be +/// validated with the failover bench before rollout. It does NOT change quorum. The bypass is +/// self-healing — one request per [`ENV_INTERNODE_OFFLINE_REPROBE_SECS`] is let through to recover a +/// peer even without a background monitor. Single-env rollback. +pub const ENV_INTERNODE_OFFLINE_BYPASS: &str = "RUSTFS_INTERNODE_OFFLINE_BYPASS"; +pub const DEFAULT_INTERNODE_OFFLINE_BYPASS: bool = false; + +/// Re-probe interval (seconds) for the offline bypass: while a peer is offline, one request is let +/// through this often to attempt recovery. Clamped to a small minimum by callers. Defaults to 5s. +pub const ENV_INTERNODE_OFFLINE_REPROBE_SECS: &str = "RUSTFS_INTERNODE_OFFLINE_REPROBE_SECS"; +pub const DEFAULT_INTERNODE_OFFLINE_REPROBE_SECS: u64 = 5; + +// Compile-time invariant: prewarm and offline-bypass are opt-in so the base build is unchanged. +const _: () = assert!(!DEFAULT_INTERNODE_PREWARM); +const _: () = assert!(!DEFAULT_INTERNODE_OFFLINE_BYPASS); + +/// Extra attempts for idempotent, read-only/reentrant control-plane RPCs (e.g. `DiskInfo`) on +/// transient network failures, with exponential backoff (grpc-optimization P3-3). +/// +/// Defaults to `0` (disabled) — retries change latency-on-failure behavior and are opt-in. **Only** +/// idempotent reads use this; write/lock RPCs (`WriteAll`/`RenameData`/`Delete*`/`Lock`/`UnLock`) +/// must never auto-retry, to preserve quorum and idempotency semantics (see `CLAUDE.md`). +pub const ENV_INTERNODE_IDEMPOTENT_READ_RETRIES: &str = "RUSTFS_INTERNODE_IDEMPOTENT_READ_RETRIES"; +pub const DEFAULT_INTERNODE_IDEMPOTENT_READ_RETRIES: usize = 0; + +// ── Control/bulk channel isolation (P1) ── +// Large `bytes`-carrying unary RPCs (ReadAll/WriteAll/ReadMultiple/BatchReadVersion) otherwise +// share the control-plane HTTP/2 connection with latency-sensitive lock/health RPCs; a large +// transfer can head-of-line block a lock RPC (G2/G5). When isolation is enabled these bulk RPCs +// are routed onto a separate per-peer channel pool. See grpc-optimization P1. + +/// Enable control/bulk internode gRPC channel isolation. +/// +/// Defaults to `false`: this touches lock-RPC transport routing (consistency-sensitive), so it +/// is opt-in and validated against a baseline before rollout. When `false`, bulk RPCs reuse the +/// control channel exactly as before, so the switch is a single-env rollback. +pub const ENV_INTERNODE_CHANNEL_ISOLATION: &str = "RUSTFS_INTERNODE_CHANNEL_ISOLATION"; +pub const DEFAULT_INTERNODE_CHANNEL_ISOLATION: bool = false; + +// Compile-time invariant: isolation is opt-in so the default build behaves exactly as before P1. +const _: () = assert!(!DEFAULT_INTERNODE_CHANNEL_ISOLATION); + +/// Number of bulk channels maintained per peer when channel isolation is enabled. +/// +/// A tonic `Channel` is a single TCP/HTTP2 connection; multiple bulk channels are round-robined +/// to relieve the single-connection throughput ceiling for large transfers. Kept small (default +/// 2) to avoid a connection storm; clamped to at least 1. Set to 1 to isolate bulk onto a single +/// dedicated connection (still separate from control). +pub const ENV_INTERNODE_BULK_CHANNELS: &str = "RUSTFS_INTERNODE_BULK_CHANNELS"; +pub const DEFAULT_INTERNODE_BULK_CHANNELS: usize = 2; + /// Profile selector for conservative internode HTTP data-plane client tuning. pub const ENV_INTERNODE_HTTP_TUNING_PROFILE: &str = "RUSTFS_INTERNODE_HTTP_TUNING_PROFILE"; pub const DEFAULT_INTERNODE_HTTP_TUNING_PROFILE: &str = "legacy"; @@ -85,6 +229,61 @@ mod tests { assert_eq!(DEFAULT_INTERNODE_HTTP_TUNING_PROFILE, "legacy"); } + #[test] + fn internode_rpc_channel_tuning_defaults() { + assert_eq!(DEFAULT_INTERNODE_RPC_HTTP2_STREAM_WINDOW_SIZE, 1024 * 1024); + assert_eq!(DEFAULT_INTERNODE_RPC_HTTP2_CONN_WINDOW_SIZE, 2 * 1024 * 1024); + // The nodelay-on and connection-window >= stream-window invariants are enforced at + // compile time next to the constant definitions. + } + + #[test] + fn internode_rpc_channel_tuning_env_names_are_stable() { + assert_eq!(ENV_INTERNODE_RPC_TCP_NODELAY, "RUSTFS_INTERNODE_RPC_TCP_NODELAY"); + assert_eq!( + ENV_INTERNODE_RPC_HTTP2_STREAM_WINDOW_SIZE, + "RUSTFS_INTERNODE_RPC_HTTP2_STREAM_WINDOW_SIZE" + ); + assert_eq!(ENV_INTERNODE_RPC_HTTP2_CONN_WINDOW_SIZE, "RUSTFS_INTERNODE_RPC_HTTP2_CONN_WINDOW_SIZE"); + assert_eq!(ENV_INTERNODE_RPC_MAX_MESSAGE_SIZE, "RUSTFS_INTERNODE_RPC_MAX_MESSAGE_SIZE"); + assert_eq!( + ENV_INTERNODE_RPC_LARGE_PAYLOAD_WARN_BYTES, + "RUSTFS_INTERNODE_RPC_LARGE_PAYLOAD_WARN_BYTES" + ); + assert_eq!(DEFAULT_INTERNODE_RPC_LARGE_PAYLOAD_WARN_BYTES, 8 * 1024 * 1024); + } + + #[test] + fn internode_channel_isolation_defaults_and_env_names() { + // The isolation-off default is asserted at compile time next to the constant definition. + assert_eq!(DEFAULT_INTERNODE_BULK_CHANNELS, 2); + assert_eq!(ENV_INTERNODE_CHANNEL_ISOLATION, "RUSTFS_INTERNODE_CHANNEL_ISOLATION"); + assert_eq!(ENV_INTERNODE_BULK_CHANNELS, "RUSTFS_INTERNODE_BULK_CHANNELS"); + } + + #[test] + fn internode_msgpack_only_env_name_is_stable() { + // The dual-write-by-default invariant is asserted at compile time next to the definition. + assert_eq!(ENV_INTERNODE_RPC_MSGPACK_ONLY, "RUSTFS_INTERNODE_RPC_MSGPACK_ONLY"); + } + + #[test] + fn internode_offline_failure_threshold_defaults_and_env_name() { + assert_eq!(DEFAULT_INTERNODE_OFFLINE_FAILURE_THRESHOLD, 3); + assert_eq!(ENV_INTERNODE_OFFLINE_FAILURE_THRESHOLD, "RUSTFS_INTERNODE_OFFLINE_FAILURE_THRESHOLD"); + } + + #[test] + fn internode_p3_lifecycle_defaults_are_opt_in() { + // The opt-in (default-off) invariants are asserted at compile time next to the definitions. + assert_eq!(DEFAULT_INTERNODE_OFFLINE_REPROBE_SECS, 5); + assert_eq!(DEFAULT_INTERNODE_IDEMPOTENT_READ_RETRIES, 0); + assert_eq!(ENV_INTERNODE_PREWARM, "RUSTFS_INTERNODE_PREWARM"); + assert_eq!(ENV_INTERNODE_OFFLINE_BYPASS, "RUSTFS_INTERNODE_OFFLINE_BYPASS"); + assert_eq!(ENV_INTERNODE_OFFLINE_REPROBE_SECS, "RUSTFS_INTERNODE_OFFLINE_REPROBE_SECS"); + assert_eq!(ENV_INTERNODE_IDEMPOTENT_READ_RETRIES, "RUSTFS_INTERNODE_IDEMPOTENT_READ_RETRIES"); + } + #[test] fn internode_timeout_env_names_are_stable() { assert_eq!(ENV_INTERNODE_CONNECT_TIMEOUT_SECS, "RUSTFS_INTERNODE_CONNECT_TIMEOUT_SECS"); diff --git a/crates/ecstore/src/cluster/rpc/client.rs b/crates/ecstore/src/cluster/rpc/client.rs index 2ffe49579..81cdb0360 100644 --- a/crates/ecstore/src/cluster/rpc/client.rs +++ b/crates/ecstore/src/cluster/rpc/client.rs @@ -16,7 +16,9 @@ use crate::cluster::rpc::{TONIC_RPC_PREFIX, gen_signature_headers}; use crate::disk::error::{DiskError, Error as DiskErrorType}; use crate::runtime::sources as runtime_sources; use http::Method; -use rustfs_protos::{create_new_channel, proto_gen::node_service::node_service_client::NodeServiceClient}; +use rustfs_protos::{ + ChannelClass, create_new_channel, get_channel_for_class, proto_gen::node_service::node_service_client::NodeServiceClient, +}; use std::{error::Error, io::ErrorKind}; use tonic::{service::interceptor::InterceptedService, transport::Channel}; use tracing::debug; @@ -29,21 +31,38 @@ pub async fn node_service_time_out_client( addr: &String, interceptor: TonicInterceptor, ) -> Result>, Box> { - // Try to get cached channel - let cached_channel = runtime_sources::cached_node_channel(addr).await; + // Default to the latency-sensitive control channel; bulk `bytes` RPCs opt in via the + // `_for_class` variant below (grpc-optimization P1). + node_service_time_out_client_for_class(addr, interceptor, ChannelClass::Control).await +} - let channel = match cached_channel { - Some(channel) => { - debug!("Using cached gRPC channel for: {}", addr); - channel - } - None => { - // No cached connection, create new one - create_new_channel(addr).await? - } +/// Build a `NodeServiceClient` bound to the [`ChannelClass`]-appropriate channel for `addr`. +/// +/// Bulk `bytes`-carrying RPCs (ReadAll/WriteAll/ReadMultiple/BatchReadVersion) pass +/// [`ChannelClass::Bulk`] so, when channel isolation is enabled, they are physically isolated +/// from lock/health RPCs; everything else uses [`ChannelClass::Control`]. When isolation is +/// disabled the two classes resolve to the same cached channel, i.e. legacy behavior. +pub async fn node_service_time_out_client_for_class( + addr: &String, + interceptor: TonicInterceptor, + class: ChannelClass, +) -> Result>, Box> { + let channel = match class { + ChannelClass::Control => match runtime_sources::cached_node_channel(addr).await { + Some(channel) => { + debug!("Using cached gRPC channel for: {}", addr); + channel + } + // No cached connection, create new one. + None => create_new_channel(addr).await?, + }, + ChannelClass::Bulk => get_channel_for_class(addr, ChannelClass::Bulk).await?, }; - Ok(NodeServiceClient::with_interceptor(channel, interceptor)) + let max_message_size = rustfs_protos::internode_rpc_max_message_size(); + Ok(NodeServiceClient::with_interceptor(channel, interceptor) + .max_decoding_message_size(max_message_size) + .max_encoding_message_size(max_message_size)) } pub async fn node_service_time_out_client_no_auth( diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index 9d8d26526..ec86440d3 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -14,6 +14,7 @@ use crate::cluster::rpc::client::{ TonicInterceptor, gen_tonic_signature_interceptor, is_network_like_disk_error, node_service_time_out_client, + node_service_time_out_client_for_class, node_service_time_out_client_no_auth, }; use crate::cluster::rpc::internode_data_transport::{ InternodeDataTransport, ReadStreamRequest, WalkDirStreamRequest, WriteStreamRequest, @@ -39,6 +40,7 @@ use bytes::Bytes; use futures::lock::Mutex; use metrics::counter; use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo}; +use rustfs_protos::ChannelClass; use rustfs_protos::evict_failed_connection; use rustfs_protos::proto_gen::node_service::RenamePartRequest; use rustfs_protos::proto_gen::node_service::{ @@ -77,6 +79,8 @@ enum FailureHealthAction { const REMOTE_DISK_OPEN_WRITE_MAX_ATTEMPTS: usize = 2; const REMOTE_DISK_OPEN_WRITE_RETRY_BACKOFF: Duration = Duration::from_millis(20); +/// Base backoff for idempotent read-only RPC retries (grpc-optimization P3-3); doubles per attempt. +const REMOTE_DISK_READ_RETRY_BASE_BACKOFF: Duration = Duration::from_millis(50); const ENV_RUSTFS_METADATA_BATCH_READ: &str = "RUSTFS_METADATA_BATCH_READ"; const LEGACY_ENV_RUSTFS_BATCH_METADATA_RPC: &str = "RUSTFS_BATCH_METADATA_RPC"; const BATCH_METADATA_RPC_OFF: &str = "off"; @@ -184,6 +188,78 @@ pub struct RemoteDisk { data_transport: Arc, } +// ── Connection lifecycle (grpc-optimization P3) ── + +/// Whether to prewarm the internode control channel in the background at construction (default off). +fn internode_prewarm_enabled() -> bool { + rustfs_utils::get_env_bool(rustfs_config::ENV_INTERNODE_PREWARM, rustfs_config::DEFAULT_INTERNODE_PREWARM) +} + +/// Whether to fast-fail RPCs to peers already marked offline (default off). +fn internode_offline_bypass_enabled() -> bool { + rustfs_utils::get_env_bool( + rustfs_config::ENV_INTERNODE_OFFLINE_BYPASS, + rustfs_config::DEFAULT_INTERNODE_OFFLINE_BYPASS, + ) +} + +/// Re-probe interval for the offline bypass (>= 1s). +fn internode_offline_reprobe_interval() -> Duration { + Duration::from_secs( + rustfs_utils::get_env_u64( + rustfs_config::ENV_INTERNODE_OFFLINE_REPROBE_SECS, + rustfs_config::DEFAULT_INTERNODE_OFFLINE_REPROBE_SECS, + ) + .max(1), + ) +} + +/// If the offline bypass is enabled and `addr` is marked offline, return a reason string to +/// fast-fail with instead of paying the connect timeout (grpc-optimization P3-2). Self-healing: +/// one request per re-probe interval is let through so the peer can recover. Shared by the data +/// path (`remote_disk`) and the lock path (`remote_locker`). +pub(crate) fn internode_offline_bypass_reason(addr: &str) -> Option { + if internode_offline_bypass_enabled() + && rustfs_io_metrics::internode_metrics::cluster_peer_should_bypass(addr, internode_offline_reprobe_interval()) + { + return Some(format!("internode peer {addr} offline; fast-fail bypass (P3)")); + } + None +} + +/// Number of extra attempts for idempotent read-only control-plane RPCs on transient network +/// failures (grpc-optimization P3-3). `0` (default) disables retries. Write/lock RPCs never retry. +fn internode_idempotent_read_retries() -> usize { + rustfs_utils::get_env_usize( + rustfs_config::ENV_INTERNODE_IDEMPOTENT_READ_RETRIES, + rustfs_config::DEFAULT_INTERNODE_IDEMPOTENT_READ_RETRIES, + ) +} + +/// Peers for which a control-channel prewarm has already been triggered, to dedup the N remote +/// disks that map to a single peer address. +static PREWARMED_PEERS: std::sync::LazyLock>> = + std::sync::LazyLock::new(|| std::sync::Mutex::new(std::collections::HashSet::new())); + +/// Best-effort background prewarm of a peer's control channel (grpc-optimization P3-1). Deduped per +/// peer; a dial failure just falls through to the existing lazy connect and recovery monitor. +fn spawn_control_channel_prewarm(addr: String) { + { + let Ok(mut prewarmed) = PREWARMED_PEERS.lock() else { + return; + }; + if !prewarmed.insert(addr.clone()) { + return; + } + } + tokio::spawn(async move { + match node_service_time_out_client_no_auth(&addr).await { + Ok(_) => debug!(addr = %addr, "internode control channel prewarmed"), + Err(err) => debug!(addr = %addr, error = %err, "internode control channel prewarm failed (best-effort)"), + } + }); +} + impl RemoteDisk { fn recovery_monitor_span(addr: &str, endpoint: &Endpoint) -> tracing::Span { tracing::info_span!( @@ -231,6 +307,12 @@ impl RemoteDisk { }; record_drive_runtime_state(ep, RuntimeDriveHealthState::Online); + // P3-1: move the connect cost off the first RPC by prewarming the control channel in the + // background. Deduped per peer, best-effort, opt-in. + if internode_prewarm_enabled() { + spawn_control_channel_prewarm(disk.addr.clone()); + } + Ok(disk) } @@ -613,6 +695,49 @@ impl RemoteDisk { self.execute_with_timeout_for_op("unknown", operation, timeout_duration).await } + /// Execute an **idempotent, read-only/reentrant** RPC with a bounded number of retries on + /// transient network failures, with exponential backoff (grpc-optimization P3-3). Retries + /// default to 0 (disabled). MUST NOT be used for write/lock RPCs — those must never auto-retry + /// (quorum/idempotency safety). The `operation` closure is re-invoked per attempt, so it must be + /// `Fn` (rebuild the request from borrowed inputs, do not move captured state out). + async fn execute_read_with_retry(&self, op: &'static str, operation: F, timeout_duration: Duration) -> Result + where + F: Fn() -> Fut, + Fut: std::future::Future>, + { + let max_retries = internode_idempotent_read_retries(); + let mut attempt = 0usize; + loop { + // Only the final attempt marks the disk faulty / evicts the channel. Earlier retries + // ignore the failure, so a transient error cannot flip the disk into a faulty + // short-circuit (which would defeat the retry) or over-count failures. + let health_action = if attempt >= max_retries { + FailureHealthAction::MarkFailure + } else { + FailureHealthAction::IgnoreFailure + }; + match self + .execute_with_timeout_for_op_and_health_action(op, &operation, timeout_duration, health_action) + .await + { + Err(err) if attempt < max_retries && is_network_like_disk_error(&err) => { + attempt += 1; + let backoff = REMOTE_DISK_READ_RETRY_BASE_BACKOFF + .saturating_mul(1u32 << u32::try_from(attempt - 1).unwrap_or(4).min(4)); + debug!( + endpoint = %self.endpoint, + addr = %self.addr, + op, + attempt, + "retrying idempotent read-only RPC after transient network error" + ); + tokio::time::sleep(backoff).await; + } + other => return other, + } + } + } + async fn execute_with_timeout_for_op( &self, op: &'static str, @@ -804,12 +929,41 @@ impl RemoteDisk { } } + /// P3-2 offline bypass: when enabled and this peer is marked offline, fast-fail instead of + /// paying the connect timeout, so the erasure layer proceeds on quorum sooner. This does not + /// change quorum. Self-healing — one request per re-probe interval is let through so the peer + /// recovers even without a background monitor. The recovery monitor's own probe path calls the + /// client directly and is unaffected. + fn offline_bypass_error(&self) -> Option { + internode_offline_bypass_reason(&self.addr).map(Error::other) + } + async fn get_client(&self) -> Result>> { + if let Some(err) = self.offline_bypass_error() { + return Err(err); + } node_service_time_out_client(&self.addr, TonicInterceptor::Signature(gen_tonic_signature_interceptor())) .await .map_err(|err| Error::other(format!("can not get client, err: {err}"))) } + /// Client for large `bytes`-carrying RPCs (ReadAll/WriteAll/ReadMultiple/BatchReadVersion). + /// Routes onto the isolated bulk channel pool so large transfers cannot head-of-line block + /// lock/health RPCs (grpc-optimization P1). Falls back to the control channel when isolation + /// is disabled. + async fn get_bulk_client(&self) -> Result>> { + if let Some(err) = self.offline_bypass_error() { + return Err(err); + } + node_service_time_out_client_for_class( + &self.addr, + TonicInterceptor::Signature(gen_tonic_signature_interceptor()), + ChannelClass::Bulk, + ) + .await + .map_err(|err| Error::other(format!("can not get client, err: {err}"))) + } + async fn disk_ref(&self) -> String { (*self.id.lock().await) .map(|id| id.to_string()) @@ -817,27 +971,56 @@ impl RemoteDisk { } } +/// Initial capacity hint (bytes) for msgpack encode buffers, sized to cover a typical single- +/// version `FileInfo` without repeated growth reallocations. Larger payloads still grow as needed. +const MSGPACK_ENCODE_CAPACITY_HINT: usize = 512; + fn encode_msgpack(value: &T) -> Result> { - let mut serializer = rmp_serde::Serializer::new(Vec::new()); + let mut serializer = rmp_serde::Serializer::new(Vec::with_capacity(MSGPACK_ENCODE_CAPACITY_HINT)); value.serialize(&mut serializer)?; Ok(serializer.into_inner()) } +/// JSON compatibility string for a dual-encoded (`_bin` + text) request field. Returns an empty +/// string when msgpack-only mode is enabled (grpc-optimization P2-1) so the redundant JSON copy is +/// not sent; otherwise the legacy JSON encoding. Only use for fields whose peer decodes `_bin` +/// first — the paired `_bin` (msgpack) field must always be sent alongside. +fn compat_json(value: &T) -> Result { + if rustfs_protos::internode_rpc_msgpack_only() { + return Ok(String::new()); + } + Ok(serde_json::to_string(value)?) +} + fn encode_msgpack_named(value: &T) -> Result> { - let mut serializer = rmp_serde::Serializer::new(Vec::new()).with_struct_map(); + let mut serializer = rmp_serde::Serializer::new(Vec::with_capacity(MSGPACK_ENCODE_CAPACITY_HINT)).with_struct_map(); value.serialize(&mut serializer)?; Ok(serializer.into_inner()) } -fn decode_msgpack_or_json(binary: &[u8], json: &str) -> Result { +fn decode_msgpack_or_json(binary: &[u8], json: &str, value_name: &'static str) -> Result { if !binary.is_empty() { let mut deserializer = rmp_serde::Deserializer::new(Cursor::new(binary)); return T::deserialize(&mut deserializer).map_err(Error::from); } + // The msgpack payload was absent, so fall back to the JSON compatibility field. This branch + // must read zero across a release window before the redundant JSON fields can be dropped (P2). + crate::cluster::rpc::runtime_sources::record_response_json_fallback(value_name); serde_json::from_str(json).map_err(Error::from) } +/// Aggregate encoded size (bytes) of a `ReadMultiple` response, preferring the msgpack payloads +/// and falling back to the JSON compatibility strings. Used to size the RPC for the payload +/// histogram / large-payload alerting (grpc-optimization P0 instrumentation). +fn read_multiple_response_payload_len(response: &ReadMultipleResponse) -> usize { + if !response.read_multiple_resps_bin.is_empty() { + response.read_multiple_resps_bin.iter().map(|buf| buf.len()).sum() + } else { + response.read_multiple_resps.iter().map(|item| item.len()).sum() + } +} + fn decode_read_multiple_response_items(response: ReadMultipleResponse, endpoint: &Endpoint) -> Result> { if !response.read_multiple_resps_bin.is_empty() { if !response.read_multiple_resps.is_empty() @@ -858,7 +1041,7 @@ fn decode_read_multiple_response_items(response: ReadMultipleResponse, endpoint: let mut read_multiple_resps = Vec::with_capacity(response.read_multiple_resps_bin.len()); for (index, buf) in response.read_multiple_resps_bin.iter().enumerate() { - let resp = decode_msgpack_or_json::(buf, "").map_err(|err| { + let resp = decode_msgpack_or_json::(buf, "", "ReadMultipleResp").map_err(|err| { Error::other(format!("decode ReadMultipleResp msgpack item {index} from {endpoint} failed: {err}")) })?; read_multiple_resps.push(resp); @@ -866,6 +1049,10 @@ fn decode_read_multiple_response_items(response: ReadMultipleResponse, endpoint: return Ok(read_multiple_resps); } + // No msgpack payloads present: the whole list fell back to the JSON compatibility field (P2). + if !response.read_multiple_resps.is_empty() { + crate::cluster::rpc::runtime_sources::record_response_json_fallback("ReadMultipleResp"); + } let mut read_multiple_resps = Vec::with_capacity(response.read_multiple_resps.len()); for (index, json_str) in response.read_multiple_resps.iter().enumerate() { let resp = serde_json::from_str::(json_str) @@ -899,7 +1086,7 @@ fn decode_batch_read_version_response_items( let mut batch_read_version_resps = Vec::with_capacity(response.batch_read_version_resps_bin.len()); for (index, buf) in response.batch_read_version_resps_bin.iter().enumerate() { - let resp = decode_msgpack_or_json::(buf, "").map_err(|err| { + let resp = decode_msgpack_or_json::(buf, "", "BatchReadVersionResp").map_err(|err| { Error::other(format!("decode BatchReadVersionResp msgpack item {index} from {endpoint} failed: {err}")) })?; batch_read_version_resps.push(resp); @@ -907,6 +1094,10 @@ fn decode_batch_read_version_response_items( return Ok(batch_read_version_resps); } + // No msgpack payloads present: the whole list fell back to the JSON compatibility field (P2). + if !response.batch_read_version_resps.is_empty() { + crate::cluster::rpc::runtime_sources::record_response_json_fallback("BatchReadVersionResp"); + } let mut batch_read_version_resps = Vec::with_capacity(response.batch_read_version_resps.len()); for (index, json_str) in response.batch_read_version_resps.iter().enumerate() { let resp = serde_json::from_str::(json_str).map_err(|err| { @@ -1207,6 +1398,10 @@ impl DiskAPI for RemoteDisk { self.execute_with_timeout( || async { + // `_bin` support for DeleteVersion is new (grpc-optimization P2); always dual-write + // JSON + msgpack until its fallback counter has read zero across a release window. + let file_info_bin = encode_msgpack(&fi)?; + let opts_bin = encode_msgpack(&opts)?; let file_info = serde_json::to_string(&fi)?; let opts = serde_json::to_string(&opts)?; @@ -1221,6 +1416,8 @@ impl DiskAPI for RemoteDisk { file_info, force_del_marker, opts, + file_info_bin: file_info_bin.into(), + opts_bin: opts_bin.into(), }); let response = client.delete_version(request).await?.into_inner(); @@ -1256,6 +1453,18 @@ impl DiskAPI for RemoteDisk { return vec![Some(DiskError::FaultyDisk); versions.len()]; } + // `_bin` support for DeleteVersions is new (grpc-optimization P2); always dual-write JSON + + // msgpack until its fallback counter has read zero across a release window. + let opts_bin = match encode_msgpack(&opts) { + Ok(opts_bin) => opts_bin, + Err(err) => { + let mut errors = Vec::with_capacity(versions.len()); + for _ in 0..versions.len() { + errors.push(Some(Error::other(err.to_string()))); + } + return errors; + } + }; let opts = match serde_json::to_string(&opts) { Ok(opts) => opts, Err(err) => { @@ -1267,6 +1476,7 @@ impl DiskAPI for RemoteDisk { } }; let mut versions_str = Vec::with_capacity(versions.len()); + let mut versions_bin = Vec::with_capacity(versions.len()); for file_info_versions in versions.iter() { versions_str.push(match serde_json::to_string(file_info_versions) { Ok(versions_str) => versions_str, @@ -1278,6 +1488,16 @@ impl DiskAPI for RemoteDisk { return errors; } }); + versions_bin.push(match encode_msgpack(file_info_versions) { + Ok(versions_bin) => Bytes::from(versions_bin), + Err(err) => { + let mut errors = Vec::with_capacity(versions.len()); + for _ in 0..versions.len() { + errors.push(Some(Error::other(err.to_string()))); + } + return errors; + } + }); } let mut client = match self.get_client().await { Ok(client) => client, @@ -1295,6 +1515,8 @@ impl DiskAPI for RemoteDisk { volume: volume.to_string(), versions: versions_str, opts, + versions_bin, + opts_bin: opts_bin.into(), }); // TODO: use Error not string @@ -1396,7 +1618,7 @@ impl DiskAPI for RemoteDisk { state = "started", "Remote disk RPC started" ); - let file_info = serde_json::to_string(&fi)?; + let file_info = compat_json(&fi)?; let file_info_bin = encode_msgpack(&fi)?; self.execute_with_timeout_for_op( @@ -1469,8 +1691,8 @@ impl DiskAPI for RemoteDisk { state = "started", "Remote disk RPC started" ); - let file_info = serde_json::to_string(&fi)?; - let opts_str = serde_json::to_string(&opts)?; + let file_info = compat_json(&fi)?; + let opts_str = compat_json(&opts)?; let file_info_bin = encode_msgpack(&fi)?; let opts_bin = encode_msgpack(opts)?; @@ -1526,7 +1748,7 @@ impl DiskAPI for RemoteDisk { state = "started", "Remote disk RPC started" ); - let opts_str = serde_json::to_string(opts)?; + let opts_str = compat_json(opts)?; let opts_bin = encode_msgpack(opts)?; self.execute_with_timeout( @@ -1551,7 +1773,7 @@ impl DiskAPI for RemoteDisk { return Err(response.error.unwrap_or_default().into()); } - let file_info = decode_msgpack_or_json::(&response.file_info_bin, &response.file_info)?; + let file_info = decode_msgpack_or_json::(&response.file_info_bin, &response.file_info, "FileInfo")?; Ok(file_info) }, @@ -1582,7 +1804,7 @@ impl DiskAPI for RemoteDisk { state = "started", "Remote disk RPC started" ); - let batch_read_version_req = serde_json::to_string(&req)?; + let batch_read_version_req = compat_json(&req)?; let batch_read_version_req_bin = encode_msgpack(&req)?; let batch_result = self @@ -1591,7 +1813,7 @@ impl DiskAPI for RemoteDisk { move || async move { let disk = self.disk_ref().await; let mut client = self - .get_client() + .get_bulk_client() .await .map_err(|err| Error::other(format!("can not get client, err: {err}")))?; let request = Request::new(BatchReadVersionRequest { @@ -1688,7 +1910,8 @@ impl DiskAPI for RemoteDisk { return Err(response.error.unwrap_or_default().into()); } - let raw_file_info = decode_msgpack_or_json::(&response.raw_file_info_bin, &response.raw_file_info)?; + let raw_file_info = + decode_msgpack_or_json::(&response.raw_file_info_bin, &response.raw_file_info, "RawFileInfo")?; Ok(raw_file_info) }, @@ -1723,7 +1946,7 @@ impl DiskAPI for RemoteDisk { self.execute_with_timeout_for_op( "rename_data", || async { - let file_info = serde_json::to_string(&fi)?; + let file_info = compat_json(&fi)?; let file_info_bin = encode_msgpack_named(&fi)?; let mut client = self .get_client() @@ -1745,8 +1968,11 @@ impl DiskAPI for RemoteDisk { return Err(response.error.unwrap_or_default().into()); } - let rename_data_resp = - decode_msgpack_or_json::(&response.rename_data_resp_bin, &response.rename_data_resp)?; + let rename_data_resp = decode_msgpack_or_json::( + &response.rename_data_resp_bin, + &response.rename_data_resp, + "RenameDataResp", + )?; Ok(rename_data_resp) }, @@ -2241,11 +2467,11 @@ impl DiskAPI for RemoteDisk { self.execute_with_timeout( || async { - let read_multiple_req = serde_json::to_string(&req)?; + let read_multiple_req = compat_json(&req)?; let read_multiple_req_bin = encode_msgpack(&req)?; let disk = self.disk_ref().await; let mut client = self - .get_client() + .get_bulk_client() .await .map_err(|err| Error::other(format!("can not get client, err: {err}")))?; let request = Request::new(ReadMultipleRequest { @@ -2260,6 +2486,10 @@ impl DiskAPI for RemoteDisk { return Err(response.error.unwrap_or_default().into()); } + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_read_multiple_recv_bytes( + read_multiple_response_payload_len(&response), + ); + let read_multiple_resps = decode_read_multiple_response_items(response, &self.endpoint)?; Ok(read_multiple_resps) @@ -2288,7 +2518,7 @@ impl DiskAPI for RemoteDisk { || async { let data_len = data.len(); let disk = self.disk_ref().await; - let mut client = self.get_client().await.map_err(|err| { + let mut client = self.get_bulk_client().await.map_err(|err| { crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_write_all_error(); Error::other(format!("can not get client, err: {err}")) })?; @@ -2339,7 +2569,7 @@ 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| { + let mut client = self.get_bulk_client().await.map_err(|err| { crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_read_all_error(); Error::other(format!("can not get client, err: {err}")) })?; @@ -2373,7 +2603,8 @@ impl DiskAPI for RemoteDisk { #[tracing::instrument(skip(self))] async fn disk_info(&self, opts: &DiskInfoOptions) -> Result { - self.execute_with_timeout_for_op( + // disk_info is idempotent/read-only, so it is eligible for the P3-3 bounded retry. + self.execute_read_with_retry( "disk_info", || async { let opts = serde_json::to_string(&opts)?; @@ -2650,6 +2881,45 @@ mod tests { assert_eq!(decoded[0].data, b"fallback"); } + #[test] + fn read_multiple_response_payload_len_prefers_msgpack_and_falls_back_to_json() { + let bin_a = encode_msgpack(&sample_read_multiple_resp("a", b"binary")).expect("msgpack should encode"); + let bin_b = encode_msgpack(&sample_read_multiple_resp("b", b"more")).expect("msgpack should encode"); + let json = serde_json::to_string(&sample_read_multiple_resp("j", b"fallback")).expect("json should encode"); + + // When msgpack bins are present, the length is their aggregate size (JSON strings ignored). + let with_bin = ReadMultipleResponse { + success: true, + read_multiple_resps: vec![json.clone()], + read_multiple_resps_bin: vec![bin_a.clone().into(), bin_b.clone().into()], + error: None, + }; + assert_eq!(read_multiple_response_payload_len(&with_bin), bin_a.len() + bin_b.len()); + + // With no msgpack bins, the JSON compatibility strings are summed instead. + let json_only = ReadMultipleResponse { + success: true, + read_multiple_resps: vec![json.clone()], + read_multiple_resps_bin: Vec::new(), + error: None, + }; + assert_eq!(read_multiple_response_payload_len(&json_only), json.len()); + + // An empty response has zero payload. + assert_eq!(read_multiple_response_payload_len(&ReadMultipleResponse::default()), 0); + } + + #[test] + fn compat_json_dual_writes_by_default() { + // msgpack-only defaults off, so compat_json returns the JSON encoding (dual-write). The + // empty-string (msgpack-only) path is exercised via the env flag in integration, not here, + // to keep this test independent of process-global env state. + let resp = sample_read_multiple_resp("file", b"data"); + let json = compat_json(&resp).expect("compat_json should encode"); + assert!(!json.is_empty()); + assert_eq!(json, serde_json::to_string(&resp).expect("json should encode")); + } + #[test] fn read_multiple_response_decode_reports_corrupt_msgpack_item() { let endpoint = sample_remote_endpoint(); diff --git a/crates/ecstore/src/cluster/rpc/remote_locker.rs b/crates/ecstore/src/cluster/rpc/remote_locker.rs index 9aefa8e27..ffdfa9834 100644 --- a/crates/ecstore/src/cluster/rpc/remote_locker.rs +++ b/crates/ecstore/src/cluster/rpc/remote_locker.rs @@ -77,6 +77,12 @@ impl RemoteClient { } pub async fn get_client(&self) -> Result>> { + // P3-2 offline bypass (now covering the lock path too): fast-fail a peer already marked + // offline instead of paying the connect timeout, so dsync reaches quorum sooner. Does not + // change quorum; the self-healing re-probe keeps the peer recoverable. + if let Some(reason) = crate::cluster::rpc::remote_disk::internode_offline_bypass_reason(&self.addr) { + return Err(LockError::internal(reason)); + } node_service_time_out_client(&self.addr, TonicInterceptor::Signature(gen_tonic_signature_interceptor())) .await .map_err(|err| LockError::internal(format!("can not get client, err: {err}"))) diff --git a/crates/ecstore/src/cluster/rpc/runtime_sources.rs b/crates/ecstore/src/cluster/rpc/runtime_sources.rs index de69c516c..0b21769c2 100644 --- a/crates/ecstore/src/cluster/rpc/runtime_sources.rs +++ b/crates/ecstore/src/cluster/rpc/runtime_sources.rs @@ -13,8 +13,9 @@ // limitations under the License. use rustfs_io_metrics::internode_metrics::{ - INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_OPERATION_PUT_FILE_STREAM, - INTERNODE_TRANSPORT_BACKEND_GRPC, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, global_internode_metrics, + INTERNODE_MSGPACK_DIRECTION_RESPONSE, INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_READ_MULTIPLE, + INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_TRANSPORT_BACKEND_GRPC, + INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, global_internode_metrics, }; #[cfg(test)] @@ -70,6 +71,43 @@ pub(crate) fn record_remote_disk_grpc_read_all_recv_bytes(bytes: usize) { INTERNODE_TRANSPORT_BACKEND_GRPC, bytes, ); + record_grpc_payload_size(INTERNODE_OPERATION_GRPC_READ_ALL, bytes); +} + +pub(crate) fn record_remote_disk_grpc_read_multiple_recv_bytes(bytes: usize) { + global_internode_metrics().record_recv_bytes_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_MULTIPLE, + INTERNODE_TRANSPORT_BACKEND_GRPC, + bytes, + ); + record_grpc_payload_size(INTERNODE_OPERATION_GRPC_READ_MULTIPLE, bytes); +} + +/// Payload-size threshold (bytes) above which a unary internode gRPC response is counted as a +/// "large payload" for alerting. Env-overridable via `RUSTFS_INTERNODE_RPC_LARGE_PAYLOAD_WARN_BYTES`. +fn internode_rpc_large_payload_warn_bytes() -> usize { + rustfs_utils::get_env_usize( + rustfs_config::ENV_INTERNODE_RPC_LARGE_PAYLOAD_WARN_BYTES, + rustfs_config::DEFAULT_INTERNODE_RPC_LARGE_PAYLOAD_WARN_BYTES, + ) +} + +/// Record the payload size of a completed unary gRPC RPC into the operation histogram, and +/// flag it as a large payload when it crosses the configured threshold. This instrumentation +/// sizes which `bytes`-carrying RPCs contend with latency-sensitive control-plane traffic on +/// the shared channel, feeding the P1 channel-isolation decision (see docs/grpc-optimization). +fn record_grpc_payload_size(operation: &'static str, bytes: usize) { + let metrics = global_internode_metrics(); + metrics.record_operation_payload_bytes(operation, INTERNODE_TRANSPORT_BACKEND_GRPC, bytes); + if bytes >= internode_rpc_large_payload_warn_bytes() { + metrics.record_large_operation_payload(operation, INTERNODE_TRANSPORT_BACKEND_GRPC); + } +} + +/// Count a client-side response decode that fell back to the JSON compatibility field because the +/// msgpack `_bin` payload was absent (grpc-optimization P2). `message` is the value name. +pub(crate) fn record_response_json_fallback(message: &'static str) { + global_internode_metrics().record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_RESPONSE, message); } #[cfg(test)] diff --git a/crates/io-metrics/src/internode_metrics.rs b/crates/io-metrics/src/internode_metrics.rs index 9e9dbf6a3..a16e21603 100644 --- a/crates/io-metrics/src/internode_metrics.rs +++ b/crates/io-metrics/src/internode_metrics.rs @@ -13,27 +13,36 @@ // limitations under the License. use metrics::{counter, gauge}; +use std::collections::HashMap; use std::sync::{ - Arc, LazyLock, + Arc, LazyLock, Mutex, atomic::{AtomicU64, Ordering}, }; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use std::time::{Duration, Instant, 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"; +pub const INTERNODE_OPERATION_GRPC_READ_MULTIPLE: &str = "grpc_read_multiple"; pub const INTERNODE_TRANSPORT_BACKEND_TCP_HTTP: &str = "tcp-http"; pub const INTERNODE_TRANSPORT_BACKEND_GRPC: &str = "grpc"; pub const INTERNODE_TRANSPORT_BACKEND_UNKNOWN: &str = "unknown"; +/// Direction of a msgpack/JSON codec decode, for the JSON-fallback counter: a server decoding a +/// peer's request vs a client decoding a peer's response (grpc-optimization P2). +pub const INTERNODE_MSGPACK_DIRECTION_REQUEST: &str = "request"; +pub const INTERNODE_MSGPACK_DIRECTION_RESPONSE: &str = "response"; + const OPERATION_LABEL: &str = "operation"; const BACKEND_LABEL: &str = "backend"; const CLASSIFICATION_LABEL: &str = "classification"; const STAGE_LABEL: &str = "stage"; const DOMINANT_ERROR_LABEL: &str = "dominant_error"; const HTTP_VERSION_LABEL: &str = "http_version"; +const DIRECTION_LABEL: &str = "direction"; +const MESSAGE_LABEL: &str = "message"; 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"; @@ -47,6 +56,9 @@ const INTERNODE_OPERATION_HTTP_VERSIONS_TOTAL: &str = "rustfs_system_network_int const INTERNODE_OPERATION_STALL_TIMEOUTS_TOTAL: &str = "rustfs_system_network_internode_operation_stall_timeouts_total"; const INTERNODE_OPERATION_WRITE_SHUTDOWN_ERRORS_TOTAL: &str = "rustfs_system_network_internode_operation_write_shutdown_errors_total"; +const INTERNODE_OPERATION_PAYLOAD_BYTES: &str = "rustfs_system_network_internode_operation_payload_bytes"; +const INTERNODE_OPERATION_LARGE_PAYLOADS_TOTAL: &str = "rustfs_system_network_internode_operation_large_payloads_total"; +const INTERNODE_MSGPACK_JSON_FALLBACK_TOTAL: &str = "rustfs_system_network_internode_msgpack_json_fallback_total"; const ERASURE_WRITE_QUORUM_FAILURES_TOTAL: &str = "rustfs_system_storage_erasure_write_quorum_failures_total"; #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -113,6 +125,14 @@ pub const INTERNODE_OPERATION_METRICS: &[InternodeOperationMetricDescriptor] = & name: ERASURE_WRITE_QUORUM_FAILURES_TOTAL, labels: QUORUM_FAILURE_LABELS, }, + InternodeOperationMetricDescriptor { + name: INTERNODE_OPERATION_PAYLOAD_BYTES, + labels: OPERATION_BACKEND_LABELS, + }, + InternodeOperationMetricDescriptor { + name: INTERNODE_OPERATION_LARGE_PAYLOADS_TOTAL, + labels: OPERATION_BACKEND_LABELS, + }, ]; #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] @@ -315,6 +335,31 @@ impl InternodeMetrics { .increment(1); } + /// Record the payload size (bytes) of a completed internode operation into a histogram + /// keyed by operation+backend. Used to size which unary `bytes`-carrying RPCs + /// (`ReadAll`/`ReadMultiple`/`WriteAll`) would benefit from being moved off the shared + /// control-plane channel (see docs/grpc-optimization P1). + pub fn record_operation_payload_bytes(&self, operation: &'static str, backend: &'static str, bytes: usize) { + metrics::histogram!(INTERNODE_OPERATION_PAYLOAD_BYTES, OPERATION_LABEL => operation, BACKEND_LABEL => backend) + .record(bytes as f64); + } + + /// Increment the large-payload counter for an operation+backend whose payload exceeded the + /// caller-configured warning threshold. Feeds alerting on large unary RPCs that contend with + /// latency-sensitive control-plane traffic on the shared connection. + pub fn record_large_operation_payload(&self, operation: &'static str, backend: &'static str) { + counter!(INTERNODE_OPERATION_LARGE_PAYLOADS_TOTAL, OPERATION_LABEL => operation, BACKEND_LABEL => backend).increment(1); + } + + /// Count a decode that fell back to the JSON compatibility field because the msgpack `_bin` + /// payload was absent. Both internode RPC directions dual-encode msgpack + JSON today; this + /// counter must read zero across a release window before the redundant JSON fields can be + /// dropped (grpc-optimization P2). `direction` is [`INTERNODE_MSGPACK_DIRECTION_REQUEST`] or + /// [`INTERNODE_MSGPACK_DIRECTION_RESPONSE`]; `message` is the low-cardinality value name. + pub fn record_msgpack_json_fallback(&self, direction: &'static str, message: &'static str) { + counter!(INTERNODE_MSGPACK_JSON_FALLBACK_TOTAL, DIRECTION_LABEL => direction, MESSAGE_LABEL => message).increment(1); + } + pub fn record_erasure_write_quorum_failure(&self, stage: &'static str, dominant_error: &'static str) { counter!( ERASURE_WRITE_QUORUM_FAILURES_TOTAL, @@ -386,6 +431,107 @@ pub fn global_internode_metrics() -> &'static Arc { &GLOBAL_INTERNODE_METRICS } +// ── Cluster peer online/offline health (grpc-optimization P3) ── +// Tracks reachability of each internode peer and exposes the count of offline peers as a gauge, +// for parity with MinIO's `minio_cluster_servers_offline_total`. This is pure observability: it +// does not change peer selection or quorum. A peer flips offline after a configured number of +// consecutive failures and back online on the next successful dial. + +/// Gauge: number of internode peers currently considered offline. +const CLUSTER_SERVERS_OFFLINE_TOTAL: &str = "rustfs_cluster_servers_offline_total"; + +#[derive(Debug)] +struct PeerHealthState { + online: bool, + consecutive_failures: u32, + /// Last time a request was let through to re-probe an offline peer (grpc-optimization P3 + /// offline bypass). `None` means "not yet re-probed since going offline". + last_reprobe: Option, +} + +impl Default for PeerHealthState { + fn default() -> Self { + // A newly observed peer is assumed online until it accrues failures. + Self { + online: true, + consecutive_failures: 0, + last_reprobe: None, + } + } +} + +static CLUSTER_PEER_HEALTH: LazyLock>> = LazyLock::new(|| Mutex::new(HashMap::new())); + +fn publish_offline_gauge(peers: &HashMap) { + let offline = peers.values().filter(|peer| !peer.online).count(); + gauge!(CLUSTER_SERVERS_OFFLINE_TOTAL).set(offline as f64); +} + +/// Record that a cluster peer is reachable: mark it online and reset its consecutive-failure +/// counter. Called on a successful dial to `addr`. +pub fn record_peer_reachable(addr: &str) { + // Recover from a poisoned lock so peer-health tracking and the offline gauge never stall permanently. + let mut peers = CLUSTER_PEER_HEALTH.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let entry = peers.entry(addr.to_string()).or_default(); + entry.online = true; + entry.consecutive_failures = 0; + entry.last_reprobe = None; + publish_offline_gauge(&peers); +} + +/// Record a failed interaction with a cluster peer (dial failure or RPC-triggered eviction). After +/// `failure_threshold` (>= 1) consecutive failures the peer flips offline. +pub fn record_peer_unreachable(addr: &str, failure_threshold: u32) { + // Recover from a poisoned lock so peer-health tracking and the offline gauge never stall permanently. + let mut peers = CLUSTER_PEER_HEALTH.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let entry = peers.entry(addr.to_string()).or_default(); + entry.consecutive_failures = entry.consecutive_failures.saturating_add(1); + if entry.consecutive_failures >= failure_threshold.max(1) { + entry.online = false; + } + publish_offline_gauge(&peers); +} + +/// Whether a cluster peer is currently considered offline (known and marked offline). +pub fn cluster_peer_is_offline(addr: &str) -> bool { + let peers = CLUSTER_PEER_HEALTH.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + peers.get(addr).map(|peer| !peer.online).unwrap_or(false) +} + +/// Decide whether to fast-fail (bypass) an offline peer instead of attempting to reach it +/// (grpc-optimization P3 offline bypass). Returns `true` to bypass. +/// +/// Self-healing: for an offline peer this returns `true` most of the time, but lets one request +/// through every `reprobe_interval` (returning `false` and recording the re-probe time) so the peer +/// can recover via a normal dial even if no background monitor is running. Online peers are never +/// bypassed. +pub fn cluster_peer_should_bypass(addr: &str, reprobe_interval: Duration) -> bool { + let mut peers = CLUSTER_PEER_HEALTH.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + let Some(entry) = peers.get_mut(addr) else { + return false; + }; + if entry.online { + return false; + } + let now = Instant::now(); + let due = match entry.last_reprobe { + None => true, + Some(last) => now.duration_since(last) >= reprobe_interval, + }; + if due { + // Let this request through to re-probe; do not bypass. + entry.last_reprobe = Some(now); + false + } else { + true + } +} + +#[cfg(test)] +fn cluster_peer_online(addr: &str) -> Option { + CLUSTER_PEER_HEALTH.lock().ok()?.get(addr).map(|peer| peer.online) +} + #[cfg(test)] mod tests { use super::*; @@ -450,7 +596,7 @@ mod tests { #[test] fn operation_metric_descriptors_include_backend_and_operation_labels() { - assert_eq!(INTERNODE_OPERATION_METRICS.len(), 13); + assert_eq!(INTERNODE_OPERATION_METRICS.len(), 15); for metric in &INTERNODE_OPERATION_METRICS[..6] { assert_eq!(metric.labels, &[OPERATION_LABEL, BACKEND_LABEL]); } @@ -465,6 +611,9 @@ mod tests { assert_eq!(metric.labels, &[OPERATION_LABEL, BACKEND_LABEL]); } assert_eq!(INTERNODE_OPERATION_METRICS[12].labels, &[STAGE_LABEL, DOMINANT_ERROR_LABEL]); + // Payload histogram + large-payload counter carry operation+backend labels. + assert_eq!(INTERNODE_OPERATION_METRICS[13].labels, &[OPERATION_LABEL, BACKEND_LABEL]); + assert_eq!(INTERNODE_OPERATION_METRICS[14].labels, &[OPERATION_LABEL, BACKEND_LABEL]); } #[test] @@ -511,6 +660,98 @@ mod tests { INTERNODE_OPERATION_METRICS[12].name, "rustfs_system_storage_erasure_write_quorum_failures_total" ); + assert_eq!( + INTERNODE_OPERATION_METRICS[13].name, + "rustfs_system_network_internode_operation_payload_bytes" + ); + assert_eq!( + INTERNODE_OPERATION_METRICS[14].name, + "rustfs_system_network_internode_operation_large_payloads_total" + ); + assert_eq!(INTERNODE_OPERATION_GRPC_READ_MULTIPLE, "grpc_read_multiple"); + assert_eq!( + INTERNODE_MSGPACK_JSON_FALLBACK_TOTAL, + "rustfs_system_network_internode_msgpack_json_fallback_total" + ); + assert_eq!(INTERNODE_MSGPACK_DIRECTION_REQUEST, "request"); + assert_eq!(INTERNODE_MSGPACK_DIRECTION_RESPONSE, "response"); + } + + #[test] + fn msgpack_json_fallback_counter_records_without_panicking() { + // Smoke test: the counter accepts both directions and a static message label. + let metrics = InternodeMetrics::default(); + metrics.record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_REQUEST, "FileInfo"); + metrics.record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_RESPONSE, "RawFileInfo"); + } + + #[test] + fn cluster_peer_flips_offline_after_threshold_and_back_online() { + // Unique addr keeps this independent of the process-global registry / other tests. + let addr = "http://cluster-peer-health-unit-test:9000"; + assert_eq!(cluster_peer_online(addr), None); + + record_peer_unreachable(addr, 3); + record_peer_unreachable(addr, 3); + assert_eq!(cluster_peer_online(addr), Some(true), "still online below threshold"); + + record_peer_unreachable(addr, 3); + assert_eq!(cluster_peer_online(addr), Some(false), "offline at threshold"); + + record_peer_reachable(addr); + assert_eq!(cluster_peer_online(addr), Some(true), "back online after a reachable dial"); + } + + #[test] + fn cluster_peer_threshold_is_clamped_to_at_least_one() { + let addr = "http://cluster-peer-health-clamp-test:9000"; + // A zero threshold must not mean "never offline"; one failure suffices. + record_peer_unreachable(addr, 0); + assert_eq!(cluster_peer_online(addr), Some(false)); + record_peer_reachable(addr); + assert_eq!(cluster_peer_online(addr), Some(true)); + } + + #[test] + fn cluster_servers_offline_total_name_is_stable() { + assert_eq!(CLUSTER_SERVERS_OFFLINE_TOTAL, "rustfs_cluster_servers_offline_total"); + } + + #[test] + fn cluster_peer_should_bypass_is_self_healing() { + let addr = "http://cluster-peer-bypass-selfheal-test:9000"; + let interval = Duration::from_secs(30); + + // Online peer: never bypassed. + record_peer_reachable(addr); + assert!(!cluster_peer_should_bypass(addr, interval)); + + // Take it offline (threshold 1 for the test). + record_peer_unreachable(addr, 1); + assert!(cluster_peer_is_offline(addr)); + + // First decision after going offline is a re-probe (not bypassed)... + assert!(!cluster_peer_should_bypass(addr, interval)); + // ...and subsequent ones within the interval are bypassed. + assert!(cluster_peer_should_bypass(addr, interval)); + assert!(cluster_peer_should_bypass(addr, interval)); + + // A zero interval always allows a re-probe (never strands the peer). + assert!(!cluster_peer_should_bypass(addr, Duration::ZERO)); + + // Recovery clears bypass entirely. + record_peer_reachable(addr); + assert!(!cluster_peer_should_bypass(addr, interval)); + } + + #[test] + fn cluster_peer_should_bypass_ignores_unknown_and_online_peers() { + let addr = "http://cluster-peer-bypass-unknown-test:9000"; + // Unknown peer: not bypassed. + assert!(!cluster_peer_should_bypass(addr, Duration::from_secs(5))); + // Known-online peer: not bypassed. + record_peer_reachable(addr); + assert!(!cluster_peer_should_bypass(addr, Duration::from_secs(5))); } #[test] diff --git a/crates/protos/src/generated/proto_gen/node_service.rs b/crates/protos/src/generated/proto_gen/node_service.rs index 773603369..68b095a5f 100644 --- a/crates/protos/src/generated/proto_gen/node_service.rs +++ b/crates/protos/src/generated/proto_gen/node_service.rs @@ -586,6 +586,10 @@ pub struct DeleteVersionRequest { pub force_del_marker: bool, #[prost(string, tag = "6")] pub opts: ::prost::alloc::string::String, + #[prost(bytes = "bytes", tag = "7")] + pub file_info_bin: ::prost::bytes::Bytes, + #[prost(bytes = "bytes", tag = "8")] + pub opts_bin: ::prost::bytes::Bytes, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct DeleteVersionResponse { @@ -606,6 +610,10 @@ pub struct DeleteVersionsRequest { pub versions: ::prost::alloc::vec::Vec<::prost::alloc::string::String>, #[prost(string, tag = "4")] pub opts: ::prost::alloc::string::String, + #[prost(bytes = "bytes", repeated, tag = "5")] + pub versions_bin: ::prost::alloc::vec::Vec<::prost::bytes::Bytes>, + #[prost(bytes = "bytes", tag = "6")] + pub opts_bin: ::prost::bytes::Bytes, } #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)] pub struct DeleteVersionsResponse { diff --git a/crates/protos/src/lib.rs b/crates/protos/src/lib.rs index eadd2a69b..f51e298c5 100644 --- a/crates/protos/src/lib.rs +++ b/crates/protos/src/lib.rs @@ -19,11 +19,12 @@ mod generated; mod runtime_sources; use proto_gen::node_service::node_service_client::NodeServiceClient; -use rustfs_common::{cache_connection, evict_connection_with_log_level}; +use rustfs_common::{cache_connection, cached_connection, evict_connection_with_log_level}; use std::{ collections::HashMap, error::Error, sync::LazyLock, + sync::atomic::{AtomicUsize, Ordering}, time::{Duration, Instant}, }; use tokio::sync::Mutex; @@ -101,6 +102,128 @@ fn internode_rpc_timeout() -> Duration { )) } +fn internode_rpc_tcp_nodelay() -> bool { + rustfs_utils::get_env_bool( + rustfs_config::ENV_INTERNODE_RPC_TCP_NODELAY, + rustfs_config::DEFAULT_INTERNODE_RPC_TCP_NODELAY, + ) +} + +/// HTTP/2 initial stream window size for the client channel, or `None` to use the +/// library default (a configured value of `0` opts out). +fn internode_rpc_http2_stream_window() -> Option { + match rustfs_utils::get_env_u32( + rustfs_config::ENV_INTERNODE_RPC_HTTP2_STREAM_WINDOW_SIZE, + rustfs_config::DEFAULT_INTERNODE_RPC_HTTP2_STREAM_WINDOW_SIZE, + ) { + 0 => None, + v => Some(v), + } +} + +/// HTTP/2 initial connection window size for the client channel, or `None` to use the +/// library default (a configured value of `0` opts out). +fn internode_rpc_http2_conn_window() -> Option { + match rustfs_utils::get_env_u32( + rustfs_config::ENV_INTERNODE_RPC_HTTP2_CONN_WINDOW_SIZE, + rustfs_config::DEFAULT_INTERNODE_RPC_HTTP2_CONN_WINDOW_SIZE, + ) { + 0 => None, + v => Some(v), + } +} + +/// Maximum encoded/decoded internode gRPC message size (bytes), shared by the client +/// `NodeServiceClient` and server `NodeServiceServer`. Defaults to +/// [`DEFAULT_GRPC_SERVER_MESSAGE_LEN`] (100 MiB) when the env var is unset. +pub fn internode_rpc_max_message_size() -> usize { + rustfs_utils::get_env_usize(rustfs_config::ENV_INTERNODE_RPC_MAX_MESSAGE_SIZE, DEFAULT_GRPC_SERVER_MESSAGE_LEN) +} + +/// Whether internode metadata RPCs should send only the msgpack `_bin` payloads and leave the JSON +/// compatibility strings empty (grpc-optimization P2-1). Shared by the client (`remote_disk`) and +/// server (`node_service`) send paths. Defaults to `false` (dual-write); see +/// [`rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY`] and the convergence runbook before enabling. +pub fn internode_rpc_msgpack_only() -> bool { + rustfs_utils::get_env_bool( + rustfs_config::ENV_INTERNODE_RPC_MSGPACK_ONLY, + rustfs_config::DEFAULT_INTERNODE_RPC_MSGPACK_ONLY, + ) +} + +/// Consecutive-failure threshold after which an internode peer is marked offline (grpc-optimization +/// P3). Drives the `rustfs_cluster_servers_offline_total` gauge. +fn internode_offline_failure_threshold() -> u32 { + rustfs_utils::get_env_u32( + rustfs_config::ENV_INTERNODE_OFFLINE_FAILURE_THRESHOLD, + rustfs_config::DEFAULT_INTERNODE_OFFLINE_FAILURE_THRESHOLD, + ) +} + +/// Class of internode gRPC channel used to route an RPC (grpc-optimization P1). +/// +/// - [`ChannelClass::Control`]: latency-sensitive control-plane RPCs (locks, health, small +/// metadata). Always uses the per-peer control connection keyed by the bare address. +/// - [`ChannelClass::Bulk`]: large `bytes`-carrying unary RPCs +/// (`ReadAll`/`WriteAll`/`ReadMultiple`/`BatchReadVersion`). When channel isolation is enabled +/// these are routed onto a separate per-peer bulk connection pool so a large transfer cannot +/// head-of-line block a lock RPC on the shared HTTP/2 connection. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ChannelClass { + Control, + Bulk, +} + +/// Whether control/bulk channel isolation is enabled (env-gated, default off for safe rollout). +fn channel_isolation_enabled() -> bool { + rustfs_utils::get_env_bool( + rustfs_config::ENV_INTERNODE_CHANNEL_ISOLATION, + rustfs_config::DEFAULT_INTERNODE_CHANNEL_ISOLATION, + ) +} + +/// Number of bulk channels maintained per peer (>= 1). +fn bulk_channel_pool_size() -> usize { + rustfs_utils::get_env_usize(rustfs_config::ENV_INTERNODE_BULK_CHANNELS, rustfs_config::DEFAULT_INTERNODE_BULK_CHANNELS).max(1) +} + +/// Round-robin cursor for selecting a bulk channel within a peer's pool. A single global cursor +/// is sufficient: it advances once per bulk acquisition and `% pool_size` spreads consecutive +/// acquisitions across the pool. +static BULK_CHANNEL_CURSOR: AtomicUsize = AtomicUsize::new(0); + +/// Connection-cache key for the `idx`-th bulk channel to `addr`. The NUL separator cannot appear +/// in a URL, so a bulk key never collides with the control key (the bare `addr`). +fn bulk_cache_key(addr: &str, idx: usize) -> String { + format!("{addr}\u{0}bulk\u{0}{idx}") +} + +/// Acquire a cached-or-newly-dialed channel for the given peer and channel class. +/// +/// For [`ChannelClass::Control`] (or whenever isolation is disabled) this is exactly the legacy +/// behavior: reuse the cached control channel keyed by `addr`, else dial a new one. For +/// [`ChannelClass::Bulk`] with isolation enabled, a channel is round-robin selected from the +/// per-peer bulk pool and dialed on demand, physically isolating large transfers from +/// control-plane RPCs. +pub async fn get_channel_for_class(addr: &str, class: ChannelClass) -> Result> { + if class == ChannelClass::Control || !channel_isolation_enabled() { + if let Some(channel) = cached_connection(addr).await { + debug!("Using cached control gRPC channel for: {}", addr); + return Ok(channel); + } + return create_new_channel(addr).await; + } + + let pool_size = bulk_channel_pool_size(); + let idx = BULK_CHANNEL_CURSOR.fetch_add(1, Ordering::Relaxed) % pool_size; + let cache_key = bulk_cache_key(addr, idx); + if let Some(channel) = cached_connection(&cache_key).await { + debug!("Using cached bulk gRPC channel {} for: {}", idx, addr); + return Ok(channel); + } + build_channel(addr, &cache_key).await +} + /// Creates a new gRPC channel with optimized keepalive settings for cluster resilience. /// /// This function is designed to detect dead peers quickly using env-configurable @@ -111,7 +234,18 @@ fn internode_rpc_timeout() -> Duration { /// - HTTP/2 keepalive timeout: `DEFAULT_INTERNODE_HTTP2_KEEPALIVE_TIMEOUT_SECS` (3s) /// - RPC timeout: `DEFAULT_INTERNODE_RPC_TIMEOUT_SECS` (10s) pub async fn create_new_channel(addr: &str) -> Result> { - debug!("Creating new gRPC channel to: {}", addr); + // The control channel is cached under the bare address, preserving the legacy key. + build_channel(addr, addr).await +} + +/// Dial a new gRPC channel to `dial_addr` and cache it under `cache_key`. +/// +/// `dial_addr` is the real peer URL used for the TCP/TLS connection and hostname verification; +/// `cache_key` is the connection-cache key. They are identical for control channels and differ +/// for isolated bulk channels (see [`bulk_cache_key`]), letting several physically distinct +/// channels to the same peer be cached independently. +async fn build_channel(dial_addr: &str, cache_key: &str) -> Result> { + debug!("Creating new gRPC channel to: {} (cache key: {})", dial_addr, cache_key); let dial_started_at = Instant::now(); let connect_timeout = internode_connect_timeout(); let tcp_keepalive = internode_tcp_keepalive(); @@ -119,11 +253,13 @@ pub async fn create_new_channel(addr: &str) -> Result> { let http2_keepalive_timeout = internode_http2_keep_alive_timeout(); let rpc_timeout = internode_rpc_timeout(); - let mut connector = Endpoint::from_shared(addr.to_string())? + let mut connector = Endpoint::from_shared(dial_addr.to_string())? // Fast connection timeout for dead peer detection .connect_timeout(connect_timeout) // TCP-level keepalive - OS will probe connection .tcp_keepalive(Some(tcp_keepalive)) + // Disable Nagle so latency-sensitive control-plane RPCs (locks/health) are not batched + .tcp_nodelay(internode_rpc_tcp_nodelay()) // HTTP/2 PING frames for application-layer health check .http2_keep_alive_interval(http2_keepalive_interval) // How long to wait for PING ACK before considering connection dead @@ -133,22 +269,32 @@ pub async fn create_new_channel(addr: &str) -> Result> { // Overall timeout for any RPC - fail fast on unresponsive peers .timeout(rpc_timeout); + // Raise HTTP/2 flow-control windows above the 64KiB library default so larger unary + // responses (ReadMultiple/BatchReadVersion) are not throttled by the BDP. Mirrors the + // server-side window tuning in `rustfs/src/server/http.rs`. + if let Some(stream_window) = internode_rpc_http2_stream_window() { + connector = connector.initial_stream_window_size(stream_window); + } + if let Some(conn_window) = internode_rpc_http2_conn_window() { + connector = connector.initial_connection_window_size(conn_window); + } + let outbound_tls = runtime_sources::outbound_tls_state().await; let generation = outbound_tls.generation.0; let mut stale_generation = false; { let generation_cache = TLS_GENERATION_CACHE.lock().await; - if let Some(cached_generation) = generation_cache.get(addr) + if let Some(cached_generation) = generation_cache.get(cache_key) && *cached_generation != generation { stale_generation = true; } } - if addr.starts_with(RUSTFS_HTTPS_PREFIX) { + if dial_addr.starts_with(RUSTFS_HTTPS_PREFIX) { if let Some(cert_pem) = outbound_tls.root_ca_pem.as_ref() { let ca = Certificate::from_pem(cert_pem); // Derive the hostname from the HTTPS URL for TLS hostname verification. - let domain = addr + let domain = dial_addr .trim_start_matches(RUSTFS_HTTPS_PREFIX) .split('/') .next() @@ -168,11 +314,11 @@ pub async fn create_new_channel(addr: &str) -> Result> { ClientTlsConfig::new().ca_certificate(ca) }; connector = connector.tls_config(tls)?; - debug!("Configured TLS with custom root certificate for: {}", addr); + debug!("Configured TLS with custom root certificate for: {}", dial_addr); } else { // No custom root CA published — fall back to system roots. // This is the expected path when no TLS path is configured. - debug!("No custom root certificate configured; using system roots for TLS: {}", addr); + debug!("No custom root certificate configured; using system roots for TLS: {}", dial_addr); connector = connector.tls_config(ClientTlsConfig::new())?; } } @@ -180,25 +326,32 @@ pub async fn create_new_channel(addr: &str) -> Result> { let channel = match connector.connect().await { Ok(channel) => { runtime_sources::record_grpc_dial_result(dial_started_at.elapsed(), true); + // A successful dial marks the peer online (grpc-optimization P3). Keyed by the real + // peer address, so control and bulk channels to one peer share health state. + runtime_sources::record_peer_reachable(dial_addr); channel } Err(err) => { runtime_sources::record_grpc_dial_result(dial_started_at.elapsed(), false); + runtime_sources::record_peer_unreachable(dial_addr, internode_offline_failure_threshold()); return Err(err.into()); } }; - cache_connection(addr.to_string(), channel.clone()).await; + cache_connection(cache_key.to_string(), channel.clone()).await; { let mut generation_cache = TLS_GENERATION_CACHE.lock().await; - enforce_tls_generation_cache_bound(&mut generation_cache, generation, addr); - generation_cache.insert(addr.to_string(), generation); + enforce_tls_generation_cache_bound(&mut generation_cache, generation, cache_key); + generation_cache.insert(cache_key.to_string(), generation); } if stale_generation { runtime_sources::record_stale_grpc_channel_tls_generation(); } - debug!("Successfully created and cached gRPC channel to: {}", addr); + debug!( + "Successfully created and cached gRPC channel to: {} (cache key: {})", + dial_addr, cache_key + ); Ok(channel) } @@ -237,6 +390,21 @@ pub async fn evict_failed_connection_with_log_level(addr: &str, log_level: Conne }; evict_connection_with_log_level(addr, cache_log_level).await; TLS_GENERATION_CACHE.lock().await.remove(addr); + + // An RPC-triggered eviction is a peer-failure signal; feed it into the online/offline state so + // the peer flips offline after enough consecutive failures (grpc-optimization P3). + runtime_sources::record_peer_unreachable(addr, internode_offline_failure_threshold()); + + // A peer failure typically affects every channel to that peer, so also drop any isolated + // bulk channels rather than leaving a half-dead connection cached. Round-robin selection + // means the caller cannot know which bulk index it hit, so evict the whole pool. + if channel_isolation_enabled() { + for idx in 0..bulk_channel_pool_size() { + let cache_key = bulk_cache_key(addr, idx); + evict_connection_with_log_level(&cache_key, cache_log_level).await; + TLS_GENERATION_CACHE.lock().await.remove(&cache_key); + } + } } #[cfg(test)] @@ -264,4 +432,37 @@ mod tests { assert!(!rustfs_common::has_cached_connection(addr).await); } + + #[test] + fn bulk_cache_key_is_distinct_and_cannot_collide_with_control_key() { + let addr = "https://node-a:9000"; + let k0 = bulk_cache_key(addr, 0); + let k1 = bulk_cache_key(addr, 1); + assert_ne!(k0, k1); + assert_ne!(k0, addr); + assert!(k0.starts_with(addr)); + // The NUL separator cannot appear in a URL, so a bulk key never equals a real address. + assert!(k0.contains('\u{0}')); + } + + #[test] + fn bulk_channel_pool_size_is_at_least_one() { + // Even without env configuration the pool size is clamped to a usable minimum. + assert!(bulk_channel_pool_size() >= 1); + } + + #[tokio::test] + async fn get_channel_for_class_bulk_reuses_control_cache_when_isolation_disabled() { + // Isolation defaults off, so a Bulk request must reuse the control channel keyed by the + // bare address and must NOT create a separate bulk-keyed entry (zero behavior change). + let addr = "http://get-channel-isolation-off-test"; + let channel = Endpoint::from_static("http://127.0.0.1:1").connect_lazy(); + cache_connection(addr.to_string(), channel).await; + + let acquired = get_channel_for_class(addr, ChannelClass::Bulk).await; + assert!(acquired.is_ok()); + assert!(!rustfs_common::has_cached_connection(&bulk_cache_key(addr, 0)).await); + + evict_failed_connection_with_log_level(addr, ConnectionEvictionLogLevel::Debug).await; + } } diff --git a/crates/protos/src/node.proto b/crates/protos/src/node.proto index b7f59cce3..5b667c8f6 100644 --- a/crates/protos/src/node.proto +++ b/crates/protos/src/node.proto @@ -405,6 +405,10 @@ message DeleteVersionRequest { string file_info = 4; bool force_del_marker = 5; string opts = 6; + // msgpack payloads mirroring the JSON fields above (grpc-optimization P2). Senders dual-write + // both; receivers prefer the *_bin form and fall back to the JSON string when it is empty. + bytes file_info_bin = 7; + bytes opts_bin = 8; } message DeleteVersionResponse { @@ -418,6 +422,10 @@ message DeleteVersionsRequest { string volume = 2; repeated string versions = 3; string opts = 4; + // msgpack payloads mirroring the JSON fields above (grpc-optimization P2). Senders dual-write + // both; receivers prefer the *_bin form and fall back to the JSON string when it is empty. + repeated bytes versions_bin = 5; + bytes opts_bin = 6; } message DeleteVersionsResponse { diff --git a/crates/protos/src/runtime_sources.rs b/crates/protos/src/runtime_sources.rs index 3bf1d61eb..135e3a9f5 100644 --- a/crates/protos/src/runtime_sources.rs +++ b/crates/protos/src/runtime_sources.rs @@ -29,3 +29,14 @@ pub(crate) fn record_stale_grpc_channel_tls_generation() { pub(crate) fn record_grpc_dial_result(duration: Duration, success: bool) { global_internode_metrics().record_dial_result(duration, success); } + +/// Mark an internode peer reachable (online) after a successful dial (grpc-optimization P3). +pub(crate) fn record_peer_reachable(addr: &str) { + rustfs_io_metrics::internode_metrics::record_peer_reachable(addr); +} + +/// Record a peer failure (dial failure or RPC-triggered eviction); flips the peer offline once it +/// crosses `failure_threshold` consecutive failures (grpc-optimization P3). +pub(crate) fn record_peer_unreachable(addr: &str, failure_threshold: u32) { + rustfs_io_metrics::internode_metrics::record_peer_unreachable(addr, failure_threshold); +} diff --git a/docs/operations/internode-grpc-benchmark-runbook.md b/docs/operations/internode-grpc-benchmark-runbook.md new file mode 100644 index 000000000..97157d30a --- /dev/null +++ b/docs/operations/internode-grpc-benchmark-runbook.md @@ -0,0 +1,97 @@ +# Internode gRPC Optimization — A/B Benchmark Runbook + +Reproducible procedure to collect **before/after** artifacts for each internode gRPC +optimization stage (grpc-optimization P0–P3). Every stage is env-gated, so "before" and +"after" are the *same binary* with different env — no rebuild between runs. + +> Live runs need a multi-node cluster (Docker or ≥2 rustfs endpoints), a load tool +> (`warp` or `s3bench`), and a Prometheus scrape of `/metrics`. They are not runnable in a +> single-process sandbox. Capture artifacts on a real cluster. + +## One-click driver + +`scripts/run_internode_grpc_ab_bench.sh --stage --phase [-- ]` +wraps the env matrix below: it writes the stage/phase **server** env to +`/server-env.sh`, then runs the right underlying bench into +`target/bench/internode-transport/-/`. + +```bash +# P1 A/B (restart the cluster with each phase's server-env.sh between the two runs): +scripts/run_internode_grpc_ab_bench.sh --stage p1 --phase before -- --access-key AK --secret-key SK --metrics-url http://node1:9000/metrics +scripts/run_internode_grpc_ab_bench.sh --stage p1 --phase after -- --access-key AK --secret-key SK --metrics-url http://node1:9000/metrics +# P3 failover A/B (docker four-node): +scripts/run_internode_grpc_ab_bench.sh --stage p3 --phase after +``` + +`RUSTFS_INTERNODE_*` are **server** env: for the load-driven stages (p0/p1/p2) source the emitted +`server-env.sh` on every node and restart rustfs *before* the run — the driver cannot mutate an +already-running server. Use `--dry-run` to preview the env and command. + +## Harness + +- Throughput / latency: `scripts/run_internode_transport_baseline.sh` (drives + `run_object_batch_bench.sh`; writes `target/bench/internode-transport-/`). Pass + `--metrics-url ` to also capture internode metric deltas. +- Failover / offline: `scripts/run_four_node_cluster_failover_bench.sh` (spins up a 4-node + compose cluster, kills `FAILOVER_NODE`, benchmarks; writes + `target/bench/four-node-failover-/`). + +Store the paired artifacts as `target/bench/internode-transport/{baseline,after-P0,after-P1,after-P3}/` +(the paths the design docs reference). `target/` is gitignored — attach the artifacts to the PR. + +## Metrics to capture (Prometheus) + +| Metric | Stage signal | +|---|---| +| `rustfs_system_network_internode_operation_duration_ms{operation,backend}` | control-plane RTT (P0), lock/bulk latency | +| `rustfs_system_network_internode_operation_payload_bytes` | payload size distribution (P0/P1 sizing) | +| `rustfs_system_network_internode_operation_large_payloads_total` | large unary RPCs sharing the channel (P1 target) | +| `rustfs_system_network_internode_dial_avg_time_nanos`, `..._dial_errors_total` | connect cost (P3 prewarm) | +| `rustfs_system_network_internode_msgpack_json_fallback_total{direction,message}` | must be **0** before enabling msgpack-only (P2) | +| `rustfs_cluster_servers_offline_total` | offline detection correctness (P3 bypass) | +| lock p99 (lock metrics) | P1 head-of-line-blocking win | + +## Per-stage env matrix + +Run **before** with the stage's env at its baseline column, **after** with the enabled +column, everything else at defaults. Roll a restart between runs. + +| Stage | Env | before (baseline) | after (enabled) | +|---|---|---|---| +| P0 nodelay | `RUSTFS_INTERNODE_RPC_TCP_NODELAY` | `false` | `true` (default) | +| P0 stream window | `RUSTFS_INTERNODE_RPC_HTTP2_STREAM_WINDOW_SIZE` | `0` | unset (1 MiB) | +| P0 conn window | `RUSTFS_INTERNODE_RPC_HTTP2_CONN_WINDOW_SIZE` | `0` | unset (2 MiB) | +| P0 msg limit | `RUSTFS_INTERNODE_RPC_MAX_MESSAGE_SIZE` | `4194304` | unset (100 MiB) | +| P1 isolation | `RUSTFS_INTERNODE_CHANNEL_ISOLATION` | `false` (default) | `true` | +| P1 bulk pool | `RUSTFS_INTERNODE_BULK_CHANNELS` | `1` | `2`–`4` | +| P2 msgpack-only | `RUSTFS_INTERNODE_RPC_MSGPACK_ONLY` | `false` (default) | `true` (only after fallback counter = 0 across a window) | +| P3 prewarm | `RUSTFS_INTERNODE_PREWARM` | `false` (default) | `true` | +| P3 offline bypass | `RUSTFS_INTERNODE_OFFLINE_BYPASS` | `false` (default) | `true` | +| P3 reprobe / threshold | `RUSTFS_INTERNODE_OFFLINE_REPROBE_SECS` / `RUSTFS_INTERNODE_OFFLINE_FAILURE_THRESHOLD` | defaults | `5` / `3` | + +## Procedure per stage + +1. **Baseline**: start the cluster with the stage's env at the *before* column. Run the + relevant bench; save to `.../baseline/` (or `.../after-P{n-1}/` when chaining stages). +2. **After**: restart with the *after* column; re-run the identical bench; save to + `.../after-P{n}/`. +3. Diff the object-bench summaries and the metric deltas. + +- **P0** — `run_internode_transport_baseline.sh` with `--sizes 4KiB,1MiB,16MiB,128MiB` and + `--concurrencies 1,16,64`. Expect: small-RPC `duration_ms` (DiskInfo/Ping) down (nodelay), + large-metadata (ReadMultiple/BatchReadVersion) throughput up (windows). Functional: a + `>4 MiB` multi-version `xl.meta` no longer fails `out_of_range`. +- **P1** — mixed workload (large `ReadAll` + high-frequency `Refresh`). Acceptance gate from + the design doc: **lock p99 down ≥ 20%** with `RUSTFS_INTERNODE_CHANNEL_ISOLATION=true`. +- **P2** — observe `msgpack_json_fallback_total` across a release window; it must stay **0** + before flipping `RUSTFS_INTERNODE_RPC_MSGPACK_ONLY=true` (see the msgpack convergence + runbook). Codec allocation via a `dhat`/`heaptrack` micro-run. +- **P3** — cold-start: first cross-node op latency should drop ~one connect RTT with prewarm. + Failover: `run_four_node_cluster_failover_bench.sh`, kill a node with + `RUSTFS_INTERNODE_OFFLINE_BYPASS=true`; expect faster failover and a correct + `rustfs_cluster_servers_offline_total` (1 while the node is down, back to 0 after recovery). + +## Rollback + +Every stage is a single-env rollback (set the env back to its baseline column and restart). +No wire-format is broken except P2 stage 2 (proto field removal), which is a separate release. diff --git a/docs/operations/internode-msgpack-json-convergence-runbook.md b/docs/operations/internode-msgpack-json-convergence-runbook.md new file mode 100644 index 000000000..268f6da73 --- /dev/null +++ b/docs/operations/internode-msgpack-json-convergence-runbook.md @@ -0,0 +1,145 @@ +# Internode msgpack/JSON Convergence Runbook + +Operational runbook for retiring the redundant JSON compatibility fields on internode +gRPC metadata RPCs (grpc-optimization **P2-1**). This is a **cross-version** change: it +proceeds strictly by observation-gated stages, never in one step. + +## Background + +Internode RPCs dual-encode each metadata value as **both**: + +- a msgpack binary field (`*_bin`, e.g. `file_info_bin`), and +- a JSON compatibility string (e.g. `file_info`). + +Decoders prefer the `_bin` payload and fall back to the JSON string only when `_bin` is +empty (`decode_msgpack_or_json`). The dual-write costs bandwidth and CPU. Before the JSON +fields can be dropped, the fallback branch must be proven **unused** in production — +otherwise a rolling upgrade with mixed node versions could read an emptied field. + +## The observation metric (already shipped) + +``` +rustfs_system_network_internode_msgpack_json_fallback_total{direction, message} +``` + +Incremented whenever a decode falls back to the JSON field because the msgpack payload was +absent. + +- `direction="request"` — a server decoding a peer's request (`node_service/disk.rs`). +- `direction="response"` — a client decoding a peer's response (`cluster/rpc/remote_disk.rs`), + including the list-level `ReadMultiple` / `BatchReadVersion` fallbacks. +- `message` — the value name, e.g. `FileInfo`, `RawFileInfo`, `ReadMultipleResp`. + +## Stage 0 — Observe (current stage) + +Ship the current release (which contains the counter) and let it run for **at least one +full release window** across the whole fleet. The counter must stay at **zero**. + +Confirm zero across the observation window (adjust `[30d]` to the window length): + +```promql +sum by (direction, message) ( + increase(rustfs_system_network_internode_msgpack_json_fallback_total[30d]) +) +``` + +Every series must be `0`. A non-zero value means some peer is still emitting an empty +`_bin` (an old node, or a message whose sender does not fill `_bin`) — investigate the +`{direction, message}` label before proceeding. + +Standing alert (keep enabled through all stages): + +```yaml +- alert: InternodeMsgpackJsonFallback + expr: sum by (direction, message) (increase(rustfs_system_network_internode_msgpack_json_fallback_total[15m])) > 0 + for: 5m + labels: { severity: warning } + annotations: + summary: "Internode RPC fell back to JSON decode ({{ $labels.direction }}/{{ $labels.message }})" + description: "A peer sent an empty msgpack _bin payload. Do NOT advance msgpack-only convergence while this fires." +``` + +## Field → peer-decoder audit + +The send-side change (Stage 1) may only empty a JSON field whose **peer decodes `_bin` +first**. The following mapping is verified against the current code. + +### Convergence-ready (peer decodes `_bin` first) + +| Direction | Message / field | Peer decoder | +|---|---|---| +| request | `WriteMetadata.file_info` | `FileInfo` (node_service/disk.rs) | +| request | `UpdateMetadata.file_info` | `FileInfo` | +| request | `UpdateMetadata.opts` | `UpdateMetadataOpts` | +| request | `RenameData.file_info` | `FileInfo` | +| request | `ReadMultiple.read_multiple_req` | `ReadMultipleReq` | +| request | `BatchReadVersion.batch_read_version_req` | `BatchReadVersionReq` | +| request | `Read*.opts` | `ReadOptions` | +| response | `ReadVersion.file_info` | `FileInfo` (cluster/rpc/remote_disk.rs) | +| response | `ReadXL.raw_file_info` | `RawFileInfo` | +| response | `RenameData.rename_data_resp` | `RenameDataResp` | +| response | `ReadMultiple` resp list | per-item + list fallback | +| response | `BatchReadVersion` resp list | per-item + list fallback | + +### `_bin` support added THIS release — converge after their own window + +The `DeleteVersion`/`DeleteVersions` protos had **no `_bin` fields**. They gained additive +`*_bin` fields plus bin-first server decoders in this release, and the client now dual-writes +them. They are **kept out** of the msgpack-only set (always dual-write) until their own +fallback counter has read zero across a window with the new decoders fully deployed. + +| Direction | Message / field | Status | +|---|---|---| +| request | `DeleteVersion.file_info` (`FileInfo`) | `_bin` added; dual-write; converge after window | +| request | `DeleteVersion.opts` (`DeleteOptions`) | `_bin` added; dual-write; converge after window | +| request | `DeleteVersions.versions` (`FileInfoVersions`) | `_bin` added; dual-write; converge after window | +| request | `DeleteVersions.opts` (`DeleteOptions`) | `_bin` added; dual-write; converge after window | + +> Any `*_bin` proto field not in the tables above must be mapped to a confirmed `_bin`-first +> peer decoder before it is added to the convergence set. + +### Still JSON-only (no `_bin` field) + +| Direction | Message / field | Note | +|---|---|---| +| response | `DeleteVersion.raw_file_info` | proto has no `_bin`; needs an additive proto field before it can converge. | + +## Stage 1 — Stop writing JSON (env-gated, after Stage 0 reads zero) + +The send-side lever is **implemented** and gated by the default-off env flag +`RUSTFS_INTERNODE_RPC_MSGPACK_ONLY`. When enabled, the convergence-ready fields above send +only `_bin` and leave the JSON string empty; the `_bin` payload is always sent and decoders +keep the JSON read fallback unchanged. The delete fields are excluded (dual-write) per the +section above. + +Only enable it **after** Stage 0 has read zero for a full window across the fleet: + +1. Ship with the flag **off** (no behavior change). +2. Enable it on one node (`RUSTFS_INTERNODE_RPC_MSGPACK_ONLY=true`, restart) and watch the + fallback counter for a soak period. If it stays zero, enable fleet-wide. +3. **Rollback:** set `RUSTFS_INTERNODE_RPC_MSGPACK_ONLY=false` (or unset) and restart. No + wire-format was broken in this stage, so rollback is immediate and safe. + +## Stage 2 — Remove the proto JSON fields (next release, N+1) + +Only after Stage 1 has been stable with the flag on for a full window and the counter is +still zero. + +1. Mark the retired text fields `reserved` in `crates/protos/src/node.proto` (never reuse + the field numbers) and delete the JSON read-fallback branches; codec becomes msgpack-only. +2. This is a hard wire-format change — it requires the mixed-version upgrade rehearsal + (four-node scripts) to pass, and it cannot be rolled back by env alone. + +## Rollback matrix + +| Stage | Wire-format broken? | Rollback | +|---|---|---| +| 0 Observe | no | n/a (metric only) | +| 1 msgpack-only send | no | `RUSTFS_INTERNODE_RPC_MSGPACK_ONLY=false` + restart | +| 2 remove fields | yes | redeploy prior release; field numbers stay `reserved` | + +## Related + +- Codec + counter implementation: commit `feat(internode): P2 msgpack/JSON codec observability + encode buffer presizing`. +- Decoders: `decode_msgpack_or_json` in `crates/ecstore/src/cluster/rpc/remote_disk.rs` (client) + and `rustfs/src/storage/rpc/node_service/disk.rs` (server). diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index f87e2eee9..0b1852c51 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -69,6 +69,7 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::task::{Context, Poll}; use std::time::Duration; use tokio::net::{TcpListener, TcpStream}; +use tonic::service::interceptor::InterceptedService; use tonic::{Request, Status}; use tower::{Service, ServiceBuilder}; use tower_http::add_extension::AddExtensionLayer; @@ -1047,7 +1048,19 @@ fn process_connection( // Note: NodeService is not Clone (holds LocalPeerS3Client), and the SwiftService // type is feature-gated, so we cannot pre-build the full hybrid service. // The construction cost is negligible (struct wrapping only, no I/O). - let rpc_service = NodeServiceServer::with_interceptor(make_server(), check_auth); + // Align the server codec limit with the client (both default to + // `DEFAULT_GRPC_SERVER_MESSAGE_LEN`, 100 MiB) so `bytes`-carrying unary RPCs are not + // capped by tonic's 4 MiB default. Env-overridable via RUSTFS_INTERNODE_RPC_MAX_MESSAGE_SIZE. + // The codec size limits live on `NodeServiceServer`, so set them before wrapping the + // service in the auth interceptor (which returns an `InterceptedService` without those + // methods). + let rpc_max_message_size = rustfs_protos::internode_rpc_max_message_size(); + let rpc_service = InterceptedService::new( + NodeServiceServer::new(make_server()) + .max_decoding_message_size(rpc_max_message_size) + .max_encoding_message_size(rpc_max_message_size), + check_auth, + ); #[cfg(feature = "swift")] let http_service = SwiftService::new(true, None, s3_service); diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index d7ed95fff..aafdedd3d 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/rustfs/src/storage/rpc/node_service.rs @@ -1902,6 +1902,7 @@ mod tests { file_info: "{}".to_string(), force_del_marker: false, opts: "{}".to_string(), + ..Default::default() }); let response = service.delete_version(request).await; @@ -1923,6 +1924,7 @@ mod tests { file_info: "invalid json".to_string(), force_del_marker: false, opts: "{}".to_string(), + ..Default::default() }); let response = service.delete_version(request).await; @@ -1944,6 +1946,7 @@ mod tests { file_info: "{}".to_string(), force_del_marker: false, opts: "invalid json".to_string(), + ..Default::default() }); let response = service.delete_version(request).await; @@ -1963,6 +1966,7 @@ mod tests { volume: "test-volume".to_string(), versions: vec!["{}".to_string()], opts: "{}".to_string(), + ..Default::default() }); let response = service.delete_versions(request).await; @@ -1982,6 +1986,7 @@ mod tests { volume: "test-volume".to_string(), versions: vec!["invalid json".to_string()], opts: "{}".to_string(), + ..Default::default() }); let response = service.delete_versions(request).await; @@ -2001,6 +2006,7 @@ mod tests { volume: "test-volume".to_string(), versions: vec!["{}".to_string()], opts: "invalid json".to_string(), + ..Default::default() }); let response = service.delete_versions(request).await; diff --git a/rustfs/src/storage/rpc/node_service/disk.rs b/rustfs/src/storage/rpc/node_service/disk.rs index b6da6e29b..de37815dc 100644 --- a/rustfs/src/storage/rpc/node_service/disk.rs +++ b/rustfs/src/storage/rpc/node_service/disk.rs @@ -21,7 +21,8 @@ use crate::storage::storage_api::runtime_sources_consumer::runtime_sources; use bytes::Bytes; use rustfs_filemeta::FileInfo; use rustfs_io_metrics::internode_metrics::{ - INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC, + INTERNODE_MSGPACK_DIRECTION_REQUEST, INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, global_internode_metrics, }; use rustfs_protos::proto_gen::node_service::*; use serde::de::DeserializeOwned; @@ -29,18 +30,29 @@ use std::io::Cursor; use tonic::{Request, Response, Status}; use tracing::debug; -fn decode_msgpack_or_json(binary: &[u8], json: &str, value_name: &str) -> std::result::Result { +/// Initial capacity hint (bytes) for msgpack encode buffers, sized to cover a typical single- +/// version `FileInfo` without repeated growth reallocations. Larger payloads still grow as needed. +const MSGPACK_ENCODE_CAPACITY_HINT: usize = 512; + +fn decode_msgpack_or_json( + binary: &[u8], + json: &str, + value_name: &'static str, +) -> std::result::Result { if !binary.is_empty() { let mut deserializer = rmp_serde::Deserializer::new(Cursor::new(binary)); return T::deserialize(&mut deserializer) .map_err(|err| DiskError::other(format!("decode {value_name} msgpack failed: {err}"))); } + // The msgpack payload was absent, so fall back to the JSON compatibility field. This branch + // must read zero across a release window before the redundant JSON fields can be dropped (P2). + global_internode_metrics().record_msgpack_json_fallback(INTERNODE_MSGPACK_DIRECTION_REQUEST, value_name); serde_json::from_str(json).map_err(|err| DiskError::other(format!("decode {value_name} failed: {err}"))) } fn encode_msgpack(value: &T, value_name: &str) -> std::result::Result, DiskError> { - let mut serializer = rmp_serde::Serializer::new(Vec::new()); + let mut serializer = rmp_serde::Serializer::new(Vec::with_capacity(MSGPACK_ENCODE_CAPACITY_HINT)); value .serialize(&mut serializer) .map_err(|err| DiskError::other(format!("encode {value_name} msgpack failed: {err}")))?; @@ -48,13 +60,23 @@ fn encode_msgpack(value: &T, value_name: &str) -> std::resu } fn encode_msgpack_named(value: &T, value_name: &str) -> std::result::Result, DiskError> { - let mut serializer = rmp_serde::Serializer::new(Vec::new()).with_struct_map(); + let mut serializer = rmp_serde::Serializer::new(Vec::with_capacity(MSGPACK_ENCODE_CAPACITY_HINT)).with_struct_map(); value .serialize(&mut serializer) .map_err(|err| DiskError::other(format!("encode {value_name} named msgpack failed: {err}")))?; Ok(serializer.into_inner()) } +/// JSON compatibility string for a dual-encoded response field. Returns an empty string when +/// msgpack-only mode is enabled (grpc-optimization P2-1) so the redundant JSON copy is not sent; +/// otherwise the legacy JSON encoding. The paired `_bin` (msgpack) field is always sent. +fn compat_response_json(value: &T) -> std::result::Result { + if rustfs_protos::internode_rpc_msgpack_only() { + return Ok(String::new()); + } + serde_json::to_string(value) +} + fn encode_read_multiple_response_payloads( read_multiple_resps: &[ReadMultipleResp], ) -> std::result::Result<(Vec, Vec), DiskError> { @@ -63,7 +85,7 @@ fn encode_read_multiple_response_payloads( for read_multiple_resp in read_multiple_resps { read_multiple_resps_json.push( - serde_json::to_string(read_multiple_resp) + compat_response_json(read_multiple_resp) .map_err(|err| DiskError::other(format!("encode ReadMultipleResp json failed: {err}")))?, ); read_multiple_resps_bin.push(Bytes::from(encode_msgpack(read_multiple_resp, "ReadMultipleResp")?)); @@ -80,7 +102,7 @@ fn encode_batch_read_version_response_payloads( for batch_read_version_resp in batch_read_version_resps { batch_read_version_resps_json.push( - serde_json::to_string(batch_read_version_resp) + compat_response_json(batch_read_version_resp) .map_err(|err| DiskError::other(format!("encode BatchReadVersionResp json failed: {err}")))?, ); batch_read_version_resps_bin.push(Bytes::from(encode_msgpack(batch_read_version_resp, "BatchReadVersionResp")?)); @@ -292,8 +314,9 @@ impl NodeService { let request = request.into_inner(); if let Some(disk) = self.find_disk(&request.disk).await { let mut versions = Vec::with_capacity(request.versions.len()); - for version in request.versions.iter() { - match serde_json::from_str::(version) { + for (index, version) in request.versions.iter().enumerate() { + let version_bin = request.versions_bin.get(index).map(|b| b.as_ref()).unwrap_or(&[]); + match decode_msgpack_or_json::(version_bin, version, "FileInfoVersions") { Ok(version) => versions.push(version), Err(err) => { return Ok(Response::new(DeleteVersionsResponse { @@ -304,7 +327,7 @@ impl NodeService { } }; } - let opts = match serde_json::from_str::(&request.opts) { + let opts = match decode_msgpack_or_json::(&request.opts_bin, &request.opts, "DeleteOptions") { Ok(opts) => opts, Err(err) => { return Ok(Response::new(DeleteVersionsResponse { @@ -345,7 +368,7 @@ impl NodeService { ) -> Result, Status> { let request = request.into_inner(); if let Some(disk) = self.find_disk(&request.disk).await { - let file_info = match serde_json::from_str::(&request.file_info) { + let file_info = match decode_msgpack_or_json::(&request.file_info_bin, &request.file_info, "FileInfo") { Ok(file_info) => file_info, Err(err) => { return Ok(Response::new(DeleteVersionResponse { @@ -355,7 +378,7 @@ impl NodeService { })); } }; - let opts = match serde_json::from_str::(&request.opts) { + let opts = match decode_msgpack_or_json::(&request.opts_bin, &request.opts, "DeleteOptions") { Ok(opts) => opts, Err(err) => { return Ok(Response::new(DeleteVersionResponse { @@ -401,7 +424,7 @@ impl NodeService { if let Some(disk) = self.find_disk(&request.disk).await { match disk.read_xl(&request.volume, &request.path, request.read_data).await { Ok(raw_file_info) => { - let raw_file_info_json = serde_json::to_string(&raw_file_info); + let raw_file_info_json = compat_response_json(&raw_file_info); let raw_file_info_bin = encode_msgpack(&raw_file_info, "RawFileInfo"); match (raw_file_info_json, raw_file_info_bin) { (Ok(raw_file_info), Ok(raw_file_info_bin)) => Ok(Response::new(ReadXlResponse { @@ -463,7 +486,7 @@ impl NodeService { .await { Ok(file_info) => { - let file_info_json = serde_json::to_string(&file_info); + let file_info_json = compat_response_json(&file_info); let file_info_bin = encode_msgpack(&file_info, "FileInfo"); match (file_info_json, file_info_bin) { (Ok(file_info), Ok(file_info_bin)) => Ok(Response::new(ReadVersionResponse { @@ -768,7 +791,7 @@ impl NodeService { .await { Ok(rename_data_resp) => { - let rename_data_resp_json = serde_json::to_string(&rename_data_resp); + let rename_data_resp_json = compat_response_json(&rename_data_resp); let rename_data_resp_bin = encode_msgpack_named(&rename_data_resp, "RenameDataResp"); match (rename_data_resp_json, rename_data_resp_bin) { (Ok(rename_data_resp), Ok(rename_data_resp_bin)) => Ok(Response::new(RenameDataResponse { diff --git a/scripts/run_internode_grpc_ab_bench.sh b/scripts/run_internode_grpc_ab_bench.sh new file mode 100755 index 000000000..7325797a1 --- /dev/null +++ b/scripts/run_internode_grpc_ab_bench.sh @@ -0,0 +1,164 @@ +#!/usr/bin/env bash +set -euo pipefail + +# One-click A/B benchmark driver for the internode gRPC optimization stages (P0-P3). +# +# For a given --stage and --phase it: +# 1. computes the RUSTFS_INTERNODE_* *server* env for that stage/phase (per the runbook matrix), +# 2. writes it to /server-env.sh and prints it, +# 3. runs the appropriate underlying bench into a labeled . +# +# IMPORTANT: RUSTFS_INTERNODE_* are SERVER env. The load-driven stages (p0/p1/p2) benchmark an +# ALREADY-RUNNING cluster, so you must (re)start rustfs on every node with the emitted env BEFORE +# running the "after" (and the baseline "before"). This script cannot change a running server. +# The p3 (docker four-node) path exports the env so a compose that forwards it can pick it up. +# +# See docs/operations/internode-grpc-benchmark-runbook.md for metrics and acceptance gates. + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +PROJECT_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)" +TRANSPORT_BENCH="${PROJECT_ROOT}/scripts/run_internode_transport_baseline.sh" +FAILOVER_BENCH="${PROJECT_ROOT}/scripts/run_four_node_cluster_failover_bench.sh" + +STAGE="" +PHASE="" +OUT_ROOT="${PROJECT_ROOT}/target/bench/internode-transport" +DRY_RUN=false +PASSTHROUGH=() + +usage() { + cat <<'USAGE' +Usage: + scripts/run_internode_grpc_ab_bench.sh --stage --phase [options] [-- ] + +Required: + --stage Which optimization stage to A/B. + --phase baseline (feature off) or enabled (feature on). + +Options: + --out-root Artifact root. Default: target/bench/internode-transport + --dry-run Print the env + the command, do not run the bench. + -h, --help Show this help. + +Everything after `--` is passed through to the underlying bench: + p0/p1/p2 -> run_internode_transport_baseline.sh (e.g. --access-key AK --secret-key SK --metrics-url URL) + p3 -> run_four_node_cluster_failover_bench.sh + +Examples: + # P1 baseline then enabled (restart the cluster with the emitted env between the two): + scripts/run_internode_grpc_ab_bench.sh --stage p1 --phase before -- --access-key AK --secret-key SK + scripts/run_internode_grpc_ab_bench.sh --stage p1 --phase after -- --access-key AK --secret-key SK + + # P3 failover A/B (docker four-node): + scripts/run_internode_grpc_ab_bench.sh --stage p3 --phase after +USAGE +} + +while [[ $# -gt 0 ]]; do + case "$1" in + --stage) STAGE="$2"; shift 2 ;; + --phase) PHASE="$2"; shift 2 ;; + --out-root) OUT_ROOT="$2"; shift 2 ;; + --dry-run) DRY_RUN=true; shift ;; + -h|--help) usage; exit 0 ;; + --) shift; PASSTHROUGH=("$@"); break ;; + *) echo "Unknown argument: $1" >&2; usage; exit 2 ;; + esac +done + +[[ -n "${STAGE}" && -n "${PHASE}" ]] || { echo "ERROR: --stage and --phase are required" >&2; usage; exit 2; } +case "${PHASE}" in before|after) ;; *) echo "ERROR: --phase must be before|after" >&2; exit 2 ;; esac + +# Emit the RUSTFS_INTERNODE_* env for the stage/phase as `KEY=VALUE` lines. Empty output means +# "cluster defaults" (nothing to set). "before" = feature disabled, "after" = feature enabled. +stage_env() { + local stage="$1" phase="$2" + case "${stage}" in + p0) + if [[ "${phase}" == "before" ]]; then + printf '%s\n' \ + 'RUSTFS_INTERNODE_RPC_TCP_NODELAY=false' \ + 'RUSTFS_INTERNODE_RPC_HTTP2_STREAM_WINDOW_SIZE=0' \ + 'RUSTFS_INTERNODE_RPC_HTTP2_CONN_WINDOW_SIZE=0' \ + 'RUSTFS_INTERNODE_RPC_MAX_MESSAGE_SIZE=4194304' + fi + # after: P0 defaults are the enabled state -> no overrides needed. + ;; + p1) + if [[ "${phase}" == "after" ]]; then + printf '%s\n' \ + 'RUSTFS_INTERNODE_CHANNEL_ISOLATION=true' \ + 'RUSTFS_INTERNODE_BULK_CHANNELS=2' + else + printf '%s\n' 'RUSTFS_INTERNODE_CHANNEL_ISOLATION=false' + fi + ;; + p2) + if [[ "${phase}" == "after" ]]; then + # Only enable after the msgpack json-fallback counter has read zero across a window. + printf '%s\n' 'RUSTFS_INTERNODE_RPC_MSGPACK_ONLY=true' + else + printf '%s\n' 'RUSTFS_INTERNODE_RPC_MSGPACK_ONLY=false' + fi + ;; + p3) + if [[ "${phase}" == "after" ]]; then + printf '%s\n' \ + 'RUSTFS_INTERNODE_PREWARM=true' \ + 'RUSTFS_INTERNODE_OFFLINE_BYPASS=true' \ + 'RUSTFS_INTERNODE_OFFLINE_REPROBE_SECS=5' \ + 'RUSTFS_INTERNODE_OFFLINE_FAILURE_THRESHOLD=3' + else + printf '%s\n' \ + 'RUSTFS_INTERNODE_PREWARM=false' \ + 'RUSTFS_INTERNODE_OFFLINE_BYPASS=false' + fi + ;; + *) echo "ERROR: unknown --stage '${stage}' (expected p0|p1|p2|p3)" >&2; exit 2 ;; + esac +} + +ENV_LINES="$(stage_env "${STAGE}" "${PHASE}")" +OUT_DIR="${OUT_ROOT}/${STAGE}-${PHASE}" +mkdir -p "${OUT_DIR}" +ENV_FILE="${OUT_DIR}/server-env.sh" + +{ + echo "# RustFS server env for internode stage ${STAGE} / phase ${PHASE}." + echo "# Source this on EVERY rustfs node before starting the server, then run the bench." + if [[ -n "${ENV_LINES}" ]]; then + while IFS= read -r line; do echo "export ${line}"; done <<<"${ENV_LINES}" + else + echo "# (cluster defaults — no overrides needed for this stage/phase)" + fi +} >"${ENV_FILE}" + +echo "== internode gRPC A/B: stage=${STAGE} phase=${PHASE} ==" +echo "-- server env (also written to ${ENV_FILE}):" +sed 's/^/ /' "${ENV_FILE}" +echo "-- artifacts: ${OUT_DIR}" + +# Export for the p3 docker path (best-effort: only takes effect if the compose forwards the env). +if [[ -n "${ENV_LINES}" ]]; then + while IFS= read -r line; do export "${line?}"; done <<<"${ENV_LINES}" +fi + +run() { + if [[ "${DRY_RUN}" == "true" ]]; then + echo "-- dry-run, would execute:"; printf ' %q ' "$@"; echo + return 0 + fi + "$@" +} + +case "${STAGE}" in + p0|p1|p2) + echo "-- NOTE: (re)start rustfs on all nodes with the env above before this measurement is valid." + [[ -x "${TRANSPORT_BENCH}" ]] || { echo "ERROR: ${TRANSPORT_BENCH} not found/executable" >&2; exit 1; } + run "${TRANSPORT_BENCH}" --out-dir "${OUT_DIR}" "${PASSTHROUGH[@]}" + ;; + p3) + [[ -x "${FAILOVER_BENCH}" ]] || { echo "ERROR: ${FAILOVER_BENCH} not found/executable" >&2; exit 1; } + OUT_DIR="${OUT_DIR}" run "${FAILOVER_BENCH}" "${PASSTHROUGH[@]}" + ;; +esac