From 48e136030851bb44af5a326f3ad24aafd125ee88 Mon Sep 17 00:00:00 2001 From: "Matthew C. Morgan" Date: Wed, 2 Apr 2025 20:52:40 -0400 Subject: [PATCH] refactor for clarity --- .../aws_resource_management/core.py | 151 +------ .../aws_resource_management/discovery.py | 395 ++---------------- .../managers/__init__.py | 1 + .../aws_resource_management/managers/base.py | 252 ++++++----- .../aws_resource_management/managers/ec2.py | 295 +++++++------ .../aws_resource_management/managers/ecr.py | 118 ++++++ .../aws_resource_management/managers/eks.py | 68 +++ .../aws_resource_management/managers/emr.py | 167 ++++---- .../aws_resource_management/managers/rds.py | 302 +++++++------ .../aws_resource_management/reporting.py | 135 ++++++ 10 files changed, 952 insertions(+), 932 deletions(-) create mode 100644 local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/ecr.py create mode 100644 local-app/python-tools/gfl-resource-actions/aws_resource_management/reporting.py diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/core.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/core.py index 4cf64809..5ab3a97d 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/core.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/core.py @@ -23,14 +23,13 @@ EMRManager, RDSManager, ) +from aws_resource_management.reporting import print_resource_summary, initialize_stats logger = setup_logging() config = get_config() - class ResourceManager: """Main resource manager that orchestrates operations across accounts and resources.""" - def __init__( self, profile_name: Optional[str] = None, @@ -56,7 +55,6 @@ def __init__( self.partition = partition # Cache for discovered regions by account self.region_cache = {} - # If no partition was specified, we'll auto-detect based on credentials # when we process the first account @@ -88,34 +86,13 @@ def process_accounts( try: if exclude_accounts is None: exclude_accounts = [] - if exclude_resources is None: exclude_resources = [] - if exclude_regions is None: exclude_regions = [] - # Initialize stats if not provided if stats is None: - stats = { - "accounts_processed": 0, - "accounts_skipped": 0, - "resources_processed": 0, - "resources_skipped": 0, - "regions_processed": 0, - "regions_by_account": {}, - } - - # Initialize resource-specific counters - for resource_type in ["ec2", "rds", "eks", "emr"]: - for stat_type in [ - "found", - "skipped", - "stopped", - "started", - "errors", - ]: - stats[f"{resource_type}_{stat_type}"] = 0 + stats = initialize_stats() # Get account list - either from profiles or Organizations API if self.use_profiles: @@ -132,9 +109,8 @@ def process_accounts( # Process each account for account in accounts: - account_id = account.get("account_id") account_name = account.get("account_name", "Unknown") - + account_id = account.get("account_id") # Skip excluded accounts if account_id in exclude_accounts: logger.info( @@ -155,8 +131,6 @@ def process_accounts( # Determine regions to scan for this account account_regions = regions - - # If auto-discovery of regions is enabled, get all enabled regions for this account if self.discover_regions: try: logger.info( @@ -174,7 +148,6 @@ def process_accounts( logger.info( f"Found {len(account_regions)} enabled regions in account {account_id}" ) - # Cache discovered regions self.region_cache[account_id] = account_regions except Exception as e: @@ -198,8 +171,8 @@ def process_accounts( ] # Track regions processed for this account - stats["regions_by_account"][account_id] = account_regions stats["regions_processed"] += len(account_regions) + stats["regions_by_account"][account_id] = account_regions # Process this account with its regions logger.info( @@ -211,11 +184,10 @@ def process_accounts( credentials=credentials, regions=account_regions, action=action, - dry_run=dry_run, exclude_resources=exclude_resources, + dry_run=dry_run, stats=stats, ) - stats["accounts_processed"] += 1 except Exception as e: @@ -239,97 +211,16 @@ def process_accounts( logger.debug( f"Account {account_id} regions: {', '.join(account_regions)}" ) - - # Print detailed resource summary self._print_resource_summary(stats, action) - return stats except KeyboardInterrupt: - # Log the interruption but re-raise to ensure application exit logger.warning("Account processing interrupted by user") raise def _print_resource_summary(self, stats: Dict[str, Any], action: str) -> None: - """ - Print a detailed summary of resources processed, broken down by resource type. - - Args: - stats: Statistics dictionary - action: Action that was performed (stop or start) - """ - logger.info(f"\n{'=' * 30} SUMMARY {'=' * 30}") - logger.info( - f"ACTION: {action.upper()} {'(DRY RUN)' if stats.get('dry_run', False) else ''}" - ) - - # General stats - logger.info(f"\nGENERAL STATISTICS:") - logger.info(f"Accounts processed: {stats.get('accounts_processed', 0)}") - logger.info(f"Accounts skipped: {stats.get('accounts_skipped', 0)}") - logger.info(f"Regions processed: {stats.get('regions_processed', 0)}") - - # EC2 resources - logger.info(f"\nEC2 INSTANCES:") - logger.info(f"Found: {stats.get('ec2_found', 0)}") - if action == "stop": - logger.info(f"Stopped: {stats.get('ec2_stopped', 0)}") - else: - logger.info(f"Started: {stats.get('ec2_started', 0)}") - logger.info(f"Skipped: {stats.get('ec2_skipped', 0)}") - logger.info(f"Errors: {stats.get('ec2_errors', 0)}") - - # RDS resources - logger.info(f"\nRDS INSTANCES:") - logger.info(f"Found: {stats.get('rds_found', 0)}") - if action == "stop": - logger.info(f"Stopped: {stats.get('rds_stopped', 0)}") - else: - logger.info(f"Started: {stats.get('rds_started', 0)}") - logger.info(f"Skipped: {stats.get('rds_skipped', 0)}") - logger.info(f"Errors: {stats.get('rds_errors', 0)}") - - # RDS engine breakdown if available - rds_engines = stats.get("rds_engines", {}) - if rds_engines: - logger.info( - f"RDS engines: {', '.join(f'{engine}({count})' for engine, count in rds_engines.items())}" - ) - - # EKS resources - logger.info(f"\nEKS CLUSTERS:") - logger.info(f"Found: {stats.get('eks_found', 0)}") - if action == "stop": - logger.info(f"Stopped: {stats.get('eks_stopped', 0)}") - else: - logger.info(f"Started: {stats.get('eks_started', 0)}") - logger.info(f"Skipped: {stats.get('eks_skipped', 0)}") - logger.info(f"Errors: {stats.get('eks_errors', 0)}") - - # EMR resources - logger.info(f"\nEMR CLUSTERS:") - logger.info(f"Found: {stats.get('emr_found', 0)}") - if action == "stop": - logger.info(f"Stopped: {stats.get('emr_stopped', 0)}") - else: - logger.info(f"Started: {stats.get('emr_started', 0)}") - logger.info(f"Skipped: {stats.get('emr_skipped', 0)}") - logger.info(f"Errors: {stats.get('emr_errors', 0)}") - - # Error summary if there were any errors - if stats.get("errors", []): - logger.info(f"\nERROR SUMMARY:") - logger.info(f"Total errors: {len(stats.get('errors', []))}") - for i, error in enumerate( - stats.get("errors", [])[:5], 1 - ): # Show first 5 errors - logger.info(f" {i}. {error}") - if len(stats.get("errors", [])) > 5: - logger.info( - f" ... and {len(stats.get('errors', [])) - 5} more errors (see log for details)" - ) - - logger.info(f"{'=' * 68}") + """Print a summary of the resources processed.""" + print_resource_summary(stats, action, stats.get("dry_run", False)) def _process_single_account( self, @@ -408,7 +299,6 @@ def _process_single_account( f"No regions specified, using defaults for partition {self.partition}: {', '.join(regions)}" ) - # Only log this once - removing duplicate message logger.info( f"Processing account: {account_name} ({account_id}) in {len(regions)} regions" ) @@ -416,14 +306,12 @@ def _process_single_account( # Define special handling for EMR clusters - remove 'STOPPED' from valid states as it's not accepted by the API emr_states = None if "emr" not in exclude_resources: - # Only include valid EMR states for the ListClusters API call - # Note: 'STOPPED' is not a valid state for ListClusters API emr_states = [ "STARTING", "BOOTSTRAPPING", "RUNNING", - "WAITING", "TERMINATING", + "WAITING", ] logger.debug( f"Using valid EMR states for discovery: {', '.join(emr_states)}" @@ -464,6 +352,8 @@ def _process_single_account( "emr_stopped", "emr_started", "emr_errors", + "ecr_images_found", + "ecr_old_images_found", ]: if stat_type not in stats: stats[stat_type] = 0 @@ -477,7 +367,6 @@ def _process_single_account( stats["errors"] = [] # Normalize state and status fields for all resource types - # For EC2 instances ec2_instances = resources.get("ec2_instances", []) for instance in ec2_instances: if "state" not in instance and "status" in instance: @@ -485,10 +374,9 @@ def _process_single_account( elif "status" not in instance and "state" in instance: instance["status"] = instance["state"] elif "state" not in instance and "status" not in instance: - instance["state"] = "unknown" instance["status"] = "unknown" + instance["state"] = "unknown" - # For RDS instances rds_instances = resources.get("rds_instances", []) for instance in rds_instances: if "state" not in instance and "status" in instance: @@ -496,10 +384,9 @@ def _process_single_account( elif "status" not in instance and "state" in instance: instance["status"] = instance["state"] elif "state" not in instance and "status" not in instance: - instance["state"] = "unknown" instance["status"] = "unknown" + instance["state"] = "unknown" - # For EKS clusters eks_clusters = resources.get("eks_clusters", []) for cluster in eks_clusters: if "state" not in cluster and "status" in cluster: @@ -507,10 +394,9 @@ def _process_single_account( elif "status" not in cluster and "state" in cluster: cluster["status"] = cluster["state"] elif "state" not in cluster and "status" not in cluster: - cluster["state"] = "unknown" cluster["status"] = "unknown" + cluster["state"] = "unknown" - # For EMR clusters emr_clusters = resources.get("emr_clusters", []) for cluster in emr_clusters: if "state" not in cluster: @@ -520,10 +406,9 @@ def _process_single_account( # Count service-managed EC2 instances eks_tag = self.config.get("eks_tag", "kubernetes.io/cluster/") - emr_tag = self.config.get("emr_tag", "aws:elasticmapreduce:job-flow-id") exclusion_tag = self.config.get("exclusion_tag", "DoNotStop") + emr_tag = self.config.get("emr_tag", "aws:elasticmapreduce:job-flow-id") - # Continue with existing code eks_managed = sum( 1 for i in resources.get("ec2_instances", []) @@ -546,8 +431,10 @@ def _process_single_account( stats["ec2_skipped"] += excluded_ec2 stats["rds_found"] += len(resources.get("rds_instances", [])) - stats["eks_found"] += len(resources.get("eks_clusters", [])) stats["emr_found"] += len(resources.get("emr_clusters", [])) + stats["eks_found"] += len(resources.get("eks_clusters", [])) + stats["ecr_images_found"] += len(resources.get("ecr_images", [])) + stats["ecr_old_images_found"] += len(resources.get("ecr_old_images", [])) # Track RDS engines for instance in resources.get("rds_instances", []): @@ -571,6 +458,10 @@ def _process_single_account( logger.info( f"Account {account_id} - Found {len(resources.get('emr_clusters', []))} EMR clusters" ) + logger.info( + f"Account {account_id} - Found {len(resources.get('ecr_images', []))} ECR images " + f"({len(resources.get('ecr_old_images', []))} older than 1 year)" + ) # Initialize resource managers with account name ec2_manager = EC2Manager(credentials, account_id, account_name) diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/discovery.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/discovery.py index d327060f..e7c1c183 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/discovery.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/discovery.py @@ -16,6 +16,13 @@ ) from aws_resource_management.config_manager import get_config from aws_resource_management.logging_setup import setup_logging +from aws_resource_management.managers import ( + ECRManager, + EC2Manager, + EKSManager, + EMRManager, + RDSManager, +) from botocore.exceptions import ClientError, EndpointConnectionError logger = setup_logging() @@ -494,369 +501,51 @@ def get_account_resources( regions: List[str], exclude_resources: List[str], account_id: str, - account_name: str, + account_name: Optional[str] = None, emr_cluster_states: Optional[List[str]] = None, ) -> Dict[str, List[Dict[str, Any]]]: """ - Discover resources in an AWS account across multiple regions. - - Args: - credentials: AWS credentials for the account - regions: List of regions to scan - exclude_resources: List of resource types to exclude - account_id: AWS account ID - account_name: AWS account name - emr_cluster_states: Optional list of EMR cluster states to filter by - - Returns: - Dictionary of discovered resources by type + Get all resources for an account across multiple regions. """ - # Use default regions if empty regions list - if not regions: - # First detect partition to get appropriate regions - partition = detect_partition_from_credentials(credentials) - # Get regions for the detected partition - regions = get_regions_for_partition(partition) - logger.info( - f"Using default regions for partition {partition} in account {account_id}" - ) - - logger.info( - f"Discovering resources in account {account_id} ({account_name}) across {len(regions)} regions" - ) - resources = { "ec2_instances": [], "rds_instances": [], "eks_clusters": [], "emr_clusters": [], + "ecr_images": [], + "ecr_old_images": [], } - # Create a ResourceDiscovery instance for this account - discovery = ResourceDiscovery(credentials, account_id) - - # Use parallel processing for regions to speed up discovery - with ThreadPoolExecutor(max_workers=min(10, len(regions))) as executor: - # Submit tasks for each region and resource type - futures = {} - - for region in regions: - logger.debug(f"Scanning region {region} in account {account_id}") - - # Get EC2 instances if not excluded - if "ec2" not in exclude_resources: - future = executor.submit(discovery._discover_ec2, region) - futures[(region, "ec2")] = future - - # Get RDS instances if not excluded - if "rds" not in exclude_resources: - future = executor.submit(discovery._discover_rds, region) - futures[(region, "rds")] = future - - # Get EKS clusters if not excluded - if "eks" not in exclude_resources: - future = executor.submit(discovery._discover_eks, region) - futures[(region, "eks")] = future - - # Get EMR clusters if not excluded - if "emr" not in exclude_resources: - future = executor.submit( - discovery._discover_emr, region, emr_cluster_states - ) - futures[(region, "emr")] = future - - # Collect results as they complete - for (region, resource_type), future in futures.items(): - try: - result = future.result() - if resource_type == "ec2": - resources["ec2_instances"].extend(result) - elif resource_type == "rds": - resources["rds_instances"].extend(result) - elif resource_type == "eks": - resources["eks_clusters"].extend(result) - elif resource_type == "emr": - resources["emr_clusters"].extend(result) - logger.debug( - f"Found {len(result)} {resource_type} resources in {region}" - ) - except Exception as e: - logger.error(f"Error discovering {resource_type} in {region}: {str(e)}") - - # Log summary - logger.info( - f"Account {account_id} - Found {len(resources['ec2_instances'])} EC2 instances" - ) - logger.info( - f"Account {account_id} - Found {len(resources['rds_instances'])} RDS instances" - ) - logger.info( - f"Account {account_id} - Found {len(resources['eks_clusters'])} EKS clusters" - ) - logger.info( - f"Account {account_id} - Found {len(resources['emr_clusters'])} EMR clusters" - ) + # Discover EC2 instances + if "ec2" not in exclude_resources: + logger.info(f"Discovering EC2 instances for account {account_id} in {len(regions)} regions") + ec2_manager = EC2Manager(credentials, account_id, account_name) + resources["ec2_instances"] = ec2_manager.discover_instances(regions) + + # Discover RDS instances + if "rds" not in exclude_resources: + logger.info(f"Discovering RDS instances for account {account_id} in {len(regions)} regions") + rds_manager = RDSManager(credentials, account_id, account_name) + resources["rds_instances"] = rds_manager.discover_instances(regions) + + # Discover EKS clusters + if "eks" not in exclude_resources: + logger.info(f"Discovering EKS clusters for account {account_id} in {len(regions)} regions") + eks_manager = EKSManager(credentials, account_id, account_name) + resources["eks_clusters"] = eks_manager.discover_clusters(regions) + + # Discover EMR clusters + if "emr" not in exclude_resources: + logger.info(f"Discovering EMR clusters for account {account_id} in {len(regions)} regions") + emr_manager = EMRManager(credentials, account_id, account_name) + resources["emr_clusters"] = emr_manager.discover_clusters(regions, emr_cluster_states) + + # Discover ECR images + if "ecr" not in exclude_resources: + logger.info(f"Discovering ECR images for account {account_id} in {len(regions)} regions") + ecr_manager = ECRManager(credentials, account_id, account_name) + result = ecr_manager.discover_images_by_age(regions) + resources["ecr_images"] = result["all_images"] + resources["ecr_old_images"] = result["old_images"] return resources - - -def discover_ec2_instances( - credentials: Dict[str, str], region: str, account_id: str -) -> List[Dict[str, Any]]: - """ - Discover EC2 instances in a region. - - Args: - credentials: AWS credentials - region: AWS region - account_id: AWS account ID - - Returns: - List of EC2 instance dictionaries - """ - ec2_client = boto3.client("ec2", region_name=region, **credentials) - instances = [] - - try: - paginator = ec2_client.get_paginator("describe_instances") - - for page in paginator.paginate(): - for reservation in page["Reservations"]: - for instance in reservation["Instances"]: - # Skip terminated instances - if instance["State"]["Name"] == "terminated": - continue - - # Extract tags as a dictionary - tags = {} - if "Tags" in instance: - tags = {tag["Key"]: tag["Value"] for tag in instance["Tags"]} - - # Create a simplified instance object - instance_obj = { - "id": instance["InstanceId"], - "type": instance["InstanceType"], - "state": instance["State"]["Name"], - "region": region, - "account_id": account_id, - "tags": tags, - "name": tags.get("Name", instance["InstanceId"]), - } - - instances.append(instance_obj) - - return instances - except ClientError as e: - logger.error(f"Error discovering EC2 instances in {region}: {str(e)}") - return [] - - -def discover_rds_instances( - credentials: Dict[str, str], region: str, account_id: str -) -> List[Dict[str, Any]]: - """ - Discover RDS instances in a region. - - Args: - credentials: AWS credentials - region: AWS region - account_id: AWS account ID - - Returns: - List of RDS instance dictionaries - """ - rds_client = boto3.client("rds", region_name=region, **credentials) - instances = [] - - try: - paginator = rds_client.get_paginator("describe_db_instances") - - for page in paginator.paginate(): - for instance in page["DBInstances"]: - # Create a simplified instance object - instance_obj = { - "id": instance["DBInstanceIdentifier"], - "engine": instance["Engine"], - "state": instance["DBInstanceStatus"], - "region": region, - "account_id": account_id, - "name": instance["DBInstanceIdentifier"], - } - - instances.append(instance_obj) - - return instances - except ClientError as e: - logger.error(f"Error discovering RDS instances in {region}: {str(e)}") - return [] - - -def discover_eks_clusters( - credentials: Dict[str, str], region: str, account_id: str -) -> List[Dict[str, Any]]: - """ - Discover EKS clusters in a region. - - Args: - credentials: AWS credentials - region: AWS region - account_id: AWS account ID - - Returns: - List of EKS cluster dictionaries - """ - eks_client = boto3.client("eks", region_name=region, **credentials) - clusters = [] - - try: - response = eks_client.list_clusters() - - for cluster_name in response["clusters"]: - cluster_details = eks_client.describe_cluster(name=cluster_name)["cluster"] - - # Create a simplified cluster object - cluster_obj = { - "id": cluster_name, - "name": cluster_name, - "status": cluster_details["status"], - "region": region, - "account_id": account_id, - "version": cluster_details["version"], - } - - clusters.append(cluster_obj) - - # Handle pagination if there are more clusters - while "nextToken" in response: - response = eks_client.list_clusters(nextToken=response["nextToken"]) - for cluster_name in response["clusters"]: - cluster_details = eks_client.describe_cluster(name=cluster_name)[ - "cluster" - ] - - # Create a simplified cluster object - cluster_obj = { - "id": cluster_name, - "name": cluster_name, - "status": cluster_details["status"], - "region": region, - "account_id": account_id, - "version": cluster_details["version"], - } - - clusters.append(cluster_obj) - - return clusters - except ClientError as e: - logger.error(f"Error discovering EKS clusters in {region}: {str(e)}") - return [] - - -def discover_emr_clusters( - credentials: Dict[str, str], - region: str, - account_id: str, - cluster_states: Optional[List[str]] = None, -) -> List[Dict[str, Any]]: - """ - Discover EMR clusters in a region. - - Args: - credentials: AWS credentials - region: AWS region - account_id: AWS account ID - cluster_states: Optional list of EMR cluster states to filter by - - Returns: - List of EMR cluster dictionaries - """ - emr_client = boto3.client("emr", region_name=region, **credentials) - clusters = [] - - try: - paginator = emr_client.get_paginator("list_clusters") - - # Only list active clusters with valid states - # Default cluster states, 'STOPPED' is not a valid state for ListClusters API - # See: https://docs.aws.amazon.com/emr/latest/APIReference/API_ListClusters.html - if cluster_states is None: - cluster_states = [ - "STARTING", - "BOOTSTRAPPING", - "RUNNING", - "WAITING", - "TERMINATING", - ] - - try: - for page in paginator.paginate(ClusterStates=cluster_states): - for cluster in page.get("Clusters", []): - try: - # Safely extract cluster state with fallbacks - if ( - "Status" in cluster - and isinstance(cluster["Status"], dict) - and "State" in cluster["Status"] - ): - state = cluster["Status"]["State"] - else: - state = "UNKNOWN" - - # Create a simplified cluster object with both state and status fields - cluster_obj = { - "id": cluster.get("Id", "unknown-id"), - "name": cluster.get("Name", "Unknown"), - "state": state, - "status": state, # Duplicate to ensure both fields exist - "region": region, - "account_id": account_id, - } - - clusters.append(cluster_obj) - except Exception as e: - logger.warning( - f"Error processing EMR cluster in {region}: {str(e)}" - ) - # Continue with next cluster - continue - - except Exception as e: - logger.warning(f"Error during EMR pagination in {region}: {str(e)}") - # Try using list_clusters without pagination as fallback - try: - response = emr_client.list_clusters(ClusterStates=cluster_states) - for cluster in response.get("Clusters", []): - # Safely extract cluster state with fallbacks - if ( - "Status" in cluster - and isinstance(cluster["Status"], dict) - and "State" in cluster["Status"] - ): - state = cluster["Status"]["State"] - else: - state = "UNKNOWN" - - # Create a simplified cluster object with both state and status fields - cluster_obj = { - "id": cluster.get("Id", "unknown-id"), - "name": cluster.get("Name", "Unknown"), - "state": state, - "status": state, # Duplicate to ensure both fields exist - "region": region, - "account_id": account_id, - } - - clusters.append(cluster_obj) - except Exception as nested_e: - logger.error( - f"Fallback EMR cluster retrieval failed in {region}: {str(nested_e)}" - ) - - return clusters - except ClientError as e: - logger.error(f"Error discovering EMR clusters in {region}: {str(e)}") - return [] - except Exception as e: - # Added more generic exception handling - logger.error(f"Unexpected error discovering EMR clusters in {region}: {str(e)}") - return [] diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/__init__.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/__init__.py index e2526461..0cbfac73 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/__init__.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/__init__.py @@ -3,6 +3,7 @@ """ from aws_resource_management.managers.base import ResourceManager +from aws_resource_management.managers.ecr import ECRManager from aws_resource_management.managers.ec2 import EC2Manager from aws_resource_management.managers.eks import EKSManager from aws_resource_management.managers.emr import EMRManager diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/base.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/base.py index a5602032..9a85ec99 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/base.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/base.py @@ -1,32 +1,23 @@ """ -Base resource manager class that all specific managers inherit from. +Base resource manager class for AWS resources. """ -import datetime -from abc import ABC, abstractmethod +import logging +import time +from datetime import datetime from typing import Any, Dict, List, Optional, Union import boto3 from aws_resource_management.config_manager import get_config -from aws_resource_management.logging_setup import setup_logging -logger = setup_logging() +logger = logging.getLogger(__name__) config = get_config() - -class ResourceManager(ABC): - """ - Base class for all resource managers. - - This abstract class defines the common interface and functionality - that all resource managers should implement. - """ +class ResourceManager: + """Base class for AWS resource managers.""" def __init__( - self, - credentials: Dict[str, str], - account_id: str, - account_name: Optional[str] = None, + self, credentials: Dict[str, str], account_id: str, account_name: Optional[str] = None ): """ Initialize the resource manager. @@ -39,132 +30,175 @@ def __init__( self.credentials = credentials self.account_id = account_id self.account_name = account_name - self.resource_type = "generic" # Override in subclass - - def create_client(self, service_name: str, region_name: str) -> Any: + self.clients = {} # Cache for boto3 clients + + def get_boto3_client(self, service_name: str, region: str) -> Any: """ - Create a boto3 client for the specified AWS service. + Get a boto3 client for the specified service and region. Args: service_name: AWS service name - region_name: AWS region name - - Returns: - boto3 client - """ - return boto3.client( - service_name, - region_name=region_name, - aws_access_key_id=self.credentials.get("aws_access_key_id"), - aws_secret_access_key=self.credentials.get("aws_secret_access_key"), - aws_session_token=self.credentials.get("aws_session_token"), - ) - - def get_timestamp(self) -> str: - """ - Get current timestamp in ISO format. + region: AWS region Returns: - Timestamp string - """ - return datetime.datetime.now().isoformat() - + Boto3 client object or None if creation fails + """ + client_key = f"{service_name}-{region}" + if client_key in self.clients: + return self.clients[client_key] + + try: + client = boto3.client( + service_name, + region_name=region, + aws_access_key_id=self.credentials.get("aws_access_key_id"), + aws_secret_access_key=self.credentials.get("aws_secret_access_key"), + aws_session_token=self.credentials.get("aws_session_token"), + ) + self.clients[client_key] = client + return client + except Exception as e: + logger.error(f"Error creating {service_name} client in {region}: {e}") + return None + + # Alias for backwards compatibility + create_client = get_boto3_client + def should_exclude(self, resource: Dict[str, Any]) -> bool: """ - Check if a resource should be excluded from operations based on tags. + Check if a resource should be excluded based on tags. Args: - resource: Resource dictionary with a 'tags' key + resource: Resource dictionary with tags Returns: - True if resource should be excluded, False otherwise + True if the resource should be excluded, False otherwise """ tags = resource.get("tags", {}) exclusion_tag = config.get("exclusion_tag") - - # Check if the exclusion tag exists - if exclusion_tag in tags: - return True - - return False - - def log_action( - self, - resource_id: str, - region: str, - action: str, - details: str = "", - resource_name: Optional[str] = None, - dry_run: bool = False, - existing_schedule: Optional[str] = None, - ) -> None: + + return exclusion_tag in tags + + def get_timestamp(self) -> str: """ - Log an action performed on a resource. + Get current timestamp in ISO 8601 format. - Args: - resource_id: Resource ID - region: AWS region - action: Action name (e.g., start, stop) - details: Additional details about the action - resource_name: Resource name (optional) - dry_run: Whether this was a dry run - existing_schedule: Existing schedule if any (optional) + Returns: + ISO 8601 timestamp string """ - resource_info = ( - f"'{resource_name}' ({resource_id})" if resource_name else resource_id - ) - dry_run_prefix = "[DRY RUN] " if dry_run else "" - schedule_info = f", Schedule: {existing_schedule}" if existing_schedule else "" - - logger.info( - f"{dry_run_prefix}{action.upper()} {self.resource_type} {resource_info} in {region}{schedule_info} {details}" - ) - + return datetime.utcnow().isoformat() + def get_partition_for_region(self, region: str) -> str: """ - Get the AWS partition for a given region. + Get AWS partition for a region. Args: region: AWS region Returns: - AWS partition (e.g., aws, aws-us-gov) + AWS partition string (aws, aws-us-gov, aws-cn) """ - if region.startswith("us-gov"): + if region.startswith("us-gov-"): return "aws-us-gov" - elif region.startswith("cn"): + elif region.startswith("cn-"): return "aws-cn" else: return "aws" - - @abstractmethod - def stop( - self, resources: List[Dict[str, Any]], dry_run: bool = False - ) -> Dict[str, int]: + + def log_action( + self, + resource_id: str, + region: str, + action: str, + resource_name: Optional[str] = None, + details: Optional[str] = None, + dry_run: bool = False, + existing_schedule: Optional[str] = None + ) -> None: """ - Stop resources. + Log an action taken on a resource. Args: - resources: List of resource dictionaries - dry_run: If True, only simulate the action - - Returns: - Dictionary with success and error counts - """ - pass + resource_id: Resource ID + region: AWS region + action: Action name (start, stop, etc.) + resource_name: Resource name (optional) + details: Action details (optional) + dry_run: Whether this was a dry run + existing_schedule: Existing schedule tag value (optional) + """ + name_str = f" ({resource_name})" if resource_name else "" + action_str = f"{action.upper()}" + details_str = f": {details}" if details else "" + schedule_str = f" [Schedule: {existing_schedule}]" if existing_schedule else "" + dry_run_str = "[DRY RUN] " if dry_run else "" + + logger.info( + f"{dry_run_str}{action_str} {self.resource_type} {resource_id}{name_str} " + f"in {region}{details_str}{schedule_str}" + ) - @abstractmethod - def start( - self, resources: List[Dict[str, Any]], dry_run: bool = False - ) -> Dict[str, int]: + def paginate_boto3(self, client, method_name, result_key, **kwargs): """ - Start resources. - + Helper method to handle pagination for boto3 API calls. + Args: - resources: List of resource dictionaries - dry_run: If True, only simulate the action - + client: Boto3 client + method_name: API method name to call + result_key: Key in the response that contains the results + **kwargs: Additional arguments to pass to the method + Returns: - Dictionary with success and error counts - """ - pass + Combined list of all items across pages + """ + method = getattr(client, method_name) + items = [] + + try: + # Try to use paginator if available + try: + paginator = client.get_paginator(method_name) + for page in paginator.paginate(**kwargs): + if result_key in page: + items.extend(page[result_key]) + return items + except (AttributeError, client.exceptions.ClientError): + # Fall back to manual pagination if paginator not available + pass + + # Manual pagination + response = method(**kwargs) + if result_key in response: + items.extend(response[result_key]) + + # Check for different pagination keys + pagination_keys = [ + "NextToken", "nextToken", "Marker", "marker", + "next_token", "next_marker" + ] + + pagination_key = next( + (k for k in pagination_keys if k in response), None + ) + + while pagination_key and response.get(pagination_key): + kwargs[pagination_key] = response[pagination_key] + response = method(**kwargs) + if result_key in response: + items.extend(response[result_key]) + + return items + except Exception as e: + logger.error( + f"Error paginating through {method_name} for {client._service_model.service_name}: {str(e)}" + ) + return items + + # Abstract methods that should be implemented by subclasses + def start(self, resources: List[Dict[str, Any]], dry_run: bool = False) -> Dict[str, int]: + """Start resources - abstract method to be implemented by subclasses.""" + raise NotImplementedError("Subclasses must implement start()") + + def stop(self, resources: List[Dict[str, Any]], dry_run: bool = False) -> Dict[str, int]: + """Stop resources - abstract method to be implemented by subclasses.""" + raise NotImplementedError("Subclasses must implement stop()") diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/ec2.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/ec2.py index d795b19e..4368975c 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/ec2.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/ec2.py @@ -3,6 +3,7 @@ """ from typing import Any, Dict, List, Optional +from collections import defaultdict from aws_resource_management.config_manager import get_config from aws_resource_management.logging_setup import setup_logging @@ -21,160 +22,210 @@ def __init__( account_id: str, account_name: Optional[str] = None, ): - """ - Initialize EC2 manager. - - Args: - credentials: AWS credentials dictionary - account_id: AWS account ID - account_name: AWS account name (optional) - """ + """Initialize EC2 manager.""" super().__init__(credentials, account_id, account_name) self.resource_type = "ec2_instance" + self.MAX_INSTANCES_PER_API_CALL = 50 def should_exclude(self, instance: Dict[str, Any]) -> bool: - """ - Check if EC2 instance should be excluded based on tags. - - Args: - instance: Instance dictionary with tags - - Returns: - True if the instance should be excluded, False otherwise - """ + """Check if EC2 instance should be excluded based on tags.""" tags = instance.get("tags", {}) - - # Check for explicit exclusion tag exclusion_tag = config.get("exclusion_tag") + eks_tag = config.get("eks_tag") + emr_tag = config.get("emr_tag") + if exclusion_tag in tags: - logger.info( - f"Skipping EC2 instance {instance['id']} with exclusion tag {exclusion_tag}" - ) + logger.debug(f"Skipping EC2 instance {instance['id']} with exclusion tag {exclusion_tag}") return True - - # Skip instances that are part of EKS clusters - eks_tag = config.get("eks_tag") - if eks_tag in tags: - logger.info( - f"Skipping EC2 instance {instance['id']} with tag {eks_tag}={tags[eks_tag]} (EKS managed node)" - ) + + if any(tag.startswith(eks_tag) for tag in tags): + logger.debug(f"Skipping EC2 instance {instance['id']} - EKS managed node") return True - - # Skip instances that are part of EMR clusters - emr_tag = config.get("emr_tag") + if emr_tag in tags: - logger.info( - f"Skipping EC2 instance {instance['id']} with tag {emr_tag}={tags[emr_tag]} (EMR managed node)" - ) + logger.debug(f"Skipping EC2 instance {instance['id']} - EMR managed node") return True return False - def stop( - self, instances: List[Dict[str, Any]], dry_run: bool = False - ) -> Dict[str, int]: - """ - Stop running EC2 instances. - - Args: - instances: List of EC2 instance dictionaries - dry_run: If True, only simulate the action - - Returns: - Dictionary with success and error counts - """ - success_count = 0 - error_count = 0 - + def stop(self, instances: List[Dict[str, Any]], dry_run: bool = False) -> Dict[str, int]: + """Stop running EC2 instances in batches for efficiency.""" + # Group instances by region for batch processing + running_instances_by_region = defaultdict(list) + stats = {"success": 0, "errors": 0, "skipped": 0} + + # Filter instances - keep only running instances that should not be excluded for instance in instances: if self.should_exclude(instance): + stats["skipped"] += 1 continue if instance["state"] == "running": + running_instances_by_region[instance["region"]].append(instance) + else: + logger.debug(f"Skipping EC2 instance {instance['id']} in state {instance['state']} (not running)") + stats["skipped"] += 1 + + timestamp = self.get_timestamp() + + # Process each region's instances in batches + for region, region_instances in running_instances_by_region.items(): + ec2_client = self.get_boto3_client("ec2", region) + if not ec2_client: + logger.error(f"Failed to create EC2 client for region {region}") + stats["errors"] += len(region_instances) + continue + + # Process instances in batches + for i in range(0, len(region_instances), self.MAX_INSTANCES_PER_API_CALL): + batch = region_instances[i:i + self.MAX_INSTANCES_PER_API_CALL] + instance_ids = [instance["id"] for instance in batch] + try: - region = instance["region"] - ec2_client = self.create_client("ec2", region) - - # Create ISO 8601 timestamp - timestamp = self.get_timestamp() - - # Tag the instance before stopping - logger.info( - f"{'[DRY RUN] Would tag' if dry_run else 'Tagging'} EC2 instance {instance['id']} with {config.get('stop_tag')}={timestamp}" - ) + # Tag all instances in batch if not dry_run: + logger.info(f"Tagging {len(instance_ids)} EC2 instances in {region}") ec2_client.create_tags( - Resources=[instance["id"]], - Tags=[{"Key": config.get("stop_tag"), "Value": timestamp}], + Resources=instance_ids, + Tags=[{"Key": config.get("stop_tag"), "Value": timestamp}] ) - - # Stop the instance - logger.info( - f"{'[DRY RUN] Would stop' if dry_run else 'Stopping'} EC2 instance {instance['id']} in region {region}" - ) + + # Stop all instances in batch if not dry_run: - ec2_client.stop_instances(InstanceIds=[instance["id"]]) - - # Log with detailed information - self.log_action(instance["id"], region, "stop", dry_run=dry_run) - success_count += 1 - + logger.info(f"Stopping {len(instance_ids)} EC2 instances in {region}") + ec2_client.stop_instances(InstanceIds=instance_ids) + else: + logger.info(f"[DRY RUN] Would stop {len(instance_ids)} EC2 instances in {region}") + + # Log individual instances for tracking purposes + for instance in batch: + self.log_action( + instance["id"], + region, + "stop", + resource_name=instance.get("name", "Unnamed"), + dry_run=dry_run + ) + + stats["success"] += len(batch) except Exception as e: - logger.error( - f"Failed to {'simulate stopping' if dry_run else 'stop'} EC2 instance {instance['id']}: {e}" - ) - error_count += 1 - - return {"success": success_count, "errors": error_count} - - def start( - self, instances: List[Dict[str, Any]], dry_run: bool = False - ) -> Dict[str, int]: - """ - Start stopped EC2 instances. - - Args: - instances: List of EC2 instance dictionaries - dry_run: If True, only simulate the action - - Returns: - Dictionary with success and error counts - """ - success_count = 0 - error_count = 0 - + logger.error(f"Batch EC2 stop operation failed in {region}: {e}") + stats["errors"] += len(batch) + + logger.info(f"EC2 stop summary: {stats['success']} stopped, {stats['skipped']} skipped, {stats['errors']} errors") + return stats + + def start(self, instances: List[Dict[str, Any]], dry_run: bool = False) -> Dict[str, int]: + """Start stopped EC2 instances in batches for efficiency.""" + # Group instances by region for batch processing + stopped_instances_by_region = defaultdict(list) + stats = {"success": 0, "errors": 0, "skipped": 0} + + # Filter instances - keep only stopped instances that should not be excluded for instance in instances: if self.should_exclude(instance): + stats["skipped"] += 1 continue if instance["state"] == "stopped": + stopped_instances_by_region[instance["region"]].append(instance) + else: + logger.debug(f"Skipping EC2 instance {instance['id']} in state {instance['state']} (not stopped)") + stats["skipped"] += 1 + + # Process each region's instances in batches + for region, region_instances in stopped_instances_by_region.items(): + ec2_client = self.get_boto3_client("ec2", region) + if not ec2_client: + logger.error(f"Failed to create EC2 client for region {region}") + stats["errors"] += len(region_instances) + continue + + # Process instances in batches + for i in range(0, len(region_instances), self.MAX_INSTANCES_PER_API_CALL): + batch = region_instances[i:i + self.MAX_INSTANCES_PER_API_CALL] + instance_ids = [instance["id"] for instance in batch] + try: - region = instance["region"] - ec2_client = self.create_client("ec2", region) - - logger.info( - f"{'[DRY RUN] Would start' if dry_run else 'Starting'} EC2 instance {instance['id']} in region {region}" - ) + # Start all instances in batch if not dry_run: - ec2_client.start_instances(InstanceIds=[instance["id"]]) - - # Remove the stop tag after successful start - logger.info( - f"Removing {config.get('stop_tag')} tag from EC2 instance {instance['id']}" - ) + logger.info(f"Starting {len(instance_ids)} EC2 instances in {region}") + ec2_client.start_instances(InstanceIds=instance_ids) + else: + logger.info(f"[DRY RUN] Would start {len(instance_ids)} EC2 instances in {region}") + + # Remove tags in batch + if not dry_run: + logger.info(f"Removing stop tag from {len(instance_ids)} EC2 instances in {region}") ec2_client.delete_tags( - Resources=[instance["id"]], - Tags=[{"Key": config.get("stop_tag")}], + Resources=instance_ids, + Tags=[{"Key": config.get("stop_tag")}] ) - - # Log with detailed information - self.log_action(instance["id"], region, "start", dry_run=dry_run) - success_count += 1 - + + # Log individual instances for tracking purposes + for instance in batch: + self.log_action( + instance["id"], + region, + "start", + resource_name=instance.get("name", "Unnamed"), + dry_run=dry_run + ) + + stats["success"] += len(batch) except Exception as e: - logger.error( - f"Failed to {'simulate starting' if dry_run else 'start'} EC2 instance {instance['id']}: {e}" - ) - error_count += 1 - - return {"success": success_count, "errors": error_count} + logger.error(f"Batch EC2 start operation failed in {region}: {e}") + stats["errors"] += len(batch) + + logger.info(f"EC2 start summary: {stats['success']} started, {stats['skipped']} skipped, {stats['errors']} errors") + return stats + + def discover_instances(self, regions: List[str]) -> List[Dict[str, Any]]: + """Discover EC2 instances across multiple regions.""" + all_instances = [] + + for region in regions: + try: + ec2_client = self.get_boto3_client('ec2', region) + if not ec2_client: + continue + + # Use pagination helper from base class + reservations = self.paginate_boto3( + ec2_client, + 'describe_instances', + 'Reservations' + ) + + # Extract and process instances + instances = [] + for reservation in reservations: + for instance in reservation.get("Instances", []): + # Convert tags to dictionary + tags = {} + for tag in instance.get("Tags", []): + tags[tag["Key"]] = tag["Value"] + + # Create a standardized instance dictionary + instances.append({ + "id": instance["InstanceId"], + "name": tags.get("Name", "Unnamed"), + "type": instance["InstanceType"], + "status": instance["State"]["Name"], + "state": instance["State"]["Name"], + "region": region, + "tags": tags, + "launch_time": instance["LaunchTime"].isoformat() if "LaunchTime" in instance else None, + "public_ip": instance.get("PublicIpAddress"), + "private_ip": instance.get("PrivateIpAddress"), + "accountId": self.account_id, + "accountName": self.account_name + }) + + logger.info(f"Found {len(instances)} EC2 instances in {region}") + all_instances.extend(instances) + + except Exception as e: + logger.error(f"Error discovering EC2 instances in {region}: {str(e)}") + + return all_instances diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/ecr.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/ecr.py new file mode 100644 index 00000000..9989b76f --- /dev/null +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/ecr.py @@ -0,0 +1,118 @@ +from typing import Dict, List, Any, Optional +import logging +from datetime import datetime, timedelta + +from aws_resource_management.managers.base import ResourceManager + +logger = logging.getLogger(__name__) + + +class ECRManager(ResourceManager): + """Manager for Amazon ECR (Elastic Container Registry) resources.""" + + def __init__(self, credentials: Dict[str, str], account_id: str, account_name: Optional[str] = None): + """Initialize the ECR resource manager.""" + super().__init__(credentials, account_id, account_name) + self.resource_type = "ecr_image" + + def start(self, resources: List[Dict[str, Any]], dry_run: bool = False) -> Dict[str, int]: + """No-op implementation as ECR images don't have a start operation.""" + return {"success": 0, "errors": 0, "skipped": len(resources)} + + def stop(self, resources: List[Dict[str, Any]], dry_run: bool = False) -> Dict[str, int]: + """No-op implementation as ECR images don't have a stop operation.""" + return {"success": 0, "errors": 0, "skipped": len(resources)} + + def discover_images_by_age(self, regions: List[str], age_threshold_days: int = 365) -> Dict[str, List[Dict[str, Any]]]: + """Discover ECR images across regions, categorized by age.""" + all_images = [] + old_images = [] + + for region in regions: + images_result = self._discover_images_in_region(region, age_threshold_days) + all_images.extend(images_result['all_images']) + old_images.extend(images_result['old_images']) + + logger.info(f"Found {len(all_images)} ECR images across {len(regions)} regions " + f"({len(old_images)} older than {age_threshold_days} days)") + + return { + 'all_images': all_images, + 'old_images': old_images + } + + def _discover_images_in_region(self, region: str, age_threshold_days: int = 365) -> Dict[str, List[Dict[str, Any]]]: + """Discover ECR images in a specific region, categorized by age.""" + ecr_client = self.get_boto3_client("ecr", region) + if not ecr_client: + return {'all_images': [], 'old_images': []} + + all_images = [] + old_images = [] + threshold_date = datetime.now() - timedelta(days=age_threshold_days) + + try: + # Get all repositories efficiently using pagination helper + repositories = self.paginate_boto3( + ecr_client, + 'describe_repositories', + 'repositories' + ) + + repository_names = [repo['repositoryName'] for repo in repositories] + logger.info(f"Found {len(repository_names)} ECR repositories in {region}") + + # For each repository, efficiently get all image IDs + for repo_name in repository_names: + try: + # Get image IDs using pagination helper + image_ids = self.paginate_boto3( + ecr_client, + 'list_images', + 'imageIds', + repositoryName=repo_name + ) + + # Process images in chunks due to API limitations + chunk_size = 100 + for i in range(0, len(image_ids), chunk_size): + chunk = image_ids[i:i + chunk_size] + if not chunk: + continue + + try: + # Get details for all images in this chunk + response = ecr_client.describe_images( + repositoryName=repo_name, + imageIds=chunk + ) + + # Process each image + for image_detail in response.get('imageDetails', []): + # Add metadata + image_detail['repositoryName'] = repo_name + image_detail['region'] = region + image_detail['accountId'] = self.account_id + if self.account_name: + image_detail['accountName'] = self.account_name + + all_images.append(image_detail) + + # Check if image is older than threshold + if ('imagePushedAt' in image_detail and + image_detail['imagePushedAt'] < threshold_date): + old_images.append(image_detail) + except Exception as e: + logger.error(f"Error describing images for repository {repo_name} chunk {i//chunk_size + 1}: {e}") + except Exception as e: + logger.error(f"Error processing repository {repo_name}: {e}") + + logger.info(f"Found {len(all_images)} ECR images in {region} ({len(old_images)} older than {age_threshold_days} days)") + + except Exception as e: + logger.error(f"Error discovering ECR images in {region}: {e}") + + return { + 'all_images': all_images, + 'old_images': old_images + } diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/eks.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/eks.py index dcc12aca..3a9913b7 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/eks.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/eks.py @@ -503,3 +503,71 @@ def _scale_nodegroups( logger.error( f"Error listing nodegroups for cluster {cluster_name} in region {region}: {e}" ) + + def discover_clusters(self, regions: List[str]) -> List[Dict[str, Any]]: + """ + Discover EKS clusters across multiple regions. + + Args: + regions: List of AWS regions to scan + + Returns: + List of EKS cluster dictionaries + """ + all_clusters = [] + + for region in regions: + try: + eks_client = self.get_boto3_client('eks', region) + if not eks_client: + logger.warning(f"Could not create EKS client in {region}") + continue + + # Get clusters + response = eks_client.list_clusters() + cluster_names = response.get("clusters", []) + + # Handle pagination + while "nextToken" in response: + response = eks_client.list_clusters(nextToken=response["nextToken"]) + cluster_names.extend(response.get("clusters", [])) + + # Get detailed information for each cluster + clusters = [] + for name in cluster_names: + try: + # Get cluster details + cluster = eks_client.describe_cluster(name=name)["cluster"] + + # Get tags + tags = cluster.get("tags", {}) + + # Create a simplified cluster dictionary + cluster_dict = { + "id": name, + "name": name, + "arn": cluster.get("arn", f"arn:aws:eks:{region}:{self.account_id}:cluster/{name}"), + "status": cluster.get("status", "UNKNOWN"), + "state": cluster.get("status", "UNKNOWN"), # Add both for consistency + "region": region, + "tags": tags, + "version": cluster.get("version"), + "endpoint": cluster.get("endpoint"), + "created_at": cluster.get("createdAt"), + "accountId": self.account_id, + } + if self.account_name: + cluster_dict["accountName"] = self.account_name + + clusters.append(cluster_dict) + except Exception as e: + logger.warning(f"Error getting details for EKS cluster {name}: {e}") + + logger.info(f"Found {len(clusters)} EKS clusters in {region}") + all_clusters.extend(clusters) + + except Exception as e: + logger.error(f"Error discovering EKS clusters in {region}: {str(e)}") + + logger.info(f"Found a total of {len(all_clusters)} EKS clusters across all regions") + return all_clusters diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/emr.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/emr.py index dcb6cad7..20c091d9 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/emr.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/emr.py @@ -33,108 +33,83 @@ def __init__( super().__init__(credentials, account_id, account_name) self.resource_type = "emr_cluster" - def discover_clusters( - self, region: str, cluster_states: List[str] = None - ) -> List[Dict[str, Any]]: + def discover_clusters(self, regions: List[str], cluster_states: Optional[List[str]] = None) -> List[Dict[str, Any]]: """ - Discover EMR clusters in the given region with the specified states. - + Discover EMR clusters across multiple regions. + Args: - region: AWS region - cluster_states: List of EMR cluster states to filter by - + regions: List of AWS regions to scan + cluster_states: List of cluster states to include (optional) + Returns: List of EMR cluster dictionaries """ - try: - # Default states if not provided - EMR API only accepts specific states - if not cluster_states: - cluster_states = [ - "STARTING", - "BOOTSTRAPPING", - "RUNNING", - "WAITING", - "TERMINATING", - ] - - logger.debug( - f"Discovering EMR clusters in {region} with states: {', '.join(cluster_states)}" - ) - - emr_client = self.create_client("emr", region) - clusters = [] - - # ListClusters API only returns a subset of data, with pagination - paginator = emr_client.get_paginator("list_clusters") - page_iterator = paginator.paginate(ClusterStates=cluster_states) - - for page in page_iterator: - if "Clusters" in page: - for cluster_summary in page["Clusters"]: - # Get detailed info for each cluster - try: - cluster_id = cluster_summary.get("Id") - if not cluster_id: - logger.warning("Skipping cluster with missing ID") - continue - - # Format basic info - cluster = { - "id": cluster_id, - "name": cluster_summary.get("Name", "Unknown"), - "status": cluster_summary.get("Status", {}).get( - "State", "UNKNOWN" - ), - "region": region, - "account_id": self.account_id, - "account_name": self.account_name, - } - - # Get tags - try: - response = emr_client.describe_cluster( - ClusterId=cluster_id - ) - if ( - "Cluster" in response - and "Tags" in response["Cluster"] - ): - tags = {} - for tag in response["Cluster"]["Tags"]: - tags[tag.get("Key")] = tag.get("Value") - cluster["tags"] = tags - except Exception as e: - logger.warning( - f"Could not get tags for cluster {cluster_id}: {str(e)}" - ) - cluster["tags"] = {} - - clusters.append(cluster) - - except Exception as e: - logger.warning(f"Error processing EMR cluster: {str(e)}") - - logger.info(f"Discovered {len(clusters)} EMR clusters in {region}") - return clusters - - except ClientError as e: - error_code = e.response.get("Error", {}).get("Code") - if ( - error_code == "AccessDeniedException" - or error_code == "UnauthorizedOperation" - ): - logger.warning(f"Access denied to EMR in region {region}: {str(e)}") - elif ( - error_code == "EndpointConnectionError" - or error_code == "UnknownEndpoint" - ): - logger.debug(f"EMR not supported in region {region}") - else: + all_clusters = [] + + # Default cluster states if not provided + if cluster_states is None: + cluster_states = ["STARTING", "BOOTSTRAPPING", "RUNNING", "WAITING", "TERMINATING"] + + for region in regions: + try: + emr_client = self.get_boto3_client('emr', region) + if not emr_client: + logger.warning(f"Could not create EMR client in {region}") + continue + + # Get clusters + response = emr_client.list_clusters(ClusterStates=cluster_states) + clusters = response.get("Clusters", []) + + # Handle pagination + while "Marker" in response: + response = emr_client.list_clusters( + Marker=response["Marker"], + ClusterStates=cluster_states + ) + clusters.extend(response.get("Clusters", [])) + + # Process each cluster + processed_clusters = [] + for cluster in clusters: + try: + # Get cluster details + cluster_details = emr_client.describe_cluster(ClusterId=cluster["Id"]) + cluster_info = cluster_details.get("Cluster", {}) + + # Convert tags + tags = {} + for tag in cluster_info.get("Tags", []): + tags[tag.get("Key", "")] = tag.get("Value", "") + + # Create a simplified cluster dictionary + cluster_dict = { + "id": cluster["Id"], + "name": cluster.get("Name", "Unnamed"), + "status": cluster.get("Status", {}).get("State", "UNKNOWN"), + "state": cluster.get("Status", {}).get("State", "UNKNOWN"), + "region": region, + "tags": tags, + "creation_time": cluster.get("Status", {}).get("Timeline", {}).get("CreationDateTime"), + "termination_time": cluster.get("Status", {}).get("Timeline", {}).get("EndDateTime"), + "cluster_type": cluster_info.get("InstanceCollectionType", "Unknown"), + "accountId": self.account_id, + } + if self.account_name: + cluster_dict["accountName"] = self.account_name + + processed_clusters.append(cluster_dict) + except Exception as e: + logger.warning(f"Error getting details for EMR cluster {cluster['Id']}: {e}") + + logger.info(f"Found {len(processed_clusters)} EMR clusters in {region}") + all_clusters.extend(processed_clusters) + + except Exception as e: logger.error(f"Error discovering EMR clusters in {region}: {str(e)}") - return [] - except Exception as e: - logger.error(f"Error discovering EMR clusters in {region}: {str(e)}") - return [] + + logger.info(f"Found a total of {len(all_clusters)} EMR clusters across all regions") + return all_clusters def discover_all_clusters(self, regions: List[str]) -> List[Dict[str, Any]]: """ diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/rds.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/rds.py index 89885c0c..6127b1a3 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/rds.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/rds.py @@ -3,6 +3,7 @@ """ from typing import Any, Dict, List, Optional +from collections import defaultdict from aws_resource_management.config_manager import get_config from aws_resource_management.logging_setup import setup_logging @@ -21,147 +22,204 @@ def __init__( account_id: str, account_name: Optional[str] = None, ): - """ - Initialize RDS manager. - - Args: - credentials: AWS credentials dictionary - account_id: AWS account ID - account_name: AWS account name (optional) - """ + """Initialize RDS manager.""" super().__init__(credentials, account_id, account_name) self.resource_type = "rds_instance" + self.MAX_DB_OPERATIONS = 20 # Conservative limit for RDS batch operations def stop( self, instances: List[Dict[str, Any]], dry_run: bool = False ) -> Dict[str, int]: - """ - Stop available RDS instances. - - Args: - instances: List of RDS instance dictionaries - dry_run: If True, only simulate the action - - Returns: - Dictionary with success and error counts - """ - logger.info(f"Evaluating {len(instances)} RDS instances for stop action") - stopped_count = 0 - skipped_count = 0 - error_count = 0 - + """Stop available RDS instances in batches where possible.""" + # Group instances by region + available_instances_by_region = defaultdict(list) + stats = {"success": 0, "errors": 0, "skipped": 0} + + # Filter instances - keep only those in 'available' state and not excluded for instance in instances: - # Skip instances with the exclusion tag if self.should_exclude(instance): - logger.info( - f"Skipping RDS instance {instance['id']} with exclusion tag {config.get('exclusion_tag')}" - ) - skipped_count += 1 + stats["skipped"] += 1 continue - - # Log status for debugging - logger.debug(f"RDS instance {instance['id']} status: {instance['status']}") - + if instance["status"] == "available": - try: - region = instance["region"] - rds_client = self.create_client("rds", region) - - # Create ISO 8601 timestamp - timestamp = self.get_timestamp() - - # Tag the instance before stopping - logger.info( - f"{'[DRY RUN] Would tag' if dry_run else 'Tagging'} RDS instance {instance['id']} with {config.get('stop_tag')}={timestamp}" - ) - if not dry_run: - rds_client.add_tags_to_resource( - ResourceName=f"arn:{self.get_partition_for_region(region)}:rds:{region}:{self.account_id}:db:{instance['id']}", - Tags=[{"Key": config.get("stop_tag"), "Value": timestamp}], - ) - - logger.info( - f"{'[DRY RUN] Would stop' if dry_run else 'Stopping'} RDS instance {instance['id']} in region {region}" - ) - if not dry_run: - rds_client.stop_db_instance(DBInstanceIdentifier=instance["id"]) - - # Log with detailed information - self.log_action(instance["id"], region, "stop", dry_run=dry_run) - - stopped_count += 1 - - except Exception as e: - logger.error( - f"Failed to {'simulate stopping' if dry_run else 'stop'} RDS instance {instance['id']}: {e}" - ) - error_count += 1 + available_instances_by_region[instance["region"]].append(instance) else: - logger.debug( - f"Skipping RDS instance {instance['id']} with non-available status: {instance['status']}" - ) - skipped_count += 1 - - logger.info( - f"RDS stop summary: {stopped_count} instances processed, {skipped_count} instances skipped" - ) + logger.debug(f"Skipping RDS instance {instance['id']} in state {instance['status']} (not available)") + stats["skipped"] += 1 + + timestamp = self.get_timestamp() + + # Process each region's instances + for region, region_instances in available_instances_by_region.items(): + rds_client = self.get_boto3_client("rds", region) + if not rds_client: + logger.error(f"Failed to create RDS client for region {region}") + stats["errors"] += len(region_instances) + continue + + # Process instances in batches + for i in range(0, len(region_instances), self.MAX_DB_OPERATIONS): + batch = region_instances[i:i + self.MAX_DB_OPERATIONS] + + # Unfortunately, RDS doesn't support bulk tagging or stopping + # Process each instance individually but within a batch context + for instance in batch: + try: + db_id = instance["id"] + arn = f"arn:{self.get_partition_for_region(region)}:rds:{region}:{self.account_id}:db:{db_id}" + + # Tag the instance before stopping + if not dry_run: + logger.debug(f"Tagging RDS instance {db_id}") + rds_client.add_tags_to_resource( + ResourceName=arn, + Tags=[{"Key": config.get("stop_tag"), "Value": timestamp}], + ) + + logger.info(f"Stopping RDS instance {db_id} in {region}") + rds_client.stop_db_instance(DBInstanceIdentifier=db_id) + else: + logger.info(f"[DRY RUN] Would stop RDS instance {db_id} in {region}") + + self.log_action( + db_id, + region, + "stop", + resource_name=instance.get("name"), + dry_run=dry_run + ) + stats["success"] += 1 + except Exception as e: + logger.error(f"Failed to stop RDS instance {instance['id']}: {e}") + stats["errors"] += 1 - return {"success": stopped_count, "errors": error_count} + logger.info(f"RDS stop summary: {stats['success']} stopped, {stats['skipped']} skipped, {stats['errors']} errors") + return stats def start( self, instances: List[Dict[str, Any]], dry_run: bool = False ) -> Dict[str, int]: - """ - Start stopped RDS instances. - - Args: - instances: List of RDS instance dictionaries - dry_run: If True, only simulate the action - - Returns: - Dictionary with success and error counts - """ - started_count = 0 - error_count = 0 - + """Start stopped RDS instances in batches where possible.""" + # Group instances by region + stopped_instances_by_region = defaultdict(list) + stats = {"success": 0, "errors": 0, "skipped": 0} + + # Filter instances - keep only those in 'stopped' state and not excluded for instance in instances: - # Skip instances with the exclusion tag if self.should_exclude(instance): - logger.info( - f"Skipping RDS instance {instance['id']} with exclusion tag {config.get('exclusion_tag')}" - ) + stats["skipped"] += 1 continue - + if instance["status"] == "stopped": - try: - region = instance["region"] - rds_client = self.create_client("rds", region) - - logger.info( - f"{'[DRY RUN] Would start' if dry_run else 'Starting'} RDS instance {instance['id']} in region {region}" - ) - if not dry_run: - rds_client.start_db_instance( - DBInstanceIdentifier=instance["id"] - ) - - # Remove the stop tag after successful start - logger.info( - f"Removing {config.get('stop_tag')} tag from RDS instance {instance['id']}" + stopped_instances_by_region[instance["region"]].append(instance) + else: + logger.debug(f"Skipping RDS instance {instance['id']} in state {instance['status']} (not stopped)") + stats["skipped"] += 1 + + # Process each region's instances + for region, region_instances in stopped_instances_by_region.items(): + rds_client = self.get_boto3_client("rds", region) + if not rds_client: + logger.error(f"Failed to create RDS client for region {region}") + stats["errors"] += len(region_instances) + continue + + # Process instances in batches + for i in range(0, len(region_instances), self.MAX_DB_OPERATIONS): + batch = region_instances[i:i + self.MAX_DB_OPERATIONS] + + # Unfortunately, RDS doesn't support bulk operations + for instance in batch: + try: + db_id = instance["id"] + arn = f"arn:{self.get_partition_for_region(region)}:rds:{region}:{self.account_id}:db:{db_id}" + + # Start the instance + if not dry_run: + logger.info(f"Starting RDS instance {db_id} in {region}") + rds_client.start_db_instance(DBInstanceIdentifier=db_id) + + # Remove the stop tag + logger.debug(f"Removing stop tag from RDS instance {db_id}") + rds_client.remove_tags_from_resource( + ResourceName=arn, + TagKeys=[config.get("stop_tag")], + ) + else: + logger.info(f"[DRY RUN] Would start RDS instance {db_id} in {region}") + + self.log_action( + db_id, + region, + "start", + resource_name=instance.get("name"), + dry_run=dry_run ) - rds_client.remove_tags_from_resource( - ResourceName=f"arn:{self.get_partition_for_region(region)}:rds:{region}:{self.account_id}:db:{instance['id']}", - TagKeys=[config.get("stop_tag")], + stats["success"] += 1 + except Exception as e: + logger.error(f"Failed to start RDS instance {instance['id']}: {e}") + stats["errors"] += 1 + + logger.info(f"RDS start summary: {stats['success']} started, {stats['skipped']} skipped, {stats['errors']} errors") + return stats + + def discover_instances(self, regions: List[str]) -> List[Dict[str, Any]]: + """Discover RDS instances across regions using efficient pagination.""" + all_instances = [] + + for region in regions: + try: + rds_client = self.get_boto3_client('rds', region) + if not rds_client: + continue + + # Use pagination helper from base class + db_instances = self.paginate_boto3( + rds_client, + 'describe_db_instances', + 'DBInstances' + ) + + # Process each instance + instances = [] + for db in db_instances: + # Get tags efficiently + try: + tags_response = rds_client.list_tags_for_resource( + ResourceName=db["DBInstanceArn"] ) - - # Log with detailed information - self.log_action(instance["id"], region, "start", dry_run=dry_run) - started_count += 1 - - except Exception as e: - logger.error( - f"Failed to {'simulate starting' if dry_run else 'start'} RDS instance {instance['id']}: {e}" - ) - error_count += 1 - - return {"success": started_count, "errors": error_count} + tags = { + item["Key"]: item["Value"] + for item in tags_response.get("TagList", []) + } + except Exception as e: + logger.warning(f"Error getting tags for RDS instance {db['DBInstanceIdentifier']}: {e}") + tags = {} + + # Create standardized instance dictionary + instances.append({ + "id": db["DBInstanceIdentifier"], + "name": db["DBInstanceIdentifier"], + "engine": db["Engine"], + "engine_version": db.get("EngineVersion"), + "status": db["DBInstanceStatus"], + "state": db["DBInstanceStatus"], + "region": region, + "tags": tags, + "instance_class": db.get("DBInstanceClass"), + "storage": db.get("AllocatedStorage"), + "multi_az": db.get("MultiAZ", False), + "public": db.get("PubliclyAccessible", False), + "endpoint": db.get("Endpoint", {}).get("Address"), + "port": db.get("Endpoint", {}).get("Port"), + "accountId": self.account_id, + "accountName": self.account_name + }) + + logger.info(f"Found {len(instances)} RDS instances in {region}") + all_instances.extend(instances) + + except Exception as e: + logger.error(f"Error discovering RDS instances in {region}: {str(e)}") + + return all_instances diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/reporting.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/reporting.py new file mode 100644 index 00000000..dce2fc3e --- /dev/null +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/reporting.py @@ -0,0 +1,135 @@ +""" +Reporting functionality for AWS resource management. +""" + +import logging +from typing import Dict, Any, List + +logger = logging.getLogger(__name__) + +def print_resource_summary(stats: Dict[str, Any], action: str, dry_run: bool = False) -> None: + """ + Print a detailed summary of resources processed, broken down by resource type. + + Args: + stats: Statistics dictionary + action: Action that was performed (stop or start) + dry_run: Whether this was a dry run + """ + logger.info(f"\n{'=' * 30} SUMMARY {'=' * 30}") + logger.info( + f"ACTION: {action.upper()} {'(DRY RUN)' if dry_run else ''}" + ) + + # General stats + logger.info(f"\nGENERAL STATISTICS:") + logger.info(f"Accounts processed: {stats.get('accounts_processed', 0)}") + logger.info(f"Accounts skipped: {stats.get('accounts_skipped', 0)}") + logger.info(f"Regions processed: {stats.get('regions_processed', 0)}") + + # EC2 resources + logger.info(f"\nEC2 INSTANCES:") + logger.info(f"Found: {stats.get('ec2_found', 0)}") + if action == "stop": + logger.info(f"Stopped: {stats.get('ec2_stopped', 0)}") + else: + logger.info(f"Started: {stats.get('ec2_started', 0)}") + logger.info(f"Skipped: {stats.get('ec2_skipped', 0)}") + logger.info(f"Errors: {stats.get('ec2_errors', 0)}") + + # RDS resources + logger.info(f"\nRDS INSTANCES:") + logger.info(f"Found: {stats.get('rds_found', 0)}") + if action == "stop": + logger.info(f"Stopped: {stats.get('rds_stopped', 0)}") + else: + logger.info(f"Started: {stats.get('rds_started', 0)}") + logger.info(f"Skipped: {stats.get('rds_skipped', 0)}") + logger.info(f"Errors: {stats.get('rds_errors', 0)}") + + # RDS engine breakdown if available + rds_engines = stats.get("rds_engines", {}) + if rds_engines: + logger.info( + f"RDS engines: {', '.join(f'{engine}({count})' for engine, count in rds_engines.items())}" + ) + + # EKS resources + logger.info(f"\nEKS CLUSTERS:") + logger.info(f"Found: {stats.get('eks_found', 0)}") + if action == "stop": + logger.info(f"Stopped: {stats.get('eks_stopped', 0)}") + else: + logger.info(f"Started: {stats.get('eks_started', 0)}") + logger.info(f"Skipped: {stats.get('eks_skipped', 0)}") + logger.info(f"Errors: {stats.get('eks_errors', 0)}") + + # EMR resources + logger.info(f"\nEMR CLUSTERS:") + logger.info(f"Found: {stats.get('emr_found', 0)}") + if action == "stop": + logger.info(f"Stopped: {stats.get('emr_stopped', 0)}") + else: + logger.info(f"Started: {stats.get('emr_started', 0)}") + logger.info(f"Skipped: {stats.get('emr_skipped', 0)}") + logger.info(f"Errors: {stats.get('emr_errors', 0)}") + + # ECR images + logger.info(f"\nECR IMAGES:") + logger.info(f"Total images found: {stats.get('ecr_images_found', 0)}") + logger.info(f"Images older than 1 year: {stats.get('ecr_old_images_found', 0)}") + + # Error summary if there were any errors + print_error_summary(stats.get("errors", [])) + + logger.info(f"{'=' * 68}") + +def print_error_summary(errors: List[str]) -> None: + """ + Print a summary of errors that occurred during processing. + + Args: + errors: List of error messages + """ + if errors: + logger.info(f"\nERROR SUMMARY:") + logger.info(f"Total errors: {len(errors)}") + for i, error in enumerate(errors[:5], 1): # Show first 5 errors + logger.info(f" {i}. {error}") + if len(errors) > 5: + logger.info( + f" ... and {len(errors) - 5} more errors (see log for details)" + ) + +def initialize_stats() -> Dict[str, Any]: + """ + Initialize a statistics dictionary with default values. + + Returns: + A dictionary with initialized statistics counters + """ + stats = { + "accounts_processed": 0, + "accounts_skipped": 0, + "resources_processed": 0, + "resources_skipped": 0, + "regions_processed": 0, + "regions_by_account": {}, + "errors": [], + "rds_engines": {}, + "ecr_images_found": 0, + "ecr_old_images_found": 0, + } + + # Initialize resource-specific counters + for resource_type in ["ecr", "ec2", "rds", "eks", "emr"]: + for stat_type in [ + "found", + "skipped", + "stopped", + "started", + "errors", + ]: + stats[f"{resource_type}_{stat_type}"] = 0 + + return stats