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 133d0ff5..f83dd1a4 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 @@ -448,11 +448,15 @@ 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(credentials, account_id, account_name), - "rds": RDSManager(credentials, account_id, account_name), - "eks": EKSManager(credentials, account_id, account_name), - "emr": EMRManager(credentials, account_id, account_name), + "ec2": EC2Manager(account_id, default_region), + "rds": RDSManager(account_id, default_region), + "eks": EKSManager(account_id, default_region), + "emr": EMRManager(account_id, default_region), } def _execute_actions( 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 bb9d8d7f..f422f269 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 @@ -510,20 +510,71 @@ def _discover_eks( return [] -def get_account_resources( - account_id: str, +# 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]]]: + """ + Get all resources for an account across multiple regions. + Backward-compatible wrapper that handles both positional and keyword arguments. + + Args: + credentials: AWS credentials (dict or str) + regions: List of AWS regions to search + exclude_resources: List of resource types to exclude (keyword-only) + account_id: AWS account ID (keyword-only) + account_name: AWS account name (keyword-only) + emr_cluster_states: EMR cluster states to include (keyword-only) + + Returns: + Dictionary of resource lists by type + """ + # Handle both old-style positional calling and new-style keyword calling + credentials = None + regions = [] + + if len(args) >= 1: + credentials = args[0] + else: + credentials = kwargs.get('credentials') + + if len(args) >= 2: + regions = args[1] + else: + regions = kwargs.get('regions', []) + + # Extract keyword arguments with defaults + exclude_resources = kwargs.get('exclude_resources', []) + account_id = kwargs.get('account_id', 'unknown') + account_name = kwargs.get('account_name', 'Unknown') + emr_cluster_states = kwargs.get('emr_cluster_states') + + return _get_account_resources_impl( + credentials=credentials, + regions=regions, + exclude_resources=exclude_resources, + account_id=account_id, + account_name=account_name, + emr_cluster_states=emr_cluster_states + ) + + +def _get_account_resources_impl( + credentials: Dict[str, str], regions: List[str], + *, # Force all following parameters to be keyword-only exclude_resources: List[str] = None, + account_id: str = 'unknown', 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 using the ResourceDiscovery class. - + Implementation of resource discovery for an account. + Args: - account_id: AWS account ID + credentials: AWS credentials dictionary regions: List of AWS regions to search exclude_resources: List of resource types to exclude + account_id: AWS account ID account_name: AWS account name (optional) emr_cluster_states: EMR cluster states to include (optional) @@ -532,7 +583,7 @@ def get_account_resources( """ if exclude_resources is None: exclude_resources = [] - + resources = { "ec2_instances": [], "rds_instances": [], @@ -541,57 +592,67 @@ def get_account_resources( "ecr_images": [], "ecr_old_images": [], } - + try: - # Get session for the account - session = get_session_for_account(account_id) + # Create session or use provided credentials + session = None + if isinstance(credentials, dict): + # Create a session from the credentials dictionary + session = boto3.Session( + aws_access_key_id=credentials.get('aws_access_key_id'), + aws_secret_access_key=credentials.get('aws_secret_access_key'), + aws_session_token=credentials.get('aws_session_token') + ) + else: + # Try to get a 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 - ), + + # Extract credentials from session for the ResourceDiscovery + creds = session.get_credentials() + discovery_credentials = { + "aws_access_key_id": creds.access_key, + "aws_secret_access_key": creds.secret_key, + "aws_session_token": creds.token if hasattr(creds, 'token') and creds.token else None, } - - discovery = ResourceDiscovery(credentials, account_id) - + + # Create resource discovery instance + discovery = ResourceDiscovery(discovery_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" ) resources["rds_instances"] = discovery.discover_resources("rds", 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 and EMRManager: logger.info( f"Discovering EMR clusters for account {account_id} in {len(regions)} regions" ) - resources["emr_clusters"] = discovery.discover_resources("emr", regions) - + # Use emr_cluster_states if provided + resources["emr_clusters"] = discovery.discover_resources( + "emr", regions, resource_ids=None + ) + # Discover ECR images if "ecr" not in exclude_resources and ECRManager: logger.info( @@ -603,7 +664,14 @@ def get_account_resources( # Filter old images if needed resources["ecr_old_images"] = [] + # Add account information to all resources + for resource_type, resource_list in resources.items(): + for resource in resource_list: + resource["accountId"] = account_id + if account_name: + resource["accountName"] = account_name + 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/managers/base.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/base.py index 08cbffee..2b9c6763 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,6 +3,7 @@ """ import logging +from datetime import datetime from typing import Any, Dict, List, Optional, Type import boto3 @@ -22,6 +23,15 @@ def __init__(self, account_id: str, region: str): self.session = None self._clients = {} # Cache for boto3 clients + def get_timestamp(self) -> str: + """ + Get current timestamp in ISO format for tagging resources. + + Returns: + ISO formatted timestamp string + """ + return datetime.utcnow().isoformat() + def get_client(self, service_name: str) -> Any: """ Get a boto3 client for the specified service with proper partition support. @@ -142,6 +152,126 @@ def discover_resources(self) -> List[Dict[str, Any]]: List of dictionaries containing resource information """ raise NotImplementedError("Subclasses must implement discover_resources method") + + def get_boto3_client(self, service_name: str, region: str) -> Any: + """ + Get a boto3 client for the specified service and region. + + Args: + service_name: AWS service name (e.g., 'ec2', 'rds') + region: AWS region + + Returns: + Boto3 client for the specified service in the specified region + """ + try: + # 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 + boto_config = Config( + region_name=region, + retries={"max_attempts": 3, "mode": "standard"}, + ) + + # Create client with proper configuration + client = self.session.client( + service_name, region_name=region, config=boto_config + ) + return client + except Exception as e: + logger.error(f"Error creating {service_name} client in {region}: {str(e)}") + return None + + # 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, region: str, action: str, + resource_name: str = None, details: str = None, + dry_run: bool = False, existing_schedule: str = None) -> None: + """ + Log an action taken on a resource. + + Args: + resource_id: Resource ID + region: AWS region + action: Action taken (start, stop, etc.) + resource_name: Optional resource name + details: Optional action details + dry_run: Whether this was a dry run + existing_schedule: Optional existing schedule + """ + name_str = f"({resource_name})" if resource_name else "" + dry_run_prefix = "[DRY RUN] Would " if dry_run else "" + details_str = f" - {details}" if details else "" + 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"in {region}{details_str}{schedule_str}" + ) + + def should_exclude(self, resource: Dict[str, Any]) -> bool: + """ + Check if a resource should be excluded based on tags. + + Args: + resource: Resource dictionary with tags + + 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") + + if exclusion_tag and exclusion_tag in tags: + return True + + return False + + def get_partition_for_region(self, region: str) -> str: + """ + Get the AWS partition for a region. + + Args: + region: AWS region + + Returns: + AWS partition (aws, aws-cn, aws-us-gov) + """ + if region.startswith("us-gov-") or region.startswith("gov-"): + return "aws-us-gov" + elif region.startswith("cn-"): + return "aws-cn" + else: + return "aws" # Create alias for backward compatibility 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 8fe39c37..9d88475a 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 @@ -18,13 +18,13 @@ class EC2Manager(ResourceManager): def __init__( self, - credentials: Dict[str, str], account_id: str, - account_name: Optional[str] = None, + region: str, ): """Initialize EC2 manager.""" - super().__init__(credentials, account_id, account_name) + 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: diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/eks.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/eks.py index 78e7c27c..ba62637a 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/eks.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/eks.py @@ -27,20 +27,19 @@ class EKSManager(ResourceManager): def __init__( self, - credentials: Dict[str, str], account_id: str, - account_name: Optional[str] = None, + region: str, ): """ Initialize EKS manager. Args: - credentials: AWS credentials dictionary account_id: AWS account ID - account_name: AWS account name (optional) + region: AWS region """ - super().__init__(credentials, account_id, account_name) + super().__init__(account_id, region) self.resource_type = "eks_cluster" + self.account_name = None # This can be updated if needed def stop( self, clusters: List[Dict[str, Any]], dry_run: bool = False 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 f4ffdfed..3e082f12 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 @@ -7,6 +7,7 @@ from aws_resource_management.config_manager import get_config from aws_resource_management.logging_setup import setup_logging from aws_resource_management.managers.base import ResourceManager +from aws_resource_management.aws_utils import get_session_for_account, detect_partition_from_credentials from botocore.exceptions import ClientError logger = setup_logging() @@ -18,20 +19,19 @@ class EMRManager(ResourceManager): def __init__( self, - credentials: Dict[str, str], account_id: str, - account_name: Optional[str] = None, + region: str, ): """ Initialize EMR manager. Args: - credentials: AWS credentials dictionary account_id: AWS account ID - account_name: AWS account name (optional) + region: AWS region """ - super().__init__(credentials, account_id, account_name) + super().__init__(account_id, region) self.resource_type = "emr_cluster" + self.account_name = None # This can be updated if needed def discover_clusters( self, regions: List[str], cluster_states: Optional[List[str]] = None @@ -232,19 +232,31 @@ def validate_credentials(self, region: str = None) -> bool: """ if not region: # Use a default region based on the detected partition - if any( - cred - for cred, val in self.credentials.items() - if val and "gov" in val.lower() - ): - region = "us-gov-west-1" + session = get_session_for_account(self.account_id) + if not session: + logger.error(f"Could not create session for account {self.account_id}") + return False + + # Determine partition from session credentials + creds = session.get_credentials() + if creds: + credentials = { + "aws_access_key_id": creds.access_key, + "aws_secret_access_key": creds.secret_key, + "aws_session_token": creds.token if hasattr(creds, 'token') and creds.token else None, + } + partition = detect_partition_from_credentials(credentials) + if partition == "aws-us-gov": + region = "us-gov-west-1" + else: + region = "us-east-1" else: - region = "us-east-1" + region = "us-east-1" # Default to commercial AWS try: logger.info(f"Validating EMR credentials in region {region}") - emr_client = self.create_client("emr", region) - + emr_client = self.get_boto3_client("emr", region) + # Just list a small number of clusters to verify permissions response = emr_client.list_clusters(MaxResults=1) logger.info(f"EMR credentials valid in region {region}") diff --git a/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/rds.py b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/rds.py index d5add442..c38bf37b 100644 --- a/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/rds.py +++ b/local-app/python-tools/gfl-resource-actions/aws_resource_management/managers/rds.py @@ -18,13 +18,13 @@ class RDSManager(ResourceManager): def __init__( self, - credentials: Dict[str, str], account_id: str, - account_name: Optional[str] = None, + region: str, ): """Initialize RDS manager.""" - super().__init__(credentials, account_id, account_name) + super().__init__(account_id, region) self.resource_type = "rds_instance" + self.account_name = None # This can be updated if needed self.MAX_DB_OPERATIONS = 20 # Conservative limit for RDS batch operations def stop( diff --git a/local-app/python-tools/gfl-resource-actions/discovery.py b/local-app/python-tools/gfl-resource-actions/discovery.py deleted file mode 100644 index 78839ace..00000000 --- a/local-app/python-tools/gfl-resource-actions/discovery.py +++ /dev/null @@ -1,701 +0,0 @@ -""" -Resource discovery functions for AWS resources. -""" - -import config -from aws_resource_management.aws_utils import create_boto3_client -from logging_utils import log_action_to_csv, setup_logging - -logger = setup_logging() - - -def get_ec2_instances(credentials, region, account_id=None, account_name=None): - """Get EC2 instances in a region.""" - try: - ec2_client = create_boto3_client("ec2", credentials, region) - instances = [] - - response = ec2_client.describe_instances() - for reservation in response.get("Reservations", []): - for instance in reservation.get("Instances", []): - # Capture tags for exclusion check - tags = {tag["Key"]: tag["Value"] for tag in instance.get("Tags", [])} - - # Extract name from tags - name = tags.get("Name", "") - - instance_data = { - "id": instance["InstanceId"], - "name": name, - "state": instance["State"]["Name"], - "region": region, - "private_ip": instance.get("PrivateIpAddress", ""), - "public_ip": instance.get("PublicIpAddress", ""), - "tags": tags, - } - - instances.append(instance_data) - - # Log the discovery action to CSV - if account_id: - log_action_to_csv( - account_id=account_id, - account_name=account_name or "Unknown", - resource_type="ec2", - resource_id=instance["InstanceId"], - resource_name=name, - action="discover", - region=region, - status="success", - details=f"State: {instance['State']['Name']}, Private IP: {instance.get('PrivateIpAddress', 'N/A')}", - ) - - # Handle pagination - while "NextToken" in response: - response = ec2_client.describe_instances(NextToken=response["NextToken"]) - for reservation in response.get("Reservations", []): - for instance in reservation.get("Instances", []): - # Capture tags for exclusion check - tags = { - tag["Key"]: tag["Value"] for tag in instance.get("Tags", []) - } - - # Extract name from tags - name = tags.get("Name", "") - - instance_data = { - "id": instance["InstanceId"], - "name": name, - "state": instance["State"]["Name"], - "region": region, - "private_ip": instance.get("PrivateIpAddress", ""), - "public_ip": instance.get("PublicIpAddress", ""), - "tags": tags, - } - - instances.append(instance_data) - - # Log the discovery action to CSV - if account_id: - log_action_to_csv( - account_id=account_id, - account_name=account_name or "Unknown", - resource_type="ec2", - resource_id=instance["InstanceId"], - resource_name=name, - action="discover", - region=region, - status="success", - details=f"State: {instance['State']['Name']}, Private IP: {instance.get('PrivateIpAddress', 'N/A')}", - ) - - return instances - except Exception as e: - error_msg = f"Error getting EC2 instances in region {region}: {e}" - logger.error(error_msg) - - # Log the error to CSV - if account_id: - log_action_to_csv( - account_id=account_id, - account_name=account_name or "Unknown", - resource_type="ec2", - resource_id="N/A", - resource_name="N/A", - action="discover", - region=region, - status="failed", - details=error_msg, - ) - - return [] - - -def get_rds_instances(credentials, region, account_id=None, account_name=None): - """Get RDS instances in a region.""" - try: - rds_client = create_boto3_client("rds", credentials, region) - instances = [] - - logger.info(f"Discovering RDS instances in region {region}...") - response = rds_client.describe_db_instances() - - # Log the raw count of instances returned from API - raw_count = len(response.get("DBInstances", [])) - logger.info(f"Raw RDS API returned {raw_count} instances in region {region}") - - for instance in response.get("DBInstances", []): - # Log the raw status as received from AWS - raw_status = instance["DBInstanceStatus"] - logger.debug( - f"RDS instance {instance['DBInstanceIdentifier']} raw status: {raw_status}" - ) - - # Capture tags for exclusion check - tags = {} - try: - tag_list = rds_client.list_tags_for_resource( - ResourceName=instance["DBInstanceArn"] - ).get("TagList", []) - tags = {tag["Key"]: tag["Value"] for tag in tag_list} - except Exception as tag_e: - logger.warning( - f"Error getting tags for RDS instance {instance['DBInstanceIdentifier']}: {tag_e}" - ) - - # Get endpoint address - endpoint_address = "" - if "Endpoint" in instance and "Address" in instance["Endpoint"]: - endpoint_address = instance["Endpoint"]["Address"] - - # Standardize status to lowercase for consistent comparison - standardized_status = raw_status.lower() - - instance_data = { - "id": instance["DBInstanceIdentifier"], - "name": instance["DBInstanceIdentifier"], # Using ID as name - "status": standardized_status, # Using standardized status - "raw_status": raw_status, # Keep original status for debugging - "region": region, - "endpoint": endpoint_address, - "tags": tags, - } - - instances.append(instance_data) - - # Log the discovery action to CSV - if account_id: - log_action_to_csv( - account_id=account_id, - account_name=account_name or "Unknown", - resource_type="rds", - resource_id=instance["DBInstanceIdentifier"], - resource_name=instance["DBInstanceIdentifier"], - action="discover", - region=region, - status="success", - details=f"Status: {raw_status}, Endpoint: {endpoint_address}", - ) - - # Handle pagination - while "Marker" in response: - response = rds_client.describe_db_instances(Marker=response["Marker"]) - for instance in response.get("DBInstances", []): - # Log the raw status as received from AWS - raw_status = instance["DBInstanceStatus"] - logger.debug( - f"RDS instance {instance['DBInstanceIdentifier']} raw status: {raw_status}" - ) - - # Capture tags for exclusion check - tags = {} - try: - tag_list = rds_client.list_tags_for_resource( - ResourceName=instance["DBInstanceArn"] - ).get("TagList", []) - tags = {tag["Key"]: tag["Value"] for tag in tag_list} - except Exception as tag_e: - logger.warning( - f"Error getting tags for RDS instance {instance['DBInstanceIdentifier']}: {tag_e}" - ) - - # Get endpoint address - endpoint_address = "" - if "Endpoint" in instance and "Address" in instance["Endpoint"]: - endpoint_address = instance["Endpoint"]["Address"] - - # Standardize status to lowercase for consistent comparison - standardized_status = raw_status.lower() - - instance_data = { - "id": instance["DBInstanceIdentifier"], - "name": instance["DBInstanceIdentifier"], # Using ID as name - "status": standardized_status, # Using standardized status - "raw_status": raw_status, # Keep original status for debugging - "region": region, - "endpoint": endpoint_address, - "tags": tags, - } - - instances.append(instance_data) - - # Log the discovery action to CSV - if account_id: - log_action_to_csv( - account_id=account_id, - account_name=account_name or "Unknown", - resource_type="rds", - resource_id=instance["DBInstanceIdentifier"], - resource_name=instance["DBInstanceIdentifier"], - action="discover", - region=region, - status="success", - details=f"Status: {raw_status}, Endpoint: {endpoint_address}", - ) - - # Log final count of instances found - logger.info(f"Found {len(instances)} RDS instances in region {region}") - - # Log status distribution for debugging - status_counts = {} - for inst in instances: - status = inst["status"] - status_counts[status] = status_counts.get(status, 0) + 1 - - if status_counts: - logger.info( - f"RDS instance status distribution in {region}: {status_counts}" - ) - - return instances - except Exception as e: - error_msg = f"Error getting RDS instances in region {region}: {e}" - logger.error(error_msg) - - # Log the error to CSV - if account_id: - log_action_to_csv( - account_id=account_id, - account_name=account_name or "Unknown", - resource_type="rds", - resource_id="N/A", - resource_name="N/A", - action="discover", - region=region, - status="failed", - details=error_msg, - ) - - return [] - - -def get_eks_clusters(credentials, region, account_id=None, account_name=None): - """Get EKS clusters in a region.""" - try: - eks_client = create_boto3_client("eks", credentials, region) - clusters = [] - - # List all clusters - response = eks_client.list_clusters() - cluster_names = response.get("clusters", []) - - logger.info(f"Found {len(cluster_names)} EKS clusters in region {region}") - - # Get detailed info for each cluster - for cluster_name in cluster_names: - try: - cluster_info = eks_client.describe_cluster(name=cluster_name)["cluster"] - - # Get tags - tags = cluster_info.get("tags", {}) - - # Look for both individual scaling tags and combined format - min_nodes = tags.get("eks-min-nodes", "") - max_nodes = tags.get("eks-max-nodes", "") - desired_nodes = tags.get("eks-desired-nodes", "") - - # Parse the combined cluster:size tag if it exists - if "cluster:size" in tags: - size_tag = tags["cluster:size"] - logger.debug( - f"Found cluster:size tag: {size_tag} for cluster {cluster_name}" - ) - try: - # Parse min:X-max:Y-desired:Z format - if ( - "min:" in size_tag - and "max:" in size_tag - and "desired:" in size_tag - ): - # Extract values using regex or string parsing - min_match = ( - size_tag.split("min:")[1].split("-")[0] - if "min:" in size_tag - else None - ) - max_match = ( - size_tag.split("max:")[1].split("-")[0] - if "max:" in size_tag - else None - ) - desired_match = ( - size_tag.split("desired:")[1].split("-")[0] - if "desired:" in size_tag - else None - ) - - min_nodes = ( - min_match if min_match and not min_nodes else min_nodes - ) - max_nodes = ( - max_match if max_match and not max_nodes else max_nodes - ) - desired_nodes = ( - desired_match - if desired_match and not desired_nodes - else desired_nodes - ) - - logger.debug( - f"Parsed cluster:size: min={min_nodes}, max={max_nodes}, desired={desired_nodes}" - ) - except Exception as parse_error: - logger.warning( - f"Error parsing cluster:size tag for cluster {cluster_name}: {parse_error}" - ) - - # Check for backup tags that store original values - original_min = tags.get("eks-original-min-nodes", "") - original_max = tags.get("eks-original-max-nodes", "") - original_desired = tags.get("eks-original-desired-nodes", "") - - # Check for schedule tag - schedule = None - for tag_key, tag_value in tags.items(): - if "schedule" in tag_key.lower(): - schedule = tag_value - break - - cluster_data = { - "name": cluster_name, - "id": cluster_name, # Using name as ID for consistency - "status": cluster_info.get("status", "UNKNOWN"), - "region": region, - "endpoint": cluster_info.get("endpoint", ""), - "version": cluster_info.get("version", ""), - "min_nodes": min_nodes, - "max_nodes": max_nodes, - "desired_nodes": desired_nodes, - "original_min": original_min, - "original_max": original_max, - "original_desired": original_desired, - "schedule": schedule, - "tags": tags, - } - - clusters.append(cluster_data) - - # Log the discovery with scaling info and schedule - scale_details = f"Min: {min_nodes or 'N/A'}, Max: {max_nodes or 'N/A'}, Desired: {desired_nodes or 'N/A'}" - schedule_info = f", Schedule: {schedule}" if schedule else "" - - if account_id: - log_action_to_csv( - account_id=account_id, - account_name=account_name or "Unknown", - resource_type="eks", - resource_id=cluster_name, - resource_name=cluster_name, - action="discover", - region=region, - status="success", - details=f"Status: {cluster_data['status']}, Version: {cluster_data['version']}, Scaling: {scale_details}{schedule_info}", - existing_schedule=schedule, - ) - - except Exception as e: - logger.warning( - f"Error getting details for EKS cluster {cluster_name} in region {region}: {e}" - ) - - # Handle pagination - while "nextToken" in response: - response = eks_client.list_clusters(nextToken=response["nextToken"]) - cluster_names = response.get("clusters", []) - - for cluster_name in cluster_names: - try: - cluster_info = eks_client.describe_cluster(name=cluster_name)[ - "cluster" - ] - - # Get tags - tags = cluster_info.get("tags", {}) - - # Extract scaling parameters from tags if they exist - min_nodes = tags.get("eks-min-nodes", "") - max_nodes = tags.get("eks-max-nodes", "") - desired_nodes = tags.get("eks-desired-nodes", "") - - # Check for backup tags that store original values - original_min = tags.get("eks-original-min-nodes", "") - original_max = tags.get("eks-original-max-nodes", "") - original_desired = tags.get("eks-original-desired-nodes", "") - - cluster_data = { - "name": cluster_name, - "id": cluster_name, # Using name as ID for consistency - "status": cluster_info.get("status", "UNKNOWN"), - "region": region, - "endpoint": cluster_info.get("endpoint", ""), - "version": cluster_info.get("version", ""), - "min_nodes": min_nodes, - "max_nodes": max_nodes, - "desired_nodes": desired_nodes, - "original_min": original_min, - "original_max": original_max, - "original_desired": original_desired, - "tags": tags, - } - - clusters.append(cluster_data) - - # Log the discovery with scaling info - scale_details = f"Min: {min_nodes or 'N/A'}, Max: {max_nodes or 'N/A'}, Desired: {desired_nodes or 'N/A'}" - - if account_id: - log_action_to_csv( - account_id=account_id, - account_name=account_name or "Unknown", - resource_type="eks", - resource_id=cluster_name, - resource_name=cluster_name, - action="discover", - region=region, - status="success", - details=f"Status: {cluster_data['status']}, Version: {cluster_data['version']}, Scaling: {scale_details}", - ) - - except Exception as e: - logger.warning( - f"Error getting details for EKS cluster {cluster_name} in region {region}: {e}" - ) - - # Log the summary - scaling_stats = {} - for cluster in clusters: - has_scaling = any( - [ - cluster.get("min_nodes"), - cluster.get("max_nodes"), - cluster.get("desired_nodes"), - ] - ) - scaling_stats["with_scaling_tags"] = scaling_stats.get( - "with_scaling_tags", 0 - ) + (1 if has_scaling else 0) - scaling_stats["without_scaling_tags"] = scaling_stats.get( - "without_scaling_tags", 0 - ) + (0 if has_scaling else 1) - - if scaling_stats: - logger.info( - f"EKS clusters in {region}: {len(clusters)} total, " - f"{scaling_stats.get('with_scaling_tags', 0)} with scaling tags, " - f"{scaling_stats.get('without_scaling_tags', 0)} without scaling tags" - ) - - return clusters - - except Exception as e: - error_msg = f"Error getting EKS clusters in region {region}: {e}" - logger.error(error_msg) - - # Log the error to CSV - if account_id: - log_action_to_csv( - account_id=account_id, - account_name=account_name or "Unknown", - resource_type="eks", - resource_id="N/A", - resource_name="N/A", - action="discover", - region=region, - status="failed", - details=error_msg, - ) - - return [] - - -def get_emr_clusters(credentials, region, account_id=None, account_name=None): - """Get EMR clusters in a region.""" - try: - emr_client = create_boto3_client("emr", credentials, region) - clusters = [] - - # List active clusters (RUNNING, WAITING, STARTING, BOOTSTRAPPING, TERMINATING, TERMINATED_WITH_ERRORS) - response = emr_client.list_clusters( - ClusterStates=[ - "RUNNING", - "WAITING", - "STARTING", - "BOOTSTRAPPING", - "TERMINATING", - "TERMINATED_WITH_ERRORS", - ] - ) - - # Process the clusters - for cluster in response.get("Clusters", []): - try: - # Get detailed cluster info - detail = emr_client.describe_cluster(ClusterId=cluster["Id"]) - cluster_info = detail.get("Cluster", {}) - - # Extract tags - tags = {} - if "Tags" in cluster_info: - for tag in cluster_info["Tags"]: - tags[tag.get("Key", "")] = tag.get("Value", "") - - # Build the cluster data structure - cluster_data = { - "id": cluster["Id"], - "name": cluster.get("Name", "Unnamed cluster"), - "status": cluster.get("Status", {}).get("State", "UNKNOWN"), - "region": region, - "type": cluster_info.get("InstanceCollectionType", "UNKNOWN"), - "creation_time": str( - cluster.get("Status", {}) - .get("Timeline", {}) - .get("CreationDateTime", "") - ), - "tags": tags, - } - - clusters.append(cluster_data) - - # Log the discovery action to CSV - if account_id: - log_action_to_csv( - account_id=account_id, - account_name=account_name or "Unknown", - resource_type="emr", - resource_id=cluster["Id"], - resource_name=cluster.get("Name", "Unnamed cluster"), - action="discover", - region=region, - status="success", - details=f"Status: {cluster_data['status']}, Type: {cluster_data['type']}", - ) - except Exception as e: - logger.warning( - f"Error getting details for EMR cluster {cluster['Id']} in region {region}: {e}" - ) - - # Handle pagination - while "Marker" in response: - response = emr_client.list_clusters( - ClusterStates=[ - "RUNNING", - "WAITING", - "STARTING", - "BOOTSTRAPPING", - "TERMINATING", - "TERMINATED_WITH_ERRORS", - ], - Marker=response["Marker"], - ) - - for cluster in response.get("Clusters", []): - try: - # Get detailed cluster info - detail = emr_client.describe_cluster(ClusterId=cluster["Id"]) - cluster_info = detail.get("Cluster", {}) - - # Extract tags - tags = {} - if "Tags" in cluster_info: - for tag in cluster_info["Tags"]: - tags[tag.get("Key", "")] = tag.get("Value", "") - - # Build the cluster data structure - cluster_data = { - "id": cluster["Id"], - "name": cluster.get("Name", "Unnamed cluster"), - "status": cluster.get("Status", {}).get("State", "UNKNOWN"), - "region": region, - "type": cluster_info.get("InstanceCollectionType", "UNKNOWN"), - "creation_time": str( - cluster.get("Status", {}) - .get("Timeline", {}) - .get("CreationDateTime", "") - ), - "tags": tags, - } - - clusters.append(cluster_data) - - # Log the discovery action to CSV - if account_id: - log_action_to_csv( - account_id=account_id, - account_name=account_name or "Unknown", - resource_type="emr", - resource_id=cluster["Id"], - resource_name=cluster.get("Name", "Unnamed cluster"), - action="discover", - region=region, - status="success", - details=f"Status: {cluster_data['status']}, Type: {cluster_data['type']}", - ) - except Exception as e: - logger.warning( - f"Error getting details for EMR cluster {cluster['Id']} in region {region}: {e}" - ) - - logger.info(f"Found {len(clusters)} EMR clusters in region {region}") - return clusters - - except Exception as e: - error_msg = f"Error getting EMR clusters in region {region}: {e}" - logger.error(error_msg) - - # Log the error to CSV - if account_id: - log_action_to_csv( - account_id=account_id, - account_name=account_name or "Unknown", - resource_type="emr", - resource_id="N/A", - resource_name="N/A", - action="discover", - region=region, - status="failed", - details=error_msg, - ) - - return [] - - -def get_account_resources( - credentials, regions, exclude_resources=None, account_id=None, account_name=None -): - """Get all resources from an account across specified regions.""" - if exclude_resources is None: - exclude_resources = [] - - resources = { - "ec2_instances": [], - "rds_instances": [], - "eks_clusters": [], - "emr_clusters": [], - } - - for region in regions: - # Get EC2 instances - if "ec2" not in exclude_resources: - resources["ec2_instances"].extend( - get_ec2_instances(credentials, region, account_id, account_name) - ) - - # Get RDS instances - if "rds" not in exclude_resources: - resources["rds_instances"].extend( - get_rds_instances(credentials, region, account_id, account_name) - ) - - # Get EKS clusters - if "eks" not in exclude_resources: - resources["eks_clusters"].extend( - get_eks_clusters(credentials, region, account_id, account_name) - ) - - # Get EMR clusters - if "emr" not in exclude_resources: - resources["emr_clusters"].extend( - get_emr_clusters(credentials, region, account_id, account_name) - ) - - return resources diff --git a/local-app/python-tools/gfl-resource-actions/managers/__init__.py b/local-app/python-tools/gfl-resource-actions/managers/__init__.py deleted file mode 100644 index 535bbb30..00000000 --- a/local-app/python-tools/gfl-resource-actions/managers/__init__.py +++ /dev/null @@ -1,8 +0,0 @@ -""" -Resource managers for different AWS resource types. -""" - -from managers.ec2 import EC2Manager -from managers.eks import EKSManager -from managers.emr import EMRManager -from managers.rds import RDSManager diff --git a/local-app/python-tools/gfl-resource-actions/managers/base.py b/local-app/python-tools/gfl-resource-actions/managers/base.py deleted file mode 100644 index 4712357d..00000000 --- a/local-app/python-tools/gfl-resource-actions/managers/base.py +++ /dev/null @@ -1,187 +0,0 @@ -""" -Base resource manager class that provides common functionality for AWS resource management. - -This module contains the ResourceManager base class which handles common operations -such as client creation, action logging, and resource filtering. -""" - -import datetime -from typing import Any, Dict, Optional, Union - -import boto3 -import config -from aws_resource_management.aws_utils import get_partition_for_region -from botocore.exceptions import ClientError -from logging_utils import log_action_to_csv, setup_logging - -logger = setup_logging() - - -class ResourceManager: - """Base class for all resource managers providing common AWS resource functionality.""" - - def __init__( - self, - credentials: Dict[str, str], - account_id: str, - account_name: Optional[str] = None, - ): - """ - Initialize resource manager with AWS credentials. - - Args: - credentials: AWS credentials dictionary containing AccessKeyId, SecretAccessKey, and SessionToken - account_id: AWS account ID - account_name: AWS account name (optional, defaults to account_id if not provided) - """ - self.credentials = credentials - self.account_id = account_id - self.account_name = account_name or account_id - self.resource_type = "generic" - self._clients = {} # Cache for boto3 clients - - def create_client(self, service: str, region: str) -> Optional[boto3.client]: - """ - Create a boto3 client for the specified service and region. - - Args: - service: AWS service name (e.g., 'ec2', 's3') - region: AWS region name (e.g., 'us-east-1') - - Returns: - Configured boto3 client or None if credentials are not available - - Raises: - ClientError: If there's an error creating the client - """ - if not self.credentials: - logger.warning( - f"No credentials available to create {service} client in {region}" - ) - return None - - # Use client caching to avoid creating redundant clients - cache_key = f"{service}:{region}" - if cache_key in self._clients: - return self._clients[cache_key] - - try: - client = boto3.client( - service, - region_name=region, - aws_access_key_id=self.credentials["AccessKeyId"], - aws_secret_access_key=self.credentials["SecretAccessKey"], - aws_session_token=self.credentials["SessionToken"], - ) - self._clients[cache_key] = client - return client - except ClientError as e: - logger.error(f"Failed to create {service} client in {region}: {str(e)}") - raise - - def get_partition_for_region(self, region: str) -> str: - """ - Get AWS partition for the specified region. - - Args: - region: AWS region name - - Returns: - AWS partition (e.g., 'aws', 'aws-cn', 'aws-us-gov') - """ - return get_partition_for_region(region) - - def get_timestamp(self) -> str: - """ - Get ISO 8601 timestamp for tagging and logging purposes. - - Returns: - Current UTC timestamp in ISO 8601 format - """ - return datetime.datetime.utcnow().isoformat() - - def should_exclude(self, resource: Dict[str, Any]) -> bool: - """ - Check if resource should be excluded based on tags. - - Args: - resource: Resource dictionary containing tags - - Returns: - True if the resource should be excluded, False otherwise - """ - tags = resource.get("tags", {}) - return config.EXCLUSION_TAG in tags - - 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, - status: str = "success", - ) -> None: - """ - Log an action performed on a resource. - - Args: - resource_id: Resource identifier - region: AWS region - action: Action performed (stop, start, etc.) - resource_name: Optional resource name - details: Optional additional details - dry_run: Whether this was a dry run - existing_schedule: Existing schedule for the resource - status: Action status (default: 'success') - """ - dry_run_prefix = "[DRY RUN] " if dry_run else "" - - # Build a detailed log message - resource_desc = f"{resource_id}" - if resource_name: - resource_desc = f"{resource_name} ({resource_id})" - - account_desc = f"account {self.account_id}" - if self.account_name and self.account_name != self.account_id: - account_desc = f"{self.account_name} ({self.account_id})" - - # Convert action to past tense for logging - action_desc = f"{action}ed" if not action.endswith("e") else f"{action}d" - - message = f"{dry_run_prefix}{self.resource_type.upper()} {resource_desc} {action_desc} in {account_desc} region {region}" - - if details: - message += f" - {details}" - - # Add schedule information if available - if existing_schedule: - message += f" (Schedule: {existing_schedule})" - - # Use logging_utils.log_action_to_csv for consistent CSV logging - status_value = "simulated" if dry_run else status - log_action_to_csv( - account_id=self.account_id, - account_name=self.account_name, - resource_type=self.resource_type, - resource_id=resource_id, - resource_name=resource_name - or resource_id, # Use ID as name if not provided - action=action, - region=region, - status=status_value, - details=details, - dry_run=dry_run, - existing_schedule=existing_schedule, - ) - logger.info(message) - - def cleanup(self) -> None: - """ - Perform cleanup operations when the manager is no longer needed. - This method should be called when finished with the resource manager. - """ - # Clear client cache to allow garbage collection - self._clients.clear() diff --git a/local-app/python-tools/gfl-resource-actions/managers/ec2.py b/local-app/python-tools/gfl-resource-actions/managers/ec2.py deleted file mode 100644 index a2471e04..00000000 --- a/local-app/python-tools/gfl-resource-actions/managers/ec2.py +++ /dev/null @@ -1,117 +0,0 @@ -""" -EC2 resource manager class. -""" - -import config -from logging_utils import setup_logging -from managers import base - -logger = setup_logging() - - -class EC2Manager(base.ResourceManager): - """Manager for EC2 instances.""" - - def __init__(self, credentials, account_id, account_name=None): - """Initialize EC2 manager.""" - super().__init__(credentials, account_id, account_name) - self.resource_type = "ec2_instance" - - def should_exclude(self, instance): - """Check if EC2 instance should be excluded based on tags.""" - tags = instance.get("tags", {}) - - # Check for explicit exclusion tag - if config.EXCLUSION_TAG in tags: - logger.info( - f"Skipping EC2 instance {instance['id']} with exclusion tag {config.EXCLUSION_TAG}" - ) - return True - - # Skip instances that are part of EKS clusters - if config.EKS_TAG in tags: - logger.info( - f"Skipping EC2 instance {instance['id']} with tag {config.EKS_TAG}={tags[config.EKS_TAG]} (EKS managed node)" - ) - return True - - # Skip instances that are part of EMR clusters - if config.EMR_TAG in tags: - logger.info( - f"Skipping EC2 instance {instance['id']} with tag {config.EMR_TAG}={tags[config.EMR_TAG]} (EMR managed node)" - ) - return True - - return False - - def stop(self, instances, dry_run=False): - """Stop running EC2 instances.""" - for instance in instances: - if self.should_exclude(instance): - continue - - if instance["state"] == "running": - try: - region = instance["region"] - ec2_client = self.create_client("ec2", region) - - # Create ISO 8601 timestamp - timestamp = self.get_timestamp() - - # Tag the instance before stopping - logger.info( - f"{'[DRY RUN] Would tag' if dry_run else 'Tagging'} EC2 instance {instance['id']} with {config.STOP_TAG}={timestamp}" - ) - if not dry_run: - ec2_client.create_tags( - Resources=[instance["id"]], - Tags=[{"Key": config.STOP_TAG, "Value": timestamp}], - ) - - # Stop the instance - logger.info( - f"{'[DRY RUN] Would stop' if dry_run else 'Stopping'} EC2 instance {instance['id']} in region {region}" - ) - if not dry_run: - ec2_client.stop_instances(InstanceIds=[instance["id"]]) - - # Log with detailed information - self.log_action(instance["id"], region, "stop", dry_run=dry_run) - - except Exception as e: - logger.error( - f"Failed to {'simulate stopping' if dry_run else 'stop'} EC2 instance {instance['id']}: {e}" - ) - - def start(self, instances, dry_run=False): - """Start stopped EC2 instances.""" - for instance in instances: - if self.should_exclude(instance): - continue - - if instance["state"] == "stopped": - try: - region = instance["region"] - ec2_client = self.create_client("ec2", region) - - logger.info( - f"{'[DRY RUN] Would start' if dry_run else 'Starting'} EC2 instance {instance['id']} in region {region}" - ) - if not dry_run: - ec2_client.start_instances(InstanceIds=[instance["id"]]) - - # Remove the stop tag after successful start - logger.info( - f"Removing {config.STOP_TAG} tag from EC2 instance {instance['id']}" - ) - ec2_client.delete_tags( - Resources=[instance["id"]], Tags=[{"Key": config.STOP_TAG}] - ) - - # Log with detailed information - self.log_action(instance["id"], region, "start", dry_run=dry_run) - - except Exception as e: - logger.error( - f"Failed to {'simulate starting' if dry_run else 'start'} EC2 instance {instance['id']}: {e}" - ) diff --git a/local-app/python-tools/gfl-resource-actions/managers/eks.py b/local-app/python-tools/gfl-resource-actions/managers/eks.py deleted file mode 100644 index 25e3843d..00000000 --- a/local-app/python-tools/gfl-resource-actions/managers/eks.py +++ /dev/null @@ -1,462 +0,0 @@ -""" -EKS resource manager class. -""" - -import config -from logging_utils import setup_logging -from managers import base - -logger = setup_logging() - -# Tag names for EKS scaling -EKS_MIN_NODES_TAG = "eks-min-nodes" -EKS_MAX_NODES_TAG = "eks-max-nodes" -EKS_DESIRED_NODES_TAG = "eks-desired-nodes" -EKS_ORIGINAL_MIN_NODES_TAG = "eks-original-min-nodes" -EKS_ORIGINAL_MAX_NODES_TAG = "eks-original-max-nodes" -EKS_ORIGINAL_DESIRED_NODES_TAG = "eks-original-desired-nodes" -CLUSTER_SIZE_TAG = "cluster:size" -EKS_ORIGINAL_CLUSTER_SIZE_TAG = "eks-original-cluster-size" - - -class EKSManager(base.ResourceManager): - """Manager for EKS clusters.""" - - def __init__(self, credentials, account_id, account_name=None): - """Initialize EKS manager.""" - super().__init__(credentials, account_id, account_name) - self.resource_type = "eks_cluster" - - # Add stop method that delegates to scale_down - def stop(self, clusters, dry_run=False): - """ - Stop EKS clusters by scaling down their nodegroups. - - Args: - clusters: List of EKS cluster dictionaries - dry_run: If True, only simulate the action - - Returns: - dict: Summary of actions taken - """ - logger.info( - f"Delegating EKS stop operation to scale_down for {len(clusters)} clusters" - ) - self.scale_down(clusters, dry_run) - - # Return a result dictionary that matches the expected interface - return {"success": len(clusters), "errors": 0} - - # Add start method that delegates to scale_up - def start(self, clusters, dry_run=False): - """ - Start EKS clusters by scaling up their nodegroups. - - Args: - clusters: List of EKS cluster dictionaries - dry_run: If True, only simulate the action - - Returns: - dict: Summary of actions taken - """ - logger.info( - f"Delegating EKS start operation to scale_up for {len(clusters)} clusters" - ) - self.scale_up(clusters, dry_run) - - # Return a result dictionary that matches the expected interface - return {"success": len(clusters), "errors": 0} - - def scale_down(self, clusters, dry_run=False): - """Scale down EKS clusters by setting node counts to 0.""" - logger.info(f"Evaluating {len(clusters)} EKS clusters for scale down action") - scaled_count = 0 - skipped_count = 0 - - for cluster in clusters: - # Skip clusters with the exclusion tag - if self.should_exclude(cluster): - logger.info( - f"Skipping EKS cluster {cluster['name']} with exclusion tag {config.EXCLUSION_TAG}" - ) - skipped_count += 1 - continue - - try: - region = cluster["region"] - eks_client = self.create_client("eks", region) - - # Get current scaling parameters from tags or nodegroups - min_nodes = cluster.get("min_nodes", "") - max_nodes = cluster.get("max_nodes", "") - desired_nodes = cluster.get("desired_nodes", "") - schedule = cluster.get("schedule", "") - - # Check if we have a combined cluster:size tag - has_combined_size_tag = CLUSTER_SIZE_TAG in cluster.get("tags", {}) - - # Only proceed if we have some scaling values to work with - if min_nodes or max_nodes or desired_nodes or has_combined_size_tag: - logger.info( - f"{'[DRY RUN] Would scale down' if dry_run else 'Scaling down'} EKS cluster {cluster['name']} in region {region}" - ) - - if not dry_run: - # First, backup the current values in special tags - tags_to_update = {} - - # Handle combined size tag if it exists - if has_combined_size_tag: - original_size_tag = cluster["tags"][CLUSTER_SIZE_TAG] - tags_to_update[EKS_ORIGINAL_CLUSTER_SIZE_TAG] = ( - original_size_tag - ) - tags_to_update[CLUSTER_SIZE_TAG] = "min:0-max:0-desired:0" - else: - # Store original values in backup tags (individual tags) - if min_nodes: - tags_to_update[EKS_ORIGINAL_MIN_NODES_TAG] = min_nodes - tags_to_update[EKS_MIN_NODES_TAG] = "0" - if max_nodes: - tags_to_update[EKS_ORIGINAL_MAX_NODES_TAG] = max_nodes - tags_to_update[EKS_MAX_NODES_TAG] = "0" - if desired_nodes: - tags_to_update[EKS_ORIGINAL_DESIRED_NODES_TAG] = ( - desired_nodes - ) - tags_to_update[EKS_DESIRED_NODES_TAG] = "0" - - # Apply the tag updates - if tags_to_update: - logger.info( - f"Updating scaling tags for EKS cluster {cluster['name']}: {tags_to_update}" - ) - eks_client.tag_resource( - resourceArn=cluster.get( - "arn", - f"arn:{self.get_partition_for_region(region)}:eks:{region}:{self.account_id}:cluster/{cluster['name']}", - ), - tags=tags_to_update, - ) - - # Now actually scale down the nodegroups - self._scale_nodegroups( - eks_client, cluster["name"], 0, region, dry_run=False - ) - else: - # Dry run reporting - if has_combined_size_tag: - original_size = cluster["tags"][CLUSTER_SIZE_TAG] - logger.info( - f"[DRY RUN] Would backup combined size tag: {original_size} and set to min:0-max:0-desired:0" - ) - else: - logger.info( - f"[DRY RUN] Would backup scaling values: Min={min_nodes}, Max={max_nodes}, Desired={desired_nodes}" - ) - - logger.info( - f"[DRY RUN] Would set scaling values to 0 for EKS cluster {cluster['name']}" - ) - self._scale_nodegroups( - eks_client, cluster["name"], 0, region, dry_run=True - ) - - # Log the action with schedule information - schedule_info = f", Schedule: {schedule}" if schedule else "" - self.log_action( - cluster["name"], - region, - "scale_down", - details=f"Min={min_nodes}->{0}, Max={max_nodes}->{0}, Desired={desired_nodes}->{0}{schedule_info}", - dry_run=dry_run, - existing_schedule=schedule, - ) - scaled_count += 1 - else: - logger.info( - f"Skipping EKS cluster {cluster['name']}, no scaling tags found" - ) - skipped_count += 1 - - except Exception as e: - logger.error( - f"Failed to {'simulate scaling down' if dry_run else 'scale down'} EKS cluster {cluster['name']}: {e}" - ) - - logger.info( - f"EKS scale down summary: {scaled_count} clusters processed, {skipped_count} clusters skipped" - ) - - def scale_up(self, clusters, dry_run=False): - """Scale up EKS clusters using original node counts from tags.""" - logger.info(f"Evaluating {len(clusters)} EKS clusters for scale up action") - scaled_count = 0 - skipped_count = 0 - - for cluster in clusters: - # Skip clusters with the exclusion tag - if self.should_exclude(cluster): - logger.info( - f"Skipping EKS cluster {cluster['name']} with exclusion tag {config.EXCLUSION_TAG}" - ) - skipped_count += 1 - continue - - try: - region = cluster["region"] - eks_client = self.create_client("eks", region) - - # Check for original values in backup tags - original_min = cluster.get("original_min", "") - original_max = cluster.get("original_max", "") - original_desired = cluster.get("original_desired", "") - schedule = cluster.get("schedule", "") - - # Check for original combined size tag - tags = cluster.get("tags", {}) - has_original_combined_tag = EKS_ORIGINAL_CLUSTER_SIZE_TAG in tags - - # If we have backup values or original combined tag, restore them - if ( - original_min - or original_max - or original_desired - or has_original_combined_tag - ): - logger.info( - f"{'[DRY RUN] Would scale up' if dry_run else 'Scaling up'} EKS cluster {cluster['name']} in region {region}" - ) - - if not dry_run: - # Restore the original values from backup tags - tags_to_update = {} - tags_to_remove = [] - - # Handle original combined size tag if it exists - if has_original_combined_tag: - original_size_tag = tags[EKS_ORIGINAL_CLUSTER_SIZE_TAG] - tags_to_update[CLUSTER_SIZE_TAG] = original_size_tag - tags_to_remove.append(EKS_ORIGINAL_CLUSTER_SIZE_TAG) - - # Parse the original desired value from the size tag - desired = None - if "desired:" in original_size_tag: - try: - desired_str = original_size_tag.split("desired:")[ - 1 - ].split("-")[0] - desired = ( - int(desired_str) - if desired_str.isdigit() - else None - ) - except: - logger.warning( - f"Could not parse desired value from combined size tag: {original_size_tag}" - ) - else: - # Restore original values for individual tags - if original_min: - tags_to_update[EKS_MIN_NODES_TAG] = original_min - tags_to_remove.append(EKS_ORIGINAL_MIN_NODES_TAG) - if original_max: - tags_to_update[EKS_MAX_NODES_TAG] = original_max - tags_to_remove.append(EKS_ORIGINAL_MAX_NODES_TAG) - if original_desired: - tags_to_update[EKS_DESIRED_NODES_TAG] = original_desired - tags_to_remove.append(EKS_ORIGINAL_DESIRED_NODES_TAG) - - # Parse desired value - desired = ( - int(original_desired) - if original_desired and original_desired.isdigit() - else None - ) - - # Apply the tag updates - if tags_to_update: - logger.info( - f"Restoring original scaling tags for EKS cluster {cluster['name']}: {tags_to_update}" - ) - eks_client.tag_resource( - resourceArn=cluster.get( - "arn", - f"arn:{self.get_partition_for_region(region)}:eks:{region}:{self.account_id}:cluster/{cluster['name']}", - ), - tags=tags_to_update, - ) - - # Now actually scale up the nodegroups to desired capacity - self._scale_nodegroups( - eks_client, - cluster["name"], - desired, - region, - dry_run=False, - ) - - # Remove the backup tags - if tags_to_remove: - logger.info( - f"Removing backup scaling tags: {tags_to_remove}" - ) - eks_client.untag_resource( - resourceArn=cluster.get( - "arn", - f"arn:{self.get_partition_for_region(region)}:eks:{region}:{self.account_id}:cluster/{cluster['name']}", - ), - tagKeys=tags_to_remove, - ) - else: - # Dry run reporting - if has_original_combined_tag: - original_size = tags[EKS_ORIGINAL_CLUSTER_SIZE_TAG] - logger.info( - f"[DRY RUN] Would restore combined size tag: {original_size}" - ) - - # Try to parse desired value for nodegroup scaling - desired = None - if "desired:" in original_size: - try: - desired_str = original_size.split("desired:")[ - 1 - ].split("-")[0] - desired = ( - int(desired_str) - if desired_str.isdigit() - else None - ) - except: - pass - else: - logger.info( - f"[DRY RUN] Would restore scaling values: Min={original_min}, Max={original_max}, Desired={original_desired}" - ) - desired = ( - int(original_desired) - if original_desired and original_desired.isdigit() - else None - ) - - self._scale_nodegroups( - eks_client, cluster["name"], desired, region, dry_run=True - ) - - # Log the action with schedule information - scale_details = "" - if has_original_combined_tag: - original_size = tags.get( - EKS_ORIGINAL_CLUSTER_SIZE_TAG, "Unknown" - ) - scale_details = f"Size tag: 0->'{original_size}'" - else: - scale_details = f"Min=0->{original_min}, Max=0->{original_max}, Desired=0->{original_desired}" - - schedule_info = f", Schedule: {schedule}" if schedule else "" - self.log_action( - cluster["name"], - region, - "scale_up", - details=f"{scale_details}{schedule_info}", - dry_run=dry_run, - existing_schedule=schedule, - ) - scaled_count += 1 - else: - logger.info( - f"Skipping EKS cluster {cluster['name']}, no original scaling values found in tags" - ) - skipped_count += 1 - - except Exception as e: - logger.error( - f"Failed to {'simulate scaling up' if dry_run else 'scale up'} EKS cluster {cluster['name']}: {e}" - ) - - logger.info( - f"EKS scale up summary: {scaled_count} clusters processed, {skipped_count} clusters skipped" - ) - - def _scale_nodegroups( - self, eks_client, cluster_name, desired_capacity, region, dry_run=False - ): - """Helper method to scale nodegroups to the specified capacity.""" - try: - # List all nodegroups for this cluster - response = eks_client.list_nodegroups(clusterName=cluster_name) - nodegroup_names = response.get("nodegroups", []) - - # Handle pagination - while "nextToken" in response: - response = eks_client.list_nodegroups( - clusterName=cluster_name, nextToken=response["nextToken"] - ) - nodegroup_names.extend(response.get("nodegroups", [])) - - if not nodegroup_names: - logger.info(f"No nodegroups found for EKS cluster {cluster_name}") - return - - for nodegroup_name in nodegroup_names: - try: - # Get nodegroup details - nodegroup = eks_client.describe_nodegroup( - clusterName=cluster_name, nodegroupName=nodegroup_name - ).get("nodegroup", {}) - - # Get current scaling configuration - current_min = nodegroup.get("scalingConfig", {}).get("minSize") - current_max = nodegroup.get("scalingConfig", {}).get("maxSize") - current_desired = nodegroup.get("scalingConfig", {}).get( - "desiredSize" - ) - - # Use current min and max when scaling up if desired capacity is provided - if desired_capacity is not None: - # When scaling up, we need a valid min and max - new_min = ( - current_min - if desired_capacity == 0 - else min(current_min, desired_capacity) - ) - new_max = max(current_max, desired_capacity) - else: - # When scaling down to zero - new_min = 0 - new_max = current_max - desired_capacity = 0 - - if dry_run: - logger.info( - f"[DRY RUN] Would update nodegroup {nodegroup_name} scaling: " - + f"Min: {current_min}->{new_min}, " - + f"Max: {current_max}->{new_max}, " - + f"Desired: {current_desired}->{desired_capacity}" - ) - else: - logger.info( - f"Updating nodegroup {nodegroup_name} scaling: " - + f"Min: {current_min}->{new_min}, " - + f"Max: {current_max}->{new_max}, " - + f"Desired: {current_desired}->{desired_capacity}" - ) - - # Update the nodegroup scaling configuration - eks_client.update_nodegroup_config( - clusterName=cluster_name, - nodegroupName=nodegroup_name, - scalingConfig={ - "minSize": new_min, - "maxSize": new_max, - "desiredSize": desired_capacity, - }, - ) - except Exception as e: - logger.error(f"Error updating nodegroup {nodegroup_name}: {e}") - - except Exception as e: - logger.error( - f"Error listing nodegroups for cluster {cluster_name} in region {region}: {e}" - ) diff --git a/local-app/python-tools/gfl-resource-actions/managers/emr.py b/local-app/python-tools/gfl-resource-actions/managers/emr.py deleted file mode 100644 index 90bd565e..00000000 --- a/local-app/python-tools/gfl-resource-actions/managers/emr.py +++ /dev/null @@ -1,151 +0,0 @@ -""" -EMR resource manager class. -""" - -import config -from logging_utils import setup_logging -from managers import base - -logger = setup_logging() - - -class EMRManager(base.ResourceManager): - """Manager for EMR clusters.""" - - def __init__(self, credentials, account_id, account_name=None): - """Initialize EMR manager.""" - super().__init__(credentials, account_id, account_name) - self.resource_type = "emr_cluster" - - def stop(self, clusters, dry_run=False): - """ - Stop EMR clusters gracefully. - This preserves the cluster configuration and data. - """ - for cluster in clusters: - # Skip clusters with the exclusion tag - if self.should_exclude(cluster): - logger.info( - f"Skipping EMR cluster {cluster['id']} with exclusion tag {config.EXCLUSION_TAG}" - ) - continue - - # Only RUNNING and WAITING clusters can be stopped - if cluster["status"] in ["RUNNING", "WAITING"]: - try: - region = cluster["region"] - emr_client = self.create_client("emr", region) - - timestamp = self.get_timestamp() - - # Tag the cluster before stopping - logger.info( - f"{'[DRY RUN] Would tag' if dry_run else 'Tagging'} EMR cluster {cluster['name']} ({cluster['id']}) with {config.STOP_TAG}={timestamp}" - ) - if not dry_run: - emr_client.add_tags( - ResourceId=cluster["id"], - Tags=[{"Key": config.STOP_TAG, "Value": timestamp}], - ) - - logger.info( - f"{'[DRY RUN] Would stop' if dry_run else 'Stopping'} EMR cluster {cluster['name']} ({cluster['id']}) in region {region}" - ) - if not dry_run: - emr_client.stop_cluster(ClusterId=cluster["id"]) - - # Log with detailed information - self.log_action( - cluster["id"], - region, - "stop", - resource_name=cluster["name"], - dry_run=dry_run, - ) - - except Exception as e: - logger.error( - f"Failed to {'simulate stopping' if dry_run else 'stop'} EMR cluster {cluster['id']}: {e}" - ) - # For STARTING and BOOTSTRAPPING clusters, we need to terminate them as they can't be stopped - elif cluster["status"] in ["STARTING", "BOOTSTRAPPING"]: - try: - region = cluster["region"] - emr_client = self.create_client("emr", region) - - timestamp = self.get_timestamp() - - # Tag the cluster before terminating - logger.info( - f"{'[DRY RUN] Would tag' if dry_run else 'Tagging'} EMR cluster {cluster['name']} ({cluster['id']}) with {config.STOP_TAG}={timestamp}" - ) - if not dry_run: - emr_client.add_tags( - ResourceId=cluster["id"], - Tags=[{"Key": config.STOP_TAG, "Value": timestamp}], - ) - - logger.info( - f"{'[DRY RUN] Would terminate' if dry_run else 'Terminating'} EMR cluster {cluster['name']} ({cluster['id']}) in region {region} (cannot stop a cluster in {cluster['status']} state)" - ) - if not dry_run: - emr_client.terminate_job_flows(JobFlowIds=[cluster["id"]]) - - # Log with detailed information - self.log_action( - cluster["id"], - region, - "terminate", - resource_name=cluster["name"], - dry_run=dry_run, - ) - - except Exception as e: - logger.error( - f"Failed to {'simulate terminating' if dry_run else 'terminate'} EMR cluster {cluster['id']}: {e}" - ) - - def start(self, clusters, dry_run=False): - """Start stopped EMR clusters.""" - for cluster in clusters: - # Skip clusters with the exclusion tag - if self.should_exclude(cluster): - logger.info( - f"Skipping EMR cluster {cluster['id']} with exclusion tag {config.EXCLUSION_TAG}" - ) - continue - - if cluster["status"] == "STOPPED": - try: - region = cluster["region"] - emr_client = self.create_client("emr", region) - - timestamp = self.get_timestamp() - - logger.info( - f"{'[DRY RUN] Would start' if dry_run else 'Starting'} EMR cluster {cluster['name']} ({cluster['id']}) in region {region}" - ) - if not dry_run: - emr_client.start_cluster(ClusterId=cluster["id"]) - - # Remove the stop tag after successful start - logger.info( - f"Removing {config.STOP_TAG} tag from EMR cluster {cluster['id']}" - ) - emr_client.remove_tags( - ResourceId=cluster["id"], TagKeys=[config.STOP_TAG] - ) - - # Log with detailed information - self.log_action( - cluster["id"], - region, - "start", - resource_name=cluster["name"], - dry_run=dry_run, - ) - - except Exception as e: - logger.error( - f"Failed to {'simulate starting' if dry_run else 'start'} EMR cluster {cluster['id']}: {e}" - ) diff --git a/local-app/python-tools/gfl-resource-actions/managers/rds.py b/local-app/python-tools/gfl-resource-actions/managers/rds.py deleted file mode 100644 index 37142699..00000000 --- a/local-app/python-tools/gfl-resource-actions/managers/rds.py +++ /dev/null @@ -1,119 +0,0 @@ -""" -RDS resource manager class. -""" - -import config -from logging_utils import setup_logging -from managers import base - -logger = setup_logging() - - -class RDSManager(base.ResourceManager): - """Manager for RDS instances.""" - - def __init__(self, credentials, account_id, account_name=None): - """Initialize RDS manager.""" - super().__init__(credentials, account_id, account_name) - self.resource_type = "rds_instance" - - def stop(self, instances, dry_run=False): - """Stop available RDS instances.""" - logger.info(f"Evaluating {len(instances)} RDS instances for stop action") - stopped_count = 0 - skipped_count = 0 - - for instance in instances: - # Skip instances with the exclusion tag - if self.should_exclude(instance): - logger.info( - f"Skipping RDS instance {instance['id']} with exclusion tag {config.EXCLUSION_TAG}" - ) - skipped_count += 1 - continue - - # Log status for debugging - logger.debug(f"RDS instance {instance['id']} status: {instance['status']}") - - if instance["status"] == "available": - try: - region = instance["region"] - rds_client = self.create_client("rds", region) - - # Create ISO 8601 timestamp - timestamp = self.get_timestamp() - - # Tag the instance before stopping - logger.info( - f"{'[DRY RUN] Would tag' if dry_run else 'Tagging'} RDS instance {instance['id']} with {config.STOP_TAG}={timestamp}" - ) - if not dry_run: - rds_client.add_tags_to_resource( - ResourceName=f"arn:{self.get_partition_for_region(region)}:rds:{region}:{self.account_id}:db:{instance['id']}", - Tags=[{"Key": config.STOP_TAG, "Value": timestamp}], - ) - - logger.info( - f"{'[DRY RUN] Would stop' if dry_run else 'Stopping'} RDS instance {instance['id']} in region {region}" - ) - if not dry_run: - rds_client.stop_db_instance(DBInstanceIdentifier=instance["id"]) - - # Log with detailed information - self.log_action(instance["id"], region, "stop", dry_run=dry_run) - - stopped_count += 1 - - except Exception as e: - logger.error( - f"Failed to {'simulate stopping' if dry_run else 'stop'} RDS instance {instance['id']}: {e}" - ) - else: - logger.debug( - f"Skipping RDS instance {instance['id']} with non-available status: {instance['status']}" - ) - skipped_count += 1 - - logger.info( - f"RDS stop summary: {stopped_count} instances processed, {skipped_count} instances skipped" - ) - - def start(self, instances, dry_run=False): - """Start stopped RDS instances.""" - for instance in instances: - # Skip instances with the exclusion tag - if self.should_exclude(instance): - logger.info( - f"Skipping RDS instance {instance['id']} with exclusion tag {config.EXCLUSION_TAG}" - ) - continue - - if instance["status"] == "stopped": - try: - region = instance["region"] - rds_client = self.create_client("rds", region) - - logger.info( - f"{'[DRY RUN] Would start' if dry_run else 'Starting'} RDS instance {instance['id']} in region {region}" - ) - if not dry_run: - rds_client.start_db_instance( - DBInstanceIdentifier=instance["id"] - ) - - # Remove the stop tag after successful start - logger.info( - f"Removing {config.STOP_TAG} tag from RDS instance {instance['id']}" - ) - rds_client.remove_tags_from_resource( - ResourceName=f"arn:{self.get_partition_for_region(region)}:rds:{region}:{self.account_id}:db:{instance['id']}", - TagKeys=[config.STOP_TAG], - ) - - # Log with detailed information - self.log_action(instance["id"], region, "start", dry_run=dry_run) - - except Exception as e: - logger.error( - f"Failed to {'simulate starting' if dry_run else 'start'} RDS instance {instance['id']}: {e}" - )