mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-29 23:37:04 +00:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 881a663765 | |||
| 495e1d3b87 |
@@ -1,224 +1,107 @@
|
||||
# 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.
|
||||
|
||||
name: Windows Filesystem Tests
|
||||
|
||||
name: Scanner cache read origin preflight
|
||||
on:
|
||||
push:
|
||||
branches: [ main ]
|
||||
paths:
|
||||
- "crates/ecstore/src/disk/**"
|
||||
- "crates/ecstore/src/layout/**"
|
||||
- "crates/ecstore/src/store/init_format.rs"
|
||||
- "crates/ecstore/src/set_disk/ops/locking.rs"
|
||||
- "crates/ecstore/src/object_api/write_commit_context.rs"
|
||||
- "crates/ecstore/src/bucket/utils.rs"
|
||||
- "crates/ecstore/Cargo.toml"
|
||||
- "crates/utils/src/dunce.rs"
|
||||
- "crates/utils/src/os/**"
|
||||
- "crates/utils/Cargo.toml"
|
||||
- "crates/utils/src/path.rs"
|
||||
- "rustfs/build.rs"
|
||||
- "rustfs/Cargo.toml"
|
||||
- "rustfs/src/main.rs"
|
||||
- "rustfs/src/server/runtime.rs"
|
||||
- "rustfs/src/startup_*.rs"
|
||||
- "Cargo.toml"
|
||||
- "Cargo.lock"
|
||||
- ".github/actions/setup/**"
|
||||
- ".github/workflows/windows-filesystem.yml"
|
||||
pull_request:
|
||||
branches: [ main ]
|
||||
paths:
|
||||
- "crates/ecstore/src/disk/**"
|
||||
- "crates/ecstore/src/layout/**"
|
||||
- "crates/ecstore/src/store/init_format.rs"
|
||||
- "crates/ecstore/src/set_disk/ops/locking.rs"
|
||||
- "crates/ecstore/src/object_api/write_commit_context.rs"
|
||||
- "crates/ecstore/src/bucket/utils.rs"
|
||||
- "crates/ecstore/Cargo.toml"
|
||||
- "crates/utils/src/dunce.rs"
|
||||
- "crates/utils/src/os/**"
|
||||
- "crates/utils/Cargo.toml"
|
||||
- "crates/utils/src/path.rs"
|
||||
- "rustfs/build.rs"
|
||||
- "rustfs/Cargo.toml"
|
||||
- "rustfs/src/main.rs"
|
||||
- "rustfs/src/server/runtime.rs"
|
||||
- "rustfs/src/startup_*.rs"
|
||||
- "Cargo.toml"
|
||||
- "Cargo.lock"
|
||||
- ".github/actions/setup/**"
|
||||
- ".github/workflows/windows-filesystem.yml"
|
||||
workflow_dispatch:
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
||||
concurrency:
|
||||
group: ${{ github.workflow }}-${{ github.event.pull_request.number || github.ref }}
|
||||
cancel-in-progress: true
|
||||
|
||||
group: scanner-cache-origin-${{ github.ref }}
|
||||
cancel-in-progress: false
|
||||
env:
|
||||
CARGO_TERM_COLOR: always
|
||||
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
|
||||
RUST_BACKTRACE: 1
|
||||
|
||||
CARGO_TERM_COLOR: never
|
||||
CARGO_INCREMENTAL: '0'
|
||||
CARGO_BUILD_JOBS: '1'
|
||||
RUST_BACKTRACE: '0'
|
||||
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: 'true'
|
||||
NO_PROXY: 127.0.0.1,localhost
|
||||
jobs:
|
||||
rename-safety:
|
||||
name: Rename Safety
|
||||
runs-on: windows-latest
|
||||
timeout-minutes: 60
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
|
||||
with:
|
||||
persist-credentials: false
|
||||
|
||||
- name: Setup Rust environment
|
||||
uses: ./.github/actions/setup
|
||||
with:
|
||||
rust-version: stable
|
||||
cache-shared-key: build-x86_64-pc-windows-msvc
|
||||
cache-save-if: 'false'
|
||||
install-build-packaging-tools: 'false'
|
||||
install-test-tools: 'false'
|
||||
|
||||
- name: Check production Windows dependencies
|
||||
shell: pwsh
|
||||
run: cargo check -p rustfs-ecstore --lib
|
||||
|
||||
- name: Test guarded rename publication
|
||||
shell: pwsh
|
||||
run: cargo test -p rustfs-ecstore --lib rename_all_ -- --nocapture
|
||||
|
||||
- name: Test multipart parent durability barriers
|
||||
shell: pwsh
|
||||
run: cargo test -p rustfs-ecstore --lib multipart_owned_record_parent_barriers_ -- --nocapture
|
||||
|
||||
- name: Test Windows path conversion
|
||||
shell: pwsh
|
||||
run: "cargo test -p rustfs-utils --no-default-features --features path --lib dunce:: -- --nocapture"
|
||||
|
||||
- name: Test Windows volume and disk-space probes
|
||||
shell: pwsh
|
||||
run: |
|
||||
# Map a drive letter to the temp drive's administrative share so the
|
||||
# remote-volume probe also covers mapped network drives.
|
||||
$drive = (Split-Path -Qualifier ([System.IO.Path]::GetTempPath())).TrimEnd(':')
|
||||
net use R: "\\localhost\$drive`$" /persistent:no
|
||||
if ($LASTEXITCODE -eq 0) {
|
||||
$env:RUSTFS_TEST_WINDOWS_MAPPED_DRIVE = 'R:'
|
||||
} else {
|
||||
Write-Output "::warning::could not map R: to \\localhost\$drive`$; mapped-drive test will skip"
|
||||
}
|
||||
cargo test -p rustfs-utils --features os --lib os:: -- --nocapture
|
||||
$status = $LASTEXITCODE
|
||||
if ($env:RUSTFS_TEST_WINDOWS_MAPPED_DRIVE) { net use R: /delete /y | Out-Null }
|
||||
exit $status
|
||||
|
||||
- name: Test Windows handle guards
|
||||
shell: pwsh
|
||||
run: cargo test -p rustfs-ecstore --lib windows_ -- --nocapture
|
||||
|
||||
- name: Test endpoint path parsing
|
||||
shell: pwsh
|
||||
run: "cargo test -p rustfs-ecstore --lib layout::endpoint:: -- --nocapture"
|
||||
|
||||
- name: Test startup temporary-directory cleanup
|
||||
shell: pwsh
|
||||
run: cargo test -p rustfs-ecstore --lib cleanup_tmp_on_startup_ -- --nocapture
|
||||
|
||||
- name: Test object key Windows admission
|
||||
shell: pwsh
|
||||
run: "cargo test -p rustfs-ecstore --lib bucket::utils:: -- --nocapture"
|
||||
|
||||
- name: Test Windows file-name rules
|
||||
shell: pwsh
|
||||
run: "cargo test -p rustfs-utils --no-default-features --features path --lib dunce:: -- --nocapture"
|
||||
|
||||
- name: Test fresh format publication
|
||||
shell: pwsh
|
||||
run: cargo test -p rustfs-ecstore --lib fresh_format_load_initializes_all_disks -- --nocapture
|
||||
|
||||
# The native Windows main thread only gets the 1 MiB PE stack reserve, while
|
||||
# Linux and macOS hand out 8 MiB. Startup runs the whole root future on the
|
||||
# thread that calls `block_on`, so a debug server once overflowed before it
|
||||
# listened on Windows only. Start a real server to keep that covered.
|
||||
startup-smoke:
|
||||
name: Startup Smoke
|
||||
runs-on: windows-latest
|
||||
scoped-tests:
|
||||
runs-on: ubuntu-latest
|
||||
timeout-minutes: 90
|
||||
steps:
|
||||
- name: Checkout repository
|
||||
uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
|
||||
- name: Checkout
|
||||
uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a
|
||||
with:
|
||||
fetch-depth: 3
|
||||
persist-credentials: false
|
||||
|
||||
- name: Setup Rust environment
|
||||
uses: ./.github/actions/setup
|
||||
clean: false
|
||||
- name: Verify source snapshot
|
||||
shell: bash
|
||||
run: |
|
||||
set -euo pipefail
|
||||
workflow_sha=$(git rev-parse HEAD)
|
||||
source_sha=$(git rev-parse HEAD^)
|
||||
base_sha=$(git rev-parse HEAD^^)
|
||||
source_tree=$(git rev-parse "${source_sha}^{tree}")
|
||||
source_diff=$(git diff --full-index --binary "$base_sha" "$source_sha" | sha256sum | cut -d' ' -f1)
|
||||
source_files=$(git diff --name-only "$base_sha" "$source_sha")
|
||||
workflow_files=$(git diff --name-only "$source_sha" "$workflow_sha")
|
||||
expected_files=$(printf '%s\n' 'crates/scanner/src/data_usage_define.rs' 'crates/scanner/src/data_usage_define/persistence.rs' 'crates/scanner/src/data_usage_define/tests.rs' 'crates/scanner/src/data_usage_define/tests/save_metrics.rs')
|
||||
printf 'head=%s source=%s base=%s tree=%s diff=%s\n' "$workflow_sha" "$source_sha" "$base_sha" "$source_tree" "$source_diff"
|
||||
[[ "$workflow_sha" == "$GITHUB_SHA" ]]
|
||||
[[ "$source_sha" == '495e1d3b873cdd2be2f9ffd6eb76e095010832b9' ]]
|
||||
[[ "$base_sha" == '7a3bf8cdbd781116a0970585b995ade668c0e662' ]]
|
||||
[[ "$source_tree" == 'c8bd60069f74095d5de1efaee916e177041a9e2e' ]]
|
||||
[[ "$source_diff" == '0a433216abecfc91416729e77cf5b16b45ffb04400701216b6aaf92321eeaf82' ]]
|
||||
[[ "$source_files" == "$expected_files" ]]
|
||||
[[ "$workflow_files" == '.github/workflows/windows-filesystem.yml' ]]
|
||||
[[ -z "$(git status --porcelain)" ]]
|
||||
- name: Install system dependencies
|
||||
run: |
|
||||
sudo apt-get update
|
||||
sudo apt-get install -y build-essential pkg-config libssl-dev ripgrep
|
||||
- name: Install protoc
|
||||
uses: rustfs/setup-protoc@a3705324d8f9bf5b6c3573fb6cf8ae421db55dd6
|
||||
with:
|
||||
rust-version: stable
|
||||
cache-shared-key: build-x86_64-pc-windows-msvc
|
||||
cache-save-if: 'false'
|
||||
install-build-packaging-tools: 'false'
|
||||
install-test-tools: 'false'
|
||||
|
||||
- name: Build debug server binary
|
||||
shell: pwsh
|
||||
run: cargo build --locked -p rustfs --bin rustfs
|
||||
|
||||
- name: Print version
|
||||
shell: pwsh
|
||||
version: '35.1'
|
||||
repo-token: ${{ github.token }}
|
||||
- name: Install flatc
|
||||
uses: Nugine/setup-flatc@698800de72a96bfb22cf60431dc21a2ff9a7e07b
|
||||
with:
|
||||
version: '25.12.19'
|
||||
- name: Install Rust toolchain
|
||||
uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8
|
||||
with:
|
||||
toolchain: stable
|
||||
- name: Run the named tests
|
||||
shell: bash
|
||||
run: |
|
||||
& .\target\debug\rustfs.exe --version
|
||||
if ($LASTEXITCODE -ne 0) { throw "rustfs --version exited with $LASTEXITCODE" }
|
||||
|
||||
- name: Start server and wait for readiness
|
||||
shell: pwsh
|
||||
env:
|
||||
RUSTFS_ACCESS_KEY: rustfsadmin
|
||||
RUSTFS_SECRET_KEY: rustfsadmin
|
||||
run: |
|
||||
$volume = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke'
|
||||
$stdout = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke.out.log'
|
||||
$stderr = Join-Path $env:RUNNER_TEMP 'rustfs-startup-smoke.err.log'
|
||||
New-Item -ItemType Directory -Force -Path $volume | Out-Null
|
||||
$server = Start-Process -FilePath .\target\debug\rustfs.exe `
|
||||
-ArgumentList @('--address', '127.0.0.1:9000', "`"$volume`"") `
|
||||
-RedirectStandardOutput $stdout -RedirectStandardError $stderr `
|
||||
-NoNewWindow -PassThru
|
||||
# Cache the process handle now; otherwise ExitCode reads back empty.
|
||||
$null = $server.Handle
|
||||
$ready = $false
|
||||
try {
|
||||
for ($i = 0; $i -lt 180; $i++) {
|
||||
if ($server.HasExited) { break }
|
||||
try {
|
||||
$response = Invoke-WebRequest -Uri 'http://127.0.0.1:9000/health/ready' -UseBasicParsing -NoProxy -TimeoutSec 2
|
||||
if ($response.StatusCode -eq 200) { $ready = $true; break }
|
||||
} catch {}
|
||||
Start-Sleep -Seconds 1
|
||||
}
|
||||
} finally {
|
||||
if (-not $server.HasExited) { Stop-Process -Id $server.Id -Force }
|
||||
$server.WaitForExit()
|
||||
Write-Host '--- server stdout (tail) ---'
|
||||
Get-Content $stdout -Tail 100 -ErrorAction SilentlyContinue
|
||||
Write-Host '--- server stderr (tail) ---'
|
||||
Get-Content $stderr -Tail 100 -ErrorAction SilentlyContinue
|
||||
}
|
||||
if (-not $ready) {
|
||||
throw ("rustfs did not report ready on /health/ready (exit code: {0})" -f $server.ExitCode)
|
||||
}
|
||||
set -euo pipefail
|
||||
df -h .
|
||||
output_dir="$RUNNER_TEMP/scanner-cache-read-origin"
|
||||
mkdir -p "$output_dir"
|
||||
tests=(
|
||||
data_usage_define::tests::cache_read_origin_reports_logical_slot_and_legacy_location
|
||||
data_usage_define::tests::save_metrics::read_origin_counts_successful_loads_with_fixed_labels
|
||||
data_usage_define::tests::corrupt_primary_cache_loads_valid_backup
|
||||
data_usage_define::tests::corrupt_cache_without_valid_backup_rebuilds_with_cas_revisions
|
||||
data_usage_define::tests::backup_cache_load_fails_closed_without_main_revision_quorum
|
||||
data_usage_define::tests::recovered_main_cache_wins_over_stale_backup
|
||||
data_usage_define::tests::usage_cache_wire_format_is_pinned
|
||||
)
|
||||
printf '%s\n' "${tests[@]}" > "$output_dir/selection.txt"
|
||||
: > "$output_dir/results.txt"
|
||||
overall=0
|
||||
for test_name in "${tests[@]}"; do
|
||||
log="$output_dir/${test_name//::/_}.log"
|
||||
set +e
|
||||
cargo test --locked -p rustfs-scanner --lib -- --exact "$test_name" > "$log" 2>&1
|
||||
status=$?
|
||||
set -e
|
||||
selected=$(grep -Fxc 'running 1 test' "$log" || true)
|
||||
passed=$(grep -Fxc "test $test_name ... ok" "$log" || true)
|
||||
summary=$(grep -Fc 'test result: ok. 1 passed; 0 failed;' "$log" || true)
|
||||
printf 'test=%s exit=%s selected=%s passed=%s summary=%s\n' "$test_name" "$status" "$selected" "$passed" "$summary" | tee -a "$output_dir/results.txt"
|
||||
tail -n 35 "$log"
|
||||
if [[ "$status" -ne 0 || "$selected" -ne 1 || "$passed" -ne 1 || "$summary" -ne 1 ]]; then
|
||||
overall=1
|
||||
break
|
||||
fi
|
||||
done
|
||||
exit "$overall"
|
||||
- name: Preserve scoped test logs
|
||||
if: always()
|
||||
uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a
|
||||
with:
|
||||
name: scanner-cache-origin-${{ github.run_id }}
|
||||
path: ${{ runner.temp }}/scanner-cache-read-origin/
|
||||
if-no-files-found: warn
|
||||
|
||||
@@ -67,11 +67,12 @@ const METRIC_CACHE_SAVE_RETRY_TOTAL: &str = "rustfs_scanner_cache_save_retry_tot
|
||||
const METRIC_CACHE_SAVE_PAYLOAD_BYTES_TOTAL: &str = "rustfs_scanner_cache_save_payload_bytes_total";
|
||||
const METRIC_CACHE_SAVE_DURATION_SECONDS: &str = "rustfs_scanner_cache_save_duration_seconds";
|
||||
const METRIC_CACHE_BACKUP_REVISION_FAILURE_TOTAL: &str = "rustfs_scanner_cache_backup_revision_failure_total";
|
||||
const METRIC_CACHE_READ_ORIGIN_TOTAL: &str = "rustfs_scanner_cache_read_origin_total";
|
||||
const LOG_COMPONENT_SCANNER: &str = "scanner";
|
||||
const LOG_SUBSYSTEM_CACHE: &str = "cache";
|
||||
const EVENT_SCANNER_CACHE_LOAD_STATE: &str = "scanner_cache_load_state";
|
||||
const EVENT_SCANNER_CACHE_SAVE_STATE: &str = "scanner_cache_save_state";
|
||||
static CACHE_SAVE_METRICS_ONCE: Once = Once::new();
|
||||
static CACHE_METRICS_ONCE: Once = Once::new();
|
||||
|
||||
pub const DATA_USAGE_SCAN_CHECKPOINT_VERSION: u16 = 1;
|
||||
pub const DATA_USAGE_RAW_ENUMERATION_CURSOR_VERSION: u16 = 1;
|
||||
@@ -204,6 +205,54 @@ pub(crate) struct DataUsageCacheRevisions {
|
||||
backup: Option<DataUsageCacheRevision>,
|
||||
}
|
||||
|
||||
/// The observed result of trying the logical primary cache slot.
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub(crate) enum DataUsageCachePrimaryRead {
|
||||
Missing,
|
||||
Corrupt,
|
||||
}
|
||||
|
||||
/// Process-local read provenance, not evidence of freshness or authority.
|
||||
/// Legacy objects may come from the old data-usage bucket. Empty may mean
|
||||
/// missing or corrupt objects; neither state certifies a complete snapshot.
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
pub(crate) enum DataUsageCacheReadOrigin {
|
||||
Primary {
|
||||
legacy: bool,
|
||||
},
|
||||
Backup {
|
||||
primary: DataUsageCachePrimaryRead,
|
||||
legacy: bool,
|
||||
},
|
||||
Empty,
|
||||
}
|
||||
|
||||
impl DataUsageCacheReadOrigin {
|
||||
fn metric_label(self) -> &'static str {
|
||||
match self {
|
||||
Self::Primary { legacy: false } => "primary",
|
||||
Self::Primary { legacy: true } => "legacy_primary",
|
||||
Self::Backup {
|
||||
primary: DataUsageCachePrimaryRead::Missing,
|
||||
legacy: false,
|
||||
} => "backup_primary_missing",
|
||||
Self::Backup {
|
||||
primary: DataUsageCachePrimaryRead::Missing,
|
||||
legacy: true,
|
||||
} => "legacy_backup_primary_missing",
|
||||
Self::Backup {
|
||||
primary: DataUsageCachePrimaryRead::Corrupt,
|
||||
legacy: false,
|
||||
} => "backup_primary_corrupt",
|
||||
Self::Backup {
|
||||
primary: DataUsageCachePrimaryRead::Corrupt,
|
||||
legacy: true,
|
||||
} => "legacy_backup_primary_corrupt",
|
||||
Self::Empty => "empty",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub static DATA_USAGE_BUCKET: LazyLock<String> =
|
||||
LazyLock::new(|| format!("{RUSTFS_META_BUCKET}{SLASH_SEPARATOR}{BUCKET_META_PREFIX}"));
|
||||
|
||||
@@ -1271,8 +1320,12 @@ impl DataUsageCache {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn ensure_cache_save_metrics_registered() {
|
||||
CACHE_SAVE_METRICS_ONCE.call_once(|| {
|
||||
fn ensure_cache_metrics_registered() {
|
||||
CACHE_METRICS_ONCE.call_once(|| {
|
||||
describe_counter!(
|
||||
METRIC_CACHE_READ_ORIGIN_TOTAL,
|
||||
"Successful scanner cache loads by observed logical origin; does not prove cache freshness."
|
||||
);
|
||||
describe_counter!(
|
||||
METRIC_CACHE_SAVE_ATTEMPT_TOTAL,
|
||||
"Total scanner data usage cache save attempts by result and cache type."
|
||||
|
||||
@@ -70,6 +70,7 @@ struct DataUsageCacheLoadResult {
|
||||
cache: DataUsageCache,
|
||||
main_revision: DataUsageCacheRevision,
|
||||
backup_revision: Option<DataUsageCacheRevision>,
|
||||
origin: DataUsageCacheReadOrigin,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
@@ -97,6 +98,7 @@ impl DataUsageCache {
|
||||
/// If the object is not found, a nil error with empty data usage cache is returned.
|
||||
pub async fn load<S: ScannerObjectIO>(&mut self, store: Arc<S>, name: &str) -> StorageResult<()> {
|
||||
let loaded = Self::load_cache(store, name).await?;
|
||||
Self::record_cache_read_origin(loaded.origin);
|
||||
*self = loaded.cache;
|
||||
Ok(())
|
||||
}
|
||||
@@ -106,6 +108,17 @@ impl DataUsageCache {
|
||||
store: Arc<S>,
|
||||
name: &str,
|
||||
) -> StorageResult<DataUsageCacheRevisions> {
|
||||
self.load_with_revisions_and_origin(store, name)
|
||||
.await
|
||||
.map(|(revisions, _)| revisions)
|
||||
}
|
||||
|
||||
/// Origin is an observation of this load, not proof of current coverage.
|
||||
pub(crate) async fn load_with_revisions_and_origin<S: ScannerObjectIO>(
|
||||
&mut self,
|
||||
store: Arc<S>,
|
||||
name: &str,
|
||||
) -> StorageResult<(DataUsageCacheRevisions, DataUsageCacheReadOrigin)> {
|
||||
let backup_name = format!("{name}.bkp");
|
||||
let backup_path = path_join_buf(&[BUCKET_META_PREFIX, &backup_name]);
|
||||
let loaded = Self::load_cache(store.clone(), name).await?;
|
||||
@@ -131,9 +144,16 @@ impl DataUsageCache {
|
||||
},
|
||||
};
|
||||
let main = loaded.main_revision;
|
||||
let origin = loaded.origin;
|
||||
Self::record_cache_read_origin(origin);
|
||||
*self = loaded.cache;
|
||||
|
||||
Ok(DataUsageCacheRevisions { main, backup })
|
||||
Ok((DataUsageCacheRevisions { main, backup }, origin))
|
||||
}
|
||||
|
||||
fn record_cache_read_origin(origin: DataUsageCacheReadOrigin) {
|
||||
Self::ensure_cache_metrics_registered();
|
||||
counter!(METRIC_CACHE_READ_ORIGIN_TOTAL, "origin" => origin.metric_label()).increment(1);
|
||||
}
|
||||
|
||||
pub(crate) async fn read_revisions<S: ScannerObjectIO>(store: Arc<S>, name: &str) -> StorageResult<DataUsageCacheRevisions> {
|
||||
@@ -156,10 +176,12 @@ impl DataUsageCache {
|
||||
cache,
|
||||
revision: Some(main_revision),
|
||||
} => {
|
||||
let legacy = matches!(&main_revision, DataUsageCacheRevision::Missing);
|
||||
return Ok(DataUsageCacheLoadResult {
|
||||
cache: *cache,
|
||||
main_revision,
|
||||
backup_revision: None,
|
||||
origin: DataUsageCacheReadOrigin::Primary { legacy },
|
||||
});
|
||||
}
|
||||
DataUsageCacheLoadAttempt::Loaded { revision: None, .. } => {
|
||||
@@ -193,10 +215,17 @@ impl DataUsageCache {
|
||||
let main_revision = main_revision.ok_or_else(|| {
|
||||
Error::other(format!("scanner cache main revision is unavailable while loading backup: {name}"))
|
||||
})?;
|
||||
let primary = if main_state == DataUsageCacheLoadState::Corrupt {
|
||||
DataUsageCachePrimaryRead::Corrupt
|
||||
} else {
|
||||
DataUsageCachePrimaryRead::Missing
|
||||
};
|
||||
let legacy = matches!(&backup_revision, DataUsageCacheRevision::Missing);
|
||||
return Ok(DataUsageCacheLoadResult {
|
||||
cache: *cache,
|
||||
main_revision,
|
||||
backup_revision: Some(backup_revision),
|
||||
origin: DataUsageCacheReadOrigin::Backup { primary, legacy },
|
||||
});
|
||||
}
|
||||
DataUsageCacheLoadState::Loaded
|
||||
@@ -220,6 +249,7 @@ impl DataUsageCache {
|
||||
main_revision: main_revision
|
||||
.ok_or_else(|| Error::other(format!("scanner cache missing state has no revision: {name}")))?,
|
||||
backup_revision,
|
||||
origin: DataUsageCacheReadOrigin::Empty,
|
||||
});
|
||||
}
|
||||
(DataUsageCacheLoadState::Corrupt, DataUsageCacheLoadState::Missing)
|
||||
@@ -239,6 +269,7 @@ impl DataUsageCache {
|
||||
main_revision: main_revision
|
||||
.ok_or_else(|| Error::other(format!("scanner cache corrupt state has no revision: {name}")))?,
|
||||
backup_revision,
|
||||
origin: DataUsageCacheReadOrigin::Empty,
|
||||
});
|
||||
}
|
||||
_ => {}
|
||||
@@ -518,7 +549,7 @@ impl DataUsageCache {
|
||||
revision: Option<DataUsageCacheRevision>,
|
||||
expected_epoch: Option<u64>,
|
||||
) -> StorageResult<()> {
|
||||
Self::ensure_cache_save_metrics_registered();
|
||||
Self::ensure_cache_metrics_registered();
|
||||
let path_type = Self::cache_path_type(path);
|
||||
let path = path.to_string();
|
||||
let sha256hex = (revision.is_some() && !buf.is_empty())
|
||||
|
||||
@@ -168,8 +168,10 @@ impl AsyncRead for PrefixThenErrorReader {
|
||||
#[derive(Debug)]
|
||||
struct CacheReadStore {
|
||||
main: CacheReadBody,
|
||||
main_missing: bool,
|
||||
backup: Option<Vec<u8>>,
|
||||
legacy: Option<Vec<u8>>,
|
||||
legacy_backup: Option<Vec<u8>>,
|
||||
puts: AtomicUsize,
|
||||
#[cfg(feature = "diagnostic-read-version-caller")]
|
||||
entry_sources: Mutex<Vec<Option<crate::storage_api::ecstore_diagnostic::ReadVersionEntrySource>>>,
|
||||
@@ -185,8 +187,10 @@ impl CacheReadStore {
|
||||
fn new(main: CacheReadBody, backup: Option<Vec<u8>>) -> Self {
|
||||
Self {
|
||||
main,
|
||||
main_missing: false,
|
||||
backup,
|
||||
legacy: None,
|
||||
legacy_backup: None,
|
||||
puts: AtomicUsize::new(0),
|
||||
#[cfg(feature = "diagnostic-read-version-caller")]
|
||||
entry_sources: Mutex::new(Vec::new()),
|
||||
@@ -239,8 +243,12 @@ impl ObjectIO for CacheReadStore {
|
||||
.await
|
||||
.push(crate::storage_api::ecstore_diagnostic::current_read_version_entry_source());
|
||||
if bucket == DATA_USAGE_BUCKET.as_str() {
|
||||
return self
|
||||
.legacy
|
||||
let legacy = if object.ends_with(".bkp") {
|
||||
&self.legacy_backup
|
||||
} else {
|
||||
&self.legacy
|
||||
};
|
||||
return legacy
|
||||
.clone()
|
||||
.map(|data| Self::reader(CacheReadBody::Bytes(data), "legacy-etag"))
|
||||
.ok_or(Error::FileNotFound);
|
||||
@@ -252,6 +260,9 @@ impl ObjectIO for CacheReadStore {
|
||||
let main_path = path_join_buf(&[BUCKET_META_PREFIX, DATA_USAGE_CACHE_NAME]);
|
||||
let backup_path = format!("{main_path}.bkp");
|
||||
if object == main_path {
|
||||
if self.main_missing {
|
||||
return Err(Error::FileNotFound);
|
||||
}
|
||||
return Ok(Self::reader(self.main.clone(), "main-etag"));
|
||||
}
|
||||
if object == backup_path {
|
||||
@@ -712,6 +723,143 @@ async fn corrupt_primary_cache_loads_valid_backup() {
|
||||
assert_eq!(store.puts.load(Ordering::SeqCst), 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cache_read_origin_reports_logical_slot_and_legacy_location() {
|
||||
let mut primary = DataUsageCache::default();
|
||||
primary.info.name = "primary".to_owned();
|
||||
let mut backup = DataUsageCache::default();
|
||||
backup.info.name = "backup".to_owned();
|
||||
backup.info.snapshot_complete = true;
|
||||
let primary_bytes = primary.marshal_msg().expect("serialize primary");
|
||||
let backup_bytes = backup.marshal_msg().expect("serialize backup");
|
||||
let empty_bytes = DataUsageCache::default().marshal_msg().expect("serialize empty cache");
|
||||
let corrupt = b"not-msgpack".to_vec();
|
||||
|
||||
let cases = [
|
||||
(
|
||||
"primary",
|
||||
Some(primary_bytes.clone()),
|
||||
Some(backup_bytes.clone()),
|
||||
None,
|
||||
None,
|
||||
DataUsageCacheReadOrigin::Primary { legacy: false },
|
||||
primary_bytes.clone(),
|
||||
),
|
||||
(
|
||||
"legacy primary",
|
||||
None,
|
||||
None,
|
||||
Some(primary_bytes.clone()),
|
||||
None,
|
||||
DataUsageCacheReadOrigin::Primary { legacy: true },
|
||||
primary_bytes,
|
||||
),
|
||||
(
|
||||
"backup after missing",
|
||||
None,
|
||||
Some(backup_bytes.clone()),
|
||||
None,
|
||||
None,
|
||||
DataUsageCacheReadOrigin::Backup {
|
||||
primary: DataUsageCachePrimaryRead::Missing,
|
||||
legacy: false,
|
||||
},
|
||||
backup_bytes.clone(),
|
||||
),
|
||||
(
|
||||
"legacy backup after missing",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
Some(backup_bytes.clone()),
|
||||
DataUsageCacheReadOrigin::Backup {
|
||||
primary: DataUsageCachePrimaryRead::Missing,
|
||||
legacy: true,
|
||||
},
|
||||
backup_bytes.clone(),
|
||||
),
|
||||
(
|
||||
"backup after corrupt",
|
||||
Some(corrupt.clone()),
|
||||
Some(backup_bytes.clone()),
|
||||
None,
|
||||
None,
|
||||
DataUsageCacheReadOrigin::Backup {
|
||||
primary: DataUsageCachePrimaryRead::Corrupt,
|
||||
legacy: false,
|
||||
},
|
||||
backup_bytes.clone(),
|
||||
),
|
||||
(
|
||||
"legacy backup after corrupt",
|
||||
Some(corrupt.clone()),
|
||||
None,
|
||||
None,
|
||||
Some(backup_bytes.clone()),
|
||||
DataUsageCacheReadOrigin::Backup {
|
||||
primary: DataUsageCachePrimaryRead::Corrupt,
|
||||
legacy: true,
|
||||
},
|
||||
backup_bytes,
|
||||
),
|
||||
(
|
||||
"missing objects",
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
DataUsageCacheReadOrigin::Empty,
|
||||
empty_bytes.clone(),
|
||||
),
|
||||
(
|
||||
"corrupt primary without backup",
|
||||
Some(corrupt),
|
||||
None,
|
||||
None,
|
||||
None,
|
||||
DataUsageCacheReadOrigin::Empty,
|
||||
empty_bytes,
|
||||
),
|
||||
];
|
||||
|
||||
for (name, main, current_backup, legacy_primary, legacy_backup, expected_origin, expected_bytes) in cases {
|
||||
let has_main = main.is_some();
|
||||
let has_backup = current_backup.is_some();
|
||||
let mut fixture = CacheReadStore::new(CacheReadBody::Bytes(main.unwrap_or_default()), current_backup);
|
||||
fixture.main_missing = !has_main;
|
||||
fixture.legacy = legacy_primary;
|
||||
fixture.legacy_backup = legacy_backup;
|
||||
let store = Arc::new(fixture);
|
||||
let mut loaded = DataUsageCache::default();
|
||||
let (revisions, origin) = loaded
|
||||
.load_with_revisions_and_origin(store.clone(), DATA_USAGE_CACHE_NAME)
|
||||
.await
|
||||
.expect("fixture cache load");
|
||||
|
||||
assert_eq!(origin, expected_origin, "origin for {name}");
|
||||
assert_eq!(loaded.marshal_msg().expect("re-encode cache"), expected_bytes, "cache body for {name}");
|
||||
assert_eq!(
|
||||
revisions.main,
|
||||
if has_main {
|
||||
DataUsageCacheRevision::Etag("main-etag".to_owned())
|
||||
} else {
|
||||
DataUsageCacheRevision::Missing
|
||||
},
|
||||
"primary CAS revision for {name}"
|
||||
);
|
||||
assert_eq!(
|
||||
revisions.backup,
|
||||
Some(if has_backup {
|
||||
DataUsageCacheRevision::Etag("backup-etag".to_owned())
|
||||
} else {
|
||||
DataUsageCacheRevision::Missing
|
||||
}),
|
||||
"backup CAS revision for {name}"
|
||||
);
|
||||
assert_eq!(store.puts.load(Ordering::SeqCst), 0, "reading must not write for {name}");
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn corrupt_cache_without_valid_backup_rebuilds_with_cas_revisions() {
|
||||
for backup in [None, Some(b"also-not-msgpack".to_vec())] {
|
||||
|
||||
@@ -7,17 +7,18 @@ use std::sync::atomic::AtomicU64;
|
||||
|
||||
const PAYLOAD_BYTES: &str = "rustfs_scanner_cache_save_payload_bytes_total";
|
||||
const ENCODED_BYTES: &str = "rustfs_scanner_cache_serialized_bytes_total";
|
||||
const READ_ORIGIN: &str = "rustfs_scanner_cache_read_origin_total";
|
||||
|
||||
#[derive(Default)]
|
||||
struct SaveRecorder(StdMutex<HashMap<metrics::Key, Arc<AtomicU64>>>);
|
||||
struct CacheRecorder(StdMutex<HashMap<metrics::Key, Arc<AtomicU64>>>);
|
||||
|
||||
impl metrics::Recorder for SaveRecorder {
|
||||
impl metrics::Recorder for CacheRecorder {
|
||||
fn describe_counter(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
|
||||
fn describe_gauge(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
|
||||
fn describe_histogram(&self, _: metrics::KeyName, _: Option<metrics::Unit>, _: metrics::SharedString) {}
|
||||
|
||||
fn register_counter(&self, key: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Counter {
|
||||
if ![PAYLOAD_BYTES, ENCODED_BYTES].contains(&key.name()) {
|
||||
if ![PAYLOAD_BYTES, ENCODED_BYTES, READ_ORIGIN].contains(&key.name()) {
|
||||
return metrics::Counter::noop();
|
||||
}
|
||||
metrics::Counter::from_arc(self.0.lock().expect("save recorder").entry(key.clone()).or_default().clone())
|
||||
@@ -36,7 +37,7 @@ async fn check_save(store: BackupFallbackStore, revisions: Option<DataUsageCache
|
||||
let store = Arc::new(store);
|
||||
let cache = DataUsageCache::default();
|
||||
let encoded = rmp_serde::to_vec(&cache).expect("independent encoded body");
|
||||
let recorder = SaveRecorder::default();
|
||||
let recorder = CacheRecorder::default();
|
||||
let save = async {
|
||||
match revisions {
|
||||
Some(revisions) => {
|
||||
@@ -85,6 +86,49 @@ async fn check_save(store: BackupFallbackStore, revisions: Option<DataUsageCache
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn read_origin_counts_successful_loads_with_fixed_labels() {
|
||||
let bytes = DataUsageCache::default().marshal_msg().expect("serialize backup");
|
||||
let corrupt_primary = CacheReadStore::new(CacheReadBody::Bytes(b"not-msgpack".to_vec()), Some(bytes.clone()));
|
||||
let mut legacy_backup = CacheReadStore::new(CacheReadBody::Bytes(Vec::new()), None);
|
||||
legacy_backup.main_missing = true;
|
||||
legacy_backup.legacy_backup = Some(bytes);
|
||||
let mut empty = CacheReadStore::new(CacheReadBody::Bytes(Vec::new()), None);
|
||||
empty.main_missing = true;
|
||||
let recorder = CacheRecorder::default();
|
||||
|
||||
for (store, with_revisions) in [(corrupt_primary, false), (legacy_backup, true), (empty, true)] {
|
||||
let store = Arc::new(store);
|
||||
let mut loaded = DataUsageCache::default();
|
||||
let read = async {
|
||||
if with_revisions {
|
||||
loaded.load_with_revisions(store, DATA_USAGE_CACHE_NAME).await.map(|_| ())
|
||||
} else {
|
||||
loaded.load(store, DATA_USAGE_CACHE_NAME).await
|
||||
}
|
||||
};
|
||||
let mut read = std::pin::pin!(read);
|
||||
std::future::poll_fn(|cx| metrics::with_local_recorder(&recorder, || read.as_mut().poll(cx)))
|
||||
.await
|
||||
.expect("fixture load");
|
||||
}
|
||||
|
||||
let unavailable = Arc::new(BackupFallbackStore::new(Vec::new(), false));
|
||||
let mut loaded = DataUsageCache::default();
|
||||
let mut failed_read = std::pin::pin!(loaded.load_with_revisions(unavailable, DATA_USAGE_CACHE_NAME));
|
||||
let error = std::future::poll_fn(|cx| metrics::with_local_recorder(&recorder, || failed_read.as_mut().poll(cx)))
|
||||
.await
|
||||
.expect_err("an unreadable primary must not count as a fallback");
|
||||
assert!(matches!(error, Error::ErasureReadQuorum));
|
||||
|
||||
let rows = recorder.0.lock().expect("cache recorder");
|
||||
assert_eq!(rows.len(), 3, "one fixed origin label per successful load");
|
||||
for origin in ["backup_primary_corrupt", "legacy_backup_primary_missing", "empty"] {
|
||||
let key = metrics::Key::from_parts(READ_ORIGIN, vec![metrics::Label::new("origin", origin)]);
|
||||
assert_eq!(rows[&key].load(Ordering::Relaxed), 1, "one observation for {origin}");
|
||||
}
|
||||
}
|
||||
|
||||
fn revisions(backup: bool) -> DataUsageCacheRevisions {
|
||||
DataUsageCacheRevisions {
|
||||
main: DataUsageCacheRevision::Missing,
|
||||
|
||||
Reference in New Issue
Block a user