You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by ps...@apache.org on 2026/08/14 05:47:49 UTC
(opendal-ofs) 11/12: feat(harness): add reproducible managed sync conditions
This is an automated email from the ASF dual-hosted git repository.
PsiACE pushed a commit to branch managed-sync-eval
in repository https://gitbox.apache.org/repos/asf/opendal-ofs.git
commit 95dd1640a5c6c6d8c7b7429bd5fd88f9d5575c26
Author: PsiACE <ps...@apache.org>
AuthorDate: Fri Aug 14 13:38:43 2026 +0800
feat(harness): add reproducible managed sync conditions
---
fixtures/managed-sync/compose.yaml | 9 +
xtask/src/main.rs | 14 +-
xtask/src/managed_sync.rs | 2 +
xtask/src/managed_sync/behavior.rs | 5 +-
xtask/src/managed_sync/bub.rs | 2 +-
xtask/src/managed_sync/evaluation.rs | 327 +++++++++++++++++++++++++++
xtask/src/managed_sync/fixture.rs | 265 ++++++++++++++++++++--
xtask/src/managed_sync/scale.rs | 5 +-
xtask/src/managed_sync/scale/random_write.rs | 4 +-
9 files changed, 614 insertions(+), 19 deletions(-)
diff --git a/fixtures/managed-sync/compose.yaml b/fixtures/managed-sync/compose.yaml
index f6536e2..cdb3f60 100644
--- a/fixtures/managed-sync/compose.yaml
+++ b/fixtures/managed-sync/compose.yaml
@@ -62,5 +62,14 @@ services:
NO_PROXY: minio,127.0.0.1,localhost
no_proxy: minio,127.0.0.1,localhost
+ toxiproxy:
+ image: ghcr.io/shopify/toxiproxy:2.12.0
+ environment:
+ NO_PROXY: minio,toxiproxy,127.0.0.1,localhost
+ no_proxy: minio,toxiproxy,127.0.0.1,localhost
+ ports:
+ - "127.0.0.1:${OFS_MANAGED_SYNC_PROXY_PORT:-19001}:8666"
+ - "127.0.0.1:${OFS_MANAGED_SYNC_PROXY_ADMIN_PORT:-19002}:8474"
+
volumes:
minio-data:
diff --git a/xtask/src/main.rs b/xtask/src/main.rs
index a184eb2..8774de8 100644
--- a/xtask/src/main.rs
+++ b/xtask/src/main.rs
@@ -127,6 +127,9 @@ struct CommandManagedSyncScale {
#[arg(long, default_value_t = 1024, help = "Random-write file size in MiB.")]
file_size_mib: u64,
+
+ #[command(flatten)]
+ evaluation: managed_sync::EvaluationOptions,
}
impl CommandManagedSyncScale {
@@ -138,6 +141,7 @@ impl CommandManagedSyncScale {
self.duration_seconds,
self.sync_interval_seconds,
self.file_size_mib,
+ self.evaluation,
);
}
}
@@ -246,11 +250,19 @@ struct CommandManagedSyncBehavior {
#[arg(long, help = "Leave the fixture running after the command exits.")]
keep: bool,
+
+ #[command(flatten)]
+ evaluation: managed_sync::EvaluationOptions,
}
impl CommandManagedSyncBehavior {
fn run(self) {
- managed_sync::run_fixture(self.keep, self.case.as_deref(), &self.profile);
+ managed_sync::run_fixture(
+ self.keep,
+ self.case.as_deref(),
+ &self.profile,
+ self.evaluation,
+ );
}
}
diff --git a/xtask/src/managed_sync.rs b/xtask/src/managed_sync.rs
index a150295..654c274 100644
--- a/xtask/src/managed_sync.rs
+++ b/xtask/src/managed_sync.rs
@@ -20,9 +20,11 @@
mod behavior;
mod bub;
mod cli;
+mod evaluation;
mod fixture;
mod scale;
pub(crate) use behavior::run as run_fixture;
pub(crate) use bub::run as run_bub_e2e;
+pub(crate) use evaluation::EvaluationOptions;
pub(crate) use scale::run as run_scale;
diff --git a/xtask/src/managed_sync/behavior.rs b/xtask/src/managed_sync/behavior.rs
index ab7b112..ba1cf68 100644
--- a/xtask/src/managed_sync/behavior.rs
+++ b/xtask/src/managed_sync/behavior.rs
@@ -25,13 +25,14 @@ use std::fs;
use std::path::Path;
use super::cli::{BuildProfile, Ofs};
+use super::evaluation::EvaluationOptions;
use super::fixture::Fixture;
-pub(crate) fn run(keep: bool, case: Option<&str>, profile: &str) {
+pub(crate) fn run(keep: bool, case: Option<&str>, profile: &str, evaluation: EvaluationOptions) {
let profile = BuildProfile::parse(profile);
let ofs = Ofs::debug(profile);
ofs.build();
- let fixture = Fixture::new(keep, profile.name()).start();
+ let fixture = Fixture::new(keep, profile.name(), evaluation).start();
fixture.create_bucket();
match case {
Some("admission") => lifecycle::admission(&fixture, ofs),
diff --git a/xtask/src/managed_sync/bub.rs b/xtask/src/managed_sync/bub.rs
index 6a1af7b..010f1a7 100644
--- a/xtask/src/managed_sync/bub.rs
+++ b/xtask/src/managed_sync/bub.rs
@@ -30,7 +30,7 @@ pub(crate) fn run(keep: bool) {
.ok()
.filter(|value| !value.is_empty())
.expect("BUB_API_KEY is required for the Bub end-to-end scenario");
- let fixture = Fixture::new(keep, "bub").start_bub();
+ let fixture = Fixture::new(keep, "bub", Default::default()).start_bub();
fixture.create_bucket();
let sessions = fixture.container_storage_url("bub/sessions");
let skills = fixture.container_storage_url("bub/skills");
diff --git a/xtask/src/managed_sync/evaluation.rs b/xtask/src/managed_sync/evaluation.rs
new file mode 100644
index 0000000..c36e415
--- /dev/null
+++ b/xtask/src/managed_sync/evaluation.rs
@@ -0,0 +1,327 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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.
+
+//! Explicit network and billing conditions for Managed Sync evaluation.
+
+use clap::Args;
+use std::io::{Read as _, Write as _};
+use std::net::TcpStream;
+use std::sync::mpsc;
+use std::thread;
+use std::time::Duration;
+
+pub(super) const TOXIPROXY_NAME: &str = "managed-sync-minio";
+
+pub(super) struct ChaosController {
+ stop: mpsc::Sender<()>,
+ worker: thread::JoinHandle<ChaosReport>,
+}
+
+#[derive(Clone, Debug)]
+pub(super) struct ChaosReport {
+ pub(super) schedule_seed: String,
+ pub(super) faults_injected: u64,
+}
+
+#[derive(Args, Clone, Debug, Default)]
+pub(crate) struct EvaluationOptions {
+ #[arg(
+ long,
+ default_value_t = 0,
+ help = "Mean object-store round-trip delay in milliseconds."
+ )]
+ network_rtt_ms: u64,
+
+ #[arg(
+ long,
+ default_value_t = 0,
+ help = "Random variation around --network-rtt-ms in milliseconds."
+ )]
+ network_jitter_ms: u64,
+
+ #[arg(
+ long,
+ default_value_t = 0,
+ help = "Per-connection, per-direction bandwidth limit in decimal KB/s; 0 is unlimited."
+ )]
+ connection_bandwidth_kb_per_second: u64,
+
+ #[arg(long, help = "Seed for deterministic connection-reset scheduling.")]
+ chaos_seed: Option<u64>,
+
+ #[arg(
+ long,
+ default_value_t = 0,
+ help = "Connection-reset probability per scheduling tick, in millionths."
+ )]
+ chaos_reset_rate_per_million: u32,
+
+ #[arg(
+ long,
+ default_value_t = 250,
+ help = "Deterministic chaos scheduling tick in milliseconds."
+ )]
+ chaos_tick_ms: u64,
+
+ #[arg(
+ long,
+ default_value_t = 100,
+ help = "Duration of each injected connection outage in milliseconds."
+ )]
+ chaos_outage_ms: u64,
+
+ #[arg(
+ long,
+ default_value_t = 0.0,
+ help = "Price per million read requests in USD."
+ )]
+ read_request_usd_per_million: f64,
+
+ #[arg(
+ long,
+ default_value_t = 0.0,
+ help = "Price per million write or list requests in USD."
+ )]
+ write_list_request_usd_per_million: f64,
+
+ #[arg(
+ long,
+ default_value_t = 0.0,
+ help = "Object-store egress price per GiB in USD."
+ )]
+ egress_usd_per_gib: f64,
+
+ #[arg(
+ long,
+ default_value_t = 0.0,
+ help = "Object-store ingress price per GiB in USD."
+ )]
+ ingress_usd_per_gib: f64,
+
+ #[arg(
+ long,
+ default_value_t = 0.0,
+ help = "Storage price per GiB-month in USD."
+ )]
+ storage_usd_per_gib_month: f64,
+}
+
+impl EvaluationOptions {
+ pub(super) fn validate(&self) {
+ assert!(
+ self.network_jitter_ms <= self.network_rtt_ms,
+ "network jitter must not exceed mean RTT"
+ );
+ assert!(
+ self.connection_bandwidth_kb_per_second <= i64::MAX as u64,
+ "connection bandwidth exceeds the fault proxy range"
+ );
+ assert!(
+ self.chaos_reset_rate_per_million <= 1_000_000,
+ "chaos reset rate must not exceed one million"
+ );
+ assert!(
+ self.chaos_reset_rate_per_million == 0 || self.chaos_seed.is_some(),
+ "--chaos-seed is required when connection resets are enabled"
+ );
+ assert!(
+ self.chaos_reset_rate_per_million == 0 || self.chaos_tick_ms != 0,
+ "chaos tick must be positive when connection resets are enabled"
+ );
+ for (name, value) in [
+ ("read request price", self.read_request_usd_per_million),
+ (
+ "write/list request price",
+ self.write_list_request_usd_per_million,
+ ),
+ ("egress price", self.egress_usd_per_gib),
+ ("ingress price", self.ingress_usd_per_gib),
+ ("storage price", self.storage_usd_per_gib_month),
+ ] {
+ assert!(
+ value.is_finite() && value >= 0.0,
+ "{name} must be finite and non-negative"
+ );
+ }
+ }
+
+ pub(super) const fn network_enabled(&self) -> bool {
+ self.network_rtt_ms != 0
+ || self.network_jitter_ms != 0
+ || self.connection_bandwidth_kb_per_second != 0
+ || self.chaos_reset_rate_per_million != 0
+ }
+
+ pub(super) const fn network_rtt_ms(&self) -> u64 {
+ self.network_rtt_ms
+ }
+
+ pub(super) const fn network_jitter_ms(&self) -> u64 {
+ self.network_jitter_ms
+ }
+
+ pub(super) const fn connection_bandwidth_kb_per_second(&self) -> u64 {
+ self.connection_bandwidth_kb_per_second
+ }
+
+ pub(super) const fn chaos_seed(&self) -> Option<u64> {
+ self.chaos_seed
+ }
+
+ pub(super) const fn chaos_reset_rate_per_million(&self) -> u32 {
+ self.chaos_reset_rate_per_million
+ }
+
+ pub(super) const fn chaos_tick_ms(&self) -> u64 {
+ self.chaos_tick_ms
+ }
+
+ pub(super) const fn chaos_outage_ms(&self) -> u64 {
+ self.chaos_outage_ms
+ }
+
+ pub(super) const fn billing_enabled(&self) -> bool {
+ self.read_request_usd_per_million != 0.0
+ || self.write_list_request_usd_per_million != 0.0
+ || self.egress_usd_per_gib != 0.0
+ || self.ingress_usd_per_gib != 0.0
+ || self.storage_usd_per_gib_month != 0.0
+ }
+
+ pub(super) const fn read_request_usd_per_million(&self) -> f64 {
+ self.read_request_usd_per_million
+ }
+
+ pub(super) const fn write_list_request_usd_per_million(&self) -> f64 {
+ self.write_list_request_usd_per_million
+ }
+
+ pub(super) const fn egress_usd_per_gib(&self) -> f64 {
+ self.egress_usd_per_gib
+ }
+
+ pub(super) const fn ingress_usd_per_gib(&self) -> f64 {
+ self.ingress_usd_per_gib
+ }
+
+ pub(super) const fn storage_usd_per_gib_month(&self) -> f64 {
+ self.storage_usd_per_gib_month
+ }
+
+ pub(super) fn start_chaos(
+ &self,
+ admin_port: u16,
+ root: &str,
+ stage: &str,
+ ) -> Option<ChaosController> {
+ let rate = self.chaos_reset_rate_per_million;
+ if rate == 0 {
+ return None;
+ }
+ let seed = self
+ .chaos_seed
+ .expect("validated chaos configuration has a seed");
+ let mut hasher = blake3::Hasher::new();
+ hasher.update(b"ofs-managed-sync-chaos\0");
+ hasher.update(&seed.to_le_bytes());
+ hasher.update(&(root.len() as u64).to_le_bytes());
+ hasher.update(root.as_bytes());
+ hasher.update(&(stage.len() as u64).to_le_bytes());
+ hasher.update(stage.as_bytes());
+ let schedule_seed = hasher.finalize();
+ let mut schedule = blake3::Hasher::new();
+ schedule.update(b"ofs-managed-sync-chaos-events\0");
+ schedule.update(schedule_seed.as_bytes());
+ let mut schedule = schedule.finalize_xof();
+ let tick = Duration::from_millis(self.chaos_tick_ms);
+ let outage = Duration::from_millis(self.chaos_outage_ms);
+ let reset_threshold = u64::from(rate) * (1_u64 << 32) / 1_000_000;
+ let (stop, receiver) = mpsc::channel();
+ let worker = thread::spawn(move || {
+ let mut faults_injected = 0_u64;
+ loop {
+ match receiver.recv_timeout(tick) {
+ Ok(()) | Err(mpsc::RecvTimeoutError::Disconnected) => break,
+ Err(mpsc::RecvTimeoutError::Timeout) => {}
+ }
+ let mut random = [0_u8; size_of::<u32>()];
+ schedule.fill(&mut random);
+ if u64::from(u32::from_le_bytes(random)) >= reset_threshold {
+ continue;
+ }
+ set_proxy_enabled(admin_port, false)
+ .unwrap_or_else(|error| panic!("inject Managed Sync outage: {error}"));
+ faults_injected += 1;
+ let stopped = match receiver.recv_timeout(outage) {
+ Ok(()) | Err(mpsc::RecvTimeoutError::Disconnected) => true,
+ Err(mpsc::RecvTimeoutError::Timeout) => false,
+ };
+ set_proxy_enabled(admin_port, true)
+ .unwrap_or_else(|error| panic!("end Managed Sync outage: {error}"));
+ if stopped {
+ break;
+ }
+ }
+ ChaosReport {
+ schedule_seed: schedule_seed.to_hex().to_string(),
+ faults_injected,
+ }
+ });
+ Some(ChaosController { stop, worker })
+ }
+}
+
+impl ChaosController {
+ pub(super) fn stop(self) -> ChaosReport {
+ let _ = self.stop.send(());
+ self.worker
+ .join()
+ .expect("Managed Sync chaos controller did not panic")
+ }
+}
+
+fn set_proxy_enabled(port: u16, enabled: bool) -> Result<(), String> {
+ let address = format!("127.0.0.1:{port}");
+ let mut stream = TcpStream::connect_timeout(
+ &address.parse().expect("loopback proxy address is valid"),
+ Duration::from_secs(1),
+ )
+ .map_err(|error| error.to_string())?;
+ stream
+ .set_read_timeout(Some(Duration::from_secs(1)))
+ .map_err(|error| error.to_string())?;
+ stream
+ .set_write_timeout(Some(Duration::from_secs(1)))
+ .map_err(|error| error.to_string())?;
+ let body = format!("{{\"enabled\":{enabled}}}");
+ let request = format!(
+ "POST /proxies/{TOXIPROXY_NAME} HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
+ body.len()
+ );
+ stream
+ .write_all(request.as_bytes())
+ .map_err(|error| error.to_string())?;
+ let mut response = [0; 32];
+ let read = stream
+ .read(&mut response)
+ .map_err(|error| error.to_string())?;
+ if response[..read].starts_with(b"HTTP/1.1 200") {
+ Ok(())
+ } else {
+ Err(String::from_utf8_lossy(&response[..read]).into_owned())
+ }
+}
diff --git a/xtask/src/managed_sync/fixture.rs b/xtask/src/managed_sync/fixture.rs
index 2cf9e00..ff91fdc 100644
--- a/xtask/src/managed_sync/fixture.rs
+++ b/xtask/src/managed_sync/fixture.rs
@@ -22,21 +22,28 @@ use std::env;
use std::fs;
use std::io::{Read as _, Write as _};
use std::net::TcpStream;
+use std::panic::{AssertUnwindSafe, catch_unwind, resume_unwind};
use std::path::{Path, PathBuf};
use std::process::Command;
use std::thread;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use super::cli::{output_text, run_logged};
+use super::evaluation::{ChaosController, ChaosReport, EvaluationOptions, TOXIPROXY_NAME};
const DEFAULT_MINIO_PORT: u16 = 19_000;
+const DEFAULT_PROXY_PORT: u16 = 19_001;
+const DEFAULT_PROXY_ADMIN_PORT: u16 = 19_002;
const FIXTURE_READY_TIMEOUT: Duration = Duration::from_secs(30);
pub(super) struct Fixture {
compose_file: PathBuf,
keep: bool,
+ evaluation: EvaluationOptions,
minio_port: u16,
project: String,
+ proxy_admin_port: u16,
+ proxy_port: u16,
started: bool,
variant: String,
}
@@ -149,7 +156,12 @@ fn is_executable(_metadata: &fs::Metadata) -> bool {
}
impl Fixture {
- pub(crate) fn new(keep: bool, variant: impl Into<String>) -> Self {
+ pub(crate) fn new(
+ keep: bool,
+ variant: impl Into<String>,
+ evaluation: EvaluationOptions,
+ ) -> Self {
+ evaluation.validate();
let minio_port = env::var("OFS_MANAGED_SYNC_MINIO_PORT")
.map(|value| {
value
@@ -157,12 +169,36 @@ impl Fixture {
.unwrap_or_else(|_| panic!("invalid OFS_MANAGED_SYNC_MINIO_PORT: {value}"))
})
.unwrap_or(DEFAULT_MINIO_PORT);
+ let proxy_port = env::var("OFS_MANAGED_SYNC_PROXY_PORT")
+ .map(|value| {
+ value
+ .parse()
+ .unwrap_or_else(|_| panic!("invalid OFS_MANAGED_SYNC_PROXY_PORT: {value}"))
+ })
+ .unwrap_or(DEFAULT_PROXY_PORT);
+ let proxy_admin_port = env::var("OFS_MANAGED_SYNC_PROXY_ADMIN_PORT")
+ .map(|value| {
+ value.parse().unwrap_or_else(|_| {
+ panic!("invalid OFS_MANAGED_SYNC_PROXY_ADMIN_PORT: {value}")
+ })
+ })
+ .unwrap_or(DEFAULT_PROXY_ADMIN_PORT);
+ assert!(
+ !evaluation.network_enabled()
+ || (proxy_port != minio_port
+ && proxy_admin_port != minio_port
+ && proxy_admin_port != proxy_port),
+ "Managed Sync fixture ports must differ"
+ );
let workspace = PathBuf::from(env!("CARGO_WORKSPACE_DIR"));
Self {
compose_file: workspace.join("fixtures/managed-sync/compose.yaml"),
keep,
+ evaluation,
minio_port,
project: format!("opendal-ofs-managed-sync-{}", std::process::id()),
+ proxy_admin_port,
+ proxy_port,
started: false,
variant: variant.into(),
}
@@ -170,8 +206,19 @@ impl Fixture {
pub(super) fn start(mut self) -> Self {
self.started = true;
- run_logged(self.compose().args(["up", "--detach", "minio"]));
- self.wait_until_ready();
+ if self.evaluation.network_enabled() {
+ run_logged(
+ self.compose()
+ .args(["up", "--detach", "minio", "toxiproxy"]),
+ );
+ } else {
+ run_logged(self.compose().args(["up", "--detach", "minio"]));
+ }
+ self.wait_until_ready(self.minio_port);
+ if self.evaluation.network_enabled() {
+ self.configure_network();
+ self.wait_until_ready(self.proxy_port);
+ }
self
}
@@ -193,25 +240,132 @@ impl Fixture {
"bub-a",
"bub-b",
]));
- self.wait_until_ready();
+ self.wait_until_ready(self.minio_port);
self
}
- fn wait_until_ready(&self) {
+ fn wait_until_ready(&self, port: u16) {
let deadline = Instant::now() + FIXTURE_READY_TIMEOUT;
+ let response_timeout = Duration::from_millis(
+ self.evaluation
+ .network_rtt_ms()
+ .saturating_add(self.evaluation.network_jitter_ms())
+ .saturating_add(1_000),
+ );
while Instant::now() < deadline {
- if minio_is_ready(self.minio_port) {
+ if minio_is_ready(port, response_timeout) {
return;
}
thread::sleep(Duration::from_millis(200));
}
panic!(
"MinIO did not become ready on 127.0.0.1:{} within {} seconds",
- self.minio_port,
+ port,
FIXTURE_READY_TIMEOUT.as_secs()
);
}
+ fn configure_network(&self) {
+ let deadline = Instant::now() + FIXTURE_READY_TIMEOUT;
+ loop {
+ let output = self
+ .toxiproxy_cli()
+ .args([
+ "create",
+ "--listen",
+ "0.0.0.0:8666",
+ "--upstream",
+ "minio:9000",
+ TOXIPROXY_NAME,
+ ])
+ .output()
+ .expect("configure Managed Sync fault proxy");
+ if output.status.success() {
+ break;
+ }
+ assert!(
+ Instant::now() < deadline,
+ "configure Managed Sync fault proxy failed: {}",
+ output_text(&output.stderr)
+ );
+ thread::sleep(Duration::from_millis(200));
+ }
+ if self.evaluation.network_rtt_ms() != 0 || self.evaluation.network_jitter_ms() != 0 {
+ let latency = format!("latency={}", self.evaluation.network_rtt_ms());
+ let jitter = format!("jitter={}", self.evaluation.network_jitter_ms());
+ self.add_toxic([
+ "toxic",
+ "add",
+ "--toxicName",
+ "network-rtt",
+ "--type",
+ "latency",
+ "--attribute",
+ &latency,
+ "--attribute",
+ &jitter,
+ "--downstream",
+ TOXIPROXY_NAME,
+ ]);
+ }
+ let bandwidth = self.evaluation.connection_bandwidth_kb_per_second();
+ if bandwidth != 0 {
+ let rate = format!("rate={bandwidth}");
+ self.add_toxic([
+ "toxic",
+ "add",
+ "--toxicName",
+ "upload-bandwidth",
+ "--type",
+ "bandwidth",
+ "--attribute",
+ &rate,
+ "--upstream",
+ TOXIPROXY_NAME,
+ ]);
+ self.add_toxic([
+ "toxic",
+ "add",
+ "--toxicName",
+ "download-bandwidth",
+ "--type",
+ "bandwidth",
+ "--attribute",
+ &rate,
+ "--downstream",
+ TOXIPROXY_NAME,
+ ]);
+ }
+ }
+
+ fn add_toxic<const N: usize>(&self, arguments: [&str; N]) {
+ let output = self
+ .toxiproxy_cli()
+ .args(arguments)
+ .output()
+ .expect("add Managed Sync network condition");
+ assert!(
+ output.status.success(),
+ "add Managed Sync network condition failed: {}",
+ output_text(&output.stderr)
+ );
+ }
+
+ fn toxiproxy_cli(&self) -> Command {
+ let mut command = self.compose();
+ command.args([
+ "run",
+ "--rm",
+ "--no-deps",
+ "--entrypoint",
+ "/toxiproxy-cli",
+ "toxiproxy",
+ "--host",
+ "http://toxiproxy:8474",
+ ]);
+ command
+ }
+
pub(crate) fn create_bucket(&self) {
run_logged(self.compose().args([
"run",
@@ -235,7 +389,7 @@ impl Fixture {
pub(crate) fn storage_url(&self, root: &str) -> String {
format!(
"s3://managed-sync/{root}?endpoint=http%3A%2F%2F127.0.0.1%3A{}®ion=us-east-1",
- self.minio_port
+ self.endpoint_port()
)
}
@@ -285,7 +439,11 @@ impl Fixture {
) -> T {
let before = self.provider_snapshot();
let started = Instant::now();
- let result = action();
+ let chaos = self
+ .evaluation
+ .start_chaos(self.proxy_admin_port, root, stage);
+ let result = catch_unwind(AssertUnwindSafe(action));
+ let chaos = chaos.map(ChaosController::stop);
let elapsed = started.elapsed();
let provider = self.provider_snapshot().difference(&before);
let inventory = self.storage_inventory(root);
@@ -297,8 +455,13 @@ impl Fixture {
logical,
&provider,
&inventory,
+ &self.evaluation,
+ chaos.as_ref(),
);
- result
+ match result {
+ Ok(result) => result,
+ Err(payload) => resume_unwind(payload),
+ }
}
fn provider_snapshot(&self) -> ProviderSnapshot {
@@ -416,7 +579,7 @@ impl Fixture {
pub(super) fn container_storage_url(&self, root: &str) -> String {
format!(
"s3://managed-sync/{root}?endpoint=http%3A%2F%2F127.0.0.1%3A{}®ion=us-east-1",
- self.minio_port
+ self.endpoint_port()
)
}
@@ -454,6 +617,11 @@ impl Fixture {
command
.env("OFS_BUB_IMAGE", self.bub_image())
.env("OFS_MANAGED_SYNC_MINIO_PORT", self.minio_port.to_string())
+ .env(
+ "OFS_MANAGED_SYNC_PROXY_ADMIN_PORT",
+ self.proxy_admin_port.to_string(),
+ )
+ .env("OFS_MANAGED_SYNC_PROXY_PORT", self.proxy_port.to_string())
.args(["--project-name", &self.project, "--file"])
.arg(&self.compose_file);
command
@@ -463,6 +631,14 @@ impl Fixture {
format!("{}-bub:local", self.project)
}
+ const fn endpoint_port(&self) -> u16 {
+ if self.evaluation.network_enabled() {
+ self.proxy_port
+ } else {
+ self.minio_port
+ }
+ }
+
fn stop(&self) -> bool {
self.compose()
.args(["down", "--volumes", "--remove-orphans"])
@@ -507,7 +683,7 @@ pub(super) fn container_http_proxy() -> Option<String> {
.find_map(|name| env::var(name).ok().filter(|value| !value.is_empty()))
}
-fn minio_is_ready(port: u16) -> bool {
+fn minio_is_ready(port: u16, response_timeout: Duration) -> bool {
let address = format!("127.0.0.1:{port}");
let Ok(mut stream) = TcpStream::connect_timeout(
&address.parse().expect("loopback fixture address is valid"),
@@ -515,7 +691,7 @@ fn minio_is_ready(port: u16) -> bool {
) else {
return false;
};
- let _ = stream.set_read_timeout(Some(Duration::from_secs(1)));
+ let _ = stream.set_read_timeout(Some(response_timeout));
if stream
.write_all(
b"GET /minio/health/ready HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n",
@@ -620,6 +796,8 @@ fn report_observation(
logical: LogicalIo,
provider: &ProviderSnapshot,
inventory: &StorageInventory,
+ evaluation: &EvaluationOptions,
+ chaos: Option<&ChaosReport>,
) {
let mut data_objects = 0_u64;
let mut data_bytes = 0_u64;
@@ -646,6 +824,7 @@ fn report_observation(
let amplification = |provider_bytes: u64, logical_bytes: u64| {
(logical_bytes != 0).then(|| provider_bytes as f64 / logical_bytes as f64)
};
+ let billing = estimate_billing(evaluation, provider, inventory);
println!(
"managed-sync observation: {}",
serde_json::json!({
@@ -672,6 +851,66 @@ fn report_observation(
"metadata_bytes": metadata_bytes,
"classes": classes,
},
+ "network": {
+ "rtt_ms": evaluation.network_rtt_ms(),
+ "jitter_ms": evaluation.network_jitter_ms(),
+ "connection_bandwidth_kb_per_second": evaluation.connection_bandwidth_kb_per_second(),
+ },
+ "chaos": chaos.map(|report| serde_json::json!({
+ "seed": evaluation.chaos_seed(),
+ "schedule_seed": report.schedule_seed,
+ "reset_rate_per_million": evaluation.chaos_reset_rate_per_million(),
+ "tick_ms": evaluation.chaos_tick_ms(),
+ "outage_ms": evaluation.chaos_outage_ms(),
+ "faults_injected": report.faults_injected,
+ })),
+ "billing": billing,
})
);
}
+
+fn estimate_billing(
+ evaluation: &EvaluationOptions,
+ provider: &ProviderSnapshot,
+ inventory: &StorageInventory,
+) -> Option<serde_json::Value> {
+ if !evaluation.billing_enabled() {
+ return None;
+ }
+ let (read_requests, write_list_requests) =
+ provider
+ .requests
+ .iter()
+ .fold((0_u64, 0_u64), |(reads, writes), (name, count)| {
+ if name.starts_with("Get") || name.starts_with("Head") {
+ (reads + count, writes)
+ } else {
+ (reads, writes + count)
+ }
+ });
+ let stored_bytes = inventory
+ .classes
+ .values()
+ .map(|class| class.bytes)
+ .sum::<u64>();
+ let request_usd = read_requests as f64 / 1_000_000.0
+ * evaluation.read_request_usd_per_million()
+ + write_list_requests as f64 / 1_000_000.0
+ * evaluation.write_list_request_usd_per_million();
+ let gib = 1024.0 * 1024.0 * 1024.0;
+ let egress_usd = provider.read_bytes as f64 / gib * evaluation.egress_usd_per_gib();
+ let ingress_usd = provider.write_bytes as f64 / gib * evaluation.ingress_usd_per_gib();
+ let transfer_usd = egress_usd + ingress_usd;
+ let storage_usd_per_month = stored_bytes as f64 / gib * evaluation.storage_usd_per_gib_month();
+ Some(serde_json::json!({
+ "read_requests": read_requests,
+ "write_list_requests": write_list_requests,
+ "request_usd": request_usd,
+ "egress_usd": egress_usd,
+ "ingress_usd": ingress_usd,
+ "transfer_usd": transfer_usd,
+ "operation_usd": request_usd + transfer_usd,
+ "stored_bytes": stored_bytes,
+ "storage_usd_per_month": storage_usd_per_month,
+ }))
+}
diff --git a/xtask/src/managed_sync/scale.rs b/xtask/src/managed_sync/scale.rs
index 03b2e13..2cb7ba6 100644
--- a/xtask/src/managed_sync/scale.rs
+++ b/xtask/src/managed_sync/scale.rs
@@ -25,6 +25,7 @@ use std::path::{Path, PathBuf};
use std::time::{Instant, SystemTime, UNIX_EPOCH};
use super::cli::{BuildProfile, Ofs, output_text, require_success};
+use super::evaluation::EvaluationOptions;
use super::fixture::{Fixture, LogicalIo};
const TINY_FILE_COUNT: u64 = 1_000_000;
@@ -43,6 +44,7 @@ pub(crate) fn run(
duration_seconds: u64,
sync_interval_seconds: u64,
file_size_mib: u64,
+ evaluation: EvaluationOptions,
) {
if scenario == "random-write" {
random_write::run(
@@ -51,6 +53,7 @@ pub(crate) fn run(
duration_seconds,
sync_interval_seconds,
file_size_mib,
+ evaluation,
);
return;
}
@@ -58,7 +61,7 @@ pub(crate) fn run(
let build_profile = BuildProfile::parse(build_profile);
let ofs = Ofs::release(build_profile);
ofs.build();
- let fixture = Fixture::new(keep, build_profile.name()).start();
+ let fixture = Fixture::new(keep, build_profile.name(), evaluation).start();
fixture.create_bucket();
let work = ScaleRoot::new(keep);
let replicas = [
diff --git a/xtask/src/managed_sync/scale/random_write.rs b/xtask/src/managed_sync/scale/random_write.rs
index 933cd18..24884f2 100644
--- a/xtask/src/managed_sync/scale/random_write.rs
+++ b/xtask/src/managed_sync/scale/random_write.rs
@@ -28,6 +28,7 @@ use std::time::{Duration, Instant};
use super::{ScaleRoot, require_same_tree, write_xof};
use crate::managed_sync::cli::{BuildProfile, Ofs, output_text, require_success};
+use crate::managed_sync::evaluation::EvaluationOptions;
use crate::managed_sync::fixture::{Fixture, LogicalIo};
const MIB: u64 = 1024 * 1024;
@@ -41,13 +42,14 @@ pub(super) fn run(
duration_seconds: u64,
sync_interval_seconds: u64,
file_size_mib: u64,
+ evaluation: EvaluationOptions,
) {
let workload = Workload::new(duration_seconds, sync_interval_seconds, file_size_mib);
let build_profile = BuildProfile::parse(build_profile);
let ofs = Ofs::release(build_profile);
ofs.build();
- let fixture = Fixture::new(keep, build_profile.name()).start();
+ let fixture = Fixture::new(keep, build_profile.name(), evaluation).start();
fixture.create_bucket();
let work = ScaleRoot::new(keep);
let source = work.path.join("replica-a");