Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
295 changes: 293 additions & 2 deletions crates/cli/src/commands/admin/ilm.rs
Original file line number Diff line number Diff line change
@@ -1,8 +1,12 @@
//! ILM administration commands.

use clap::{Args, Subcommand};
use rc_core::admin::{AdminApi, ManualTransitionRunRequest, ManualTransitionRunResponse};
use rc_core::Error;
use rc_core::admin::{
AdminApi, ManualTransitionJobResponse, ManualTransitionRunRequest, ManualTransitionRunResponse,
};
use serde::Serialize;
use tokio::time::{Duration, Instant, sleep};

use super::{emit_observability_error, get_admin_client};
use crate::exit_code::ExitCode;
Expand All @@ -22,6 +26,15 @@ pub enum IlmCommands {
pub enum TransitionCommands {
/// Run bounded lifecycle transition evaluation for existing objects
Run(ManualTransitionRunArgs),

/// Show durable lifecycle transition job status
Status(ManualTransitionJobArgs),

/// Request cancellation for a durable lifecycle transition job
Cancel(ManualTransitionJobArgs),

/// Wait until a durable lifecycle transition job reaches a terminal state
Wait(ManualTransitionWaitArgs),
}

#[derive(Args, Debug)]
Expand All @@ -44,6 +57,10 @@ pub struct ManualTransitionRunArgs {
#[arg(long)]
pub dry_run: bool,

/// Start a durable background job and return its job endpoints
#[arg(long = "async")]
pub async_mode: bool,

/// Maximum number of object versions to scan
#[arg(long, default_value_t = 10_000, value_parser = clap::value_parser!(u64).range(1..=MAX_MANUAL_TRANSITION_OBJECTS))]
pub max_objects: u64,
Expand All @@ -53,6 +70,32 @@ pub struct ManualTransitionRunArgs {
pub max_duration_seconds: Option<u64>,
}

#[derive(Args, Debug)]
pub struct ManualTransitionJobArgs {
/// Alias name of the server
pub alias: String,

/// Durable manual transition job ID returned by `transition run --async`
pub job_id: String,
}

#[derive(Args, Debug)]
pub struct ManualTransitionWaitArgs {
/// Alias name of the server
pub alias: String,

/// Durable manual transition job ID returned by `transition run --async`
pub job_id: String,

/// Seconds between status polls
#[arg(long, default_value_t = 2, value_parser = clap::value_parser!(u64).range(1..=60))]
pub poll_interval_seconds: u64,

/// Maximum seconds to wait before returning an error
#[arg(long, value_parser = clap::value_parser!(u64).range(1..=86_400))]
pub timeout_seconds: Option<u64>,
}

#[derive(Debug, Serialize)]
struct ManualTransitionRunSuccessOutput<'a> {
schema_version: u8,
Expand All @@ -62,11 +105,29 @@ struct ManualTransitionRunSuccessOutput<'a> {
data: &'a ManualTransitionRunResponse,
}

#[derive(Debug, Serialize)]
struct ManualTransitionJobSuccessOutput<'a> {
schema_version: u8,
#[serde(rename = "type")]
output_type: &'static str,
status: &'static str,
data: &'a ManualTransitionJobResponse,
}

pub async fn execute(command: IlmCommands, formatter: &Formatter) -> ExitCode {
match command {
IlmCommands::Transition(TransitionCommands::Run(args)) => {
execute_manual_transition_run(args, formatter).await
}
IlmCommands::Transition(TransitionCommands::Status(args)) => {
execute_manual_transition_status(args, formatter).await
}
IlmCommands::Transition(TransitionCommands::Cancel(args)) => {
execute_manual_transition_cancel(args, formatter).await
}
IlmCommands::Transition(TransitionCommands::Wait(args)) => {
execute_manual_transition_wait(args, formatter).await
}
}
}

Expand All @@ -88,7 +149,13 @@ async fn execute_manual_transition_run(
max_duration_seconds: args.max_duration_seconds,
};

