ironflow_cli/commands/
logs.rs

1//! Log retrieval and streaming via SSE.
2
3use std::io::Write as _;
4
5use anyhow::Result;
6use clap::Args;
7use futures_util::StreamExt;
8use ironflow_sdk::IronflowClient;
9use ironflow_sdk::client::ListRunLogsFilter;
10use uuid::Uuid;
11
12/// Arguments for the `logs` command.
13#[derive(Debug, Args)]
14pub struct LogsArgs {
15    /// Run UUID to retrieve logs for.
16    pub run_id: Uuid,
17    /// Keep streaming until the run reaches a terminal state.
18    #[arg(long)]
19    pub follow: bool,
20    /// Filter by step ID.
21    #[arg(long)]
22    pub step_id: Option<Uuid>,
23    /// Filter by output stream (`stdout`, `stderr`, `system`).
24    #[arg(long)]
25    pub stream: Option<String>,
26    /// Maximum number of entries to return per page (only without --follow).
27    #[arg(long)]
28    pub limit: Option<u32>,
29}
30
31/// Terminal event types that signal the run is done.
32const TERMINAL_EVENTS: &[&str] = &["run_completed", "run_failed", "run_cancelled"];
33
34/// Execute the `logs` command.
35///
36/// Without `--follow`, retrieves persisted logs via `GET /api/v1/runs/:id/logs`.
37/// With `--follow`, streams live logs via SSE until the run reaches a terminal state.
38///
39/// # Errors
40///
41/// Returns an error on API or SSE connection failure.
42pub async fn execute(client: &IronflowClient, args: &LogsArgs, json_mode: bool) -> Result<()> {
43    if args.follow {
44        return execute_follow(client, args, json_mode).await;
45    }
46
47    let mut out = std::io::stdout().lock();
48    let mut cursor: Option<Uuid> = None;
49
50    loop {
51        let filter = ListRunLogsFilter {
52            step_id: args.step_id,
53            stream: args.stream.as_deref(),
54            cursor,
55            limit: args.limit,
56        };
57
58        let response = client.get_run_logs(args.run_id, &filter).await?;
59
60        for entry in &response.data {
61            if json_mode {
62                writeln!(out, "{}", serde_json::to_string(&entry)?)?;
63            } else {
64                writeln!(
65                    out,
66                    "[{}] [{}] {}",
67                    entry.step_name, entry.stream, entry.line
68                )?;
69            }
70        }
71
72        let has_more = response
73            .meta
74            .as_ref()
75            .and_then(|m| m.extra.get("has_more"))
76            .and_then(|v| v.as_bool())
77            .unwrap_or(false);
78
79        if !has_more {
80            break;
81        }
82
83        cursor = response
84            .meta
85            .as_ref()
86            .and_then(|m| m.extra.get("next_cursor"))
87            .and_then(|v| v.as_str())
88            .and_then(|s| s.parse().ok());
89    }
90
91    Ok(())
92}
93
94/// Stream logs via SSE (--follow mode).
95async fn execute_follow(client: &IronflowClient, args: &LogsArgs, json_mode: bool) -> Result<()> {
96    let mut stream = client.events(Some(args.run_id), None).await?;
97    let mut out = std::io::stdout().lock();
98
99    while let Some(event) = stream.next().await {
100        match event {
101            Ok(ev) => {
102                if json_mode {
103                    let obj = serde_json::json!({
104                        "event": ev.event_type,
105                        "data": ev.data,
106                    });
107                    writeln!(out, "{}", serde_json::to_string(&obj)?)?;
108                } else {
109                    writeln!(out, "[{}] {}", ev.event_type, ev.data)?;
110                }
111
112                if TERMINAL_EVENTS.contains(&ev.event_type.as_str()) {
113                    break;
114                }
115            }
116            Err(e) => {
117                return Err(anyhow::anyhow!("SSE stream error: {e}"));
118            }
119        }
120    }
121
122    Ok(())
123}
ironflow_cli/commands/logs.rs

ironflow_cli/commands/
logs.rs
