# Configuración Kubernetes para reducir fragmentación GPU
# Usa affinity rules para consolidar trabajos multi-GPU en mismo nodo
apiVersion: v1
kind: Pod
metadata:
name: training-job-8gpus
spec:
# Requerir todas las GPUs en el mismo nodo (topologyKey: hostname)
affinity:
    podAffinity:
    requiredDuringSchedulingIgnoredDuringExecution:
        - labelSelector:
            matchExpressions:
            - key: job-name
                operator: In
                values:
                - training-job-8gpus
        topologyKey: kubernetes.io/hostname

    # Preferir nodos con más GPUs libres (reduce fragmentación futura)
    nodeAffinity:
    preferredDuringSchedulingIgnoredDuringExecution:
        - weight: 100
        preference:
            matchExpressions:
            - key: gpu.nvidia.com/free-count
                operator: Gt
                values:
                - "4"  # Preferir nodos con >4 GPUs libres

containers:
    - name: pytorch-training
    image: nvcr.io/nvidia/pytorch:24.01-py3
    resources:
        limits:
        nvidia.com/gpu: 8  # Solicitar 8 GPUs
    # Variables de entorno para training distribuido
    env:
        - name: NCCL_DEBUG
        value: "INFO"
        - name: NCCL_IB_DISABLE
        value: "0"  # Habilitar InfiniBand si está disponible 

#!/bin/bash
# Script para auditar fragmentación GPU en clúster Kubernetes

echo "=== AUDITORÍA FRAGMENTACIÓN GPU ==="
echo ""

# 1. GPUs totales por nodo
echo "1. Distribución GPUs por nodo:"
kubectl get nodes -o json | jq -r '.items[] | select(.status.capacity."nvidia.com/gpu" != null) | "\(.metadata.name): \(.status.capacity."nvidia.com/gpu") GPUs totales"'

echo ""

# 2. GPUs libres por nodo (aproximación basada en pods running)
echo "2. GPUs libres estimadas por nodo:"
for node in $(kubectl get nodes -o json | jq -r '.items[] | select(.status.capacity."nvidia.com/gpu" != null) | .metadata.name'); do
    total=$(kubectl get node $node -o json | jq -r '.status.capacity."nvidia.com/gpu"')
    used=$(kubectl get pods --all-namespaces --field-selector spec.nodeName=$node -o json | jq '[.items[].spec.containers[].resources.limits."nvidia.com/gpu" // "0" | tonumber] | add')
    free=$((total - used))
    echo "$node: $free GPUs libres de $total totales"
done

echo ""

# 3. Trabajos en cola esperando GPUs
echo "3. Trabajos en cola (pending por falta GPUs):"
kubectl get pods --all-namespaces --field-selector status.phase=Pending -o json | jq -r '.items[] | select(.status.conditions[]? | select(.reason == "Unschedulable" and (.message | contains("Insufficient nvidia.com/gpu")))) | "\(.metadata.namespace)/\(.metadata.name): requiere \(.spec.containers[0].resources.limits."nvidia.com/gpu" // "N/A") GPUs"'

echo ""
echo "=== FIN AUDITORÍA ===" 

import torch
import torch.distributed as dist
from torch.nn.parallel import DistributedDataParallel as DDP
import os
import time
from pathlib import Path

