Interactive Log Filtering for Container Log

README.md

@@ -78,7 +78,8 @@ lazy-ecs will automatically use the standard AWS credentials chain:
 ### Container-Level Features 🚀
 
 - ✅ **Container log viewing** - Display recent logs with timestamps from CloudWatch
-- ✅ **Container log live tail viewing** - Display logs live tail with timestamps from CloudWatch
+- ✅ **Container log live tail viewing** - Real-time log streaming with instant keyboard shortcuts
+- ✅ **Log filtering** - CloudWatch filter patterns (include/exclude) during live tail
 - ✅ **Basic container details** - Show container name, image, CPU/memory configuration
 - ✅ **Show environment variables & secrets** - Display environment variables and secrets configuration (without exposing secret values)
 - ✅ **Show port mappings** - Display container port configurations and networking
@@ -124,8 +125,8 @@ lazy-ecs will automatically use the standard AWS credentials chain:
 ### Advanced Features 🎯
 
 - ⬜ **Enhanced log features**:
-  - ⬜ Search/filter logs by keywords or time range
-  - ✅ Follow logs in real-time (tail -f style) - complex UI implementation
+  - ✅ Search/filter logs by keywords (CloudWatch patterns with include/exclude)
+  - ✅ Follow logs in real-time (tail -f style) with responsive keyboard shortcuts
   - ⬜ Download logs to file
 - ⬜ **Monitoring integration**:
   - ⬜ Show CloudWatch metrics for containers/tasks

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "lazy-ecs"
-version = "0.2.1"
+version = "0.3.0"
 description = "A CLI tool for working with AWS services"
 readme = "README.md"
 authors = [

src/lazy_ecs/core/utils.py

@@ -2,13 +2,38 @@
 
 from __future__ import annotations
 
+import atexit
+import select
+import sys
+import threading
 from collections.abc import Iterator
-from contextlib import contextmanager
+from contextlib import contextmanager, suppress
 from typing import TYPE_CHECKING, Literal
 
 from rich.console import Console
 from rich.spinner import Spinner
 
+# Try to import Unix-specific terminal control modules
+try:
+    import termios
+    import tty
+
+    HAS_TERMIOS = True
+    # Store original terminal settings globally
+    _original_terminal_settings = None
+    if sys.stdin.isatty():
+        _original_terminal_settings = termios.tcgetattr(sys.stdin.fileno())
+
+        # Register cleanup on exit
+        def restore_terminal() -> None:
+            if _original_terminal_settings and sys.stdin.isatty():
+                with suppress(Exception):
+                    termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, _original_terminal_settings)
+
+        atexit.register(restore_terminal)
+except ImportError:
+    HAS_TERMIOS = False
+
 if TYPE_CHECKING:
     from mypy_boto3_ecs.client import ECSClient
 
