Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: add throttle requests functionality for simulate #105

Merged
merged 2 commits into from
Nov 7, 2023
Merged

feat: add throttle requests functionality for simulate #105

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
7c5b2a8
feat: add throttle requests functionality for simulate
dbcfd Nov 7, 2023
1e4250e
fix: clippy and test
dbcfd Nov 7, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 33 additions & 0 deletions operator/src/simulation/controller.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -172,6 +172,7 @@ async fn reconcile(
run_time: spec.run_time.to_owned(),
nonce: status.nonce,
job_image_config: job_image_config.clone(),
throttle_requests: spec.throttle_requests,
};

apply_manager(cx.clone(), &ns, simulation.clone(), manager_config).await?;
Expand Down Expand Up @@ -779,4 +780,36 @@ mod tests {
.expect("reconciler");
timeout_after_1s(mocksrv).await;
}
#[tokio::test]
#[traced_test]
async fn reconcile_throttle() {
let mock_rpc_client = MockIpfsRpcClientTest::new();
let (testctx, api_handle) = Context::test(mock_rpc_client);
let fakeserver = ApiServerVerifier::new(api_handle);
let simulation = Simulation::test().with_spec(SimulationSpec {
throttle_requests: Some(100),
..Default::default()
});
let mut stub = Stub::default();
stub.manager_job.patch(expect![[r#"
--- original
+++ modified
@@ -74,6 +74,10 @@
{
"name": "DID_PRIVATE_KEY",
"value": "86dce513cf0a37d4acd6d2c2e00fe4b95e0e655ca51e1a890808f5fa6f4fe65a"
+ },
+ {
+ "name": "SIMULATE_THROTTLE_REQUESTS",
+ "value": "100"
}
],
"image": "public.ecr.aws/r5b3e0r5/3box/keramik-runner:latest",
"#]]);
let mocksrv = stub.run(fakeserver);
reconcile(Arc::new(simulation), testctx)
.await
.expect("reconciler");
timeout_after_1s(mocksrv).await;
}
}
131 changes: 68 additions & 63 deletions operator/src/simulation/manager.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,11 +32,78 @@ pub struct ManagerConfig {
pub scenario: String,
pub users: u32,
pub run_time: u32,
pub throttle_requests: Option<usize>,
pub nonce: u32,
pub job_image_config: JobImageConfig,
}

pub fn manager_job_spec(config: ManagerConfig) -> JobSpec {
let mut env_vars = vec![
EnvVar {
name: "RUNNER_OTLP_ENDPOINT".to_owned(),
value: Some("http://otel:4317".to_owned()),
..Default::default()
},
EnvVar {
name: "RUST_LOG".to_owned(),
value: Some("info,keramik_runner=trace".to_owned()),
..Default::default()
},
EnvVar {
name: "SIMULATE_SCENARIO".to_owned(),
value: Some(config.scenario.to_owned()),
..Default::default()
},
EnvVar {
name: "SIMULATE_MANAGER".to_owned(),
value: Some("true".to_owned()),
..Default::default()
},
EnvVar {
name: "SIMULATE_PEERS_PATH".to_owned(),
value: Some("/keramik-peers/peers.json".to_owned()),
..Default::default()
},
EnvVar {
name: "SIMULATE_TARGET_PEER".to_owned(),
value: Some(0.to_string()),
..Default::default()
},
EnvVar {
name: "SIMULATE_NONCE".to_owned(),
value: Some(config.nonce.to_string()),
..Default::default()
},
EnvVar {
name: "SIMULATE_USERS".to_owned(),
value: Some(config.users.to_string()),
..Default::default()
},
EnvVar {
name: "SIMULATE_RUN_TIME".to_owned(),
value: Some(format!("{}m", config.run_time)),
..Default::default()
},
EnvVar {
name: "DID_KEY".to_owned(),
value: Some("did:key:z6Mkqn5jbycThHcBtakJZ8fHBQ2oVRQhXQEdQk5ZK2NDtNZA".to_owned()),
..Default::default()
},
EnvVar {
name: "DID_PRIVATE_KEY".to_owned(),
value: Some(
"86dce513cf0a37d4acd6d2c2e00fe4b95e0e655ca51e1a890808f5fa6f4fe65a".to_owned(),
),
..Default::default()
},
];
if let Some(throttle_requests) = config.throttle_requests {
env_vars.push(EnvVar {
name: "SIMULATE_THROTTLE_REQUESTS".to_owned(),
value: Some(throttle_requests.to_string()),
..Default::default()
})
}
JobSpec {
backoff_limit: Some(4),
template: PodTemplateSpec {
Expand All @@ -58,69 +125,7 @@ pub fn manager_job_spec(config: ManagerConfig) -> JobSpec {
"/usr/bin/keramik-runner".to_owned(),
"simulate".to_owned(),
]),
env: Some(vec![
EnvVar {
name: "RUNNER_OTLP_ENDPOINT".to_owned(),
value: Some("http://otel:4317".to_owned()),
..Default::default()
},
EnvVar {
name: "RUST_LOG".to_owned(),
value: Some("info,keramik_runner=trace".to_owned()),
..Default::default()
},
EnvVar {
name: "SIMULATE_SCENARIO".to_owned(),
value: Some(config.scenario.to_owned()),
..Default::default()
},
EnvVar {
name: "SIMULATE_MANAGER".to_owned(),
value: Some("true".to_owned()),
..Default::default()
},
EnvVar {
name: "SIMULATE_PEERS_PATH".to_owned(),
value: Some("/keramik-peers/peers.json".to_owned()),
..Default::default()
},
EnvVar {
name: "SIMULATE_TARGET_PEER".to_owned(),
value: Some(0.to_string()),
..Default::default()
},
EnvVar {
name: "SIMULATE_NONCE".to_owned(),
value: Some(config.nonce.to_string()),
..Default::default()
},
EnvVar {
name: "SIMULATE_USERS".to_owned(),
value: Some(config.users.to_string()),
..Default::default()
},
EnvVar {
name: "SIMULATE_RUN_TIME".to_owned(),
value: Some(format!("{}m", config.run_time)),
..Default::default()
},
EnvVar {
name: "DID_KEY".to_owned(),
value: Some(
"did:key:z6Mkqn5jbycThHcBtakJZ8fHBQ2oVRQhXQEdQk5ZK2NDtNZA"
.to_owned(),
),
..Default::default()
},
EnvVar {
name: "DID_PRIVATE_KEY".to_owned(),
value: Some(
"86dce513cf0a37d4acd6d2c2e00fe4b95e0e655ca51e1a890808f5fa6f4fe65a"
.to_owned(),
),
..Default::default()
},
]),
env: Some(env_vars),
volume_mounts: Some(vec![VolumeMount {
mount_path: "/keramik-peers".to_owned(),
name: "keramik-peers".to_owned(),
Expand Down
2 changes: 2 additions & 0 deletions operator/src/simulation/spec.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@ pub struct SimulationSpec {
pub image: Option<String>,
/// Pull policy for image.
pub image_pull_policy: Option<String>,
/// Throttle requests (per second) for a simulation
pub throttle_requests: Option<usize>,
}

/// Current status of a simulation.
Expand Down
10 changes: 9 additions & 1 deletion runner/src/simulate.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,10 @@ pub struct Opts {
/// All workers and manager must be given the same nonce.
#[arg(long, env = "SIMULATE_NONCE")]
nonce: u64,

/// Option to throttle requests (per second) for load control
#[arg(long, env = "SIMULATE_THROTTLE_REQUESTS", default_value = "None")]
throttle_requests: Option<usize>,
}

#[derive(Debug, Clone, Copy)]
Expand Down Expand Up @@ -140,6 +144,7 @@ pub async fn simulate(opts: Opts) -> Result<()> {
.get(opts.target_peer)
.ok_or_else(|| anyhow!("target peer too large, not enough peers"))?,
)?,
opts.throttle_requests,
)
};

Expand Down Expand Up @@ -171,7 +176,7 @@ fn manager_config(count: usize, users: usize, run_time: String) -> GooseConfigur
config.run_time = run_time;
config
}
fn worker_config(target_peer_addr: String) -> GooseConfiguration {
fn worker_config(target_peer_addr: String, throttle_requests: Option<usize>) -> GooseConfiguration {
let mut config = GooseConfiguration::default();
config.request_log = "request.log".to_owned();
config.log_level = 2;
Expand All @@ -181,6 +186,9 @@ fn worker_config(target_peer_addr: String) -> GooseConfiguration {
// domain name explicitly.
config.manager_host = "manager.goose".to_owned();
config.manager_port = 5115;
if let Some(throttle_requests) = throttle_requests {
config.throttle_requests = throttle_requests
}
config
}

Expand Down
Loading