From a6a8fed3132a700f112793d6ad17a19c05b48bb9 Mon Sep 17 00:00:00 2001 From: "Matthew C. Morgan" Date: Thu, 10 Jul 2025 17:42:19 -0400 Subject: [PATCH] latest --- .../gfl-resource-actions/Makefile | 2 +- .../aws_resource_management/cli.py | 18 + .../aws_resource_management/reporting.py | 200 +++-- .../utils/file_utils.py | 8 +- .../python-tools/gfl-resource-actions/plan.md | 802 ------------------ 5 files changed, 175 insertions(+), 855 deletions(-) delete mode 100644 local-app/python-tools/gfl-resource-actions/plan.md diff --git a/local-app/python-tools/gfl-resource-actions/Makefile b/local-app/python-tools/gfl-resource-actions/Makefile index 283f0dd9..ef7d0f8e 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 ecr --csv-output-dir ./ --csv-export + $(PYTHON) -m $(PACKAGE_NAME).cli --stop --dry-run --resource-type ecr --csv-output-dir ./ --csv-export --csv-per-account # Run to stop resources run-stop: 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 00ad1420..eaeffd50 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 @@ -119,6 +119,11 @@ def parse_args(): "--csv-prefix", help="Prefix for CSV filenames", ) + csv_group.add_argument( + "--csv-per-account", + action="store_true", + help="Create separate CSV files per account in addition to consolidated CSV", + ) return parser.parse_args() @@ -189,11 +194,24 @@ def process_command(args: argparse.Namespace) -> None: # Export to CSV logger.info(f"Exporting resources to CSV in directory: {args.csv_output_dir}") + + # Export consolidated CSV csv_files = export_resources_to_csv( resources=discovered_resources, output_dir=args.csv_output_dir, prefix=args.csv_prefix, + per_account=False, # Always create consolidated first ) + + # Export per-account CSVs if requested + if args.csv_per_account: + per_account_files = export_resources_to_csv( + resources=discovered_resources, + output_dir=args.csv_output_dir, + prefix=args.csv_prefix, + per_account=True, + ) + csv_files.update(per_account_files) # Log success message with file locations if csv_files: 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 207e7725..14bbfe80 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 @@ -179,6 +179,7 @@ def export_resources_to_csv( resources: Dict[str, List[Dict[str, Any]]], output_dir: Optional[str] = None, prefix: Optional[str] = None, + per_account: bool = False, ) -> Dict[str, str]: """ Export discovered resources to CSV files. @@ -187,6 +188,7 @@ def export_resources_to_csv( resources: Dictionary of resource lists by type output_dir: Directory to save CSV files (defaults to current directory) prefix: Optional prefix for CSV filenames + per_account: If True, create separate CSV files per account Returns: Dictionary mapping resource types to their CSV file paths @@ -197,59 +199,159 @@ def export_resources_to_csv( # Ensure output directory exists output_dir = ensure_directory(output_dir) + if per_account: + return _export_per_account_csv(resources, output_dir, prefix) + else: + 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 + + +def _export_per_account_csv( + resources: Dict[str, List[Dict[str, Any]]], + output_dir: str, + prefix: Optional[str] = None, +) -> Dict[str, str]: + """ + Export resources to separate CSV files per account. + + Returns: + Dictionary mapping account IDs to their CSV file paths + """ csv_files = {} - # Process each resource type + # Group all resources by account + account_resources = {} 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 - ) + for resource in resource_list: + account_id = resource.get('accountId', 'unknown') + account_name = resource.get('accountName', account_id) - logger.info( - f"Exported {len(resource_list)} {resource_type} resources to {filepath}" - ) - csv_files[resource_type] = filepath + if account_id not in account_resources: + account_resources[account_id] = { + 'account_name': account_name, + 'resources': {} + } + + if resource_type not in account_resources[account_id]['resources']: + account_resources[account_id]['resources'][resource_type] = [] + + account_resources[account_id]['resources'][resource_type].append(resource) + + # Create CSV file for each account + for account_id, account_data in account_resources.items(): + account_name = account_data['account_name'] + + # Generate account-specific filename + safe_account_name = account_name.replace(' ', '_').replace('(', '').replace(')', '') + account_filename = generate_timestamp_filename( + f"resources_account_{account_id}_{safe_account_name}", + prefix + ) + account_filepath = os.path.join(output_dir, account_filename) + + # Combine all resource types for this account into one CSV + _write_account_csv(account_filepath, account_data['resources'], account_id, account_name) - except Exception as e: - logger.error(f"Error exporting {resource_type} to CSV: {e}") + csv_files[f"account_{account_id}"] = account_filepath + logger.info(f"Exported resources for account {account_name} ({account_id}) to {account_filepath}") return csv_files + + +def _write_account_csv( + filepath: str, + resources_by_type: Dict[str, List[Dict[str, Any]]], + account_id: str, + account_name: str +) -> None: + """Write all resources for an account to a single CSV file.""" + all_resources = [] + + # Flatten all resource types into a single list + for resource_type, resource_list in resources_by_type.items(): + for resource in resource_list: + # Add resource type column + resource_copy = resource.copy() + resource_copy['resource_type'] = resource_type + all_resources.append(resource_copy) + + if not all_resources: + return + + # Get all unique columns + all_keys = set() + for resource in all_resources: + all_keys.update(resource.keys()) + + # Define column order + priority_fields = [ + "resource_type", "id", "name", "accountId", "accountName", + "region", "status", "type", "tags" + ] + headers = [h for h in priority_fields if h in all_keys] + headers.extend(sorted(k for k in all_keys if k not in priority_fields)) + + # Initialize CSV with headers + initialize_csv_file(filepath, headers, overwrite=True) + + # Write all resources + for resource in all_resources: + # Handle tags formatting + if "tags" in resource and isinstance(resource["tags"], dict): + resource["tags"] = format_tags_for_csv(resource["tags"]) + + write_csv_row(filepath, {k: resource.get(k, "") for k in headers}, headers) diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/file_utils.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/file_utils.py index 15cd04a3..4e3b2e26 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/file_utils.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/utils/file_utils.py @@ -34,22 +34,24 @@ def ensure_directory(directory_path: Union[str, Path]) -> str: def generate_timestamp_filename( - base_name: str, prefix: Optional[str] = None, extension: str = "csv" + base_name: str, prefix: Optional[str] = None, extension: str = "csv", account_suffix: Optional[str] = None ) -> str: """ - Generate a filename with timestamp. + Generate a filename with timestamp and optional account suffix. Args: base_name: Base name for the file prefix: Optional prefix extension: File extension without dot + account_suffix: Optional account-specific suffix 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}" + account_str = f"_{account_suffix}" if account_suffix else "" + return f"{prefix_str}{base_name}{account_str}_{timestamp}.{extension}" def get_logs_directory(base_dir: Optional[str] = None) -> str: diff --git a/local-app/python-tools/gfl-resource-actions/plan.md b/local-app/python-tools/gfl-resource-actions/plan.md deleted file mode 100644 index 6276a040..00000000 --- a/local-app/python-tools/gfl-resource-actions/plan.md +++ /dev/null @@ -1,802 +0,0 @@ -# AWS Resource Management Refactoring Plan - -## Phase 1: Core Module Creation ✅ - -We've successfully completed the first phase of the refactoring plan by creating a consolidated `utils` package with specialized utility modules: - -1. **aws_core_utils.py** - AWS credential management and session handling - - Credential acquisition and caching - - Session management - - Account discovery - -2. **region_utils.py** - Region and partition handling - - Region validation and filtering - - Partition detection - - Region/partition mapping - -3. **api_utils.py** - AWS API interaction patterns - - Pagination utilities - - Error handling helpers - - Resource normalization - -4. **file_utils.py** - File operations - - CSV handling - - JSON file operations - - Directory management - -5. **logging_utils.py** - Logging configuration - - Context-aware logging - - Operation logging with timing - - JSON formatting - -6. **config_utils.py** - Configuration management - - Config loading and merging - - Default configuration - - Config updates - -## Phase 2: Migration (Next Steps) - -The second phase involves updating imports throughout the codebase to use the new consolidated modules: - -1. Update import statements in: - - `core.py` - - `discovery.py` - - `reporting.py` - - `resource_registry.py` - - `managers/*.py` - -2. Add deprecation warnings to the old utility modules: - - Update `aws_utils.py` to import from new modules and add deprecation warning - - Update `discovery_utils.py` to import from new modules and add deprecation warning - - Update `csv_utils.py` to import from new modules and add deprecation warning - - Update `logging_setup.py` to import from new modules and add deprecation warning - - Update `config_manager.py` to import from new modules and add deprecation warning - -3. Add comprehensive tests: - - Create test cases for each new utility module - - Verify backward compatibility of functions - - Test in multiple AWS environments (Commercial, GovCloud, China) - -## Phase 3: Cleanup - -1. Remove deprecated modules: - - Remove original utility modules once all code has been updated to use the new modules - - Update documentation to reflect the new structure - - Clean up any remaining references - -2. Final validation: - - Run full test suite - - Verify all functionality in actual AWS environments - - Test with multiple AWS partitions - -## Implementation Notes - -1. **Backward Compatibility**: - - The new modules maintain the same function signatures as the original functions - - The `utils/__init__.py` file exposes commonly used functions for convenient import - - Old modules are kept but will import from new modules for backward compatibility - -2. **Improved Features**: - - Better error handling in API calls - - More consistent caching mechanisms - - Optimized pagination handling - - Clearer separation of concerns - - More comprehensive type annotations - -3. **Structure Benefits**: - - Eliminates circular dependencies - - Reduces code duplication - - Improves maintainability - - Simplifies future enhancements - -## Migration Guide for Developers - -When working with the AWS Resource Management codebase: - -1. **Prefer importing from `utils` package**: - ```python - # Old approach - from aws_resource_management.aws_utils import get_credentials - - # New approach - from aws_resource_management.utils import get_credentials - ``` - -2. **For specialized functions, import from specific module**: - ```python - # For region-specific functionality - from aws_resource_management.utils.region_utils import filter_regions_for_partition - - # For API-specific functionality - from aws_resource_management.utils.api_utils import retry_with_backoff - ``` - -3. **Use the comprehensive logging system**: - ```python - from aws_resource_management.utils import logger, log_operation - - with log_operation(logger, "Creating AWS session"): - session = get_session_for_account(account_id, region) - ``` - -Step-by-Step Improvement Plan for AWS Resource Management Tool -1. ✅ Code Organization and Structure (COMPLETED) -Currently, the codebase is well-organized with separate modules for different resource managers: - -✅ Implemented proper Python packaging with a setup.py file for easier installation -✅ Added type hints throughout the codebase for better IDE support and code quality -✅ Created a dedicated config management module with environment variable support -✅ Separated CLI functionality from core business logic for better testability -✅ Added AWS SSO profile support for credential management -✅ Added multi-region discovery and support for checking all regions per account - -2. Error Handling and Logging -The logging implementation could be enhanced: - -2.1 Structured Logging: -- Expand the existing JSONFormatter class in logging_setup.py to include additional context: - - Add AWS-specific fields: account_id, region, resource_type, resource_id, request_id - - Include operation context: operation_name, operation_status, duration_ms - - Add correlation_id field to track related log entries across a single operation - - Support custom context through thread-local storage or LogRecord extra parameter - - Include hostname, process ID, and thread ID for distributed debugging - -- Implementation approach: - - Create a LoggingContext class to store and manage contextual information - - Add a log_with_context() utility function that includes current context - - Update the JSONFormatter.format() method to incorporate context - - Add context manager for tracking operation timing and results - - Create logging filter to automatically enrich logs with AWS context - -- Configuration improvements: - - Add support for different log formats per handler (JSON for files, human-readable for console) - - Include log rotation by size and time with configurable retention - - Support configurable log destinations (file, console, syslog, CloudWatch Logs) - - Add option to disable certain context fields for privacy/security - -- Sample implementation structure: - ```python - # Context management - class LoggingContext: - _context = threading.local() - @classmethod - def set(cls, **kwargs): ... - @classmethod - def get(cls): ... - @classmethod - def clear(cls): ... - - # Context manager for operations - @contextmanager - def log_operation(name, **context): - start = time.time() - LoggingContext.set(operation_name=name, **context) - try: - yield - duration = (time.time() - start) * 1000 - LoggingContext.set(operation_status="success", duration_ms=duration) - logger.info(f"Operation {name} completed successfully") - except Exception as e: - duration = (time.time() - start) * 1000 - LoggingContext.set(operation_status="failed", duration_ms=duration) - logger.error(f"Operation {name} failed: {e}") - raise - finally: - LoggingContext.clear() - ``` - -- Handling sensitive information: - - Create a SanitizingFormatter that masks sensitive fields (access keys, passwords) - - Define a list of sensitive field patterns to detect and mask - - Use regex patterns to identify potential sensitive data in unstructured log messages - - Add configuration options to control sanitization levels - -2.2 Granular Log Levels: -- Implementation details for each log level: - - DEBUG: - - AWS API request/response bodies and headers - - Parameter values for all functions - - Resource discovery detailed progress - - Authentication token acquisition - - Configuration loading and resolution steps - - Connection attempts and retries - - - INFO: - - Resource operations start/completion (e.g., "Starting EC2 instance i-123456") - - Resource state changes (e.g., "RDS instance changed to 'stopping'") - - Important configuration values loaded - - Number of resources discovered/affected - - Operation timing information - - Profile and region information in use - - - WARNING: - - Resources excluded due to tags - - Missing optional configuration - - Slow API responses - - Resources in unexpected states - - Deprecated API usage - - Retrying operations after recoverable errors - - Missing tags or metadata - - - ERROR: - - Failed operations on specific resources - - Authentication or permission issues - - Invalid parameters or configurations - - API rate limiting or throttling - - Network connectivity issues to specific regions - - Resource not found - - - CRITICAL: - - Complete failure to authenticate - - Fatal configuration errors - - Inability to access any AWS services - - Unrecoverable program state - -- Implementation approach: - - Create logging helper functions for consistent usage: - ```python - def log_resource_operation(logger, level, operation, resource_type, resource_id, - region, details=None, exception=None): - """Log resource operations with consistent structure.""" - msg = f"{operation} {resource_type} {resource_id} in {region}" - extra = { - "resource_type": resource_type, - "resource_id": resource_id, - "operation": operation, - "region": region - } - if details: - extra["details"] = details - msg += f": {details}" - if exception: - extra["exception"] = str(exception) - msg += f" (failed: {exception})" - logger.log(level, msg, extra=extra) - ``` - - - Add log level configuration: - - Command line option to override log level - - Environment variable for log level - - Configuration file setting - - Different log levels for console vs file output - - - Create module-level logger constants with appropriate levels: - ```python - # Base module logger - logger = logging.getLogger('aws_resource_management') - - # Component-specific loggers with potential different levels - api_logger = logger.getChild('api') # For AWS API calls - discovery_logger = logger.getChild('discovery') # For resource discovery - operation_logger = logger.getChild('operation') # For resource operations - ``` - - - Add utility function to adjust verbosity: - ```python - def set_verbosity(level_name: str) -> None: - """ - Set the verbosity level of the application. - - Arguments: - level_name: One of 'debug', 'info', 'warning', 'error', 'critical' - """ - level = getattr(logging, level_name.upper()) - logger.setLevel(level) - - # Adjust component loggers as needed - if level <= logging.DEBUG: - api_logger.setLevel(logging.DEBUG) # Always show API details in debug mode - else: - api_logger.setLevel(logging.INFO) # Otherwise just show API operation summaries - ``` - -2.3 Error Handling Module: -- Create aws_errors.py module with custom exception hierarchy: - - BaseAWSResourceError (base exception): - - Common attributes for all errors: error_code, message, details - - Methods for serializing errors to JSON - - Support for contextual information (account_id, region, etc.) - - - ConfigurationError (config issues): - - Missing required configuration - - Invalid configuration values - - Configuration file parsing errors - - Environment variable conflicts - - - AuthenticationError (credential issues): - - Invalid credentials - - Expired credentials - - Missing credentials - - Insufficient permissions - - - ResourceOperationError (resource actions): - - Failed state transitions - - Resource in incompatible state - - Resource already in target state - - Dependencies preventing operation - - Resource-specific error subclasses (EC2Error, RDSError, etc.) - - - NetworkError (connectivity issues): - - Connection timeouts - - Rate limiting/throttling - - DNS resolution failures - - Endpoint unavailability - - - ValidationError (input validation): - - Invalid parameter types - - Value range violations - - Missing required parameters - - Incompatible parameter combinations - -- Implementation approach: - - Define error codes for each exception type: - ```python - class ErrorCode(Enum): - # General errors (1-99) - UNKNOWN_ERROR = 1 - INTERNAL_ERROR = 2 - - # Configuration errors (100-199) - CONFIG_NOT_FOUND = 100 - CONFIG_PARSE_ERROR = 101 - CONFIG_VALIDATION_ERROR = 102 - - # Authentication errors (200-299) - AUTH_INVALID_CREDENTIALS = 200 - AUTH_EXPIRED_CREDENTIALS = 201 - AUTH_INSUFFICIENT_PERMISSIONS = 202 - - # Resource operation errors (300-399) - RESOURCE_NOT_FOUND = 300 - RESOURCE_STATE_CONFLICT = 301 - RESOURCE_DEPENDENCY_ERROR = 302 - - # Service-specific error ranges - # EC2 errors (1000-1099) - EC2_INVALID_STATE = 1000 - EC2_INSTANCE_NOT_FOUND = 1001 - - # RDS errors (1100-1199) - RDS_INVALID_STATE = 1100 - RDS_INSTANCE_NOT_FOUND = 1101 - ``` - - - Add contextual information to exceptions: - ```python - class BaseAWSResourceError(Exception): - """Base exception for all AWS Resource Management errors.""" - - def __init__( - self, - message: str, - error_code: ErrorCode = ErrorCode.UNKNOWN_ERROR, - account_id: Optional[str] = None, - region: Optional[str] = None, - resource_type: Optional[str] = None, - resource_id: Optional[str] = None, - details: Optional[Dict[str, Any]] = None, - original_exception: Optional[Exception] = None - ): - self.message = message - self.error_code = error_code - self.account_id = account_id - self.region = region - self.resource_type = resource_type - self.resource_id = resource_id - self.details = details or {} - self.original_exception = original_exception - - # Build detailed error message - detailed_msg = f"[{error_code.name}] {message}" - if resource_type and resource_id: - detailed_msg += f" (Resource: {resource_type}/{resource_id})" - if region: - detailed_msg += f" in region {region}" - - super().__init__(detailed_msg) - - def to_dict(self) -> Dict[str, Any]: - """Convert exception to a dictionary for serialization.""" - result = { - "error_code": self.error_code.value, - "error_name": self.error_code.name, - "message": self.message, - } - - # Add contextual information if available - for field in ["account_id", "region", "resource_type", "resource_id"]: - if getattr(self, field): - result[field] = getattr(self, field) - - # Add any additional details - if self.details: - result["details"] = self.details - - return result - ``` - - - Add utility functions for error handling: - ```python - def handle_boto3_error( - boto3_error: botocore.exceptions.ClientError, - resource_type: Optional[str] = None, - resource_id: Optional[str] = None, - region: Optional[str] = None, - operation: Optional[str] = None - ) -> BaseAWSResourceError: - """Convert boto3 errors to our custom exceptions with proper context.""" - error_code = boto3_error.response.get("Error", {}).get("Code", "Unknown") - error_message = boto3_error.response.get("Error", {}).get("Message", str(boto3_error)) - - # Map boto3 error codes to our error codes - if error_code == "AuthFailure": - return AuthenticationError( - f"Authentication failed: {error_message}", - error_code=ErrorCode.AUTH_INVALID_CREDENTIALS, - resource_type=resource_type, - resource_id=resource_id, - region=region, - details={"operation": operation}, - original_exception=boto3_error - ) - - # Handle throttling/rate limiting - if error_code == "Throttling" or error_code == "ThrottlingException": - return NetworkError( - f"API rate limit exceeded: {error_message}", - error_code=ErrorCode.API_THROTTLING, - resource_type=resource_type, - resource_id=resource_id, - region=region, - details={"operation": operation}, - original_exception=boto3_error - ) - - # Default to ResourceOperationError for other cases - return ResourceOperationError( - f"AWS operation failed: {error_message}", - error_code=ErrorCode.from_boto3_code(error_code), - resource_type=resource_type, - resource_id=resource_id, - region=region, - details={ - "operation": operation, - "boto3_error_code": error_code - }, - original_exception=boto3_error - ) - ``` - -- Error code mapping: - - Map AWS/boto3 error codes to internal error codes - - Create utility to convert between different error representations - - Support HTTP status codes for API responses - -- Integration with logging: - - Log exceptions with appropriate levels based on severity - - Include stack traces for unexpected errors - - Provide consistent error reporting format - -2.4 Stack Trace and Debug Information: -- Comprehensive exception handling: - - Use `traceback` module to capture and format stack traces - - Include stack trace context in log messages at DEBUG level - - Create wrapper functions for common try/except patterns - - Record exception chain for nested exceptions - -- Implementation approach for stack trace logging: - ```python - def log_exception( - logger, - exc: Exception, - msg: str = "An exception occurred", - level: int = logging.ERROR, - include_traceback: bool = True, - include_cause_chain: bool = True, - context: Optional[Dict[str, Any]] = None - ) -> None: - """ - Log an exception with stack trace at the specified level. - - Args: - logger: Logger to use - exc: The exception to log - msg: Message to log with the exception - level: Log level to use - include_traceback: Whether to include the stack trace - include_cause_chain: Whether to include cause chain for chained exceptions - context: Additional context to include in the log - """ - exc_info = sys.exc_info() if include_traceback else None - - # Extract info from exception - extra = context or {} - extra['exception_type'] = exc.__class__.__name__ - extra['exception_msg'] = str(exc) - - # Format cause chain if requested and available - if include_cause_chain and exc.__cause__: - causes = [] - current = exc.__cause__ - while current: - causes.append(f"{current.__class__.__name__}: {str(current)}") - current = current.__cause__ - - extra['exception_causes'] = causes - - # Log the exception - logger.log(level, msg, exc_info=exc_info, extra=extra) - ``` - -- AWS API request/response logging: - - Create a boto3 event hook for logging API calls - - Log request parameters and response data - - Sanitize sensitive information in requests/responses - - Track AWS request IDs for correlation - - Calculate and log API latency metrics - -- Implementation approach for boto3 request logging: - ```python - def add_request_logging_to_session(session: boto3.Session, log_level: int = logging.DEBUG) -> None: - """ - Add request/response logging hooks to a boto3 session. - - Args: - session: boto3 Session to instrument - log_level: Log level to use for API requests/responses - """ - api_logger = logging.getLogger('aws_resource_management.api') - - def log_request(request, **kwargs): - # Create a unique ID for this request for correlation - request_id = str(uuid.uuid4()) - request.context['request_log_id'] = request_id - - # Get sanitized parameters (remove sensitive info) - params = _sanitize_params(request.context.get('params', {})) - - # Log the request details - api_logger.log( - log_level, - f"AWS API Request: {request.context.get('operation_name')}", - extra={ - 'request_log_id': request_id, - 'service': request.context.get('client_meta').service_model.service_name, - 'operation': request.context.get('operation_name'), - 'params': params, - 'region': request.context.get('client_region'), - 'endpoint_url': request.context.get('client_meta').endpoint_url, - } - ) - - def log_response(request, response, **kwargs): - # Get the request ID we set earlier - request_id = getattr(request.context, 'request_log_id', 'unknown') - - # Get AWS request ID from response - aws_request_id = response.get('ResponseMetadata', {}).get('RequestId', 'unknown') - - # Calculate latency - latency_ms = response.get('ResponseMetadata', {}).get('HTTPHeaders', {}).get('x-amz-request-latency', -1) - - # Log the response - api_logger.log( - log_level, - f"AWS API Response: {request.context.get('operation_name')}", - extra={ - 'request_log_id': request_id, - 'aws_request_id': aws_request_id, - 'status_code': response.get('ResponseMetadata', {}).get('HTTPStatusCode'), - 'latency_ms': latency_ms, - 'service': request.context.get('client_meta').service_model.service_name, - 'operation': request.context.get('operation_name'), - } - ) - - # Register the event handlers with boto3 - session.events.register('before-send.*.*.', log_request) - session.events.register('after-call.*.*.', log_response) - ``` - -- Utility functions for error handling and reporting: - - Create consistent error message formatting - - Extract useful error information from boto3 exceptions - - Format error messages for CLI output vs. logs - - Handle known error patterns with custom messages - - Retrieve error documentation links for common errors - -- Implementation approach for error reporting: - ```python - def format_error_for_display( - error: Exception, - verbose: bool = False, - include_help: bool = True - ) -> str: - """ - Format an error for display to users. - - Args: - error: The exception to format - verbose: Whether to include detailed information - include_help: Whether to include help text for known errors - - Returns: - Formatted error message - """ - # Handle our custom exceptions - if isinstance(error, BaseAWSResourceError): - message = f"Error ({error.error_code.name}): {error.message}" - - # Add context for resource errors if available - if error.resource_type and error.resource_id: - message += f"\nResource: {error.resource_type}/{error.resource_id}" - if error.region: - message += f"\nRegion: {error.region}" - - # Add details for verbose mode - if verbose and error.details: - message += "\nDetails:" - for k, v in error.details.items(): - message += f"\n {k}: {v}" - - # Add help text for known errors - if include_help: - help_text = ERROR_HELP_TEXT.get(error.error_code) - if help_text: - message += f"\n\nTroubleshooting:\n{help_text}" - - return message - - # Handle boto3 ClientError - if isinstance(error, botocore.exceptions.ClientError): - error_code = error.response.get("Error", {}).get("Code", "Unknown") - error_message = error.response.get("Error", {}).get("Message", str(error)) - - message = f"AWS Error ({error_code}): {error_message}" - - # Add request ID if available for troubleshooting - request_id = error.response.get("ResponseMetadata", {}).get("RequestId") - if request_id and verbose: - message += f"\nAWS Request ID: {request_id}" - - # Add help for common boto3 errors - if include_help: - help_text = BOTO3_ERROR_HELP.get(error_code) - if help_text: - message += f"\n\nTroubleshooting:\n{help_text}" - - return message - - # Default case for other exceptions - return f"Error: {str(error)}" - ``` - -- Retry mechanism with exponential backoff: - - Implement decorator for retryable functions - - Configure retry count, delay, and backoff factor - - Define which exceptions are retryable - - Log retry attempts and backoff periods - - Customize backoff strategy (exponential, linear, etc.) - -- Implementation approach for retry decorator: - ```python - def retry_with_backoff( - max_retries: int = 3, - initial_backoff: float = 1.0, - backoff_factor: float = 2.0, - max_backoff: float = 30.0, - retryable_exceptions: Tuple[Type[Exception], ...] = ( - botocore.exceptions.ClientError, - requests.exceptions.RequestException, - ), - should_retry_cb: Optional[Callable[[Exception], bool]] = None - ): - """ - Decorator for retrying functions with exponential backoff. - - Args: - max_retries: Maximum number of retries - initial_backoff: Initial backoff time in seconds - backoff_factor: Factor to multiply backoff by after each retry - max_backoff: Maximum backoff time in seconds - retryable_exceptions: Tuple of exceptions that should trigger a retry - should_retry_cb: Optional callback to determine if an exception is retryable - - Returns: - Decorated function - """ - def decorator(func): - @functools.wraps(func) - def wrapper(*args, **kwargs): - retry_logger = logging.getLogger('aws_resource_management.retry') - - retry_count = 0 - backoff = initial_backoff - - while True: - try: - return func(*args, **kwargs) - except retryable_exceptions as e: - # Check if we should retry this specific exception - should_retry = True - if should_retry_cb: - should_retry = should_retry_cb(e) - - # If we've reached max retries or shouldn't retry, re-raise - if retry_count >= max_retries or not should_retry: - raise - - # Calculate backoff with jitter for this retry - jittered_backoff = backoff * (0.9 + 0.2 * random.random()) - actual_backoff = min(jittered_backoff, max_backoff) - - # Log the retry attempt - retry_logger.warning( - f"Retrying {func.__name__} after error: {str(e)}. " - f"Retry {retry_count + 1}/{max_retries} in {actual_backoff:.2f}s" - ) - - # Wait before retrying - time.sleep(actual_backoff) - - # Update for next iteration - retry_count += 1 - backoff = backoff * backoff_factor - return wrapper - return decorator - ``` - -3. Testing and Quality Assurance -The codebase would benefit from: - -Unit tests for core functionality -Integration tests for AWS interactions (using moto library) -Implement pre-commit hooks for code quality checks -Add CI/CD pipeline for automated testing -4. Enhanced Features -Several valuable features could be added: - -Support for resource tagging based on schedule -Cost calculation for resources being managed -Additional resource types (Lambda, S3, DynamoDB) -Resource dependency management -Schedule-based operations (cron-like syntax) -Consistent signal handling (ensuring a single interrupt exits the application) -✅ AWS SSO Profile support for authentication without role assumption: - - Automatic detection and use of AWS SSO profiles in ~/.aws/config - - Support for specifying specific profiles with --profile option - - Prioritization of administrator profiles for each account - - Fallback to role assumption when profiles are not available -✅ Multi-region discovery and management with smart defaults: - - Auto-discovery of all enabled regions per account - - Support for region filtering with include/exclude patterns - - Parallel region checking for improved performance - - Partition-aware region discovery (commercial AWS, GovCloud, China) - - Process all regions in a single account before moving to the next account - - Complete account-level processing to maintain context and improve error handling - - Improved error handling for authentication failures during region discovery - - Graceful degradation to default regions when discovery fails - - Timeout handling for region discovery to prevent hanging operations - - Thread-safe region discovery with proper exception handling - -5. Security Improvements -Security could be strengthened by: - -Parameter validation and sanitization -Secret management for credentials -Role assumption with MFA support -Audit logging of all actions -Access control based on AWS profiles and roles -Enhanced credential management: - - Support for credential chaining (try profile, then role assumption) - - Automated credential refresh for long-running operations - - Validation of permissions before operations - - Secure handling of temporary credentials - -6. Documentation -Documentation could be improved with: - -Comprehensive API documentation -User guides with common scenarios -Architecture diagrams -Contributing guidelines