@@ -79,3 +104,46 @@ def paginate_aws_list(
         results.extend(page.get(result_key, []))
 
     return results
+
+
+def wait_for_keypress(stop_event: threading.Event) -> str | None:
+    """Wait for a single keypress in a non-blocking manner.
+
+    Returns the key pressed, or None if stop_event is set.
+    This runs in a separate thread to allow checking for keypresses without blocking.
+    """
+    if HAS_TERMIOS and sys.stdin.isatty():
+        # Unix/Linux/macOS with terminal support
+        fd = sys.stdin.fileno()
+        old_settings = termios.tcgetattr(fd)
+        try:
+            # Set terminal to cbreak mode for single-character input
+            tty.setcbreak(fd)
+
+            # Check for input with timeout
+            while not stop_event.is_set():
+                # Use select to check if input is available (0.01 second timeout for responsiveness)
+                if select.select([sys.stdin], [], [], 0.01)[0]:
+                    char = sys.stdin.read(1)
+                    # Handle Ctrl-C properly
+                    if char == "\x03":  # Ctrl-C
+                        raise KeyboardInterrupt()
+                    return char
+
+            return None
+        except (Exception, KeyboardInterrupt):
+            # Restore settings before re-raising
+            termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
+            if isinstance(sys.exc_info()[1], KeyboardInterrupt):
+                raise
+            return None
+        finally:
+            # Always restore terminal settings
+            with suppress(Exception):
+                termios.tcsetattr(fd, termios.TCSADRAIN, old_settings)
+    else:
+        # Fallback for Windows or when termios is not available
+        try:
+            return sys.stdin.read(1)
+        except Exception:
+            return None

src/lazy_ecs/features/container/container.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 from collections.abc import Generator
+from contextlib import suppress
 from os import environ
 from typing import TYPE_CHECKING, Any
 
@@ -15,6 +16,7 @@
     from mypy_boto3_ecs.type_defs import ContainerDefinitionOutputTypeDef, TaskDefinitionTypeDef
     from mypy_boto3_logs.client import CloudWatchLogsClient
     from mypy_boto3_logs.type_defs import (
+        FilteredLogEventTypeDef,
         LiveTailSessionLogEventTypeDef,
         OutputLogEventTypeDef,
         StartLiveTailResponseStreamTypeDef,
@@ -106,6 +108,20 @@ def get_container_logs(self, log_group: str, log_stream: str, lines: int = 50) -
         )
         return response.get("events", [])
 
+    def get_container_logs_filtered(
+        self, log_group: str, log_stream: str, filter_pattern: str, lines: int = 50
+    ) -> list[FilteredLogEventTypeDef]:
+        """Get container logs with CloudWatch filter pattern applied."""
+        if not self.logs_client:
+            return []
+        response = self.logs_client.filter_log_events(
+            logGroupName=log_group,
+            logStreamNames=[log_stream],
+            filterPattern=filter_pattern,
+            limit=lines,
+        )
+        return response.get("events", [])
+
     def get_live_container_logs_tail(
         self, log_group: str, log_stream: str, event_filter_pattern: str = ""
     ) -> Generator[StartLiveTailResponseStreamTypeDef | LiveTailSessionLogEventTypeDef]:
@@ -127,14 +143,20 @@ def get_live_container_logs_tail(
             logEventFilterPattern=event_filter_pattern,
         )
         response_stream = response.get("responseStream")
-        for event in response_stream:
-            if "sessionStart" in event:
-                continue
-            elif "sessionUpdate" in event:
-                log_events = event.get("sessionUpdate", {}).get("sessionResults", [])
-                yield from log_events
-            else:
-                yield event
+        try:
+            for event in response_stream:
+                if "sessionStart" in event:
+                    continue
+                elif "sessionUpdate" in event:
+                    log_events = event.get("sessionUpdate", {}).get("sessionResults", [])
+                    yield from log_events
+                else:
+                    yield event
+        finally:
+            # Properly close the response stream
+            if hasattr(response_stream, "close"):
+                with suppress(Exception):
+                    response_stream.close()
 
     def list_log_groups(self, cluster_name: str, container_name: str) -> list[str]:
         if not self.logs_client:

src/lazy_ecs/features/container/models.py

@@ -0,0 +1,45 @@
+"""Data models for container operations."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime
+from enum import Enum
+from typing import Any
+
+
+class Action(Enum):
+    """Actions for log tailing interaction."""
+
+    STOP = "s"
+    FILTER = "f"
+    CLEAR = "c"
+
+    @classmethod
+    def from_key(cls, key: str) -> Action | None:
+        """Convert keyboard key to action."""
+        for action in cls:
+            if action.value == key:
+                return action
+        return None
+
+
+@dataclass
+class LogEvent:
+    """Represents a log event from CloudWatch."""
+
+    timestamp: int | None
+    message: str
+    event_id: str | None = None
+
+    @property
+    def key(self) -> tuple[Any, ...] | str:
+        """Get unique key for deduplication."""
+        return self.event_id if self.event_id else (self.timestamp, self.message)
+
+    def format(self) -> str:
+        """Format the log event for display."""
+        if self.timestamp:
+            dt = datetime.fromtimestamp(self.timestamp / 1000)
+            return f"[{dt.strftime('%H:%M:%S')}] {self.message}"
+        return self.message