1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

# Calculadora básica de costos S3
def calculate_s3_costs(storage_gb, requests_per_month, data_transfer_gb):
    """
    Calcula costos estimados de S3 (región us-east-1)
    """
    
    # Precios aproximados (pueden variar)
    storage_cost_per_gb = 0.023  # Standard
    request_cost_per_1000 = 0.0004  # GET requests
    transfer_cost_per_gb = 0.09  # Primeros 10TB
    
    storage_cost = storage_gb * storage_cost_per_gb
    request_cost = (requests_per_month / 1000) * request_cost_per_1000
    transfer_cost = data_transfer_gb * transfer_cost_per_gb
    
    total_cost = storage_cost + request_cost + transfer_cost
    
    return {
        'storage': storage_cost,
        'requests': request_cost,
        'transfer': transfer_cost,
        'total': total_cost
    }

# Ejemplo de uso
monthly_costs = calculate_s3_costs(
    storage_gb=1000,
    requests_per_month=100000,
    data_transfer_gb=500
)

print(f"Costo mensual estimado: ${monthly_costs['total']:.2f}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

# Configuración de bucket S3 con múltiples clases
resource "aws_s3_bucket" "optimized_bucket" {
  bucket = "mi-bucket-optimizado"
  
  tags = {
    Environment = "production"
    CostCenter  = "engineering"
  }
}

resource "aws_s3_bucket_lifecycle_configuration" "lifecycle" {
  bucket = aws_s3_bucket.optimized_bucket.id

  rule {
    id     = "cost_optimization"
    status = "Enabled"

    # Transición a IA después de 30 días
    transition {
      days          = 30
      storage_class = "STANDARD_IA"
    }

    # Transición a Glacier después de 90 días
    transition {
      days          = 90
      storage_class = "GLACIER"
    }

    # Transición a Deep Archive después de 365 días
    transition {
      days          = 365
      storage_class = "DEEP_ARCHIVE"
    }

    # Eliminar versiones incompletas después de 7 días
    abort_incomplete_multipart_upload {
      days_after_initiation = 7
    }
  }

  # Regla para logs
  rule {
    id     = "logs_lifecycle"
    status = "Enabled"

    filter {
      prefix = "logs/"
    }

    transition {
      days          = 7
      storage_class = "STANDARD_IA"
    }

    transition {
      days          = 30
      storage_class = "GLACIER"
    }

    expiration {
      days = 2555  # 7 años
    }
  }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

{
  "Rules": [
    {
      "ID": "IntelligentTiering",
      "Status": "Enabled",
      "Filter": {
        "Prefix": "data/"
      },
      "Transitions": [
        {
          "Days": 0,
          "StorageClass": "INTELLIGENT_TIERING"
        }
      ]
    },
    {
      "ID": "LogsArchival",
      "Status": "Enabled",
      "Filter": {
        "And": {
          "Prefix": "logs/",
          "Tags": [
            {
              "Key": "LogType",
              "Value": "application"
            }
          ]
        }
      },
      "Transitions": [
        {
          "Days": 30,
          "StorageClass": "STANDARD_IA"
        },
        {
          "Days": 90,
          "StorageClass": "GLACIER"
        }
      ],
      "Expiration": {
        "Days": 2555
      }
    }
  ]
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

import boto3
import json
from datetime import datetime, timedelta

def lambda_handler(event, context):
    """
    Función Lambda para optimización automática de S3
    """
    
    s3 = boto3.client('s3')
    cloudwatch = boto3.client('cloudwatch')
    
    bucket_name = event['bucket_name']
    
    # Analizar patrones de acceso
    access_patterns = analyze_access_patterns(bucket_name)
    
    # Aplicar optimizaciones
    optimizations = apply_optimizations(bucket_name, access_patterns)
    
    # Enviar métricas a CloudWatch
    send_metrics(cloudwatch, bucket_name, optimizations)
    
    return {
        'statusCode': 200,
        'body': json.dumps({
            'bucket': bucket_name,
            'optimizations_applied': len(optimizations),
            'estimated_savings': calculate_savings(optimizations)
        })
    }

def analyze_access_patterns(bucket_name):
    """
    Analiza patrones de acceso usando S3 Analytics
    """
    s3 = boto3.client('s3')
    
    try:
        # Obtener configuración de analytics
        response = s3.get_bucket_analytics_configuration(
            Bucket=bucket_name,
            Id='AccessPatternAnalysis'
        )
        
        # Procesar datos de analytics
        return process_analytics_data(response)
        
    except s3.exceptions.NoSuchConfiguration:
        # Crear configuración de analytics si no existe
        create_analytics_configuration(bucket_name)
        return {}

def apply_optimizations(bucket_name, patterns):
    """
    Aplica optimizaciones basadas en patrones de acceso
    """
    s3 = boto3.client('s3')
    optimizations = []
    
    for prefix, pattern in patterns.items():
        if pattern['access_frequency'] == 'low':
            # Mover a IA si no se ha accedido en 30 días
            if pattern['days_since_access'] > 30:
                optimization = {
                    'prefix': prefix,
                    'action': 'move_to_ia',
                    'estimated_savings': pattern['size_gb'] * 0.01
                }
                optimizations.append(optimization)
        
        elif pattern['access_frequency'] == 'archive':
            # Mover a Glacier si no se ha accedido en 90 días
            if pattern['days_since_access'] > 90:
                optimization = {
                    'prefix': prefix,
                    'action': 'move_to_glacier',
                    'estimated_savings': pattern['size_gb'] * 0.019
                }
                optimizations.append(optimization)
    
    return optimizations

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

# Optimización de requests con caching
import boto3
from botocore.config import Config
import redis

class OptimizedS3Client:
    def __init__(self, region='us-east-1'):
        # Configurar cliente S3 con retry optimizado
        config = Config(
            region_name=region,
            retries={
                'max_attempts': 3,
                'mode': 'adaptive'
            },
            max_pool_connections=50
        )
        
        self.s3 = boto3.client('s3', config=config)
        self.cache = redis.Redis(host='localhost', port=6379, db=0)
        self.cache_ttl = 300  # 5 minutos
    
    def get_object_with_cache(self, bucket, key):
        """
        Obtiene objeto con cache para reducir requests
        """
        cache_key = f"s3:{bucket}:{key}"
        
        # Intentar obtener del cache
        cached_data = self.cache.get(cache_key)
        if cached_data:
            return json.loads(cached_data)
        
        # Si no está en cache, obtener de S3
        try:
            response = self.s3.get_object(Bucket=bucket, Key=key)
            data = response['Body'].read()
            
            # Guardar en cache
            self.cache.setex(
                cache_key, 
                self.cache_ttl, 
                json.dumps({
                    'data': data.decode('utf-8'),
                    'last_modified': response['LastModified'].isoformat(),
                    'etag': response['ETag']
                })
            )
            
            return data
            
        except Exception as e:
            print(f"Error obteniendo objeto: {e}")
            return None
    
    def batch_operations(self, operations):
        """
        Agrupa operaciones para reducir requests
        """
        results = []
        
        # Agrupar por tipo de operación
        get_operations = [op for op in operations if op['type'] == 'get']
        put_operations = [op for op in operations if op['type'] == 'put']
        
        # Procesar GET operations en paralelo
        if get_operations:
            results.extend(self._parallel_gets(get_operations))
        
        # Procesar PUT operations en batch
        if put_operations:
            results.extend(self._batch_puts(put_operations))
        
        return results

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

import boto3
from datetime import datetime, timedelta

def setup_s3_cost_monitoring(bucket_name):
    """
    Configura monitoreo de costos para S3
    """
    cloudwatch = boto3.client('cloudwatch')
    
    # Crear alarma para costos altos
    cloudwatch.put_metric_alarm(
        AlarmName=f'S3-HighCosts-{bucket_name}',
        ComparisonOperator='GreaterThanThreshold',
        EvaluationPeriods=1,
        MetricName='EstimatedCharges',
        Namespace='AWS/Billing',
        Period=86400,  # 24 horas
        Statistic='Maximum',
        Threshold=100.0,  # $100
        ActionsEnabled=True,
        AlarmActions=[
            'arn:aws:sns:us-east-1:123456789012:s3-cost-alerts'
        ],
        AlarmDescription='Alerta cuando los costos de S3 superan $100',
        Dimensions=[
            {
                'Name': 'Currency',
                'Value': 'USD'
            },
            {
                'Name': 'ServiceName',
                'Value': 'AmazonS3'
            }
        ]
    )
    
    # Crear dashboard personalizado
    dashboard_body = {
        "widgets": [
            {
                "type": "metric",
                "properties": {
                    "metrics": [
                        ["AWS/S3", "BucketSizeBytes", "BucketName", bucket_name, "StorageType", "StandardStorage"],
                        [".", "NumberOfObjects", ".", ".", ".", "AllStorageTypes"]
                    ],
                    "period": 86400,
                    "stat": "Average",
                    "region": "us-east-1",
                    "title": f"S3 Storage Metrics - {bucket_name}"
                }
            }
        ]
    }
    
    cloudwatch.put_dashboard(
        DashboardName=f'S3-Costs-{bucket_name}',
        DashboardBody=json.dumps(dashboard_body)
    )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

#!/bin/bash

# Script para análisis de costos S3
BUCKET_NAME=$1
REGION=${2:-us-east-1}

if [ -z "$BUCKET_NAME" ]; then
    echo "Uso: $0 <bucket-name> [region]"
    exit 1
fi

echo "Analizando costos para bucket: $BUCKET_NAME"
echo "=========================================="

# Obtener tamaño total del bucket
TOTAL_SIZE=$(aws s3api list-objects-v2 \
    --bucket $BUCKET_NAME \
    --query 'sum(Contents[].Size)' \
    --output text)

if [ "$TOTAL_SIZE" != "None" ]; then
    TOTAL_SIZE_GB=$((TOTAL_SIZE / 1024 / 1024 / 1024))
    echo "Tamaño total: ${TOTAL_SIZE_GB} GB"
    
    # Calcular costo estimado
    STORAGE_COST=$(echo "scale=2; $TOTAL_SIZE_GB * 0.023" | bc)
    echo "Costo estimado de almacenamiento: \$${STORAGE_COST}/mes"
else
    echo "Bucket vacío o no accesible"
fi

# Analizar distribución por clases de almacenamiento
echo -e "\nDistribución por clases de almacenamiento:"
echo "----------------------------------------"

aws s3api list-objects-v2 \
    --bucket $BUCKET_NAME \
    --query 'Contents[].{Key:Key,Size:Size,StorageClass:StorageClass}' \
    --output table

# Identificar objetos grandes
echo -e "\nObjetos más grandes (top 10):"
echo "----------------------------"

aws s3api list-objects-v2 \
    --bucket $BUCKET_NAME \
    --query 'reverse(sort_by(Contents, &Size))[0:9].{Key:Key,Size:Size}' \
    --output table

# Sugerir optimizaciones
echo -e "\nSugerencias de optimización:"
echo "---------------------------"
echo "1. Revisar objetos grandes para posible compresión"
echo "2. Implementar lifecycle policies para datos antiguos"
echo "3. Considerar Intelligent Tiering para datos con patrones de acceso variables"
echo "4. Evaluar la necesidad de versionado en objetos grandes"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

import gzip
import boto3
from io import BytesIO

def upload_compressed_object(bucket, key, data):
    """
    Sube objeto comprimido para ahorrar espacio
    """
    s3 = boto3.client('s3')
    
    # Comprimir datos
    compressed_data = BytesIO()
    with gzip.GzipFile(fileobj=compressed_data, mode='wb') as gz:
        if isinstance(data, str):
            gz.write(data.encode('utf-8'))
        else:
            gz.write(data)
    
    compressed_data.seek(0)
    
    # Subir con metadata de compresión
    s3.put_object(
        Bucket=bucket,
        Key=key,
        Body=compressed_data.getvalue(),
        ContentEncoding='gzip',
        Metadata={
            'compression': 'gzip',
            'original-size': str(len(data))
        }
    )
    
    compression_ratio = len(compressed_data.getvalue()) / len(data)
    print(f"Compresión: {compression_ratio:.2%} del tamaño original")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

def cleanup_old_versions(bucket_name, days_to_keep=30):
    """
    Elimina versiones antiguas para reducir costos
    """
    s3 = boto3.client('s3')
    cutoff_date = datetime.now() - timedelta(days=days_to_keep)
    
    paginator = s3.get_paginator('list_object_versions')
    
    for page in paginator.paginate(Bucket=bucket_name):
        if 'Versions' in page:
            for version in page['Versions']:
                if version['LastModified'].replace(tzinfo=None) < cutoff_date:
                    if not version['IsLatest']:
                        s3.delete_object(
                            Bucket=bucket_name,
                            Key=version['Key'],
                            VersionId=version['VersionId']
                        )
                        print(f"Eliminada versión antigua: {version['Key']} - {version['VersionId']}")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

def analyze_access_patterns(bucket_name, days=30):
    """
    Analiza patrones de acceso usando CloudTrail
    """
    cloudtrail = boto3.client('cloudtrail')
    
    end_time = datetime.now()
    start_time = end_time - timedelta(days=days)
    
    events = cloudtrail.lookup_events(
        LookupAttributes=[
            {
                'AttributeKey': 'ResourceName',
                'AttributeValue': bucket_name
            }
        ],
        StartTime=start_time,
        EndTime=end_time
    )
    
    access_stats = {}
    
    for event in events['Events']:
        if event['EventName'] in ['GetObject', 'PutObject']:
            key = event.get('Resources', [{}])[0].get('ResourceName', '')
            if key not in access_stats:
                access_stats[key] = {
                    'get_count': 0,
                    'put_count': 0,
                    'last_access': None
                }
            
            if event['EventName'] == 'GetObject':
                access_stats[key]['get_count'] += 1
            else:
                access_stats[key]['put_count'] += 1
            
            access_stats[key]['last_access'] = event['EventTime']
    
    return access_stats

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

import boto3
from datetime import datetime, timedelta

def get_s3_cost_breakdown(days=30):
    """
    Obtiene desglose de costos S3 usando Cost Explorer
    """
    ce = boto3.client('ce')
    
    end_date = datetime.now().strftime('%Y-%m-%d')
    start_date = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
    
    response = ce.get_cost_and_usage(
        TimePeriod={
            'Start': start_date,
            'End': end_date
        },
        Granularity='DAILY',
        Metrics=['BlendedCost'],
        GroupBy=[
            {
                'Type': 'DIMENSION',
                'Key': 'SERVICE'
            }
        ],
        Filter={
            'Dimensions': {
                'Key': 'SERVICE',
                'Values': ['Amazon Simple Storage Service']
            }
        }
    )
    
    return response['ResultsByTime']