mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-01 02:52:15 +00:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 overtrue
|
3926d108f2 | ||
|
overtrue
|
9a2c3a0a1d |
@@ -12,20 +12,18 @@
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
# Companion to ci.yml for the required "Test and Lint" and "Quick Checks"
|
||||
# status checks.
|
||||
# Companion to ci.yml for the required "Test and Lint" status check.
|
||||
#
|
||||
# ci.yml skips docs-only pull requests via paths-ignore, but the branch
|
||||
# ruleset requires checks named "Test and Lint" and "Quick Checks" — without
|
||||
# this workflow a docs-only PR would wait on those checks forever. This
|
||||
# workflow triggers on exactly the paths ci.yml ignores and reports success
|
||||
# under the same job names. Mixed PRs trigger both workflows and the real
|
||||
# checks still gate: a required check with any failing run blocks the merge.
|
||||
# ruleset requires a check named "Test and Lint" — without this workflow a
|
||||
# docs-only PR would wait on that check forever. This workflow triggers on
|
||||
# exactly the paths ci.yml ignores and reports an instant success under the
|
||||
# same job name. Mixed PRs trigger both workflows and the real check still
|
||||
# gates: a required check with any failing run blocks the merge.
|
||||
# https://docs.github.com/en/repositories/configuring-branches-and-merges-in-your-repository/defining-the-mergeability-of-pull-requests/troubleshooting-required-status-checks#handling-skipped-but-required-checks
|
||||
#
|
||||
# Keep the paths list below in sync with the pull_request paths-ignore list
|
||||
# in ci.yml, and keep the quick-checks steps below byte-identical to the
|
||||
# quick-checks job in ci.yml (see the comment on that job).
|
||||
# in ci.yml.
|
||||
|
||||
name: Continuous Integration (docs only)
|
||||
|
||||
@@ -54,63 +52,9 @@ permissions:
|
||||
contents: read
|
||||
|
||||
jobs:
|
||||
# Deliberately NOT a bare `echo`. Once "Quick Checks" becomes a required
|
||||
# check, ci.yml gates every expensive job behind it, so a mixed PR reports
|
||||
# two check runs with this name: the real one (45-51s) and this companion.
|
||||
# GitHub has no written contract for how it picks between same-named
|
||||
# required check runs ("latest wins" vs "any failure blocks"), so instead of
|
||||
# relying on ordering we make both runs execute the same commands against
|
||||
# the same merge ref — their conclusions are then necessarily identical and
|
||||
# the choice does not matter. Keep these steps byte-identical to the
|
||||
# quick-checks job in ci.yml (a guard script that asserts this, and the paths
|
||||
# sync below, is tracked in rustfs/backlog#1603).
|
||||
#
|
||||
# For a genuinely docs-only PR this adds no strictness (no code changed, so
|
||||
# fmt and the guards always pass) and costs ~50s of ubuntu-latest.
|
||||
quick-checks:
|
||||
name: Quick Checks
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 10
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
|
||||
|
||||
- name: Install ripgrep
|
||||
run: sudo apt-get update && sudo apt-get install -y ripgrep
|
||||
|
||||
- name: Install Rust toolchain
|
||||
uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable
|
||||
with:
|
||||
components: rustfmt
|
||||
|
||||
- name: Check code formatting
|
||||
run: cargo fmt --all --check
|
||||
|
||||
- name: Check unsafe code allowances
|
||||
run: ./scripts/check_unsafe_code_allowances.sh
|
||||
|
||||
- name: Check layered dependencies
|
||||
run: ./scripts/check_layer_dependencies.sh
|
||||
|
||||
- name: Check architecture migration rules
|
||||
run: ./scripts/check_architecture_migration_rules.sh
|
||||
|
||||
- name: Check tokio io-uring feature guard
|
||||
run: ./scripts/check_no_tokio_io_uring.sh
|
||||
|
||||
- name: Check extension schema boundaries
|
||||
run: ./scripts/check_extension_schema_boundaries.sh
|
||||
|
||||
- name: Check body-cache whitelist guard
|
||||
run: ./scripts/check_body_cache_whitelist.sh
|
||||
|
||||
- name: Check no planning docs committed
|
||||
run: ./scripts/check_no_planning_docs.sh
|
||||
|
||||
test-and-lint:
|
||||
name: Test and Lint
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 10
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
|
||||
|
||||
@@ -140,26 +140,13 @@ jobs:
|
||||
test-and-lint:
|
||||
name: Test and Lint
|
||||
if: github.event_name != 'pull_request' || github.event.action != 'closed'
|
||||
needs: [ quick-checks ]
|
||||
runs-on: sm-standard-4
|
||||
timeout-minutes: 90
|
||||
# Both lines are required. Job-level `permissions` replaces the workflow
|
||||
# block rather than merging with it, so declaring only `actions: write`
|
||||
# would drop `contents: read` and break this job's checkout and the
|
||||
# repo-token the setup action hands to setup-protoc.
|
||||
permissions:
|
||||
contents: read
|
||||
actions: write
|
||||
env:
|
||||
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
|
||||
with:
|
||||
# This job's token can cancel runs and delete Actions caches. Checkout
|
||||
# otherwise writes it into .git/config, where a PR's own build.rs or
|
||||
# proc-macro could read it back out.
|
||||
persist-credentials: false
|
||||
|
||||
- name: Setup Rust environment
|
||||
uses: ./.github/actions/setup
|
||||
@@ -290,48 +277,6 @@ jobs:
|
||||
- name: Run rebalance/decommission migration proofs
|
||||
run: ./scripts/check_migration_gate_count.sh
|
||||
|
||||
# Early stop. Once this job has failed the PR cannot merge, so the sibling
|
||||
# lanes are burning runners on a result nobody can act on: on run
|
||||
# 30674613104 three lanes had already failed while Test and Lint and the
|
||||
# rio-v2 variant kept going past 70 minutes.
|
||||
#
|
||||
# Only this job may cancel. The lanes that are NOT required checks
|
||||
# (protocols, ILM, e2e, s3-tests) must never hold that power: a flake in
|
||||
# one of them would turn the required "Test and Lint" into `cancelled`,
|
||||
# which blocks the merge. Today a maintainer can merge with sftp red, and
|
||||
# that has to stay true.
|
||||
#
|
||||
# These steps run last so the `if: always()` artifact upload above still
|
||||
# captures logs and diagnostics before the run goes away.
|
||||
- name: Annotate early-stop reason
|
||||
if: failure() && github.event_name == 'pull_request'
|
||||
run: |
|
||||
echo "## CI early-stop" >> "$GITHUB_STEP_SUMMARY"
|
||||
echo "Job \`${GITHUB_JOB}\` (Test and Lint) failed; cancelling run ${GITHUB_RUN_ID} to free runners." >> "$GITHUB_STEP_SUMMARY"
|
||||
echo "Sibling jobs showing **cancelled** were stopped by this job, not by their own failure." >> "$GITHUB_STEP_SUMMARY"
|
||||
|
||||
# curl rather than `gh`: every existing `gh` call in this repo runs on
|
||||
# ubuntu-latest, and the sm-standard-* images are custom and trimmed (they
|
||||
# ship no C toolchain, see the e2e job below), so `gh` is not known to
|
||||
# exist here.
|
||||
#
|
||||
# Fork PRs are excluded explicitly instead of relying on the error path:
|
||||
# their GITHUB_TOKEN is forced read-only and job-level permissions cannot
|
||||
# raise it, so the call would always 403. Skipping keeps their logs clean.
|
||||
- name: Cancel run on failure (same-repo PR only)
|
||||
if: >-
|
||||
failure() && github.event_name == 'pull_request'
|
||||
&& github.event.pull_request.head.repo.full_name == github.repository
|
||||
continue-on-error: true
|
||||
env:
|
||||
GH_TOKEN: ${{ secrets.GITHUB_TOKEN }}
|
||||
run: |
|
||||
curl -fsS -X POST \
|
||||
-H "Authorization: Bearer ${GH_TOKEN}" \
|
||||
-H "Accept: application/vnd.github+json" \
|
||||
-H "X-GitHub-Api-Version: 2022-11-28" \
|
||||
"${GITHUB_API_URL}/repos/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}/cancel" || true
|
||||
|
||||
# Dedicated serial lane for the ILM / lifecycle integration tests. These tests
|
||||
# drive the object layer through process-global singletons (the GLOBAL_ENV
|
||||
# ECStore, the global tier-config manager, background-expiry workers) and bind
|
||||
@@ -345,7 +290,6 @@ jobs:
|
||||
test-ilm-integration-serial:
|
||||
name: ILM Integration (serial)
|
||||
if: github.event_name != 'pull_request' || github.event.action != 'closed'
|
||||
needs: [ quick-checks ]
|
||||
runs-on: sm-standard-4
|
||||
timeout-minutes: 45
|
||||
env:
|
||||
@@ -383,7 +327,6 @@ jobs:
|
||||
test-and-lint-rio-v2:
|
||||
name: Test and Lint (rio-v2)
|
||||
if: github.event_name != 'pull_request' || github.event.action != 'closed'
|
||||
needs: [ quick-checks ]
|
||||
runs-on: sm-standard-4
|
||||
timeout-minutes: 60
|
||||
env:
|
||||
@@ -411,17 +354,10 @@ jobs:
|
||||
test-and-lint-protocols:
|
||||
name: "Test and Lint (${{ matrix.features.name }})"
|
||||
if: github.event_name != 'pull_request' || github.event.action != 'closed'
|
||||
needs: [ quick-checks ]
|
||||
runs-on: sm-standard-4
|
||||
timeout-minutes: 60
|
||||
strategy:
|
||||
# On a PR, one failing protocol leg is enough to know the PR is not ready,
|
||||
# so stop the sibling leg instead of paying another ~40 minutes for it.
|
||||
# Everywhere else (main pushes, the merge queue, the weekly schedule) keep
|
||||
# the full signal: there we want to know whether swift AND sftp are broken,
|
||||
# not just whichever failed first. This is the only part of the early-stop
|
||||
# work that also covers fork PRs, since it needs no token.
|
||||
fail-fast: ${{ github.event_name == 'pull_request' }}
|
||||
fail-fast: false
|
||||
matrix:
|
||||
features:
|
||||
- name: swift
|
||||
@@ -453,7 +389,6 @@ jobs:
|
||||
build-rustfs-debug-binary:
|
||||
name: Build RustFS Debug Binary
|
||||
if: github.event_name != 'pull_request' || github.event.action != 'closed'
|
||||
needs: [ quick-checks ]
|
||||
runs-on: sm-standard-4
|
||||
timeout-minutes: 30
|
||||
env:
|
||||
@@ -484,7 +419,6 @@ jobs:
|
||||
build-rustfs-debug-binary-rio-v2:
|
||||
name: Build RustFS Debug Binary (rio-v2)
|
||||
if: github.event_name != 'pull_request' || github.event.action != 'closed'
|
||||
needs: [ quick-checks ]
|
||||
runs-on: sm-standard-4
|
||||
timeout-minutes: 30
|
||||
env:
|
||||
@@ -514,14 +448,6 @@ jobs:
|
||||
|
||||
uring-integration:
|
||||
name: io_uring Integration (real)
|
||||
# The pull_request trigger includes `closed` purely so the concurrency
|
||||
# group cancels in-flight runs of a closed PR; every other job opts out of
|
||||
# that run with this guard (or is skipped through its `needs` chain). This
|
||||
# job had neither, so each closed/merged PR really ran the whole io_uring
|
||||
# suite (measured 4m17s / 7m19s / 7m31s on runs 30678272341 / 30678117601 /
|
||||
# 30662728539) and kept the cancellation run in progress for minutes.
|
||||
if: github.event_name != 'pull_request' || github.event.action != 'closed'
|
||||
needs: [ quick-checks ]
|
||||
# GitHub-hosted ubuntu-latest runs a recent kernel with io_uring and, unlike
|
||||
# a container, applies no seccomp filter that would block io_uring_setup — so
|
||||
# the probe succeeds and the tests exercise the real UringBackend/FdCache/
|
||||
@@ -820,13 +746,7 @@ jobs:
|
||||
# evaluates ILM within ~2s of the due time, well inside the poll window.
|
||||
s3-lifecycle-behavior-tests:
|
||||
name: S3 Lifecycle Behavior Tests
|
||||
# Also gated on e2e-tests, matching s3-implemented-tests: when the e2e smoke
|
||||
# suite is already red this lane cannot tell us anything new, and it holds a
|
||||
# sm-standard-4 for up to 30 minutes doing so. Both lanes only download the
|
||||
# prebuilt debug binary (no cargo build), and s3-implemented-tests — which
|
||||
# already waits on e2e-tests — finishes later anyway, so a green PR's total
|
||||
# wall clock is unchanged.
|
||||
needs: [ build-rustfs-debug-binary, e2e-tests ]
|
||||
needs: [ build-rustfs-debug-binary ]
|
||||
runs-on: sm-standard-4
|
||||
timeout-minutes: 30
|
||||
steps:
|
||||
|
||||
@@ -26,7 +26,6 @@
|
||||
|
||||
use super::common::*;
|
||||
use aws_sdk_s3::Client;
|
||||
use aws_sdk_s3::error::ProvideErrorMetadata;
|
||||
use aws_sdk_s3::primitives::{ByteStream, DateTimeFormat};
|
||||
use aws_sdk_s3::types::{
|
||||
CompletedMultipartUpload, CompletedPart, Delete, MetadataDirective, ObjectIdentifier, ObjectLockLegalHoldStatus,
|
||||
@@ -2121,127 +2120,6 @@ async fn test_multipart_default_retention_fixed_at_create() {
|
||||
// Versioning Auto-Enable Tests
|
||||
// ============================================================================
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn test_unretained_object_lock_object_delete_and_bucket_cleanup() {
|
||||
init_logging();
|
||||
info!("🧪 Test: Unretained Object Lock object delete and bucket cleanup (Issue #5339)");
|
||||
|
||||
let mut env = ObjectLockTestEnvironment::new()
|
||||
.await
|
||||
.expect("failed to create Object Lock test environment");
|
||||
env.start_rustfs().await.expect("failed to start RustFS");
|
||||
|
||||
let bucket = "test-object-lock-delete-cleanup";
|
||||
let key = "unretained-object";
|
||||
|
||||
env.create_object_lock_bucket(bucket)
|
||||
.await
|
||||
.expect("failed to create Object Lock bucket");
|
||||
let client = env.s3_client();
|
||||
|
||||
let put_response = client
|
||||
.put_object()
|
||||
.bucket(bucket)
|
||||
.key(key)
|
||||
.body(ByteStream::from_static(b"unretained data"))
|
||||
.send()
|
||||
.await
|
||||
.expect("failed to upload unretained object");
|
||||
let object_version_id = put_response
|
||||
.version_id()
|
||||
.expect("Object Lock buckets must create versioned objects")
|
||||
.to_string();
|
||||
|
||||
let delete_response = client
|
||||
.delete_object()
|
||||
.bucket(bucket)
|
||||
.key(key)
|
||||
.send()
|
||||
.await
|
||||
.expect("failed to create delete marker");
|
||||
assert_eq!(delete_response.delete_marker(), Some(true));
|
||||
let delete_marker_version_id = delete_response
|
||||
.version_id()
|
||||
.expect("Deleting without a version ID must create a delete marker")
|
||||
.to_string();
|
||||
|
||||
let get_error = client
|
||||
.get_object()
|
||||
.bucket(bucket)
|
||||
.key(key)
|
||||
.send()
|
||||
.await
|
||||
.expect_err("GET must not return an object hidden by a delete marker");
|
||||
assert_eq!(get_error.raw_response().map(|response| response.status().as_u16()), Some(404));
|
||||
assert_eq!(get_error.as_service_error().and_then(|error| error.code()), Some("NoSuchKey"));
|
||||
|
||||
let listed_objects = client
|
||||
.list_objects_v2()
|
||||
.bucket(bucket)
|
||||
.send()
|
||||
.await
|
||||
.expect("failed to list current objects");
|
||||
assert!(
|
||||
listed_objects.contents().iter().all(|object| object.key() != Some(key)),
|
||||
"ListObjectsV2 must hide objects whose latest version is a delete marker"
|
||||
);
|
||||
|
||||
let listed_versions = client
|
||||
.list_object_versions()
|
||||
.bucket(bucket)
|
||||
.send()
|
||||
.await
|
||||
.expect("failed to list object versions");
|
||||
assert!(
|
||||
listed_versions
|
||||
.versions()
|
||||
.iter()
|
||||
.any(|version| version.key() == Some(key) && version.version_id() == Some(object_version_id.as_str())),
|
||||
"The data version must remain until it is explicitly deleted"
|
||||
);
|
||||
assert!(
|
||||
listed_versions
|
||||
.delete_markers()
|
||||
.iter()
|
||||
.any(|marker| marker.key() == Some(key) && marker.version_id() == Some(delete_marker_version_id.as_str())),
|
||||
"ListObjectVersions must expose the delete marker"
|
||||
);
|
||||
|
||||
client
|
||||
.delete_object()
|
||||
.bucket(bucket)
|
||||
.key(key)
|
||||
.version_id(object_version_id)
|
||||
.send()
|
||||
.await
|
||||
.expect("failed to delete the data version");
|
||||
client
|
||||
.delete_object()
|
||||
.bucket(bucket)
|
||||
.key(key)
|
||||
.version_id(delete_marker_version_id)
|
||||
.send()
|
||||
.await
|
||||
.expect("failed to delete the delete marker");
|
||||
|
||||
let remaining_versions = client
|
||||
.list_object_versions()
|
||||
.bucket(bucket)
|
||||
.send()
|
||||
.await
|
||||
.expect("failed to list versions after cleanup");
|
||||
assert!(remaining_versions.versions().is_empty());
|
||||
assert!(remaining_versions.delete_markers().is_empty());
|
||||
|
||||
client
|
||||
.delete_bucket()
|
||||
.bucket(bucket)
|
||||
.send()
|
||||
.await
|
||||
.expect("Deleting every version must remove xl.meta so the bucket can be deleted normally");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn test_versioning_auto_enabled_with_object_lock() {
|
||||
|
||||
@@ -227,10 +227,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
|
||||
// Step 12: Check cache statistics
|
||||
println!("12. Checking cache statistics...");
|
||||
if let Some((hits, misses)) = encryption_service.cache_stats().await {
|
||||
if let Some(stats) = encryption_service.cache_stats().await {
|
||||
println!(" ✓ Cache statistics:");
|
||||
println!(" - Cache hits: {}", hits);
|
||||
println!(" - Cache misses: {}\n", misses);
|
||||
println!(" - Cached entries: {}", stats.entries);
|
||||
println!(" - Cache hits: {}", stats.hits);
|
||||
println!(" - Cache misses: {}", stats.misses);
|
||||
println!(" - Entries evicted: {}\n", stats.evictions);
|
||||
} else {
|
||||
println!(" - Cache is disabled\n");
|
||||
}
|
||||
|
||||
@@ -263,10 +263,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
|
||||
|
||||
// Step 12: Check cache statistics
|
||||
println!("12. Checking cache statistics...");
|
||||
if let Some((hits, misses)) = encryption_service.cache_stats().await {
|
||||
if let Some(stats) = encryption_service.cache_stats().await {
|
||||
println!(" ✓ Cache statistics:");
|
||||
println!(" - Cache hits: {}", hits);
|
||||
println!(" - Cache misses: {}\n", misses);
|
||||
println!(" - Cached entries: {}", stats.entries);
|
||||
println!(" - Cache hits: {}", stats.hits);
|
||||
println!(" - Cache misses: {}", stats.misses);
|
||||
println!(" - Entries evicted: {}\n", stats.evictions);
|
||||
} else {
|
||||
println!(" - Cache is disabled\n");
|
||||
}
|
||||
|
||||
+254
-38
@@ -16,11 +16,82 @@
|
||||
|
||||
use crate::types::KeyMetadata;
|
||||
use moka::future::Cache;
|
||||
use moka::notification::RemovalCause;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::time::Duration;
|
||||
|
||||
/// Default lifetime of a cached key metadata entry.
|
||||
const DEFAULT_METADATA_TTL: Duration = Duration::from_secs(300);
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Metrics
|
||||
//
|
||||
// Lookup outcomes and removals are counted here because moka exposes neither
|
||||
// hit nor miss counts; the numbers below are the cache's own, not derived.
|
||||
// Label values are exclusively static strings (lookup result, removal cause) —
|
||||
// key identifiers and key metadata must never reach a metric label.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Counter: key metadata lookups, by `result` (`hit` or `miss`).
|
||||
const METRIC_CACHE_LOOKUPS_TOTAL: &str = "rustfs_kms_metadata_cache_lookups_total";
|
||||
/// Counter: entries dropped from the cache, by `cause` (`expired`, `size`,
|
||||
/// `explicit`, `replaced`); only `expired` and `size` are true evictions, the
|
||||
/// rest are invalidations driven by key lifecycle operations.
|
||||
const METRIC_CACHE_EVICTIONS_TOTAL: &str = "rustfs_kms_metadata_cache_evictions_total";
|
||||
/// Gauge: entries currently held by the cache.
|
||||
const METRIC_CACHE_ENTRIES: &str = "rustfs_kms_metadata_cache_entries";
|
||||
|
||||
/// 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_CACHE_LOOKUPS_TOTAL, "Total KMS key metadata cache lookups, by hit or miss");
|
||||
metrics::describe_counter!(
|
||||
METRIC_CACHE_EVICTIONS_TOTAL,
|
||||
"Total entries dropped from the KMS key metadata cache, by removal cause"
|
||||
);
|
||||
metrics::describe_gauge!(METRIC_CACHE_ENTRIES, "Entries currently held by the KMS key metadata cache");
|
||||
});
|
||||
}
|
||||
|
||||
fn removal_cause_label(cause: RemovalCause) -> &'static str {
|
||||
match cause {
|
||||
RemovalCause::Expired => "expired",
|
||||
RemovalCause::Explicit => "explicit",
|
||||
RemovalCause::Replaced => "replaced",
|
||||
RemovalCause::Size => "size",
|
||||
}
|
||||
}
|
||||
|
||||
/// Cumulative cache counters. Shared with moka's eviction listener, which runs
|
||||
/// outside any `&self` borrow, hence the `Arc` and the atomics.
|
||||
#[derive(Debug, Default)]
|
||||
struct CacheCounters {
|
||||
hits: AtomicU64,
|
||||
misses: AtomicU64,
|
||||
evictions: AtomicU64,
|
||||
}
|
||||
|
||||
/// Snapshot of the KMS key metadata cache counters.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub struct KmsCacheStats {
|
||||
/// Entries currently held. Eventually consistent: moka applies pending
|
||||
/// maintenance lazily.
|
||||
pub entries: u64,
|
||||
/// Lookups served from the cache since process start.
|
||||
pub hits: u64,
|
||||
/// Lookups that fell through to the backend since process start.
|
||||
pub misses: u64,
|
||||
/// Entries dropped since process start, whatever the cause (expiry,
|
||||
/// capacity, invalidation, replacement).
|
||||
pub evictions: u64,
|
||||
}
|
||||
|
||||
/// KMS cache for storing frequently accessed keys and metadata
|
||||
pub struct KmsCache {
|
||||
key_metadata_cache: Cache<String, KeyMetadata>,
|
||||
counters: Arc<CacheCounters>,
|
||||
}
|
||||
|
||||
impl KmsCache {
|
||||
@@ -33,11 +104,26 @@ impl KmsCache {
|
||||
/// A new instance of `KmsCache`
|
||||
///
|
||||
pub fn new(capacity: u64) -> Self {
|
||||
Self::with_ttl(capacity, DEFAULT_METADATA_TTL)
|
||||
}
|
||||
|
||||
/// Create a new KMS cache with an explicit metadata time-to-live
|
||||
fn with_ttl(capacity: u64, metadata_ttl: Duration) -> Self {
|
||||
describe_metrics();
|
||||
|
||||
let counters = Arc::new(CacheCounters::default());
|
||||
let eviction_counters = Arc::clone(&counters);
|
||||
|
||||
Self {
|
||||
key_metadata_cache: Cache::builder()
|
||||
.max_capacity(capacity)
|
||||
.time_to_live(Duration::from_secs(300)) // 5 minutes default TTL
|
||||
.time_to_live(metadata_ttl)
|
||||
.eviction_listener(move |_key: Arc<String>, _metadata: KeyMetadata, cause: RemovalCause| {
|
||||
eviction_counters.evictions.fetch_add(1, Ordering::Relaxed);
|
||||
metrics::counter!(METRIC_CACHE_EVICTIONS_TOTAL, "cause" => removal_cause_label(cause)).increment(1);
|
||||
})
|
||||
.build(),
|
||||
counters,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -50,7 +136,9 @@ impl KmsCache {
|
||||
/// An `Option` containing the `KeyMetadata` if found, or `None` if not found
|
||||
///
|
||||
pub async fn get_key_metadata(&self, key_id: &str) -> Option<KeyMetadata> {
|
||||
self.key_metadata_cache.get(key_id).await
|
||||
let cached = self.key_metadata_cache.get(key_id).await;
|
||||
self.record_lookup(cached.is_some());
|
||||
cached
|
||||
}
|
||||
|
||||
/// Put key metadata into cache
|
||||
@@ -62,6 +150,7 @@ impl KmsCache {
|
||||
pub async fn put_key_metadata(&mut self, key_id: &str, metadata: &KeyMetadata) {
|
||||
self.key_metadata_cache.insert(key_id.to_string(), metadata.clone()).await;
|
||||
self.key_metadata_cache.run_pending_tasks().await;
|
||||
self.record_entry_count();
|
||||
}
|
||||
|
||||
/// Remove key metadata from cache
|
||||
@@ -71,6 +160,11 @@ impl KmsCache {
|
||||
///
|
||||
pub async fn remove_key_metadata(&mut self, key_id: &str) {
|
||||
self.key_metadata_cache.remove(key_id).await;
|
||||
|
||||
// Flush maintenance so the removal notification and the entry gauge
|
||||
// describe the cache as it is once this call returns.
|
||||
self.key_metadata_cache.run_pending_tasks().await;
|
||||
self.record_entry_count();
|
||||
}
|
||||
|
||||
/// Clear all cached entries
|
||||
@@ -79,18 +173,35 @@ impl KmsCache {
|
||||
|
||||
// Wait for invalidation to complete
|
||||
self.key_metadata_cache.run_pending_tasks().await;
|
||||
self.record_entry_count();
|
||||
}
|
||||
|
||||
/// Get cache statistics (hit count, miss count)
|
||||
/// Get cache statistics
|
||||
///
|
||||
/// # Returns
|
||||
/// A tuple containing total entries and total misses
|
||||
/// A [`KmsCacheStats`] snapshot of entry count, hits, misses and evictions
|
||||
///
|
||||
pub fn stats(&self) -> (u64, u64) {
|
||||
(
|
||||
self.key_metadata_cache.entry_count(),
|
||||
0u64, // moka doesn't provide miss count directly
|
||||
)
|
||||
pub fn stats(&self) -> KmsCacheStats {
|
||||
KmsCacheStats {
|
||||
entries: self.key_metadata_cache.entry_count(),
|
||||
hits: self.counters.hits.load(Ordering::Relaxed),
|
||||
misses: self.counters.misses.load(Ordering::Relaxed),
|
||||
evictions: self.counters.evictions.load(Ordering::Relaxed),
|
||||
}
|
||||
}
|
||||
|
||||
fn record_lookup(&self, hit: bool) {
|
||||
let (counter, result) = if hit {
|
||||
(&self.counters.hits, "hit")
|
||||
} else {
|
||||
(&self.counters.misses, "miss")
|
||||
};
|
||||
counter.fetch_add(1, Ordering::Relaxed);
|
||||
metrics::counter!(METRIC_CACHE_LOOKUPS_TOTAL, "result" => result).increment(1);
|
||||
}
|
||||
|
||||
fn record_entry_count(&self) {
|
||||
metrics::gauge!(METRIC_CACHE_ENTRIES).set(self.key_metadata_cache.entry_count() as f64);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -99,37 +210,90 @@ mod tests {
|
||||
use super::*;
|
||||
use crate::types::{KeyState, KeyUsage};
|
||||
use jiff::Zoned;
|
||||
use metrics_util::MetricKind;
|
||||
use metrics_util::debugging::{DebugValue, DebuggingRecorder};
|
||||
use std::time::Duration;
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
struct CacheInfo {
|
||||
key_metadata_count: u64,
|
||||
}
|
||||
|
||||
impl CacheInfo {
|
||||
fn total_entries(&self) -> u64 {
|
||||
self.key_metadata_count
|
||||
}
|
||||
}
|
||||
|
||||
impl KmsCache {
|
||||
fn with_ttl_for_tests(capacity: u64, metadata_ttl: Duration) -> Self {
|
||||
Self {
|
||||
key_metadata_cache: Cache::builder().max_capacity(capacity).time_to_live(metadata_ttl).build(),
|
||||
}
|
||||
}
|
||||
|
||||
fn info_for_tests(&self) -> CacheInfo {
|
||||
CacheInfo {
|
||||
key_metadata_count: self.key_metadata_cache.entry_count(),
|
||||
}
|
||||
}
|
||||
|
||||
fn contains_key_metadata_for_tests(&self, key_id: &str) -> bool {
|
||||
self.key_metadata_cache.contains_key(key_id)
|
||||
}
|
||||
}
|
||||
|
||||
fn test_metadata(key_id: &str) -> KeyMetadata {
|
||||
KeyMetadata {
|
||||
key_id: key_id.to_string(),
|
||||
key_state: KeyState::Enabled,
|
||||
key_usage: KeyUsage::EncryptDecrypt,
|
||||
description: None,
|
||||
creation_date: Zoned::now(),
|
||||
deletion_date: None,
|
||||
origin: "KMS".to_string(),
|
||||
key_manager: "CUSTOMER".to_string(),
|
||||
tags: std::collections::HashMap::new(),
|
||||
}
|
||||
}
|
||||
|
||||
type MetricEntry = (
|
||||
metrics_util::CompositeKey,
|
||||
Option<metrics::Unit>,
|
||||
Option<metrics::SharedString>,
|
||||
DebugValue,
|
||||
);
|
||||
|
||||
/// Drive `test` on a current-thread runtime under a thread-local debugging
|
||||
/// recorder and return its output plus one snapshot of everything emitted.
|
||||
///
|
||||
/// A single snapshot per test on purpose: `Snapshotter::snapshot` drains
|
||||
/// the recorded state, so taking it per assertion would leave every
|
||||
/// assertion after the first with nothing to look at.
|
||||
fn record_metrics<F, Fut, T>(test: F) -> (T, Vec<MetricEntry>)
|
||||
where
|
||||
F: FnOnce() -> Fut,
|
||||
Fut: std::future::Future<Output = T>,
|
||||
{
|
||||
let recorder = DebuggingRecorder::new();
|
||||
let snapshotter = recorder.snapshotter();
|
||||
let output = 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())
|
||||
});
|
||||
(output, 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()
|
||||
}
|
||||
|
||||
/// Last value recorded for the unlabelled gauge `name`.
|
||||
fn gauge_value(snapshot: &[MetricEntry], name: &str) -> Option<f64> {
|
||||
snapshot.iter().find_map(|(composite, _unit, _description, value)| {
|
||||
let matches = composite.kind() == MetricKind::Gauge && composite.key().name() == name;
|
||||
match (matches, value) {
|
||||
(true, DebugValue::Gauge(gauge)) => Some(**gauge),
|
||||
_ => None,
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_cache_operations() {
|
||||
let mut cache = KmsCache::new(100);
|
||||
@@ -154,19 +318,16 @@ mod tests {
|
||||
assert_eq!(retrieved.expect("metadata should be cached").key_id, "test-key-1");
|
||||
|
||||
// Test cache info
|
||||
let info = cache.info_for_tests();
|
||||
assert_eq!(info.key_metadata_count, 1);
|
||||
assert_eq!(info.total_entries(), 1);
|
||||
assert_eq!(cache.stats().entries, 1);
|
||||
|
||||
// Test cache clearing
|
||||
cache.clear().await;
|
||||
let info_after_clear = cache.info_for_tests();
|
||||
assert_eq!(info_after_clear.total_entries(), 0);
|
||||
assert_eq!(cache.stats().entries, 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_cache_with_custom_ttl() {
|
||||
let mut cache = KmsCache::with_ttl_for_tests(
|
||||
let mut cache = KmsCache::with_ttl(
|
||||
100,
|
||||
Duration::from_millis(100), // Short TTL for testing
|
||||
);
|
||||
@@ -217,4 +378,59 @@ mod tests {
|
||||
|
||||
assert!(cache.contains_key_metadata_for_tests("contains-test"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn lookups_report_real_hit_and_miss_counts() {
|
||||
let (stats, snapshot) = record_metrics(|| async {
|
||||
let mut cache = KmsCache::new(100);
|
||||
|
||||
assert!(cache.get_key_metadata("absent").await.is_none());
|
||||
cache.put_key_metadata("present", &test_metadata("present")).await;
|
||||
assert!(cache.get_key_metadata("present").await.is_some());
|
||||
assert!(cache.get_key_metadata("absent").await.is_none());
|
||||
|
||||
cache.stats()
|
||||
});
|
||||
|
||||
assert_eq!(stats.hits, 1);
|
||||
assert_eq!(stats.misses, 2);
|
||||
assert_eq!(counter_value(&snapshot, METRIC_CACHE_LOOKUPS_TOTAL, &[("result", "hit")]), 1);
|
||||
assert_eq!(counter_value(&snapshot, METRIC_CACHE_LOOKUPS_TOTAL, &[("result", "miss")]), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn removals_report_their_cause_and_the_resulting_entry_count() {
|
||||
let (stats, snapshot) = record_metrics(|| async {
|
||||
let mut cache = KmsCache::new(100);
|
||||
|
||||
cache.put_key_metadata("key", &test_metadata("key")).await;
|
||||
cache.put_key_metadata("key", &test_metadata("key")).await;
|
||||
cache.remove_key_metadata("key").await;
|
||||
|
||||
cache.stats()
|
||||
});
|
||||
|
||||
assert_eq!(counter_value(&snapshot, METRIC_CACHE_EVICTIONS_TOTAL, &[("cause", "replaced")]), 1);
|
||||
assert_eq!(counter_value(&snapshot, METRIC_CACHE_EVICTIONS_TOTAL, &[("cause", "explicit")]), 1);
|
||||
assert_eq!(stats.evictions, 2);
|
||||
assert_eq!(stats.entries, 0);
|
||||
assert_eq!(gauge_value(&snapshot, METRIC_CACHE_ENTRIES), Some(0.0));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn capacity_pressure_reports_size_evictions() {
|
||||
let (stats, snapshot) = record_metrics(|| async {
|
||||
let mut cache = KmsCache::with_ttl(1, DEFAULT_METADATA_TTL);
|
||||
|
||||
cache.put_key_metadata("first", &test_metadata("first")).await;
|
||||
cache.put_key_metadata("second", &test_metadata("second")).await;
|
||||
|
||||
cache.stats()
|
||||
});
|
||||
|
||||
assert_eq!(counter_value(&snapshot, METRIC_CACHE_EVICTIONS_TOTAL, &[("cause", "size")]), 1);
|
||||
assert_eq!(stats.evictions, 1);
|
||||
assert_eq!(stats.entries, 1);
|
||||
assert_eq!(gauge_value(&snapshot, METRIC_CACHE_ENTRIES), Some(1.0));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -86,6 +86,7 @@ pub use api_types::{
|
||||
StartKmsResponse, StopKmsResponse, TagKeyRequest, TagKeyResponse, UntagKeyRequest, UntagKeyResponse,
|
||||
UpdateKeyDescriptionRequest, UpdateKeyDescriptionResponse,
|
||||
};
|
||||
pub use cache::KmsCacheStats;
|
||||
pub use config::*;
|
||||
pub use deletion_worker::DeletionReferenceChecker;
|
||||
pub use encryption::is_data_key_envelope;
|
||||
|
||||
@@ -15,7 +15,7 @@
|
||||
//! KMS manager for handling key operations and backend coordination
|
||||
|
||||
use crate::backends::KmsBackend;
|
||||
use crate::cache::KmsCache;
|
||||
use crate::cache::{KmsCache, KmsCacheStats};
|
||||
use crate::config::KmsConfig;
|
||||
use crate::error::Result;
|
||||
use crate::types::{
|
||||
@@ -113,8 +113,8 @@ impl KmsManager {
|
||||
self.backend.list_keys(request).await
|
||||
}
|
||||
|
||||
/// Get cache statistics
|
||||
pub async fn cache_stats(&self) -> Option<(u64, u64)> {
|
||||
/// Get cache statistics, or `None` when caching is disabled
|
||||
pub async fn cache_stats(&self) -> Option<KmsCacheStats> {
|
||||
if self.enable_cache {
|
||||
let cache = self.cache.read().await;
|
||||
Some(cache.stats())
|
||||
@@ -254,9 +254,12 @@ mod tests {
|
||||
let describe_response = manager.describe_key(describe_request).await.expect("Failed to describe key");
|
||||
assert_eq!(describe_response.key_metadata.key_id, create_response.key_id);
|
||||
|
||||
// Test cache stats
|
||||
let stats = manager.cache_stats().await;
|
||||
assert!(stats.is_some());
|
||||
// Creating the key populated the cache, so the describe above was
|
||||
// served from it rather than from the backend.
|
||||
let stats = manager.cache_stats().await.expect("cache is enabled");
|
||||
assert_eq!(stats.entries, 1);
|
||||
assert_eq!(stats.hits, 1);
|
||||
assert_eq!(stats.misses, 0);
|
||||
|
||||
// Test health check
|
||||
let health = manager.health_check().await.expect("Health check failed");
|
||||
|
||||
@@ -14,6 +14,7 @@
|
||||
|
||||
//! Object encryption service for S3-compatible encryption
|
||||
|
||||
use crate::cache::KmsCacheStats;
|
||||
use crate::encryption::ciphers::{create_cipher, generate_iv};
|
||||
use crate::error::{KmsError, Result};
|
||||
use crate::manager::KmsManager;
|
||||
@@ -160,9 +161,9 @@ impl ObjectEncryptionService {
|
||||
/// Get cache statistics
|
||||
///
|
||||
/// # Returns
|
||||
/// Option with (hits, misses) if caching is enabled
|
||||
/// A [`KmsCacheStats`] snapshot if caching is enabled, `None` otherwise
|
||||
///
|
||||
pub async fn cache_stats(&self) -> Option<(u64, u64)> {
|
||||
pub async fn cache_stats(&self) -> Option<KmsCacheStats> {
|
||||
self.kms_manager.cache_stats().await
|
||||
}
|
||||
|
||||
|
||||
@@ -82,6 +82,14 @@ pub struct KmsStatusResponse {
|
||||
pub struct CacheStatsResponse {
|
||||
pub hit_count: u64,
|
||||
pub miss_count: u64,
|
||||
/// Entries currently cached. Additive field: omitted by older servers, so
|
||||
/// it must stay defaulted for consumers.
|
||||
#[serde(default)]
|
||||
pub entry_count: u64,
|
||||
/// Entries dropped since process start, whatever the cause. Additive
|
||||
/// field, same compatibility rule as `entry_count`.
|
||||
#[serde(default)]
|
||||
pub eviction_count: u64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
@@ -193,9 +201,11 @@ impl Operation for KmsStatusHandler {
|
||||
}
|
||||
};
|
||||
|
||||
let cache_stats = service.cache_stats().await.map(|(hits, misses)| CacheStatsResponse {
|
||||
hit_count: hits,
|
||||
miss_count: misses,
|
||||
let cache_stats = service.cache_stats().await.map(|stats| CacheStatsResponse {
|
||||
hit_count: stats.hits,
|
||||
miss_count: stats.misses,
|
||||
entry_count: stats.entries,
|
||||
eviction_count: stats.evictions,
|
||||
});
|
||||
let config = kms_service_manager_from_context().get_redacted_config().await;
|
||||
|
||||
|
||||
+1
-36
@@ -12,32 +12,9 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
#[cfg(all(feature = "hotpath", feature = "hotpath-alloc"))]
|
||||
use std::alloc::{GlobalAlloc, Layout};
|
||||
|
||||
#[cfg(all(feature = "hotpath", feature = "hotpath-alloc"))]
|
||||
#[derive(Default)]
|
||||
struct DefaultMiMalloc;
|
||||
|
||||
#[cfg(all(feature = "hotpath", feature = "hotpath-alloc"))]
|
||||
// SAFETY: allocation and deallocation are forwarded unchanged to MiMalloc, so
|
||||
// MiMalloc's GlobalAlloc guarantees apply to every returned pointer and layout.
|
||||
#[allow(unsafe_code)]
|
||||
unsafe impl GlobalAlloc for DefaultMiMalloc {
|
||||
unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
|
||||
// SAFETY: the caller upholds GlobalAlloc's contract for layout.
|
||||
unsafe { mimalloc::MiMalloc.alloc(layout) }
|
||||
}
|
||||
|
||||
unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
|
||||
// SAFETY: ptr and layout came from this allocator and are forwarded unchanged.
|
||||
unsafe { mimalloc::MiMalloc.dealloc(ptr, layout) }
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(all(feature = "hotpath", feature = "hotpath-alloc"))]
|
||||
#[global_allocator]
|
||||
static GLOBAL: hotpath::CountingAllocator<DefaultMiMalloc> = hotpath::CountingAllocator::new();
|
||||
static GLOBAL: hotpath::CountingAllocator = hotpath::CountingAllocator::new();
|
||||
|
||||
#[cfg(not(all(feature = "hotpath", feature = "hotpath-alloc")))]
|
||||
#[global_allocator]
|
||||
@@ -48,15 +25,3 @@ fn main() {
|
||||
|
||||
rustfs::startup_entrypoint::run_process();
|
||||
}
|
||||
|
||||
#[cfg(all(test, feature = "hotpath", feature = "hotpath-alloc"))]
|
||||
mod tests {
|
||||
#[test]
|
||||
#[allow(unsafe_code)]
|
||||
fn hotpath_allocator_uses_mimalloc() {
|
||||
let allocation = Box::new([0_u8; 64]);
|
||||
|
||||
// SAFETY: the live Box pointer is valid to inspect for heap ownership.
|
||||
assert!(unsafe { libmimalloc_sys::mi_is_in_heap_region(allocation.as_ptr().cast()) });
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user