feat(logs): support log streaming

Signed-off-by: Vjeran Grozdanic <vjeran.grozdanic@sentry.io>

Author: vgrozdanic

src/api/mod.rs

@@ -2497,7 +2497,7 @@ struct LogsResponse {
 }
 
 /// Log entry structure from the logs API
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Deserialize, Clone)]
 pub struct LogEntry {
     #[serde(rename = "sentry.item_id")]
     pub item_id: String,

src/commands/logs/list.rs

@@ -1,7 +1,9 @@
 use anyhow::Result;
 use clap::Args;
+use std::collections::HashSet;
+use std::time::Duration;
 
-use crate::api::{Api, Dataset, FetchEventsOptions};
+use crate::api::{Api, Dataset, FetchEventsOptions, LogEntry};
 use crate::config::Config;
 use crate::utils::formatting::Table;
 
@@ -14,6 +16,9 @@ const LOG_FIELDS: &[&str] = &[
     "message",
 ];
 
+/// Maximum number of log entries to keep in memory for deduplication
+const MAX_DEDUP_BUFFER_SIZE: usize = 10_000;
+
 /// Arguments for listing logs
 #[derive(Args)]
 pub(super) struct ListLogsArgs {
@@ -26,12 +31,20 @@ pub(super) struct ListLogsArgs {
     project: Option<String>,
 
     #[arg(long = "max-rows", default_value = "100")]
-    #[arg(help = "Maximum number of log entries to fetch and display (max 1000).")]
+    #[arg(help = "Maximum number of log entries to fetch and display per request (max 1000).")]
     max_rows: usize,
 
     #[arg(long = "query", default_value = "")]
     #[arg(help = "Query to filter logs. Example: \"level:error\"")]
     query: String,
+
+    #[arg(long = "live")]
+    #[arg(help = "Enable live streaming mode to continuously poll for new logs.")]
+    live: bool,
+
+    #[arg(long = "poll-interval", default_value = "2")]
+    #[arg(help = "Polling interval in seconds for live streaming mode (default: 2).")]
+    poll_interval: u64,
 }
 
 pub(super) fn execute(args: ListLogsArgs) -> Result<()> {
@@ -63,7 +76,11 @@ pub(super) fn execute(args: ListLogsArgs) -> Result<()> {
         Some(args.query.as_str())
     };
 
-    execute_single_fetch(&api, &org, &project, query, LOG_FIELDS, &args)
+    if args.live {
+        execute_live_streaming(&api, &org, &project, query, LOG_FIELDS, &args)
+    } else {
+        execute_single_fetch(&api, &org, &project, query, LOG_FIELDS, &args)
+    }
 }
 
 fn execute_single_fetch(
@@ -117,3 +134,252 @@ fn execute_single_fetch(
 
     Ok(())
 }
+
+/// Manages deduplication of log entries with a bounded buffer
+struct LogDeduplicator {
+    /// Set of seen log IDs for quick lookup
+    seen_ids: HashSet<String>,
+    /// Buffer of log entries in order (for maintaining size limit)
+    buffer: Vec<LogEntry>,
+    /// Maximum size of the buffer
+    max_size: usize,
+}
+
+impl LogDeduplicator {
+    fn new(max_size: usize) -> Self {
+        Self {
+            seen_ids: HashSet::new(),
+            buffer: Vec::new(),
+            max_size,
+        }
+    }
+
+    /// Add new logs and return only the ones that haven't been seen before
+    fn add_logs(&mut self, new_logs: Vec<LogEntry>) -> Vec<LogEntry> {
+        let mut unique_logs = Vec::new();
+
+        for log in new_logs {
+            if !self.seen_ids.contains(&log.item_id) {
+                self.seen_ids.insert(log.item_id.clone());
+                self.buffer.push(log.clone());
+                unique_logs.push(log);
+            }
+        }
+
+        // Maintain buffer size limit by removing oldest entries
+        while self.buffer.len() > self.max_size {
+            let removed_log = self.buffer.remove(0);
+            self.seen_ids.remove(&removed_log.item_id);
+        }
+
+        unique_logs
+    }
+
+    /// Get the number of unique logs seen so far
+    fn total_seen(&self) -> usize {
+        self.seen_ids.len()
+    }
+}
+
+fn execute_live_streaming(
+    api: &Api,
+    org: &str,
+    project: &str,
+    query: Option<&str>,
+    fields: &[&str],
+    args: &ListLogsArgs,
+) -> Result<()> {
+    let mut deduplicator = LogDeduplicator::new(MAX_DEDUP_BUFFER_SIZE);
+    let poll_duration = Duration::from_secs(args.poll_interval);
+    let mut consecutive_new_only_count = 0;
+    const WARNING_THRESHOLD: usize = 3; // Show warning after 3 consecutive new-only responses
+
+    println!("Starting live log streaming...");
+    println!(
+        "Polling every {} seconds. Press Ctrl+C to stop.",
+        args.poll_interval
+    );
+
+    // Print header once
+    let mut table = Table::new();
+    table
+        .title_row()
+        .add("Item ID")
+        .add("Timestamp")
+        .add("Severity")
+        .add("Message")
+        .add("Trace");
+
+    loop {
+        let options = FetchEventsOptions {
+            dataset: Dataset::OurLogs,
+            fields,
+            project_id: Some(project),
+            cursor: None,
+            query,
+            per_page: Some(args.max_rows),
+            stats_period: Some("1h"),
+            sort: Some("-timestamp"),
+        };
+
+        match api
+            .authenticated()?
+            .fetch_organization_events(org, &options)
+        {
+            Ok(logs) => {
+                let unique_logs = deduplicator.add_logs(logs);
+
+                if unique_logs.is_empty() {
+                    consecutive_new_only_count += 1;
+
+                    if consecutive_new_only_count >= WARNING_THRESHOLD && args.query.is_empty() {
+                        eprintln!(
+                            "\n⚠️  Warning: No new logs found for {consecutive_new_only_count} consecutive polls."
+                        );
+                        eprintln!("   Consider using --query to filter logs, as you may be missing some entries.");
+                        eprintln!(
+                            "   Example: --query \"level:error\" or --query \"message:*error*\""
+                        );
+
+                        // Reset counter to avoid spam
+                        consecutive_new_only_count = 0;
+                    }
+                } else {
+                    consecutive_new_only_count = 0;
+
+                    // Print new logs
+                    for log in unique_logs {
+                        let row = table.add_row();
+                        row.add(&log.item_id)
+                            .add(&log.timestamp)
+                            .add(log.severity.as_deref().unwrap_or(""))
+                            .add(log.message.as_deref().unwrap_or(""))
+                            .add(log.trace.as_deref().unwrap_or(""));
+                    }
+
+                    table.print();
+                    println!("Total unique logs seen: {}", deduplicator.total_seen());
+                }
+            }
+            Err(e) => {
+                eprintln!("Error fetching logs: {e}");
+            }
+        }
+
+        std::thread::sleep(poll_duration);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn create_test_log(id: &str, message: &str) -> LogEntry {
+        LogEntry {
+            item_id: id.to_owned(),
+            trace: None,
+            severity: Some("info".to_owned()),
+            timestamp: "2025-01-01T00:00:00Z".to_owned(),
+            message: Some(message.to_owned()),
+        }
+    }
+
+    #[test]
+    fn test_log_deduplicator_new() {
+        let deduplicator = LogDeduplicator::new(100);
+        assert_eq!(deduplicator.total_seen(), 0);
+    }
+
+    #[test]
+    fn test_log_deduplicator_add_unique_logs() {
+        let mut deduplicator = LogDeduplicator::new(10);
+
+        let log1 = create_test_log("1", "test message 1");
+        let log2 = create_test_log("2", "test message 2");
+
+        let unique_logs = deduplicator.add_logs(vec![log1.clone(), log2.clone()]);
+
+        assert_eq!(unique_logs.len(), 2);
+        assert_eq!(deduplicator.total_seen(), 2);
+    }
+
+    #[test]
+    fn test_log_deduplicator_deduplicate_logs() {
+        let mut deduplicator = LogDeduplicator::new(10);
+
+        let log1 = create_test_log("1", "test message 1");
+        let log2 = create_test_log("2", "test message 2");
+
+        // Add logs first time
+        let unique_logs1 = deduplicator.add_logs(vec![log1.clone(), log2.clone()]);
+        assert_eq!(unique_logs1.len(), 2);
+
+        // Add same logs again
+        let unique_logs2 = deduplicator.add_logs(vec![log1.clone(), log2.clone()]);
+        assert_eq!(unique_logs2.len(), 0); // Should be empty as logs already seen
+
+        assert_eq!(deduplicator.total_seen(), 2);
+    }
+
+    #[test]
+    fn test_log_deduplicator_buffer_size_limit() {
+        let mut deduplicator = LogDeduplicator::new(3);
+
+        // Add 5 logs to a buffer with max size 3
+        let logs = vec![
+            create_test_log("1", "test message 1"),
+            create_test_log("2", "test message 2"),
+            create_test_log("3", "test message 3"),
+            create_test_log("4", "test message 4"),
+            create_test_log("5", "test message 5"),
+        ];
+
+        let unique_logs = deduplicator.add_logs(logs);
+        assert_eq!(unique_logs.len(), 5);
+
+        // After adding 5 logs to a buffer with max size 3, the oldest 2 should be evicted
+        // So logs 1 and 2 should no longer be in the seen_ids set
+        // Adding them again should return them as new logs
+        let duplicate_logs = vec![
+            create_test_log("1", "test message 1"),
+            create_test_log("2", "test message 2"),
+        ];
+        let duplicate_unique_logs = deduplicator.add_logs(duplicate_logs);
+        assert_eq!(duplicate_unique_logs.len(), 2);
+
+        // Test that adding new logs still works
+        let new_logs = vec![create_test_log("6", "test message 6")];
+        let new_unique_logs = deduplicator.add_logs(new_logs);
+        assert_eq!(new_unique_logs.len(), 1);
+    }
+
+    #[test]
+    fn test_log_deduplicator_mixed_new_and_old_logs() {
+        let mut deduplicator = LogDeduplicator::new(10);
+
+        // Add initial logs
+        let initial_logs = vec![
+            create_test_log("1", "test message 1"),
+            create_test_log("2", "test message 2"),
+        ];
+        let unique_logs1 = deduplicator.add_logs(initial_logs);
+        assert_eq!(unique_logs1.len(), 2);
+
+        // Add mix of new and old logs
+        let mixed_logs = vec![
+            create_test_log("1", "test message 1"), // old
+            create_test_log("3", "test message 3"), // new
+            create_test_log("2", "test message 2"), // old
+            create_test_log("4", "test message 4"), // new
+        ];
+        let unique_logs2 = deduplicator.add_logs(mixed_logs);
+
+        // Should only return the new logs (3 and 4)
+        assert_eq!(unique_logs2.len(), 2);
+        assert_eq!(unique_logs2[0].item_id, "3");
+        assert_eq!(unique_logs2[1].item_id, "4");
+
+        assert_eq!(deduplicator.total_seen(), 4);
+        assert_eq!(deduplicator.buffer.len(), 4);
+    }
+}

tests/integration/_cases/logs/logs-list-help.trycmd

@@ -6,11 +6,17 @@ Usage: sentry-cli[EXE] logs list [OPTIONS]
 
 Options:
       --max-rows <MAX_ROWS>
-          Maximum number of log entries to fetch and display (max 1000). [default: 100]
+          Maximum number of log entries to fetch and display per request (max 1000). [default: 100]
 
       --query <QUERY>
           Query to filter logs. Example: "level:error" [default: ]
 
+      --live
+          Enable live streaming mode to continuously poll for new logs.
+
+      --poll-interval <POLL_INTERVAL>
+          Polling interval in seconds for live streaming mode (default: 2). [default: 2]
+
   -o, --org <ORG>
           The organization ID or slug.