Compare commits

..

4 Commits

Author SHA1 Message Date
overtrue 637d4f5e45 test(ci): isolate scheduled alert self-test 2026-08-22 19:33:03 +08:00
overtrue a39ce768cf ci: wire MinIO interop failure alerts 2026-08-22 17:13:50 +08:00
overtrue d3515f5109 test(ci): guard scheduled alert issue wiring 2026-08-22 17:13:50 +08:00
overtrue 8c0eaf225d ci: detect incomplete scheduled validation runs 2026-08-22 17:13:50 +08:00
11 changed files with 360 additions and 134 deletions
@@ -14,9 +14,10 @@
name: "Schedule Failure Issue" name: "Schedule Failure Issue"
description: >- description: >-
Open (or update) a tracking issue when a scheduled workflow run fails. Open (or update) a tracking issue when a scheduled workflow run fails or
does not complete normally.
Dedupes by workflow name: if an open issue titled Dedupes by workflow name: if an open issue titled
"[scheduled-failure] <workflow name>" already exists, the failure is "[scheduled-failure] <workflow name>" already exists, the result is
appended as a comment; otherwise a new issue is created. This is the appended as a comment; otherwise a new issue is created. This is the
single alerting mechanism for all scheduled pipelines (backlog#1149 ci-8). single alerting mechanism for all scheduled pipelines (backlog#1149 ci-8).
@@ -38,6 +39,26 @@ inputs:
Set to an empty string to skip labeling. Set to an empty string to skip labeling.
required: false required: false
default: "infrastructure" default: "infrastructure"
source-run-id:
description: "Run ID to report. Defaults to the current workflow run."
required: false
default: ${{ github.run_id }}
source-run-attempt:
description: "Run attempt to report. Defaults to the current attempt."
required: false
default: ${{ github.run_attempt }}
source-event:
description: "Trigger event of the run being reported."
required: false
default: ${{ github.event_name }}
source-ref-name:
description: "Ref name of the run being reported."
required: false
default: ${{ github.ref_name }}
source-sha:
description: "Commit SHA of the run being reported."
required: false
default: ${{ github.sha }}
runs: runs:
using: "composite" using: "composite"
@@ -48,17 +69,21 @@ runs:
GH_TOKEN: ${{ inputs.github-token }} GH_TOKEN: ${{ inputs.github-token }}
WORKFLOW_NAME: ${{ inputs.workflow-name }} WORKFLOW_NAME: ${{ inputs.workflow-name }}
ISSUE_LABEL: ${{ inputs.label }} ISSUE_LABEL: ${{ inputs.label }}
SOURCE_RUN_ID: ${{ inputs.source-run-id }}
SOURCE_RUN_ATTEMPT: ${{ inputs.source-run-attempt }}
SOURCE_EVENT: ${{ inputs.source-event }}
SOURCE_REF_NAME: ${{ inputs.source-ref-name }}
SOURCE_SHA: ${{ inputs.source-sha }}
run: | run: |
set -euo pipefail set -euo pipefail
title="[scheduled-failure] ${WORKFLOW_NAME}" title="[scheduled-failure] ${WORKFLOW_NAME}"
run_url="${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}" run_url="${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${SOURCE_RUN_ID}"
# Failed job names for this run attempt. The alert job runs while the # Inspect the reported run attempt. It can be the current in-workflow
# run as a whole is still in progress, so inspect the jobs that have # failure or a completed run observed by the external watchdog.
# already completed with a non-success conclusion.
failed_jobs="$(gh api \ failed_jobs="$(gh api \
"repos/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}/attempts/${GITHUB_RUN_ATTEMPT}/jobs" \ "repos/${GITHUB_REPOSITORY}/actions/runs/${SOURCE_RUN_ID}/attempts/${SOURCE_RUN_ATTEMPT}/jobs" \
--paginate \ --paginate \
--jq '.jobs[] --jq '.jobs[]
| select(.conclusion == "failure" or .conclusion == "timed_out" or .conclusion == "cancelled") | select(.conclusion == "failure" or .conclusion == "timed_out" or .conclusion == "cancelled")
@@ -68,13 +93,13 @@ runs:
fi fi
body="$(cat <<EOF body="$(cat <<EOF
Scheduled run of **${WORKFLOW_NAME}** failed. Scheduled run of **${WORKFLOW_NAME}** did not complete successfully.
- Run: ${run_url} (attempt ${GITHUB_RUN_ATTEMPT}) - Run: ${run_url} (attempt ${SOURCE_RUN_ATTEMPT})
- Event: \`${GITHUB_EVENT_NAME}\` - Event: \`${SOURCE_EVENT}\`
- Ref: \`${GITHUB_REF_NAME}\` @ \`${GITHUB_SHA}\` - Ref: \`${SOURCE_REF_NAME}\` @ \`${SOURCE_SHA}\`
Failed jobs: Non-success jobs:
${failed_jobs} ${failed_jobs}
EOF EOF
)" )"
+20
View File
@@ -1032,3 +1032,23 @@ jobs:
echo "🎉 Released $TAG successfully!" echo "🎉 Released $TAG successfully!"
echo "📄 Release URL: ${{ needs.create-release.outputs.release_url }}" echo "📄 Release URL: ${{ needs.create-release.outputs.release_url }}"
alert-on-failure:
name: Alert on scheduled failure
needs: [build-check, prepare-platform-matrix, build-rustfs, build-summary]
if: >-
always() && github.event_name == 'schedule' &&
(contains(needs.*.result, 'failure') || contains(needs.*.result, 'cancelled'))
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
issues: write
steps:
- uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with:
persist-credentials: false
- name: Open or update failure-tracking issue
uses: ./.github/actions/schedule-failure-issue
with:
github-token: ${{ secrets.GITHUB_TOKEN }}
+34
View File
@@ -1032,3 +1032,37 @@ jobs:
path: artifacts/s3tests-single/** path: artifacts/s3tests-single/**
if-no-files-found: ignore if-no-files-found: ignore
retention-days: 3 retention-days: 3
alert-on-failure:
name: Alert on scheduled failure
needs:
- typos
- quick-checks
- test-and-lint
- test-ilm-integration-serial
- test-and-lint-rio-v2
- test-and-lint-protocols
- build-rustfs-debug-binary
- build-rustfs-debug-binary-rio-v2
- uring-integration
- e2e-tests
- e2e-full
- e2e-tests-rio-v2
- s3-implemented-tests
- s3-lifecycle-behavior-tests
if: >-
always() && github.event_name == 'schedule' &&
(contains(needs.*.result, 'failure') || contains(needs.*.result, 'cancelled'))
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
issues: write
steps:
- uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with:
persist-credentials: false
- name: Open or update failure-tracking issue
uses: ./.github/actions/schedule-failure-issue
with:
github-token: ${{ secrets.GITHUB_TOKEN }}
-11
View File
@@ -90,10 +90,6 @@ on:
description: "Optional pytest -m expression" description: "Optional pytest -m expression"
required: false required: false
default: "" default: ""
testexpr:
description: "Optional pytest -k expression"
required: false
default: ""
schedule: schedule:
# Weekly full sweep (Sunday 02:00 UTC): full suite, run against BOTH the # Weekly full sweep (Sunday 02:00 UTC): full suite, run against BOTH the
# single-node and the 4-node distributed topologies (matrix below). # single-node and the 4-node distributed topologies (matrix below).
@@ -120,7 +116,6 @@ env:
XDIST: ${{ github.event.inputs.xdist || '4' }} XDIST: ${{ github.event.inputs.xdist || '4' }}
MAXFAIL: ${{ github.event.inputs.maxfail || '0' }} MAXFAIL: ${{ github.event.inputs.maxfail || '0' }}
MARKEXPR: ${{ github.event.inputs.markexpr || '' }} MARKEXPR: ${{ github.event.inputs.markexpr || '' }}
TESTEXPR: ${{ github.event.inputs.testexpr || '' }}
S3_SHARD_COUNT: ${{ github.event_name == 'schedule' && '4' || github.event.inputs.shard-count || '1' }} S3_SHARD_COUNT: ${{ github.event_name == 'schedule' && '4' || github.event.inputs.shard-count || '1' }}
TEST_TIMEOUT: "300" TEST_TIMEOUT: "300"
@@ -274,13 +269,8 @@ jobs:
EOF EOF
cat > haproxy.cfg <<'EOF' cat > haproxy.cfg <<'EOF'
global
log stdout format raw local0 info
defaults defaults
mode http mode http
log global
log-format '%ci:%cp [%tr] %ft %b/%s %TR/%Tw/%Tc/%Tr/%Ta %ST %B %tsc %HM %HP'
timeout connect 5s timeout connect 5s
timeout client 30s timeout client 30s
timeout server 30s timeout server 30s
@@ -324,7 +314,6 @@ jobs:
XDIST="${XDIST}" \ XDIST="${XDIST}" \
MAXFAIL="${MAXFAIL}" \ MAXFAIL="${MAXFAIL}" \
MARKEXPR="${MARKEXPR}" \ MARKEXPR="${MARKEXPR}" \
TESTEXPR="${TESTEXPR}" \
./scripts/s3-tests/run.sh ./scripts/s3-tests/run.sh
- name: Publish compatibility report - name: Publish compatibility report
+18
View File
@@ -121,3 +121,21 @@ jobs:
cargo nextest run --run-ignored ignored-only --no-tests=fail \ cargo nextest run --run-ignored ignored-only --no-tests=fail \
-p "$INTEROP_PACKAGE" --features "$INTEROP_FEATURES" \ -p "$INTEROP_PACKAGE" --features "$INTEROP_FEATURES" \
-E "$INTEROP_FILTER" -E "$INTEROP_FILTER"
alert-on-failure:
name: Alert on scheduled failure
needs: [minio-interop]
if: always() && github.event_name == 'schedule' && contains(needs.*.result, 'failure')
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
issues: write
steps:
- uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with:
persist-credentials: false
- name: Open or update failure-tracking issue
uses: ./.github/actions/schedule-failure-issue
with:
github-token: ${{ secrets.GITHUB_TOKEN }}
+20
View File
@@ -194,3 +194,23 @@ jobs:
- name: Run HA leader failover live checks (three-node Raft cluster in Docker) - name: Run HA leader failover live checks (three-node Raft cluster in Docker)
run: bash scripts/test/vault_ha_kms_live.sh run: bash scripts/test/vault_ha_kms_live.sh
alert-on-failure:
name: Alert on scheduled failure
needs: [build, kms-vault-lane, kms-vault-ha-failover]
if: >-
always() && github.event_name == 'schedule' &&
(contains(needs.*.result, 'failure') || contains(needs.*.result, 'cancelled'))
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
contents: read
issues: write
steps:
- uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with:
persist-credentials: false
- name: Open or update failure-tracking issue
uses: ./.github/actions/schedule-failure-issue
with:
github-token: ${{ secrets.GITHUB_TOKEN }}
@@ -0,0 +1,63 @@
# 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: Scheduled Validation Watchdog
on:
workflow_run:
workflows:
- "Security Audit"
- "Build and Release"
- "Continuous Integration"
- "coverage"
- "e2e-nightly"
- "e2e-s3tests"
- "Fuzz"
- "mint"
- "minio-interop"
- "Nightly GNU Build"
- "Performance A/B"
- "Runner Hygiene"
types: [completed]
permissions:
contents: read
jobs:
alert-on-incomplete-run:
name: Alert on incomplete scheduled run
if: >-
github.event.workflow_run.event == 'schedule' &&
github.event.workflow_run.conclusion != 'success' &&
github.event.workflow_run.conclusion != 'failure'
runs-on: ubuntu-latest
timeout-minutes: 10
permissions:
actions: read
contents: read
issues: write
steps:
- uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
with:
persist-credentials: false
- name: Open or update incomplete-run issue
uses: ./.github/actions/schedule-failure-issue
with:
github-token: ${{ secrets.GITHUB_TOKEN }}
workflow-name: ${{ github.event.workflow_run.name }}
source-run-id: ${{ github.event.workflow_run.id }}
source-run-attempt: ${{ github.event.workflow_run.run_attempt }}
source-event: ${{ github.event.workflow_run.event }}
source-ref-name: ${{ github.event.workflow_run.head_branch }}
source-sha: ${{ github.event.workflow_run.head_sha }}
@@ -15,14 +15,12 @@
use crate::common::RustFSTestClusterEnvironment; use crate::common::RustFSTestClusterEnvironment;
use aws_sdk_s3::Client; use aws_sdk_s3::Client;
use aws_sdk_s3::error::SdkError; use aws_sdk_s3::error::SdkError;
use aws_sdk_s3::types::{CorsConfiguration, CorsRule};
use bytes::Bytes; use bytes::Bytes;
use std::sync::Arc; use std::sync::Arc;
use tokio::sync::Barrier; use tokio::sync::Barrier;
use tracing::{info, warn}; use tracing::{info, warn};
const BUCKET: &str = "conditional-put-race-bucket"; const BUCKET: &str = "conditional-put-race-bucket";
const BUCKET_METADATA_RELOAD_BUCKET: &str = "bucket-metadata-reload-barrier";
async fn cleanup_object(client: &Client, key: &str) { async fn cleanup_object(client: &Client, key: &str) {
if let Err(e) = client.delete_object().bucket(BUCKET).key(key).send().await { if let Err(e) = client.delete_object().bucket(BUCKET).key(key).send().await {
@@ -30,16 +28,6 @@ async fn cleanup_object(client: &Client, key: &str) {
} }
} }
async fn assert_bucket_cors_missing(client: &Client) {
let result = client.get_bucket_cors().bucket(BUCKET_METADATA_RELOAD_BUCKET).send().await;
match result {
Err(SdkError::ServiceError(error)) => {
assert_eq!(error.err().meta().code(), Some("NoSuchCORSConfiguration"));
}
result => panic!("expected the peer to report a missing CORS configuration: {result:?}"),
}
}
async fn conditional_put( async fn conditional_put(
client: &Client, client: &Client,
key: &str, key: &str,
@@ -248,48 +236,3 @@ async fn test_conditional_put_basic_cluster() -> Result<(), Box<dyn std::error::
cleanup_object(&client, test_key).await; cleanup_object(&client, test_key).await;
Ok(()) Ok(())
} }
#[tokio::test]
async fn test_bucket_cors_write_is_visible_on_peer_before_response() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
crate::common::init_logging();
let mut cluster = RustFSTestClusterEnvironment::new(2).await?;
cluster.start().await?;
cluster.create_test_bucket(BUCKET_METADATA_RELOAD_BUCKET).await?;
let writer = cluster.create_s3_client(0)?;
let reader = cluster.create_s3_client(1)?;
assert_bucket_cors_missing(&reader).await;
let rule = CorsRule::builder()
.allowed_methods("GET")
.allowed_origins("https://example.com")
.build()?;
let configuration = CorsConfiguration::builder().cors_rules(rule).build()?;
writer
.put_bucket_cors()
.bucket(BUCKET_METADATA_RELOAD_BUCKET)
.cors_configuration(configuration)
.send()
.await?;
let response = reader.get_bucket_cors().bucket(BUCKET_METADATA_RELOAD_BUCKET).send().await?;
let rules = response.cors_rules();
assert_eq!(
rules.len(),
1,
"peer should observe the committed CORS rule before the write response returns"
);
assert_eq!(rules[0].allowed_methods(), ["GET"]);
assert_eq!(rules[0].allowed_origins(), ["https://example.com"]);
writer
.delete_bucket_cors()
.bucket(BUCKET_METADATA_RELOAD_BUCKET)
.send()
.await?;
assert_bucket_cors_missing(&reader).await;
writer.delete_bucket().bucket(BUCKET_METADATA_RELOAD_BUCKET).send().await?;
Ok(())
}
@@ -85,7 +85,6 @@ const PEER_REST_RECOVERY_MAX_ATTEMPTS: u32 = 60;
const PEER_REST_RECOVERY_MAX_BACKOFF: Duration = Duration::from_secs(30); const PEER_REST_RECOVERY_MAX_BACKOFF: Duration = Duration::from_secs(30);
const SCANNER_ACTIVITY_MAX_MESSAGE_SIZE: usize = 1024; const SCANNER_ACTIVITY_MAX_MESSAGE_SIZE: usize = 1024;
const REPLICATION_STATS_MAX_MESSAGE_SIZE: usize = 8 * 1024 * 1024; const REPLICATION_STATS_MAX_MESSAGE_SIZE: usize = 8 * 1024 * 1024;
const BUCKET_METADATA_RELOAD_TIMEOUT: Duration = Duration::from_secs(5);
/// Error for a peer that reported `success = false` without an `error_info` payload. /// Error for a peer that reported `success = false` without an `error_info` payload.
/// ///
@@ -1329,38 +1328,27 @@ impl PeerRestClient {
} }
pub async fn load_bucket_metadata(&self, bucket: &str, scanner_maintenance_change: bool) -> Result<()> { pub async fn load_bucket_metadata(&self, bucket: &str, scanner_maintenance_change: bool) -> Result<()> {
let result = tokio::time::timeout(BUCKET_METADATA_RELOAD_TIMEOUT, async { self.finalize_result(
let result = self.load_bucket_metadata_once(bucket, scanner_maintenance_change).await; async {
if let Err(err) = &result let mut client = self.get_client().await?;
&& Self::is_network_like_error(err) let mut request = Request::new(LoadBucketMetadataRequest {
{ bucket: bucket.to_string(),
self.prepare_retry().await; scanner_maintenance_change,
return self.load_bucket_metadata_once(bucket, scanner_maintenance_change).await; });
set_tonic_mutation_body_digest(&mut request)?;
let response = client.load_bucket_metadata(request).await?.into_inner();
if !response.success {
if let Some(msg) = response.error_info {
return Err(Error::other(msg));
}
return Err(peer_failure_without_details("load_bucket_metadata", Some(bucket)));
}
Ok(())
} }
result .await,
}) )
.await .await
.unwrap_or_else(|_| Err(Error::other(format!("load_bucket_metadata({bucket}) timed out"))));
self.finalize_result(result).await
}
async fn load_bucket_metadata_once(&self, bucket: &str, scanner_maintenance_change: bool) -> Result<()> {
let mut client = self.get_client().await?;
let mut request = Request::new(LoadBucketMetadataRequest {
bucket: bucket.to_string(),
scanner_maintenance_change,
});
set_tonic_mutation_body_digest(&mut request)?;
request.set_timeout(BUCKET_METADATA_RELOAD_TIMEOUT);
let response = client.load_bucket_metadata(request).await?.into_inner();
if !response.success {
if let Some(msg) = response.error_info {
return Err(Error::other(msg));
}
return Err(peer_failure_without_details("load_bucket_metadata", Some(bucket)));
}
Ok(())
} }
pub async fn delete_bucket_metadata(&self, bucket: &str) -> Result<()> { pub async fn delete_bucket_metadata(&self, bucket: &str) -> Result<()> {
+18 -22
View File
@@ -513,15 +513,13 @@ fn sr_bucket_meta_item(bucket: String, item_type: &str) -> SRBucketMeta {
} }
} }
async fn notify_bucket_metadata_reload( fn notify_bucket_metadata_reload(
bucket: String, bucket: String,
operation: &'static str, operation: &'static str,
request_context: Option<request_context::RequestContext>, request_context: Option<request_context::RequestContext>,
scanner_maintenance_change: bool, scanner_maintenance_change: bool,
) { ) {
record_local_scanner_maintenance_reload(&bucket, scanner_maintenance_change); record_local_scanner_maintenance_reload(&bucket, scanner_maintenance_change);
// Keep reload detached across request cancellation, but wait before a healthy peer can serve the previous config.
let (completed_tx, completed_rx) = tokio::sync::oneshot::channel();
spawn_background_with_context(request_context, async move { spawn_background_with_context(request_context, async move {
if let Some(notification_sys) = current_notification_system() { if let Some(notification_sys) = current_notification_system() {
let result = if scanner_maintenance_change { let result = if scanner_maintenance_change {
@@ -533,9 +531,7 @@ async fn notify_bucket_metadata_reload(
warn!(bucket = %bucket, error = %err, "failed to notify peers after {operation}"); warn!(bucket = %bucket, error = %err, "failed to notify peers after {operation}");
} }
} }
let _ = completed_tx.send(());
}); });
let _ = completed_rx.await;
} }
fn record_local_scanner_maintenance_reload(bucket: &str, scanner_maintenance_change: bool) { fn record_local_scanner_maintenance_reload(bucket: &str, scanner_maintenance_change: bool) {
@@ -1480,7 +1476,7 @@ impl DefaultBucketUsecase {
.await .await
.map_err(ApiError::from)?; .map_err(ApiError::from)?;
notify_bucket_metadata_reload(bucket.clone(), "delete bucket encryption", request_context, false).await; notify_bucket_metadata_reload(bucket.clone(), "delete bucket encryption", request_context, false);
let item = sr_bucket_meta_item(bucket.clone(), "sse-config"); let item = sr_bucket_meta_item(bucket.clone(), "sse-config");
if let Err(err) = site_replication_bucket_meta_hook(item).await { if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1512,7 +1508,7 @@ impl DefaultBucketUsecase {
.await .await
.map_err(ApiError::from)?; .map_err(ApiError::from)?;
notify_bucket_metadata_reload(bucket.clone(), "delete bucket cors", request_context, false).await; notify_bucket_metadata_reload(bucket.clone(), "delete bucket cors", request_context, false);
let item = sr_bucket_meta_item(bucket.clone(), "cors-config"); let item = sr_bucket_meta_item(bucket.clone(), "cors-config");
if let Err(err) = site_replication_bucket_meta_hook(item).await { if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1544,7 +1540,7 @@ impl DefaultBucketUsecase {
.await .await
.map_err(ApiError::from)?; .map_err(ApiError::from)?;
notify_bucket_metadata_reload(bucket.clone(), "delete bucket lifecycle", request_context, true).await; notify_bucket_metadata_reload(bucket.clone(), "delete bucket lifecycle", request_context, true);
let item = sr_bucket_meta_item(bucket.clone(), "lc-config"); let item = sr_bucket_meta_item(bucket.clone(), "lc-config");
if let Err(err) = site_replication_bucket_meta_hook(item).await { if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1576,7 +1572,7 @@ impl DefaultBucketUsecase {
.await .await
.map_err(ApiError::from)?; .map_err(ApiError::from)?;
notify_bucket_metadata_reload(bucket.clone(), "delete bucket policy", request_context, false).await; notify_bucket_metadata_reload(bucket.clone(), "delete bucket policy", request_context, false);
let item = sr_bucket_meta_item(bucket.clone(), "policy"); let item = sr_bucket_meta_item(bucket.clone(), "policy");
if let Err(err) = site_replication_bucket_meta_hook(item).await { if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1634,7 +1630,7 @@ impl DefaultBucketUsecase {
} }
drop(targets_guard); drop(targets_guard);
notify_bucket_metadata_reload(bucket.clone(), "delete bucket replication", request_context, true).await; notify_bucket_metadata_reload(bucket.clone(), "delete bucket replication", request_context, true);
let item = sr_bucket_meta_item(bucket.clone(), "replication-config"); let item = sr_bucket_meta_item(bucket.clone(), "replication-config");
if let Err(err) = site_replication_bucket_meta_hook(item).await { if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1659,7 +1655,7 @@ impl DefaultBucketUsecase {
.await .await
.map_err(ApiError::from)?; .map_err(ApiError::from)?;
notify_bucket_metadata_reload(bucket.clone(), "delete bucket tagging", request_context, false).await; notify_bucket_metadata_reload(bucket.clone(), "delete bucket tagging", request_context, false);
let item = sr_bucket_meta_item(bucket.clone(), "tags"); let item = sr_bucket_meta_item(bucket.clone(), "tags");
if let Err(err) = site_replication_bucket_meta_hook(item).await { if let Err(err) = site_replication_bucket_meta_hook(item).await {
@@ -1692,7 +1688,7 @@ impl DefaultBucketUsecase {
.await .await
.map_err(ApiError::from)?; .map_err(ApiError::from)?;
notify_bucket_metadata_reload(bucket.clone(), "delete public access block", request_context, false).await; notify_bucket_metadata_reload(bucket.clone(), "delete public access block", request_context, false);
Ok(S3Response::with_status(DeletePublicAccessBlockOutput::default(), StatusCode::NO_CONTENT)) Ok(S3Response::with_status(DeletePublicAccessBlockOutput::default(), StatusCode::NO_CONTENT))
} }
@@ -2147,7 +2143,7 @@ impl DefaultBucketUsecase {
.await .await
.map_err(ApiError::from)?; .map_err(ApiError::from)?;
notify_bucket_metadata_reload(bucket.clone(), "put bucket encryption", request_context, false).await; notify_bucket_metadata_reload(bucket.clone(), "put bucket encryption", request_context, false);
let mut item = sr_bucket_meta_item(bucket.clone(), "sse-config"); let mut item = sr_bucket_meta_item(bucket.clone(), "sse-config");
item.sse_config = Some( item.sse_config = Some(
@@ -2226,7 +2222,7 @@ impl DefaultBucketUsecase {
.await .await
.map_err(ApiError::from)?; .map_err(ApiError::from)?;
notify_bucket_metadata_reload(bucket.clone(), "put bucket lifecycle", request_context, true).await; notify_bucket_metadata_reload(bucket.clone(), "put bucket lifecycle", request_context, true);
let mut item = sr_bucket_meta_item(bucket.clone(), "lc-config"); let mut item = sr_bucket_meta_item(bucket.clone(), "lc-config");
item.expiry_lc_config = item.expiry_lc_config =
@@ -2311,7 +2307,7 @@ impl DefaultBucketUsecase {
.await .await
.map_err(ApiError::from)?; .map_err(ApiError::from)?;
notify_bucket_metadata_reload(bucket.clone(), "put bucket notification", request_context, false).await; notify_bucket_metadata_reload(bucket.clone(), "put bucket notification", request_context, false);
let region = resolve_notification_region(self.global_region(), request_region); let region = resolve_notification_region(self.global_region(), request_region);
let notify = current_notify_interface_for_context(self.context.as_deref()); let notify = current_notify_interface_for_context(self.context.as_deref());
@@ -2416,7 +2412,7 @@ impl DefaultBucketUsecase {
.await .await
.map_err(ApiError::from)?; .map_err(ApiError::from)?;
notify_bucket_metadata_reload(bucket.clone(), "put bucket policy", request_context, false).await; notify_bucket_metadata_reload(bucket.clone(), "put bucket policy", request_context, false);
let mut item = sr_bucket_meta_item(bucket.clone(), "policy"); let mut item = sr_bucket_meta_item(bucket.clone(), "policy");
item.policy = Some(serde_json::from_str(&policy).map_err(|e| s3_error!(InvalidArgument, "parse policy failed {:?}", e))?); item.policy = Some(serde_json::from_str(&policy).map_err(|e| s3_error!(InvalidArgument, "parse policy failed {:?}", e))?);
@@ -2451,7 +2447,7 @@ impl DefaultBucketUsecase {
.await .await
.map_err(ApiError::from)?; .map_err(ApiError::from)?;
notify_bucket_metadata_reload(bucket.clone(), "put bucket cors", request_context, false).await; notify_bucket_metadata_reload(bucket.clone(), "put bucket cors", request_context, false);
let mut item = sr_bucket_meta_item(bucket.clone(), "cors-config"); let mut item = sr_bucket_meta_item(bucket.clone(), "cors-config");
item.cors = item.cors =
@@ -2495,7 +2491,7 @@ impl DefaultBucketUsecase {
.map_err(ApiError::from)?; .map_err(ApiError::from)?;
drop(targets_guard); drop(targets_guard);
notify_bucket_metadata_reload(bucket.clone(), "put bucket replication", request_context, true).await; notify_bucket_metadata_reload(bucket.clone(), "put bucket replication", request_context, true);
let mut item = sr_bucket_meta_item(bucket.clone(), "replication-config"); let mut item = sr_bucket_meta_item(bucket.clone(), "replication-config");
item.replication_config = Some( item.replication_config = Some(
@@ -2535,7 +2531,7 @@ impl DefaultBucketUsecase {
.await .await
.map_err(ApiError::from)?; .map_err(ApiError::from)?;
notify_bucket_metadata_reload(bucket.clone(), "put public access block", request_context, false).await; notify_bucket_metadata_reload(bucket.clone(), "put public access block", request_context, false);
Ok(S3Response::new(PutPublicAccessBlockOutput::default())) Ok(S3Response::new(PutPublicAccessBlockOutput::default()))
} }
@@ -2564,7 +2560,7 @@ impl DefaultBucketUsecase {
.await .await
.map_err(ApiError::from)?; .map_err(ApiError::from)?;
notify_bucket_metadata_reload(bucket.clone(), "put bucket tagging", request_context, false).await; notify_bucket_metadata_reload(bucket.clone(), "put bucket tagging", request_context, false);
let mut item = sr_bucket_meta_item(bucket.clone(), "tags"); let mut item = sr_bucket_meta_item(bucket.clone(), "tags");
item.tags = Some(serialize_config(&tagging).and_then(|bytes| String::from_utf8(bytes).map_err(to_internal_error))?); item.tags = Some(serialize_config(&tagging).and_then(|bytes| String::from_utf8(bytes).map_err(to_internal_error))?);
@@ -2597,7 +2593,7 @@ impl DefaultBucketUsecase {
.await .await
.map_err(ApiError::from)?; .map_err(ApiError::from)?;
notify_bucket_metadata_reload(bucket.clone(), "put bucket versioning", request_context, false).await; notify_bucket_metadata_reload(bucket.clone(), "put bucket versioning", request_context, false);
let mut item = sr_bucket_meta_item(bucket.clone(), "version-config"); let mut item = sr_bucket_meta_item(bucket.clone(), "version-config");
item.versioning = Some( item.versioning = Some(
@@ -3048,7 +3044,7 @@ mod tests {
"{method} should identify the bucket metadata operation in reload logs" "{method} should identify the bucket metadata operation in reload logs"
); );
let expected_reload = format!( let expected_reload = format!(
"notify_bucket_metadata_reload(bucket.clone(), \"{operation}\", request_context, {scanner_maintenance_change}).await;" "notify_bucket_metadata_reload(bucket.clone(), \"{operation}\", request_context, {scanner_maintenance_change});"
); );
assert!( assert!(
body.contains(&expected_reload), body.contains(&expected_reload),
+131 -1
View File
@@ -15,6 +15,20 @@ from pathlib import Path
ROOT = Path(__file__).resolve().parents[1] ROOT = Path(__file__).resolve().parents[1]
SCHEDULED_ALERT_WORKFLOWS = (
".github/workflows/audit.yml",
".github/workflows/build.yml",
".github/workflows/ci.yml",
".github/workflows/coverage.yml",
".github/workflows/e2e-replication-nightly.yml",
".github/workflows/e2e-s3tests.yml",
".github/workflows/fuzz.yml",
".github/workflows/mint.yml",
".github/workflows/minio-interop.yml",
".github/workflows/nightly-gnu.yml",
".github/workflows/performance-ab.yml",
".github/workflows/runner-hygiene.yml",
)
def words(value: str) -> set[str]: def words(value: str) -> set[str]:
@@ -252,6 +266,72 @@ def check_profile_definitions(root: Path) -> list[str]:
return errors return errors
def check_scheduled_alerts(root: Path) -> list[str]:
errors: list[str] = []
for relative in SCHEDULED_ALERT_WORKFLOWS:
path = root / relative
try:
lines = path.read_text().splitlines()
except FileNotFoundError:
errors.append(f"{relative}: missing scheduled validation workflow")
continue
try:
start = lines.index(" alert-on-failure:") + 1
except ValueError:
errors.append(f"{relative}: missing alert-on-failure job")
continue
end = next(
(index for index in range(start, len(lines)) if re.fullmatch(r" [A-Za-z0-9_-]+:", lines[index])),
len(lines),
)
job = "\n".join(line.split("#", 1)[0] for line in lines[start:end])
required = (
"always()",
"github.event_name == 'schedule'",
"contains(needs.*.result, 'failure')",
"issues: write",
"uses: ./.github/actions/schedule-failure-issue",
"github-token: ${{ secrets.GITHUB_TOKEN }}",
)
missing = [token for token in required if token not in job]
if missing:
errors.append(f"{relative}: alert-on-failure missing {', '.join(missing)}")
watchdog_path = root / ".github/workflows/scheduled-validation-watchdog.yml"
try:
watchdog = "\n".join(
line.split("#", 1)[0] for line in watchdog_path.read_text().splitlines()
)
except FileNotFoundError:
errors.append(".github/workflows/scheduled-validation-watchdog.yml: missing completion watchdog")
return errors
for relative in SCHEDULED_ALERT_WORKFLOWS:
path = root / relative
if not path.is_file():
continue
source = path.read_text()
match = re.search(r"^name:\s*[\"']?([^\"'\n]+)", source, re.MULTILINE)
if not match:
errors.append(f"{relative}: missing workflow name")
elif f'- "{match.group(1).strip()}"' not in watchdog:
errors.append(f"{relative}: missing from scheduled completion watchdog")
required = (
"github.event.workflow_run.event == 'schedule'",
"github.event.workflow_run.conclusion != 'success'",
"github.event.workflow_run.conclusion != 'failure'",
"workflow-name: ${{ github.event.workflow_run.name }}",
"source-run-id: ${{ github.event.workflow_run.id }}",
"source-run-attempt: ${{ github.event.workflow_run.run_attempt }}",
)
missing = [token for token in required if token not in watchdog]
if missing:
errors.append(
".github/workflows/scheduled-validation-watchdog.yml: missing " + ", ".join(missing)
)
return errors
def check_profile_listing(root: Path, profile: str, listing: Path) -> list[str]: def check_profile_listing(root: Path, profile: str, listing: Path) -> list[str]:
try: try:
expected_digest = profile_selection(root, profile) expected_digest = profile_selection(root, profile)
@@ -281,6 +361,7 @@ def validate(root: Path) -> list[str]:
errors.extend(check_runner_selection(root)) errors.extend(check_runner_selection(root))
errors.extend(check_s3_tests_runner(root)) errors.extend(check_s3_tests_runner(root))
errors.extend(check_profile_definitions(root)) errors.extend(check_profile_definitions(root))
errors.extend(check_scheduled_alerts(root))
return errors return errors
@@ -363,6 +444,7 @@ class SelfTests(unittest.TestCase):
mock.patch(__name__ + ".check_fuzz_targets", return_value=[]), mock.patch(__name__ + ".check_fuzz_targets", return_value=[]),
mock.patch(__name__ + ".check_runner_selection", return_value=[]), mock.patch(__name__ + ".check_runner_selection", return_value=[]),
mock.patch(__name__ + ".check_profile_definitions", return_value=[]), mock.patch(__name__ + ".check_profile_definitions", return_value=[]),
mock.patch(__name__ + ".check_scheduled_alerts", return_value=[]),
): ):
self.assertEqual(len(validate(root)), 1) self.assertEqual(len(validate(root)), 1)
@@ -413,6 +495,54 @@ class SelfTests(unittest.TestCase):
with mock.patch.object(sys, "platform", "linux"): with mock.patch.object(sys, "platform", "linux"):
self.assertEqual(len(check_profile_listing(root, "e2e-full", listing)), 1) self.assertEqual(len(check_profile_listing(root, "e2e-full", listing)), 1)
def test_scheduled_alerts_require_completion_watchdog(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
alert = (
" alert-on-failure:\n"
" if: always() && github.event_name == 'schedule' && "
"contains(needs.*.result, 'failure')\n"
" permissions:\n"
" issues: write\n"
" steps:\n"
" - uses: ./.github/actions/schedule-failure-issue\n"
" with:\n"
" github-token: ${{ secrets.GITHUB_TOKEN }}\n"
)
names: list[str] = []
for relative in SCHEDULED_ALERT_WORKFLOWS:
path = root / relative
path.parent.mkdir(parents=True, exist_ok=True)
names.append(path.stem)
path.write_text(f'name: "{path.stem}"\n{alert}')
watchdog = root / ".github/workflows/scheduled-validation-watchdog.yml"
watchdog.write_text(
"\n".join(f'- "{name}"' for name in names)
+ "\ngithub.event.workflow_run.event == 'schedule'\n"
+ "github.event.workflow_run.conclusion != 'success'\n"
+ "github.event.workflow_run.conclusion != 'failure'\n"
+ "workflow-name: ${{ github.event.workflow_run.name }}\n"
+ "source-run-id: ${{ github.event.workflow_run.id }}\n"
+ "source-run-attempt: ${{ github.event.workflow_run.run_attempt }}\n"
)
self.assertEqual(check_scheduled_alerts(root), [])
first = root / SCHEDULED_ALERT_WORKFLOWS[0]
mutations = (
("contains(needs.*.result, 'failure')", "false"),
("issues: write", "issues: read"),
("uses: ./.github/actions/schedule-failure-issue", "uses: actions/checkout@v7"),
("github-token: ${{ secrets.GITHUB_TOKEN }}", "github-token: missing"),
)
for required, replacement in mutations:
original = first.read_text()
first.write_text(original.replace(required, replacement))
self.assertEqual(len(check_scheduled_alerts(root)), 1)
first.write_text(original)
watchdog.write_text(watchdog.read_text().replace(f'- "{names[0]}"\n', ""))
self.assertEqual(len(check_scheduled_alerts(root)), 1)
def main() -> int: def main() -> int:
if sys.argv[1:] == ["--self-test"]: if sys.argv[1:] == ["--self-test"]:
suite = unittest.defaultTestLoader.loadTestsFromTestCase(SelfTests) suite = unittest.defaultTestLoader.loadTestsFromTestCase(SelfTests)
@@ -436,7 +566,7 @@ def main() -> int:
for error in errors: for error in errors:
print(f"ERROR: {error}", file=sys.stderr) print(f"ERROR: {error}", file=sys.stderr)
return 1 return 1
print("OK: e2e modules, runner selection, fuzz matrices, profiles, and bounded diagnostics are wired") print("OK: e2e modules, runner selection, fuzz matrices, profiles, and scheduled alerts are wired")
return 0 return 0