class MultiLevelCheckpointer:
    """
    Sistema de checkpointing multi-nivel para entrenamientos distribuidos.
    Implementa estrategia Gemini-style con checkpoints en memoria, local y remoto.
    """

    def __init__(self,
                model: torch.nn.Module,
                optimizer: torch.optim.Optimizer,
                local_dir: str = "/mnt/nvme/checkpoints",
                remote_dir: str = "s3://my-bucket/checkpoints",
                memory_checkpoint_freq: int = 1,      # Cada iteración
                local_checkpoint_freq: int = 100,    # Cada 100 iters (~15 min)
                remote_checkpoint_freq: int = 500):  # Cada 500 iters (~2 hrs)

        self.model = model
        self.optimizer = optimizer
        self.local_dir = Path(local_dir)
        self.remote_dir = remote_dir
        self.memory_checkpoint_freq = memory_checkpoint_freq
        self.local_checkpoint_freq = local_checkpoint_freq
        self.remote_checkpoint_freq = remote_checkpoint_freq

        # Buffer en memoria para último checkpoint
        self.memory_checkpoint = None

        # Crear directorios
        self.local_dir.mkdir(parents=True, exist_ok=True)

    def save_checkpoint(self, iteration: int, epoch: int, loss: float):
        """
        Guarda checkpoint en el nivel apropiado según frecuencia.
        """
        checkpoint_data = {
            'iteration': iteration,
            'epoch': epoch,
            'model_state_dict': self.model.state_dict(),
            'optimizer_state_dict': self.optimizer.state_dict(),
            'loss': loss,
            'timestamp': time.time()
        }

        # NIVEL 1: Checkpoint en memoria (cada iteración)
        if iteration % self.memory_checkpoint_freq == 0:
            self.memory_checkpoint = checkpoint_data.copy()
            # No logging (demasiado verbose)

        # NIVEL 2: Checkpoint local NVMe (cada 100 iters)
        if iteration % self.local_checkpoint_freq == 0:
            local_path = self.local_dir / f"checkpoint_iter_{iteration}.pt"
            torch.save(checkpoint_data, local_path)
            print(f"✓ Checkpoint local guardado: {local_path}")

        # NIVEL 3: Checkpoint remoto S3 (cada 500 iters)
        if iteration % self.remote_checkpoint_freq == 0:
            local_path = self.local_dir / f"checkpoint_iter_{iteration}.pt"
            # Guardar localmente primero
            torch.save(checkpoint_data, local_path)
            # Upload asíncrono a S3 (no bloquea entrenamiento)
            remote_path = f"{self.remote_dir}/checkpoint_iter_{iteration}.pt"
            self._async_upload_to_s3(local_path, remote_path)
            print(f"✓ Checkpoint remoto iniciado: {remote_path}")

    def _async_upload_to_s3(self, local_path: Path, remote_path: str):
        """
        Upload asíncrono a S3 sin bloquear entrenamiento.
        En producción, usar threading o subprocess.
        """
        import subprocess
        subprocess.Popen([
            "aws", "s3", "cp", str(local_path), remote_path, "--quiet"
        ])

    def recover_from_failure(self):
        """
        Recupera desde el último checkpoint disponible.
        Orden de prioridad: memoria > local > remoto
        """
        # Intentar memoria primero
        if self.memory_checkpoint is not None:
            print("✓ Recuperando desde checkpoint EN MEMORIA (pérdida: ~segundos)")
            self.model.load_state_dict(self.memory_checkpoint['model_state_dict'])
            self.optimizer.load_state_dict(self.memory_checkpoint['optimizer_state_dict'])
            return self.memory_checkpoint['iteration']

        # Intentar checkpoints locales
        local_checkpoints = sorted(self.local_dir.glob("checkpoint_iter_*.pt"))
        if local_checkpoints:
            latest_local = local_checkpoints[-1]
            print(f"✓ Recuperando desde checkpoint LOCAL: {latest_local}")
            checkpoint = torch.load(latest_local)
            self.model.load_state_dict(checkpoint['model_state_dict'])
            self.optimizer.load_state_dict(checkpoint['optimizer_state_dict'])
            return checkpoint['iteration']

        # Intentar checkpoint remoto (último recurso)
        print("⚠️ No hay checkpoints locales, descargando desde S3...")
        # Implementar lógica download S3 aquí
        return 0


# EJEMPLO DE USO EN TRAINING LOOP
def train_with_checkpointing():
    # Setup distribuido
    dist.init_process_group(backend='nccl')

    # Modelo y optimizer
    model = MyLargeModel().cuda()
    model = DDP(model)
    optimizer = torch.optim.AdamW(model.parameters(), lr=1e-4)

    # Inicializar checkpointer
    checkpointer = MultiLevelCheckpointer(
        model=model,
        optimizer=optimizer,
        local_dir="/mnt/nvme/checkpoints",
        remote_dir="s3://my-training-bucket/llm-experiment-1"
    )

    # Intentar recuperar desde fallo previo
    start_iteration = checkpointer.recover_from_failure()

    # Training loop
    for epoch in range(10):
        for iteration, (inputs, labels) in enumerate(dataloader):
            # Skip iterations ya completadas (si recovering)
            if iteration < start_iteration:
                continue

            # Forward pass
            outputs = model(inputs)
            loss = criterion(outputs, labels)

            # Backward pass
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()

            # CHECKPOINT (automático multi-nivel)
            checkpointer.save_checkpoint(
                iteration=iteration,
                epoch=epoch,
                loss=loss.item()
            )

            if iteration % 10 == 0:
                print(f"Epoch {epoch}, Iter {iteration}, Loss: {loss.item():.4f}")

#!/usr/bin/env python3
"""
Calculadora TCO energía + refrigeración para clústeres GPU.
Compara aire vs líquido, diferentes PUEs, precios energía.
"""

