From 2a56e6c4c76a9645ad190c7e7564434967ed5b67 Mon Sep 17 00:00:00 2001 From: "Matthew C. Morgan" Date: Wed, 2 Apr 2025 23:47:00 -0400 Subject: [PATCH] cleanup more --- .../aws_resource_management/__init__.py | 17 +- .../aws_resource_management/aws_utils.py | 188 ++++++++-- .../aws_resource_management/config_manager.py | 162 ++++----- .../aws_resource_management/discovery.py | 156 +++++--- .../aws_resource_management/logging_setup.py | 257 +++++++++----- .../managers/__init__.py | 40 ++- .../aws_resource_management/managers/base.py | 332 ++++++------------ .../gfl-resource-actions/logging_utils.py | 312 ---------------- 8 files changed, 636 insertions(+), 828 deletions(-) delete mode 100644 local-app/python-tools/gfl-resource-actions/logging_utils.py diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/__init__.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/__init__.py index 7694c236..9b891146 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/__init__.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/__init__.py @@ -1,8 +1,19 @@ """ AWS Resource Management package. - -This package provides tools for managing AWS resources including stopping, -starting, and listing resources across different AWS services. """ +# Re-export key modules for easier imports +from aws_resource_management.logging_setup import ( + LoggingContext, + configure_logging, + initialize_csv_log, + log_action_to_csv, + log_operation, + log_with_context, + setup_logging, +) + +# Set up a default logger +logger = setup_logging() + __version__ = "0.1.0" 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 e39baf71..2052acce 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 @@ -8,11 +8,13 @@ import os import threading import time +from functools import lru_cache from typing import Any, Dict, List, Optional, Set, Tuple, Union import boto3 import botocore from aws_resource_management.logging_setup import setup_logging +from botocore.exceptions import ClientError, NoCredentialsError, ProfileNotFound logger = setup_logging() @@ -90,27 +92,22 @@ def safe_in(substring: Any, container: Any) -> bool: # --------------------------------------------------------------------------- -def get_partition_for_region(region: Optional[str]) -> str: - """Get the AWS partition for a region (can be None).""" - # Default for GovCloud environment - DEFAULT_PARTITION = "aws-us-gov" +@lru_cache(maxsize=128) +def get_partition_for_region(region_name: str) -> str: + """ + Determine AWS partition based on region name. - # Early return for None or non-string regions - if not is_string(region): - return DEFAULT_PARTITION + Args: + region_name: AWS region name - # Map region prefixes to partitions using simple lookups - if region.startswith("us-gov-"): + Returns: + AWS partition name (e.g., 'aws', 'aws-us-gov', 'aws-cn') + """ + if region_name.startswith("us-gov-"): return "aws-us-gov" - elif region.startswith("cn-"): + elif region_name.startswith("cn-"): return "aws-cn" - elif any( - region.startswith(prefix) - for prefix in ("us-", "eu-", "ap-", "ca-", "sa-", "af-") - ): - return "aws" - - return DEFAULT_PARTITION + return "aws" def get_regions_for_partition(partition: str) -> List[str]: @@ -353,15 +350,26 @@ def detect_partition_from_credentials(credentials: Dict[str, str]) -> str: # --------------------------------------------------------------------------- +@lru_cache(maxsize=128) def get_session_for_account( account_id: str, region: Optional[str] = None, profile_name: Optional[str] = None ) -> Optional[boto3.Session]: - """Get a boto3 session with improved caching to minimize API calls.""" + """ + Get a boto3 session for the specified account with proper role assumption. + + Args: + account_id: AWS account ID + region: AWS region name (optional) + profile_name: AWS profile name to use (optional) + + Returns: + Boto3 session with appropriate credentials + """ cache_key = ( f"session:{account_id}:{region or 'default'}:{profile_name or 'default'}" ) - # Check cache first + # Return cached session if available with _session_cache_lock: if ( cache_key in _session_cache @@ -369,28 +377,93 @@ def get_session_for_account( ): return _session_cache[cache_key]["session"] - # Get credentials - credentials = get_credentials(account_id, profile_name) - if not credentials: - return None - - # Create session + # Try getting a session using standard methods try: - session = boto3.Session( - aws_access_key_id=credentials["aws_access_key_id"], - aws_secret_access_key=credentials["aws_secret_access_key"], - aws_session_token=credentials.get("aws_session_token"), - region_name=region, - ) + # Method 1: Try using a profile named after the account ID or the specified profile + try: + session = boto3.Session( + profile_name=profile_name or account_id, region_name=region + ) + # Validate session by checking identity + sts = session.client("sts") + identity = sts.get_caller_identity() + if identity.get("Account") == account_id: + logger.debug( + f"Using profile {profile_name or account_id} for authentication" + ) - # Cache the session - with _session_cache_lock: - _session_cache[cache_key] = {"session": session, "timestamp": time.time()} + # Cache the session + with _session_cache_lock: + _session_cache[cache_key] = { + "session": session, + "timestamp": time.time(), + } + + return session + except (ProfileNotFound, NoCredentialsError, ClientError): + # Profile not found or invalid, continue to next method + pass + + # Method 2: Use credentials to create a session + credentials = get_credentials(account_id, profile_name) + if credentials: + session = boto3.Session( + aws_access_key_id=credentials["aws_access_key_id"], + aws_secret_access_key=credentials["aws_secret_access_key"], + aws_session_token=credentials.get("aws_session_token"), + region_name=region, + ) + + # Cache the session + with _session_cache_lock: + _session_cache[cache_key] = { + "session": session, + "timestamp": time.time(), + } + + logger.debug(f"Created session for account {account_id} using credentials") + return session + + # Method 3: Try to assume role in target account using default session + default_session = boto3.Session(region_name=region) + sts = default_session.client("sts") + + # Try common cross-account roles + for role_name in ["OrganizationAccountAccessRole", "AWSControlTowerExecution"]: + try: + role_arn = f"arn:aws:iam::{account_id}:role/{role_name}" + response = sts.assume_role( + RoleArn=role_arn, RoleSessionName="ResourceManagementSession" + ) + + # Create session with temporary credentials + credentials = response["Credentials"] + session = boto3.Session( + aws_access_key_id=credentials["AccessKeyId"], + aws_secret_access_key=credentials["SecretAccessKey"], + aws_session_token=credentials["SessionToken"], + region_name=region, + ) + + # Cache the session + with _session_cache_lock: + _session_cache[cache_key] = { + "session": session, + "timestamp": time.time(), + } + + logger.debug(f"Assumed {role_name} in account {account_id}") + return session + except ClientError: + # Role assumption failed, try the next role + continue + + # If all methods fail, raise exception + raise Exception(f"Could not authenticate to account {account_id}") - return session except Exception as e: - logger.error(f"Error creating session for account {account_id}: {str(e)}") - return None + logger.error(f"Failed to get session for account {account_id}: {str(e)}") + raise # --------------------------------------------------------------------------- @@ -559,6 +632,47 @@ def get_enabled_regions( return get_regions_for_partition(partition) +def list_enabled_regions( + session: boto3.Session, exclude_regions: Optional[List[str]] = None +) -> List[str]: + """ + List all enabled regions for the account, respecting partition. + + Args: + session: Boto3 session + exclude_regions: List of regions to exclude + + Returns: + List of enabled region names + """ + if exclude_regions is None: + exclude_regions = [] + + try: + ec2 = session.client( + "ec2", region_name="us-east-1" + ) # Use US East 1 for global operations + regions_response = ec2.describe_regions( + AllRegions=False + ) # Only get enabled regions + + regions = [ + r["RegionName"] + for r in regions_response["Regions"] + if r["RegionName"] not in exclude_regions + ] + return regions + except Exception as e: + logger.warning(f"Could not retrieve regions, using defaults: {str(e)}") + + # For GovCloud, provide sensible defaults + if session.region_name and session.region_name.startswith("us-gov-"): + return ["us-gov-east-1", "us-gov-west-1"] + + # Default to common commercial regions + return ["us-east-1", "us-east-2", "us-west-1", "us-west-2"] + + def is_valid_region(region: str) -> bool: """Check if a region is valid without making an API call.""" if not is_string(region): diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/config_manager.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/config_manager.py index e7a42885..5a00ba46 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/config_manager.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/config_manager.py @@ -1,6 +1,5 @@ """ -Configuration management for the AWS Resource Management tool. -Handles loading configuration from files and environment variables. +Configuration manager for AWS Resource Management. """ import os @@ -9,113 +8,76 @@ import yaml -# Default configuration values +# Default configuration DEFAULT_CONFIG = { - "exclusion_tag": "gliffy:exclude", - "stop_tag": "gliffy:stopped-at", + "action_csv_file": "resource_actions.csv", "log_level": "INFO", - "default_region": "us-gov-east-1", - "max_retries": 3, - "scheduled_tag": "gliffy:schedule", - "dry_run": False, + "max_workers": 10, + "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"], + }, } -# Environment variable prefix for overriding configuration -ENV_PREFIX = "AWS_RESOURCE_MGMT_" +# Global config cache +_config = None -class ConfigManager: - """ - Configuration manager for AWS Resource Management tool. - Handles loading config from files and environment variables. +def get_config() -> Dict[str, Any]: """ + Get configuration with defaults merged with user settings. - _instance = None - _config = None - - def __new__(cls): - """Singleton pattern to ensure only one config instance.""" - if cls._instance is None: - cls._instance = super(ConfigManager, cls).__new__(cls) - cls._instance._config = DEFAULT_CONFIG.copy() - return cls._instance - - def load_config(self, config_file: Optional[str] = None) -> Dict[str, Any]: - """ - Load configuration from file and environment variables. - - Args: - config_file: Path to the config file (YAML) - - Returns: - Configuration dictionary - """ - # Start with defaults - self._config = DEFAULT_CONFIG.copy() - - # Load from config file if provided - if config_file and Path(config_file).exists(): - try: - with open(config_file, "r") as f: - file_config = yaml.safe_load(f) - if file_config and isinstance(file_config, dict): - self._config.update(file_config) - except Exception as e: - print(f"Error loading config file: {e}") - - # Override with environment variables - for key in self._config.keys(): - env_key = f"{ENV_PREFIX}{key.upper()}" - if env_key in os.environ: - # Convert environment variable to appropriate type - env_value = os.environ[env_key] - if isinstance(self._config[key], bool): - self._config[key] = env_value.lower() in ("true", "yes", "1") - elif isinstance(self._config[key], int): - self._config[key] = int(env_value) - else: - self._config[key] = env_value - - return self._config - - def get(self, key: str, default: Any = None) -> Any: - """ - Get a configuration value by key. - - Args: - key: Configuration key - default: Default value if key doesn't exist - - Returns: - Configuration value - """ - return self._config.get(key, default) - - def set(self, key: str, value: Any) -> None: - """ - Set a configuration value. - - Args: - key: Configuration key - value: Configuration value - """ - self._config[key] = value - - def get_all(self) -> Dict[str, Any]: - """ - Get the entire configuration dictionary. - - Returns: - Configuration dictionary - """ - return self._config.copy() - - -def get_config() -> ConfigManager: + Returns: + Configuration dictionary """ - Get the configuration manager instance. + global _config + + # Return cached config if available + if _config is not None: + return _config + + # Start with default config + config = DEFAULT_CONFIG.copy() + + # Look for config files in multiple locations + config_paths = [ + os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "config.yaml"), + os.path.expanduser("~/.aws-resource-management/config.yaml"), + ] + + # Try to load and merge configs from files + for path in config_paths: + try: + if os.path.exists(path): + with open(path, "r") as f: + user_config = yaml.safe_load(f) + if user_config and isinstance(user_config, dict): + # Merge with default config + _deep_merge(config, user_config) + except Exception: + # Ignore errors reading config files + pass + + # Cache the config + _config = config + return config + + +def _deep_merge(base: Dict, update: Dict) -> Dict: + """ + Recursively merge dictionaries. + + Args: + base: Base dictionary to update + update: Dictionary with values to merge Returns: - ConfigManager instance + Updated base dictionary """ - return ConfigManager() + for key, value in update.items(): + if isinstance(value, dict) and key in base and isinstance(base[key], dict): + _deep_merge(base[key], value) + else: + base[key] = value + return base 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 175f37f5..bb9d8d7f 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,31 +1,47 @@ """ -Resource discovery module for finding AWS resources. +Resource discovery functionality. """ -import concurrent.futures +import logging from collections import defaultdict from concurrent.futures import ThreadPoolExecutor, as_completed -from typing import Any, Dict, List, Optional, Union +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_session_for_account, is_valid_region, + list_enabled_regions, ) from aws_resource_management.config_manager import get_config -from aws_resource_management.logging_setup import setup_logging +from aws_resource_management.logging_setup import log_with_context, setup_logging from aws_resource_management.managers import ( + BaseResourceManager, EC2Manager, - ECRManager, - EKSManager, - EMRManager, RDSManager, ) from botocore.exceptions import ClientError, EndpointConnectionError -logger = setup_logging() +# Try to import optional managers +try: + from aws_resource_management.managers import EKSManager +except ImportError: + EKSManager = None + +try: + from aws_resource_management.managers import EMRManager +except ImportError: + EMRManager = None + +try: + from aws_resource_management.managers import ECRManager +except ImportError: + ECRManager = None + +logger = logging.getLogger("aws_resource_management") config = get_config() # Default regions for different partitions @@ -136,9 +152,7 @@ def discover_resources( # Use thread pool to speed up discovery across regions resources = [] - with concurrent.futures.ThreadPoolExecutor( - max_workers=min(10, len(region_list)) - ) as executor: + with ThreadPoolExecutor(max_workers=min(10, len(region_list))) as executor: # Create a future for each region future_to_region = { executor.submit(discovery_method, region, resource_ids): region @@ -146,7 +160,7 @@ def discover_resources( } # Process results as they complete - for future in concurrent.futures.as_completed(future_to_region): + for future in as_completed(future_to_region): region = future_to_region[future] try: region_resources = future.result() @@ -497,16 +511,28 @@ def _discover_eks( def get_account_resources( - credentials: Dict[str, str], - regions: List[str], - exclude_resources: List[str], account_id: str, + regions: List[str], + exclude_resources: List[str] = None, account_name: Optional[str] = None, emr_cluster_states: Optional[List[str]] = None, ) -> Dict[str, List[Dict[str, Any]]]: """ - Get all resources for an account across multiple regions. + Get all resources for an account across multiple regions using the ResourceDiscovery class. + + Args: + account_id: AWS account ID + regions: List of AWS regions to search + exclude_resources: List of resource types to exclude + account_name: AWS account name (optional) + emr_cluster_states: EMR cluster states to include (optional) + + Returns: + Dictionary of resource lists by type """ + if exclude_resources is None: + exclude_resources = [] + resources = { "ec2_instances": [], "rds_instances": [], @@ -516,48 +542,68 @@ def get_account_resources( "ecr_old_images": [], } - # Discover EC2 instances - if "ec2" not in exclude_resources: - logger.info( - f"Discovering EC2 instances for account {account_id} in {len(regions)} regions" - ) - ec2_manager = EC2Manager(credentials, account_id, account_name) - resources["ec2_instances"] = ec2_manager.discover_instances(regions) + try: + # Get session for the account + session = get_session_for_account(account_id) + if not session: + logger.error(f"Could not create session for account {account_id}") + return resources + + # Create resource discovery instance + from aws_resource_management.aws_utils import get_credentials + + credentials = { + "aws_access_key_id": session.get_credentials().access_key, + "aws_secret_access_key": session.get_credentials().secret_key, + "aws_session_token": ( + session.get_credentials().token + if session.get_credentials().token + else None + ), + } + + discovery = ResourceDiscovery(credentials, account_id) + + # Discover EC2 instances + if "ec2" not in exclude_resources: + logger.info( + f"Discovering EC2 instances for account {account_id} in {len(regions)} regions" + ) + resources["ec2_instances"] = discovery.discover_resources("ec2", regions) - # Discover RDS instances - if "rds" not in exclude_resources: - logger.info( - f"Discovering RDS instances for account {account_id} in {len(regions)} regions" - ) - rds_manager = RDSManager(credentials, account_id, account_name) - resources["rds_instances"] = rds_manager.discover_instances(regions) + # Discover RDS instances + if "rds" not in exclude_resources: + logger.info( + f"Discovering RDS instances for account {account_id} in {len(regions)} regions" + ) + resources["rds_instances"] = discovery.discover_resources("rds", regions) - # Discover EKS clusters - if "eks" not in exclude_resources: - logger.info( - f"Discovering EKS clusters for account {account_id} in {len(regions)} regions" - ) - eks_manager = EKSManager(credentials, account_id, account_name) - resources["eks_clusters"] = eks_manager.discover_clusters(regions) + # Discover EKS clusters + if "eks" not in exclude_resources and EKSManager: + logger.info( + f"Discovering EKS clusters for account {account_id} in {len(regions)} regions" + ) + resources["eks_clusters"] = discovery.discover_resources("eks", regions) - # Discover EMR clusters - if "emr" not in exclude_resources: - logger.info( - f"Discovering EMR clusters for account {account_id} in {len(regions)} regions" - ) - emr_manager = EMRManager(credentials, account_id, account_name) - resources["emr_clusters"] = emr_manager.discover_clusters( - regions, emr_cluster_states - ) + # Discover EMR clusters + if "emr" not in exclude_resources and EMRManager: + logger.info( + f"Discovering EMR clusters for account {account_id} in {len(regions)} regions" + ) + resources["emr_clusters"] = discovery.discover_resources("emr", regions) - # Discover ECR images - if "ecr" not in exclude_resources: - logger.info( - f"Discovering ECR images for account {account_id} in {len(regions)} regions" - ) - ecr_manager = ECRManager(credentials, account_id, account_name) - result = ecr_manager.discover_images_by_age(regions) - resources["ecr_images"] = result["all_images"] - resources["ecr_old_images"] = result["old_images"] + # Discover ECR images + if "ecr" not in exclude_resources and ECRManager: + 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"] = [] + + except Exception as e: + logger.error(f"Error discovering resources for account {account_id}: {e}") return resources 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 2969d039..8e67c482 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,5 +1,5 @@ """ -Logging setup for AWS Resource Management tool. +Enhanced logging utilities for AWS Resource Management. """ import csv @@ -8,139 +8,228 @@ import logging import os import sys +import time +from contextlib import contextmanager from pathlib import Path -from typing import Any, Dict, Optional +from typing import Any, Callable, Dict, Optional from aws_resource_management.config_manager import get_config -class JSONFormatter(logging.Formatter): - """ - JSON formatter for structured logging. - """ +# Global context data that can be included in log messages +class LoggingContext: + _context = {} - def format(self, record: logging.LogRecord) -> str: - """ - Format the log record as a JSON string. + @classmethod + def get(cls) -> Dict[str, Any]: + """Get the current context dictionary.""" + return cls._context.copy() + + @classmethod + def set(cls, **kwargs) -> None: + """Set context values.""" + cls._context.update(kwargs) + + @classmethod + def clear(cls) -> None: + """Clear the context.""" + cls._context.clear() - Args: - record: Log record - Returns: - JSON formatted string - """ +class JSONFormatter(logging.Formatter): + """Format log records as JSON.""" + + def format(self, record: logging.LogRecord) -> str: + """Format the log record as JSON.""" log_data = { - "timestamp": datetime.fromtimestamp(record.created).isoformat(), + "timestamp": self.formatTime(record), "level": record.levelname, + "logger": record.name, "message": record.getMessage(), - "module": record.module, - "function": record.funcName, - "line": record.lineno, } - # Include exception info if present + # Add exception info if present if record.exc_info: - log_data["exception"] = { - "type": record.exc_info[0].__name__, - "message": str(record.exc_info[1]), - } + log_data["exception"] = self.formatException(record.exc_info) + + # Add context data if present + if hasattr(record, "context") and record.context: + log_data.update(record.context) - # Include any custom fields - for key, value in getattr(record, "extra_fields", {}).items(): - log_data[key] = value + # Add LoggingContext data + log_data.update(LoggingContext.get()) return json.dumps(log_data) -def setup_logging( - log_level: Optional[str] = None, +def setup_logging(name="aws_resource_management", level=logging.INFO) -> logging.Logger: + """Set up basic logging.""" + logger = logging.getLogger(name) + + # Only configure if handlers aren't already set up + if not logger.handlers: + logger.setLevel(level) + + # Create console handler with formatter + handler = logging.StreamHandler(sys.stdout) + handler.setFormatter( + logging.Formatter("%(asctime)s [%(levelname)s] %(name)s - %(message)s") + ) + + logger.addHandler(handler) + logger.propagate = False + + return logger + + +def configure_logging( + app_name: str, + log_level: str = "INFO", + json_format: bool = False, log_file: Optional[str] = None, - use_json: bool = False, + console_output: bool = True, + aws_context: Optional[Dict[str, str]] = None, ) -> logging.Logger: """ - Set up logging configuration. + Configure logging with advanced features. Args: - log_level: Log level (DEBUG, INFO, WARNING, ERROR, CRITICAL) + app_name: Application name + log_level: Log level name + json_format: Output logs as JSON log_file: Path to log file - use_json: Whether to use JSON formatting + console_output: Whether to log to console + aws_context: AWS context information Returns: - Logger instance + Configured logger instance """ - # Get root logger - logger = logging.getLogger("aws_resource_management") - - # Clear existing handlers to avoid duplicate logs - if logger.handlers: - logger.handlers = [] - - # Set log level - if not log_level: - log_level = os.environ.get("AWS_RESOURCE_MGMT_LOG_LEVEL", "INFO") + # Convert log level string to logging level + level = getattr(logging, log_level.upper(), logging.INFO) - logger.setLevel(getattr(logging, log_level)) + # Create logger + logger = logging.getLogger(app_name) + logger.setLevel(level) + logger.handlers = [] # Remove any existing handlers - # Create console handler - console_handler = logging.StreamHandler(sys.stdout) + # Set global AWS context if provided + if aws_context: + LoggingContext.set(**aws_context) - # Set formatter based on format preference - if use_json: + # Create formatter + if json_format: formatter = JSONFormatter() else: formatter = logging.Formatter( "%(asctime)s [%(levelname)s] %(name)s - %(message)s" ) - console_handler.setFormatter(formatter) - logger.addHandler(console_handler) + # Add console handler + if console_output: + console_handler = logging.StreamHandler(sys.stdout) + console_handler.setFormatter(formatter) + logger.addHandler(console_handler) - # Add file handler if log file is specified + # Add file handler if log_file: file_handler = logging.FileHandler(log_file) file_handler.setFormatter(formatter) logger.addHandler(file_handler) + # Prevent propagation to avoid duplicate logs + logger.propagate = False + return logger -class LoggerAdapter(logging.LoggerAdapter): - """ - Logger adapter for adding context to log messages. +def log_with_context(logger: logging.Logger, level: int, msg: str, **context) -> None: """ + Log with additional context data. - def __init__(self, logger: logging.Logger, context: Dict[str, Any]): - """ - Initialize logger adapter with context. - - Args: - logger: Logger instance - context: Context dictionary - """ - super().__init__(logger, context) - - def process(self, msg: str, kwargs: Dict[str, Any]) -> tuple: - """ - Process the log message by adding context. - - Args: - msg: Log message - kwargs: Keyword arguments - - Returns: - Tuple of (modified message, modified kwargs) - """ - # Add extra_fields for JSON formatter - if "extra" not in kwargs: - kwargs["extra"] = {} - - if "extra_fields" not in kwargs["extra"]: - kwargs["extra"]["extra_fields"] = {} + Args: + logger: Logger to use + level: Log level + msg: Log message + **context: Additional context data + """ + # Add logging context + combined_context = LoggingContext.get() + combined_context.update(context) + + # Create a record with extra context + extra = {"context": combined_context} + + # Log with context + logger.log(level, msg, extra=extra) + + +@contextmanager +def log_operation( + logger: logging.Logger, + operation: str, + level: int = logging.INFO, + success_level: Optional[int] = None, + failure_level: int = logging.ERROR, + **context, +): + """ + Context manager to log the start, end, and any exceptions for an operation. - for key, value in self.extra.items(): - kwargs["extra"]["extra_fields"][key] = value + Args: + logger: Logger to use + operation: Operation name + level: Log level for start/success messages + success_level: Optional different level for success message + failure_level: Log level for failure messages + **context: Additional context data + """ + success_level = success_level or level + start_time = time.time() + + # Set operation context + operation_context = { + "operation": operation, + "operation_status": "started", + } + operation_context.update(context) + + # Log start with context + log_with_context(logger, level, f"Started {operation}", **operation_context) + + try: + # Execute the operation + yield + + # Log success with duration + duration = time.time() - start_time + operation_context.update( + { + "operation_status": "completed", + "duration_seconds": round(duration, 3), + } + ) + log_with_context( + logger, + success_level, + f"Completed {operation} in {duration:.3f}s", + **operation_context, + ) - return msg, kwargs + except Exception as e: + # Log failure with exception and duration + duration = time.time() - start_time + operation_context.update( + { + "operation_status": "failed", + "duration_seconds": round(duration, 3), + "error": str(e), + "error_type": e.__class__.__name__, + } + ) + log_with_context( + logger, failure_level, f"Failed {operation}: {e}", **operation_context + ) + raise def initialize_csv_log(csv_file: Optional[str] = None) -> str: diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/__init__.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/__init__.py index 0583be56..da446c04 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/__init__.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/__init__.py @@ -1,10 +1,40 @@ """ -Resource manager modules for different AWS services. +Resource manager implementations. """ -from aws_resource_management.managers.base import ResourceManager +# Import classes for easier access from the managers package +from aws_resource_management.managers.base import BaseResourceManager, ResourceManager from aws_resource_management.managers.ec2 import EC2Manager -from aws_resource_management.managers.ecr import ECRManager -from aws_resource_management.managers.eks import EKSManager -from aws_resource_management.managers.emr import EMRManager from aws_resource_management.managers.rds import RDSManager + +# Try to import optional managers +try: + from aws_resource_management.managers.eks import EKSManager +except ImportError: + EKSManager = None + +try: + from aws_resource_management.managers.emr import EMRManager +except ImportError: + EMRManager = None + +try: + from aws_resource_management.managers.ecr import ECRManager +except ImportError: + ECRManager = None + +# Export manager types for public use +__all__ = [ + "ResourceManager", + "BaseResourceManager", + "EC2Manager", + "RDSManager", +] + +# Add optional managers if available +if EKSManager: + __all__.append("EKSManager") +if EMRManager: + __all__.append("EMRManager") +if ECRManager: + __all__.append("ECRManager") 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 9e626b2e..08cbffee 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 @@ -3,278 +3,146 @@ """ import logging -import sys -import time -from datetime import datetime -from typing import Any, Dict, List, Optional, Union +from typing import Any, Dict, List, Optional, Type import boto3 -from aws_resource_management.config_manager import get_config -from botocore.exceptions import ClientError +from aws_resource_management.aws_utils import get_session_for_account +from botocore.config import Config +from botocore.exceptions import ClientError, NoRegionError -logger = logging.getLogger(__name__) -config = get_config() +logger = logging.getLogger("aws_resource_management.managers.base") -# Define authentication failure error codes -AUTH_FAILURE_ERRORS = [ - "AuthFailure", - "InvalidClientTokenId", - "UnauthorizedOperation", - "AccessDenied", -] +class BaseResourceManager: + """Base class for all resource managers.""" -class ResourceManager: - """Base class for AWS resource managers.""" - - def __init__( - self, - credentials: Dict[str, str], - account_id: str, - account_name: Optional[str] = None, - ): - """Initialize the resource manager.""" - self.credentials = credentials + def __init__(self, account_id: str, region: str): self.account_id = account_id - self.account_name = account_name - self.clients = {} # Cache for boto3 clients + self.region = region + self.session = None + self._clients = {} # Cache for boto3 clients + + def get_client(self, service_name: str) -> Any: + """ + Get a boto3 client for the specified service with proper partition support. - def get_boto3_client(self, service_name: str, region: str) -> Any: - """Get a boto3 client for the specified service and region.""" - client_key = f"{service_name}-{region}" - if client_key in self.clients: - return self.clients[client_key] + Args: + service_name: AWS service name (e.g., 'ec2', 'rds') + + Returns: + Boto3 client for the specified service + """ + # Return cached client if available + cache_key = f"{service_name}_{self.region}" + if cache_key in self._clients: + return self._clients[cache_key] try: - client = boto3.client( - service_name, - region_name=region, - aws_access_key_id=self.credentials.get("aws_access_key_id"), - aws_secret_access_key=self.credentials.get("aws_secret_access_key"), - aws_session_token=self.credentials.get("aws_session_token"), + # Get an account-specific session with proper role assumption + if not self.session: + self.session = get_session_for_account(self.account_id) + + # Create config with retry settings and proper endpoint resolution + boto_config = Config( + region_name=self.region, + retries={"max_attempts": 3, "mode": "standard"}, + # Enable partition-specific endpoint resolution + use_dualstack_endpoint=False, + use_fips_endpoint=False, ) - # Immediately validate credentials by making a simple API call - self._validate_credentials(client, service_name, region) + # Create client with proper configuration + client = self.session.client( + service_name, region_name=self.region, config=boto_config + ) - # Cache client if validation passes - self.clients[client_key] = client + # Cache the client + self._clients[cache_key] = client return client + + except NoRegionError: + logger.error( + f"No region specified for {service_name} client and no default region found" + ) + raise except ClientError as e: - # Check if it's an authentication error - error_code = e.response.get("Error", {}).get("Code", "") - error_message = e.response.get("Error", {}).get("Message", "") + error_code = e.response.get("Error", {}).get("Code", "Unknown") + logger.error( + f"Error creating {service_name} client in {self.region}: {error_code} - {str(e)}" + ) + raise + except Exception as e: + logger.error( + f"Error creating {service_name} client in {self.region}: {str(e)}" + ) + raise - if error_code in AUTH_FAILURE_ERRORS: - logger.error( - f"Authentication failed for {service_name} in {region}: {error_message}" - ) + def _handle_api_error(self, operation: str, error: Exception) -> None: + """ + Handle AWS API errors with consistent logging. + + Args: + operation: Operation name that failed + error: Exception that occurred + """ + if isinstance(error, ClientError): + error_code = error.response.get("Error", {}).get("Code", "Unknown") + error_message = error.response.get("Error", {}).get("Message", str(error)) + + if error_code == "AuthFailure": logger.error( - f"This account likely doesn't have valid credentials. Skipping account {self.account_id}." + f"Authentication failure during {operation} in account {self.account_id} " + f"region {self.region}. Check IAM permissions." ) - # Return None to signal authentication failure - return None - else: + elif error_code == "AccessDenied": logger.error( - f"Error creating {service_name} client in {region}: {error_code} - {error_message}" + f"Access denied during {operation} in account {self.account_id} " + f"region {self.region}. Check IAM permissions." ) - return None - except Exception as e: - logger.error(f"Error creating {service_name} client in {region}: {e}") - return None - - def _validate_credentials( - self, client: Any, service_name: str, region: str - ) -> None: - """ - Validate credentials by making a minimal API call. - Raises ClientError if authentication fails. - """ - # Make service-specific minimal API call to verify credentials - try: - if service_name == "ec2": - client.describe_regions(MaxResults=5) - elif service_name == "rds": - client.describe_db_engine_versions(MaxRecords=1) - elif service_name == "eks": - client.list_clusters(maxResults=1) - elif service_name == "emr": - client.list_clusters(MaxResults=1) - elif service_name == "ecr": - client.describe_repositories(maxResults=1) else: - # For other services, don't validate (better than failing) - pass - except ClientError as e: - # Check if it's an authentication error - error_code = e.response.get("Error", {}).get("Code", "") - if error_code in AUTH_FAILURE_ERRORS: logger.error( - f"Authentication validation failed for {service_name} in {region}" + f"API error during {operation} in account {self.account_id} " + f"region {self.region}: {error_code} - {error_message}" ) - raise # Re-raise to be caught by get_boto3_client - # Otherwise ignore other errors during validation - except Exception: - # Ignore any other exceptions during validation - pass - - # Alias for backwards compatibility - create_client = get_boto3_client + else: + logger.error( + f"Error during {operation} in account {self.account_id} " + f"region {self.region}: {str(error)}" + ) - def should_exclude(self, resource: Dict[str, Any]) -> bool: + def start(self, resource_ids: List[str]) -> Dict[str, str]: """ - Check if a resource should be excluded based on tags. + Start resources. Must be implemented by derived classes. Args: - resource: Resource dictionary with tags + resource_ids: List of resource IDs to start Returns: - True if the resource should be excluded, False otherwise + Dictionary mapping resource IDs to status """ - tags = resource.get("tags", {}) - exclusion_tag = config.get("exclusion_tag") - - return exclusion_tag in tags + raise NotImplementedError("Subclasses must implement start method") - def get_timestamp(self) -> str: - """ - Get current timestamp in ISO 8601 format. - - Returns: - ISO 8601 timestamp string + def stop(self, resource_ids: List[str]) -> Dict[str, str]: """ - return datetime.utcnow().isoformat() - - def get_partition_for_region(self, region: str) -> str: - """ - Get AWS partition for a region. + Stop resources. Must be implemented by derived classes. Args: - region: AWS region + resource_ids: List of resource IDs to stop Returns: - AWS partition string (aws, aws-us-gov, aws-cn) + Dictionary mapping resource IDs to status """ - if region.startswith("us-gov-"): - return "aws-us-gov" - elif region.startswith("cn-"): - return "aws-cn" - else: - return "aws" + raise NotImplementedError("Subclasses must implement stop method") - def log_action( - self, - resource_id: str, - region: str, - action: str, - resource_name: Optional[str] = None, - details: Optional[str] = None, - dry_run: bool = False, - existing_schedule: Optional[str] = None, - ) -> None: + def discover_resources(self) -> List[Dict[str, Any]]: """ - Log an action taken on a resource. + Discover resources of this type. Must be implemented by derived classes. - Args: - resource_id: Resource ID - region: AWS region - action: Action name (start, stop, etc.) - resource_name: Resource name (optional) - details: Action details (optional) - dry_run: Whether this was a dry run - existing_schedule: Existing schedule tag value (optional) + Returns: + List of dictionaries containing resource information """ - name_str = f" ({resource_name})" if resource_name else "" - action_str = f"{action.upper()}" - details_str = f": {details}" if details else "" - schedule_str = f" [Schedule: {existing_schedule}]" if existing_schedule else "" - dry_run_str = "[DRY RUN] " if dry_run else "" - - logger.info( - f"{dry_run_str}{action_str} {self.resource_type} {resource_id}{name_str} " - f"in {region}{details_str}{schedule_str}" - ) - - def paginate_boto3(self, client, method_name, result_key, **kwargs): - """Helper method to handle pagination for boto3 API calls.""" - # If client is None (authentication failed earlier), return empty list - if client is None: - logger.warning( - f"Skipping {method_name} due to previous authentication failure" - ) - return [] - - method = getattr(client, method_name) - items = [] - - try: - # Try to use paginator if available - try: - paginator = client.get_paginator(method_name) - for page in paginator.paginate(**kwargs): - if result_key in page: - items.extend(page[result_key]) - return items - except ClientError as e: - error_code = e.response.get("Error", {}).get("Code", "") - if error_code in AUTH_FAILURE_ERRORS: - logger.error( - f"Authentication failure in paginator for {method_name}: {error_code}" - ) - return [] # Return empty result on auth failure - # Fall back to manual pagination - pass - except AttributeError: - # No paginator available, fall back to manual pagination - pass - - # Manual pagination - try: - response = method(**kwargs) - if result_key in response: - items.extend(response[result_key]) - - pagination_keys = [ - "NextToken", - "nextToken", - "Marker", - "marker", - "next_token", - "next_marker", - ] - pagination_key = next( - (k for k in pagination_keys if k in response), None - ) - - while pagination_key and response.get(pagination_key): - kwargs[pagination_key] = response[pagination_key] - response = method(**kwargs) - if result_key in response: - items.extend(response[result_key]) - - except ClientError as e: - error_code = e.response.get("Error", {}).get("Code", "") - if error_code in AUTH_FAILURE_ERRORS: - logger.error( - f"Authentication failure in manual pagination for {method_name}: {error_code}" - ) - return [] # Return empty result on auth failure - logger.error(f"Error during manual pagination for {method_name}: {e}") - - except Exception as e: - logger.error(f"Error paginating through {method_name}: {str(e)}") - - return items + raise NotImplementedError("Subclasses must implement discover_resources method") - # Abstract methods that should be implemented by subclasses - def start( - self, resources: List[Dict[str, Any]], dry_run: bool = False - ) -> Dict[str, int]: - """Start resources - abstract method to be implemented by subclasses.""" - raise NotImplementedError("Subclasses must implement start()") - def stop( - self, resources: List[Dict[str, Any]], dry_run: bool = False - ) -> Dict[str, int]: - """Stop resources - abstract method to be implemented by subclasses.""" - raise NotImplementedError("Subclasses must implement stop()") +# Create alias for backward compatibility +ResourceManager = BaseResourceManager diff --git a/local-app/python-tools/gfl-resource-actions/logging_utils.py b/local-app/python-tools/gfl-resource-actions/logging_utils.py deleted file mode 100644 index a6c16659..00000000 --- a/local-app/python-tools/gfl-resource-actions/logging_utils.py +++ /dev/null @@ -1,312 +0,0 @@ -""" -Logging utilities for resource management. -""" - -import csv -import datetime -import logging -import os -import sys -from pathlib import Path - -from config import ACTION_CSV_FILE - -# Configure log formats -DEFAULT_LOG_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" -DEFAULT_LOG_LEVEL = logging.INFO - -# Configure CSV logging -CSV_LOG_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "logs") -CSV_FIELDS = [ - "timestamp", - "account_id", - "account_name", - "resource_type", - "resource_id", - "resource_name", - "action", - "region", - "status", - "details", -] - - -def setup_logging(name="resource_management", level=logging.INFO): - """Set up and configure logging.""" - logger = logging.getLogger(name) - - # Only configure the logger if it hasn't been configured already - if not logger.handlers: - logger.setLevel(level) - - # Create console handler - console_handler = logging.StreamHandler() - console_handler.setLevel(level) - - # Create formatter - formatter = logging.Formatter( - "%(asctime)s - %(name)s - %(levelname)s - %(message)s" - ) - console_handler.setFormatter(formatter) - - # Add handler to logger - logger.addHandler(console_handler) - - # Prevent propagation to the root logger to avoid duplicate messages - logger.propagate = False - - return logger - - -def initialize_csv_log(csv_file=None): - """ - Initialize CSV log file with headers if it doesn't exist. - - Args: - csv_file (str, optional): CSV file name (default: resource_actions.csv) - - Returns: - str: Path to the CSV log file - """ - if csv_file is None: - csv_file = ACTION_CSV_FILE - - 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(CSV_FIELDS) - - return csv_path - - -def log_action_to_csv( - account_id, - account_name, - resource_type, - resource_id, - resource_name, - action, - region, - status, - details="", - dry_run=False, - existing_schedule=None, -): - """Log resource action to CSV file for tracking.""" - timestamp = datetime.datetime.now().isoformat() - - # Create directory if it doesn't exist - os.makedirs(os.path.dirname(CSV_LOG_DIR), exist_ok=True) - - # Check if file exists to determine if we need to write headers - file_exists = os.path.isfile(CSV_LOG_DIR) - - with open(CSV_LOG_DIR, "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 header if file is new - if not file_exists: - writer.writeheader() - - # 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 "", - } - ) - - -# Create a logger instance for direct import -logger = setup_logging() - -import logging -from typing import Any, Dict, Optional - -from logging_setup import LoggingContext, log_operation, log_with_context - -logger = logging.getLogger("aws_resource_management") - - -def log_resource_operation( - level: int, - operation: str, - resource_type: str, - resource_id: str, - region: str, - details: Optional[Dict[str, Any]] = None, - exception: Optional[Exception] = None, -) -> None: - """ - Log resource operations with consistent structure. - - Args: - level: Log level - operation: Operation being performed (e.g., 'start', 'stop') - resource_type: AWS resource type (e.g., 'ec2_instance', 'rds_instance') - resource_id: Resource identifier - region: AWS region - details: Optional additional details - exception: Optional exception if the operation failed - """ - msg = f"{operation} {resource_type} {resource_id} in {region}" - context = { - "resource_type": resource_type, - "resource_id": resource_id, - "operation": operation, - "region": region, - } - - if details: - context["details"] = details - msg += f": {details}" - - if exception: - context["exception"] = str(exception) - context["exception_type"] = exception.__class__.__name__ - msg += f" (failed: {exception})" - - log_with_context(logger, level, msg, **context) - - -def initialize_aws_logging( - app_name: str, - log_level: str = "INFO", - account_id: Optional[str] = None, - region: Optional[str] = None, - log_to_file: bool = False, - log_file_path: Optional[str] = None, -) -> logging.Logger: - """ - Initialize AWS-aware logging for the application. - - Args: - app_name: Application name - log_level: Log level name - account_id: AWS account ID - region: AWS default region - log_to_file: Whether to log to a file - log_file_path: Path to log file (if log_to_file is True) - - Returns: - Configured logger instance - """ - from logging_setup import configure_logging - - # Set up default log file path if needed - if log_to_file and not log_file_path: - import os - - log_dir = os.path.join( - os.path.expanduser("~"), ".aws-resource-management", "logs" - ) - os.makedirs(log_dir, exist_ok=True) - log_file_path = os.path.join(log_dir, f"{app_name}.log") - - # Configure AWS context - aws_context = {} - if account_id: - aws_context["account_id"] = account_id - if region: - aws_context["region"] = region - - # Initialize logging - return configure_logging( - app_name=app_name, - log_level=log_level, - json_format=True, - log_file=log_file_path if log_to_file else None, - console_output=True, - aws_context=aws_context, - ) - - -# Example usage function to demonstrate the structured logging features -def example_usage(): - # Initialize logging - logger = initialize_aws_logging( - app_name="resource-manager", - log_level="INFO", - account_id="123456789012", - region="us-west-2", - log_to_file=True, - ) - - # Simple logging with context - log_with_context( - logger, - logging.INFO, - "Starting application", - app_version="1.0.0", - environment="development", - ) - - # Log resource operation - log_resource_operation( - logging.INFO, - "start", - "ec2_instance", - "i-1234567890abcdef0", - "us-west-2", - details={"instance_type": "t2.micro", "target_state": "running"}, - ) - - # Use operation context manager - try: - with log_operation( - logger, - "resize_rds_instance", - logging.INFO, - resource_type="rds_instance", - resource_id="mydb-instance-1", - region="us-west-2", - ): - # Simulated operation that takes time - import time - - time.sleep(1.5) - - # Set additional context during operation - LoggingContext.set(instance_class="db.t3.large") - - # Log with the updated context - log_with_context(logger, logging.INFO, "Changed instance class") - - # Simulate successful completion - pass - except Exception as e: - # The log_operation context manager will log the exception - # and re-raise it - pass