Skip to content

feat: add RESP compatibility E2E coverage and improve task output #412

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

Merged
starrysky9959 merged 2 commits into from
Jun 1, 2026
Merged

feat: add RESP compatibility E2E coverage and improve task output #412

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
eeb7ae0
test: add resp compatibility e2e coverage
starrysky9959 Jun 1, 2026
560ac25
feat: improve task progress output
starrysky9959 Jun 1, 2026
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
11 changes: 9 additions & 2 deletions .github/workflows/test.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,10 @@ on:
description: "Concurrent execution slots per worker"
required: false
default: "4"
resp_compat_version:
description: "RESP compatibility target Redis version"
required: false
default: "7.0.0"

env:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
Expand Down Expand Up @@ -83,11 +87,13 @@ jobs:
duration="${{ github.event.inputs.duration || '20' }}"
repeat="${{ github.event.inputs.repeat || '8' }}"
inflight="${{ github.event.inputs.inflight || '4' }}"
resp_compat_version="${{ github.event.inputs.resp_compat_version || '7.0.0' }}"
else
workers="8"
duration="20"
repeat="8"
inflight="4"
resp_compat_version="7.0.0"
fi

export DOCKER_E2E_DIR="$PWD/tests/docker_ha"
Expand All @@ -99,9 +105,10 @@ jobs:
export WORKERS="${workers}"
export INFLIGHT="${inflight}"
export REPEAT="${repeat}"
export RESP_COMPAT_VERSION="${resp_compat_version}"
export SKIP_DEPS=1
export STEPS="launch,cluster-update,monitor-update,eloqctl-mutate,py-stress,go-stress,ts-stress,remove"
echo "Running E2E with WORKERS=${WORKERS} INFLIGHT=${INFLIGHT} DURATION_SECONDS=${DURATION_SECONDS} REPEAT=${REPEAT}"
export STEPS="launch,cluster-update,monitor-update,eloqctl-mutate,py-stress,go-stress,ts-stress,resp-compat,remove"
echo "Running E2E with WORKERS=${WORKERS} INFLIGHT=${INFLIGHT} DURATION_SECONDS=${DURATION_SECONDS} REPEAT=${REPEAT} RESP_COMPAT_VERSION=${RESP_COMPAT_VERSION}"
tests/e2e/devctl.sh stress "${STEPS}"