def calculate_power_cooling_tco(
    num_gpus: int,
    gpu_tdp_watts: int,
    hours_per_month: int = 720,
    electricity_price_per_kwh: float = 0.12,
    pue: float = 1.4,
    cooling_type: str = "air"  # "air" o "liquid"
):
    """
    Calcula coste mensual y anual de energía + refrigeración.
    
    Args:
        num_gpus: Número de GPUs
        gpu_tdp_watts: TDP por GPU (H100=700W, Blackwell=2700W)
        hours_per_month: Horas operación/mes (720 = 24/7)
        electricity_price_per_kwh: Precio electricidad en euros
        pue: Power Usage Effectiveness (1.1-2.5)
        cooling_type: "air" o "liquid"
    
    Returns:
        dict con breakdown costes
    """
    # Consumo GPUs base
    gpu_power_kw = (num_gpus * gpu_tdp_watts) / 1000
    gpu_energy_kwh = gpu_power_kw * hours_per_month
    gpu_cost_base = gpu_energy_kwh * electricity_price_per_kwh

    # Consumo infraestructura (CPUs, RAM, red ~30% GPUs)
    infra_multiplier = 0.30
    infra_cost = gpu_cost_base * infra_multiplier

    # Overhead refrigeración (PUE - 1.0)
    cooling_overhead = (pue - 1.0) * (gpu_cost_base + infra_cost)

    # Total energía + refrigeración
    total_power_cost = gpu_cost_base + infra_cost + cooling_overhead

    # Coste inicial refrigeración líquida (amortizado 5 años)
    if cooling_type == "liquid":
        num_racks = max(1, num_gpus // 8)  # Asumir 8 GPUs/rack
        liquid_cooling_capex = num_racks * 50000  # 50k€ por rack
        liquid_cooling_monthly = liquid_cooling_capex / 60  # Amortizar 5 años
        liquid_maintenance_monthly = num_racks * 650  # 650€/mes mantenimiento
    else:
        liquid_cooling_monthly = 0
        liquid_maintenance_monthly = 0

    total_monthly = total_power_cost + liquid_cooling_monthly + liquid_maintenance_monthly

    return {
        "gpu_cost_base_monthly": round(gpu_cost_base, 2),
        "infrastructure_cost_monthly": round(infra_cost, 2),
        "cooling_overhead_monthly": round(cooling_overhead, 2),
        "liquid_cooling_amortized_monthly": round(liquid_cooling_monthly, 2),
        "liquid_maintenance_monthly": round(liquid_maintenance_monthly, 2),
        "total_monthly": round(total_monthly, 2),
        "total_annual": round(total_monthly * 12, 2),
        "pue_used": pue,
        "cooling_type": cooling_type
    }


# EJEMPLO DE USO: Comparar aire vs líquido para 100× H100s
if __name__ == "__main__":
    print("=" * 60)
    print("COMPARACIÓN REFRIGERACIÓN: AIRE vs LÍQUIDO")
    print("Clúster: 100× GPUs H100 (700W TDP cada una)")
    print("=" * 60)
    print()

    # Escenario 1: Refrigeración por aire (PUE 1.4)
    aire = calculate_power_cooling_tco(
        num_gpus=100,
        gpu_tdp_watts=700,
        pue=1.4,
        electricity_price_per_kwh=0.12,
        cooling_type="air"
    )

    print("🌬️ REFRIGERACIÓN POR AIRE (PUE 1.4)")
    print(f"   Energía GPUs base:      {aire['gpu_cost_base_monthly']:>10}€/mes")
    print(f"   Infraestructura:        {aire['infrastructure_cost_monthly']:>10}€/mes")
    print(f"   Overhead refrigeración: {aire['cooling_overhead_monthly']:>10}€/mes")
    print(f"   {'─'*40}")
    print(f"   TOTAL MENSUAL:          {aire['total_monthly']:>10}€")
    print(f"   TOTAL ANUAL:            {aire['total_annual']:>10}€")
    print()

    # Escenario 2: Refrigeración líquida (PUE 1.2)
    liquido = calculate_power_cooling_tco(
        num_gpus=100,
        gpu_tdp_watts=700,
        pue=1.2,  # Mejor PUE con líquido
        electricity_price_per_kwh=0.12,
        cooling_type="liquid"
    )

    print("💧 REFRIGERACIÓN LÍQUIDA (PUE 1.2)")
    print(f"   Energía GPUs base:      {liquido['gpu_cost_base_monthly']:>10}€/mes")
    print(f"   Infraestructura:        {liquido['infrastructure_cost_monthly']:>10}€/mes")
    print(f"   Overhead refrigeración: {liquido['cooling_overhead_monthly']:>10}€/mes")
    print(f"   Capex amortizado:       {liquido['liquid_cooling_amortized_monthly']:>10}€/mes")
    print(f"   Mantenimiento:          {liquido['liquid_maintenance_monthly']:>10}€/mes")
    print(f"   {'─'*40}")
    print(f"   TOTAL MENSUAL:          {liquido['total_monthly']:>10}€")
    print(f"   TOTAL ANUAL:            {liquido['total_annual']:>10}€")
    print()

    # Comparación ROI
    savings_monthly = aire['total_monthly'] - liquido['total_monthly']
    savings_annual = savings_monthly * 12

    if savings_monthly > 0:
        print("✅ RESULTADO:")
        print(f"   Ahorro líquido vs aire: {savings_monthly:>10}€/mes")
        print(f"   Ahorro anual:           {savings_annual:>10}€/año")
        print()

        # Calcular break-even
        num_racks = max(1, 100 // 8)
        capex_total = num_racks * 50000
        break_even_months = capex_total / savings_monthly if savings_monthly > 0 else float('inf')

        print(f"   Capex inicial líquido:  {capex_total}€")
        print(f"   Break-even point:       {break_even_months:.1f} meses ({break_even_months/12:.1f} años)")
    else:
        print("❌ RESULTADO:")
        print(f"   Líquido NO es rentable para esta configuración")

    print()
    print("=" * 60)

import torch
from torch.utils.data import DataLoader, Dataset
import multiprocessing as mp

class OptimizedDataLoader:
    """
    DataLoader optimizado para eliminar cuellos de botella I/O.
    Combina prefetching, multiprocessing, y pin memory.
    """

    @staticmethod
    def create_fast_dataloader(
        dataset: Dataset,
        batch_size: int,
        num_workers: int = None
    ) -> DataLoader:
        """
        Crea DataLoader con configuración óptima para GPUs.
        
        Args:
            dataset: PyTorch Dataset
            batch_size: Tamaño de batch
            num_workers: Threads para data loading (None = auto-detect)
        
        Returns:
            DataLoader optimizado
        """
        # Auto-detect número óptimo de workers
        if num_workers is None:
            # Regla de oro: 2-4 workers por GPU
            num_gpus = torch.cuda.device_count()
            num_workers = num_gpus * 4

        # Configuración optimizada
        dataloader = DataLoader(
            dataset,
            batch_size=batch_size,

            # Workers para loading paralelo (evita I/O blocking)
            num_workers=num_workers,

            # Pin memory para transfers CPU→GPU más rápidos
            pin_memory=True,

            # Prefetch batches (GPU nunca espera)
            prefetch_factor=4,  # Cada worker prefetchea 4 batches

            # Persistent workers (evita overhead spawn process)
            persistent_workers=True,

            # Shuffle para mejor generalización
            shuffle=True
        )

        return dataloader


# EJEMPLO DE USO CON DATASET EN NVME LOCAL
import os
import shutil
from pathlib import Path

def cache_dataset_to_nvme(s3_path: str, nvme_path: str = "/mnt/nvme/dataset"):
    """
    Copia dataset de S3 a NVMe local antes de training.
    Solo hacer una vez al inicio.
    """
    nvme_dir = Path(nvme_path)
    nvme_dir.mkdir(parents=True, exist_ok=True)

    print(f"🔄 Copiando dataset de {s3_path} a NVMe local...")

    # Usar aws s3 sync (más rápido que boto3 para bulk transfers)
    import subprocess
    subprocess.run([
        "aws", "s3", "sync",
        s3_path, str(nvme_dir),
        "--quiet"
    ], check=True)

    print(f"✓ Dataset cacheado en {nvme_dir}")
    return str(nvme_dir)


# TRAINING LOOP CON DATA LOADING OPTIMIZADO
def train_with_optimized_io():
    # 1. Cachear dataset a NVMe (una vez)
    dataset_path = cache_dataset_to_nvme("s3://my-bucket/imagenet/")

    # 2. Crear dataset apuntando a NVMe local
    from torchvision.datasets import ImageFolder
    dataset = ImageFolder(dataset_path)

    # 3. Crear dataloader optimizado
    dataloader = OptimizedDataLoader.create_fast_dataloader(
        dataset,
        batch_size=256,
        num_workers=16  # 4 GPUs × 4 workers
    )

    # 4. Training loop
    model = MyModel().cuda()
    optimizer = torch.optim.AdamW(model.parameters())

    for epoch in range(100):
        for batch_idx, (images, labels) in enumerate(dataloader):
            # Transfer a GPU (rápido con pin_memory=True)
            images = images.cuda(non_blocking=True)
            labels = labels.cuda(non_blocking=True)

            # Forward + backward
            outputs = model(images)
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()

            # GPU nunca espera por datos gracias a prefetching