From 059197149e606d0b4a84e683a33960757e6cbbaf Mon Sep 17 00:00:00 2001 From: "Matthew C. Morgan" Date: Thu, 3 Apr 2025 19:16:49 -0400 Subject: [PATCH] outputs csv --- .../gfl-resource-actions/Makefile | 2 +- .../aws_resource_management/__main__.py | 10 - .../aws_resource_management/aws_utils.py | 100 ++--- .../aws_resource_management/cli.py | 109 +++++- .../aws_resource_management/core.py | 240 ++++++++++-- .../aws_resource_management/csv_utils.py | 106 +++++ .../aws_resource_management/discovery.py | 258 ++++++------ .../discovery_utils.py | 169 ++++++++ .../aws_resource_management/logging_setup.py | 102 ++--- .../aws_resource_management/managers/base.py | 191 ++++++--- .../aws_resource_management/managers/ec2.py | 146 ++++--- .../aws_resource_management/managers/ecr.py | 370 ++++++++++++------ .../aws_resource_management/managers/emr.py | 6 +- .../partition_utils.py | 172 ++++++++ .../aws_resource_management/reporting.py | 98 ++++- .../resource_registry.py | 197 ++++++++++ .../aws_resource_management/utils.py | 299 +------------- 17 files changed, 1715 insertions(+), 860 deletions(-) delete mode 100644 local-app/python-tools/gfl-resource-actions/aws_resource_management/__main__.py create mode 100644 local-app/python-tools/gfl-resource-actions/aws_resource_management/csv_utils.py create mode 100644 local-app/python-tools/gfl-resource-actions/aws_resource_management/discovery_utils.py create mode 100644 local-app/python-tools/gfl-resource-actions/aws_resource_management/partition_utils.py create mode 100644 local-app/python-tools/gfl-resource-actions/aws_resource_management/resource_registry.py diff --git a/local-app/python-tools/gfl-resource-actions/Makefile b/local-app/python-tools/gfl-resource-actions/Makefile index 4e61d5ee..3949e596 100644 --- a/local-app/python-tools/gfl-resource-actions/Makefile +++ b/local-app/python-tools/gfl-resource-actions/Makefile @@ -75,7 +75,7 @@ dist: # Run in dry-run mode run-dry-run: - $(PYTHON) -m $(PACKAGE_NAME).cli --stop --dry-run --resource-type all + $(PYTHON) -m $(PACKAGE_NAME).cli --stop --dry-run --resource-type ecr --csv-output-dir ./ --csv-export # Run to stop resources run-stop: diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/__main__.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/__main__.py deleted file mode 100644 index 42096049..00000000 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/__main__.py +++ /dev/null @@ -1,10 +0,0 @@ -""" -Main entry point for AWS Resource Management CLI. -""" - -import sys - -from aws_resource_management.cli_optimized import main - -if __name__ == "__main__": - sys.exit(main()) diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/aws_utils.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/aws_utils.py index adf8900f..08f9e2a8 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/aws_utils.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/aws_utils.py @@ -14,6 +14,16 @@ import boto3 import botocore from aws_resource_management.logging_setup import setup_logging +from aws_resource_management.partition_utils import ( + DEFAULT_PARTITION, + DEFAULT_REGIONS, + PARTITION_REGIONS, + get_default_regions_for_partition, + get_partition_for_region, + get_regions_for_partition, + is_region_in_partition, + is_valid_region, +) from botocore.exceptions import ClientError, NoCredentialsError, ProfileNotFound logger = setup_logging() @@ -30,24 +40,8 @@ "timestamp": 0, "all_regions": {}, "enabled_regions": {}, - "partition_regions": { - "aws-us-gov": ["us-gov-east-1", "us-gov-west-1"], - "aws-cn": ["cn-north-1", "cn-northwest-1"], - "aws": [ - "us-east-1", - "us-east-2", - "us-west-1", - "us-west-2", - "eu-west-1", - "eu-central-1", - "eu-west-2", - "eu-west-3", - "ap-northeast-1", - "ap-northeast-2", - "ap-southeast-1", - "ap-southeast-2", - ], - }, + # Use PARTITION_REGIONS from partition_utils for partition region data + "partition_regions": PARTITION_REGIONS, } # Cache expiration time in seconds (1 hour) @@ -91,33 +85,9 @@ def safe_in(substring: Any, container: Any) -> bool: # Simplified region and partition handling # --------------------------------------------------------------------------- - -@lru_cache(maxsize=128) -def get_partition_for_region(region_name: str) -> str: - """ - Determine AWS partition based on region name. - - Args: - region_name: AWS region name - - Returns: - AWS partition name (e.g., 'aws', 'aws-us-gov', 'aws-cn') - """ - if region_name.startswith("us-gov-"): - return "aws-us-gov" - elif region_name.startswith("cn-"): - return "aws-cn" - return "aws" - - -def get_regions_for_partition(partition: str) -> List[str]: - """Get regions for a partition using cached data instead of API calls.""" - if partition in _region_cache["partition_regions"]: - return _region_cache["partition_regions"][partition] - - # Default to GovCloud - return _region_cache["partition_regions"]["aws-us-gov"] - +# Export partition functions directly from partition_utils for backwards compatibility +# These explicit re-exports make it clearer that they're from partition_utils +# rather than just happening to have the same name def get_all_regions(partition: Optional[str] = None) -> List[str]: """Get all regions, using cache to minimize API calls.""" @@ -159,7 +129,7 @@ def get_all_regions(partition: Optional[str] = None) -> List[str]: return all_regions except Exception as e: logger.warning(f"Error getting regions: {str(e)}") - return get_regions_for_partition(partition or "aws-us-gov") + return get_regions_for_partition(partition or DEFAULT_PARTITION) # --------------------------------------------------------------------------- @@ -299,7 +269,7 @@ def _assume_role_for_account(account_id: str) -> Optional[Dict[str, Any]]: def detect_partition_from_credentials(credentials: Dict[str, str]) -> str: """Detect partition from credentials with minimal API calls.""" if not credentials: - return "aws-us-gov" # Default for environment + return DEFAULT_PARTITION # Default for environment from partition_utils # Check credential format first (fast, no API calls) access_key = str(credentials.get("aws_access_key_id", "")).lower() @@ -342,7 +312,7 @@ def detect_partition_from_credentials(credentials: Dict[str, str]) -> str: return "aws" except Exception: # Default to GovCloud for environment - return "aws-us-gov" + return DEFAULT_PARTITION # --------------------------------------------------------------------------- @@ -644,8 +614,8 @@ def get_enabled_regions( return regions except Exception: - # Fall back to default regions - return get_regions_for_partition(partition) + # Fall back to default regions from partition_utils + return get_default_regions_for_partition(partition) def list_enabled_regions( @@ -689,29 +659,27 @@ def list_enabled_regions( return ["us-east-1", "us-east-2", "us-west-1", "us-west-2"] +# Replace is_valid_region with import from partition_utils def is_valid_region(region: str) -> bool: - """Check if a region is valid without making an API call.""" - if not is_string(region): - return False - - # Check against our cached region lists - for partition, regions in _region_cache["partition_regions"].items(): - if region in regions: - return True - - # If not in predefined lists, make an API call as last resort - try: - boto3.session.Session().client("ec2", region_name=region) - return True - except: - return False + """ + Check if a region is valid without making an API call. + Legacy function - delegates to partition_utils.is_valid_region. + + Args: + region: Region name to check + + Returns: + True if the region is valid, False otherwise + """ + from aws_resource_management.partition_utils import is_valid_region as _is_valid_region + return _is_valid_region(region) # --------------------------------------------------------------------------- # Function aliases for backward compatibility # --------------------------------------------------------------------------- -# Alias get_enabled_regions as get_available_regions for backward compatibility +# Explicitly document this is an alias for get_enabled_regions get_available_regions = get_enabled_regions diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/cli.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/cli.py index 3e0a040c..291fdb78 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/cli.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/cli.py @@ -5,6 +5,7 @@ import argparse import logging +import os import sys import traceback from typing import Any, Dict, List, Optional @@ -12,6 +13,8 @@ from aws_resource_management.config_manager import get_config from aws_resource_management.core import ResourceManager from aws_resource_management.logging_setup import setup_logging +from aws_resource_management.reporting import export_resources_to_csv +from aws_resource_management.resource_registry import get_registry logger = setup_logging() config = get_config() @@ -25,13 +28,28 @@ def parse_args(): action_group = parser.add_mutually_exclusive_group(required=True) action_group.add_argument("--stop", action="store_true", help="Stop resources") action_group.add_argument("--start", action="store_true", help="Start resources") - - # Resource selection + action_group.add_argument("--export", action="store_true", help="Export resources to CSV without taking any action") + + # Resource type argument using registry + registry = get_registry() + resource_choices = registry.get_resource_type_choices() + resource_type_choices = [choice['value'] for choice in resource_choices] + + # Ensure 'all' is in the list of choices + if 'all' not in resource_type_choices: + resource_type_choices.append('all') + + # Format the help text with choice descriptions + resource_type_help = "\n".join( + f" {choice['value']}: {choice['help']}" for choice in resource_choices + ) + resource_type_help += "\n all: All supported resource types" + parser.add_argument( "--resource-type", - choices=["ecr", "ec2", "rds", "eks", "emr", "all"], + choices=resource_type_choices, default="all", - help="Resource type to manage (default: all)", + help=f"Resource type to manage. Available types:\n{resource_type_help}", ) # Account filtering @@ -82,6 +100,23 @@ def parse_args(): action="store_true", help="Show what would be done without making changes", ) + + # CSV Export options + csv_group = parser.add_argument_group('CSV Export Options') + csv_group.add_argument( + "--csv-export", + action="store_true", + help="Export discovered resources to CSV", + ) + csv_group.add_argument( + "--csv-output-dir", + help="Directory to save CSV files (default: current directory)", + default=os.getcwd(), + ) + csv_group.add_argument( + "--csv-prefix", + help="Prefix for CSV filenames", + ) return parser.parse_args() @@ -92,17 +127,25 @@ def main(): try: # Determine action - action = "stop" if args.stop else "start" + action = "stop" if args.stop else "start" if args.start else "export" + # Get registry of available resource types + registry = get_registry() + all_resource_types = registry.get_resource_types() + # Determine resource types to process exclude_resources = [] if args.resource_type != "all": # If specific resource type is selected, exclude all others - all_resource_types = ["ecr", "ec2", "rds", "eks", "emr"] exclude_resources = [ rt for rt in all_resource_types if rt != args.resource_type ] + # Validate that the requested resource type exists in the registry + if args.resource_type != "all" and args.resource_type not in all_resource_types: + logger.error(f"Invalid resource type: {args.resource_type}. Available types: {', '.join(all_resource_types)}") + sys.exit(1) + logger.info( f"Processing {args.resource_type} resources in {args.regions or 'all enabled'} regions for {action} action" ) @@ -123,17 +166,49 @@ def main(): logger.info(f"Processing only account {args.account}") # We'll handle this by post-filtering the account list in resource_manager - # Process accounts - single scan for all resource types - stats = resource_manager.process_accounts( - action=action, - regions=args.regions or [], - exclude_accounts=exclude_accounts, - exclude_resources=exclude_resources, - exclude_regions=args.exclude_regions or [], - dry_run=args.dry_run, - ) - - logger.info(f"Completed {action} action for {args.resource_type} resources") + # For export-only action or when CSV export is requested + if action == "export" or args.csv_export: + # Discover resources across all accounts and regions + discovered_resources = resource_manager.discover_resources( + regions=args.regions or [], + exclude_accounts=exclude_accounts, + exclude_resources=exclude_resources, + exclude_regions=args.exclude_regions or [], + specific_account=args.account + ) + + # Export to CSV + logger.info(f"Exporting resources to CSV in directory: {args.csv_output_dir}") + csv_files = export_resources_to_csv( + resources=discovered_resources, + output_dir=args.csv_output_dir, + prefix=args.csv_prefix + ) + + # Log success message with file locations + if csv_files: + logger.info(f"Successfully exported {len(csv_files)} resource type(s) to CSV") + for resource_type, filepath in csv_files.items(): + logger.info(f" - {resource_type}: {filepath}") + else: + logger.warning("No resources were exported to CSV. Check if resources were found.") + + # If action was export-only, we're done + if action == "export": + sys.exit(0) + + # Process accounts for start/stop actions + if action in ["start", "stop"]: + stats = resource_manager.process_accounts( + action=action, + regions=args.regions or [], + exclude_accounts=exclude_accounts, + exclude_resources=exclude_resources, + exclude_regions=args.exclude_regions or [], + dry_run=args.dry_run, + ) + + logger.info(f"Completed {action} action for {args.resource_type} resources") # Exit with success code sys.exit(0) 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 0e4ffcde..9e3eb1b2 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 @@ -8,7 +8,6 @@ from collections import defaultdict from typing import Any, Dict, List, Optional, Tuple, Union -from aws_resource_management import utils from aws_resource_management.aws_utils import ( detect_partition_from_credentials, get_account_list, @@ -19,13 +18,13 @@ from aws_resource_management.config_manager import get_config from aws_resource_management.discovery import get_account_resources from aws_resource_management.logging_setup import setup_logging -from aws_resource_management.managers import ( - EC2Manager, - EKSManager, - EMRManager, - RDSManager, +from aws_resource_management.partition_utils import ( + DEFAULT_PARTITION, + filter_regions_for_partition, + get_default_regions_for_partition, ) from aws_resource_management.reporting import initialize_stats, print_resource_summary +from aws_resource_management.resource_registry import get_registry from botocore.exceptions import ClientError logger = setup_logging() @@ -35,9 +34,6 @@ class ResourceManager: """Main resource manager that orchestrates operations across accounts and resources.""" - # Resource type constants for cleaner checking - RESOURCE_TYPES = ["ec2", "rds", "eks", "emr", "ecr"] - def __init__( self, profile_name: Optional[str] = None, @@ -54,6 +50,13 @@ def __init__( self.partition = partition self.region_cache = {} self.MAX_CONCURRENT_ACCOUNTS = 3 # Limit concurrent account processing + # Get resource types from registry + self.resource_registry = get_registry() + + @property + def RESOURCE_TYPES(self) -> List[str]: + """Get list of available resource types from registry.""" + return self.resource_registry.get_resource_types() def process_accounts( self, @@ -304,10 +307,14 @@ def _get_account_regions( logger.info(f"Discovering enabled regions for account {account_id}") discovered_regions = get_available_regions( credentials, self.partition - ) # Changed from get_enabled_regions - account_regions = [ - r for r in discovered_regions if r not in exclude_regions - ] + ) + # Filter discovered regions for the partition + if self.partition: + discovered_regions = filter_regions_for_partition(discovered_regions, self.partition) + + # Apply exclusions + account_regions = [r for r in discovered_regions if r not in exclude_regions] + logger.info( f"Found {len(account_regions)} enabled regions in account {account_id}" ) @@ -323,12 +330,19 @@ def _get_account_regions( ) return provided_regions return [] + # If we have provided regions, filter them for the partition elif provided_regions: - filtered_regions = [r for r in provided_regions if r not in exclude_regions] + if self.partition: + # Use centralized filter + valid_regions = filter_regions_for_partition(provided_regions, self.partition) + filtered_regions = [r for r in valid_regions if r not in exclude_regions] + else: + filtered_regions = [r for r in provided_regions if r not in exclude_regions] + if not filtered_regions: filtered_regions = self._get_default_regions(credentials) logger.info( - f"All provided regions were excluded, using defaults: {', '.join(filtered_regions)}" + f"All provided regions were excluded or invalid, using defaults: {', '.join(filtered_regions)}" ) return filtered_regions else: @@ -359,15 +373,11 @@ def _get_default_regions( except Exception as e: logger.warning(f"Failed to auto-detect partition from credentials: {e}") - # If still no partition, use commercial AWS as fallback - partition = partition or "aws" - - if partition == "aws-us-gov": - return ["us-gov-east-1", "us-gov-west-1"] - elif partition == "aws-cn": - return ["cn-north-1", "cn-northwest-1"] - else: - return ["us-east-1", "us-east-2", "us-west-1", "us-west-2"] + # If still no partition, use GovCloud AWS as fallback (default for this environment) + partition = partition or DEFAULT_PARTITION + + # Use centralized utility + return get_default_regions_for_partition(partition) def _get_emr_states(self, exclude_resources: List[str]) -> Optional[List[str]]: """Get valid EMR states for discovery.""" @@ -500,17 +510,31 @@ def _log_resource_counts( def _create_resource_managers( self, credentials: Dict[str, str], account_id: str, account_name: str ) -> Dict[str, Any]: - """Create all resource managers for an account.""" - # We're using a default region here just for manager initialization - # Each method in the manager will use the appropriate region for the operation - default_region = self._get_default_regions(credentials)[0] - - return { - "ec2": EC2Manager(account_id, default_region), - "rds": RDSManager(account_id, default_region), - "eks": EKSManager(account_id, default_region), - "emr": EMRManager(account_id, default_region), - } + """ + Create resource managers for all supported resource types. + + Args: + credentials: AWS credentials + account_id: AWS account ID + account_name: AWS account name + + Returns: + Dictionary of resource managers by type + """ + managers = {} + + for resource_type in self.RESOURCE_TYPES: + manager_class = self.resource_registry.get_manager_class(resource_type) + if manager_class: + try: + # Create an instance for each region + managers[resource_type] = {} + for region in self.region_cache.get(account_id, []): + managers[resource_type][region] = manager_class(account_id, region) + except Exception as e: + logger.error(f"Error creating {resource_type} manager: {str(e)}") + + return managers def _execute_actions( self, @@ -531,7 +555,38 @@ def safe_get_result(result: Optional[Dict[str, int]], key: str) -> int: # Execute actions on each resource type for resource_type in self.RESOURCE_TYPES: - if resource_type == "ecr": # ECR doesn't have start/stop actions + # Special handling for ECR - it has different resource keys and different actions + if resource_type == "ecr": + if resource_type in exclude_resources: + stats[f"{resource_type}_skipped"] += len(resources.get("ecr_images", [])) + continue + + # Get ECR manager + ecr_manager = managers.get(resource_type) + if not ecr_manager: + logger.warning(f"No manager found for {resource_type}, skipping") + continue + + # ECR has cleanup action instead of start/stop + if action == "stop" and resources.get("ecr_old_images"): + # Clean up old images when "stop" action is used + old_images = resources.get("ecr_old_images", []) + if old_images: + logger.info(f"Cleaning up {len(old_images)} old ECR images") + # Get a region-specific manager for each region with images + regions_with_images = set(img.get("region") for img in old_images if img.get("region")) + for region in regions_with_images: + region_images = [img for img in old_images if img.get("region") == region] + if region_images and region in ecr_manager: + result = ecr_manager[region].cleanup_old_images(region_images, dry_run) + stats["ecr_images_deleted"] = stats.get("ecr_images_deleted", 0) + safe_get_result(result, "success") + stats["ecr_errors"] = stats.get("ecr_errors", 0) + safe_get_result(result, "errors") + stats["ecr_skipped"] = stats.get("ecr_skipped", 0) + safe_get_result(result, "skipped") + else: + # For "start" action or if no old images, just note that ECR has no relevant action + stats[f"{resource_type}_skipped"] += len(resources.get("ecr_images", [])) + + # Skip the rest of the loop for ECR continue if resource_type in exclude_resources: @@ -610,3 +665,116 @@ def _log_summary(self, stats: Dict[str, Any], action: str) -> None: ) print_resource_summary(stats, action, stats.get("dry_run", False)) + + def discover_resources( + self, + regions: List[str] = None, + exclude_accounts: List[str] = None, + exclude_resources: List[str] = None, + exclude_regions: List[str] = None, + specific_account: Optional[str] = None, + ) -> Dict[str, List[Dict[str, Any]]]: + """ + Discover resources across accounts and regions. + + Args: + regions: List of regions to discover resources in + exclude_accounts: List of accounts to exclude + exclude_resources: List of resource types to exclude + exclude_regions: List of regions to exclude + specific_account: Optional specific account ID to focus on + + Returns: + Dictionary of resources by type + """ + exclude_accounts = exclude_accounts or [] + exclude_resources = exclude_resources or [] + exclude_regions = exclude_regions or [] + regions = regions or [] + + # Get account list + accounts = self._get_accounts() + + # Filter accounts if specific_account is provided + if specific_account: + accounts = [acc for acc in accounts if acc.get("account_id") == specific_account] + if not accounts: + logger.error(f"Specified account {specific_account} not found or not accessible") + return {} + + # Pre-filter accounts that we can authenticate with + accounts = self._pre_validate_accounts(accounts) + + # Apply account exclusions + accounts = [acc for acc in accounts if acc.get("account_id") not in exclude_accounts] + + if not accounts: + logger.error("No accounts available for resource discovery") + return {} + + # Initialize consolidated resources dictionary + all_resources = { + "ec2_instances": [], + "rds_instances": [], + "eks_clusters": [], + "emr_clusters": [], + "ecr_images": [], + "ecr_old_images": [], + } + + # Process each account + for account in accounts: + account_id = account.get("account_id") + account_name = account.get("account_name", "Unknown") + + logger.info(f"Discovering resources in account {account_name} ({account_id})") + + try: + # Get credentials for the account + credentials = get_credentials(account_id, self.profile_name) + if not credentials: + logger.warning(f"Could not obtain credentials for account {account_id}") + continue + + # Determine regions for this account + account_regions = self._get_account_regions( + account_id, credentials, regions, exclude_regions + ) + + if not account_regions: + logger.warning(f"No valid regions for account {account_id}") + continue + + # Auto-detect partition if not specified + partition = ( + detect_partition_from_credentials(credentials) + if not self.partition + else self.partition + ) + + # Get resources with special handling for EMR + emr_states = self._get_emr_states(exclude_resources) + + # Get all resources for the account + account_resources = get_account_resources( + credentials, + account_regions, + exclude_resources, + account_id, + account_name, + emr_cluster_states=emr_states, + ) + + # Merge resources into the consolidated dictionary + for resource_type, resources in account_resources.items(): + all_resources.setdefault(resource_type, []).extend(resources) + + except Exception as e: + logger.error(f"Error discovering resources for account {account_id}: {str(e)}") + + # Log summary of discovered resources + for resource_type, resources in all_resources.items(): + if resources: + logger.info(f"Total {resource_type}: {len(resources)}") + + return all_resources diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/csv_utils.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/csv_utils.py new file mode 100644 index 00000000..babf0a72 --- /dev/null +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/csv_utils.py @@ -0,0 +1,106 @@ +""" +Shared CSV utilities for AWS resource management. +""" + +import csv +import os +from datetime import datetime +from pathlib import Path +from typing import Any, Dict, List, Optional, Set + +def ensure_csv_dir(directory: str) -> str: + """ + Ensure the CSV directory exists and return its path. + + Args: + directory: Directory path to ensure exists + + Returns: + Absolute path to the directory + """ + path = Path(directory) + path.mkdir(parents=True, exist_ok=True) + return str(path.absolute()) + +def initialize_csv_file( + filepath: str, + headers: List[str], + overwrite: bool = False +) -> bool: + """ + Initialize a CSV file with headers if it doesn't exist or overwrite is True. + + Args: + filepath: Path to CSV file + headers: List of column headers + overwrite: Whether to overwrite existing file + + Returns: + True if file was created/initialized, False if file already existed + """ + # Create parent directories if they don't exist + os.makedirs(os.path.dirname(filepath), exist_ok=True) + + # Check if file exists + file_exists = os.path.isfile(filepath) + + # Create new file or overwrite existing file + if not file_exists or overwrite: + with open(filepath, 'w', newline='') as csvfile: + writer = csv.writer(csvfile) + writer.writerow(headers) + return True + + return False + +def write_csv_row(filepath: str, row_data: Dict[str, Any], headers: List[str]) -> None: + """ + Write a row to a CSV file. + + Args: + filepath: Path to CSV file + row_data: Dictionary containing row data + headers: List of column headers in correct order + """ + # Create file with headers if it doesn't exist + if not os.path.isfile(filepath): + initialize_csv_file(filepath, headers) + + # Write the row + with open(filepath, 'a', newline='') as csvfile: + writer = csv.DictWriter(csvfile, fieldnames=headers) + writer.writerow(row_data) + +def format_tags_for_csv(tags: Dict[str, str]) -> str: + """ + Format AWS resource tags for CSV output. + + Args: + tags: Dictionary of tag key-value pairs + + Returns: + Formatted string representation of tags + """ + if not tags: + return "" + return "; ".join([f"{k}={v}" for k, v in tags.items()]) + +def generate_timestamp_filename( + base_name: str, + prefix: Optional[str] = None, + extension: str = "csv" +) -> str: + """ + Generate a filename with timestamp. + + Args: + base_name: Base name for the file + prefix: Optional prefix + extension: File extension without dot + + Returns: + Timestamped filename + """ + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + prefix_str = f"{prefix}_" if prefix else "" + return f"{prefix_str}{base_name}_{timestamp}.{extension}" 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 e8478781..313a0564 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 @@ -1,21 +1,32 @@ """ -Resource discovery functionality. +Resource discovery functionality for AWS Resource Management. """ +import concurrent.futures import logging -from collections import defaultdict +from typing import Any, Dict, List, Optional from concurrent.futures import ThreadPoolExecutor, as_completed -from typing import Any, Dict, List, Optional, Set, Union - -import boto3 from aws_resource_management.aws_utils import ( detect_partition_from_credentials, - get_all_regions, - get_regions_for_partition, + get_available_regions, get_session_for_account, is_valid_region, list_enabled_regions, ) +from aws_resource_management.logging_setup import setup_logging +from aws_resource_management.partition_utils import ( + DEFAULT_REGIONS, + get_regions_for_partition, + is_region_in_partition, + filter_regions_for_partition, +) +from aws_resource_management.discovery_utils import ( + paginate_aws_response, # Import centralized pagination utility + format_tags, + add_account_info, + normalize_resource_state +) +import boto3 from aws_resource_management.config_manager import get_config from aws_resource_management.logging_setup import log_with_context, setup_logging from aws_resource_management.managers import ( @@ -35,7 +46,6 @@ from aws_resource_management.managers import EMRManager except ImportError: EMRManager = None - try: from aws_resource_management.managers import ECRManager except ImportError: @@ -44,27 +54,22 @@ logger = logging.getLogger("aws_resource_management") config = get_config() -# Default regions for different partitions -DEFAULT_REGIONS = { - "aws": ["us-east-1", "us-east-2", "us-west-1", "us-west-2"], - "aws-us-gov": ["us-gov-east-1", "us-gov-west-1"], - "aws-cn": ["cn-north-1", "cn-northwest-1"], -} - class ResourceDiscovery: - """Resource discovery for AWS resources.""" - - def __init__(self, credentials: Dict[str, str], account_id: str): - """ - Initialize resource discovery. - + """Discovers AWS resources across accounts and regions.""" + def __init__( + self, credentials: Dict[str, str], account_id: str, account_name: Optional[str] = None + ) -> None: + """Initialize resource discovery for an account. + Args: credentials: AWS credentials dictionary account_id: AWS account ID + account_name: AWS account name (optional) """ self.credentials = credentials self.account_id = account_id + self.account_name = account_name or account_id # Default to account_id if name not provided self.partition = detect_partition_from_credentials(credentials) logger.info( f"Resource discovery initialized for account {account_id} in partition {self.partition}" @@ -73,94 +78,53 @@ def __init__(self, credentials: Dict[str, str], account_id: str): def discover_resources( self, resource_type: str, - regions: Union[str, List[str]] = "all", + regions: List[str], resource_ids: Optional[List[str]] = None, + max_workers: int = 5, ) -> List[Dict[str, Any]]: - """ - Discover resources of the specified type. + """Discover resources of a specific type across multiple regions. Args: - resource_type: Resource type (ec2_instance, rds_instance, etc.) - regions: Regions to discover resources in, "all" for all regions - resource_ids: List of resource IDs to filter by + resource_type: Type of resource to discover (ec2, rds, eks, emr, ecr) + regions: List of regions to discover resources in + resource_ids: Optional list of specific resource IDs to discover + max_workers: Maximum number of concurrent worker threads Returns: - List of resource dictionaries + List of discovered resources """ - # Convert regions input to a list of actual region names - if regions == "all": - regions = get_regions_for_partition(self.partition) - elif isinstance(regions, str): - regions = [regions] - - # Filter regions to only include those in the detected partition - valid_regions = [] - for region in regions: - # Skip regions that don't belong to our partition - if self.partition == "aws-us-gov" and not region.startswith("us-gov-"): - logger.debug( - f"Skipping region {region} - not in {self.partition} partition" - ) - continue - elif self.partition == "aws-cn" and not region.startswith("cn-"): - logger.debug( - f"Skipping region {region} - not in {self.partition} partition" - ) - continue - elif self.partition == "aws" and ( - region.startswith("us-gov-") or region.startswith("cn-") - ): - logger.debug( - f"Skipping region {region} - not in {self.partition} partition" - ) - continue - - # Add the valid region to our list - valid_regions.append(region) - + logger.debug(f"Discovering {resource_type} resources in {len(regions)} regions") + + # Validate regions for this partition + valid_regions = filter_regions_for_partition(regions, self.partition) + # If we filtered out all regions, use default regions for the partition if not valid_regions: valid_regions = DEFAULT_REGIONS.get(self.partition, DEFAULT_REGIONS["aws"]) - logger.info( + logger.warning( f"No valid regions found for partition {self.partition}, using defaults: {', '.join(valid_regions)}" ) - - # Use the filtered valid regions + + # Use validated regions regions = valid_regions - # Determine which regions to search - region_list = [] - if regions == "all": - region_list = get_all_regions() - elif isinstance(regions, list): - region_list = [r for r in regions if is_valid_region(r)] - else: - logger.error(f"Invalid regions parameter: {regions}") - return [] - - if not region_list: - logger.warning("No valid regions specified for resource discovery") - return [] - - # Call the appropriate discovery method based on resource type + # Validate resource type discovery_method = getattr(self, f"_discover_{resource_type}", None) if not discovery_method: - logger.error( - f"No discovery method available for resource type: {resource_type}" - ) + logger.error(f"No discovery method available for resource type: {resource_type}") return [] # Use thread pool to speed up discovery across regions resources = [] - with ThreadPoolExecutor(max_workers=min(10, len(region_list))) as executor: + with concurrent.futures.ThreadPoolExecutor(max_workers=min(max_workers, len(regions))) as executor: # Create a future for each region future_to_region = { executor.submit(discovery_method, region, resource_ids): region - for region in region_list + for region in regions } # Process results as they complete - for future in as_completed(future_to_region): + for future in concurrent.futures.as_completed(future_to_region): region = future_to_region[future] try: region_resources = future.result() @@ -171,13 +135,13 @@ def discover_resources( except Exception as e: logger.error(f"Error discovering {resource_type} in {region}: {e}") + logger.info(f"Total {resource_type} resources discovered: {len(resources)}") return resources def _discover_ec2( self, region: str, resource_ids: Optional[List[str]] = None ) -> List[Dict[str, Any]]: - """ - Discover EC2 instances in a region. + """Discover EC2 instances in a region. Args: region: AWS region @@ -201,35 +165,38 @@ def _discover_ec2( filters.append({"Name": "instance-id", "Values": resource_ids}) try: - # Get instances - response = ec2_client.describe_instances(Filters=filters) - - # Extract instance information - instances = [] - for reservation in response.get("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 simplified instance dictionary - instance_dict = { - "id": instance["InstanceId"], - "name": tags.get("Name", "Unnamed"), - "type": instance["InstanceType"], - "status": instance["State"]["Name"], - "region": region, - "tags": tags, - "launch_time": ( - instance.get("LaunchTime", "").isoformat() - if instance.get("LaunchTime") - else None - ), - "public_ip": instance.get("PublicIpAddress"), - "private_ip": instance.get("PrivateIpAddress"), - } - instances.append(instance_dict) + # Define a page processor for EC2 instances + def process_ec2_page(page: Dict[str, Any]) -> List[Dict[str, Any]]: + instances = [] + for reservation in page.get("Reservations", []): + for instance in reservation.get("Instances", []): + # Convert tags to dictionary + tags = format_tags(instance.get("Tags", [])) + # Create a simplified instance dictionary + instances.append({ + "id": instance["InstanceId"], + "name": tags.get("Name", "Unnamed"), + "type": instance["InstanceType"], + "status": instance["State"]["Name"], + "region": region, + "tags": tags, + "launch_time": ( + instance.get("LaunchTime", "").isoformat() + if instance.get("LaunchTime") + else None + ), + "public_ip": instance.get("PublicIpAddress"), + "private_ip": instance.get("PrivateIpAddress"), + }) + return instances + + # Use the centralized pagination utility with our custom processor + instances = paginate_aws_response( + ec2_client, + "describe_instances", + page_processor=process_ec2_page, + Filters=filters + ) return instances @@ -240,8 +207,7 @@ def _discover_ec2( def _discover_rds( self, region: str, resource_ids: Optional[List[str]] = None ) -> List[Dict[str, Any]]: - """ - Discover RDS instances in a region. + """Discover RDS instances in a region. Args: region: AWS region @@ -262,8 +228,6 @@ def _discover_rds( try: # Get instances db_instances = [] - - # Use filters if specific IDs are provided if resource_ids: for db_id in resource_ids: try: @@ -323,8 +287,7 @@ def _discover_rds( def _discover_emr( self, region: str, resource_ids: Optional[List[str]] = None ) -> List[Dict[str, Any]]: - """ - Discover EMR clusters in a region. + """Discover EMR clusters in a region. Args: region: AWS region @@ -435,8 +398,7 @@ def _discover_emr( def _discover_eks( self, region: str, resource_ids: Optional[List[str]] = None ) -> List[Dict[str, Any]]: - """ - Discover EKS clusters in a region. + """Discover EKS clusters in a region. Args: region: AWS region @@ -512,8 +474,7 @@ def _discover_eks( def _discover_ecr( self, region: str, resource_ids: Optional[List[str]] = None ) -> List[Dict[str, Any]]: - """ - Discover ECR repositories and images in a region. + """Discover ECR repositories and images in a region. Args: region: AWS region @@ -576,7 +537,6 @@ def _discover_ecr( chunk = image_ids[i : i + chunk_size] if not chunk: continue - try: # Get details for all images in this chunk image_details = ecr_client.describe_images( @@ -591,7 +551,6 @@ def _discover_ecr( primary_tag = image["imageTags"][0] else: primary_tag = "untagged" - # Create a simplified image dictionary image_dict = { "id": image_id, @@ -626,7 +585,6 @@ def _discover_ecr( logger.error(f"Error discovering ECR repositories in {region}: {e}") return [] - # Backward compatibility wrapper for get_account_resources # This function accepts both positional and keyword arguments def get_account_resources(*args, **kwargs) -> Dict[str, List[Dict[str, Any]]]: @@ -674,7 +632,6 @@ def get_account_resources(*args, **kwargs) -> Dict[str, List[Dict[str, Any]]]: emr_cluster_states=emr_cluster_states, ) - def _get_account_resources_impl( credentials: Dict[str, str], regions: List[str], @@ -707,7 +664,7 @@ def _get_account_resources_impl( "eks_clusters": [], "emr_clusters": [], "ecr_images": [], - "ecr_old_images": [], + "ecr_old_images": [], # Add a new key for old ECR images } try: @@ -728,7 +685,7 @@ def _get_account_resources_impl( logger.error(f"Could not create session for account {account_id}") return resources - # Extract credentials from session for the ResourceDiscovery + # Extract credentials from session creds = session.get_credentials() discovery_credentials = { "aws_access_key_id": creds.access_key, @@ -739,7 +696,7 @@ def _get_account_resources_impl( } # Create resource discovery instance - discovery = ResourceDiscovery(discovery_credentials, account_id) + discovery = ResourceDiscovery(discovery_credentials, account_id, account_name) # Discover EC2 instances if "ec2" not in exclude_resources: @@ -756,32 +713,45 @@ def _get_account_resources_impl( resources["rds_instances"] = discovery.discover_resources("rds", regions) # Discover EKS clusters - if "eks" not in exclude_resources and EKSManager: + if "eks" not in exclude_resources: logger.info( f"Discovering EKS clusters for account {account_id} in {len(regions)} regions" ) - resources["eks_clusters"] = discovery.discover_resources("eks", regions) + try: + resources["eks_clusters"] = discovery.discover_resources("eks", regions) + except Exception as e: + logger.error(f"Error discovering EKS clusters: {e}") + resources["eks_clusters"] = [] # Discover EMR clusters - if "emr" not in exclude_resources and EMRManager: + if "emr" not in exclude_resources: logger.info( f"Discovering EMR clusters for account {account_id} in {len(regions)} regions" ) - # Use emr_cluster_states if provided - resources["emr_clusters"] = discovery.discover_resources( - "emr", regions, resource_ids=None - ) + try: + resources["emr_clusters"] = discovery.discover_resources("emr", regions) + except Exception as e: + logger.error(f"Error discovering EMR clusters: {e}") + resources["emr_clusters"] = [] # Discover ECR images - if "ecr" not in exclude_resources and ECRManager: + if "ecr" not in exclude_resources: logger.info( f"Discovering ECR images for account {account_id} in {len(regions)} regions" ) - ecr_resources = discovery.discover_resources("ecr", regions) - # Process ECR resources based on your application's ECR discovery structure - resources["ecr_images"] = ecr_resources - # Filter old images if needed - resources["ecr_old_images"] = [] + try: + ecr_resources = discovery.discover_resources("ecr", regions) + resources["ecr_images"] = ecr_resources + # Filter old images based on age - only if image has pushed_at attribute + resources["ecr_old_images"] = [ + img for img in ecr_resources + if img.get("is_old", False) or + (img.get("age_days", 0) > 365) + ] + except Exception as e: + logger.error(f"Error discovering ECR images: {e}") + resources["ecr_images"] = [] + resources["ecr_old_images"] = [] # Add account information to all resources for resource_type, resource_list in resources.items(): diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/discovery_utils.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/discovery_utils.py new file mode 100644 index 00000000..dff44416 --- /dev/null +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/discovery_utils.py @@ -0,0 +1,169 @@ +""" +Shared discovery utilities for AWS resource management. + +This module provides common functions used in resource discovery +across different resource types. +""" + +import logging +from typing import Any, Dict, List, Optional, Callable, Iterator, Tuple, Union + +import boto3 +from botocore.exceptions import ClientError +from botocore.paginate import PageIterator + +logger = logging.getLogger(__name__) + +def paginate_aws_response( + client: Any, + operation: str, + result_key: Optional[str] = None, + max_items: Optional[int] = None, + page_processor: Optional[Callable[[Dict[str, Any]], List[Dict[str, Any]]]] = None, + **kwargs +) -> List[Dict[str, Any]]: + """ + Paginate through AWS API responses with flexible options. + + Args: + client: Boto3 client + operation: Operation name (e.g., 'describe_instances') + result_key: Key in the response that contains the results (optional) + max_items: Maximum number of items to return (optional) + page_processor: Optional function to process each page before extracting results + **kwargs: Additional arguments for the operation + + Returns: + Combined list of results from all pages + """ + try: + # Check if the operation supports pagination + if not hasattr(client, 'get_paginator') or not hasattr(client.get_paginator, '__call__'): + # Fall back to direct call if pagination isn't supported + logger.debug(f"Pagination not supported for {operation}, making direct call") + response = getattr(client, operation)(**kwargs) + + if result_key and result_key in response: + return response[result_key] + return [response] # Return response as a list item + + # Initialize pagination + paginator = client.get_paginator(operation) + + # Configure pagination + pagination_config = {} + if max_items: + pagination_config['MaxItems'] = max_items + + page_iterator = paginator.paginate(**kwargs, PaginationConfig=pagination_config) + + # Process pages and collect results + results = [] + for page in page_iterator: + # Apply page processor if provided + if page_processor: + processed_items = page_processor(page) + results.extend(processed_items) + continue + + # Extract items using result_key if provided + if result_key: + if result_key in page: + results.extend(page[result_key]) + else: + logger.warning(f"Result key '{result_key}' not found in {operation} response") + else: + # If no result_key provided, add the entire page as a result item + results.append(page) + + return results + + except Exception as e: + logger.error(f"Error paginating {operation}: {str(e)}") + return [] + +def format_tags(tags_list: List[Dict[str, str]]) -> Dict[str, str]: + """ + Convert AWS tags list format to dictionary. + + Args: + tags_list: List of tag dictionaries with Key and Value + + Returns: + Dictionary of tag key-value pairs + """ + if not tags_list: + return {} + + return {tag.get("Key", ""): tag.get("Value", "") for tag in tags_list} + +def should_exclude_resource( + tags: Dict[str, str], + exclusion_tag: str +) -> bool: + """ + Check if a resource should be excluded based on tags. + + Args: + tags: Resource tags + exclusion_tag: Tag key to check for exclusion + + Returns: + True if resource should be excluded, False otherwise + """ + return exclusion_tag in tags + +def normalize_resource_state(resource: Dict[str, Any]) -> None: + """ + Normalize resource state/status fields for consistency. + + Args: + resource: Resource dictionary to normalize + """ + if "state" not in resource and "status" in resource: + resource["state"] = resource["status"] + elif "status" not in resource and "state" in resource: + resource["status"] = resource["state"] + elif "state" not in resource and "status" not in resource: + resource["status"] = "unknown" + resource["state"] = "unknown" + +def add_account_info( + resources: List[Dict[str, Any]], + account_id: str, + account_name: Optional[str] = None +) -> None: + """ + Add account information to resources. + + Args: + resources: List of resource dictionaries + account_id: AWS account ID + account_name: AWS account name + """ + for resource in resources: + resource["accountId"] = account_id + if account_name: + resource["accountName"] = account_name + +def safe_api_call(func, error_message: str, default_return: Any = None) -> Any: + """ + Safely execute an AWS API call with error handling. + + Args: + func: Function to execute (usually a lambda) + error_message: Error message to log on failure + default_return: Value to return on failure + + Returns: + API call result or default value on failure + """ + try: + return func() + except ClientError as e: + error_code = e.response.get("Error", {}).get("Code", "Unknown") + logger.error(f"{error_message}: {error_code} - {str(e)}") + except Exception as e: + logger.error(f"{error_message}: {str(e)}") + + return default_return diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/logging_setup.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/logging_setup.py index 8e67c482..7e4d73d5 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/logging_setup.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/logging_setup.py @@ -1,7 +1,6 @@ """ Enhanced logging utilities for AWS Resource Management. """ - import csv import datetime import json @@ -14,7 +13,7 @@ from typing import Any, Callable, Dict, Optional from aws_resource_management.config_manager import get_config - +from aws_resource_management.csv_utils import ensure_csv_dir, initialize_csv_file, write_csv_row # Global context data that can be included in log messages class LoggingContext: @@ -249,34 +248,26 @@ def initialize_csv_log(csv_file: Optional[str] = None) -> str: # Configure CSV logging directory csv_log_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "logs") + csv_log_dir = ensure_csv_dir(csv_log_dir) csv_path = os.path.join(csv_log_dir, csv_file) - # Create directory if it doesn't exist - Path(csv_log_dir).mkdir(parents=True, exist_ok=True) - - # Check if file exists, if not create with headers - file_exists = os.path.isfile(csv_path) - - if not file_exists: - with open(csv_path, "w", newline="") as f: - writer = csv.writer(f) - writer.writerow( - [ - "timestamp", - "account_id", - "account_name", - "resource_type", - "resource_id", - "resource_name", - "action", - "region", - "status", - "details", - "dry_run", - "schedule", - ] - ) - + # Initialize with headers + headers = [ + "timestamp", + "account_id", + "account_name", + "resource_type", + "resource_id", + "resource_name", + "action", + "region", + "status", + "details", + "dry_run", + "schedule", + ] + + initialize_csv_file(csv_path, headers, overwrite=False) return csv_path @@ -314,40 +305,27 @@ def log_action_to_csv( # Initialize CSV log file csv_path = initialize_csv_log() - with open(csv_path, "a", newline="") as csvfile: - fieldnames = [ - "timestamp", - "account_id", - "account_name", - "resource_type", - "resource_id", - "resource_name", - "action", - "region", - "status", - "details", - "dry_run", - "schedule", - ] - writer = csv.DictWriter(csvfile, fieldnames=fieldnames) - - # Write the log entry - writer.writerow( - { - "timestamp": timestamp, - "account_id": account_id, - "account_name": account_name, - "resource_type": resource_type, - "resource_id": resource_id, - "resource_name": resource_name, - "action": action, - "region": region, - "status": status, - "details": details, - "dry_run": "Yes" if dry_run else "No", - "schedule": existing_schedule or "", - } - ) + # Prepare row data + row_data = { + "timestamp": timestamp, + "account_id": account_id, + "account_name": account_name, + "resource_type": resource_type, + "resource_id": resource_id, + "resource_name": resource_name, + "action": action, + "region": region, + "status": status, + "details": details, + "dry_run": "Yes" if dry_run else "No", + "schedule": existing_schedule or "", + } + + # Headers must match the keys in row_data + headers = list(row_data.keys()) + + # Write to CSV + write_csv_row(csv_path, row_data, headers) # Create a logger instance for direct import 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 a532cbff..349a02c4 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 @@ -4,10 +4,19 @@ import logging from datetime import datetime -from typing import Any, Dict, List, Optional, Type +from typing import Any, Dict, List, Optional, Type, Callable import boto3 from aws_resource_management.aws_utils import get_session_for_account +from aws_resource_management.config_manager import get_config +from aws_resource_management.discovery_utils import ( + add_account_info, + normalize_resource_state, + paginate_aws_response, # Use the enhanced utility + safe_api_call, + should_exclude_resource +) +from aws_resource_management.partition_utils import get_partition_for_region from botocore.config import Config from botocore.exceptions import ClientError, NoRegionError @@ -17,11 +26,18 @@ class ResourceManager: """Base class for all resource managers.""" + # Class attributes for resource type identification and registration + # These should be overridden by subclasses + _RESOURCE_TYPE = None # e.g., 'ec2' - used for registry lookup + _DISPLAY_NAME = None # e.g., 'EC2 Instances' - user-friendly name + _DESCRIPTION = None # e.g., 'Amazon Elastic Compute Cloud instances' + def __init__(self, account_id: str, region: str): self.account_id = account_id self.region = region self.session = None self._clients = {} # Cache for boto3 clients + self.config = get_config() def get_timestamp(self) -> str: """ @@ -120,38 +136,61 @@ def _handle_api_error(self, operation: str, error: Exception) -> None: f"region {self.region}: {str(error)}" ) - def start(self, resource_ids: List[str]) -> Dict[str, str]: + def start(self, resources: List[Dict[str, Any]], dry_run: bool = False) -> Dict[str, int]: """ Start resources. Must be implemented by derived classes. Args: - resource_ids: List of resource IDs to start + resources: List of resources to start + dry_run: Whether to simulate the action Returns: - Dictionary mapping resource IDs to status + Dictionary of statistics (success, skipped, errors counts) """ raise NotImplementedError("Subclasses must implement start method") - def stop(self, resource_ids: List[str]) -> Dict[str, str]: + def stop(self, resources: List[Dict[str, Any]], dry_run: bool = False) -> Dict[str, int]: """ Stop resources. Must be implemented by derived classes. Args: - resource_ids: List of resource IDs to stop + resources: List of resources to stop + dry_run: Whether to simulate the action Returns: - Dictionary mapping resource IDs to status + Dictionary of statistics (success, skipped, errors counts) """ raise NotImplementedError("Subclasses must implement stop method") - def discover_resources(self) -> List[Dict[str, Any]]: + def discover_resources(self, regions: List[str]) -> List[Dict[str, Any]]: """ - Discover resources of this type. Must be implemented by derived classes. + Discover resources of this type across regions. + + Args: + regions: List of regions to discover resources in Returns: List of dictionaries containing resource information """ raise NotImplementedError("Subclasses must implement discover_resources method") + + def paginate_boto3( + self, client: Any, operation: str, result_key: str, **kwargs + ) -> List[Dict[str, Any]]: + """ + Paginate through AWS API responses. + + Args: + client: Boto3 client + operation: Operation name (e.g., 'describe_instances') + result_key: Key in the response that contains the results + **kwargs: Additional arguments for the operation + + Returns: + Combined list of results from all pages + """ + # Delegate to the centralized utility function + return paginate_aws_response(client, operation, result_key, **kwargs) def get_boto3_client(self, service_name: str, region: str) -> Any: """ @@ -187,31 +226,6 @@ def get_boto3_client(self, service_name: str, region: str) -> Any: # Alias for backward compatibility - many manager implementations use create_client() create_client = get_boto3_client - def paginate_boto3( - self, client: Any, operation: str, result_key: str - ) -> List[Dict[str, Any]]: - """ - Paginate through AWS API responses. - - Args: - client: Boto3 client - operation: Operation name (e.g., 'describe_instances') - result_key: Key in the response that contains the results - - Returns: - Combined list of results from all pages - """ - try: - paginator = client.get_paginator(operation) - results = [] - for page in paginator.paginate(): - if result_key in page: - results.extend(page[result_key]) - return results - except Exception as e: - logger.error(f"Error paginating {operation}: {str(e)}") - return [] - def log_action( self, resource_id: str, @@ -240,9 +254,30 @@ def log_action( schedule_str = f" [Schedule: {existing_schedule}]" if existing_schedule else "" logger.info( - f"{dry_run_prefix}{action.capitalize()} {self.resource_type} {resource_id} {name_str} " + f"{dry_run_prefix}{action.capitalize()} {self._RESOURCE_TYPE} {resource_id} {name_str} " f"in {region}{details_str}{schedule_str}" ) + + # Also log to CSV if tracking is enabled + try: + from aws_resource_management.logging_setup import log_action_to_csv + + log_action_to_csv( + account_id=self.account_id, + account_name=getattr(self, "account_name", self.account_id), + resource_type=self._RESOURCE_TYPE, + resource_id=resource_id, + resource_name=resource_name or resource_id, + action=action, + region=region, + status="simulated" if dry_run else "completed", + details=details or "", + dry_run=dry_run, + existing_schedule=existing_schedule, + ) + except ImportError: + # CSV logging not available + pass def should_exclude(self, resource: Dict[str, Any]) -> bool: """ @@ -254,31 +289,81 @@ def should_exclude(self, resource: Dict[str, Any]) -> bool: Returns: True if the resource should be excluded, False otherwise """ - from aws_resource_management.config_manager import get_config - - config = get_config() - tags = resource.get("tags", {}) - exclusion_tag = config.get("exclusion_tag") + exclusion_tag = self.config.get("exclusion_tag") if exclusion_tag and exclusion_tag in tags: return True return False - - def get_partition_for_region(self, region: str) -> str: + + def run_with_retries( + self, + func: Callable, + max_retries: int = 3, + error_message: str = "Operation failed" + ) -> Any: """ - Get the AWS partition for a region. - + Run a function with retries. + Args: - region: AWS region - + func: Function to run + max_retries: Maximum number of retries + error_message: Error message to log on failure + Returns: - AWS partition (aws, aws-cn, aws-us-gov) + Result of the function or None on failure """ - if region.startswith("us-gov-") or region.startswith("gov-"): - return "aws-us-gov" - elif region.startswith("cn-"): - return "aws-cn" - else: - return "aws" + retries = 0 + while retries <= max_retries: + try: + return func() + except Exception as e: + retries += 1 + if retries <= max_retries: + logger.warning( + f"{error_message}: {str(e)}. Retry {retries}/{max_retries}" + ) + else: + logger.error(f"{error_message}: {str(e)}. All retries failed") + return None + + def process_resources_in_batches( + self, + resources: List[Dict[str, Any]], + batch_action: Callable[[List[Dict[str, Any]]], bool], + batch_size: int = 20, + description: str = "Processing resources" + ) -> Dict[str, int]: + """ + Process resources in batches to avoid API limits. + + Args: + resources: List of resources to process + batch_action: Function to call for each batch + batch_size: Size of each batch + description: Description for logging + + Returns: + Dictionary with success and error counts + """ + stats = {"success": 0, "errors": 0, "skipped": 0} + + # Process in batches + for i in range(0, len(resources), batch_size): + batch = resources[i:i+batch_size] + logger.info(f"{description} batch {i//batch_size + 1}/{(len(resources)-1)//batch_size + 1} ({len(batch)} resources)") + + try: + if batch_action(batch): + stats["success"] += len(batch) + else: + stats["errors"] += len(batch) + except Exception as e: + logger.error(f"Error processing batch: {str(e)}") + stats["errors"] += len(batch) + + return stats + + # Use centralized partition utility + get_partition_for_region = get_partition_for_region 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 9d88475a..b44fd177 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 @@ -1,11 +1,18 @@ """ -EC2 resource manager class. +EC2 resource manager for AWS Resource Management. """ from collections import defaultdict from typing import Any, Dict, List, Optional from aws_resource_management.config_manager import get_config +from aws_resource_management.discovery_utils import ( + add_account_info, + format_tags, + normalize_resource_state, + paginate_aws_response, # Use the centralized utility + safe_api_call +) from aws_resource_management.logging_setup import setup_logging from aws_resource_management.managers.base import ResourceManager @@ -16,6 +23,11 @@ class EC2Manager(ResourceManager): """Manager for EC2 instances.""" + # Resource type information for registry + _RESOURCE_TYPE = "ec2" + _DISPLAY_NAME = "EC2 Instances" + _DESCRIPTION = "Amazon Elastic Compute Cloud instances" + def __init__( self, account_id: str, @@ -23,16 +35,15 @@ def __init__( ): """Initialize EC2 manager.""" super().__init__(account_id, region) - self.resource_type = "ec2_instance" self.account_name = None # This can be updated if needed 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.""" tags = instance.get("tags", {}) - exclusion_tag = config.get("exclusion_tag") - eks_tag = config.get("eks_tag") - emr_tag = config.get("emr_tag") + exclusion_tag = self.config.get("exclusion_tag") + eks_tag = self.config.get("eks_tag") + emr_tag = self.config.get("emr_tag") if exclusion_tag in tags: logger.debug( @@ -82,47 +93,52 @@ def stop( 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] + # Process instances in batches using the utility method + def process_batch(batch): instance_ids = [instance["id"] for instance in batch] - - try: - # Tag all instances in batch - if not dry_run: - logger.info( - f"Tagging {len(instance_ids)} EC2 instances in {region}" - ) - ec2_client.create_tags( + + # Tag and stop instances + if not dry_run: + # Using the safe_api_call utility for error handling + tag_success = safe_api_call( + lambda: ec2_client.create_tags( Resources=instance_ids, - Tags=[{"Key": config.get("stop_tag"), "Value": timestamp}], - ) - - # Stop all instances in batch - if not dry_run: - 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"Batch EC2 stop operation failed in {region}: {e}") - stats["errors"] += len(batch) + Tags=[{"Key": config.get("stop_tag"), "Value": timestamp}] + ), + f"Failed to tag EC2 instances in {region}" + ) + + stop_success = safe_api_call( + lambda: ec2_client.stop_instances(InstanceIds=instance_ids), + f"Failed to stop EC2 instances in {region}" + ) + + if not tag_success or not stop_success: + return False + + # Log individual instances + for instance in batch: + self.log_action( + instance["id"], + region, + "stop", + resource_name=instance.get("name", "Unnamed"), + dry_run=dry_run + ) + + return True + + # Using the process_resources_in_batches utility + batch_stats = self.process_resources_in_batches( + region_instances, + process_batch, + self.MAX_INSTANCES_PER_API_CALL, + f"Stopping EC2 instances in {region}" + ) + + # Merge stats + for key in ["success", "errors", "skipped"]: + stats[key] += batch_stats[key] logger.info( f"EC2 stop summary: {stats['success']} stopped, {stats['skipped']} skipped, {stats['errors']} errors" @@ -206,7 +222,7 @@ def start( ) return stats - def discover_instances(self, regions: List[str]) -> List[Dict[str, Any]]: + def discover_resources(self, regions: List[str]) -> List[Dict[str, Any]]: """Discover EC2 instances across multiple regions.""" all_instances = [] @@ -216,23 +232,16 @@ def discover_instances(self, regions: List[str]) -> List[Dict[str, Any]]: 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( - { + # Define processor function for EC2 instances + def process_ec2_page(page: Dict[str, Any]) -> List[Dict[str, Any]]: + instances = [] + for reservation in page.get("Reservations", []): + for instance in reservation.get("Instances", []): + # Convert tags to dictionary using shared utility + tags = format_tags(instance.get("Tags", [])) + + # Create a standardized instance dictionary + instances.append({ "id": instance["InstanceId"], "name": tags.get("Name", "Unnamed"), "type": instance["InstanceType"], @@ -249,9 +258,16 @@ def discover_instances(self, regions: List[str]) -> List[Dict[str, Any]]: "private_ip": instance.get("PrivateIpAddress"), "accountId": self.account_id, "accountName": self.account_name, - } - ) - + }) + return instances + + # Use centralized pagination utility with custom processor + instances = paginate_aws_response( + ec2_client, + "describe_instances", + page_processor=process_ec2_page + ) + logger.info(f"Found {len(instances)} EC2 instances in {region}") all_instances.extend(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 index b3b959be..500487d6 100644 --- 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 @@ -1,130 +1,280 @@ +""" +ECR resource manager for AWS Resource Management. +""" + import logging -from datetime import datetime, timedelta, timezone +from datetime import datetime, timedelta from typing import Any, Dict, List, Optional +from aws_resource_management.config_manager import get_config +from aws_resource_management.discovery_utils import ( + format_tags, paginate_aws_response, safe_api_call +) +from aws_resource_management.logging_setup import setup_logging from aws_resource_management.managers.base import ResourceManager -logger = logging.getLogger(__name__) +logger = setup_logging() +config = get_config() class ECRManager(ResourceManager): - """Manager for Amazon ECR (Elastic Container Registry) resources.""" + """Manager for ECR repositories and images.""" + + # Resource type information for registry + _RESOURCE_TYPE = "ecr" + _DISPLAY_NAME = "ECR Repositories" + _DESCRIPTION = "Amazon Elastic Container Registry repositories and images" def __init__( self, - credentials: Dict[str, str], account_id: str, - account_name: Optional[str] = None, + region: str, ): - """Initialize the ECR resource manager.""" - super().__init__(credentials, account_id, account_name) - self.resource_type = "ecr_image" + """Initialize ECR manager.""" + super().__init__(account_id, region) + self.account_name = None + self.old_image_threshold_days = config.get("ecr_old_image_threshold_days", 365) - 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.""" + def discover_resources(self, regions: List[str]) -> List[Dict[str, Any]]: + """ + Discover ECR repositories and their images. + + Args: + regions: List of AWS regions to scan + + Returns: + List of ECR image dictionaries + """ 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 = [] - # Create timezone-aware datetime for threshold to avoid comparison issues - threshold_date = datetime.now(timezone.utc) - timedelta(days=age_threshold_days) + try: + ecr_client = self.get_boto3_client("ecr", region) + if not ecr_client: + logger.warning(f"Could not create ECR client in {region}, skipping") + continue + + # Discover repositories + repositories = paginate_aws_response( + ecr_client, + "describe_repositories", + "repositories" + ) - try: - # Get all repositories efficiently using pagination helper - repositories = self.paginate_boto3( - ecr_client, "describe_repositories", "repositories" - ) + logger.info(f"Found {len(repositories)} ECR repositories in {region}") + + # Process each repository to get its images + for repo in repositories: + repo_name = repo.get("repositoryName", "unknown") + + try: + # Get image IDs (more efficient than getting all details at once) + image_ids = paginate_aws_response( + ecr_client, + "list_images", + "imageIds", + repositoryName=repo_name + ) + + logger.debug(f"Found {len(image_ids)} images in repository {repo_name}") + + # Process images in batches (ECR API has limits on batch size) + for i in range(0, len(image_ids), 100): + batch = image_ids[i:i + 100] + if not batch: + continue + + try: + # Get detailed information about images + image_details = safe_api_call( + lambda: ecr_client.describe_images( + repositoryName=repo_name, + imageIds=batch + ), + f"Error describing images in {repo_name}", + {"imageDetails": []} + ) + + # Process each image + for image in image_details.get("imageDetails", []): + try: + # Get primary tag (first in list) or use digest + image_tags = image.get("imageTags", []) + primary_tag = image_tags[0] if image_tags else "untagged" + + # Calculate image age + pushed_at = image.get("imagePushedAt") + is_old = False + age_days = 0 + + if pushed_at: + age = datetime.now().astimezone() - pushed_at.astimezone() + age_days = age.days + is_old = age_days > self.old_image_threshold_days + + # Create simplified image object + image_obj = { + "id": image.get("imageDigest", ""), + "name": f"{repo_name}:{primary_tag}", + "repositoryName": repo_name, + "region": region, + "tags": image_tags, + "size_mb": round(image.get("imageSizeInBytes", 0) / (1024 * 1024), 2), + "pushed_at": pushed_at.isoformat() if pushed_at else None, + "age_days": age_days, + "is_old": is_old, + "digest": image.get("imageDigest"), + "accountId": self.account_id, + "accountName": self.account_name + } + + all_images.append(image_obj) + except Exception as e: + logger.warning(f"Error processing image in {repo_name}: {e}") + + except Exception as e: + logger.warning(f"Error describing batch of images in {repo_name}: {e}") + + except Exception as e: + logger.warning(f"Error listing images in repository {repo_name}: {e}") - repository_names = [repo["repositoryName"] for repo in repositories] - logger.info( - f"Found {len(repository_names)} ECR repositories in region {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 + except Exception as e: + logger.error(f"Error discovering ECR repositories in {region}: {e}") + + logger.info(f"Discovered {len(all_images)} ECR images across {len(regions)} regions") + return all_images + + def start(self, resources: List[Dict[str, Any]], dry_run: bool = False) -> Dict[str, int]: + """ + ECR images don't have a start operation, so this is a no-op. + + Args: + resources: List of ECR image dictionaries + dry_run: Whether to simulate the action + + Returns: + Dictionary with success and error counts + """ + logger.info("ECR images don't support start operations") + return {"success": 0, "errors": 0, "skipped": len(resources)} + + def stop(self, resources: List[Dict[str, Any]], dry_run: bool = False) -> Dict[str, int]: + """ + ECR images don't have a stop operation, so this is a no-op. + + Args: + resources: List of ECR image dictionaries + dry_run: Whether to simulate the action + + Returns: + Dictionary with success and error counts + """ + logger.info("ECR images don't support stop operations") + return {"success": 0, "errors": 0, "skipped": len(resources)} + + def cleanup_old_images( + self, + images: List[Dict[str, Any]], + dry_run: bool = False, + keep_tagged: bool = True + ) -> Dict[str, int]: + """ + Delete old ECR images to save storage costs. + + Args: + images: List of ECR image dictionaries + dry_run: Whether to simulate the action + keep_tagged: Whether to preserve images with tags + + Returns: + Dictionary with success and error counts + """ + stats = {"success": 0, "errors": 0, "skipped": 0} + + # Group images by repository and region for more efficient processing + images_by_repo_region = {} + for image in images: + repo_name = image.get("repositoryName") + region = image.get("region") + + if not repo_name or not region: + stats["skipped"] += 1 + continue + + key = f"{region}:{repo_name}" + if key not in images_by_repo_region: + images_by_repo_region[key] = [] + + images_by_repo_region[key].append(image) + + # Process each repository + for key, repo_images in images_by_repo_region.items(): + region, repo_name = key.split(":", 1) + + # Only process images that meet deletion criteria + to_delete = [] + for image in repo_images: + # Skip images that aren't old + if not image.get("is_old", False): + stats["skipped"] += 1 + continue + + # Skip tagged images if keep_tagged is True + if keep_tagged and image.get("tags"): + stats["skipped"] += 1 + continue + + to_delete.append(image) + + if not to_delete: + continue + + # Get ECR client for this region + ecr_client = self.get_boto3_client("ecr", region) + if not ecr_client: + logger.error(f"Failed to create ECR client for region {region}") + stats["errors"] += len(to_delete) + continue + + # Delete images in batches (ECR API limit is 100 per call) + for i in range(0, len(to_delete), 100): + batch = to_delete[i:i + 100] + + # Create image identifiers for the batch + image_ids = [] + for image in batch: + digest = image.get("digest") + if digest: + image_ids.append({"imageDigest": digest}) + + if not image_ids: + continue + + # Delete the batch + if not dry_run: + try: + logger.info(f"Deleting {len(image_ids)} old images from {repo_name} in {region}") + ecr_client.batch_delete_image( + repositoryName=repo_name, + imageIds=image_ids + ) + stats["success"] += len(image_ids) + except Exception as e: + logger.error(f"Error deleting images from {repo_name} in {region}: {e}") + stats["errors"] += len(image_ids) + else: + logger.info(f"[DRY RUN] Would delete {len(image_ids)} old images from {repo_name} in {region}") + stats["success"] += len(image_ids) + + # Log individual images for tracking + for image in batch: + self.log_action( + image.get("digest", "unknown")[:12], + region, + "delete", + resource_name=image.get("name", "Unknown"), + details=f"Age: {image.get('age_days')} days", + dry_run=dry_run ) - - # 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 - # Make sure we're comparing compatible datetime objects - if "imagePushedAt" in image_detail: - # Both are now timezone-aware so comparison is safe - if 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} + + return stats 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 d8f23c1c..5ff40c28 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 @@ -1,5 +1,5 @@ """ -EMR resource manager class. +EMR resource manager for AWS Resource Management. """ from typing import Any, Dict, List, Optional @@ -12,6 +12,7 @@ from aws_resource_management.logging_setup import setup_logging from aws_resource_management.managers.base import ResourceManager from botocore.exceptions import ClientError +from aws_resource_management.partition_utils import get_partition_for_region logger = setup_logging() config = get_config() @@ -492,3 +493,6 @@ def start( ) return {"success": success_count, "errors": error_count} + + def _some_method_using_partition(self): + partition = get_partition_for_region(self.region) diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/partition_utils.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/partition_utils.py new file mode 100644 index 00000000..59cf2721 --- /dev/null +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/partition_utils.py @@ -0,0 +1,172 @@ +""" +Centralized AWS partition information and utilities. + +This module provides consistent partition information and utilities +for working with different AWS partitions (commercial AWS, GovCloud, China) +throughout the application. +""" + +from typing import Any, Dict, List, Optional, Set, Tuple + + +# Centralized mapping of partitions to their regions +PARTITION_REGIONS = { + "aws-us-gov": ["us-gov-east-1", "us-gov-west-1"], + "aws-cn": ["cn-north-1", "cn-northwest-1"], + "aws": [ + "us-east-1", + "us-east-2", + "us-west-1", + "us-west-2", + "eu-west-1", + "eu-central-1", + "eu-west-2", + "eu-west-3", + "ap-northeast-1", + "ap-northeast-2", + "ap-southeast-1", + "ap-southeast-2", + "ca-central-1", + "sa-east-1", + "eu-north-1", + "eu-south-1", + "af-south-1", + "ap-east-1", + "ap-south-1", + "me-south-1" + ], +} + +# Sensible defaults for each partition when no regions are specified +DEFAULT_REGIONS = { + "aws-us-gov": ["us-gov-east-1", "us-gov-west-1"], + "aws-cn": ["cn-north-1", "cn-northwest-1"], + "aws": ["us-east-1", "us-east-2", "us-west-1", "us-west-2"], +} + +# The default partition to use if no partition is detected +DEFAULT_PARTITION = "aws-us-gov" + +# Map of region prefixes to partitions for quick lookups +REGION_PREFIX_TO_PARTITION = { + "us-gov-": "aws-us-gov", + "gov-": "aws-us-gov", + "cn-": "aws-cn" +} + +# All known AWS regions (flattened from PARTITION_REGIONS) +ALL_KNOWN_REGIONS = [] +for regions in PARTITION_REGIONS.values(): + ALL_KNOWN_REGIONS.extend(regions) + + +def get_partition_for_region(region_name: str) -> str: + """ + Determine AWS partition based on region name. + + Args: + region_name: AWS region name + + Returns: + AWS partition name (e.g., 'aws', 'aws-us-gov', 'aws-cn') + """ + # Check common region prefixes first + for prefix, partition in REGION_PREFIX_TO_PARTITION.items(): + if region_name.startswith(prefix): + return partition + + # Default to standard AWS partition + return "aws" + + +def get_regions_for_partition(partition: str) -> List[str]: + """ + Get the list of regions for a specific partition. + + Args: + partition: AWS partition name + + Returns: + List of region names in the partition + """ + return PARTITION_REGIONS.get(partition, PARTITION_REGIONS["aws"]) + + +def get_default_regions_for_partition(partition: str) -> List[str]: + """ + Get the default regions commonly used for a partition. + + Args: + partition: AWS partition name + + Returns: + List of default region names + """ + return DEFAULT_REGIONS.get(partition, DEFAULT_REGIONS["aws"]) + + +def is_region_in_partition(region: str, partition: str) -> bool: + """ + Check if a region belongs to the specified partition. + + Args: + region: AWS region name + partition: AWS partition name + + Returns: + True if the region is in the partition, False otherwise + """ + return get_partition_for_region(region) == partition + + +def is_valid_region(region: str) -> bool: + """ + Check if a region is known to be valid without making an API call. + + Args: + region: Region name to validate + + Returns: + True if the region is valid, False otherwise + """ + if not isinstance(region, str): + return False + + # First check if it's in our known regions list + if region in ALL_KNOWN_REGIONS: + return True + + # Check if it follows expected naming patterns for newer regions + # Commercial regions follow patterns like us-east-1, eu-west-2 + if region.startswith(("us-", "eu-", "ap-", "sa-", "ca-", "me-", "af-")) and len(region.split("-")) >= 3: + return True + + # GovCloud regions + if region.startswith("us-gov-") and len(region.split("-")) >= 3: + return True + + # China regions + if region.startswith("cn-") and len(region.split("-")) >= 3: + return True + + return False + + +def filter_regions_for_partition(regions: List[str], partition: str) -> List[str]: + """ + Filter regions to include only those belonging to the specified partition. + + Args: + regions: List of region names + partition: AWS partition name + + Returns: + List of valid region names within the partition + """ + # First validate the regions and filter out invalid ones + valid_regions = [r for r in regions if is_valid_region(r)] + + # Then filter for the specific partition + partition_regions = [r for r in valid_regions if is_region_in_partition(r, partition)] + + return partition_regions 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 index 25c611f3..1ef4f3ed 100644 --- 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 @@ -1,9 +1,17 @@ """ Reporting functionality for AWS resource management. """ - import logging -from typing import Any, Dict, List +import os +from typing import Any, Dict, List, Optional + +from aws_resource_management.csv_utils import ( + ensure_csv_dir, + format_tags_for_csv, + generate_timestamp_filename, + initialize_csv_file, + write_csv_row, +) logger = logging.getLogger(__name__) @@ -13,7 +21,7 @@ def print_resource_summary( ) -> 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) @@ -79,17 +87,20 @@ def print_resource_summary( 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)}") + if action == "stop": + logger.info(f"Old images deleted: {stats.get('ecr_images_deleted', 0)}") + logger.info(f"Skipped: {stats.get('ecr_skipped', 0)}") + logger.info(f"Errors: {stats.get('ecr_errors', 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 """ @@ -107,7 +118,7 @@ def print_error_summary(errors: List[str]) -> None: def initialize_stats() -> Dict[str, Any]: """ Initialize a statistics dictionary with default values. - + Returns: A dictionary with initialized statistics counters """ @@ -122,10 +133,11 @@ def initialize_stats() -> Dict[str, Any]: "rds_engines": {}, "ecr_images_found": 0, "ecr_old_images_found": 0, + "ecr_images_deleted": 0, } # Initialize resource-specific counters - for resource_type in ["ecr", "ec2", "rds", "eks", "emr"]: + for resource_type in ["ec2", "rds", "eks", "emr", "ecr"]: for stat_type in [ "found", "skipped", @@ -136,3 +148,75 @@ def initialize_stats() -> Dict[str, Any]: stats[f"{resource_type}_{stat_type}"] = 0 return stats + + +def export_resources_to_csv( + resources: Dict[str, List[Dict[str, Any]]], + output_dir: Optional[str] = None, + prefix: Optional[str] = None +) -> Dict[str, str]: + """ + Export discovered resources to CSV files. + + Args: + resources: Dictionary of resource lists by type + output_dir: Directory to save CSV files (defaults to current directory) + prefix: Optional prefix for CSV filenames + + Returns: + Dictionary mapping resource types to their CSV file paths + """ + if not output_dir: + output_dir = os.getcwd() + + # Ensure output directory exists + output_dir = ensure_csv_dir(output_dir) + + csv_files = {} + + # Process each resource type + for resource_type, resource_list in resources.items(): + if not resource_list: + logger.debug(f"No {resource_type} resources to export") + continue + + # Create CSV filename using utility function + filename = generate_timestamp_filename(resource_type, prefix) + filepath = os.path.join(output_dir, filename) + + try: + # Extract all unique keys to use as CSV headers + all_keys = set() + for resource in resource_list: + all_keys.update(resource.keys()) + + # Define common fields to appear first in the CSV + common_fields = [ + "id", "name", "accountId", "accountName", "region", + "status", "type", "tags" + ] + # Sort headers with common fields first, then alphabetically + headers = [h for h in common_fields if h in all_keys] + headers.extend(sorted(k for k in all_keys if k not in common_fields)) + + # Initialize the CSV file with headers + initialize_csv_file(filepath, headers, overwrite=True) + + # Process and write each resource + for resource in resource_list: + # Handle tags special case (convert dict to string) + if "tags" in resource and isinstance(resource["tags"], dict): + resource["tags"] = format_tags_for_csv(resource["tags"]) + + # Use the write_csv_row utility + write_csv_row(filepath, + {k: resource.get(k, "") for k in headers}, + headers) + + logger.info(f"Exported {len(resource_list)} {resource_type} resources to {filepath}") + csv_files[resource_type] = filepath + + except Exception as e: + logger.error(f"Error exporting {resource_type} to CSV: {e}") + + return csv_files diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/resource_registry.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/resource_registry.py new file mode 100644 index 00000000..7c524e50 --- /dev/null +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/resource_registry.py @@ -0,0 +1,197 @@ +""" +Resource manager registry for dynamically discovering and registering AWS resource types. +""" + +import importlib +import inspect +import logging +import os +import sys +from typing import Any, Dict, List, Optional, Type + +from aws_resource_management.logging_setup import setup_logging + +logger = setup_logging() + + +class ResourceRegistry: + """ + Registry for AWS resource managers and resource type information. + Provides dynamic discovery and registration of resource managers. + """ + + _instance = None # Singleton instance + _initialized = False + + def __new__(cls): + """Ensure singleton pattern for resource registry.""" + if cls._instance is None: + cls._instance = super(ResourceRegistry, cls).__new__(cls) + return cls._instance + + def __init__(self): + """Initialize the resource registry.""" + # Only initialize once (singleton pattern) + if ResourceRegistry._initialized: + return + + self._managers = {} # Dict of resource_type -> manager_class + self._display_names = {} # Dict of resource_type -> display_name + self._descriptions = {} # Dict of resource_type -> description + + # Discover managers on initialization + self._discover_managers() + ResourceRegistry._initialized = True + + def _discover_managers(self) -> None: + """Dynamically discover manager classes from the managers package.""" + try: + # Import the managers package to scan it + import aws_resource_management.managers as managers_pkg + + # Get the package directory + package_dir = os.path.dirname(managers_pkg.__file__) + + # Find Python files in the package (excluding __init__.py) + for filename in os.listdir(package_dir): + if filename.endswith('.py') and filename != '__init__.py' and not filename.startswith('_'): + module_name = filename[:-3] # Remove '.py' + + try: + # Import the module + module = importlib.import_module(f"aws_resource_management.managers.{module_name}") + + # Scan for manager classes in the module + for name, obj in inspect.getmembers(module): + # Look for classes that have a name ending with "Manager" + # and have a "_RESOURCE_TYPE" attribute + if (inspect.isclass(obj) and name.endswith('Manager') and + hasattr(obj, '_RESOURCE_TYPE') and obj._RESOURCE_TYPE): + + resource_type = obj._RESOURCE_TYPE + display_name = getattr(obj, '_DISPLAY_NAME', resource_type.upper()) + description = getattr(obj, '_DESCRIPTION', f"AWS {resource_type.upper()} resources") + + self.register_manager(resource_type, obj, display_name, description) + logger.debug(f"Auto-discovered resource manager: {name} for type {resource_type}") + + except (ImportError, AttributeError) as e: + logger.warning(f"Error importing manager from {module_name}: {str(e)}") + + if not self._managers: + logger.warning("No resource managers were discovered automatically.") + + except Exception as e: + logger.error(f"Error discovering resource managers: {str(e)}") + + def register_manager(self, + resource_type: str, + manager_class: Type, + display_name: Optional[str] = None, + description: Optional[str] = None) -> None: + """ + Register a resource manager for a specific resource type. + + Args: + resource_type: Resource type identifier (e.g., 'ec2', 'rds') + manager_class: Manager class for the resource type + display_name: Display name for the resource type + description: Description of the resource type + """ + resource_type = resource_type.lower() + self._managers[resource_type] = manager_class + + if display_name: + self._display_names[resource_type] = display_name + else: + self._display_names[resource_type] = resource_type.upper() + + if description: + self._descriptions[resource_type] = description + else: + self._descriptions[resource_type] = f"AWS {resource_type.upper()} resources" + + logger.debug(f"Registered resource manager for {resource_type}") + + def get_manager_class(self, resource_type: str) -> Optional[Type]: + """ + Get the manager class for a specific resource type. + + Args: + resource_type: Resource type identifier + + Returns: + Manager class or None if not found + """ + return self._managers.get(resource_type.lower()) + + def get_resource_types(self) -> List[str]: + """ + Get a list of all registered resource types. + + Returns: + List of resource type identifiers + """ + return list(self._managers.keys()) + + def get_display_name(self, resource_type: str) -> str: + """ + Get the display name for a resource type. + + Args: + resource_type: Resource type identifier + + Returns: + Display name + """ + return self._display_names.get(resource_type.lower(), resource_type.upper()) + + def get_description(self, resource_type: str) -> str: + """ + Get the description for a resource type. + + Args: + resource_type: Resource type identifier + + Returns: + Description + """ + return self._descriptions.get(resource_type.lower(), f"AWS {resource_type.upper()} resources") + + def get_resource_type_choices(self) -> List[Dict[str, str]]: + """ + Get a list of resource type choices suitable for CLI argument choices. + + Returns: + List of dictionaries with 'name', 'value', and 'help' keys + """ + choices = [] + for resource_type in sorted(self.get_resource_types()): + choices.append({ + 'name': self.get_display_name(resource_type), + 'value': resource_type, + 'help': self.get_description(resource_type) + }) + + # Add the "all" option + choices.append({ + 'name': 'ALL', + 'value': 'all', + 'help': 'All supported resource types' + }) + + return choices + + +# Create singleton instance for import +registry = ResourceRegistry() + + +def get_registry() -> ResourceRegistry: + """ + Get the resource registry singleton. + + Returns: + ResourceRegistry instance + """ + return registry diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils.py index 4c0fb04b..4f583af5 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils.py @@ -2,294 +2,17 @@ AWS utility functions for resource management. """ -import os -import re -from typing import Any, Dict, List, Optional, Union +# This file is being deprecated in favor of aws_utils.py which provides more +# comprehensive implementations with better caching and error handling. +# See aws_utils.py for all functionality. -import boto3 -from aws_resource_management.config_manager import get_config -from aws_resource_management.logging_setup import setup_logging -from botocore.exceptions import ClientError +import warnings -logger = setup_logging() -config = get_config() +warnings.warn( + "The utils.py module is deprecated. Please use aws_utils.py instead.", + DeprecationWarning, + stacklevel=2 +) - -def create_boto3_client( - service: str, credentials: Dict[str, str], region: str -) -> boto3.client: - """ - Create a boto3 client for a specific service using the provided credentials. - - Args: - service: AWS service name (e.g., 'ec2', 'rds') - credentials: AWS credentials dict with access key, secret key, and token - region: AWS region name - - Returns: - Configured boto3 client - """ - return boto3.client( - service, - aws_access_key_id=credentials["AccessKeyId"], - aws_secret_access_key=credentials["SecretAccessKey"], - aws_session_token=credentials["SessionToken"], - region_name=region, - ) - - -def get_available_regions() -> List[str]: - """ - Get a list of all available AWS regions. - - Returns: - List of region names - """ - try: - ec2_client = boto3.client("ec2") - response = ec2_client.describe_regions() - return [region["RegionName"] for region in response["Regions"]] - except Exception as e: - logger.error(f"Error getting available regions: {e}") - # Return a default list of common regions as fallback - default_region = config.get("default_region") - if default_region: - return [default_region] - return ["us-gov-east-1", "us-gov-west-1"] - - -def get_partition_for_region(region: str) -> str: - """ - Determine the AWS partition for a given region. - - Args: - region: AWS region name - - Returns: - str: AWS partition name ('aws', 'aws-us-gov', 'aws-cn') - """ - if region.startswith("us-gov-"): - return "aws-us-gov" - elif region.startswith("cn-"): - return "aws-cn" - else: - return "aws" - - -def find_profile_for_account( - account_id: str, role_name: Optional[str] = None -) -> Optional[str]: - """ - Find an appropriate SSO profile for the given account ID and role name. - - Args: - account_id: AWS account ID - role_name: Preferred role name (e.g., 'AdministratorAccess', 'inf-admin-t2') - - Returns: - Profile name if found, None otherwise - """ - try: - # Try to read profiles from AWS config - import configparser - - config_file = os.path.expanduser("~/.aws/config") - if not os.path.exists(config_file): - return None - - aws_config = configparser.ConfigParser() - aws_config.read(config_file) - - # Look for profiles matching the account ID - matching_profiles = [] - for section in aws_config.sections(): - # Extract the actual profile name from section (removing 'profile ' prefix if present) - profile_name = section - if section.startswith("profile "): - profile_name = section[8:] - - # Check if this profile is for the target account ID - if ( - "sso_account_id" in aws_config[section] - and aws_config[section]["sso_account_id"] == account_id - ): - matching_profiles.append( - (profile_name, aws_config[section].get("sso_role_name", "")) - ) - - # If role_name is specified, try to find a profile with that role - if role_name and matching_profiles: - for profile, profile_role in matching_profiles: - if profile_role.lower() == role_name.lower(): - logger.info( - f"Found matching profile {profile} for account {account_id} with role {role_name}" - ) - return profile - - # If we have any matching profiles, return the first one - if matching_profiles: - # Prefer profiles with admin roles if available - admin_profiles = [p for p, r in matching_profiles if "admin" in r.lower()] - if admin_profiles: - logger.info( - f"Using profile {admin_profiles[0]} for account {account_id}" - ) - return admin_profiles[0] - - logger.info( - f"Using profile {matching_profiles[0][0]} for account {account_id}" - ) - return matching_profiles[0][0] - - return None - except Exception as e: - logger.warning(f"Error finding profile for account {account_id}: {e}") - return None - - -def create_boto3_session_from_profile(profile_name: str) -> Optional[boto3.Session]: - """ - Create a boto3 session using the specified profile. - - Args: - profile_name: AWS profile name - - Returns: - Configured boto3 session or None if failed - """ - try: - logger.info(f"Creating session using profile: {profile_name}") - session = boto3.Session(profile_name=profile_name) - # Test if the session works by getting the caller identity - sts = session.client("sts") - sts.get_caller_identity() - return session - except Exception as e: - logger.warning(f"Failed to create session with profile {profile_name}: {e}") - return None - - -def assume_role( - account_id: str, region: Optional[str] = None, role_name: Optional[str] = None -) -> Optional[Dict[str, str]]: - """ - Get credentials for the specified account, preferring SSO profiles when available. - - Args: - account_id: AWS account ID to access - region: AWS region, used to determine partition - role_name: Role name to assume (defaults to config.assume_role_name) - - Returns: - Credentials dictionary or None if failed - """ - try: - # First try using SSO profiles if available - profile_name = find_profile_for_account(account_id, role_name) - if profile_name: - session = create_boto3_session_from_profile(profile_name) - if session: - # Get credentials from the session - credentials = session.get_credentials() - frozen_credentials = credentials.get_frozen_credentials() - return { - "AccessKeyId": frozen_credentials.access_key, - "SecretAccessKey": frozen_credentials.secret_key, - "SessionToken": frozen_credentials.token, - } - - # Fall back to assuming role directly if no profile found or profile didn't work - # Use default region from config if not provided - if region is None: - region = config.get("default_region") - - # Determine the correct partition for the ARN - partition = get_partition_for_region(region) - - # Use the specified role name or fall back to the default - actual_role_name = role_name if role_name else config.get("assume_role_name") - - role_arn = f"arn:{partition}:iam::{account_id}:role/{actual_role_name}" - sts_client = boto3.client("sts", region_name=region) - - logger.info(f"Assuming role {role_arn}") - response = sts_client.assume_role( - RoleArn=role_arn, RoleSessionName="ResourceManagementSession" - ) - - return response["Credentials"] - except Exception as e: - logger.error(f"Error accessing account {account_id}: {e}") - return None - - -def create_boto3_client_for_account( - account_id: str, - service: str, - region: Optional[str] = None, - role_name: Optional[str] = None, -) -> Optional[boto3.client]: - """ - Create a boto3 client for a specific service in the specified account. - - Args: - account_id: AWS account ID - service: AWS service name (e.g., 'ec2', 'rds') - region: AWS region name - role_name: Role name to use - - Returns: - Configured boto3 client or None if failed - """ - # First try using a profile directly - profile_name = find_profile_for_account(account_id, role_name) - if profile_name: - session = create_boto3_session_from_profile(profile_name) - if session: - if region: - return session.client(service, region_name=region) - else: - return session.client(service) - - # Fall back to assume_role method - credentials = assume_role(account_id, region, role_name) - if not credentials: - return None - - return create_boto3_client(service, credentials, region) - - -def get_organization_accounts( - exclude_accounts: Optional[List[str]] = None, -) -> List[Dict[str, str]]: - """ - Get a list of accounts in the AWS organization. - - Args: - exclude_accounts: List of account IDs to exclude - - Returns: - List of account dicts with id and name - """ - if exclude_accounts is None: - exclude_accounts = [] - - try: - org_client = boto3.client("organizations") - accounts = [] - - # Get all accounts in the organization - paginator = org_client.get_paginator("list_accounts") - for page in paginator.paginate(): - for account in page["Accounts"]: - # Skip suspended accounts and accounts in exclude list - if ( - account["Status"] == "ACTIVE" - and account["Id"] not in exclude_accounts - ): - accounts.append({"id": account["Id"], "name": account["Name"]}) - - return accounts - except Exception as e: - logger.error(f"Error getting organization accounts: {e}") - return [] +# Import everything from aws_utils to maintain backward compatibility +from aws_resource_management.aws_utils import *