match client.run_manual_transition(request).await {
let result = if args.async_mode {
client.run_manual_transition_async(request).await
} else {
client.run_manual_transition(request).await
};

match result {
Ok(response) => {
if formatter.is_json() {
formatter.json(&ManualTransitionRunSuccessOutput {
Expand All @@ -112,6 +179,101 @@ async fn execute_manual_transition_run(
}
}

async fn execute_manual_transition_status(
args: ManualTransitionJobArgs,
formatter: &Formatter,
) -> ExitCode {
let client = match get_admin_client(&args.alias, formatter) {
Ok(client) => client,
Err(code) => return code,
};

match client.manual_transition_job_status(&args.job_id).await {
Ok(response) => {
print_manual_transition_job("manual_transition_job_status", &response, formatter);
ExitCode::Success
}
Err(error) => emit_observability_error(
"manual_transition_job_status",
"admin.ilm-transition-job",
"Failed to get manual transition job status",
&error,
formatter,
),
}
}

async fn execute_manual_transition_cancel(
args: ManualTransitionJobArgs,
formatter: &Formatter,
) -> ExitCode {
let client = match get_admin_client(&args.alias, formatter) {
Ok(client) => client,
Err(code) => return code,
};

match client.cancel_manual_transition_job(&args.job_id).await {
Ok(response) => {
print_manual_transition_job("manual_transition_job_cancel", &response, formatter);
ExitCode::Success
}
Err(error) => emit_observability_error(
"manual_transition_job_cancel",
"admin.ilm-transition-job",
"Failed to cancel manual transition job",
&error,
formatter,
),
}
}

async fn execute_manual_transition_wait(
args: ManualTransitionWaitArgs,
formatter: &Formatter,
) -> ExitCode {
let client = match get_admin_client(&args.alias, formatter) {
Ok(client) => client,
Err(code) => return code,
};

let deadline = args
.timeout_seconds
.map(|seconds| Instant::now() + Duration::from_secs(seconds));
let poll_interval = Duration::from_secs(args.poll_interval_seconds);

loop {
match client.manual_transition_job_status(&args.job_id).await {
Ok(response) if is_terminal_job_status(&response.status) => {
print_manual_transition_job("manual_transition_job_wait", &response, formatter);
return wait_terminal_exit_code(&response.status);
}
Ok(_) if deadline.is_some_and(|deadline| Instant::now() >= deadline) => {
let error = Error::General(format!(
"Timed out waiting for manual transition job '{}' to finish",
args.job_id
));
return emit_observability_error(
"manual_transition_job_wait",
"admin.ilm-transition-job",
"Manual transition job wait timed out",
&error,
formatter,
);
}
Ok(_) => sleep(poll_interval).await,
Err(error) => {
return emit_observability_error(
"manual_transition_job_wait",
"admin.ilm-transition-job",
"Failed while waiting for manual transition job",
&error,
formatter,
);
}
}
}
}

fn print_manual_transition_run(response: &ManualTransitionRunResponse, formatter: &Formatter) {
let report = &response.report;
formatter.println(&formatter.style_name("Manual Transition Run"));
Expand All @@ -124,6 +286,24 @@ fn print_manual_transition_run(response: &ManualTransitionRunResponse, formatter
"Mode: {}",
formatter.sanitize_text(&response.mode)
));
if let Some(job_id) = &response.job_id {
formatter.println(&format!(
"Job ID: {}",
formatter.sanitize_text(job_id)
));
}
if let Some(status_endpoint) = &response.status_endpoint {
formatter.println(&format!(
"Status: {}",
formatter.sanitize_text(status_endpoint)
));
}
if let Some(cancel_endpoint) = &response.cancel_endpoint {
formatter.println(&format!(
"Cancel: {}",
formatter.sanitize_text(cancel_endpoint)
));
}
formatter.println(&format!(
"Bucket: {}",
formatter.sanitize_text(&report.bucket)
Expand Down Expand Up @@ -159,8 +339,119 @@ fn print_manual_transition_run(response: &ManualTransitionRunResponse, formatter
));
formatter.println(&format!("Limit reached: {}", report.truncated_by_limit));
formatter.println(&format!("Duration hit: {}", report.truncated_by_duration));
if let Some(continuation_token) = &report.continuation_token {
formatter.println(&format!(
"Continuation: {}",
formatter.sanitize_text(continuation_token)
));
}
}

fn value_or_all(value: &str) -> &str {
if value.is_empty() { "all" } else { value }
}

fn print_manual_transition_job(
output_type: &'static str,
response: &ManualTransitionJobResponse,
formatter: &Formatter,
) {
if formatter.is_json() {
formatter.json(&ManualTransitionJobSuccessOutput {
schema_version: 3,
output_type,
status: "success",
data: response,
});
return;
}

let report = &response.report;
let queue = &response.queue_snapshot;
formatter.println(&formatter.style_name("Manual Transition Job"));
formatter.println("");
formatter.println(&format!(
"Status: {}",
formatter.sanitize_text(&response.status)
));
formatter.println(&format!(
"Mode: {}",
formatter.sanitize_text(&response.mode)
));
formatter.println(&format!(
"Job ID: {}",
formatter.sanitize_text(&response.job_id)
));
formatter.println(&format!(
"Bucket: {}",
formatter.sanitize_text(&response.bucket)
));
formatter.println(&format!(
"Prefix: {}",
formatter.sanitize_text(value_or_all(&response.prefix))
));
formatter.println(&format!(
"Tier: {}",
formatter.sanitize_text(response.tier.as_deref().map(value_or_all).unwrap_or("all"))
));
formatter.println(&format!("Dry run: {}", response.dry_run));
formatter.println(&format!("Cancel asked: {}", response.cancel_requested));
formatter.println(&format!(
"Created ns: {}",
response.created_at_unix_nanos
));
formatter.println(&format!(
"Updated ns: {}",
response.updated_at_unix_nanos
));
if let Some(completed_at) = response.completed_at_unix_nanos {
formatter.println(&format!("Completed ns: {completed_at}"));
}
if let Some(reason) = &response.failure_reason {
formatter.println(&format!(
"Failure: {}",
formatter.sanitize_text(reason)
));
}
formatter.println("");
formatter.println(&formatter.style_name("Report"));
formatter.println(&format!("Scanned: {}", report.scanned));
formatter.println(&format!("Eligible: {}", report.eligible));
formatter.println(&format!("Enqueued: {}", report.enqueued));
formatter.println(&format!("Completed: {}", report.transition_completed));
formatter.println(&format!("Failed: {}", report.transition_failed));
formatter.println(&format!(
"Already moved: {}",
report.skipped_already_transitioned
));
formatter.println(&format!("Cancelled: {}", report.cancelled));
formatter.println(&format!("Limit reached: {}", report.truncated_by_limit));
formatter.println(&format!("Duration hit: {}", report.truncated_by_duration));
if let Some(continuation_token) = &report.continuation_token {
formatter.println(&format!(
"Continuation: {}",
formatter.sanitize_text(continuation_token)
));
}
formatter.println("");
formatter.println(&formatter.style_name("Queue"));
formatter.println(&format!("Queued: {}", queue.queued));
formatter.println(&format!("Active: {}", queue.active));
formatter.println(&format!("Workers: {}", queue.workers));
formatter.println(&format!("Capacity: {}", queue.queue_capacity));
}

fn is_terminal_job_status(status: &str) -> bool {
matches!(
status,
"completed" | "partial" | "failed" | "cancelled" | "unknown"
)
}

fn wait_terminal_exit_code(status: &str) -> ExitCode {
if matches!(status, "completed" | "partial") {
ExitCode::Success
} else {
ExitCode::GeneralError
}
}
Loading