Compare commits

...

3 Commits

Author SHA1 Message Date
houseme a4b9c13114 test: add formal hotpath ABBA runner
Co-Authored-By: heihutu <heihutu@gmail.com>
2026-07-31 16:20:03 +08:00
houseme becad7f6b4 feat: extend hotpath coverage across crates
Add opt-in hotpath feature surfaces to every workspace crate and wire the root rustfs feature passthrough for function, allocation, and CPU profiling.

Add a focused set of function-level measurements for scanner, heal, lock, target replay, IAM, KMS, Keystone, trusted proxy, and capacity paths without adding request-scoped primitive wrappers.

Co-Authored-By: heihutu <heihutu@gmail.com>
2026-07-31 12:45:40 +08:00
Zhengchao An dd11145a26 feat(kms): record operation metrics in the retry policy engine (#5500)
* feat(kms): record operation metrics in the retry policy engine

Instrument policy::execute — the single choke point every outbound Vault
call and credential exchange already flows through — so no call site
needs its own instrumentation:

- rustfs_kms_backend_operations_total (counter): operation, op_class,
  outcome (success / fatal / budget_exhausted / deadline_exceeded /
  cancelled)
- rustfs_kms_backend_attempt_failures_total (counter): operation,
  error_class (retryable_conn / retryable_status / fatal /
  attempt_timeout)
- rustfs_kms_backend_operation_duration_seconds (histogram): wall-clock
  duration including retries and backoff
- rustfs_kms_backend_operation_attempts (histogram): attempts used

Metric labels carry only static enum values (operation names, classes,
outcomes) — never key identifiers, key material, ciphertext, or tokens.
Emission goes through the process-global metrics facade recorder, the
same pattern the rest of the workspace uses, so no new wiring is needed
in rustfs/src.

Tests drive a paused-clock runtime under a thread-local debugging
recorder, so counts, attempts, and even the recorded (virtual-clock)
durations are asserted deterministically with zero real sleeps.

Refs rustfs/backlog#1569 (part of rustfs/backlog#1562)

* test(kms): add Vault fault-injection matrix

Offline cases inject transport faults locally and are fully
deterministic: a refused connection is retried up to the configured
budget, and a stalled connection is cut off by the per-attempt timeout
instead of hanging. Ignored cases run against a real dev Vault
(RUSTFS_KMS_VAULT_ADDR) and pin the fail-closed auth behavior: an
invalid token and a missing key each resolve in exactly one attempt.

Every case asserts through the policy metrics recorded by a
thread-local debugging recorder, which doubles as the request-count
assertion even against a real server. Throttling and recoverable 5xx
responses cannot be forced on a stock dev Vault; those paths stay
pinned by the scripted-Vault wiring tests and the engine tests.

Refs rustfs/backlog#1569 (part of rustfs/backlog#1562)
2026-07-31 02:56:44 +00:00
64 changed files with 2268 additions and 31 deletions
Generated
+44
View File
@@ -3672,6 +3672,7 @@ dependencies = [
"flate2", "flate2",
"futures", "futures",
"hex", "hex",
"hotpath",
"http 1.5.0", "http 1.5.0",
"http-body-util", "http-body-util",
"hyper", "hyper",
@@ -9039,6 +9040,7 @@ dependencies = [
"const-str", "const-str",
"futures", "futures",
"hashbrown 0.17.1", "hashbrown 0.17.1",
"hotpath",
"metrics", "metrics",
"rustfs-config", "rustfs-config",
"rustfs-s3-types", "rustfs-s3-types",
@@ -9059,6 +9061,7 @@ dependencies = [
"base64-simd", "base64-simd",
"bytes", "bytes",
"crc-fast", "crc-fast",
"hotpath",
"http 1.5.0", "http 1.5.0",
"md-5 0.11.0", "md-5 0.11.0",
"pretty_assertions", "pretty_assertions",
@@ -9072,6 +9075,7 @@ name = "rustfs-common"
version = "1.0.0-beta.12" version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"chrono", "chrono",
"hotpath",
"metrics", "metrics",
"rmp-serde", "rmp-serde",
"s3s", "s3s",
@@ -9086,6 +9090,7 @@ dependencies = [
name = "rustfs-concurrency" name = "rustfs-concurrency"
version = "1.0.0-beta.12" version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"hotpath",
"insta", "insta",
"rustfs-io-core", "rustfs-io-core",
"serde", "serde",
@@ -9099,6 +9104,7 @@ name = "rustfs-config"
version = "1.0.0-beta.12" version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"const-str", "const-str",
"hotpath",
"serde", "serde",
"serde_json", "serde_json",
] ]
@@ -9109,6 +9115,7 @@ version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"base64-simd", "base64-simd",
"hmac 0.13.0", "hmac 0.13.0",
"hotpath",
"rand 0.10.2", "rand 0.10.2",
"serde", "serde",
"serde_json", "serde_json",
@@ -9124,6 +9131,7 @@ dependencies = [
"argon2", "argon2",
"base64-simd", "base64-simd",
"chacha20poly1305", "chacha20poly1305",
"hotpath",
"jsonwebtoken 11.0.0", "jsonwebtoken 11.0.0",
"pbkdf2 0.13.0", "pbkdf2 0.13.0",
"rand 0.10.2", "rand 0.10.2",
@@ -9141,6 +9149,7 @@ name = "rustfs-data-usage"
version = "1.0.0-beta.12" version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"async-trait", "async-trait",
"hotpath",
"rmp-serde", "rmp-serde",
"rustfs-filemeta", "rustfs-filemeta",
"serde", "serde",
@@ -9286,6 +9295,7 @@ dependencies = [
name = "rustfs-extension-schema" name = "rustfs-extension-schema"
version = "1.0.0-beta.12" version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"hotpath",
"serde", "serde",
"serde_json", "serde_json",
"thiserror 2.0.19", "thiserror 2.0.19",
@@ -9324,6 +9334,7 @@ dependencies = [
"async-trait", "async-trait",
"base64 0.23.0", "base64 0.23.0",
"futures", "futures",
"hotpath",
"http 1.5.0", "http 1.5.0",
"metrics", "metrics",
"rustfs-common", "rustfs-common",
@@ -9355,6 +9366,7 @@ dependencies = [
"async-trait", "async-trait",
"base64-simd", "base64-simd",
"futures", "futures",
"hotpath",
"http 1.5.0", "http 1.5.0",
"jsonwebtoken 11.0.0", "jsonwebtoken 11.0.0",
"moka", "moka",
@@ -9389,6 +9401,7 @@ name = "rustfs-io-core"
version = "1.0.0-beta.12" version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"bytes", "bytes",
"hotpath",
"memmap2", "memmap2",
"rustfs-io-metrics", "rustfs-io-metrics",
"thiserror 2.0.19", "thiserror 2.0.19",
@@ -9401,6 +9414,7 @@ name = "rustfs-io-metrics"
version = "1.0.0-beta.12" version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"criterion", "criterion",
"hotpath",
"metrics", "metrics",
"metrics-util", "metrics-util",
"num_cpus", "num_cpus",
@@ -9467,6 +9481,7 @@ version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"bytes", "bytes",
"futures", "futures",
"hotpath",
"http 1.5.0", "http 1.5.0",
"http-body 1.1.0", "http-body 1.1.0",
"http-body-util", "http-body-util",
@@ -9499,9 +9514,12 @@ dependencies = [
"base64 0.23.0", "base64 0.23.0",
"chacha20poly1305", "chacha20poly1305",
"hex", "hex",
"hotpath",
"insta", "insta",
"jiff", "jiff",
"md-5 0.11.0", "md-5 0.11.0",
"metrics",
"metrics-util",
"moka", "moka",
"rand 0.10.2", "rand 0.10.2",
"reqwest", "reqwest",
@@ -9529,6 +9547,7 @@ name = "rustfs-lifecycle"
version = "1.0.0-beta.12" version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"async-trait", "async-trait",
"hotpath",
"metrics", "metrics",
"metrics-util", "metrics-util",
"proptest", "proptest",
@@ -9553,6 +9572,7 @@ dependencies = [
"async-trait", "async-trait",
"crossbeam-queue", "crossbeam-queue",
"futures", "futures",
"hotpath",
"parking_lot", "parking_lot",
"rand 0.10.2", "rand 0.10.2",
"rustfs-io-metrics", "rustfs-io-metrics",
@@ -9574,6 +9594,7 @@ version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"chrono", "chrono",
"flate2", "flate2",
"hotpath",
"regex", "regex",
"serde", "serde",
"serde_json", "serde_json",
@@ -9591,6 +9612,7 @@ name = "rustfs-madmin"
version = "1.0.0-beta.12" version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"chrono", "chrono",
"hotpath",
"humantime", "humantime",
"hyper", "hyper",
"rmp-serde", "rmp-serde",
@@ -9611,6 +9633,7 @@ dependencies = [
"criterion", "criterion",
"form_urlencoded", "form_urlencoded",
"hashbrown 0.17.1", "hashbrown 0.17.1",
"hotpath",
"metrics", "metrics",
"percent-encoding", "percent-encoding",
"quick-xml", "quick-xml",
@@ -9640,6 +9663,7 @@ version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"criterion", "criterion",
"futures", "futures",
"hotpath",
"rustfs-config", "rustfs-config",
"rustfs-io-metrics", "rustfs-io-metrics",
"rustfs-utils", "rustfs-utils",
@@ -9658,6 +9682,7 @@ version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"bytes", "bytes",
"criterion", "criterion",
"hotpath",
"metrics", "metrics",
"metrics-util", "metrics-util",
"moka", "moka",
@@ -9680,6 +9705,7 @@ dependencies = [
"flate2", "flate2",
"futures-util", "futures-util",
"glob", "glob",
"hotpath",
"jiff", "jiff",
"libc", "libc",
"metrics", "metrics",
@@ -9765,6 +9791,7 @@ dependencies = [
"futures-util", "futures-util",
"hex", "hex",
"hmac 0.13.0", "hmac 0.13.0",
"hotpath",
"http 1.5.0", "http 1.5.0",
"http-body-util", "http-body-util",
"hyper", "hyper",
@@ -9817,6 +9844,7 @@ name = "rustfs-protos"
version = "1.0.0-beta.12" version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"flatbuffers", "flatbuffers",
"hotpath",
"prost 0.14.4", "prost 0.14.4",
"rmp-serde", "rmp-serde",
"rustfs-common", "rustfs-common",
@@ -9841,6 +9869,7 @@ version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"byteorder", "byteorder",
"bytes", "bytes",
"hotpath",
"regex", "regex",
"rmp", "rmp",
"rmp-serde", "rmp-serde",
@@ -9899,6 +9928,7 @@ dependencies = [
"chacha20poly1305", "chacha20poly1305",
"hex", "hex",
"hmac 0.13.0", "hmac 0.13.0",
"hotpath",
"minlz", "minlz",
"pin-project-lite", "pin-project-lite",
"rand 0.10.2", "rand 0.10.2",
@@ -9916,6 +9946,7 @@ dependencies = [
name = "rustfs-s3-ops" name = "rustfs-s3-ops"
version = "1.0.0-beta.12" version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"hotpath",
"rustfs-s3-types", "rustfs-s3-types",
] ]
@@ -9923,6 +9954,7 @@ dependencies = [
name = "rustfs-s3-types" name = "rustfs-s3-types"
version = "1.0.0-beta.12" version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"hotpath",
"serde", "serde",
"serde_json", "serde_json",
] ]
@@ -9937,6 +9969,7 @@ dependencies = [
"datafusion", "datafusion",
"futures", "futures",
"futures-core", "futures-core",
"hotpath",
"http 1.5.0", "http 1.5.0",
"metrics", "metrics",
"parking_lot", "parking_lot",
@@ -9965,6 +9998,7 @@ dependencies = [
"datafusion", "datafusion",
"derive_builder", "derive_builder",
"futures", "futures",
"hotpath",
"parking_lot", "parking_lot",
"rustfs-s3select-api", "rustfs-s3select-api",
"s3s", "s3s",
@@ -9982,6 +10016,7 @@ dependencies = [
"futures", "futures",
"hex-simd", "hex-simd",
"hmac 0.13.0", "hmac 0.13.0",
"hotpath",
"http 1.5.0", "http 1.5.0",
"metrics", "metrics",
"rand 0.10.2", "rand 0.10.2",
@@ -10014,6 +10049,7 @@ dependencies = [
name = "rustfs-security-governance" name = "rustfs-security-governance"
version = "1.0.0-beta.12" version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"hotpath",
"thiserror 2.0.19", "thiserror 2.0.19",
] ]
@@ -10023,6 +10059,7 @@ version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"base64-simd", "base64-simd",
"bytes", "bytes",
"hotpath",
"http 1.5.0", "http 1.5.0",
"hyper", "hyper",
"rustfs-utils", "rustfs-utils",
@@ -10039,6 +10076,7 @@ name = "rustfs-storage-api"
version = "1.0.0-beta.12" version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"async-trait", "async-trait",
"hotpath",
"insta", "insta",
"rustfs-filemeta", "rustfs-filemeta",
"serde", "serde",
@@ -10060,6 +10098,7 @@ dependencies = [
"deadpool-postgres", "deadpool-postgres",
"futures-util", "futures-util",
"hashbrown 0.17.1", "hashbrown 0.17.1",
"hotpath",
"hyper", "hyper",
"hyper-rustls", "hyper-rustls",
"lapin", "lapin",
@@ -10105,6 +10144,7 @@ dependencies = [
name = "rustfs-test-utils" name = "rustfs-test-utils"
version = "1.0.0-beta.12" version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"hotpath",
"rustfs-data-usage", "rustfs-data-usage",
"rustfs-ecstore", "rustfs-ecstore",
"rustfs-storage-api", "rustfs-storage-api",
@@ -10121,6 +10161,7 @@ name = "rustfs-tls-runtime"
version = "1.0.0-beta.12" version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"arc-swap", "arc-swap",
"hotpath",
"metrics", "metrics",
"rcgen", "rcgen",
"rustfs-common", "rustfs-common",
@@ -10142,6 +10183,7 @@ version = "1.0.0-beta.12"
dependencies = [ dependencies = [
"async-trait", "async-trait",
"axum", "axum",
"hotpath",
"http 1.5.0", "http 1.5.0",
"ipnetwork", "ipnetwork",
"metrics", "metrics",
@@ -10188,6 +10230,7 @@ dependencies = [
"hex-simd", "hex-simd",
"highway", "highway",
"hmac 0.13.0", "hmac 0.13.0",
"hotpath",
"http 1.5.0", "http 1.5.0",
"hyper", "hyper",
"local-ip-address", "local-ip-address",
@@ -10220,6 +10263,7 @@ dependencies = [
"astral-tokio-tar", "astral-tokio-tar",
"async-compression", "async-compression",
"criterion", "criterion",
"hotpath",
"tempfile", "tempfile",
"thiserror 2.0.19", "thiserror 2.0.19",
"tokio", "tokio",
+26
View File
@@ -25,7 +25,33 @@ documentation = "https://docs.rs/rustfs-audit/latest/rustfs_audit/"
keywords = ["audit", "target", "management", "fan-out", "RustFS"] keywords = ["audit", "target", "management", "fan-out", "RustFS"]
categories = ["web-programming", "development-tools", "asynchronous", "api-bindings"] categories = ["web-programming", "development-tools", "asynchronous", "api-bindings"]
[features]
default = []
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
"hotpath/futures",
"rustfs-config/hotpath",
"rustfs-s3-types/hotpath",
"rustfs-targets/hotpath",
]
hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc",
"rustfs-config/hotpath-alloc",
"rustfs-s3-types/hotpath-alloc",
"rustfs-targets/hotpath-alloc",
]
hotpath-cpu = [
"hotpath",
"hotpath/hotpath-cpu",
"rustfs-config/hotpath-cpu",
"rustfs-s3-types/hotpath-cpu",
"rustfs-targets/hotpath-cpu",
]
[dependencies] [dependencies]
hotpath.workspace = true
rustfs-targets = { workspace = true } rustfs-targets = { workspace = true }
rustfs-config = { workspace = true, features = ["audit", "server-config-model"] } rustfs-config = { workspace = true, features = ["audit", "server-config-model"] }
rustfs-s3-types = { workspace = true } rustfs-s3-types = { workspace = true }
+7
View File
@@ -28,7 +28,14 @@ documentation = "https://docs.rs/rustfs-checksums/latest/rustfs_checksum/"
[lints] [lints]
workspace = true workspace = true
[features]
default = []
hotpath = ["hotpath/hotpath"]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true
bytes = { workspace = true, features = ["serde"] } bytes = { workspace = true, features = ["serde"] }
crc-fast = { workspace = true } crc-fast = { workspace = true }
http = { workspace = true } http = { workspace = true }
+7
View File
@@ -27,7 +27,14 @@ categories = ["web-programming", "development-tools", "data-structures"]
[lints] [lints]
workspace = true workspace = true
[features]
default = []
hotpath = ["hotpath/hotpath", "hotpath/tokio"]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true
tokio = { workspace = true, features = ["fs", "rt-multi-thread"] } tokio = { workspace = true, features = ["fs", "rt-multi-thread"] }
tonic = { workspace = true, features = ["gzip", "deflate"] } tonic = { workspace = true, features = ["gzip", "deflate"] }
uuid = { workspace = true, features = ["v4", "fast-rng", "macro-diagnostics"] } uuid = { workspace = true, features = ["v4", "fast-rng", "macro-diagnostics"] }
+7
View File
@@ -13,7 +13,14 @@ categories = ["concurrency", "filesystem"]
[lints] [lints]
workspace = true workspace = true
[features]
default = []
hotpath = ["hotpath/hotpath", "hotpath/tokio", "rustfs-io-core/hotpath"]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-io-core/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-io-core/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true
# Internal crates # Internal crates
rustfs-io-core = { workspace = true } rustfs-io-core = { workspace = true }
serde = { workspace = true, features = ["derive"] } serde = { workspace = true, features = ["derive"] }
+4
View File
@@ -25,6 +25,7 @@ keywords = ["configuration", "settings", "management", "rustfs", "Minio"]
categories = ["web-programming", "development-tools", "config"] categories = ["web-programming", "development-tools", "config"]
[dependencies] [dependencies]
hotpath.workspace = true
const-str = { workspace = true, optional = true, features = ["std", "proc"] } const-str = { workspace = true, optional = true, features = ["std", "proc"] }
serde = { workspace = true, optional = true, features = ["derive"] } serde = { workspace = true, optional = true, features = ["derive"] }
serde_json = { workspace = true, optional = true, features = ["raw_value"] } serde_json = { workspace = true, optional = true, features = ["raw_value"] }
@@ -34,6 +35,9 @@ workspace = true
[features] [features]
default = ["constants"] default = ["constants"]
hotpath = ["hotpath/hotpath"]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
audit = ["dep:const-str", "constants"] audit = ["dep:const-str", "constants"]
constants = ["dep:const-str"] constants = ["dep:const-str"]
notify = ["dep:const-str", "constants"] notify = ["dep:const-str", "constants"]
+7
View File
@@ -24,7 +24,14 @@ description = "Credentials management utilities for RustFS, enabling secure hand
keywords = ["rustfs", "Minio", "credentials", "authentication", "authorization"] keywords = ["rustfs", "Minio", "credentials", "authentication", "authorization"]
categories = ["web-programming", "development-tools", "data-structures", "security"] categories = ["web-programming", "development-tools", "data-structures", "security"]
[features]
default = []
hotpath = ["hotpath/hotpath"]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true
base64-simd = { workspace = true } base64-simd = { workspace = true }
hmac = { workspace = true } hmac = { workspace = true }
rand = { workspace = true, features = ["serde"] } rand = { workspace = true, features = ["serde"] }
+4
View File
@@ -29,6 +29,7 @@ documentation = "https://docs.rs/rustfs-crypto/latest/rustfs_crypto/"
workspace = true workspace = true
[dependencies] [dependencies]
hotpath.workspace = true
aes-gcm = { workspace = true, optional = true, features = ["rand_core"] } aes-gcm = { workspace = true, optional = true, features = ["rand_core"] }
argon2 = { workspace = true, optional = true } argon2 = { workspace = true, optional = true }
chacha20poly1305 = { workspace = true, optional = true } chacha20poly1305 = { workspace = true, optional = true }
@@ -49,6 +50,9 @@ time = { workspace = true, features = ["parsing", "formatting", "macros", "serde
[features] [features]
default = ["crypto", "fips"] default = ["crypto", "fips"]
hotpath = ["hotpath/hotpath"]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
fips = [] fips = []
crypto = [ crypto = [
"dep:aes-gcm", "dep:aes-gcm",
+7
View File
@@ -27,7 +27,14 @@ categories = ["data-structures", "filesystem"]
[lints] [lints]
workspace = true workspace = true
[features]
default = []
hotpath = ["hotpath/hotpath", "rustfs-filemeta/hotpath"]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-filemeta/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-filemeta/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true
serde = { workspace = true, features = ["derive"] } serde = { workspace = true, features = ["derive"] }
rmp-serde = { workspace = true } rmp-serde = { workspace = true }
async-trait = { workspace = true } async-trait = { workspace = true }
+48
View File
@@ -25,10 +25,58 @@ workspace = true
[features] [features]
default = [] default = []
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
"hotpath/futures",
"hotpath/reqwest-0-13",
"rustfs-config/hotpath",
"rustfs-credentials/hotpath",
"rustfs-data-usage/hotpath",
"rustfs-ecstore/hotpath",
"rustfs-filemeta/hotpath",
"rustfs-lock/hotpath",
"rustfs-madmin/hotpath",
"rustfs-protos/hotpath",
"rustfs-rio/hotpath",
"rustfs-signer/hotpath",
"rustfs-utils/hotpath",
]
hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc",
"rustfs-config/hotpath-alloc",
"rustfs-credentials/hotpath-alloc",
"rustfs-data-usage/hotpath-alloc",
"rustfs-ecstore/hotpath-alloc",
"rustfs-filemeta/hotpath-alloc",
"rustfs-lock/hotpath-alloc",
"rustfs-madmin/hotpath-alloc",
"rustfs-protos/hotpath-alloc",
"rustfs-rio/hotpath-alloc",
"rustfs-signer/hotpath-alloc",
"rustfs-utils/hotpath-alloc",
]
hotpath-cpu = [
"hotpath",
"hotpath/hotpath-cpu",
"rustfs-config/hotpath-cpu",
"rustfs-credentials/hotpath-cpu",
"rustfs-data-usage/hotpath-cpu",
"rustfs-ecstore/hotpath-cpu",
"rustfs-filemeta/hotpath-cpu",
"rustfs-lock/hotpath-cpu",
"rustfs-madmin/hotpath-cpu",
"rustfs-protos/hotpath-cpu",
"rustfs-rio/hotpath-cpu",
"rustfs-signer/hotpath-cpu",
"rustfs-utils/hotpath-cpu",
]
ftps = [] ftps = []
sftp = [] sftp = []
[dependencies] [dependencies]
hotpath.workspace = true
rustfs-config = { workspace = true, features = ["constants"] } rustfs-config = { workspace = true, features = ["constants"] }
rustfs-credentials.workspace = true rustfs-credentials.workspace = true
rustfs-ecstore.workspace = true rustfs-ecstore.workspace = true
+64
View File
@@ -40,19 +40,83 @@ hotpath = [
"hotpath/async-channel", "hotpath/async-channel",
"hotpath/parking_lot", "hotpath/parking_lot",
"hotpath/reqwest-0-13", "hotpath/reqwest-0-13",
"rustfs-checksums/hotpath",
"rustfs-common/hotpath",
"rustfs-concurrency/hotpath",
"rustfs-config/hotpath",
"rustfs-credentials/hotpath",
"rustfs-data-usage/hotpath",
"rustfs-filemeta/hotpath", "rustfs-filemeta/hotpath",
"rustfs-io-metrics/hotpath",
"rustfs-lifecycle/hotpath",
"rustfs-lock/hotpath",
"rustfs-madmin/hotpath",
"rustfs-object-capacity/hotpath",
"rustfs-policy/hotpath",
"rustfs-protos/hotpath",
"rustfs-replication/hotpath",
"rustfs-rio/hotpath", "rustfs-rio/hotpath",
"rustfs-rio-v2?/hotpath",
"rustfs-s3-types/hotpath",
"rustfs-signer/hotpath",
"rustfs-storage-api/hotpath",
"rustfs-tls-runtime/hotpath",
"rustfs-utils/hotpath",
"rustfs-crypto/hotpath",
] ]
hotpath-alloc = [ hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc", "hotpath/hotpath-alloc",
"rustfs-checksums/hotpath-alloc",
"rustfs-common/hotpath-alloc",
"rustfs-concurrency/hotpath-alloc",
"rustfs-config/hotpath-alloc",
"rustfs-credentials/hotpath-alloc",
"rustfs-data-usage/hotpath-alloc",
"rustfs-filemeta/hotpath-alloc", "rustfs-filemeta/hotpath-alloc",
"rustfs-io-metrics/hotpath-alloc",
"rustfs-lifecycle/hotpath-alloc",
"rustfs-lock/hotpath-alloc",
"rustfs-madmin/hotpath-alloc",
"rustfs-object-capacity/hotpath-alloc",
"rustfs-policy/hotpath-alloc",
"rustfs-protos/hotpath-alloc",
"rustfs-replication/hotpath-alloc",
"rustfs-rio/hotpath-alloc", "rustfs-rio/hotpath-alloc",
"rustfs-rio-v2?/hotpath-alloc",
"rustfs-s3-types/hotpath-alloc",
"rustfs-signer/hotpath-alloc",
"rustfs-storage-api/hotpath-alloc",
"rustfs-tls-runtime/hotpath-alloc",
"rustfs-utils/hotpath-alloc",
"rustfs-crypto/hotpath-alloc",
] ]
hotpath-cpu = [ hotpath-cpu = [
"hotpath", "hotpath",
"hotpath/hotpath-cpu", "hotpath/hotpath-cpu",
"rustfs-checksums/hotpath-cpu",
"rustfs-common/hotpath-cpu",
"rustfs-concurrency/hotpath-cpu",
"rustfs-config/hotpath-cpu",
"rustfs-credentials/hotpath-cpu",
"rustfs-data-usage/hotpath-cpu",
"rustfs-filemeta/hotpath-cpu", "rustfs-filemeta/hotpath-cpu",
"rustfs-io-metrics/hotpath-cpu",
"rustfs-lifecycle/hotpath-cpu",
"rustfs-lock/hotpath-cpu",
"rustfs-madmin/hotpath-cpu",
"rustfs-object-capacity/hotpath-cpu",
"rustfs-policy/hotpath-cpu",
"rustfs-protos/hotpath-cpu",
"rustfs-replication/hotpath-cpu",
"rustfs-rio/hotpath-cpu", "rustfs-rio/hotpath-cpu",
"rustfs-rio-v2?/hotpath-cpu",
"rustfs-s3-types/hotpath-cpu",
"rustfs-signer/hotpath-cpu",
"rustfs-storage-api/hotpath-cpu",
"rustfs-tls-runtime/hotpath-cpu",
"rustfs-utils/hotpath-cpu",
"rustfs-crypto/hotpath-cpu",
] ]
# Exposes shared lifecycle/tier test utilities (MockWarmBackend, fault # Exposes shared lifecycle/tier test utilities (MockWarmBackend, fault
# injection, xl.meta transition assertions) via `api::tier::test_util`. # injection, xl.meta transition assertions) via `api::tier::test_util`.
+2
View File
@@ -12,6 +12,8 @@
// See the License for the specific language governing permissions and // See the License for the specific language governing permissions and
// limitations under the License. // limitations under the License.
#![recursion_limit = "256"]
/// Scope-based hotpath measurement for `#[async_trait]` methods, where /// Scope-based hotpath measurement for `#[async_trait]` methods, where
/// `#[hotpath::measure]` would only time the boxed-future construction. /// `#[hotpath::measure]` would only time the boxed-future construction.
/// The guard records wall time from this statement until the enclosing /// The guard records wall time from this statement until the enclosing
+7
View File
@@ -27,7 +27,14 @@ categories = ["web-programming", "development-tools"]
[lib] [lib]
doctest = false doctest = false
[features]
default = []
hotpath = ["hotpath/hotpath"]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true
serde = { workspace = true, features = ["derive"] } serde = { workspace = true, features = ["derive"] }
thiserror.workspace = true thiserror.workspace = true
+3 -3
View File
@@ -27,9 +27,9 @@ documentation = "https://docs.rs/rustfs-filemeta/latest/rustfs_filemeta/"
[features] [features]
default = [] default = []
hotpath = ["hotpath/hotpath", "hotpath/tokio"] hotpath = ["hotpath/hotpath", "hotpath/tokio", "rustfs-utils/hotpath"]
hotpath-alloc = ["hotpath/hotpath-alloc"] hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-utils/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"] hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-utils/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true hotpath.workspace = true
+41
View File
@@ -29,7 +29,48 @@ categories = ["web-programming", "development-tools", "filesystem"]
[lints] [lints]
workspace = true workspace = true
[features]
default = []
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
"hotpath/futures",
"rustfs-common/hotpath",
"rustfs-concurrency/hotpath",
"rustfs-config/hotpath",
"rustfs-ecstore/hotpath",
"rustfs-madmin/hotpath",
"rustfs-storage-api/hotpath",
"rustfs-utils/hotpath",
"rustfs-test-utils/hotpath",
]
hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc",
"rustfs-common/hotpath-alloc",
"rustfs-concurrency/hotpath-alloc",
"rustfs-config/hotpath-alloc",
"rustfs-ecstore/hotpath-alloc",
"rustfs-madmin/hotpath-alloc",
"rustfs-storage-api/hotpath-alloc",
"rustfs-utils/hotpath-alloc",
"rustfs-test-utils/hotpath-alloc",
]
hotpath-cpu = [
"hotpath",
"hotpath/hotpath-cpu",
"rustfs-common/hotpath-cpu",
"rustfs-concurrency/hotpath-cpu",
"rustfs-config/hotpath-cpu",
"rustfs-ecstore/hotpath-cpu",
"rustfs-madmin/hotpath-cpu",
"rustfs-storage-api/hotpath-cpu",
"rustfs-utils/hotpath-cpu",
"rustfs-test-utils/hotpath-cpu",
]
[dependencies] [dependencies]
hotpath.workspace = true
rustfs-config = { workspace = true } rustfs-config = { workspace = true }
rustfs-concurrency = { workspace = true } rustfs-concurrency = { workspace = true }
rustfs-ecstore = { workspace = true } rustfs-ecstore = { workspace = true }
+1
View File
@@ -159,6 +159,7 @@ impl ErasureSetHealer {
/// execute erasure set heal with resume /// execute erasure set heal with resume
#[tracing::instrument(skip(self, buckets), fields(set_disk_id = %set_disk_id, bucket_count = buckets.len()))] #[tracing::instrument(skip(self, buckets), fields(set_disk_id = %set_disk_id, bucket_count = buckets.len()))]
#[hotpath::measure]
pub async fn heal_erasure_set(&self, buckets: &[String], set_disk_id: &str) -> Result<()> { pub async fn heal_erasure_set(&self, buckets: &[String], set_disk_id: &str) -> Result<()> {
debug!( debug!(
target: "rustfs::heal::erasure_healer", target: "rustfs::heal::erasure_healer",
+3
View File
@@ -584,6 +584,7 @@ impl HealTask {
} }
#[tracing::instrument(skip(self), fields(task_id = %self.id, heal_type = ?self.heal_type))] #[tracing::instrument(skip(self), fields(task_id = %self.id, heal_type = ?self.heal_type))]
#[hotpath::measure]
pub async fn execute(&self) -> Result<()> { pub async fn execute(&self) -> Result<()> {
// update status and timestamps atomically to avoid race conditions // update status and timestamps atomically to avoid race conditions
let now = SystemTime::now(); let now = SystemTime::now();
@@ -759,6 +760,7 @@ impl HealTask {
// specific heal implementation method // specific heal implementation method
#[tracing::instrument(skip(self), fields(bucket = %bucket, object = %object, version_id = ?version_id))] #[tracing::instrument(skip(self), fields(bucket = %bucket, object = %object, version_id = ?version_id))]
#[hotpath::measure]
async fn heal_object(&self, bucket: &str, object: &str, version_id: Option<&str>) -> Result<()> { async fn heal_object(&self, bucket: &str, object: &str, version_id: Option<&str>) -> Result<()> {
debug!( debug!(
target: "rustfs::heal::task", target: "rustfs::heal::task",
@@ -1404,6 +1406,7 @@ impl HealTask {
self.heal_bucket_objects(bucket, prefix).await self.heal_bucket_objects(bucket, prefix).await
} }
#[hotpath::measure]
async fn heal_bucket_objects(&self, bucket: &str, prefix: &str) -> Result<()> { async fn heal_bucket_objects(&self, bucket: &str, prefix: &str) -> Result<()> {
let mut continuation_token: Option<String> = None; let mut continuation_token: Option<String> = None;
let mut scanned = 0u64; let mut scanned = 0u64;
+48
View File
@@ -28,7 +28,55 @@ documentation = "https://docs.rs/rustfs-iam/latest/rustfs_iam/"
[lints] [lints]
workspace = true workspace = true
[features]
default = []
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
"hotpath/futures",
"hotpath/reqwest-0-13",
"rustfs-config/hotpath",
"rustfs-credentials/hotpath",
"rustfs-crypto/hotpath",
"rustfs-ecstore/hotpath",
"rustfs-io-metrics/hotpath",
"rustfs-madmin/hotpath",
"rustfs-policy/hotpath",
"rustfs-storage-api/hotpath",
"rustfs-utils/hotpath",
"rustfs-test-utils/hotpath",
]
hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc",
"rustfs-config/hotpath-alloc",
"rustfs-credentials/hotpath-alloc",
"rustfs-crypto/hotpath-alloc",
"rustfs-ecstore/hotpath-alloc",
"rustfs-io-metrics/hotpath-alloc",
"rustfs-madmin/hotpath-alloc",
"rustfs-policy/hotpath-alloc",
"rustfs-storage-api/hotpath-alloc",
"rustfs-utils/hotpath-alloc",
"rustfs-test-utils/hotpath-alloc",
]
hotpath-cpu = [
"hotpath",
"hotpath/hotpath-cpu",
"rustfs-config/hotpath-cpu",
"rustfs-credentials/hotpath-cpu",
"rustfs-crypto/hotpath-cpu",
"rustfs-ecstore/hotpath-cpu",
"rustfs-io-metrics/hotpath-cpu",
"rustfs-madmin/hotpath-cpu",
"rustfs-policy/hotpath-cpu",
"rustfs-storage-api/hotpath-cpu",
"rustfs-utils/hotpath-cpu",
"rustfs-test-utils/hotpath-cpu",
]
[dependencies] [dependencies]
hotpath.workspace = true
rustfs-credentials = { workspace = true } rustfs-credentials = { workspace = true }
rustfs-config = { workspace = true, features = ["server-config-model"] } rustfs-config = { workspace = true, features = ["server-config-model"] }
tokio = { workspace = true, features = ["fs", "rt-multi-thread"] } tokio = { workspace = true, features = ["fs", "rt-multi-thread"] }
+2
View File
@@ -469,6 +469,7 @@ impl ObjectStore {
}); });
} }
#[hotpath::measure]
async fn list_all_iamconfig_items(&self) -> Result<HashMap<String, Vec<String>>> { async fn list_all_iamconfig_items(&self) -> Result<HashMap<String, Vec<String>>> {
let (tx, mut rx) = mpsc::channel::<StringOrErr>(100); let (tx, mut rx) = mpsc::channel::<StringOrErr>(100);
@@ -508,6 +509,7 @@ impl ObjectStore {
Ok(res) Ok(res)
} }
#[hotpath::measure]
async fn load_policy_doc_concurrent(&self, names: &[String], mode: LoadMode) -> Result<Vec<PolicyDoc>> { async fn load_policy_doc_concurrent(&self, names: &[String], mode: LoadMode) -> Result<Vec<PolicyDoc>> {
let mut futures = Vec::with_capacity(names.len()); let mut futures = Vec::with_capacity(names.len());
+7
View File
@@ -27,7 +27,14 @@ categories = ["development-tools", "filesystem"]
[lints] [lints]
workspace = true workspace = true
[features]
default = []
hotpath = ["hotpath/hotpath", "hotpath/tokio", "rustfs-io-metrics/hotpath"]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-io-metrics/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-io-metrics/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true
bytes = { workspace = true, features = ["serde"] } bytes = { workspace = true, features = ["serde"] }
thiserror = { workspace = true } thiserror = { workspace = true }
tokio = { workspace = true, features = ["io-util", "fs", "sync", "rt-multi-thread"] } tokio = { workspace = true, features = ["io-util", "fs", "sync", "rt-multi-thread"] }
+25
View File
@@ -28,7 +28,32 @@ categories = ["development-tools", "filesystem"]
name = "metrics_pipeline" name = "metrics_pipeline"
harness = false harness = false
[features]
default = []
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
"rustfs-common/hotpath",
"rustfs-s3-ops/hotpath",
"rustfs-utils/hotpath",
]
hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc",
"rustfs-common/hotpath-alloc",
"rustfs-s3-ops/hotpath-alloc",
"rustfs-utils/hotpath-alloc",
]
hotpath-cpu = [
"hotpath",
"hotpath/hotpath-cpu",
"rustfs-common/hotpath-cpu",
"rustfs-s3-ops/hotpath-cpu",
"rustfs-utils/hotpath-cpu",
]
[dependencies] [dependencies]
hotpath.workspace = true
metrics = { workspace = true } metrics = { workspace = true }
rustfs-common = { workspace = true } rustfs-common = { workspace = true }
rustfs-s3-ops = { workspace = true } rustfs-s3-ops = { workspace = true }
+27
View File
@@ -28,7 +28,34 @@ authors.workspace = true
[lints] [lints]
workspace = true workspace = true
[features]
default = []
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
"hotpath/futures",
"hotpath/reqwest-0-13",
"rustfs-credentials/hotpath",
"rustfs-policy/hotpath",
"rustfs-utils/hotpath",
]
hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc",
"rustfs-credentials/hotpath-alloc",
"rustfs-policy/hotpath-alloc",
"rustfs-utils/hotpath-alloc",
]
hotpath-cpu = [
"hotpath",
"hotpath/hotpath-cpu",
"rustfs-credentials/hotpath-cpu",
"rustfs-policy/hotpath-cpu",
"rustfs-utils/hotpath-cpu",
]
[dependencies] [dependencies]
hotpath.workspace = true
tokio = { workspace = true, features = ["rt", "sync"] } tokio = { workspace = true, features = ["rt", "sync"] }
reqwest = { workspace = true, features = ["json"] } reqwest = { workspace = true, features = ["json"] }
serde = { workspace = true, features = ["derive"] } serde = { workspace = true, features = ["derive"] }
+2
View File
@@ -95,6 +95,7 @@ impl KeystoneClient {
} }
/// Validate a Keystone token /// Validate a Keystone token
#[hotpath::measure]
pub async fn validate_token(&self, token: &str) -> Result<KeystoneToken> { pub async fn validate_token(&self, token: &str) -> Result<KeystoneToken> {
match self.version { match self.version {
KeystoneVersion::V3 => self.validate_token_v3(token).await, KeystoneVersion::V3 => self.validate_token_v3(token).await,
@@ -238,6 +239,7 @@ impl KeystoneClient {
} }
/// Get EC2 credentials for a user /// Get EC2 credentials for a user
#[hotpath::measure]
pub async fn get_ec2_credentials(&self, user_id: &str, project_id: Option<&str>) -> Result<Vec<EC2Credential>> { pub async fn get_ec2_credentials(&self, user_id: &str, project_id: Option<&str>) -> Result<Vec<EC2Credential>> {
let admin_token = self.get_admin_token().await?; let admin_token = self.get_admin_token().await?;
+19
View File
@@ -28,6 +28,7 @@ categories = ["cryptography", "web-programming", "authentication"]
workspace = true workspace = true
[dependencies] [dependencies]
hotpath.workspace = true
# Core dependencies # Core dependencies
async-trait = { workspace = true } async-trait = { workspace = true }
tokio = { workspace = true, features = ["fs", "io-util", "macros", "rt-multi-thread", "sync", "time"] } tokio = { workspace = true, features = ["fs", "io-util", "macros", "rt-multi-thread", "sync", "time"] }
@@ -37,6 +38,8 @@ serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true, features = ["raw_value"] } serde_json = { workspace = true, features = ["raw_value"] }
tracing = { workspace = true } tracing = { workspace = true }
thiserror = { workspace = true } thiserror = { workspace = true }
# Operation metrics emitted by the retry policy engine (crate::policy).
metrics = { workspace = true }
# Cryptography # Cryptography
aes-gcm = { workspace = true, features = ["rand_core"] } aes-gcm = { workspace = true, features = ["rand_core"] }
@@ -72,6 +75,8 @@ tokio-util = { workspace = true }
[dev-dependencies] [dev-dependencies]
anyhow = { workspace = true } anyhow = { workspace = true }
# Debugging recorder for asserting emitted metrics in tests.
metrics-util = { version = "0.20", features = ["debugging"] }
insta = { workspace = true, features = ["yaml", "json"] } insta = { workspace = true, features = ["yaml", "json"] }
tempfile = { workspace = true } tempfile = { workspace = true }
temp-env = { workspace = true } temp-env = { workspace = true }
@@ -80,3 +85,17 @@ tokio = { workspace = true, features = ["net", "test-util"] }
[features] [features]
default = [] default = []
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
"hotpath/reqwest-0-13",
"rustfs-security-governance/hotpath",
"rustfs-utils/hotpath",
]
hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc",
"rustfs-security-governance/hotpath-alloc",
"rustfs-utils/hotpath-alloc",
]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-security-governance/hotpath-cpu", "rustfs-utils/hotpath-cpu"]
+3
View File
@@ -66,16 +66,19 @@ impl KmsManager {
} }
/// Encrypt data with a master key /// Encrypt data with a master key
#[hotpath::measure]
pub async fn encrypt(&self, request: EncryptRequest) -> Result<EncryptResponse> { pub async fn encrypt(&self, request: EncryptRequest) -> Result<EncryptResponse> {
self.backend.encrypt(request).await self.backend.encrypt(request).await
} }
/// Decrypt data with a master key /// Decrypt data with a master key
#[hotpath::measure]
pub async fn decrypt(&self, request: DecryptRequest) -> Result<DecryptResponse> { pub async fn decrypt(&self, request: DecryptRequest) -> Result<DecryptResponse> {
self.backend.decrypt(request).await self.backend.decrypt(request).await
} }
/// Generate a data encryption key /// Generate a data encryption key
#[hotpath::measure]
pub async fn generate_data_key(&self, request: GenerateDataKeyRequest) -> Result<GenerateDataKeyResponse> { pub async fn generate_data_key(&self, request: GenerateDataKeyRequest) -> Result<GenerateDataKeyResponse> {
self.backend.generate_data_key(request).await self.backend.generate_data_key(request).await
} }
+508 -22
View File
@@ -27,6 +27,12 @@
//! retried automatically: a response lost after the server applied the write //! retried automatically: a response lost after the server applied the write
//! would otherwise be replayed into duplicate side effects (extra key versions, //! would otherwise be replayed into duplicate side effects (extra key versions,
//! repeated deletes). //! repeated deletes).
//!
//! Every execution also records operation metrics (attempt failures by retry
//! class, terminal outcome, attempts used, wall-clock duration) through the
//! process-global `metrics` recorder. Metric labels carry only static enum
//! values — operation names, classes, outcomes — never key identifiers, key
//! material, ciphertext, or tokens.
use std::future::Future; use std::future::Future;
use std::time::Duration; use std::time::Duration;
@@ -200,6 +206,130 @@ fn equal_jitter(rng: &mut impl RngExt, cap: Duration) -> Duration {
half + Duration::from_nanos(rng.random_range(0..=spread)) half + Duration::from_nanos(rng.random_range(0..=spread))
} }
// ---------------------------------------------------------------------------
// Metrics
//
// Every execution is recorded here, at the single choke point all backend
// calls flow through, so instrumenting a new call site costs nothing beyond
// naming its operation. Label values are exclusively static enum strings
// (operation names, classes, outcomes) — key identifiers, key material,
// ciphertext, and tokens must never reach a metric label.
// ---------------------------------------------------------------------------
/// Counter: operations executed, by `operation`, `op_class`, and `outcome`.
const METRIC_OPERATIONS_TOTAL: &str = "rustfs_kms_backend_operations_total";
/// Counter: failed attempts, by `operation` and `error_class` (including
/// `attempt_timeout` for attempts cut off by the per-attempt timeout).
const METRIC_ATTEMPT_FAILURES_TOTAL: &str = "rustfs_kms_backend_attempt_failures_total";
/// Histogram: wall-clock duration of a whole operation (attempts plus
/// backoff), in seconds, by `operation` and `outcome`.
const METRIC_OPERATION_DURATION_SECONDS: &str = "rustfs_kms_backend_operation_duration_seconds";
/// Histogram: attempts one operation used before completing, by `operation`
/// and `outcome`.
const METRIC_OPERATION_ATTEMPTS: &str = "rustfs_kms_backend_operation_attempts";
impl OpClass {
fn as_label(self) -> &'static str {
match self {
OpClass::ReadIdempotent => "read_idempotent",
OpClass::MutatingNonIdempotent => "mutating_non_idempotent",
OpClass::Auth => "auth",
}
}
}
impl ErrorClass {
fn as_label(self) -> &'static str {
match self {
ErrorClass::RetryableConn => "retryable_conn",
ErrorClass::RetryableStatus => "retryable_status",
ErrorClass::Fatal => "fatal",
}
}
}
/// How one policy execution terminated, for the `outcome` metric label.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Outcome {
Success,
/// A fatal-classified failure ended the operation on its first observation.
Fatal,
/// The attempt budget ran out; the last failure was retryable (including a
/// timed-out final attempt).
BudgetExhausted,
/// The operation deadline ran out before another attempt could complete.
DeadlineExceeded,
Cancelled,
}
impl Outcome {
fn as_label(self) -> &'static str {
match self {
Outcome::Success => "success",
Outcome::Fatal => "fatal",
Outcome::BudgetExhausted => "budget_exhausted",
Outcome::DeadlineExceeded => "deadline_exceeded",
Outcome::Cancelled => "cancelled",
}
}
}
/// Register metric descriptions once per process.
fn describe_metrics() {
static DESCRIBE: std::sync::Once = std::sync::Once::new();
DESCRIBE.call_once(|| {
metrics::describe_counter!(
METRIC_OPERATIONS_TOTAL,
"Total KMS backend operations executed under the operation policy, by operation, operation class, and outcome"
);
metrics::describe_counter!(
METRIC_ATTEMPT_FAILURES_TOTAL,
"Total failed KMS backend attempts, by operation and retry classification"
);
metrics::describe_histogram!(
METRIC_OPERATION_DURATION_SECONDS,
"Wall-clock duration of KMS backend operations including retries and backoff, in seconds"
);
metrics::describe_histogram!(
METRIC_OPERATION_ATTEMPTS,
"Number of attempts a KMS backend operation used before completing"
);
});
}
/// Record one failed attempt with its retry classification.
fn record_attempt_failure(operation: &'static str, error_class: &'static str) {
metrics::counter!(
METRIC_ATTEMPT_FAILURES_TOTAL,
"operation" => operation,
"error_class" => error_class
)
.increment(1);
}
/// Record the terminal outcome of one policy execution.
fn record_operation(operation: &'static str, class: OpClass, outcome: Outcome, attempts: u32, elapsed: Duration) {
metrics::counter!(
METRIC_OPERATIONS_TOTAL,
"operation" => operation,
"op_class" => class.as_label(),
"outcome" => outcome.as_label()
)
.increment(1);
metrics::histogram!(
METRIC_OPERATION_DURATION_SECONDS,
"operation" => operation,
"outcome" => outcome.as_label()
)
.record(elapsed.as_secs_f64());
metrics::histogram!(
METRIC_OPERATION_ATTEMPTS,
"operation" => operation,
"outcome" => outcome.as_label()
)
.record(f64::from(attempts));
}
/// Run `attempt` under the policy. /// Run `attempt` under the policy.
/// ///
/// Each attempt is bounded by `attempt_timeout` (further capped by whatever is /// Each attempt is bounded by `attempt_timeout` (further capped by whatever is
@@ -230,13 +360,37 @@ where
/// [`execute`] with an injectable jitter source so tests can pin deterministic /// [`execute`] with an injectable jitter source so tests can pin deterministic
/// backoff durations instead of asserting around random sleeps. /// backoff durations instead of asserting around random sleeps.
pub(crate) async fn execute_with_jitter<T, F, Fut, J>( pub(crate) async fn execute_with_jitter<T, F, Fut, J>(
operation: &'static str,
class: OpClass,
policy: &RetryPolicy,
cancel: &CancellationToken,
jitter: J,
attempt: F,
) -> Result<T>
where
F: FnMut() -> Fut,
Fut: Future<Output = std::result::Result<T, AttemptError>>,
J: FnMut(Duration) -> Duration,
{
describe_metrics();
let started = Instant::now();
let mut attempts_made = 0u32;
let (outcome, result) = drive_attempts(operation, class, policy, cancel, jitter, attempt, &mut attempts_made).await;
record_operation(operation, class, outcome, attempts_made, started.elapsed());
result
}
/// The attempt loop behind [`execute_with_jitter`], returning the terminal
/// outcome alongside the result so the caller can record it exactly once.
async fn drive_attempts<T, F, Fut, J>(
operation: &'static str, operation: &'static str,
class: OpClass, class: OpClass,
policy: &RetryPolicy, policy: &RetryPolicy,
cancel: &CancellationToken, cancel: &CancellationToken,
mut jitter: J, mut jitter: J,
mut attempt: F, mut attempt: F,
) -> Result<T> attempts_made: &mut u32,
) -> (Outcome, Result<T>)
where where
F: FnMut() -> Fut, F: FnMut() -> Fut,
Fut: Future<Output = std::result::Result<T, AttemptError>>, Fut: Future<Output = std::result::Result<T, AttemptError>>,
@@ -247,48 +401,68 @@ where
let mut attempt_no = 0u32; let mut attempt_no = 0u32;
loop { loop {
attempt_no += 1;
if cancel.is_cancelled() { if cancel.is_cancelled() {
return Err(KmsError::operation_cancelled(format!( return (
"{operation} cancelled before attempt {attempt_no}" Outcome::Cancelled,
))); Err(KmsError::operation_cancelled(format!(
"{operation} cancelled before attempt {}",
attempt_no + 1
))),
);
} }
let remaining = deadline.saturating_duration_since(Instant::now()); let remaining = deadline.saturating_duration_since(Instant::now());
if remaining.is_zero() { if remaining.is_zero() {
return Err(KmsError::operation_timed_out(format!( return (
"{operation} exceeded operation deadline of {:?}", Outcome::DeadlineExceeded,
policy.op_deadline Err(KmsError::operation_timed_out(format!(
))); "{operation} exceeded operation deadline of {:?}",
policy.op_deadline
))),
);
} }
attempt_no += 1;
*attempts_made = attempt_no;
let attempt_budget = policy.attempt_timeout.min(remaining); let attempt_budget = policy.attempt_timeout.min(remaining);
let outcome = tokio::select! { let outcome = tokio::select! {
biased; biased;
_ = cancel.cancelled() => { _ = cancel.cancelled() => {
return Err(KmsError::operation_cancelled(format!("{operation} cancelled during attempt {attempt_no}"))); return (
Outcome::Cancelled,
Err(KmsError::operation_cancelled(format!("{operation} cancelled during attempt {attempt_no}"))),
);
} }
outcome = tokio::time::timeout(attempt_budget, attempt()) => outcome, outcome = tokio::time::timeout(attempt_budget, attempt()) => outcome,
}; };
let failure = match outcome { let failure = match outcome {
Ok(Ok(value)) => return Ok(value), Ok(Ok(value)) => return (Outcome::Success, Ok(value)),
Ok(Err(failure)) => failure, Ok(Err(failure)) => {
Err(_) => AttemptError { record_attempt_failure(operation, failure.class.as_label());
class: ErrorClass::RetryableConn, failure
error: KmsError::operation_timed_out(format!( }
"{operation} attempt {attempt_no} timed out after {attempt_budget:?}" Err(_) => {
)), record_attempt_failure(operation, "attempt_timeout");
}, AttemptError {
class: ErrorClass::RetryableConn,
error: KmsError::operation_timed_out(format!(
"{operation} attempt {attempt_no} timed out after {attempt_budget:?}"
)),
}
}
}; };
if failure.class == ErrorClass::Fatal || attempt_no >= max_attempts { if failure.class == ErrorClass::Fatal {
return Err(failure.error); return (Outcome::Fatal, Err(failure.error));
}
if attempt_no >= max_attempts {
return (Outcome::BudgetExhausted, Err(failure.error));
} }
let backoff = jitter(backoff_cap(policy, attempt_no)); let backoff = jitter(backoff_cap(policy, attempt_no));
if backoff >= deadline.saturating_duration_since(Instant::now()) { if backoff >= deadline.saturating_duration_since(Instant::now()) {
// Not enough deadline budget left for another attempt. // Not enough deadline budget left for another attempt.
return Err(failure.error); return (Outcome::DeadlineExceeded, Err(failure.error));
} }
tracing::warn!( tracing::warn!(
operation, operation,
@@ -300,7 +474,10 @@ where
tokio::select! { tokio::select! {
biased; biased;
_ = cancel.cancelled() => { _ = cancel.cancelled() => {
return Err(KmsError::operation_cancelled(format!("{operation} cancelled during retry backoff"))); return (
Outcome::Cancelled,
Err(KmsError::operation_cancelled(format!("{operation} cancelled during retry backoff"))),
);
} }
_ = tokio::time::sleep(backoff) => {} _ = tokio::time::sleep(backoff) => {}
} }
@@ -628,4 +805,313 @@ mod tests {
assert_eq!(classify_vaultrs(&ClientError::ResponseEmptyError), ErrorClass::Fatal); assert_eq!(classify_vaultrs(&ClientError::ResponseEmptyError), ErrorClass::Fatal);
assert_eq!(classify_vaultrs(&ClientError::InvalidLoginMethodError), ErrorClass::Fatal); assert_eq!(classify_vaultrs(&ClientError::InvalidLoginMethodError), ErrorClass::Fatal);
} }
// -- Metric emission ----------------------------------------------------
//
// Each test installs a thread-local debugging recorder and drives a
// paused-clock current-thread runtime inside it, so the emitted metrics
// (including virtual-clock durations) are fully deterministic.
use metrics_util::MetricKind;
use metrics_util::debugging::{DebugValue, DebuggingRecorder};
type MetricEntry = (
metrics_util::CompositeKey,
Option<metrics::Unit>,
Option<metrics::SharedString>,
DebugValue,
);
/// Run `test` on a paused current-thread runtime under a debugging
/// recorder and return one snapshot of everything it emitted.
///
/// A single snapshot per test on purpose: `Snapshotter::snapshot` drains
/// the recorded state, so taking it per assertion would only show the
/// first assertion any data.
fn record_metrics<Out>(test: impl FnOnce() -> std::pin::Pin<Box<dyn Future<Output = Out>>>) -> (Vec<MetricEntry>, Out) {
let recorder = DebuggingRecorder::new();
let snapshotter = recorder.snapshotter();
let out = metrics::with_local_recorder(&recorder, || {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_time()
.start_paused(true)
.build()
.expect("current-thread runtime must build");
runtime.block_on(test())
});
(snapshotter.snapshot().into_vec(), out)
}
fn labels_match(key: &metrics::Key, labels: &[(&str, &str)]) -> bool {
labels
.iter()
.all(|(label, expected)| key.labels().any(|l| l.key() == *label && l.value() == *expected))
}
fn counter_value(snapshot: &[MetricEntry], name: &str, labels: &[(&str, &str)]) -> u64 {
snapshot
.iter()
.filter_map(|(composite, _unit, _description, value)| {
let matches = composite.kind() == MetricKind::Counter
&& composite.key().name() == name
&& labels_match(composite.key(), labels);
match (matches, value) {
(true, DebugValue::Counter(count)) => Some(*count),
_ => None,
}
})
.sum()
}
fn histogram_values(snapshot: &[MetricEntry], name: &str, labels: &[(&str, &str)]) -> Vec<f64> {
snapshot
.iter()
.filter_map(|(composite, _unit, _description, value)| {
let matches = composite.kind() == MetricKind::Histogram
&& composite.key().name() == name
&& labels_match(composite.key(), labels);
match (matches, value) {
(true, DebugValue::Histogram(values)) => Some(values),
_ => None,
}
})
.flatten()
.map(|value| value.into_inner())
.collect()
}
#[test]
fn metrics_record_retried_success_with_attempts_and_duration() {
let calls_in_test = Arc::new(AtomicU32::new(0));
let (snapshot, ()) = record_metrics(move || {
Box::pin(async move {
let policy = policy_of(1_000, 60_000, 3, 100, 2_000);
let cancel = CancellationToken::new();
let calls_in_attempt = calls_in_test.clone();
execute_with_jitter("metrics_read", OpClass::ReadIdempotent, &policy, &cancel, full_jitter, move || {
let calls = calls_in_attempt.clone();
async move {
if calls.fetch_add(1, Ordering::SeqCst) < 2 {
Err(AttemptError {
class: ErrorClass::RetryableStatus,
error: KmsError::backend_error("throttled (429)"),
})
} else {
Ok(())
}
}
})
.await
.expect("retries within budget must succeed");
})
});
assert_eq!(
counter_value(
&snapshot,
METRIC_OPERATIONS_TOTAL,
&[
("operation", "metrics_read"),
("op_class", "read_idempotent"),
("outcome", "success")
]
),
1
);
assert_eq!(
counter_value(
&snapshot,
METRIC_ATTEMPT_FAILURES_TOTAL,
&[("operation", "metrics_read"), ("error_class", "retryable_status")]
),
2
);
assert_eq!(
histogram_values(
&snapshot,
METRIC_OPERATION_ATTEMPTS,
&[("operation", "metrics_read"), ("outcome", "success")]
),
vec![3.0]
);
// Full-cap backoffs of 100ms and 200ms on the paused clock.
let durations = histogram_values(
&snapshot,
METRIC_OPERATION_DURATION_SECONDS,
&[("operation", "metrics_read"), ("outcome", "success")],
);
assert_eq!(durations.len(), 1);
assert!((durations[0] - 0.3).abs() < 1e-9, "expected 0.3s of virtual backoff, got {durations:?}");
}
#[test]
fn metrics_record_fatal_outcome_with_single_attempt() {
let (snapshot, ()) = record_metrics(|| {
Box::pin(async {
let policy = policy_of(1_000, 60_000, 5, 100, 2_000);
let cancel = CancellationToken::new();
let result: Result<()> =
execute_with_jitter("metrics_fatal", OpClass::ReadIdempotent, &policy, &cancel, full_jitter, || async {
Err(AttemptError {
class: ErrorClass::Fatal,
error: KmsError::access_denied("permission denied (403)"),
})
})
.await;
result.expect_err("a fatal failure must end the operation");
})
});
assert_eq!(
counter_value(
&snapshot,
METRIC_OPERATIONS_TOTAL,
&[("operation", "metrics_fatal"), ("outcome", "fatal")]
),
1
);
assert_eq!(
counter_value(
&snapshot,
METRIC_ATTEMPT_FAILURES_TOTAL,
&[("operation", "metrics_fatal"), ("error_class", "fatal")]
),
1
);
assert_eq!(
histogram_values(
&snapshot,
METRIC_OPERATION_ATTEMPTS,
&[("operation", "metrics_fatal"), ("outcome", "fatal")]
),
vec![1.0]
);
}
#[test]
fn metrics_record_mutating_budget_exhausted_after_one_attempt() {
let (snapshot, ()) = record_metrics(|| {
Box::pin(async {
let policy = policy_of(1_000, 60_000, 5, 100, 2_000);
let cancel = CancellationToken::new();
let result: Result<()> = execute_with_jitter(
"metrics_rotate",
OpClass::MutatingNonIdempotent,
&policy,
&cancel,
full_jitter,
|| async { Err(retryable_conn_error()) },
)
.await;
result.expect_err("a mutating operation must not retry a retryable failure");
})
});
assert_eq!(
counter_value(
&snapshot,
METRIC_OPERATIONS_TOTAL,
&[
("operation", "metrics_rotate"),
("op_class", "mutating_non_idempotent"),
("outcome", "budget_exhausted")
]
),
1
);
assert_eq!(
histogram_values(
&snapshot,
METRIC_OPERATION_ATTEMPTS,
&[("operation", "metrics_rotate"), ("outcome", "budget_exhausted")]
),
vec![1.0]
);
}
#[test]
fn metrics_record_timeouts_and_deadline_outcome() {
let (snapshot, ()) = record_metrics(|| {
Box::pin(async {
// Hung attempts: 10s each against a 25s deadline (see
// total_duration_never_exceeds_deadline for the timeline).
let policy = policy_of(10_000, 25_000, 5, 100, 2_000);
let cancel = CancellationToken::new();
let result: Result<()> =
execute_with_jitter("metrics_hung", OpClass::ReadIdempotent, &policy, &cancel, full_jitter, || {
std::future::pending::<AttemptResult<()>>()
})
.await;
result.expect_err("hung attempts must exhaust the deadline");
})
});
assert_eq!(
counter_value(
&snapshot,
METRIC_OPERATIONS_TOTAL,
&[("operation", "metrics_hung"), ("outcome", "deadline_exceeded")]
),
1
);
assert_eq!(
counter_value(
&snapshot,
METRIC_ATTEMPT_FAILURES_TOTAL,
&[("operation", "metrics_hung"), ("error_class", "attempt_timeout")]
),
3
);
let durations = histogram_values(
&snapshot,
METRIC_OPERATION_DURATION_SECONDS,
&[("operation", "metrics_hung"), ("outcome", "deadline_exceeded")],
);
assert_eq!(durations.len(), 1);
assert!((durations[0] - 25.0).abs() < 1e-9, "expected the full 25s deadline, got {durations:?}");
}
#[test]
fn metrics_record_cancelled_outcome() {
let calls_in_test = Arc::new(AtomicU32::new(0));
let (snapshot, ()) = record_metrics(move || {
Box::pin(async move {
let policy = policy_of(1_000, 600_000, 5, 10_000, 10_000);
let cancel = CancellationToken::new();
let canceller = cancel.clone();
tokio::spawn(async move {
tokio::time::sleep(Duration::from_millis(500)).await;
canceller.cancel();
});
let calls_in_attempt = calls_in_test.clone();
let result: Result<()> =
execute_with_jitter("metrics_cancel", OpClass::ReadIdempotent, &policy, &cancel, full_jitter, move || {
let calls = calls_in_attempt.clone();
async move {
calls.fetch_add(1, Ordering::SeqCst);
Err(retryable_conn_error())
}
})
.await;
result.expect_err("cancellation must abort the backoff");
})
});
assert_eq!(
counter_value(
&snapshot,
METRIC_OPERATIONS_TOTAL,
&[("operation", "metrics_cancel"), ("outcome", "cancelled")]
),
1
);
assert_eq!(
histogram_values(
&snapshot,
METRIC_OPERATION_ATTEMPTS,
&[("operation", "metrics_cancel"), ("outcome", "cancelled")]
),
vec![1.0]
);
}
} }
+255
View File
@@ -0,0 +1,255 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//! Fault-injection matrix for the Vault backend operation policy.
//!
//! Offline cases run against locally injected transport faults (a closed
//! port, a listener that never responds) — deterministic, no external
//! dependencies. Real-Vault cases are `#[ignore]`d and need a dev Vault
//! (default `http://127.0.0.1:8200`, override with `RUSTFS_KMS_VAULT_ADDR`).
//!
//! Throttling (429) and recoverable 5xx responses cannot be forced on a stock
//! dev Vault, so their retry and metric behavior is pinned deterministically
//! by the scripted-Vault wiring tests in `backends::vault` and the engine
//! tests in `policy.rs`. Pointing `RUSTFS_KMS_VAULT_ADDR` at a
//! fault-injecting proxy reuses the ignored cases here unchanged.
//!
//! Every case installs a thread-local debugging metrics recorder and drives a
//! current-thread runtime inside it, so the policy metrics double as the
//! request-count assertion even against a real server.
use std::time::Duration;
use metrics_util::MetricKind;
use metrics_util::debugging::{DebugValue, DebuggingRecorder};
use rustfs_kms::backends::KmsClient;
use rustfs_kms::backends::vault::VaultKmsClient;
use rustfs_kms::{KmsConfig, KmsError, VaultAuthMethod, VaultConfig};
const OPERATIONS_TOTAL: &str = "rustfs_kms_backend_operations_total";
const ATTEMPT_FAILURES_TOTAL: &str = "rustfs_kms_backend_attempt_failures_total";
fn vault_config(address: &str, token: &str) -> VaultConfig {
VaultConfig {
address: address.to_string(),
auth_method: VaultAuthMethod::Token {
token: token.to_string(),
},
namespace: None,
mount_path: "transit".to_string(),
kv_mount: "secret".to_string(),
key_path_prefix: "rustfs/kms/fault-injection".to_string(),
tls: None,
}
}
fn kms_config(attempt_timeout: Duration, retry_attempts: u32) -> KmsConfig {
KmsConfig {
timeout: attempt_timeout,
retry_attempts,
..KmsConfig::default()
}
}
type MetricEntry = (
metrics_util::CompositeKey,
Option<metrics::Unit>,
Option<metrics::SharedString>,
DebugValue,
);
/// Run `test` on a current-thread runtime under a debugging metrics recorder
/// and return one snapshot of everything it emitted.
///
/// A single snapshot per test on purpose: `Snapshotter::snapshot` drains the
/// recorded state, so taking it per assertion would only show the first
/// assertion any data.
fn record_metrics(test: impl FnOnce() -> std::pin::Pin<Box<dyn std::future::Future<Output = ()>>>) -> Vec<MetricEntry> {
let recorder = DebuggingRecorder::new();
let snapshotter = recorder.snapshotter();
metrics::with_local_recorder(&recorder, || {
let runtime = tokio::runtime::Builder::new_current_thread()
.enable_all()
.build()
.expect("current-thread runtime must build");
runtime.block_on(test());
});
snapshotter.snapshot().into_vec()
}
/// Sum of counters with `name` whose labels include every `(key, value)` pair.
fn counter_value(snapshot: &[MetricEntry], name: &str, labels: &[(&str, &str)]) -> u64 {
snapshot
.iter()
.filter_map(|(composite, _unit, _description, value)| {
let key = composite.key();
let matches = composite.kind() == MetricKind::Counter
&& key.name() == name
&& labels
.iter()
.all(|(label, expected)| key.labels().any(|l| l.key() == *label && l.value() == *expected));
match (matches, value) {
(true, DebugValue::Counter(count)) => Some(*count),
_ => None,
}
})
.sum()
}
/// Connection refused: connection-class failures are retried up to the
/// configured budget, then surface as a backend error.
#[test]
fn connection_refused_is_retried_within_budget() {
// Reserve a loopback port and release it so nothing is listening there.
let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("reserve a loopback port");
let address = format!("http://{}", listener.local_addr().expect("reserved port addr"));
drop(listener);
let snapshot = record_metrics(|| {
Box::pin(async move {
let client = VaultKmsClient::new(vault_config(&address, "unused"), &kms_config(Duration::from_secs(2), 2))
.await
.expect("client construction performs no network calls");
let error = client
.describe_key("fault-injection-refused", None)
.await
.expect_err("a refused connection must fail the operation");
assert!(matches!(error, KmsError::BackendError { .. }), "got {error:?}");
})
});
assert_eq!(
counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "retryable_conn")]),
2,
"both budgeted attempts must observe the refused connection"
);
assert_eq!(counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "budget_exhausted")]), 1);
// The static-token login records its own success; the Vault read must not.
assert_eq!(
counter_value(
&snapshot,
OPERATIONS_TOTAL,
&[("operation", "vault_kv2_read_key"), ("outcome", "success")]
),
0
);
}
/// Stalled connection: a server that accepts but never responds is cut off by
/// the per-attempt timeout (either the policy timer or the equally sized HTTP
/// client timeout, whichever fires first) instead of hanging forever.
#[test]
fn stalled_connection_is_cut_off_by_the_attempt_timeout() {
let snapshot = record_metrics(|| {
Box::pin(async move {
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("bind stall listener");
let address = format!("http://{}", listener.local_addr().expect("stall listener addr"));
// Accept and park every connection without ever responding.
tokio::spawn(async move {
let mut parked = Vec::new();
loop {
let Ok((socket, _)) = listener.accept().await else { return };
parked.push(socket);
}
});
let client = VaultKmsClient::new(vault_config(&address, "unused"), &kms_config(Duration::from_millis(250), 1))
.await
.expect("client construction performs no network calls");
let error = client
.describe_key("fault-injection-stalled", None)
.await
.expect_err("a stalled request must be cut off by the attempt timeout");
assert!(
matches!(error, KmsError::OperationTimedOut { .. } | KmsError::BackendError { .. }),
"got {error:?}"
);
})
});
// The policy timer reports attempt_timeout; the client-level HTTP timeout
// surfaces as a connection-class failure. Either way it is exactly one
// attempt that was cut off.
let cut_off = counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "attempt_timeout")])
+ counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "retryable_conn")]);
assert_eq!(cut_off, 1, "the single budgeted attempt must be cut off by a timeout");
assert_eq!(counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "budget_exhausted")]), 1);
}
fn real_vault_address() -> String {
std::env::var("RUSTFS_KMS_VAULT_ADDR").unwrap_or_else(|_| "http://127.0.0.1:8200".to_string())
}
/// Invalid token against a real Vault: the 403 is fatal — exactly one
/// attempt, no retry, and the operation fails closed.
#[test]
#[ignore] // Requires a running Vault dev server
fn real_vault_invalid_token_is_fatal_and_never_retried() {
let snapshot = record_metrics(|| {
Box::pin(async {
let config = vault_config(&real_vault_address(), "fault-injection-invalid-token");
let client = VaultKmsClient::new(config, &kms_config(Duration::from_secs(5), 3))
.await
.expect("client construction performs no network calls");
let error = client
.describe_key("fault-injection-forbidden", None)
.await
.expect_err("an invalid token must be rejected");
assert!(matches!(error, KmsError::BackendError { .. }), "got {error:?}");
})
});
assert_eq!(
counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "fatal")]),
1,
"a 403 must be observed by exactly one attempt"
);
assert_eq!(counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "fatal")]), 1);
assert_eq!(
counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "retryable_status")]),
0,
"an auth failure must never be classified as retryable"
);
}
/// Healthy read against a real Vault: a missing key resolves in one attempt
/// (404 is fatal for retry purposes) and records a fatal outcome rather than
/// burning the retry budget.
#[test]
#[ignore] // Requires a running Vault dev server
fn real_vault_missing_key_is_resolved_in_one_attempt() {
let token = std::env::var("RUSTFS_KMS_VAULT_TOKEN").unwrap_or_else(|_| "dev-only-token".to_string());
let snapshot = record_metrics(|| {
Box::pin(async move {
let config = vault_config(&real_vault_address(), &token);
let client = VaultKmsClient::new(config, &kms_config(Duration::from_secs(5), 3))
.await
.expect("client construction performs no network calls");
let error = client
.describe_key("fault-injection-definitely-missing", None)
.await
.expect_err("a missing key must resolve to key-not-found");
assert!(matches!(error, KmsError::KeyNotFound { .. }), "got {error:?}");
})
});
assert_eq!(
counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "fatal")]),
1,
"a 404 must be observed by exactly one attempt"
);
assert_eq!(counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "fatal")]), 1);
}
+28
View File
@@ -25,7 +25,35 @@ keywords = ["lifecycle", "storage", "rustfs", "Minio"]
categories = ["web-programming", "development-tools", "filesystem"] categories = ["web-programming", "development-tools", "filesystem"]
documentation = "https://docs.rs/rustfs-lifecycle/latest/rustfs_lifecycle/" documentation = "https://docs.rs/rustfs-lifecycle/latest/rustfs_lifecycle/"
[features]
default = []
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
"rustfs-common/hotpath",
"rustfs-config/hotpath",
"rustfs-replication/hotpath",
"rustfs-storage-api/hotpath",
]
hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc",
"rustfs-common/hotpath-alloc",
"rustfs-config/hotpath-alloc",
"rustfs-replication/hotpath-alloc",
"rustfs-storage-api/hotpath-alloc",
]
hotpath-cpu = [
"hotpath",
"hotpath/hotpath-cpu",
"rustfs-common/hotpath-cpu",
"rustfs-config/hotpath-cpu",
"rustfs-replication/hotpath-cpu",
"rustfs-storage-api/hotpath-cpu",
]
[dependencies] [dependencies]
hotpath.workspace = true
async-trait.workspace = true async-trait.workspace = true
metrics.workspace = true metrics.workspace = true
rustfs-common.workspace = true rustfs-common.workspace = true
+14
View File
@@ -28,7 +28,21 @@ documentation = "https://docs.rs/rustfs-lock/latest/rustfs_lock/"
[lints] [lints]
workspace = true workspace = true
[features]
default = []
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
"hotpath/futures",
"hotpath/parking_lot",
"rustfs-io-metrics/hotpath",
"rustfs-utils/hotpath",
]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-io-metrics/hotpath-alloc", "rustfs-utils/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-io-metrics/hotpath-cpu", "rustfs-utils/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true
rustfs-io-metrics = { workspace = true } rustfs-io-metrics = { workspace = true }
rustfs-utils = { workspace = true } rustfs-utils = { workspace = true }
async-trait.workspace = true async-trait.workspace = true
+4
View File
@@ -540,6 +540,7 @@ impl DistributedLock {
} }
/// Acquire a lock and return a RAII guard /// Acquire a lock and return a RAII guard
#[hotpath::measure]
pub(crate) async fn acquire_guard(&self, request: &LockRequest) -> Result<Option<DistributedLockGuard>> { pub(crate) async fn acquire_guard(&self, request: &LockRequest) -> Result<Option<DistributedLockGuard>> {
if self.clients.is_empty() { if self.clients.is_empty() {
return Err(LockError::internal("No lock clients available")); return Err(LockError::internal("No lock clients available"));
@@ -626,6 +627,7 @@ impl DistributedLock {
} }
/// Convenience: acquire exclusive lock as a guard /// Convenience: acquire exclusive lock as a guard
#[hotpath::measure]
pub async fn lock_guard( pub async fn lock_guard(
&self, &self,
resource: ObjectKey, resource: ObjectKey,
@@ -640,6 +642,7 @@ impl DistributedLock {
} }
/// Convenience: acquire exclusive lock with expected contention logs suppressed /// Convenience: acquire exclusive lock with expected contention logs suppressed
#[hotpath::measure]
pub async fn lock_guard_quiet( pub async fn lock_guard_quiet(
&self, &self,
resource: ObjectKey, resource: ObjectKey,
@@ -655,6 +658,7 @@ impl DistributedLock {
} }
/// Convenience: acquire shared lock as a guard /// Convenience: acquire shared lock as a guard
#[hotpath::measure]
pub async fn rlock_guard( pub async fn rlock_guard(
&self, &self,
resource: ObjectKey, resource: ObjectKey,
+7
View File
@@ -32,7 +32,14 @@ doctest = false
name = "la-dump-anchors" name = "la-dump-anchors"
path = "src/bin/la_dump_anchors.rs" path = "src/bin/la_dump_anchors.rs"
[features]
default = []
hotpath = ["hotpath/hotpath"]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true
chrono = { workspace = true, features = ["serde"] } chrono = { workspace = true, features = ["serde"] }
flate2 = { workspace = true } flate2 = { workspace = true }
regex = { workspace = true } regex = { workspace = true }
+7
View File
@@ -28,7 +28,14 @@ documentation = "https://docs.rs/rustfs-madmin/latest/rustfs_madmin/"
[lints] [lints]
workspace = true workspace = true
[features]
default = []
hotpath = ["hotpath/hotpath"]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true
chrono = { workspace = true, features = ["serde"] } chrono = { workspace = true, features = ["serde"] }
humantime.workspace = true humantime.workspace = true
hyper = { workspace = true, features = ["http2", "http1", "server"] } hyper = { workspace = true, features = ["http2", "http1", "server"] }
+31
View File
@@ -26,9 +26,40 @@ categories = ["web-programming", "development-tools", "filesystem"]
documentation = "https://docs.rs/rustfs-notify/latest/rustfs_notify/" documentation = "https://docs.rs/rustfs-notify/latest/rustfs_notify/"
[features] [features]
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
"rustfs-config/hotpath",
"rustfs-ecstore/hotpath",
"rustfs-s3-ops/hotpath",
"rustfs-s3-types/hotpath",
"rustfs-targets/hotpath",
"rustfs-utils/hotpath",
]
hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc",
"rustfs-config/hotpath-alloc",
"rustfs-ecstore/hotpath-alloc",
"rustfs-s3-ops/hotpath-alloc",
"rustfs-s3-types/hotpath-alloc",
"rustfs-targets/hotpath-alloc",
"rustfs-utils/hotpath-alloc",
]
hotpath-cpu = [
"hotpath",
"hotpath/hotpath-cpu",
"rustfs-config/hotpath-cpu",
"rustfs-ecstore/hotpath-cpu",
"rustfs-s3-ops/hotpath-cpu",
"rustfs-s3-types/hotpath-cpu",
"rustfs-targets/hotpath-cpu",
"rustfs-utils/hotpath-cpu",
]
demo-examples = [] demo-examples = []
[dependencies] [dependencies]
hotpath.workspace = true
rustfs-config = { workspace = true, features = ["notify", "server-config-model"] } rustfs-config = { workspace = true, features = ["notify", "server-config-model"] }
rustfs-ecstore = { workspace = true } rustfs-ecstore = { workspace = true }
rustfs-s3-types = { workspace = true } rustfs-s3-types = { workspace = true }
+26
View File
@@ -34,7 +34,33 @@ harness = false
[lints] [lints]
workspace = true workspace = true
[features]
default = []
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
"hotpath/futures",
"rustfs-config/hotpath",
"rustfs-io-metrics/hotpath",
"rustfs-utils/hotpath",
]
hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc",
"rustfs-config/hotpath-alloc",
"rustfs-io-metrics/hotpath-alloc",
"rustfs-utils/hotpath-alloc",
]
hotpath-cpu = [
"hotpath",
"hotpath/hotpath-cpu",
"rustfs-config/hotpath-cpu",
"rustfs-io-metrics/hotpath-cpu",
"rustfs-utils/hotpath-cpu",
]
[dependencies] [dependencies]
hotpath.workspace = true
rustfs-config = { workspace = true, features = ["constants"] } rustfs-config = { workspace = true, features = ["constants"] }
rustfs-io-metrics = { workspace = true } rustfs-io-metrics = { workspace = true }
rustfs-utils = { workspace = true, features = ["os"] } rustfs-utils = { workspace = true, features = ["os"] }
@@ -1035,6 +1035,7 @@ impl HybridCapacityManager {
/// ///
/// Joiners subscribe to the watch channel *before* releasing the mutex, which guarantees /// Joiners subscribe to the watch channel *before* releasing the mutex, which guarantees
/// they cannot miss the completion notification even if the leader finishes very quickly. /// they cannot miss the completion notification even if the leader finishes very quickly.
#[hotpath::measure]
pub async fn refresh_or_join<F, Fut>(&self, source: DataSource, refresh_fn: F) -> Result<CapacityUpdate, String> pub async fn refresh_or_join<F, Fut>(&self, source: DataSource, refresh_fn: F) -> Result<CapacityUpdate, String>
where where
F: FnOnce() -> Fut, F: FnOnce() -> Fut,
@@ -1142,6 +1143,7 @@ impl HybridCapacityManager {
} }
/// Start a background refresh if one is not already in flight. /// Start a background refresh if one is not already in flight.
#[hotpath::measure]
pub async fn spawn_refresh_if_needed<F, Fut>(self: Arc<Self>, source: DataSource, refresh_fn: F) -> bool pub async fn spawn_refresh_if_needed<F, Fut>(self: Arc<Self>, source: DataSource, refresh_fn: F) -> bool
where where
F: FnOnce() -> Fut + Send + 'static, F: FnOnce() -> Fut + Send + 'static,
+1
View File
@@ -338,6 +338,7 @@ pub async fn select_capacity_refresh_disks(
} }
} }
#[hotpath::measure]
pub async fn refresh_capacity_with_scope(disks: Vec<CapacityDiskRef>, dirty_subset: bool) -> Result<CapacityUpdate, String> { pub async fn refresh_capacity_with_scope(disks: Vec<CapacityDiskRef>, dirty_subset: bool) -> Result<CapacityUpdate, String> {
let scan_started_at = Instant::now(); let scan_started_at = Instant::now();
let report = calculate_data_dir_used_capacity_report(&disks) let report = calculate_data_dir_used_capacity_report(&disks)
+7
View File
@@ -27,7 +27,14 @@ categories = ["web-programming", "development-tools"]
[lib] [lib]
doctest = false doctest = false
[features]
default = []
hotpath = ["hotpath/hotpath", "hotpath/tokio"]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true
bytes = { workspace = true, features = ["serde"] } bytes = { workspace = true, features = ["serde"] }
metrics = { workspace = true } metrics = { workspace = true }
moka = { workspace = true, features = ["future"] } moka = { workspace = true, features = ["future"] }
+45
View File
@@ -27,6 +27,50 @@ documentation = "https://docs.rs/rustfs-obs/latest/rustfs_obs/"
[features] [features]
default = [] default = []
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
"hotpath/futures",
"hotpath/crossbeam",
"rustfs-audit/hotpath",
"rustfs-common/hotpath",
"rustfs-config/hotpath",
"rustfs-ecstore/hotpath",
"rustfs-iam/hotpath",
"rustfs-io-metrics/hotpath",
"rustfs-notify/hotpath",
"rustfs-security-governance/hotpath",
"rustfs-storage-api/hotpath",
"rustfs-utils/hotpath",
]
hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc",
"rustfs-audit/hotpath-alloc",
"rustfs-common/hotpath-alloc",
"rustfs-config/hotpath-alloc",
"rustfs-ecstore/hotpath-alloc",
"rustfs-iam/hotpath-alloc",
"rustfs-io-metrics/hotpath-alloc",
"rustfs-notify/hotpath-alloc",
"rustfs-security-governance/hotpath-alloc",
"rustfs-storage-api/hotpath-alloc",
"rustfs-utils/hotpath-alloc",
]
hotpath-cpu = [
"hotpath",
"hotpath/hotpath-cpu",
"rustfs-audit/hotpath-cpu",
"rustfs-common/hotpath-cpu",
"rustfs-config/hotpath-cpu",
"rustfs-ecstore/hotpath-cpu",
"rustfs-iam/hotpath-cpu",
"rustfs-io-metrics/hotpath-cpu",
"rustfs-notify/hotpath-cpu",
"rustfs-security-governance/hotpath-cpu",
"rustfs-storage-api/hotpath-cpu",
"rustfs-utils/hotpath-cpu",
]
# Tokio runtime-level telemetry. Requires a `--cfg tokio_unstable` build; the # Tokio runtime-level telemetry. Requires a `--cfg tokio_unstable` build; the
# build script fails the compile when that flag is missing. Off by default so # build script fails the compile when that flag is missing. Off by default so
# ordinary builds neither pay for nor depend on Tokio's unstable API. # ordinary builds neither pay for nor depend on Tokio's unstable API.
@@ -58,6 +102,7 @@ required-features = ["dial9"]
workspace = true workspace = true
[dependencies] [dependencies]
hotpath.workspace = true
rustfs-audit = { workspace = true } rustfs-audit = { workspace = true }
rustfs-common = { workspace = true } rustfs-common = { workspace = true }
rustfs-config = { workspace = true, features = ["observability"] } rustfs-config = { workspace = true, features = ["observability"] }
+21 -3
View File
@@ -30,9 +30,27 @@ workspace = true
[features] [features]
default = [] default = []
hotpath = ["hotpath/hotpath", "hotpath/reqwest-0-13"] hotpath = [
hotpath-alloc = ["hotpath/hotpath-alloc"] "hotpath/hotpath",
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"] "hotpath/reqwest-0-13",
"rustfs-config/hotpath",
"rustfs-credentials/hotpath",
"rustfs-crypto/hotpath",
]
hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc",
"rustfs-config/hotpath-alloc",
"rustfs-credentials/hotpath-alloc",
"rustfs-crypto/hotpath-alloc",
]
hotpath-cpu = [
"hotpath",
"hotpath/hotpath-cpu",
"rustfs-config/hotpath-cpu",
"rustfs-credentials/hotpath-cpu",
"rustfs-crypto/hotpath-cpu",
]
[dependencies] [dependencies]
rustfs-credentials = { workspace = true } rustfs-credentials = { workspace = true }
+50
View File
@@ -30,6 +30,55 @@ workspace = true
[features] [features]
default = [] default = []
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
"hotpath/futures",
"rustfs-config/hotpath",
"rustfs-credentials/hotpath",
"rustfs-ecstore?/hotpath",
"rustfs-iam/hotpath",
"rustfs-keystone?/hotpath",
"rustfs-policy/hotpath",
"rustfs-rio?/hotpath",
"rustfs-storage-api/hotpath",
"rustfs-tls-runtime?/hotpath",
"rustfs-trusted-proxies?/hotpath",
"rustfs-utils/hotpath",
"rustfs-test-utils/hotpath",
]
hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc",
"rustfs-config/hotpath-alloc",
"rustfs-credentials/hotpath-alloc",
"rustfs-ecstore?/hotpath-alloc",
"rustfs-iam/hotpath-alloc",
"rustfs-keystone?/hotpath-alloc",
"rustfs-policy/hotpath-alloc",
"rustfs-rio?/hotpath-alloc",
"rustfs-storage-api/hotpath-alloc",
"rustfs-tls-runtime?/hotpath-alloc",
"rustfs-trusted-proxies?/hotpath-alloc",
"rustfs-utils/hotpath-alloc",
"rustfs-test-utils/hotpath-alloc",
]
hotpath-cpu = [
"hotpath",
"hotpath/hotpath-cpu",
"rustfs-config/hotpath-cpu",
"rustfs-credentials/hotpath-cpu",
"rustfs-ecstore?/hotpath-cpu",
"rustfs-iam/hotpath-cpu",
"rustfs-keystone?/hotpath-cpu",
"rustfs-policy/hotpath-cpu",
"rustfs-rio?/hotpath-cpu",
"rustfs-storage-api/hotpath-cpu",
"rustfs-tls-runtime?/hotpath-cpu",
"rustfs-trusted-proxies?/hotpath-cpu",
"rustfs-utils/hotpath-cpu",
"rustfs-test-utils/hotpath-cpu",
]
ftps = ["dep:libunftp", "dep:unftp-core", "dep:rustls", "dep:rustfs-tls-runtime", "dep:subtle"] ftps = ["dep:libunftp", "dep:unftp-core", "dep:rustls", "dep:rustfs-tls-runtime", "dep:subtle"]
swift = [ swift = [
"dep:rustfs-keystone", "dep:rustfs-keystone",
@@ -62,6 +111,7 @@ webdav = ["dep:dav-server", "dep:hyper", "dep:hyper-util", "dep:http-body-util",
sftp = ["dep:russh", "dep:russh-sftp", "dep:uuid", "dep:subtle", "dep:tokio-util", "dep:socket2"] sftp = ["dep:russh", "dep:russh-sftp", "dep:uuid", "dep:subtle", "dep:tokio-util", "dep:socket2"]
[dependencies] [dependencies]
hotpath.workspace = true
# Core RustFS dependencies # Core RustFS dependencies
rustfs-iam = { workspace = true } rustfs-iam = { workspace = true }
rustfs-credentials = { workspace = true } rustfs-credentials = { workspace = true }
+31
View File
@@ -32,7 +32,38 @@ workspace = true
name = "gproto" name = "gproto"
path = "src/main.rs" path = "src/main.rs"
[features]
default = []
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
"rustfs-common/hotpath",
"rustfs-config/hotpath",
"rustfs-io-metrics/hotpath",
"rustfs-tls-runtime/hotpath",
"rustfs-utils/hotpath",
]
hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc",
"rustfs-common/hotpath-alloc",
"rustfs-config/hotpath-alloc",
"rustfs-io-metrics/hotpath-alloc",
"rustfs-tls-runtime/hotpath-alloc",
"rustfs-utils/hotpath-alloc",
]
hotpath-cpu = [
"hotpath",
"hotpath/hotpath-cpu",
"rustfs-common/hotpath-cpu",
"rustfs-config/hotpath-cpu",
"rustfs-io-metrics/hotpath-cpu",
"rustfs-tls-runtime/hotpath-cpu",
"rustfs-utils/hotpath-cpu",
]
[dependencies] [dependencies]
hotpath.workspace = true
rustfs-common.workspace = true rustfs-common.workspace = true
rustfs-io-metrics.workspace = true rustfs-io-metrics.workspace = true
rustfs-config.workspace = true rustfs-config.workspace = true
+7
View File
@@ -25,7 +25,14 @@ keywords = ["replication", "storage", "rustfs", "Minio"]
categories = ["web-programming", "development-tools", "filesystem"] categories = ["web-programming", "development-tools", "filesystem"]
documentation = "https://docs.rs/rustfs-replication/latest/rustfs_replication/" documentation = "https://docs.rs/rustfs-replication/latest/rustfs_replication/"
[features]
default = []
hotpath = ["hotpath/hotpath", "hotpath/tokio"]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true
bytes = { workspace = true, features = ["serde"] } bytes = { workspace = true, features = ["serde"] }
byteorder.workspace = true byteorder.workspace = true
regex.workspace = true regex.workspace = true
+25
View File
@@ -28,7 +28,32 @@ documentation = "https://docs.rs/rustfs-rio-v2/latest/rustfs_rio_v2/"
[lints] [lints]
workspace = true workspace = true
[features]
default = []
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
"rustfs-rio/hotpath",
"rustfs-utils/hotpath",
"rustfs-filemeta/hotpath",
]
hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc",
"rustfs-rio/hotpath-alloc",
"rustfs-utils/hotpath-alloc",
"rustfs-filemeta/hotpath-alloc",
]
hotpath-cpu = [
"hotpath",
"hotpath/hotpath-cpu",
"rustfs-rio/hotpath-cpu",
"rustfs-utils/hotpath-cpu",
"rustfs-filemeta/hotpath-cpu",
]
[dependencies] [dependencies]
hotpath.workspace = true
aes-gcm = { workspace = true, features = ["rand_core"] } aes-gcm = { workspace = true, features = ["rand_core"] }
bytes = { workspace = true, features = ["serde"] } bytes = { workspace = true, features = ["serde"] }
chacha20poly1305.workspace = true chacha20poly1305.workspace = true
+26 -3
View File
@@ -30,9 +30,32 @@ workspace = true
[features] [features]
default = [] default = []
hotpath = ["hotpath/hotpath", "hotpath/tokio", "hotpath/futures", "hotpath/reqwest-0-13"] hotpath = [
hotpath-alloc = ["hotpath/hotpath-alloc"] "hotpath/hotpath",
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"] "hotpath/tokio",
"hotpath/futures",
"hotpath/reqwest-0-13",
"rustfs-config/hotpath",
"rustfs-io-metrics/hotpath",
"rustfs-tls-runtime/hotpath",
"rustfs-utils/hotpath",
]
hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc",
"rustfs-config/hotpath-alloc",
"rustfs-io-metrics/hotpath-alloc",
"rustfs-tls-runtime/hotpath-alloc",
"rustfs-utils/hotpath-alloc",
]
hotpath-cpu = [
"hotpath",
"hotpath/hotpath-cpu",
"rustfs-config/hotpath-cpu",
"rustfs-io-metrics/hotpath-cpu",
"rustfs-tls-runtime/hotpath-cpu",
"rustfs-utils/hotpath-cpu",
]
[dependencies] [dependencies]
hotpath.workspace = true hotpath.workspace = true
+7
View File
@@ -27,7 +27,14 @@ categories = ["data-structures"]
[lints] [lints]
workspace = true workspace = true
[features]
default = []
hotpath = ["hotpath/hotpath", "rustfs-s3-types/hotpath"]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-s3-types/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-s3-types/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true
rustfs-s3-types = { workspace = true } rustfs-s3-types = { workspace = true }
[lib] [lib]
+7
View File
@@ -27,7 +27,14 @@ categories = ["data-structures"]
[lints] [lints]
workspace = true workspace = true
[features]
default = []
hotpath = ["hotpath/hotpath"]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true
serde = { workspace = true, features = ["derive"] } serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true, features = ["raw_value"] } serde_json = { workspace = true, features = ["raw_value"] }
+30
View File
@@ -28,7 +28,37 @@ documentation = "https://docs.rs/rustfs-s3select-api/latest/rustfs_s3select_api/
[lints] [lints]
workspace = true workspace = true
[features]
default = []
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
"hotpath/futures",
"hotpath/parking_lot",
"rustfs-common/hotpath",
"rustfs-ecstore/hotpath",
"rustfs-storage-api/hotpath",
"rustfs-test-utils/hotpath",
]
hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc",
"rustfs-common/hotpath-alloc",
"rustfs-ecstore/hotpath-alloc",
"rustfs-storage-api/hotpath-alloc",
"rustfs-test-utils/hotpath-alloc",
]
hotpath-cpu = [
"hotpath",
"hotpath/hotpath-cpu",
"rustfs-common/hotpath-cpu",
"rustfs-ecstore/hotpath-cpu",
"rustfs-storage-api/hotpath-cpu",
"rustfs-test-utils/hotpath-cpu",
]
[dependencies] [dependencies]
hotpath.workspace = true
metrics = { workspace = true } metrics = { workspace = true }
async-trait.workspace = true async-trait.workspace = true
bytes = { workspace = true, features = ["serde"] } bytes = { workspace = true, features = ["serde"] }
+13
View File
@@ -28,7 +28,20 @@ documentation = "https://docs.rs/rustfs-s3select-query/latest/rustfs_s3select_qu
[lints] [lints]
workspace = true workspace = true
[features]
default = []
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
"hotpath/futures",
"hotpath/parking_lot",
"rustfs-s3select-api/hotpath",
]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-s3select-api/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-s3select-api/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true
rustfs-s3select-api = { workspace = true } rustfs-s3select-api = { workspace = true }
async-recursion = { workspace = true } async-recursion = { workspace = true }
async-trait.workspace = true async-trait.workspace = true
+41
View File
@@ -29,7 +29,48 @@ documentation = "https://docs.rs/rustfs-scanner/latest/rustfs_scanner/"
[lints] [lints]
workspace = true workspace = true
[features]
default = []
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
"hotpath/futures",
"rustfs-common/hotpath",
"rustfs-config/hotpath",
"rustfs-credentials/hotpath",
"rustfs-data-usage/hotpath",
"rustfs-ecstore/hotpath",
"rustfs-filemeta/hotpath",
"rustfs-storage-api/hotpath",
"rustfs-utils/hotpath",
]
hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc",
"rustfs-common/hotpath-alloc",
"rustfs-config/hotpath-alloc",
"rustfs-credentials/hotpath-alloc",
"rustfs-data-usage/hotpath-alloc",
"rustfs-ecstore/hotpath-alloc",
"rustfs-filemeta/hotpath-alloc",
"rustfs-storage-api/hotpath-alloc",
"rustfs-utils/hotpath-alloc",
]
hotpath-cpu = [
"hotpath",
"hotpath/hotpath-cpu",
"rustfs-common/hotpath-cpu",
"rustfs-config/hotpath-cpu",
"rustfs-credentials/hotpath-cpu",
"rustfs-data-usage/hotpath-cpu",
"rustfs-ecstore/hotpath-cpu",
"rustfs-filemeta/hotpath-cpu",
"rustfs-storage-api/hotpath-cpu",
"rustfs-utils/hotpath-cpu",
]
[dependencies] [dependencies]
hotpath.workspace = true
rustfs-config = { workspace = true, features = ["server-config-model"] } rustfs-config = { workspace = true, features = ["server-config-model"] }
rustfs-common = { workspace = true } rustfs-common = { workspace = true }
rustfs-credentials = { workspace = true } rustfs-credentials = { workspace = true }
+1
View File
@@ -2531,6 +2531,7 @@ where
} }
#[instrument(skip_all)] #[instrument(skip_all)]
#[hotpath::measure]
async fn run_data_scanner_cycle( async fn run_data_scanner_cycle(
ctx: &CancellationToken, ctx: &CancellationToken,
storeapi: &Arc<ECStore>, storeapi: &Arc<ECStore>,
+7
View File
@@ -30,5 +30,12 @@ doctest = false
[lints] [lints]
workspace = true workspace = true
[features]
default = []
hotpath = ["hotpath/hotpath"]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true
thiserror = { workspace = true } thiserror = { workspace = true }
+7
View File
@@ -25,7 +25,14 @@ keywords = ["digital-signature", "verification", "integrity", "rustfs", "Minio"]
categories = ["web-programming", "development-tools", "cryptography"] categories = ["web-programming", "development-tools", "cryptography"]
documentation = "https://docs.rs/rustfs-signer/latest/rustfs_signer/" documentation = "https://docs.rs/rustfs-signer/latest/rustfs_signer/"
[features]
default = []
hotpath = ["hotpath/hotpath", "rustfs-utils/hotpath"]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-utils/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-utils/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true
tracing.workspace = true tracing.workspace = true
bytes = { workspace = true, features = ["serde"] } bytes = { workspace = true, features = ["serde"] }
http.workspace = true http.workspace = true
+7
View File
@@ -27,7 +27,14 @@ categories = ["web-programming", "development-tools", "filesystem"]
[lib] [lib]
doctest = false doctest = false
[features]
default = []
hotpath = ["hotpath/hotpath", "hotpath/tokio", "rustfs-filemeta/hotpath"]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-filemeta/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-filemeta/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true
async-trait.workspace = true async-trait.workspace = true
# Storage-facing replication contracts are isolated in src/replication.rs until # Storage-facing replication contracts are isolated in src/replication.rs until
# the underlying wire types can move without creating a replication/storage-api cycle. # the underlying wire types can move without creating a replication/storage-api cycle.
+34
View File
@@ -11,7 +11,41 @@ keywords = ["file-system", "notification", "target", "rustfs", "Minio"]
categories = ["web-programming", "development-tools", "filesystem"] categories = ["web-programming", "development-tools", "filesystem"]
documentation = "https://docs.rs/rustfs-targets/latest/rustfs_targets/" documentation = "https://docs.rs/rustfs-targets/latest/rustfs_targets/"
[features]
default = []
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
"hotpath/futures",
"hotpath/parking_lot",
"hotpath/reqwest-0-13",
"rustfs-config/hotpath",
"rustfs-extension-schema/hotpath",
"rustfs-s3-types/hotpath",
"rustfs-tls-runtime/hotpath",
"rustfs-utils/hotpath",
]
hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc",
"rustfs-config/hotpath-alloc",
"rustfs-extension-schema/hotpath-alloc",
"rustfs-s3-types/hotpath-alloc",
"rustfs-tls-runtime/hotpath-alloc",
"rustfs-utils/hotpath-alloc",
]
hotpath-cpu = [
"hotpath",
"hotpath/hotpath-cpu",
"rustfs-config/hotpath-cpu",
"rustfs-extension-schema/hotpath-cpu",
"rustfs-s3-types/hotpath-cpu",
"rustfs-tls-runtime/hotpath-cpu",
"rustfs-utils/hotpath-cpu",
]
[dependencies] [dependencies]
hotpath.workspace = true
rustfs-config = { workspace = true, features = ["notify", "audit", "server-config-model"] } rustfs-config = { workspace = true, features = ["notify", "audit", "server-config-model"] }
rustfs-extension-schema = { workspace = true } rustfs-extension-schema = { workspace = true }
rustfs-tls-runtime = { workspace = true } rustfs-tls-runtime = { workspace = true }
+5
View File
@@ -522,6 +522,7 @@ fn snapshot_from_delivery(target_id: TargetID, delivery: TargetDeliverySnapshot)
} }
} }
#[hotpath::measure]
pub async fn init_target_and_optionally_start_replay<E, F, G>( pub async fn init_target_and_optionally_start_replay<E, F, G>(
target: Box<dyn Target<E> + Send + Sync>, target: Box<dyn Target<E> + Send + Sync>,
on_replay_start: F, on_replay_start: F,
@@ -557,6 +558,7 @@ where
Some((shared, cancel)) Some((shared, cancel))
} }
#[hotpath::measure]
pub(crate) async fn prepare_target<E>( pub(crate) async fn prepare_target<E>(
target: Box<dyn Target<E> + Send + Sync>, target: Box<dyn Target<E> + Send + Sync>,
cancellation: Option<&CancellationToken>, cancellation: Option<&CancellationToken>,
@@ -598,6 +600,7 @@ where
type ActivatedTarget<E> = (SharedTarget<E>, Option<(mpsc::Sender<()>, JoinHandle<()>)>); type ActivatedTarget<E> = (SharedTarget<E>, Option<(mpsc::Sender<()>, JoinHandle<()>)>);
#[hotpath::measure]
pub async fn activate_targets_with_replay<E, F, Fut>( pub async fn activate_targets_with_replay<E, F, Fut>(
targets: Vec<Box<dyn Target<E> + Send + Sync>>, targets: Vec<Box<dyn Target<E> + Send + Sync>>,
mut activate_one: F, mut activate_one: F,
@@ -670,6 +673,7 @@ fn seed_interval_start(now: tokio::time::Instant, interval: Duration) -> tokio::
now.checked_sub(interval).unwrap_or(now) now.checked_sub(interval).unwrap_or(now)
} }
#[hotpath::measure]
async fn stream_replay_worker<E>( async fn stream_replay_worker<E>(
store: &mut (dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send), store: &mut (dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send),
target: SharedTarget<E>, target: SharedTarget<E>,
@@ -805,6 +809,7 @@ async fn stream_replay_worker<E>(
/// Returns `true` if a cancel signal was observed while processing (e.g. during /// Returns `true` if a cancel signal was observed while processing (e.g. during
/// retry backoff), so the caller can stop promptly instead of continuing to /// retry backoff), so the caller can stop promptly instead of continuing to
/// drain a store that a replacement worker may already own. /// drain a store that a replacement worker may already own.
#[hotpath::measure]
async fn process_replay_batch<E>( async fn process_replay_batch<E>(
store: &(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send), store: &(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send),
batch_keys: &mut Vec<Key>, batch_keys: &mut Vec<Key>,
+2
View File
@@ -85,6 +85,7 @@ fn classify_probe_error(err: &reqwest::Error) -> TargetHealthReason {
TargetHealthReason::Unreachable TargetHealthReason::Unreachable
} }
#[hotpath::measure]
async fn probe_health_url(client: &Client, health_check_url: &Url) -> TargetHealth { async fn probe_health_url(client: &Client, health_check_url: &Url) -> TargetHealth {
match tokio::time::timeout(WEBHOOK_HEALTH_TIMEOUT, client.head(health_check_url.as_str()).send()).await { match tokio::time::timeout(WEBHOOK_HEALTH_TIMEOUT, client.head(health_check_url.as_str()).send()).await {
Ok(Ok(_)) => TargetHealth::online(TargetHealthReason::Reachable), Ok(Ok(_)) => TargetHealth::online(TargetHealthReason::Reachable),
@@ -480,6 +481,7 @@ where
build_queued_payload(event) build_queued_payload(event)
} }
#[hotpath::measure]
async fn send_body(&self, body: Vec<u8>, meta: &QueuedPayloadMeta) -> Result<(), TargetError> { async fn send_body(&self, body: Vec<u8>, meta: &QueuedPayloadMeta) -> Result<(), TargetError> {
debug!( debug!(
event = EVENT_WEBHOOK_DELIVERY_STATE, event = EVENT_WEBHOOK_DELIVERY_STATE,
+25
View File
@@ -25,7 +25,32 @@ keywords = ["testing", "storage", "rustfs", "Minio"]
categories = ["development-tools", "filesystem"] categories = ["development-tools", "filesystem"]
documentation = "https://docs.rs/rustfs-test-utils/latest/rustfs_test_utils/" documentation = "https://docs.rs/rustfs-test-utils/latest/rustfs_test_utils/"
[features]
default = []
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
"rustfs-ecstore/hotpath",
"rustfs-storage-api/hotpath",
"rustfs-data-usage/hotpath",
]
hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc",
"rustfs-ecstore/hotpath-alloc",
"rustfs-storage-api/hotpath-alloc",
"rustfs-data-usage/hotpath-alloc",
]
hotpath-cpu = [
"hotpath",
"hotpath/hotpath-cpu",
"rustfs-ecstore/hotpath-cpu",
"rustfs-storage-api/hotpath-cpu",
"rustfs-data-usage/hotpath-cpu",
]
[dependencies] [dependencies]
hotpath.workspace = true
rustfs-ecstore = { workspace = true } rustfs-ecstore = { workspace = true }
rustfs-storage-api = { workspace = true } rustfs-storage-api = { workspace = true }
tokio = { workspace = true, features = ["fs", "rt-multi-thread"] } tokio = { workspace = true, features = ["fs", "rt-multi-thread"] }
+7
View File
@@ -27,7 +27,14 @@ categories = ["network-programming", "web-programming", "development-tools"]
[lints] [lints]
workspace = true workspace = true
[features]
default = []
hotpath = ["hotpath/hotpath", "hotpath/tokio", "rustfs-common/hotpath", "rustfs-config/hotpath"]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-common/hotpath-alloc", "rustfs-config/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-common/hotpath-cpu", "rustfs-config/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true
rustfs-common.workspace = true rustfs-common.workspace = true
rustfs-config.workspace = true rustfs-config.workspace = true
arc-swap.workspace = true arc-swap.workspace = true
+13
View File
@@ -24,7 +24,20 @@ description = " RustFS Trusted Proxies module provides secure and efficient mana
keywords = ["trusted-proxies", "network-security", "rustfs", "proxy-management"] keywords = ["trusted-proxies", "network-security", "rustfs", "proxy-management"]
categories = ["network-programming", "security", "web-programming"] categories = ["network-programming", "security", "web-programming"]
[features]
default = []
hotpath = [
"hotpath/hotpath",
"hotpath/tokio",
"hotpath/reqwest-0-13",
"rustfs-config/hotpath",
"rustfs-utils/hotpath",
]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-config/hotpath-alloc", "rustfs-utils/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-config/hotpath-cpu", "rustfs-utils/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true
async-trait = { workspace = true } async-trait = { workspace = true }
axum = { workspace = true } axum = { workspace = true }
http = { workspace = true } http = { workspace = true }
@@ -227,6 +227,7 @@ pub struct GoogleCloudIpRanges;
impl GoogleCloudIpRanges { impl GoogleCloudIpRanges {
/// Fetches the latest Google Cloud IP ranges from their official source. /// Fetches the latest Google Cloud IP ranges from their official source.
#[hotpath::measure]
pub async fn fetch() -> Result<Vec<IpNetwork>, AppError> { pub async fn fetch() -> Result<Vec<IpNetwork>, AppError> {
let client = Client::builder() let client = Client::builder()
.timeout(Duration::from_secs(10)) .timeout(Duration::from_secs(10))
+4
View File
@@ -25,6 +25,7 @@ keywords = ["utilities", "hashing", "compression", "network", "rustfs"]
categories = ["web-programming", "development-tools", "cryptography"] categories = ["web-programming", "development-tools", "cryptography"]
[dependencies] [dependencies]
hotpath.workspace = true
base64-simd = { workspace = true, optional = true } base64-simd = { workspace = true, optional = true }
blake2 = { workspace = true, optional = true } blake2 = { workspace = true, optional = true }
brotli = { workspace = true, optional = true } brotli = { workspace = true, optional = true }
@@ -76,6 +77,9 @@ workspace = true
[features] [features]
default = ["ip"] # features that are enabled by default default = ["ip"] # features that are enabled by default
hotpath = ["hotpath/hotpath", "hotpath/tokio", "hotpath/futures", "hotpath/reqwest-0-13"]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
ip = ["dep:local-ip-address"] # ip characteristics and their dependencies ip = ["dep:local-ip-address"] # ip characteristics and their dependencies
net = ["ip", "dep:url", "dep:netif", "dep:futures", "dep:transform-stream", "dep:bytes", "dep:hyper", "dep:tokio"] # network features with DNS resolver net = ["ip", "dep:url", "dep:netif", "dep:futures", "dep:transform-stream", "dep:bytes", "dep:hyper", "dep:tokio"] # network features with DNS resolver
egress = ["ip", "dep:reqwest", "dep:tokio", "dep:url"] egress = ["ip", "dep:reqwest", "dep:tokio", "dep:url"]
+7
View File
@@ -32,7 +32,14 @@ doctest = false
name = "zip_benchmark" name = "zip_benchmark"
harness = false harness = false
[features]
default = []
hotpath = ["hotpath/hotpath", "hotpath/tokio"]
hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc"]
hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu"]
[dependencies] [dependencies]
hotpath.workspace = true
async-compression = { workspace = true, features = [ async-compression = { workspace = true, features = [
"tokio", "tokio",
"bzip2", "bzip2",
+112
View File
@@ -60,25 +60,137 @@ hotpath = [
"hotpath/crossbeam", "hotpath/crossbeam",
"hotpath/parking_lot", "hotpath/parking_lot",
"hotpath/reqwest-0-13", "hotpath/reqwest-0-13",
"rustfs-audit/hotpath",
"rustfs-common/hotpath",
"rustfs-concurrency/hotpath",
"rustfs-config/hotpath",
"rustfs-credentials/hotpath",
"rustfs-crypto/hotpath",
"rustfs-data-usage/hotpath",
"rustfs-ecstore/hotpath", "rustfs-ecstore/hotpath",
"rustfs-extension-schema/hotpath",
"rustfs-filemeta/hotpath", "rustfs-filemeta/hotpath",
"rustfs-heal/hotpath",
"rustfs-iam/hotpath",
"rustfs-io-core/hotpath",
"rustfs-io-metrics/hotpath",
"rustfs-keystone/hotpath",
"rustfs-kms/hotpath",
"rustfs-lock/hotpath",
"rustfs-log-analyzer/hotpath",
"rustfs-madmin/hotpath",
"rustfs-notify/hotpath",
"rustfs-object-capacity/hotpath",
"rustfs-object-data-cache/hotpath",
"rustfs-obs/hotpath",
"rustfs-policy/hotpath", "rustfs-policy/hotpath",
"rustfs-protocols/hotpath",
"rustfs-protos/hotpath",
"rustfs-rio/hotpath", "rustfs-rio/hotpath",
"rustfs-s3-ops/hotpath",
"rustfs-s3-types/hotpath",
"rustfs-s3select-api/hotpath",
"rustfs-s3select-query/hotpath",
"rustfs-scanner/hotpath",
"rustfs-security-governance/hotpath",
"rustfs-signer/hotpath",
"rustfs-storage-api/hotpath",
"rustfs-targets/hotpath",
"rustfs-tls-runtime/hotpath",
"rustfs-trusted-proxies/hotpath",
"rustfs-utils/hotpath",
"rustfs-zip/hotpath",
"rustfs-test-utils/hotpath",
] ]
hotpath-alloc = [ hotpath-alloc = [
"hotpath",
"hotpath/hotpath-alloc", "hotpath/hotpath-alloc",
"rustfs-audit/hotpath-alloc",
"rustfs-common/hotpath-alloc",
"rustfs-concurrency/hotpath-alloc",
"rustfs-config/hotpath-alloc",
"rustfs-credentials/hotpath-alloc",
"rustfs-crypto/hotpath-alloc",
"rustfs-data-usage/hotpath-alloc",
"rustfs-ecstore/hotpath-alloc", "rustfs-ecstore/hotpath-alloc",
"rustfs-extension-schema/hotpath-alloc",
"rustfs-filemeta/hotpath-alloc", "rustfs-filemeta/hotpath-alloc",
"rustfs-heal/hotpath-alloc",
"rustfs-iam/hotpath-alloc",
"rustfs-io-core/hotpath-alloc",
"rustfs-io-metrics/hotpath-alloc",
"rustfs-keystone/hotpath-alloc",
"rustfs-kms/hotpath-alloc",
"rustfs-lock/hotpath-alloc",
"rustfs-log-analyzer/hotpath-alloc",
"rustfs-madmin/hotpath-alloc",
"rustfs-notify/hotpath-alloc",
"rustfs-object-capacity/hotpath-alloc",
"rustfs-object-data-cache/hotpath-alloc",
"rustfs-obs/hotpath-alloc",
"rustfs-policy/hotpath-alloc", "rustfs-policy/hotpath-alloc",
"rustfs-protocols/hotpath-alloc",
"rustfs-protos/hotpath-alloc",
"rustfs-rio/hotpath-alloc", "rustfs-rio/hotpath-alloc",
"rustfs-s3-ops/hotpath-alloc",
"rustfs-s3-types/hotpath-alloc",
"rustfs-s3select-api/hotpath-alloc",
"rustfs-s3select-query/hotpath-alloc",
"rustfs-scanner/hotpath-alloc",
"rustfs-security-governance/hotpath-alloc",
"rustfs-signer/hotpath-alloc",
"rustfs-storage-api/hotpath-alloc",
"rustfs-targets/hotpath-alloc",
"rustfs-tls-runtime/hotpath-alloc",
"rustfs-trusted-proxies/hotpath-alloc",
"rustfs-utils/hotpath-alloc",
"rustfs-zip/hotpath-alloc",
"rustfs-test-utils/hotpath-alloc",
] ]
hotpath-cpu = [ hotpath-cpu = [
"hotpath", "hotpath",
"hotpath/hotpath-cpu", "hotpath/hotpath-cpu",
"rustfs-audit/hotpath-cpu",
"rustfs-common/hotpath-cpu",
"rustfs-concurrency/hotpath-cpu",
"rustfs-config/hotpath-cpu",
"rustfs-credentials/hotpath-cpu",
"rustfs-crypto/hotpath-cpu",
"rustfs-data-usage/hotpath-cpu",
"rustfs-ecstore/hotpath-cpu", "rustfs-ecstore/hotpath-cpu",
"rustfs-extension-schema/hotpath-cpu",
"rustfs-filemeta/hotpath-cpu", "rustfs-filemeta/hotpath-cpu",
"rustfs-heal/hotpath-cpu",
"rustfs-iam/hotpath-cpu",
"rustfs-io-core/hotpath-cpu",
"rustfs-io-metrics/hotpath-cpu",
"rustfs-keystone/hotpath-cpu",
"rustfs-kms/hotpath-cpu",
"rustfs-lock/hotpath-cpu",
"rustfs-log-analyzer/hotpath-cpu",
"rustfs-madmin/hotpath-cpu",
"rustfs-notify/hotpath-cpu",
"rustfs-object-capacity/hotpath-cpu",
"rustfs-object-data-cache/hotpath-cpu",
"rustfs-obs/hotpath-cpu",
"rustfs-policy/hotpath-cpu", "rustfs-policy/hotpath-cpu",
"rustfs-protocols/hotpath-cpu",
"rustfs-protos/hotpath-cpu",
"rustfs-rio/hotpath-cpu", "rustfs-rio/hotpath-cpu",
"rustfs-s3-ops/hotpath-cpu",
"rustfs-s3-types/hotpath-cpu",
"rustfs-s3select-api/hotpath-cpu",
"rustfs-s3select-query/hotpath-cpu",
"rustfs-scanner/hotpath-cpu",
"rustfs-security-governance/hotpath-cpu",
"rustfs-signer/hotpath-cpu",
"rustfs-storage-api/hotpath-cpu",
"rustfs-targets/hotpath-cpu",
"rustfs-tls-runtime/hotpath-cpu",
"rustfs-trusted-proxies/hotpath-cpu",
"rustfs-utils/hotpath-cpu",
"rustfs-zip/hotpath-cpu",
"rustfs-test-utils/hotpath-cpu",
] ]
[lints] [lints]
+428
View File
@@ -0,0 +1,428 @@
#!/usr/bin/env bash
# Formal Linux / production-cluster ABBA runner for the hotpath warp matrix.
#
# This script is intentionally a thin orchestrator around the existing
# run_object_batch_bench_enhanced.sh load driver and hotpath_warp_ab_gate.sh
# relative-budget gate. It runs each durability/workload cell as:
#
# A1 baseline -> B1 candidate -> B2 candidate -> A2 baseline
#
# Candidate legs are compared against A1. The final A2 leg is also compared
# against A1 to quantify baseline drift separately from candidate deltas.
set -euo pipefail
PROJECT_ROOT="$(git rev-parse --show-toplevel)"
ENHANCED_BENCH="${PROJECT_ROOT}/scripts/run_object_batch_bench_enhanced.sh"
GATE="${PROJECT_ROOT}/scripts/hotpath_warp_ab_gate.sh"
BASELINE_BIN=""
CANDIDATE_BIN=""
ENDPOINT=""
DEPLOY_HOOK=""
HEALTH_PATH="/health"
ADDRESS="127.0.0.1:9000"
DATA_ROOT="/tmp/rustfs-hotpath-abba"
DISKS=4
ACCESS_KEY="rustfsadmin"
SECRET_KEY="rustfsadmin"
REGION="us-east-1"
WARP_BIN="warp"
CONCURRENCY=8
DURATION="60s"
ROUNDS=3
COOLDOWN_SECS=20
HEALTH_TIMEOUT_SECS=180
FAIL_PCT=10
WARN_PCT=5
ALLOW_REGRESSION=false
EXEMPTION_REASON="deliberate correctness tradeoff"
OUT_DIR="${PROJECT_ROOT}/target/hotpath-abba/$(date -u +%Y%m%dT%H%M%SZ 2>/dev/null || echo run)"
DRY_RUN=false
WORKLOADS=(
"put-4kib|put|4KiB"
"put-4mib|put|4MiB"
"get-4kib|get|4KiB"
"get-4mib|get|4MiB"
"get-10mib|get|10MiB"
"mixed-256k|mixed|256KiB"
)
DRIVE_SYNC_MATRIX=("sync-on|true" "sync-off|false")
usage() {
cat <<'USAGE'
Usage: scripts/run_hotpath_warp_abba.sh --baseline-bin <path> --candidate-bin <path> [options]
Formal ABBA mode for Linux runners or production-like clusters. The schedule is
A1 baseline -> B1 candidate -> B2 candidate -> A2 baseline for every workload
and drive-sync cell.
Required:
--baseline-bin <path> Baseline RustFS binary.
--candidate-bin <path> Candidate RustFS binary.
Local Linux runner mode:
--address <host:port> Local RustFS address (default 127.0.0.1:9000).
--disks <n> Throwaway local disks per node (default 4).
--data-root <path> Local disk root (default /tmp/rustfs-hotpath-abba).
Production / cluster mode:
--endpoint <host:port> Existing cluster endpoint. Enables external mode.
--deploy-hook <cmd> Command run before each ABBA leg. It receives:
HOTPATH_ABBA_LEG=A1|B1|B2|A2
HOTPATH_ABBA_PHASE=baseline|candidate
HOTPATH_ABBA_BINARY=<baseline/candidate binary>
HOTPATH_ABBA_DRIVE_SYNC=true|false
--health-path <path> Readiness path (default /health).
Benchmark:
--duration <dur> warp duration per cell (default 60s).
--rounds <n> rounds per cell; must be >= 3 (default 3).
--cooldown <n> cooldown seconds between rounds/sizes (default 20).
--concurrency <n> warp concurrency (default 8).
--warp-bin <path> warp binary (default warp).
Credentials:
--access-key <value> S3 access key (default rustfsadmin).
--secret-key <value> S3 secret key (default rustfsadmin).
--region <value> S3 region (default us-east-1).
Gate:
--fail-pct <n> Regression budget that fails gate (default 10).
--warn-pct <n> Regression budget that warns (default 5).
--allow-regression Downgrade candidate gate FAIL to WARN.
--exemption-reason <s> Reason recorded when allow-regression is used.
Output:
--out-dir <path> Output dir (default target/hotpath-abba/<ts>).
--dry-run Print commands without starting servers or warp.
-h, --help
Outputs:
<out-dir>/abba_schedule.csv
<out-dir>/candidate_gate.md
<out-dir>/baseline_drift_gate.md
<out-dir>/summary.md
<out-dir>/<workload>/<sync>/<leg>/{median_summary.csv,baseline_compare.csv}
USAGE
}
die() {
echo "error: $*" >&2
exit 2
}
log() {
printf '[hotpath-abba] %s\n' "$*" >&2
}
run() {
if [[ "$DRY_RUN" == "true" ]]; then
{ printf 'DRY-RUN:'; printf ' %q' "$@"; printf '\n'; } >&2
return 0
fi
"$@"
}
validate_positive_int() {
local value="$1" name="$2"
[[ "$value" =~ ^[0-9]+$ && "$value" -gt 0 ]] || die "$name must be a positive integer"
}
while [[ $# -gt 0 ]]; do
case "$1" in
--baseline-bin) BASELINE_BIN="$2"; shift 2 ;;
--candidate-bin) CANDIDATE_BIN="$2"; shift 2 ;;
--endpoint) ENDPOINT="$2"; shift 2 ;;
--deploy-hook) DEPLOY_HOOK="$2"; shift 2 ;;
--health-path) HEALTH_PATH="$2"; shift 2 ;;
--address) ADDRESS="$2"; shift 2 ;;
--data-root) DATA_ROOT="$2"; shift 2 ;;
--disks) DISKS="$2"; shift 2 ;;
--access-key) ACCESS_KEY="$2"; shift 2 ;;
--secret-key) SECRET_KEY="$2"; shift 2 ;;
--region) REGION="$2"; shift 2 ;;
--warp-bin) WARP_BIN="$2"; shift 2 ;;
--concurrency) CONCURRENCY="$2"; shift 2 ;;
--duration) DURATION="$2"; shift 2 ;;
--rounds) ROUNDS="$2"; shift 2 ;;
--cooldown) COOLDOWN_SECS="$2"; shift 2 ;;
--health-timeout) HEALTH_TIMEOUT_SECS="$2"; shift 2 ;;
--fail-pct) FAIL_PCT="$2"; shift 2 ;;
--warn-pct) WARN_PCT="$2"; shift 2 ;;
--allow-regression) ALLOW_REGRESSION=true; shift ;;
--exemption-reason) EXEMPTION_REASON="$2"; shift 2 ;;
--out-dir) OUT_DIR="$2"; shift 2 ;;
--dry-run) DRY_RUN=true; shift ;;
-h|--help) usage; exit 0 ;;
*) die "unknown argument: $1" ;;
esac
done
validate_positive_int "$DISKS" "--disks"
validate_positive_int "$CONCURRENCY" "--concurrency"
validate_positive_int "$ROUNDS" "--rounds"
validate_positive_int "$COOLDOWN_SECS" "--cooldown"
validate_positive_int "$HEALTH_TIMEOUT_SECS" "--health-timeout"
validate_positive_int "$FAIL_PCT" "--fail-pct"
validate_positive_int "$WARN_PCT" "--warn-pct"
[[ "$ROUNDS" -ge 3 ]] || die "--rounds must be >= 3 for formal ABBA evidence"
[[ -n "$BASELINE_BIN" ]] || die "--baseline-bin is required"
[[ -n "$CANDIDATE_BIN" ]] || die "--candidate-bin is required"
[[ "$DRY_RUN" == "true" || -x "$BASELINE_BIN" ]] || die "baseline binary is not executable: $BASELINE_BIN"
[[ "$DRY_RUN" == "true" || -x "$CANDIDATE_BIN" ]] || die "candidate binary is not executable: $CANDIDATE_BIN"
[[ -x "$ENHANCED_BENCH" ]] || die "missing load driver: $ENHANCED_BENCH"
[[ -x "$GATE" ]] || die "missing gate: $GATE"
if [[ "$DRY_RUN" != "true" ]] && ! command -v "$WARP_BIN" >/dev/null 2>&1; then
die "warp not found on PATH; install warp or pass --warp-bin"
fi
EXTERNAL=false
if [[ -n "$ENDPOINT" ]]; then
EXTERNAL=true
ADDRESS="$ENDPOINT"
[[ -n "$DEPLOY_HOOK" ]] || log "warning: external mode without --deploy-hook; binaries must be swapped out of band"
fi
mkdir -p "$OUT_DIR"
SERVER_LOG_DIR="$OUT_DIR/server-logs"
run mkdir -p "$SERVER_LOG_DIR"
SERVER_PID=""
SERVER_LOG=""
tear_down() {
[[ "$EXTERNAL" == "true" ]] && return 0
[[ -n "$SERVER_PID" ]] || return 0
run kill "$SERVER_PID" 2>/dev/null || true
SERVER_PID=""
}
trap tear_down EXIT INT TERM
dump_server_log() {
[[ -n "$SERVER_LOG" && -f "$SERVER_LOG" ]] || return 0
echo "----- last 80 lines of $SERVER_LOG -----" >&2
tail -n 80 "$SERVER_LOG" >&2 || true
echo "----------------------------------------" >&2
}
wait_health() {
[[ "$DRY_RUN" == "true" ]] && return 0
local i
for ((i = 0; i < HEALTH_TIMEOUT_SECS; i++)); do
if [[ "$EXTERNAL" != "true" && -n "$SERVER_PID" ]] && ! kill -0 "$SERVER_PID" 2>/dev/null; then
echo "error: rustfs server (pid $SERVER_PID) exited before becoming healthy after ${i}s" >&2
dump_server_log
return 1
fi
if curl -fsS "http://${ADDRESS}${HEALTH_PATH}" >/dev/null 2>&1; then
return 0
fi
sleep 1
done
echo "error: endpoint http://${ADDRESS}${HEALTH_PATH} did not become healthy within ${HEALTH_TIMEOUT_SECS}s" >&2
dump_server_log
return 1
}
binary_for_leg() {
case "$1" in
A1|A2) echo "$BASELINE_BIN" ;;
B1|B2) echo "$CANDIDATE_BIN" ;;
*) die "unknown ABBA leg: $1" ;;
esac
}
phase_for_leg() {
case "$1" in
A1|A2) echo "baseline" ;;
B1|B2) echo "candidate" ;;
*) die "unknown ABBA leg: $1" ;;
esac
}
bring_up() {
local leg="$1" drive_sync="$2"
local phase bin
phase="$(phase_for_leg "$leg")"
bin="$(binary_for_leg "$leg")"
if [[ "$EXTERNAL" == "true" ]]; then
if [[ -n "$DEPLOY_HOOK" ]]; then
log "deploy hook: leg=$leg phase=$phase drive_sync=$drive_sync"
HOTPATH_ABBA_LEG="$leg" HOTPATH_ABBA_PHASE="$phase" HOTPATH_ABBA_BINARY="$bin" HOTPATH_ABBA_DRIVE_SYNC="$drive_sync" \
run bash -c "$DEPLOY_HOOK"
fi
wait_health
return 0
fi
local node_dir="$DATA_ROOT/$leg-sync-$drive_sync"
local disks=() d
for ((d = 1; d <= DISKS; d++)); do
disks+=("$node_dir/d$d")
done
run mkdir -p "${disks[@]}"
SERVER_LOG="$SERVER_LOG_DIR/$leg-sync-$drive_sync.log"
if [[ "$DRY_RUN" == "true" ]]; then
{ printf 'DRY-RUN: RUSTFS_DRIVE_SYNC_ENABLE=%s %q server' "$drive_sync" "$bin"
printf ' %q' "${disks[@]}"; printf '\n'; } >&2
SERVER_PID="dry-run"
return 0
fi
cat >"$SERVER_LOG_DIR/$leg-sync-$drive_sync.env" <<EOF
leg=$leg
phase=$phase
drive_sync=$drive_sync
binary=$bin
address=$ADDRESS
disks=${disks[*]}
health_url=http://${ADDRESS}${HEALTH_PATH}
health_timeout_secs=$HEALTH_TIMEOUT_SECS
uname=$(uname -a 2>/dev/null || echo unknown)
warp_version=$("$WARP_BIN" --version 2>/dev/null | head -n1 || echo unknown)
EOF
RUSTFS_UNSAFE_BYPASS_DISK_CHECK=true \
RUSTFS_ADDRESS="$ADDRESS" \
RUSTFS_ACCESS_KEY="$ACCESS_KEY" \
RUSTFS_SECRET_KEY="$SECRET_KEY" \
RUSTFS_REGION="$REGION" \
RUSTFS_CONSOLE_ENABLE=false \
RUSTFS_DRIVE_SYNC_ENABLE="$drive_sync" \
"$bin" server "${disks[@]}" >"$SERVER_LOG" 2>&1 &
SERVER_PID=$!
wait_health
}
measure() {
local leg="$1" workload="$2" mode="$3" size="$4" sync_label="$5" baseline_csv="${6:-}"
local cell="$OUT_DIR/$workload/$sync_label/$leg"
local args=(
--tool warp --warp-bin "$WARP_BIN" --warp-mode "$mode"
--endpoint "$ADDRESS" --access-key "$ACCESS_KEY" --secret-key "$SECRET_KEY"
--region "$REGION" --sizes "$size" --concurrency "$CONCURRENCY"
--duration "$DURATION" --rounds "$ROUNDS" --cooldown-secs "$COOLDOWN_SECS"
--out-dir "$cell"
)
[[ -n "$baseline_csv" ]] && args+=(--baseline-csv "$baseline_csv")
run "$ENHANCED_BENCH" "${args[@]}" >&2
echo "$cell"
}
write_schedule_header() {
echo "sync_label,drive_sync,workload,mode,size,leg,phase,binary,out_dir" >"$OUT_DIR/abba_schedule.csv"
}
append_schedule() {
local sync_label="$1" drive_sync="$2" workload="$3" mode="$4" size="$5" leg="$6"
local phase bin
phase="$(phase_for_leg "$leg")"
bin="$(binary_for_leg "$leg")"
echo "$sync_label,$drive_sync,$workload,$mode,$size,$leg,$phase,$bin,$OUT_DIR/$workload/$sync_label/$leg" >>"$OUT_DIR/abba_schedule.csv"
}
write_manifest() {
cat >"$OUT_DIR/manifest.env" <<EOF
generated_at_utc=$(date -u +%Y-%m-%dT%H:%M:%SZ 2>/dev/null || echo unknown)
runner=$(uname -srm 2>/dev/null || echo unknown)
schedule=ABBA
rounds=$ROUNDS
duration=$DURATION
cooldown_secs=$COOLDOWN_SECS
concurrency=$CONCURRENCY
baseline_bin=$BASELINE_BIN
candidate_bin=$CANDIDATE_BIN
external=$EXTERNAL
endpoint=$ADDRESS
warp_version=$("$WARP_BIN" --version 2>/dev/null | head -n1 || echo unknown)
EOF
}
declare -a CANDIDATE_COMPARE_CSVS=()
declare -a DRIFT_COMPARE_CSVS=()
write_manifest
write_schedule_header
for ds_spec in "${DRIVE_SYNC_MATRIX[@]}"; do
IFS='|' read -r sync_label drive_sync <<<"$ds_spec"
for leg in A1 B1 B2 A2; do
log "=== $sync_label leg $leg ($(phase_for_leg "$leg")) ==="
bring_up "$leg" "$drive_sync"
for wl_spec in "${WORKLOADS[@]}"; do
IFS='|' read -r workload mode size <<<"$wl_spec"
append_schedule "$sync_label" "$drive_sync" "$workload" "$mode" "$size" "$leg"
baseline_csv=""
if [[ "$leg" != "A1" ]]; then
baseline_csv="$OUT_DIR/$workload/$sync_label/A1/median_summary.csv"
fi
cell="$(measure "$leg" "$workload" "$mode" "$size" "$sync_label" "$baseline_csv")"
case "$leg" in
B1|B2) CANDIDATE_COMPARE_CSVS+=("$cell/baseline_compare.csv") ;;
A2) DRIFT_COMPARE_CSVS+=("$cell/baseline_compare.csv") ;;
esac
done
tear_down
done
done
gate_args=(--fail-pct "$FAIL_PCT" --warn-pct "$WARN_PCT" --markdown "$OUT_DIR/candidate_gate.md")
for csv in "${CANDIDATE_COMPARE_CSVS[@]}"; do
gate_args+=(--compare-csv "$csv")
done
[[ "$ALLOW_REGRESSION" == "true" ]] && gate_args+=(--allow-regression --exemption-reason "$EXEMPTION_REASON")
drift_gate_args=(--fail-pct "$FAIL_PCT" --warn-pct "$WARN_PCT" --markdown "$OUT_DIR/baseline_drift_gate.md")
for csv in "${DRIFT_COMPARE_CSVS[@]}"; do
drift_gate_args+=(--compare-csv "$csv")
done
if [[ "$DRY_RUN" == "true" ]]; then
log "dry-run complete; candidate compare CSVs=${#CANDIDATE_COMPARE_CSVS[@]} baseline drift CSVs=${#DRIFT_COMPARE_CSVS[@]}"
{ printf 'DRY-RUN:'; printf ' %q' "$GATE" "${gate_args[@]}"; printf '\n'; } >&2
{ printf 'DRY-RUN:'; printf ' %q' "$GATE" "${drift_gate_args[@]}"; printf '\n'; } >&2
exit 0
fi
log "applying candidate relative-budget gate"
set +e
"$GATE" "${gate_args[@]}"
candidate_status=$?
set -e
log "applying A2-vs-A1 baseline drift gate"
set +e
"$GATE" "${drift_gate_args[@]}"
drift_status=$?
set -e
cat >"$OUT_DIR/summary.md" <<EOF
# Hotpath Warp ABBA Summary
- schedule: A1 baseline -> B1 candidate -> B2 candidate -> A2 baseline
- runner: $(uname -srm 2>/dev/null || echo unknown)
- warp: $("$WARP_BIN" --version 2>/dev/null | head -n1 || echo unknown)
- matrix: duration=$DURATION rounds=$ROUNDS cooldown=$COOLDOWN_SECS disks=$DISKS concurrency=$CONCURRENCY
- endpoint: $ADDRESS
- baseline binary: $BASELINE_BIN
- candidate binary: $CANDIDATE_BIN
- candidate gate: $OUT_DIR/candidate_gate.md (exit $candidate_status)
- baseline drift gate: $OUT_DIR/baseline_drift_gate.md (exit $drift_status)
Interpretation:
- Treat candidate gate failures as actionable only when the A2-vs-A1 drift gate is PASS or the affected workload's A2 drift is materially smaller than the B1/B2 candidate delta.
- If both candidate and baseline drift fail on the same workload, rerun with longer duration, more rounds, or a quieter runner before assigning causality.
EOF
log "summary written to $OUT_DIR/summary.md"
if [[ "$candidate_status" -ne 0 || "$drift_status" -ne 0 ]]; then
exit 1
fi