Add CloudWatch metrics display for ECS services

README.md

@@ -177,8 +177,8 @@ lazy-ecs will automatically use the standard AWS credentials chain:
   - ✅ Follow logs in real-time (tail -f style) with responsive keyboard shortcuts
   - ⬜ Download logs to file
 - ⬜ **Monitoring integration**:
-  - ⬜ Show CloudWatch metrics (CPU/Memory utilization) for services and tasks
-  - ⬜ Display resource usage trends to identify spikes, leaks, and throttling
+  - ✅ Show CloudWatch metrics (CPU/Memory utilization) - Display current values, averages, and peaks
+  - ⬜ Add sparkline visualization - Inline Unicode trend indicators for quick visual assessment
 - ⬜ **Port forwarding to container** - Direct local connection to container ports for debugging
 - ⬜ **Multi-region support** - Work with ECS across different AWS regions
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "lazy-ecs"
-version = "0.3.2"
+version = "0.4.0"
 description = "A CLI tool for working with AWS services"
 readme = "README.md"
 authors = [
@@ -47,7 +47,7 @@ dev-dependencies = [
   "ruff==0.13.0",
   "pre-commit==4.3.0",
   "pyrefly==0.32.0",
-  "boto3-stubs[ecs,logs,sts]==1.40.30",
+  "boto3-stubs[ecs,logs,sts,cloudwatch]==1.40.30",
 ]
 
 [tool.ruff]

src/lazy_ecs/__init__.py

@@ -6,6 +6,7 @@
 from rich.console import Console
 
 if TYPE_CHECKING:
+    from mypy_boto3_cloudwatch.client import CloudWatchClient
     from mypy_boto3_ecs import ECSClient
     from mypy_boto3_logs.client import CloudWatchLogsClient
     from mypy_boto3_sts.client import STSClient
@@ -32,7 +33,8 @@ def main() -> None:
         ecs_client = _create_aws_client(args.profile)
         logs_client = _create_logs_client(args.profile)
         sts_client = _create_sts_client(args.profile)
-        ecs_service = ECSService(ecs_client, sts_client, logs_client)
+        cloudwatch_client = _create_cloudwatch_client(args.profile)
+        ecs_service = ECSService(ecs_client, sts_client, logs_client, cloudwatch_client)
         navigator = ECSNavigator(ecs_service)
 
         _navigate_clusters(navigator, ecs_service)
@@ -85,6 +87,19 @@ def _create_sts_client(profile_name: str | None) -> "STSClient":
     return boto3.client("sts", config=config)
 
 
+def _create_cloudwatch_client(profile_name: str | None) -> "CloudWatchClient":
+    """Create optimized CloudWatch client with connection pooling."""
+    config = Config(
+        max_pool_connections=5,  # Same config as ECS client
+        retries={"max_attempts": 2, "mode": "adaptive"},
+    )
+
+    if profile_name:
+        session = boto3.Session(profile_name=profile_name)
+        return session.client("cloudwatch", config=config)
+    return boto3.client("cloudwatch", config=config)
+
+
 def _navigate_clusters(navigator: ECSNavigator, ecs_service: ECSService) -> None:
     """Handle cluster-level navigation with back support."""
     while True:
@@ -144,6 +159,10 @@ def _navigate_services(navigator: ECSNavigator, ecs_service: ECSService, cluster
             navigator.show_service_events(cluster_name, selected_service)
             # Continue the loop to show the menu again
 
+        elif selection_type == "action" and action_name == "show_metrics":
+            navigator.show_service_metrics(cluster_name, selected_service)
+            # Continue the loop to show the menu again
+
 
 def _handle_task_features(
     navigator: ECSNavigator, cluster_name: str, task_arn: str, task_details: TaskDetails | None, service_name: str

src/lazy_ecs/aws_service.py

@@ -5,14 +5,15 @@
 from collections.abc import Callable, Generator
 from typing import TYPE_CHECKING, Any
 
-from .core.types import LogConfig, ServiceEvent, ServiceInfo, TaskDetails, TaskInfo
+from .core.types import LogConfig, ServiceEvent, ServiceInfo, ServiceMetrics, TaskDetails, TaskInfo
 from .features.cluster.cluster import ClusterService
 from .features.container.container import ContainerService
 from .features.service.actions import ServiceActions
 from .features.service.service import ServiceService
 from .features.task.task import TaskService
 
 if TYPE_CHECKING:
+    from mypy_boto3_cloudwatch.client import CloudWatchClient
     from mypy_boto3_ecs.client import ECSClient
     from mypy_boto3_logs.client import CloudWatchLogsClient
     from mypy_boto3_logs.type_defs import (
@@ -31,9 +32,11 @@ def __init__(
         ecs_client: ECSClient,
         sts_client: STSClient | None = None,
         logs_client: CloudWatchLogsClient | None = None,
+        cloudwatch_client: CloudWatchClient | None = None,
     ) -> None:
         self.ecs_client = ecs_client
         self.sts_client = sts_client
+        self.cloudwatch_client = cloudwatch_client
         # Initialize feature services
         self._cluster = ClusterService(ecs_client)
         self._service = ServiceService(ecs_client)
@@ -130,3 +133,11 @@ def get_service_events(self, cluster_name: str, service_name: str) -> list[Servi
     def force_new_deployment(self, cluster_name: str, service_name: str) -> bool:
         """Force a new deployment for a service."""
         return self._service_actions.force_new_deployment(cluster_name, service_name)
+
+    def get_service_metrics(self, cluster_name: str, service_name: str, hours: int = 1) -> ServiceMetrics | None:
+        """Get CloudWatch metrics (CPU/Memory utilization) for a service."""
+        if not self.cloudwatch_client:
+            return None
+        from .features.service.metrics import get_service_metrics
+
+        return get_service_metrics(self.cloudwatch_client, cluster_name, service_name, hours)

src/lazy_ecs/core/types.py

@@ -70,3 +70,15 @@ class TaskHistoryDetails(TypedDict):
     started_at: datetime | None
     stopped_at: datetime | None
     containers: list[ContainerHistoryInfo]
+
+
+class MetricStatistics(TypedDict):
+    current: float
+    average: float
+    maximum: float
+    minimum: float
+
+
+class ServiceMetrics(TypedDict):
+    cpu: MetricStatistics
+    memory: MetricStatistics

src/lazy_ecs/features/service/metrics.py

@@ -0,0 +1,118 @@
+"""CloudWatch metrics operations for ECS services."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from typing import TYPE_CHECKING
+
+from ...core.types import MetricStatistics, ServiceMetrics
+
+if TYPE_CHECKING:
+    from mypy_boto3_cloudwatch.client import CloudWatchClient
+    from mypy_boto3_cloudwatch.type_defs import DatapointTypeDef
+
+
+def get_service_metrics(
+    cloudwatch_client: CloudWatchClient,
+    cluster_name: str,
+    service_name: str,
+    hours: int = 1,
+) -> ServiceMetrics | None:
+    """Fetch CPU and Memory utilization metrics for an ECS service from CloudWatch."""
+    utc_now = datetime.now(tz=UTC)
+    start_time = utc_now - timedelta(hours=hours)
+    end_time = utc_now
+
+    cpu_stats = _get_metric_statistics(
+        cloudwatch_client=cloudwatch_client,
+        cluster_name=cluster_name,
+        service_name=service_name,
+        metric_name="CPUUtilization",
+        start_time=start_time,
+        end_time=end_time,
+    )
+
+    memory_stats = _get_metric_statistics(
+        cloudwatch_client=cloudwatch_client,
+        cluster_name=cluster_name,
+        service_name=service_name,
+        metric_name="MemoryUtilization",
+        start_time=start_time,
+        end_time=end_time,
+    )
+
+    if cpu_stats is None or memory_stats is None:
+        return None
+
+    return {"cpu": cpu_stats, "memory": memory_stats}
+
+
+def format_metrics_display(metrics: ServiceMetrics) -> list[str]:
+    """Format service metrics into human-readable display lines."""
+    lines = []
+
+    cpu = metrics["cpu"]
+    memory = metrics["memory"]
+
+    cpu_line = (
+        f"CPU:    Current: {cpu['current']:5.1f}%  |  Avg: {cpu['average']:5.1f}%  |  "
+        f"Peak: {cpu['maximum']:5.1f}%  |  Low: {cpu['minimum']:5.1f}%"
+    )
+    memory_line = (
+        f"Memory: Current: {memory['current']:5.1f}%  |  Avg: {memory['average']:5.1f}%  |  "
+        f"Peak: {memory['maximum']:5.1f}%  |  Low: {memory['minimum']:5.1f}%"
+    )
+
+    lines.append(cpu_line)
+    lines.append(memory_line)
+
+    return lines
+
+
+def _get_metric_statistics(
+    cloudwatch_client: CloudWatchClient,
+    cluster_name: str,
+    service_name: str,
+    metric_name: str,
+    start_time: datetime,
+    end_time: datetime,
+) -> MetricStatistics | None:
+    """Fetch statistics for a single metric."""
+    response = cloudwatch_client.get_metric_statistics(
+        Namespace="AWS/ECS",
+        MetricName=metric_name,
+        Dimensions=[
+            {"Name": "ClusterName", "Value": cluster_name},
+            {"Name": "ServiceName", "Value": service_name},
+        ],
+        StartTime=start_time,
+        EndTime=end_time,
+        Period=300,
+        Statistics=["Average", "Maximum", "Minimum"],
+    )
+
+    datapoints: list[DatapointTypeDef] = response.get("Datapoints", [])
+
+    if not datapoints:
+        return None
+
+    sorted_datapoints = sorted(datapoints, key=lambda x: x.get("Timestamp", datetime.min), reverse=True)
+
+    current_datapoint = sorted_datapoints[0]
+    current_value = current_datapoint.get("Average", 0.0)
+
+    all_averages = [dp.get("Average", 0.0) for dp in datapoints if "Average" in dp]
+    average_value = sum(all_averages) / len(all_averages) if all_averages else 0.0
+
+    all_maximums = [dp.get("Maximum", 0.0) for dp in datapoints if "Maximum" in dp]
+    maximum_value = max(all_maximums) if all_maximums else 0.0
+
+    all_minimums = [dp.get("Minimum", 0.0) for dp in datapoints if "Minimum" in dp]
+    minimum_value = min(all_minimums) if all_minimums else 0.0
+
+    return {
+        "current": current_value,
+        "average": average_value,
+        "maximum": maximum_value,
+        "minimum": minimum_value,
+    }

src/lazy_ecs/features/service/ui.py

@@ -8,9 +8,10 @@
 
 from ...core.base import BaseUIComponent
 from ...core.navigation import select_with_auto_pagination
-from ...core.types import TaskInfo
+from ...core.types import ServiceMetrics, TaskInfo
 from ...core.utils import show_spinner
 from .actions import ServiceActions
+from .metrics import format_metrics_display
 from .service import ServiceService
 
 console = Console()
@@ -44,6 +45,7 @@ def select_service_action(self, service_name: str, task_info: list[TaskInfo]) ->
             choices.append({"name": task["name"], "value": f"task:show_details:{task['value']}"})
 
         choices.append({"name": "📋 Show service events", "value": "action:show_events"})
+        choices.append({"name": "📊 Show metrics", "value": "action:show_metrics"})
         choices.append({"name": "🚀 Force new deployment", "value": "action:force_deployment"})
 
         return select_with_auto_pagination(
@@ -109,6 +111,13 @@ def display_service_events(self, cluster_name: str, service_name: str) -> None:
 
         console.print(table)
 
+    def display_service_metrics(self, service_name: str, metrics: ServiceMetrics) -> None:
+        """Display service metrics."""
+        lines = format_metrics_display(metrics)
+        console.print(f"\n[bold cyan]Metrics for service '{service_name}' (last hour):[/bold cyan]")
+        for line in lines:
+            console.print(line)
+
 
 def _get_event_type_style(event_type: str) -> str:
     event_styles = {

src/lazy_ecs/ui.py

@@ -96,6 +96,20 @@ def handle_force_deployment(self, cluster_name: str, service_name: str) -> None:
     def show_service_events(self, cluster_name: str, service_name: str) -> None:
         return self._service_ui.display_service_events(cluster_name, service_name)
 
+    def show_service_metrics(self, cluster_name: str, service_name: str) -> None:
+        """Fetch and display service metrics."""
+        with show_spinner():
+            metrics = self.ecs_service.get_service_metrics(cluster_name, service_name, hours=1)
+
+        if metrics:
+            self._service_ui.display_service_metrics(service_name, metrics)
+        else:
+            console.print(f"\n⚠️ No metrics available for service '{service_name}'", style="yellow")
+            console.print("This could mean:", style="dim")
+            console.print("  - The service has no running tasks", style="dim")
+            console.print("  - CloudWatch metrics are not yet available", style="dim")
+            console.print("  - The service was recently created", style="dim")
+
     def show_task_history(self, cluster_name: str, service_name: str) -> None:
         self._task_ui.display_task_history(cluster_name, service_name)