- name: Cleanup Docker
Expand Down
3 changes: 3 additions & 0 deletions .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,3 +50,6 @@ tests/**/exported.yaml
tests/**/node_modules/
tests/**/__pycache__/
tests/**/go-cluster-tls-stress
tests/e2e/cmd-stress-*.log
tests/e2e/*.generated.yaml
.tmp-e2e-download-cache/
27 changes: 18 additions & 9 deletions src/cluster_mgr/src/cli/task/task_base.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -425,8 +425,6 @@ impl TaskMgr {
if execution_value.contains_key("_FINISH_SIGNAL") {
break;
}
let start = "=> ";
let end = "---------------------------";
let cmd = execution_value
.get(CMD)
.map(|v| TaskArgValue::into_inner_value::<String>(v.clone()))
Expand All @@ -435,17 +433,24 @@ impl TaskMgr {
.get(CMD_STATUS)
.map(|v| TaskArgValue::into_inner_value::<i32>(v.clone()))
.unwrap_or(0);
let status = if status_code == 0 {
"Success".green().to_string()
} else {
"Failure".red().to_string()
};
let cmd_out = execution_value
.get(CMD_OUTPUT)
.map(|v| TaskArgValue::into_inner_value::<String>(v.clone()))
.unwrap_or_else(|| "".to_string());
let s = format!("{start}{task_id}\n{cmd}\n{status}; {cmd_out}\n{end}\n");

if verbose || status_code != 0 {
let status = if status_code == 0 {
format!("{}", "✓ Success".green())
} else {
format!("{}", "✗ Failed".red())
};
let s = format!(
" ┌─ {}\n │ {}\n │ {}\n │ {}\n └────\n",
task_id.dimmed(),
cmd.dimmed(),
status,
if cmd_out.is_empty() { "" } else { &cmd_out }
);
Comment on lines +447 to +453

@coderabbitai coderabbitai Bot Jun 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Empty CMD_OUTPUT still renders a blank line.

The current implementation displays │ even when cmd_out is empty (line 452). The AI summary indicates CMD_OUTPUT should only be appended when non-empty, but this logic renders an extra blank line in the box layout.

🔧 Proposed fix to conditionally include CMD_OUTPUT line 
-                        let s = format!(
-                            "  ┌─ {}\n  │ {}\n  │ {}\n  │ {}\n  └────\n",
-                            task_id.dimmed(),
-                            cmd.dimmed(),
-                            status,
-                            if cmd_out.is_empty() { "" } else { &cmd_out }
-                        );
+                        let output_line = if cmd_out.is_empty() {
+                            String::new()
+                        } else {
+                            format!("  │ {}\n", cmd_out)
+                        };
+                        let s = format!(
+                            "  ┌─ {}\n  │ {}\n  │ {}\n{}\n  └────\n",
+                            task_id.dimmed(),
+                            cmd.dimmed(),
+                            status,
+                            output_line.trim_end()
+                        );
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/cluster_mgr/src/cli/task/task_base.rs` around lines 447 - 453, The
formatted box always emits a "│ " line for CMD_OUTPUT even when cmd_out is
empty; update the construction of s (the formatted string around task_id, cmd,
status, cmd_out) so the CMD_OUTPUT line is only included when cmd_out is
non-empty: locate the formatting that builds s (references: task_id, cmd,
status, cmd_out) and either branch to two format variants (one with the "  │
{}\n" line and one without) or build s incrementally (push the CMD_OUTPUT line
only if !cmd_out.is_empty()) so no blank "  │ " line is rendered and the box
borders remain correct.

match writer {
Some(ref mut w) => w.write_all(s.as_bytes())?,
None => println!("{s}"),
Expand All @@ -464,7 +469,11 @@ impl TaskMgr {
.map(|id| id.host.clone())
.unwrap_or_else(|| "unknown".to_string());
let s = format!(
"=> {task_id}\nAction: {action}\nHost: {host}\nFailure: {}\n---------------------------\n",
" {} {}\n Action: {}\n Host: {}\n Detail: {}\n",
"✗".red(),
task_id.red(),
action,
host,
summarize_error(&err_msg)
);
match writer {
Expand Down
183 changes: 102 additions & 81 deletions src/cluster_mgr/src/cli/task/task_controller.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ use chrono::DateTime;
use chrono::Utc;
use futures_async_stream::try_stream;
use itertools::Itertools;
use owo_colors::OwoColorize;
use std::sync::Arc;
use tracing::{error, info, warn};

Expand Down Expand Up @@ -48,78 +49,90 @@ impl TaskController {
.collect_vec();

if task_ids.iter().all(|id| id.cmd == "ssh-check") {
return "Check SSH connectivity".to_string();
return "Checking SSH connectivity".to_string();
}
if task_ids.iter().all(|id| id.cmd == "run_deps") {
return "Install runtime dependencies".to_string();
return "Installing runtime dependencies".to_string();
}
if task_ids.iter().all(|id| id.cmd == "check") {
return "Check port availability".to_string();
return "Checking port availability".to_string();
}
if task_ids
.iter()
.all(|id| id.cmd == "deploy" && id.task == "mkdir")
{
return "Create remote directories".to_string();
return "Creating remote directories".to_string();
}
if task_ids
.iter()
.all(|id| id.cmd == "deploy" && id.task.ends_with("_download"))
{
return "Download release packages".to_string();
return "Downloading release packages".to_string();
}
if task_ids.iter().all(|id| id.cmd == "extract") {
return "Extract release packages locally".to_string();
return "Extracting release packages".to_string();
}
if task_ids.iter().all(|id| {
matches!(id.cmd.as_str(), "deploy" | "update")
&& (id.task.contains("upload") || id.task.starts_with("deploy_eloq_all_"))
}) {
return "Sync binaries and support files".to_string();
return "Syncing binaries and config files".to_string();
}
if task_ids.iter().all(|id| id.cmd == "config-update") {
return "Upload generated configuration".to_string();
return "Uploading configuration files".to_string();
}
if task_ids.iter().all(|id| id.cmd == "install") {
return "Bootstrap cluster metadata".to_string();
return "Bootstrapping cluster metadata".to_string();
}
if task_ids.iter().all(|id| id.cmd == "start") {
return "Start services".to_string();
return "Starting services".to_string();
}
if task_ids.iter().all(|id| id.cmd == "stop") {
return "Stopping services".to_string();
}
if task_ids.iter().all(|id| id.cmd == "restart") {
return "Restarting services".to_string();
}
if task_ids.iter().all(|id| id.cmd == "topology") {
return "Refresh cluster topology".to_string();
return "Refreshing cluster topology".to_string();
}
if task_ids.iter().all(|id| id.cmd == "monitor") {
if task_ids.iter().all(|id| id.task.contains("-stop-")) {
return "Stop monitor components".to_string();
return "Stopping monitor components".to_string();
}
if task_ids.iter().all(|id| id.task.contains("-status-")) {
return "Check monitor status".to_string();
return "Checking monitor status".to_string();
}
if task_ids.iter().all(|id| id.task.contains("-start-")) {
return "Start monitor components".to_string();
return "Starting monitor components".to_string();
}
return "Manage monitor components".to_string();
return "Managing monitor components".to_string();
}
if task_ids
.iter()
.all(|id| id.cmd == "remove" && id.task.starts_with("clean_log@"))
{
return "Delete log service data".to_string();
return "Deleting log service data".to_string();
}
if task_ids
.iter()
.all(|id| id.cmd == "remove" && id.task.starts_with("clean@"))
{
return "Delete cluster install directories".to_string();
return "Deleting cluster files".to_string();
}
if task_ids.iter().all(|id| {
id.cmd == "backup"
&& (id.task.starts_with("delete-")
|| id.task.starts_with("remove-local-")
|| id.task.starts_with("clean-backup"))
}) {
return "Delete backup data".to_string();
return "Deleting backup data".to_string();
}
if task_ids.iter().all(|id| id.cmd == "backup") {
return "Running backup".to_string();
}
if task_ids.iter().all(|id| id.cmd == "scale") {
return "Scaling cluster nodes".to_string();
}

Self::stage_label(task_group)
Expand Down Expand Up @@ -198,67 +211,68 @@ impl TaskController {
Config::Proxy(cfg) => cfg.proxy_service.proxy_name.clone(),
};

splits
.iter()
//.enumerate()
.for_each(|execution_context| {
let tx_arc = Arc::new(&self.tx);
let task_group_arc = Arc::new(task_group.clone());
let name = name.clone();
let join = tokio::task::spawn(async move {
let task = &execution_context.task;
let task_input = execution_context.task_input.clone();
let task_host = &execution_context.task_host;
let task_id = task.identifier();
if is_verbose_task_output() {
println!(" -> {}", task_id.format_string());
}
let start_time = std::time::Instant::now();
let execution_rs = task.execute(task_host.clone(), task_input).await;
let elapsed = start_time.elapsed();
info!("Task {:?} execution complete", task_id);
let conn_tuple = task_host.ssh_conn_tuple();
let post_execute_rs = post_task_execute!(
execution_rs,
name,
task_id.as_json_string(),
task_group_arc.as_str(),
conn_tuple.2
);
info!("Save Task {:?} execution status complete", task_id);
assert!(post_execute_rs.is_ok());
let result = match execution_rs {
Ok(rs) => {
if is_verbose_task_output() {
println!(
" ok {:.1}s {}",
elapsed.as_secs_f32(),
task_id.format_string()
);
}
TaskResultEnum::Success(rs)
splits.iter().for_each(|execution_context| {
let tx_arc = Arc::new(&self.tx);
let task_group_arc = Arc::new(task_group.clone());
let name = name.clone();
let join = tokio::task::spawn(async move {
let task = &execution_context.task;
let task_input = execution_context.task_input.clone();
let task_host = &execution_context.task_host;
let task_id = task.identifier();
if is_verbose_task_output() {
println!(" {} {}", "→".cyan(), user_friendly_task_summary(&task_id));
}
let start_time = std::time::Instant::now();
let execution_rs = task.execute(task_host.clone(), task_input).await;
let elapsed = start_time.elapsed();
info!("Task {:?} execution complete", task_id);
let conn_tuple = task_host.ssh_conn_tuple();
let post_execute_rs = post_task_execute!(
execution_rs,
name,
task_id.as_json_string(),
task_group_arc.as_str(),
conn_tuple.2
);
info!("Save Task {:?} execution status complete", task_id);
assert!(post_execute_rs.is_ok());
let result = match execution_rs {
Ok(rs) => {
if is_verbose_task_output() {
println!(
" {} {:.1}s {}",
"✓".green(),
elapsed.as_secs_f32(),
user_friendly_task_summary(&task_id)
);
}
Err(err_msg) => {
let err_msg_str = err_msg.to_string();
error!("Task {:?} execution fail {:?}", task_id, err_msg_str);
if is_verbose_task_output() {
println!(" FAIL {} -- {}", task_id.format_string(), err_msg_str);
} else {
println!(" failed: {}", user_friendly_task_summary(&task_id));
}
TaskResultEnum::Error(err_msg_str)
TaskResultEnum::Success(rs)
}
Err(err_msg) => {
let err_msg_str = err_msg.to_string();
error!("Task {:?} execution fail {:?}", task_id, err_msg_str);
if is_verbose_task_output() {
eprintln!(
" {} {} -- {}",
"✗".red(),
user_friendly_task_summary(&task_id),
summarize_error(&err_msg_str)
);
}
};
let task_pair = TaskResultPair {
task_id: task_id.format_string(),
result,
};
let send_rs = tx_arc.send(task_pair.clone());
assert!(send_rs.is_ok());
task_pair
});
joins.push(join);
TaskResultEnum::Error(err_msg_str)
}
};
let task_pair = TaskResultPair {
task_id: task_id.format_string(),
result,
};
let send_rs = tx_arc.send(task_pair.clone());
assert!(send_rs.is_ok());
task_pair
});
joins.push(join);
});

let join_result = futures::future::join_all(joins).await;
let task_result = join_result
Expand Down Expand Up @@ -318,6 +332,11 @@ impl TaskController {
let split = TaskController::split_task(&task_execution_context);
let total_stages = split.len();
let mut task_result_vec = vec![];

if !is_verbose_task_output() {
println!();
}

for (stage_idx, task_split) in split.into_iter().enumerate() {
let stage_num = stage_idx + 1;
let task_names: Vec<String> = task_split
Expand All @@ -331,10 +350,12 @@ impl TaskController {
);
} else {
let stage_label = Self::stage_label_from_tasks(&task_group_string, task_split);
let progress = format!("[{stage_num}/{total_stages}]");
println!(
"[{stage_num}/{total_stages}] {} ({})",
stage_label,
Self::stage_summary(task_split.len())
"{} {} {}",
progress.cyan(),
stage_label.bold(),
format!("({})", Self::stage_summary(task_split.len())).dimmed()
);
}
let rs = self
Expand All @@ -344,14 +365,14 @@ impl TaskController {
for pair in rs.iter() {
if let TaskResultEnum::Error(err) = &pair.result {
if !is_verbose_task_output() {
println!(" reason: {}", summarize_error(err));
eprintln!(" {} {}", "Error:".red().bold(), summarize_error(err));
}
bail!("operation failed: {err}");
}
}
}
if !is_verbose_task_output() {
println!(" done");
println!(" {} {}", "✓".green(), "done".dimmed());
}
task_result_vec.push(rs);
}
Expand Down
Loading
Loading