from .base_agent import BaseAgent from ..aws_connector import AWSConnector class EC2Agent(BaseAgent): """ An agent specialized in handling AWS EC2 tasks. """ def execute(self, command: str, **kwargs): """ Executes a given command related to EC2. """ if command == 'list_instances': return self._list_instances(**kwargs) elif command == 'start_instances': return self._start_instances(**kwargs) elif command == 'stop_instances': return self._stop_instances(**kwargs) elif command == 'terminate_instances': pass # Placeholder for confirmation logic elif command == 'create_instance': print(f"You are about to create a new EC2 instance named '{kwargs.get('name')}' with default settings.") confirm = input("This will create a new Security Group. Are you sure you want to proceed? (yes/no): ") if confirm.lower() == 'yes': return self._create_instance_smart(**kwargs) else: return {"status": "cancelled", "message": "Create instance command cancelled by user."} elif command == 'list_idle_instances': region = kwargs.get('region', '') print(f"You are about to list potentially idle EC2 instances in {region}.") confirm = input("This will retrieve CloudWatch metrics. Are you sure you want to proceed? (yes/no): ") if confirm.lower() == 'yes': return self._list_idle_instances(**kwargs) else: return {"status": "cancelled", "message": "List idle instances command cancelled by user."} else: raise NotImplementedError(f"Command '{command}' is not supported by EC2Agent.") def _create_instance_smart(self, region: str, name: str, os: str, vpc_id: str = None, subnet_id: str = None, instance_type: str = 't3.micro'): """ Creates an EC2 instance by automatically handling dependencies. - Finds the latest AMI for the specified OS (Ubuntu). - Creates a new Security Group with HTTP and SSH access. - Launches the instance in the specified VPC/Subnet or the default VPC. """ print(f"EC2Agent: Beginning Smart Create for instance '{name}' in region {region}...") try: ec2_client = AWSConnector.get_client('ec2', region_name=region) # 1. Determine VPC and Subnet if vpc_id is None or subnet_id is None: # Fallback to default VPC if not provided\n vpcs = ec2_client.describe_vpcs(Filters=[{'Name': 'isDefault', 'Values': ['true']}]).get('Vpcs', []) if not vpcs: return {"status": "error", "message": "No default VPC found in the region."} vpc_id = vpcs[0]['VpcId'] subnets = ec2_client.describe_subnets(Filters=[{'Name': 'vpc-id', 'Values': [vpc_id]}]).get('Subnets', []) if not subnets: return {"status": "error", "message": f"No subnets found in the default VPC '{vpc_id}'."} subnet_id = subnets[0]['SubnetId'] # 2. Find latest AMI based on OS ami_id = None if os.lower() == 'ubuntu': ami_response = ec2_client.describe_images( Owners=['099720109477'], # Canonical's owner ID Filters=[ {'Name': 'name', 'Values': ['ubuntu/images/hvm-ssd/ubuntu-jammy-22.04-amd64-server-*']}, {'Name': 'virtualization-type', 'Values': ['hvm']} ], MostRecent=True ) if ami_response.get('Images'): ami_id = ami_response['Images'][0]['ImageId'] else: return {"status": "error", "message": "Could not find latest Ubuntu 22.04 AMI."} elif os.lower() == 'amazon_linux_2': ami_response = ec2_client.describe_images( Owners=['137112412989'], # Amazon's owner ID for Amazon Linux Filters=[ {'Name': 'name', 'Values': ['amzn2-ami-hvm-*-x86_64-gp2']}, {'Name': 'virtualization-type', 'Values': ['hvm']} ], MostRecent=True ) if ami_response.get('Images'): ami_id = ami_response['Images'][0]['ImageId'] else: return {"status": "error", "message": "Could not find latest Amazon Linux 2 AMI."} else: return {"status": "error", "message": f"Unsupported OS: {os}. Only 'ubuntu' and 'amazon_linux_2' are supported for smart create."} # 3. Create Security Group sg_name = f"{name}-sg" sg = ec2_client.create_security_group(GroupName=sg_name, Description=f"Security group for {name}", VpcId=vpc_id) sg_id = sg['GroupId'] ec2_client.authorize_security_group_ingress( GroupId=sg_id, IpPermissions=[ {'IpProtocol': 'tcp', 'FromPort': 22, 'ToPort': 22, 'IpRanges': [{'CidrIp': '0.0.0.0/0'}]}, {'IpProtocol': 'tcp', 'FromPort': 80, 'ToPort': 80, 'IpRanges': [{'CidrIp': '0.0.0.0/0'}]} ] ) # 4. Launch Instance instance_response = ec2_client.run_instances( ImageId=ami_id, InstanceType=instance_type, MinCount=1, MaxCount=1, SecurityGroupIds=[sg_id], SubnetId=subnet_id, TagSpecifications=[{'ResourceType': 'instance', 'Tags': [{'Key': 'Name', 'Value': name}]}] ) instance_id = instance_response['Instances'][0]['InstanceId'] return { "status": "success", "message": f"Successfully initiated creation of instance '{name}'.", "instance_id": instance_id, "ami_id": ami_id, "security_group_id": sg_id, "vpc_id": vpc_id, "subnet_id": subnet_id } except Exception as e: print(f"Error during smart instance creation: {e}") # Clean up created resources if something failed mid-way (optional, but good practice) if 'sg_id' in locals(): try: ec2_client.delete_security_group(GroupId=sg_id) print(f"Cleaned up security group '{sg_id}'.") except Exception as cleanup_e: print(f"Error during cleanup: {cleanup_e}") return {"status": "error", "message": str(e)} def _list_instances(self, region: str): # ... (code is unchanged) pass def _start_instances(self, region: str, instance_ids: list): # ... (code is unchanged) pass def _stop_instances(self, region: str, instance_ids: list): # ... (code is unchanged) pass def _terminate_instances(self, region: str, instance_ids: list): # ... (code is unchanged) pass def _list_idle_instances(self, region: str, threshold: float = 5.0, period_days: int = 7): """ Lists potentially idle EC2 instances based on low CPU utilization. """ print(f"EC2Agent: Listing potentially idle instances in region {region} (CPU < {threshold}% for {period_days} days)...") try: ec2_client = AWSConnector.get_client('ec2', region_name=region) cloudwatch_client = AWSConnector.get_client('cloudwatch', region_name=region) # Get all running instances running_instances = [] response = ec2_client.describe_instances(Filters=[{'Name': 'instance-state-name', 'Values': ['running']}]).get('Reservations', []) for reservation in response: for instance in reservation['Instances']: running_instances.append(instance['InstanceId']) idle_instances = [] end_time = datetime.utcnow() start_time = end_time - timedelta(days=period_days) for instance_id in running_instances: response = cloudwatch_client.get_metric_statistics( Namespace='AWS/EC2', MetricName='CPUUtilization', Dimensions=[ {'Name': 'InstanceId', 'Value': instance_id} ], StartTime=start_time, EndTime=end_time, Period=3600 * 24, # Daily average Statistics=['Average'] ) if response['Datapoints']: avg_cpu = sum([dp['Average'] for dp in response['Datapoints']]) / len(response['Datapoints']) if avg_cpu < threshold: idle_instances.append({'InstanceId': instance_id, 'AverageCPUUtilization': f'{avg_cpu:.2f}%'}) if not idle_instances: return {"status": "success", "message": f"No idle EC2 instances found in region {region}."} return {"status": "success", "message": f"Found {len(idle_instances)} potentially idle EC2 instances in region {region}.", "idle_instances": idle_instances} except Exception as e: print(f"Error listing idle instances: {e}") return {"status": "error", "message": str(e)}