gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 16:28:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(internode): optimize gRPC transport (#4337)

Browse Source 
* 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 <heihutu@gmail.com>

* 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 <heihutu@gmail.com>

* 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 <heihutu@gmail.com>

* 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 <heihutu@gmail.com>

* 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 <heihutu@gmail.com>

* 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 <heihutu@gmail.com>

* 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 <heihutu@gmail.com>

* 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 <heihutu@gmail.com>

* 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 <heihutu@gmail.com>

* 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 <p0|p1|p2|p3> and --phase
<before|after>, it emits the stage/phase RUSTFS_INTERNODE_* server env to
<out-dir>/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/<stage>-<phase>/.

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 <heihutu@gmail.com>

* 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 <heihutu@gmail.com>

* 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 <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>
This commit is contained in:
houseme
2026-07-07 05:18:31 +08:00
committed by GitHub
parent 8f11222a63
commit 6f613317f6
17 changed files with 1571 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.docker/compose/docker-compose.cluster.local-build.yml
+56
View File
@@ -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:
crates/config/src/constants/internode.rs
+199
View File
@@ -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");
crates/ecstore/src/cluster/rpc/client.rs
+32 -13
View File
@@ -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<NodeServiceClient<InterceptedService<Channel, TonicInterceptor>>, Box<dyn Error>> {
// 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<NodeServiceClient<InterceptedService<Channel, TonicInterceptor>>, Box<dyn Error>> {
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(
crates/ecstore/src/cluster/rpc/remote_disk.rs
+291 -21
View File
@@ -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<dyn InternodeDataTransport>,
}
// ── 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<String> {
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::Mutex<std::collections::HashSet<String>>> =
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<T, F, Fut>(&self, op: &'static str, operation: F, timeout_duration: Duration) -> Result<T>
where
F: Fn() -> Fut,
Fut: std::future::Future<Output = Result<T>>,
{
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<T, F, Fut>(
&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<Error> {
internode_offline_bypass_reason(&self.addr).map(Error::other)
}
async fn get_client(&self) -> Result<NodeServiceClient<InterceptedService<Channel, TonicInterceptor>>> {
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<NodeServiceClient<InterceptedService<Channel, TonicInterceptor>>> {
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<T: Serialize>(value: &T) -> Result<Vec<u8>> {
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<T: Serialize>(value: &T) -> Result<String> {
if rustfs_protos::internode_rpc_msgpack_only() {
return Ok(String::new());
}
Ok(serde_json::to_string(value)?)
}
fn encode_msgpack_named<T: Serialize>(value: &T) -> Result<Vec<u8>> {
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<T: DeserializeOwned>(binary: &[u8], json: &str) -> Result<T> {
fn decode_msgpack_or_json<T: DeserializeOwned>(binary: &[u8], json: &str, value_name: &'static str) -> Result<T> {
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<Vec<ReadMultipleResp>> {
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::<ReadMultipleResp>(buf, "").map_err(|err| {
let resp = decode_msgpack_or_json::<ReadMultipleResp>(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::<ReadMultipleResp>(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::<BatchReadVersionResp>(buf, "").map_err(|err| {
let resp = decode_msgpack_or_json::<BatchReadVersionResp>(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::<BatchReadVersionResp>(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::<FileInfo>(&response.file_info_bin, &response.file_info)?;
let file_info = decode_msgpack_or_json::<FileInfo>(&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::<RawFileInfo>(&response.raw_file_info_bin, &response.raw_file_info)?;
let raw_file_info =
decode_msgpack_or_json::<RawFileInfo>(&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::<RenameDataResp>(&response.rename_data_resp_bin, &response.rename_data_resp)?;
let rename_data_resp = decode_msgpack_or_json::<RenameDataResp>(
&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<DiskInfo> {
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();
crates/ecstore/src/cluster/rpc/remote_locker.rs
+6
View File
@@ -77,6 +77,12 @@ impl RemoteClient {
}
pub async fn get_client(&self) -> Result<NodeServiceClient<InterceptedService<Channel, TonicInterceptor>>> {
// 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}")))
crates/ecstore/src/cluster/rpc/runtime_sources.rs
+40 -2
View File
@@ -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)]
crates/io-metrics/src/internode_metrics.rs
+244 -3
View File
@@ -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<InternodeMetrics> {
&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<Instant>,
}
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<Mutex<HashMap<String, PeerHealthState>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
fn publish_offline_gauge(peers: &HashMap<String, PeerHealthState>) {
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<bool> {
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]
crates/protos/src/generated/proto_gen/node_service.rs
+8
View File
@@ -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 {
crates/protos/src/lib.rs
+213 -12
View File
@@ -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<u32> {
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<u32> {
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<Channel, Box<dyn Error>> {
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<Channel, Box<dyn Error>> {
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<Channel, Box<dyn Error>> {
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<Channel, Box<dyn Error>> {
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<Channel, Box<dyn Error>> {
// 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<Channel, Box<dyn Error>> {
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<Channel, Box<dyn Error>> {
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;
}
}
crates/protos/src/node.proto
+8
View File
@@ -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 {
crates/protos/src/runtime_sources.rs
+11
View File
@@ -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);
}
docs/operations/internode-grpc-benchmark-runbook.md
+97
View File
@@ -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 <p0|p1|p2|p3> --phase <before|after> [-- <bench args>]`
wraps the env matrix below: it writes the stage/phase **server** env to
`<out-dir>/server-env.sh`, then runs the right underlying bench into
`target/bench/internode-transport/<stage>-<phase>/`.
```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-<ts>/`). Pass
`--metrics-url <prometheus>` 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-<ts>/`).
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.
docs/operations/internode-msgpack-json-convergence-runbook.md
+145
View File
@@ -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).
rustfs/src/server/http.rs
+14 -1
View File
@@ -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);
rustfs/src/storage/rpc/node_service.rs
+6
View File
@@ -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;
rustfs/src/storage/rpc/node_service/disk.rs
+37 -14
View File
@@ -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<T: DeserializeOwned>(binary: &[u8], json: &str, value_name: &str) -> std::result::Result<T, DiskError> {
/// 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<T: DeserializeOwned>(
binary: &[u8],
json: &str,
value_name: &'static str,
) -> std::result::Result<T, DiskError> {
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<T: serde::Serialize>(value: &T, value_name: &str) -> std::result::Result<Vec<u8>, 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<T: serde::Serialize>(value: &T, value_name: &str) -> std::resu
}
fn encode_msgpack_named<T: serde::Serialize>(value: &T, value_name: &str) -> std::result::Result<Vec<u8>, 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<T: serde::Serialize>(value: &T) -> std::result::Result<String, serde_json::Error> {
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<String>, Vec<Bytes>), 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::<FileInfoVersions>(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::<FileInfoVersions>(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::<DeleteOptions>(&request.opts) {
let opts = match decode_msgpack_or_json::<DeleteOptions>(&request.opts_bin, &request.opts, "DeleteOptions") {
Ok(opts) => opts,
Err(err) => {
return Ok(Response::new(DeleteVersionsResponse {
@@ -345,7 +368,7 @@ impl NodeService {
) -> Result<Response<DeleteVersionResponse>, Status> {
let request = request.into_inner();
if let Some(disk) = self.find_disk(&request.disk).await {
let file_info = match serde_json::from_str::<FileInfo>(&request.file_info) {
let file_info = match decode_msgpack_or_json::<FileInfo>(&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::<DeleteOptions>(&request.opts) {
let opts = match decode_msgpack_or_json::<DeleteOptions>(&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 {
scripts/run_internode_grpc_ab_bench.sh
Executable
+164
View File
@@ -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 <out-dir>/server-env.sh and prints it,
# 3. runs the appropriate underlying bench into a labeled <out-dir>.
#
# 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 <p0|p1|p2|p3> --phase <before|after> [options] [-- <bench args>]
Required:
--stage <p0|p1|p2|p3> Which optimization stage to A/B.
--phase <before|after> baseline (feature off) or enabled (feature on).
Options:
--out-root <dir> 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