feat: Add streaming logs support with --follow flag (Issue #70) (#404)

* refactor: extract profile commands from main.rs to dedicated module

- Create commands/profile.rs with all profile command handlers
- Move 660 lines of profile logic out of main.rs
- Reduce main.rs from 1,589 lines to 929 lines (41.5% reduction)
- Split into focused handlers: list, path, show, set, remove, validate, default-enterprise, default-cloud
- Update commands/mod.rs to include profile module
- Remove unused Context import from main.rs
- All 45 tests passing
- No clippy warnings

Main.rs now focuses on CLI parsing, initialization, and command routing.

Closes #402

* feat: add streaming logs support with --follow flag

Implements real-time log streaming for Redis Enterprise using polling-based approach.

## Changes

### redis-enterprise crate
- Add tokio-stream, futures, async-stream dependencies
- Implement stream_logs() method in LogsHandler
- Uses polling with configurable interval to simulate streaming
- Tracks last timestamp to avoid duplicate entries

### redisctl CLI
- Add --follow/-f flag to 'logs list' command
- Add --poll-interval flag (default: 2 seconds)
- Implement handle_stream_logs for real-time output
- Support Ctrl+C graceful shutdown
- Format logs as they arrive in table format
- Support JMESPath filtering on streamed logs

## Usage

```bash
# Stream logs in real-time (like tail -f)
redisctl enterprise logs list --follow

# Custom poll interval
redisctl enterprise logs list --follow --poll-interval 5

# With filtering
redisctl enterprise logs list --follow -q 'type'
```

## Implementation Details

Since Redis Enterprise API doesn't natively support streaming, this:
- Polls the /v1/logs endpoint at regular intervals
- Tracks the last timestamp seen to fetch only new entries
- Uses async-stream for ergonomic async generator syntax
- Handles Ctrl+C for graceful shutdown

## Testing
- All 67 tests passing (5 cloud, 17 enterprise, 45 redisctl)
- No clippy warnings
- Formatted with cargo fmt

Addresses #70

Author: joshrotenberg

Cargo.lock

@@ -52,6 +52,8 @@ allow-dirty = ["ci"]
 anyhow = "1.0"
 thiserror = "2.0"
 tokio = { version = "1.40", features = ["full"] }
+tokio-stream = "0.1"
+futures = "0.3"
 serde = { version = "1.0", features = ["derive"] }
 tar = "0.4"
 flate2 = "1.0"

Cargo.toml

@@ -16,13 +16,16 @@ readme = "../../README.md"
 
 [dependencies]
 async-trait = { workspace = true }
+async-stream = "0.3"
 reqwest = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 serde_path_to_error = "0.1"
 serde_urlencoded = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
+tokio-stream = { workspace = true }
+futures = { workspace = true }
 tracing = { workspace = true }
 anyhow = { workspace = true }
 base64 = { workspace = true }

crates/redis-enterprise/Cargo.toml

@@ -4,11 +4,16 @@
 //! - Query cluster logs
 //! - Configure log levels
 //! - Export log data
+//! - Stream logs in real-time (via polling)
 
 use crate::client::RestClient;
 use crate::error::Result;
+use futures::stream::Stream;
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
+use std::pin::Pin;
+use std::time::Duration;
+use tokio::time::sleep;
 
 /// Log entry (cluster event)
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -67,4 +72,60 @@ impl LogsHandler {
             self.client.get("/v1/logs").await
         }
     }
+
+    /// Stream logs in real-time by polling
+    ///
+    /// Since Redis Enterprise API doesn't support native streaming, this polls
+    /// the logs endpoint at regular intervals and yields new log entries.
+    ///
+    /// # Arguments
+    /// * `poll_interval` - Time to wait between polls (default: 2 seconds)
+    /// * `limit` - Maximum number of logs to fetch per poll (default: 100)
+    ///
+    /// # Returns
+    /// A stream of log entries that can be consumed with `while let Some(entry) = stream.next().await`
+    pub fn stream_logs(
+        &self,
+        poll_interval: Duration,
+        limit: Option<u32>,
+    ) -> Pin<Box<dyn Stream<Item = Result<LogEntry>> + Send + '_>> {
+        Box::pin(async_stream::stream! {
+            let mut last_time: Option<String> = None;
+
+            loop {
+                // Build query - get logs after the last timestamp we saw
+                let query = LogsQuery {
+                    stime: last_time.clone(),
+                    etime: None,
+                    order: Some("asc".to_string()), // Ascending so we get chronological order
+                    limit,
+                    offset: None,
+                };
+
+                // Fetch logs
+                match self.list(Some(query)).await {
+                    Ok(entries) => {
+                        if !entries.is_empty() {
+                            // Update last_time to the timestamp of the last entry
+                            if let Some(last_entry) = entries.last() {
+                                last_time = Some(last_entry.time.clone());
+                            }
+
+                            // Yield each entry
+                            for entry in entries {
+                                yield Ok(entry);
+                            }
+                        }
+                    }
+                    Err(e) => {
+                        yield Err(e);
+                        break;
+                    }
+                }
+
+                // Wait before next poll
+                sleep(poll_interval).await;
+            }
+        })
+    }
 }

crates/redis-enterprise/src/logs.rs

@@ -27,6 +27,8 @@ clap = { workspace = true }
 clap_complete = "4.5"
 anyhow = { workspace = true }
 tokio = { workspace = true }
+tokio-stream = { workspace = true }
+futures = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
 serde = { workspace = true }

