Kinesin/waterfall
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Adding example for agent execution

Browse Source
This commit is contained in:
Kinesin Data Technologies Incorporated
2022-10-05 10:35:48 -03:00
parent d82b000f9b
commit 923025bc49
3 changed files with 40 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/config_wfw.json
+17
View File
@@ -0,0 +1,17 @@
{
"storage": {
"type": "redis",
"url": "redis://localhost",
"prefix": "world"
},
"executor": {
"type": "agent",
"targets": [
{
"base_url": "http://localhost:2504/api/v1",
"resources": { "cores": 10 }
}
]
}
}
examples/world.json
+6 -6
View File
@@ -7,9 +7,9 @@
},
"tasks": {
"task_a": {
"up": { "command": "/usr//bin/touch ${HOME}/task_a_${yyyymmdd}", "resources": { "cores": 1 } },
"down": { "command": "/bin/rm ${HOME}/task_a_${yyyymmdd}", "resources": { "cores": 1 } },
"check": { "command": "/bin/test -e ${HOME}/task_a_${yyyymmdd}", "resources": { "cores": 1 } },
"up": { "command": "/usr//bin/touch ${HOME}/task_a_${yyyymmdd}${hhmmss}", "resources": { "cores": 1 } },
"down": { "command": "/bin/rm ${HOME}/task_a_${yyyymmdd}${hhmmss}", "resources": { "cores": 1 } },
"check": { "command": "/bin/test -e ${HOME}/task_a_${yyyymmdd}${hhmmss}", "resources": { "cores": 1 } },
"provides": [ "task_a" ],
@@ -21,9 +21,9 @@
"valid_to": "2022-01-08T09:00:00"
},
"task_b": {
"up": { "command": "/usr//bin/touch ${HOME}/task_b_${yyyymmdd}", "resources": { "cores": 1 } },
"down": { "command": "/bin/rm ${HOME}/task_b_${yyyymmdd}", "resources": { "cores": 1 } },
"check": { "command": "/bin/test -e ${HOME}/task_b_${yyyymmdd}", "resources": { "cores": 1 } },
"up": { "command": "/usr//bin/touch ${HOME}/task_b_${yyyymmdd}${hhmmss}", "resources": { "cores": 1 } },
"down": { "command": "/bin/rm ${HOME}/task_b_${yyyymmdd}${hhmmss}", "resources": { "cores": 1 } },
"check": { "command": "/bin/test -e ${HOME}/task_b_${yyyymmdd}${hhmmss}", "resources": { "cores": 1 } },
"provides": [ "task_b" ],
"requires": [ { "resource": "task_a", "offset": 0 } ],
src/executors/agent_executor.rs
+15 -20
View File
@@ -115,7 +115,7 @@ async fn submit_task(
output_options: TaskOutputOptions,
client: reqwest::Client,
varmap: VarMap,
) -> TaskAttempt {
) -> Result<TaskAttempt> {
let submit_url = format!("{}/run", base_url);
let mut attempt = TaskAttempt::new();
let submission = TaskSubmission {
@@ -130,28 +130,21 @@ async fn submit_task(
attempt
.executor
.push(format!("Executed on agent at {}", base_url));
Ok(attempt)
} else {
attempt.succeeded = false;
attempt.infra_failure = true;
attempt.executor.push(format!(
Err(anyhow!(
"Unable to dispatch to agent at {}: {:?}",
base_url,
result.text().await.unwrap()
));
))
}
}
Err(e) => {
warn!("Failed to submit task: {:?}", e);
attempt.succeeded = false;
attempt.infra_failure = true;
attempt.executor.push(format!(
Err(e) => Err(anyhow!(
"Unable to dispatch to agent at {}: {:?}",
base_url, e
));
base_url,
e
)),
}
}
attempt
}
// async fn select_target() -> Option<usize> {}
@@ -211,11 +204,12 @@ async fn start_agent_executor(
}) {
// There is a remote agent with capacity
Some((tid, target)) => {
info!("Dispatching job to {}", target.base_url);
target.current_resources.sub(&resources).unwrap();
let base_url = target.base_url.clone();
let submit_client = client.clone();
running.push(tokio::spawn(async move {
let attempt = submit_task(
let res = submit_task(
base_url,
details,
output_options,
@@ -223,8 +217,11 @@ async fn start_agent_executor(
varmap,
)
.await;
let rc = attempt.succeeded;
let mut rc = false;
if let Ok(attempt) = res {
response.send(attempt).unwrap();
rc = true;
}
(tid, resources, rc)
}));
break;
@@ -233,13 +230,11 @@ async fn start_agent_executor(
None => {
// Give the outstanding tasks a chance to complete or agents
// recover
tokio::time::sleep(tokio::time::Duration::from_millis(250)).await;
info!("Waiting to run message");
tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
// Refresh any disabled targets
for (tid, target) in targets.iter_mut().enumerate() {
if target.enabled {
info!("Skipping {} as it is enabled", target.base_url);
continue;
}
target.refresh_resources(&client).await;