Ejercicio: Procesamiento de Imágenes con AWS Batch y Step Functions ​

# infrastructure/containers/image_processor/process_image.py
import sys
import os
import boto3
import cv2
import numpy as np
from PIL import Image

def process_image(input_path, output_path):
    """
    Procesa una imagen aplicando transformaciones.
    """
    # Leer imagen
    image = cv2.imread(input_path)
    
    # Aplicar transformaciones
    resized = cv2.resize(image, (800, 600))
    
    # Aplicar filtro
    filtered = cv2.GaussianBlur(resized, (5,5), 0)
    
    # Guardar resultado
    cv2.imwrite(output_path, filtered)

def main():
    # Obtener parámetros
    input_bucket = sys.argv[1]
    output_bucket = sys.argv[2]
    
    s3 = boto3.client('s3')
    
    # Descargar imagen
    input_path = '/tmp/input.jpg'
    s3.download_file(input_bucket, os.environ['INPUT_KEY'], input_path)
    
    # Procesar
    output_path = '/tmp/output.jpg'
    process_image(input_path, output_path)
    
    # Subir resultado
    s3.upload_file(output_path, output_bucket, 
                  f"processed/{os.environ['INPUT_KEY']}")

if __name__ == "__main__":
    main()

# infrastructure/vpc/vpc_setup.py
import boto3

def create_vpc():
    ec2 = boto3.client('ec2')
    
    # Crear VPC
    vpc = ec2.create_vpc(
        CidrBlock='10.0.0.0/16',
        TagSpecifications=[
            {
                'ResourceType': 'vpc',
                'Tags': [
                    {
                        'Key': 'Name',
                        'Value': 'BatchVPC'
                    }
                ]
            }
        ]
    )
    
    # Crear subredes
    subnet1 = ec2.create_subnet(
        VpcId=vpc['Vpc']['VpcId'],
        CidrBlock='10.0.1.0/24',
        AvailabilityZone='us-east-1a'
    )
    
    subnet2 = ec2.create_subnet(
        VpcId=vpc['Vpc']['VpcId'],
        CidrBlock='10.0.2.0/24',
        AvailabilityZone='us-east-1b'
    )
    
    return {
        'vpc_id': vpc['Vpc']['VpcId'],
        'subnet_ids': [
            subnet1['Subnet']['SubnetId'],
            subnet2['Subnet']['SubnetId']
        ]
    }

# src/lambdas/job_submitter/handler.py
import json
import boto3

def validate_input(event, context):
    required_params = ['inputBucket', 'outputBucket', 'imageKey']
    
    # Validar parámetros
    for param in required_params:
        if param not in event:
            raise Exception(f"Missing required parameter: {param}")
            
    # Validar existencia de imagen
    s3 = boto3.client('s3')
    try:
        s3.head_object(
            Bucket=event['inputBucket'],
            Key=event['imageKey']
        )
    except:
        raise Exception("Input image does not exist")
        
    return event

# src/lambdas/job_monitor/handler.py
import boto3

def check_batch_job(event, context):
    batch = boto3.client('batch')
    
    response = batch.describe_jobs(
        jobs=[event['jobId']]
    )
    
    job = response['jobs'][0]
    
    return {
        'jobId': job['jobId'],
        'status': job['status'],
        'statusReason': job.get('statusReason', ''),
        'exitCode': job.get('container', {}).get('exitCode')
    }

# src/notifications/setup_sns.py
import boto3

def setup_notifications():
    sns = boto3.client('sns')
    
    # Crear tópico
    topic = sns.create_topic(
        Name='ImageProcessingNotifications'
    )
    
    # Crear suscripción
    sns.subscribe(
        TopicArn=topic['TopicArn'],
        Protocol='email',
        Endpoint='admin@example.com'
    )
    
    return topic['TopicArn']

# src/lambdas/notifier/handler.py
import boto3
import json

def notify_success(event, context):
    sns = boto3.client('sns')
    
    message = {
        'status': 'SUCCESS',
        'jobId': event['jobId'],
        'outputLocation': f"s3://{event['outputBucket']}/processed/{event['imageKey']}"
    }
    
    sns.publish(
        TopicArn='arn:aws:sns:region:account:ImageProcessingNotifications',
        Message=json.dumps(message),
        Subject='Image Processing Complete'
    )
    
    return event

# src/monitoring/metrics.py
import boto3

def publish_metrics(job_status, processing_time):
    cloudwatch = boto3.client('cloudwatch')
    
    # Publicar métricas
    cloudwatch.put_metric_data(
        Namespace='ImageProcessing',
        MetricData=[
            {
                'MetricName': 'JobStatus',
                'Value': 1 if job_status == 'SUCCEEDED' else 0,
                'Unit': 'Count'
            },
            {
                'MetricName': 'ProcessingTime',
                'Value': processing_time,
                'Unit': 'Seconds'
            }
        ]
    )

# scripts/test_workflow.py
import boto3
import json
import time

def test_workflow():
    # Iniciar Step Function
    sfn = boto3.client('stepfunctions')
    
    input_data = {
        "inputBucket": "input-bucket",
        "outputBucket": "output-bucket",
        "imageKey": "test/image.jpg"
    }
    
    execution = sfn.start_execution(
        stateMachineArn='arn:aws:states:region:account:stateMachine:ImageProcessing',
        input=json.dumps(input_data)
    )
    
    # Monitorear ejecución
    while True:
        status = sfn.describe_execution(
            executionArn=execution['executionArn']
        )
        
        if status['status'] in ['SUCCEEDED', 'FAILED']:
            print(f"Workflow completed with status: {status['status']}")
            break
            
        time.sleep(30)