mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-22 04:16:38 +00:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 98c4675617 | |||
| eec34331de |
@@ -30,6 +30,7 @@ use reqwest::StatusCode;
|
||||
use rustfs_signer::constants::UNSIGNED_PAYLOAD;
|
||||
use rustfs_signer::sign_v4;
|
||||
use s3s::Body;
|
||||
use serde_json;
|
||||
use std::ffi::OsStr;
|
||||
use std::fs as stdfs;
|
||||
use std::io::ErrorKind;
|
||||
@@ -1583,6 +1584,156 @@ impl Drop for RustFSTestClusterEnvironment {
|
||||
}
|
||||
}
|
||||
|
||||
/// Send a SigV4-signed HTTP request and return the raw `reqwest::Response`.
|
||||
///
|
||||
/// Unlike [`signed_s3_request`], this variant accepts `body: Option<Vec<u8>>`
|
||||
/// (binary-safe) and reorders parameters so that `access_key`/`secret_key`
|
||||
/// appear before the body — matching the convention used by the replication
|
||||
/// extension and object-lambda e2e suites.
|
||||
pub(crate) async fn signed_request(
|
||||
method: http::Method,
|
||||
url: &str,
|
||||
access_key: &str,
|
||||
secret_key: &str,
|
||||
body: Option<Vec<u8>>,
|
||||
content_type: Option<&str>,
|
||||
) -> Result<reqwest::Response, Box<dyn std::error::Error + Send + Sync>> {
|
||||
let uri = url.parse::<http::Uri>()?;
|
||||
let authority = uri.authority().ok_or("request URL missing authority")?.to_string();
|
||||
let mut request = http::Request::builder().method(method.clone()).uri(uri);
|
||||
request = request.header(HOST, authority);
|
||||
request = request.header("x-amz-content-sha256", UNSIGNED_PAYLOAD);
|
||||
if let Some(content_type) = content_type {
|
||||
request = request.header(CONTENT_TYPE, content_type);
|
||||
}
|
||||
|
||||
let content_len = body.as_ref().map(|body| body.len() as i64).unwrap_or_default();
|
||||
let signed = sign_v4(request.body(Body::empty())?, content_len, access_key, secret_key, "", "us-east-1");
|
||||
|
||||
let reqwest_method = reqwest::Method::from_bytes(method.as_str().as_bytes())?;
|
||||
let client = local_http_client();
|
||||
let mut request_builder = client.request(reqwest_method, url);
|
||||
for (name, value) in signed.headers() {
|
||||
request_builder = request_builder.header(name, value);
|
||||
}
|
||||
if let Some(body) = body {
|
||||
request_builder = request_builder.body(body);
|
||||
}
|
||||
|
||||
Ok(request_builder.send().await?)
|
||||
}
|
||||
|
||||
/// Like [`signed_request`], but uses a caller-supplied `reqwest::Client`
|
||||
/// instead of the shared [`local_http_client`].
|
||||
pub(crate) async fn signed_request_with_client(
|
||||
client: &reqwest::Client,
|
||||
method: http::Method,
|
||||
url: &str,
|
||||
access_key: &str,
|
||||
secret_key: &str,
|
||||
body: Option<Vec<u8>>,
|
||||
content_type: Option<&str>,
|
||||
) -> Result<reqwest::Response, Box<dyn std::error::Error + Send + Sync>> {
|
||||
let uri = url.parse::<http::Uri>()?;
|
||||
let authority = uri.authority().ok_or("request URL missing authority")?.to_string();
|
||||
let mut request = http::Request::builder().method(method.clone()).uri(uri);
|
||||
request = request.header(HOST, authority);
|
||||
request = request.header("x-amz-content-sha256", UNSIGNED_PAYLOAD);
|
||||
if let Some(content_type) = content_type {
|
||||
request = request.header(CONTENT_TYPE, content_type);
|
||||
}
|
||||
|
||||
let content_len = body.as_ref().map(|body| body.len() as i64).unwrap_or_default();
|
||||
let signed = sign_v4(request.body(Body::empty())?, content_len, access_key, secret_key, "", "us-east-1");
|
||||
|
||||
let reqwest_method = reqwest::Method::from_bytes(method.as_str().as_bytes())?;
|
||||
let mut request_builder = client.request(reqwest_method, url);
|
||||
for (name, value) in signed.headers() {
|
||||
request_builder = request_builder.header(name, value);
|
||||
}
|
||||
if let Some(body) = body {
|
||||
request_builder = request_builder.body(body);
|
||||
}
|
||||
|
||||
Ok(request_builder.send().await?)
|
||||
}
|
||||
|
||||
/// Like [`signed_request`], but includes a `session_token` in the
|
||||
/// `x-amz-security-token` header and passes it to the SigV4 signer.
|
||||
pub(crate) async fn signed_request_with_session_token(
|
||||
method: http::Method,
|
||||
url: &str,
|
||||
access_key: &str,
|
||||
secret_key: &str,
|
||||
session_token: &str,
|
||||
body: Option<Vec<u8>>,
|
||||
content_type: Option<&str>,
|
||||
) -> Result<reqwest::Response, Box<dyn std::error::Error + Send + Sync>> {
|
||||
let uri = url.parse::<http::Uri>()?;
|
||||
let authority = uri.authority().ok_or("request URL missing authority")?.to_string();
|
||||
let mut request = http::Request::builder().method(method.clone()).uri(uri);
|
||||
request = request.header(HOST, authority);
|
||||
request = request.header("x-amz-content-sha256", UNSIGNED_PAYLOAD);
|
||||
if !session_token.is_empty() {
|
||||
request = request.header("x-amz-security-token", session_token);
|
||||
}
|
||||
if let Some(content_type) = content_type {
|
||||
request = request.header(CONTENT_TYPE, content_type);
|
||||
}
|
||||
|
||||
let content_len = body.as_ref().map(|body| body.len() as i64).unwrap_or_default();
|
||||
let signed = sign_v4(
|
||||
request.body(Body::empty())?,
|
||||
content_len,
|
||||
access_key,
|
||||
secret_key,
|
||||
session_token,
|
||||
"us-east-1",
|
||||
);
|
||||
|
||||
let reqwest_method = reqwest::Method::from_bytes(method.as_str().as_bytes())?;
|
||||
let client = local_http_client();
|
||||
let mut request_builder = client.request(reqwest_method, url);
|
||||
for (name, value) in signed.headers() {
|
||||
request_builder = request_builder.header(name, value);
|
||||
}
|
||||
if let Some(body) = body {
|
||||
request_builder = request_builder.body(body);
|
||||
}
|
||||
|
||||
Ok(request_builder.send().await?)
|
||||
}
|
||||
|
||||
/// Create a new user via the admin API.
|
||||
pub(crate) async fn admin_create_user(
|
||||
env: &RustFSTestEnvironment,
|
||||
username: &str,
|
||||
secret_key: &str,
|
||||
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
|
||||
let url = format!("{}/rustfs/admin/v3/add-user?accessKey={}", env.url, username);
|
||||
let body = serde_json::json!({
|
||||
"secretKey": secret_key,
|
||||
"status": "enabled"
|
||||
});
|
||||
let response = signed_request(
|
||||
http::Method::PUT,
|
||||
&url,
|
||||
&env.access_key,
|
||||
&env.secret_key,
|
||||
Some(body.to_string().into_bytes()),
|
||||
Some("application/json"),
|
||||
)
|
||||
.await?;
|
||||
|
||||
if response.status() != reqwest::StatusCode::OK {
|
||||
let status = response.status();
|
||||
let body = response.text().await.unwrap_or_default();
|
||||
return Err(format!("create user failed: {status} {body}").into());
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
@@ -380,24 +380,10 @@ mod tests {
|
||||
cluster.start_node(1).await?;
|
||||
|
||||
let status_url = format!("{}/rustfs/admin/v3/background-heal/status", cluster.nodes[0].url);
|
||||
let mut recovered = serde_json::Value::Null;
|
||||
for _ in 0..60 {
|
||||
let status_body = signed_admin_post(&status_url, None, &cluster.access_key, &cluster.secret_key).await?;
|
||||
assert!(
|
||||
!status_body.contains("MissingContentLength"),
|
||||
"background heal status should not fail without an explicit Content-Length: {status_body}"
|
||||
);
|
||||
recovered = serde_json::from_str(&status_body)
|
||||
.map_err(|err| format!("background heal status is not JSON ({err}): {status_body}"))?;
|
||||
if recovered["clusterStatusComplete"] == serde_json::Value::Bool(true) {
|
||||
break;
|
||||
}
|
||||
sleep(Duration::from_secs(1)).await;
|
||||
}
|
||||
assert_eq!(
|
||||
recovered["clusterStatusComplete"],
|
||||
serde_json::Value::Bool(true),
|
||||
"cluster heal status should recover before root heal starts: {recovered}"
|
||||
let status_body = signed_admin_post(&status_url, None, &cluster.access_key, &cluster.secret_key).await?;
|
||||
assert!(
|
||||
!status_body.contains("MissingContentLength"),
|
||||
"background heal status should not fail without an explicit Content-Length: {status_body}"
|
||||
);
|
||||
|
||||
let heal_body = r#"{"recursive":true,"dryRun":false,"remove":false,"recreate":true,"scanMode":2,"updateParity":false,"nolock":false}"#;
|
||||
|
||||
@@ -40,7 +40,7 @@ use std::time::Duration;
|
||||
use tokio::fs;
|
||||
use tokio::net::TcpStream;
|
||||
use tokio::time::sleep;
|
||||
use tracing::{debug, error, info};
|
||||
use tracing::{debug, error, info, warn};
|
||||
|
||||
// KMS-specific constants
|
||||
pub const TEST_BUCKET: &str = "kms-test-bucket";
|
||||
@@ -177,6 +177,49 @@ pub async fn get_kms_status(
|
||||
Ok(status)
|
||||
}
|
||||
|
||||
/// Poll the KMS status endpoint until the backend reports ready or the timeout
|
||||
/// expires. Replaces hard-coded `sleep(Duration::from_secs(3))` startup waits
|
||||
/// with an active readiness probe so tests start as soon as KMS is usable
|
||||
/// (typically < 1 s) instead of always waiting the full 3 s.
|
||||
///
|
||||
/// Uses exponential back-off starting at 200 ms (doubling each attempt, capped
|
||||
/// at 1 s) up to a total wall-clock budget of 5 s.
|
||||
pub async fn wait_for_kms_ready(
|
||||
base_url: &str,
|
||||
access_key: &str,
|
||||
secret_key: &str,
|
||||
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
|
||||
let total_deadline = Duration::from_secs(5);
|
||||
let start = tokio::time::Instant::now();
|
||||
let mut backoff = Duration::from_millis(200);
|
||||
let max_backoff = Duration::from_secs(1);
|
||||
let mut first_attempt = true;
|
||||
|
||||
loop {
|
||||
if !first_attempt {
|
||||
if start.elapsed() >= total_deadline {
|
||||
return Err("KMS failed to become ready within 5 seconds".into());
|
||||
}
|
||||
sleep(backoff).await;
|
||||
backoff = (backoff * 2).min(max_backoff);
|
||||
}
|
||||
first_attempt = false;
|
||||
|
||||
match get_kms_status(base_url, access_key, secret_key).await {
|
||||
Ok(status) => {
|
||||
info!("KMS is ready (status: {})", status);
|
||||
return Ok(());
|
||||
}
|
||||
Err(e) => {
|
||||
if start.elapsed() >= total_deadline {
|
||||
return Err(format!("KMS did not become ready within 5 s: last error: {e}").into());
|
||||
}
|
||||
warn!(error = %e, elapsed_ms = start.elapsed().as_millis() as u64, "KMS not ready yet, retrying…");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Create a default KMS key for testing and return the created key ID
|
||||
pub async fn create_default_key(
|
||||
base_url: &str,
|
||||
@@ -861,6 +904,13 @@ impl LocalKMSTestEnvironment {
|
||||
Ok(default_key_id.to_string())
|
||||
}
|
||||
|
||||
/// Poll the KMS status endpoint until the backend reports ready.
|
||||
///
|
||||
/// Prefer this over a fixed `sleep` after calling `start_rustfs_for_local_kms`.
|
||||
pub async fn wait_for_kms_ready(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
|
||||
wait_for_kms_ready(&self.base_env.url, &self.base_env.access_key, &self.base_env.secret_key).await
|
||||
}
|
||||
|
||||
/// Configure Local KMS backend with a predefined default key
|
||||
pub async fn configure_local_kms(&self) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
|
||||
// Use a fixed, predictable default key ID
|
||||
|
||||
@@ -12,12 +12,11 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use crate::common::{RustFSTestClusterEnvironment, RustFSTestEnvironment, init_logging, local_http_client};
|
||||
use crate::common::{RustFSTestClusterEnvironment, RustFSTestEnvironment, init_logging, local_http_client, signed_request};
|
||||
use aws_sdk_s3::primitives::ByteStream;
|
||||
use http::header::{CONTENT_TYPE, HOST};
|
||||
use reqwest::StatusCode;
|
||||
use rustfs_signer::constants::UNSIGNED_PAYLOAD;
|
||||
use rustfs_signer::{pre_sign_v4, sign_v4};
|
||||
use rustfs_signer::pre_sign_v4;
|
||||
use rustfs_utils::egress::ENV_OUTBOUND_ALLOW_ORIGINS;
|
||||
use s3s::Body;
|
||||
use std::collections::HashMap;
|
||||
@@ -227,39 +226,6 @@ async fn presigned_get_request(
|
||||
Ok(local_http_client().get(signed.uri().to_string()).send().await?)
|
||||
}
|
||||
|
||||
async fn signed_request(
|
||||
method: http::Method,
|
||||
url: &str,
|
||||
access_key: &str,
|
||||
secret_key: &str,
|
||||
body: Option<Vec<u8>>,
|
||||
content_type: Option<&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 mut request = http::Request::builder().method(method.clone()).uri(uri);
|
||||
request = request.header(HOST, authority);
|
||||
request = request.header("x-amz-content-sha256", UNSIGNED_PAYLOAD);
|
||||
if let Some(content_type) = content_type {
|
||||
request = request.header(CONTENT_TYPE, content_type);
|
||||
}
|
||||
|
||||
let content_len = body.as_ref().map(|body| body.len() as i64).unwrap_or_default();
|
||||
let signed = sign_v4(request.body(Body::empty())?, content_len, access_key, secret_key, "", "us-east-1");
|
||||
|
||||
let reqwest_method = reqwest::Method::from_bytes(method.as_str().as_bytes())?;
|
||||
let client = local_http_client();
|
||||
let mut request_builder = client.request(reqwest_method, url);
|
||||
for (name, value) in signed.headers() {
|
||||
request_builder = request_builder.header(name, value);
|
||||
}
|
||||
if let Some(body) = body {
|
||||
request_builder = request_builder.body(body);
|
||||
}
|
||||
|
||||
Ok(request_builder.send().await?)
|
||||
}
|
||||
|
||||
async fn configure_webhook_target(
|
||||
env: &RustFSTestEnvironment,
|
||||
target_name: &str,
|
||||
|
||||
@@ -13,8 +13,9 @@
|
||||
// limitations under the License.
|
||||
|
||||
use crate::common::{
|
||||
RustFSTestEnvironment, awscurl_available, awscurl_post_sts_form_urlencoded, init_logging, local_http_client,
|
||||
replication_fast_env, rustfs_binary_path,
|
||||
RustFSTestEnvironment, admin_create_user, awscurl_available, awscurl_post_sts_form_urlencoded, init_logging,
|
||||
local_http_client, replication_fast_env, rustfs_binary_path, signed_request, signed_request_with_client,
|
||||
signed_request_with_session_token,
|
||||
};
|
||||
use crate::fake_s3_target::{
|
||||
FAKE_ACCESS_KEY, FAKE_SECRET_KEY, FakeS3Target, FaultAction as FakeTargetFault, Operation as FakeTargetOperation,
|
||||
@@ -35,7 +36,7 @@ use base64::{Engine, engine::general_purpose::STANDARD as BASE64_STANDARD};
|
||||
use bytes::Bytes;
|
||||
use flate2::read::GzDecoder;
|
||||
use futures::{Stream, StreamExt};
|
||||
use http::header::{CONTENT_ENCODING, CONTENT_TYPE, HOST};
|
||||
use http::header::CONTENT_ENCODING;
|
||||
use http_body_util::{BodyExt, Full};
|
||||
use hyper::body::Incoming;
|
||||
use hyper::server::conn::http1;
|
||||
@@ -56,9 +57,6 @@ use rustfs_madmin::{
|
||||
AddServiceAccountReq, ListServiceAccountsResp, PeerInfo, PeerSite, ReplicateAddStatus, ReplicateEditStatus,
|
||||
ReplicateRemoveStatus, SRRemoveReq, SRResyncOpStatus, SRStatusInfo, SiteReplicationInfo, SyncStatus,
|
||||
};
|
||||
use rustfs_signer::constants::UNSIGNED_PAYLOAD;
|
||||
use rustfs_signer::sign_v4;
|
||||
use s3s::Body;
|
||||
use s3s::header::X_AMZ_REPLICATION_STATUS;
|
||||
use sha2::{Digest, Sha256};
|
||||
use std::collections::BTreeMap;
|
||||
@@ -387,116 +385,6 @@ struct ReplicationResetStatusTarget {
|
||||
object: String,
|
||||
}
|
||||
|
||||
async fn signed_request(
|
||||
method: http::Method,
|
||||
url: &str,
|
||||
access_key: &str,
|
||||
secret_key: &str,
|
||||
body: Option<Vec<u8>>,
|
||||
content_type: Option<&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 mut request = http::Request::builder().method(method.clone()).uri(uri);
|
||||
request = request.header(HOST, authority);
|
||||
request = request.header("x-amz-content-sha256", UNSIGNED_PAYLOAD);
|
||||
if let Some(content_type) = content_type {
|
||||
request = request.header(CONTENT_TYPE, content_type);
|
||||
}
|
||||
|
||||
let content_len = body.as_ref().map(|body| body.len() as i64).unwrap_or_default();
|
||||
let signed = sign_v4(request.body(Body::empty())?, content_len, access_key, secret_key, "", "us-east-1");
|
||||
|
||||
let reqwest_method = reqwest::Method::from_bytes(method.as_str().as_bytes())?;
|
||||
let client = local_http_client();
|
||||
let mut request_builder = client.request(reqwest_method, url);
|
||||
for (name, value) in signed.headers() {
|
||||
request_builder = request_builder.header(name, value);
|
||||
}
|
||||
if let Some(body) = body {
|
||||
request_builder = request_builder.body(body);
|
||||
}
|
||||
|
||||
Ok(request_builder.send().await?)
|
||||
}
|
||||
|
||||
async fn signed_request_with_client(
|
||||
client: &reqwest::Client,
|
||||
method: http::Method,
|
||||
url: &str,
|
||||
access_key: &str,
|
||||
secret_key: &str,
|
||||
body: Option<Vec<u8>>,
|
||||
content_type: Option<&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 mut request = http::Request::builder().method(method.clone()).uri(uri);
|
||||
request = request.header(HOST, authority);
|
||||
request = request.header("x-amz-content-sha256", UNSIGNED_PAYLOAD);
|
||||
if let Some(content_type) = content_type {
|
||||
request = request.header(CONTENT_TYPE, content_type);
|
||||
}
|
||||
|
||||
let content_len = body.as_ref().map(|body| body.len() as i64).unwrap_or_default();
|
||||
let signed = sign_v4(request.body(Body::empty())?, content_len, access_key, secret_key, "", "us-east-1");
|
||||
|
||||
let reqwest_method = reqwest::Method::from_bytes(method.as_str().as_bytes())?;
|
||||
let mut request_builder = client.request(reqwest_method, url);
|
||||
for (name, value) in signed.headers() {
|
||||
request_builder = request_builder.header(name, value);
|
||||
}
|
||||
if let Some(body) = body {
|
||||
request_builder = request_builder.body(body);
|
||||
}
|
||||
|
||||
Ok(request_builder.send().await?)
|
||||
}
|
||||
|
||||
async fn signed_request_with_session_token(
|
||||
method: http::Method,
|
||||
url: &str,
|
||||
access_key: &str,
|
||||
secret_key: &str,
|
||||
session_token: &str,
|
||||
body: Option<Vec<u8>>,
|
||||
content_type: Option<&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 mut request = http::Request::builder().method(method.clone()).uri(uri);
|
||||
request = request.header(HOST, authority);
|
||||
request = request.header("x-amz-content-sha256", UNSIGNED_PAYLOAD);
|
||||
if !session_token.is_empty() {
|
||||
request = request.header("x-amz-security-token", session_token);
|
||||
}
|
||||
if let Some(content_type) = content_type {
|
||||
request = request.header(CONTENT_TYPE, content_type);
|
||||
}
|
||||
|
||||
let content_len = body.as_ref().map(|body| body.len() as i64).unwrap_or_default();
|
||||
let signed = sign_v4(
|
||||
request.body(Body::empty())?,
|
||||
content_len,
|
||||
access_key,
|
||||
secret_key,
|
||||
session_token,
|
||||
"us-east-1",
|
||||
);
|
||||
|
||||
let reqwest_method = reqwest::Method::from_bytes(method.as_str().as_bytes())?;
|
||||
let client = local_http_client();
|
||||
let mut request_builder = client.request(reqwest_method, url);
|
||||
for (name, value) in signed.headers() {
|
||||
request_builder = request_builder.header(name, value);
|
||||
}
|
||||
if let Some(body) = body {
|
||||
request_builder = request_builder.body(body);
|
||||
}
|
||||
|
||||
Ok(request_builder.send().await?)
|
||||
}
|
||||
|
||||
fn extract_xml_tag(xml: &str, tag: &str) -> Option<String> {
|
||||
let open = format!("<{tag}>");
|
||||
let close = format!("</{tag}>");
|
||||
@@ -1016,35 +904,6 @@ fn create_user_s3_client(env: &RustFSTestEnvironment, access_key: &str, secret_k
|
||||
Client::from_conf(config)
|
||||
}
|
||||
|
||||
async fn admin_create_user(
|
||||
env: &RustFSTestEnvironment,
|
||||
username: &str,
|
||||
secret_key: &str,
|
||||
) -> Result<(), Box<dyn Error + Send + Sync>> {
|
||||
let url = format!("{}/rustfs/admin/v3/add-user?accessKey={}", env.url, username);
|
||||
let body = serde_json::json!({
|
||||
"secretKey": secret_key,
|
||||
"status": "enabled"
|
||||
});
|
||||
let response = signed_request(
|
||||
http::Method::PUT,
|
||||
&url,
|
||||
&env.access_key,
|
||||
&env.secret_key,
|
||||
Some(body.to_string().into_bytes()),
|
||||
Some("application/json"),
|
||||
)
|
||||
.await?;
|
||||
|
||||
if response.status() != StatusCode::OK {
|
||||
let status = response.status();
|
||||
let body = response.text().await.unwrap_or_default();
|
||||
return Err(format!("create user failed: {status} {body}").into());
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn admin_add_canned_policy(
|
||||
env: &RustFSTestEnvironment,
|
||||
policy_name: &str,
|
||||
|
||||
+32
-138
@@ -12,18 +12,16 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use std::{collections::HashMap, time::Duration};
|
||||
|
||||
use jiff::Timestamp;
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use crate::heal_commands::HealResultItem;
|
||||
|
||||
/// Bitflag helper for service trace categories.
|
||||
///
|
||||
/// Each variant occupies a single bit so that a `TraceType` value can represent
|
||||
/// an arbitrary combination of categories via bitwise OR.
|
||||
#[derive(Debug, Clone, Copy, Serialize, Deserialize, Default)]
|
||||
pub struct TraceType(u64);
|
||||
|
||||
impl TraceType {
|
||||
// Define some constants
|
||||
pub const OS: TraceType = TraceType(1 << 0);
|
||||
pub const STORAGE: TraceType = TraceType(1 << 1);
|
||||
pub const S3: TraceType = TraceType(1 << 2);
|
||||
@@ -40,15 +38,13 @@ impl TraceType {
|
||||
pub const FTP: TraceType = TraceType(1 << 13);
|
||||
pub const ILM: TraceType = TraceType(1 << 14);
|
||||
|
||||
// MetricsAll must be last.
|
||||
/// All trace categories combined. Must be updated when adding new variants.
|
||||
pub const ALL: TraceType = TraceType((1 << 15) - 1);
|
||||
|
||||
pub fn new(t: u64) -> Self {
|
||||
Self(t)
|
||||
}
|
||||
}
|
||||
|
||||
impl TraceType {
|
||||
pub fn contains(&self, x: &TraceType) -> bool {
|
||||
(self.0 & x.0) == x.0
|
||||
}
|
||||
@@ -76,140 +72,38 @@ impl TraceType {
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
|
||||
pub struct TraceInfo {
|
||||
#[serde(rename = "type")]
|
||||
trace_type: u64,
|
||||
#[serde(rename = "nodename")]
|
||||
node_name: String,
|
||||
#[serde(rename = "funcname")]
|
||||
func_name: String,
|
||||
#[serde(rename = "time")]
|
||||
time: Timestamp,
|
||||
#[serde(rename = "path")]
|
||||
path: String,
|
||||
#[serde(rename = "dur")]
|
||||
duration: Duration,
|
||||
#[serde(rename = "bytes", skip_serializing_if = "Option::is_none")]
|
||||
bytes: Option<i64>,
|
||||
#[serde(rename = "msg", skip_serializing_if = "Option::is_none")]
|
||||
message: Option<String>,
|
||||
#[serde(rename = "error", skip_serializing_if = "Option::is_none")]
|
||||
error: Option<String>,
|
||||
#[serde(rename = "custom", skip_serializing_if = "Option::is_none")]
|
||||
custom: Option<HashMap<String, String>>,
|
||||
#[serde(rename = "http", skip_serializing_if = "Option::is_none")]
|
||||
http: Option<TraceHTTPStats>,
|
||||
#[serde(rename = "healResult", skip_serializing_if = "Option::is_none")]
|
||||
heal_result: Option<HealResultItem>,
|
||||
}
|
||||
|
||||
impl TraceInfo {
|
||||
pub fn mask(&self) -> u64 {
|
||||
TraceType::new(self.trace_type).mask()
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
|
||||
pub struct TraceInfoLegacy {
|
||||
trace_info: TraceInfo,
|
||||
#[serde(rename = "request")]
|
||||
req_info: Option<TraceRequestInfo>,
|
||||
#[serde(rename = "response")]
|
||||
resp_info: Option<TraceResponseInfo>,
|
||||
#[serde(rename = "stats")]
|
||||
call_stats: Option<TraceCallStats>,
|
||||
#[serde(rename = "storageStats")]
|
||||
storage_stats: Option<StorageStats>,
|
||||
#[serde(rename = "osStats")]
|
||||
os_stats: Option<OSStats>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
|
||||
pub struct StorageStats {
|
||||
path: String,
|
||||
duration: Duration,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
|
||||
pub struct OSStats {
|
||||
path: String,
|
||||
duration: Duration,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
|
||||
pub struct TraceHTTPStats {
|
||||
req_info: TraceRequestInfo,
|
||||
resp_info: TraceResponseInfo,
|
||||
call_stats: TraceCallStats,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
|
||||
pub struct TraceCallStats {
|
||||
input_bytes: i32,
|
||||
output_bytes: i32,
|
||||
latency: Duration,
|
||||
time_to_first_byte: Duration,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
|
||||
pub struct TraceRequestInfo {
|
||||
time: Timestamp,
|
||||
proto: String,
|
||||
method: String,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
path: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
raw_query: Option<String>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
headers: Option<HashMap<String, String>>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
body: Option<Vec<u8>>,
|
||||
client: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
|
||||
pub struct TraceResponseInfo {
|
||||
time: Timestamp,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
headers: Option<HashMap<String, String>>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
body: Option<Vec<u8>>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
status_code: Option<i32>,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn trace_timestamps_serialize_as_rfc3339_utc() {
|
||||
let timestamp = Timestamp::constant(1_700_000_000, 123_456_000);
|
||||
let trace = TraceInfo {
|
||||
time: timestamp,
|
||||
http: Some(TraceHTTPStats {
|
||||
req_info: TraceRequestInfo {
|
||||
time: timestamp,
|
||||
..Default::default()
|
||||
},
|
||||
resp_info: TraceResponseInfo {
|
||||
time: timestamp,
|
||||
..Default::default()
|
||||
},
|
||||
..Default::default()
|
||||
}),
|
||||
..Default::default()
|
||||
};
|
||||
fn trace_type_contains_and_overlaps() {
|
||||
let mut combined = TraceType::default();
|
||||
combined.merge(&TraceType::S3);
|
||||
combined.merge(&TraceType::HEALING);
|
||||
|
||||
let value = serde_json::to_value(trace).expect("trace should serialize");
|
||||
assert_eq!(value["time"], "2023-11-14T22:13:20.123456Z");
|
||||
assert_eq!(value["http"]["req_info"]["time"], "2023-11-14T22:13:20.123456Z");
|
||||
assert_eq!(value["http"]["resp_info"]["time"], "2023-11-14T22:13:20.123456Z");
|
||||
let trace: TraceInfo = serde_json::from_value(value).expect("trace should deserialize");
|
||||
assert_eq!(trace.time, timestamp);
|
||||
let http = trace.http.expect("http trace should deserialize");
|
||||
assert_eq!(http.req_info.time, timestamp);
|
||||
assert_eq!(http.resp_info.time, timestamp);
|
||||
assert!(combined.contains(&TraceType::S3));
|
||||
assert!(combined.contains(&TraceType::HEALING));
|
||||
assert!(!combined.contains(&TraceType::SCANNER));
|
||||
assert!(combined.overlaps(&TraceType::S3));
|
||||
assert!(combined.overlaps(&TraceType::HEALING));
|
||||
assert!(!combined.overlaps(&TraceType::SCANNER));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn trace_type_set_if() {
|
||||
let mut tt = TraceType::default();
|
||||
tt.set_if(true, &TraceType::OS);
|
||||
tt.set_if(false, &TraceType::S3);
|
||||
assert!(tt.contains(&TraceType::OS));
|
||||
assert!(!tt.contains(&TraceType::S3));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn trace_type_single_type() {
|
||||
assert!(TraceType::S3.single_type());
|
||||
let mut combined = TraceType::S3;
|
||||
combined.merge(&TraceType::HEALING);
|
||||
assert!(!combined.single_type());
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user