crates/redisctl/Cargo.toml

@@ -4,6 +4,22 @@ use clap::Subcommand;
 pub enum LogsCommands {
     /// List cluster event logs
     #[command(visible_alias = "ls")]
+    #[command(after_help = "EXAMPLES:
+    # Get recent logs
+    redisctl enterprise logs list
+
+    # Get logs since a specific time
+    redisctl enterprise logs list --since 2024-01-01T00:00:00Z
+
+    # Stream logs in real-time (like tail -f)
+    redisctl enterprise logs list --follow
+
+    # Stream logs with custom poll interval
+    redisctl enterprise logs list --follow --poll-interval 5
+
+    # Limit number of logs per fetch
+    redisctl enterprise logs list --limit 50
+")]
     List {
         /// Start time (ISO 8601 format)
         #[arg(long)]
@@ -24,5 +40,13 @@ pub enum LogsCommands {
         /// Number of events to skip (for pagination)
         #[arg(long)]
         offset: Option<u32>,
+
+        /// Follow log output (stream in real-time like tail -f)
+        #[arg(long, short = 'f')]
+        follow: bool,
+
+        /// Poll interval in seconds when following logs
+        #[arg(long, default_value = "2", requires = "follow")]
+        poll_interval: u64,
     },
 }

crates/redisctl/src/commands/enterprise/logs.rs

@@ -7,7 +7,10 @@ use crate::cli::OutputFormat;
 use crate::commands::enterprise::logs::LogsCommands;
 use crate::connection::ConnectionManager;
 use crate::error::Result as CliResult;
+use futures::StreamExt;
 use redis_enterprise::logs::LogsQuery;
+use std::time::Duration;
+use tokio::signal;
 
 /// Parameters for log list operation
 struct LogListParams {
@@ -16,6 +19,8 @@ struct LogListParams {
     order: Option<String>,
     limit: Option<u32>,
     offset: Option<u32>,
+    follow: bool,
+    poll_interval: u64,
 }
 
 pub async fn handle_logs_commands(
@@ -32,13 +37,17 @@ pub async fn handle_logs_commands(
             order,
             limit,
             offset,
+            follow,
+            poll_interval,
         } => {
             let params = LogListParams {
                 since: since.clone(),
                 until: until.clone(),
                 order: order.clone(),
                 limit: *limit,
                 offset: *offset,
+                follow: *follow,
+                poll_interval: *poll_interval,
             };
             handle_list_logs(conn_mgr, profile_name, params, output_format, query).await
         }
@@ -55,6 +64,12 @@ async fn handle_list_logs(
     let client = conn_mgr.create_enterprise_client(profile_name).await?;
     let handler = redis_enterprise::LogsHandler::new(client);
 
+    // Handle streaming mode
+    if params.follow {
+        return handle_stream_logs(handler, params, output_format, query).await;
+    }
+
+    // Normal (non-streaming) mode
     let logs_query = if params.since.is_some()
         || params.until.is_some()
         || params.order.is_some()
@@ -92,3 +107,79 @@ async fn handle_list_logs(
 
     Ok(())
 }
+
+async fn handle_stream_logs(
+    handler: redis_enterprise::LogsHandler,
+    params: LogListParams,
+    output_format: OutputFormat,
+    query: Option<&str>,
+) -> CliResult<()> {
+    // Only support table/auto format for streaming (JSON/YAML don't make sense for streams)
+    if !matches!(output_format, OutputFormat::Auto | OutputFormat::Table) {
+        return Err(RedisCtlError::InvalidInput {
+            message: "Streaming logs (--follow) only supports table output format".to_string(),
+        });
+    }
+
+    let poll_interval = Duration::from_secs(params.poll_interval);
+    let mut stream = handler.stream_logs(poll_interval, params.limit);
+
+    println!("Streaming logs (Ctrl+C to stop)...\n");
+
+    loop {
+        tokio::select! {
+            // Handle Ctrl+C
+            _ = signal::ctrl_c() => {
+                println!("\nStopping log stream...");
+                break;
+            }
+            // Handle next log entry
+            entry_result = stream.next() => {
+                match entry_result {
+                    Some(Ok(entry)) => {
+                        // Format and print the log entry
+                        let entry_json = serde_json::to_value(&entry)?;
+
+                        // Apply JMESPath query if provided
+                        let output_data = if let Some(q) = query {
+                            crate::commands::enterprise::utils::apply_jmespath(&entry_json, q)?
+                        } else {
+                            entry_json
+                        };
+
+                        // Print each entry as it arrives
+                        // For table format, print a simple formatted line
+                        if let Some(time) = output_data.get("time").and_then(|t| t.as_str()) {
+                            let event_type = output_data.get("type")
+                                .and_then(|t| t.as_str())
+                                .unwrap_or("unknown");
+
+                            // Print timestamp and event type
+                            print!("[{}] {}", time, event_type);
+
+                            // Print any extra fields
+                            if let Some(obj) = output_data.as_object() {
+                                for (key, value) in obj {
+                                    if key != "time" && key != "type" {
+                                        print!(" {}={}", key, value);
+                                    }
+                                }
+                            }
+                            println!();
+                        }
+                    }
+                    Some(Err(e)) => {
+                        eprintln!("Error fetching logs: {}", e);
+                        break;
+                    }
+                    None => {
+                        // Stream ended
+                        break;
+                    }
+                }
+            }
+        }
+    }
+
+    Ok(())
+}