mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-07 05:43:14 +00:00
fix(ecstore): avoid offline disks on admin timeout (#3263)
* fix(ecstore): avoid offline disks on admin timeout * fix(ecstore): handle poisoned admin cache locks * fix(ecstore): recover poisoned admin cache on success --------- Co-authored-by: houseme <housemecn@gmail.com> Co-authored-by: 安正超 <anzhengchao@gmail.com>
This commit is contained in:
cxymds
@@ -0,0 +1,208 @@
|
||||
// Copyright 2026 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.
|
||||
|
||||
use crate::common::{RustFSTestClusterEnvironment, init_logging, local_http_client};
|
||||
use aws_sdk_s3::primitives::ByteStream;
|
||||
use http::header::HOST;
|
||||
use reqwest::StatusCode;
|
||||
use rustfs_madmin::{ITEM_OFFLINE, InfoMessage, StorageInfo};
|
||||
use rustfs_signer::constants::UNSIGNED_PAYLOAD;
|
||||
use rustfs_signer::sign_v4;
|
||||
use s3s::Body;
|
||||
use serial_test::serial;
|
||||
use std::error::Error;
|
||||
use std::process::Command;
|
||||
use tokio::time::{Duration, sleep, timeout};
|
||||
use uuid::Uuid;
|
||||
|
||||
const BUCKET: &str = "issue-2525-admin-timeout";
|
||||
|
||||
async fn signed_admin_get(
|
||||
url: &str,
|
||||
access_key: &str,
|
||||
secret_key: &str,
|
||||
) -> Result<reqwest::Response, Box<dyn Error + Send + Sync>> {
|
||||
let uri = url.parse::<http::Uri>()?;
|
||||
let authority = uri.authority().ok_or("request URL missing authority")?.to_string();
|
||||
let request = http::Request::builder()
|
||||
.method(http::Method::GET)
|
||||
.uri(uri)
|
||||
.header(HOST, authority)
|
||||
.header("x-amz-content-sha256", UNSIGNED_PAYLOAD)
|
||||
.body(Body::empty())?;
|
||||
let signed = sign_v4(request, 0, access_key, secret_key, "", "us-east-1");
|
||||
|
||||
let mut builder = local_http_client().get(url);
|
||||
for (name, value) in signed.headers() {
|
||||
builder = builder.header(name, value);
|
||||
}
|
||||
|
||||
Ok(builder.send().await?)
|
||||
}
|
||||
|
||||
async fn fetch_info(cluster: &RustFSTestClusterEnvironment) -> Result<InfoMessage, Box<dyn Error + Send + Sync>> {
|
||||
let url = format!("{}/rustfs/admin/v3/info", cluster.nodes[0].url);
|
||||
let response = signed_admin_get(&url, &cluster.access_key, &cluster.secret_key).await?;
|
||||
parse_json_response(response, "cluster info").await
|
||||
}
|
||||
|
||||
async fn fetch_storage_info(cluster: &RustFSTestClusterEnvironment) -> Result<StorageInfo, Box<dyn Error + Send + Sync>> {
|
||||
let url = format!("{}/rustfs/admin/v3/storageinfo", cluster.nodes[0].url);
|
||||
let response = signed_admin_get(&url, &cluster.access_key, &cluster.secret_key).await?;
|
||||
parse_json_response(response, "storage info").await
|
||||
}
|
||||
|
||||
async fn parse_json_response<T: serde::de::DeserializeOwned>(
|
||||
response: reqwest::Response,
|
||||
label: &str,
|
||||
) -> Result<T, Box<dyn Error + Send + Sync>> {
|
||||
let status = response.status();
|
||||
let body = response.bytes().await?;
|
||||
if status != StatusCode::OK {
|
||||
return Err(format!("{label} request failed: {status} {}", String::from_utf8_lossy(body.as_ref())).into());
|
||||
}
|
||||
|
||||
Ok(serde_json::from_slice(&body)?)
|
||||
}
|
||||
|
||||
fn signal_process(pid: u32, signal: &str) -> Result<(), Box<dyn Error + Send + Sync>> {
|
||||
let output = Command::new("kill").arg(format!("-{signal}")).arg(pid.to_string()).output()?;
|
||||
if output.status.success() {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
Err(format!("kill -{signal} {pid} failed: {}", String::from_utf8_lossy(&output.stderr)).into())
|
||||
}
|
||||
|
||||
fn offline_server_count(info: &InfoMessage) -> usize {
|
||||
info.servers
|
||||
.as_ref()
|
||||
.map(|servers| servers.iter().filter(|server| server.state == ITEM_OFFLINE).count())
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
#[serial]
|
||||
async fn test_single_admin_timeout_does_not_immediately_mark_peer_offline() -> Result<(), Box<dyn Error + Send + Sync>> {
|
||||
init_logging();
|
||||
|
||||
let mut cluster = RustFSTestClusterEnvironment::new(4).await?;
|
||||
cluster.start().await?;
|
||||
cluster.create_test_bucket(BUCKET).await?;
|
||||
|
||||
let warm_info = fetch_info(&cluster).await?;
|
||||
assert_eq!(
|
||||
warm_info
|
||||
.backend
|
||||
.as_ref()
|
||||
.map(|backend| backend.offline_disks)
|
||||
.unwrap_or_default(),
|
||||
0,
|
||||
"cluster should start with zero offline disks"
|
||||
);
|
||||
|
||||
let warm_storage = fetch_storage_info(&cluster).await?;
|
||||
assert_eq!(
|
||||
warm_storage.backend.offline_disks.sum(),
|
||||
0,
|
||||
"storage info should start with zero offline disks"
|
||||
);
|
||||
|
||||
let suspended_pid = cluster.nodes[1]
|
||||
.process
|
||||
.as_ref()
|
||||
.ok_or("cluster node 1 process missing")?
|
||||
.id();
|
||||
signal_process(suspended_pid, "STOP")?;
|
||||
|
||||
let resume = tokio::spawn(async move {
|
||||
sleep(Duration::from_secs(6)).await;
|
||||
signal_process(suspended_pid, "CONT")
|
||||
});
|
||||
|
||||
let client = cluster.create_s3_client(0)?;
|
||||
let key = format!("issue-2525/{}", Uuid::new_v4().simple());
|
||||
let payload = b"admin timeout regression payload".to_vec();
|
||||
|
||||
let during = timeout(Duration::from_secs(20), async {
|
||||
tokio::join!(fetch_info(&cluster), fetch_storage_info(&cluster), async {
|
||||
client
|
||||
.put_object()
|
||||
.bucket(BUCKET)
|
||||
.key(&key)
|
||||
.body(ByteStream::from(payload.clone()))
|
||||
.send()
|
||||
.await?;
|
||||
Ok::<(), Box<dyn Error + Send + Sync>>(())
|
||||
})
|
||||
})
|
||||
.await;
|
||||
|
||||
resume.await??;
|
||||
|
||||
let (info_during, storage_during, put_result) =
|
||||
during.map_err(|_| "timed out while waiting for admin queries during suspended peer")?;
|
||||
let info_during = info_during?;
|
||||
let storage_during = storage_during?;
|
||||
put_result?;
|
||||
|
||||
assert_eq!(
|
||||
info_during
|
||||
.backend
|
||||
.as_ref()
|
||||
.map(|backend| backend.offline_disks)
|
||||
.unwrap_or_default(),
|
||||
0,
|
||||
"single admin timeout must not synthesize offline disks in /info"
|
||||
);
|
||||
assert_eq!(
|
||||
offline_server_count(&info_during),
|
||||
0,
|
||||
"single admin timeout must not mark any server offline in /info"
|
||||
);
|
||||
assert_eq!(
|
||||
storage_during.backend.offline_disks.sum(),
|
||||
0,
|
||||
"single admin timeout must not synthesize offline disks in /storageinfo"
|
||||
);
|
||||
assert!(
|
||||
storage_during.disks.iter().all(|disk| disk.state != ITEM_OFFLINE),
|
||||
"single admin timeout must not synthesize offline disk entries in /storageinfo"
|
||||
);
|
||||
|
||||
let recovered_info = fetch_info(&cluster).await?;
|
||||
assert_eq!(
|
||||
recovered_info
|
||||
.backend
|
||||
.as_ref()
|
||||
.map(|backend| backend.offline_disks)
|
||||
.unwrap_or_default(),
|
||||
0,
|
||||
"offline disk count should stay at zero after the peer resumes"
|
||||
);
|
||||
|
||||
let body = client
|
||||
.get_object()
|
||||
.bucket(BUCKET)
|
||||
.key(&key)
|
||||
.send()
|
||||
.await?
|
||||
.body
|
||||
.collect()
|
||||
.await?
|
||||
.into_bytes();
|
||||
assert_eq!(body.as_ref(), b"admin timeout regression payload");
|
||||
|
||||
Ok(())
|
||||
}
|
||||
@@ -149,4 +149,7 @@ mod snowball_auto_extract_test;
|
||||
#[cfg(test)]
|
||||
mod namespace_lock_quorum_test;
|
||||
|
||||
#[cfg(test)]
|
||||
mod admin_timeout_regression_test;
|
||||
|
||||
pub mod tls_gen;
|
||||
|
||||
Reference in New Issue
Block a user