Postmortem: Over-Provisioned GPU Instances Cost Us $1M in 2026 — We Fixed It with NVIDIA L4 and AWS Inferentia 3

import boto3
import pandas as pd
from datetime import datetime, timedelta
import json
import logging
from typing import Dict, List, Optional

# Configure logging for audit trails
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.FileHandler("gpu_cost_audit.log"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

class GPUCostMonitor:
    def __init__(self, region: str = "us-east-1"):
        try:
            self.ec2 = boto3.client("ec2", region_name=region)
            self.cloudwatch = boto3.client("cloudwatch", region_name=region)
            self.ce = boto3.client("ce", region_name=region)  # Cost Explorer
            self.region = region
            logger.info(f"Initialized GPU cost monitor for region {region}")
        except Exception as e:
            logger.error(f"Failed to initialize AWS clients: {str(e)}")
            raise RuntimeError(f"AWS client initialization failed: {str(e)}")

    def get_gpu_instances(self) -> List[Dict]:
        """Fetch all running instances with GPU accelerators"""
        try:
            response = self.ec2.describe_instances(
                Filters=[
                    {"Name": "instance-state-name", "Values": ["running"]},
                    {"Name": "instance-type", "Values": ["p4d.*", "p5.*", "g5.*", "g6.*"]}  # GPU instance families
                ]
            )
            instances = []
            for reservation in response.get("Reservations", []):
                for instance in reservation.get("Instances", []):
                    instances.append({
                        "instance_id": instance["InstanceId"],
                        "instance_type": instance["InstanceType"],
                        "launch_time": instance["LaunchTime"].isoformat(),
                        "tags": {tag["Key"]: tag["Value"] for tag in instance.get("Tags", [])}
                    })
            logger.info(f"Found {len(instances)} running GPU instances")
            return instances
        except Exception as e:
            logger.error(f"Failed to fetch GPU instances: {str(e)}")
            return []

    def get_gpu_utilization(self, instance_id: str, hours: int = 24) -> Optional[float]:
        """Get average GPU utilization over the last N hours"""
        try:
            end_time = datetime.utcnow()
            start_time = end_time - timedelta(hours=hours)
            response = self.cloudwatch.get_metric_statistics(
                Namespace="AWS/EC2",
                MetricName="GPUUtilization",
                Dimensions=[{"Name": "InstanceId", "Value": instance_id}],
                StartTime=start_time,
                EndTime=end_time,
                Period=3600,  # 1 hour periods
                Statistics=["Average"]
            )
            datapoints = response.get("Datapoints", [])
            if not datapoints:
                logger.warning(f"No GPU utilization data for {instance_id}")
                return None
            avg_util = sum(dp["Average"] for dp in datapoints) / len(datapoints)
            return round(avg_util, 2)
        except Exception as e:
            logger.error(f"Failed to get utilization for {instance_id}: {str(e)}")
            return None

    def calculate_waste(self, instance_type: str, utilization: float, hours_running: int) -> float:
        """Calculate wasted spend for an instance based on utilization"""
        # On-demand pricing as of 2026-09 (source: AWS pricing API)
        pricing = {
            "p4d.24xlarge": 32.77,  # A100 8x
            "p5.48xlarge": 65.28,   # H100 8x
            "g5.12xlarge": 5.67,    # A10G 4x
            "g6.24xlarge": 9.87     # L4 8x
        }
        hourly_cost = pricing.get(instance_type, 0)
        if hourly_cost == 0:
            logger.warning(f"No pricing data for {instance_type}")
            return 0.0
        # Assume 80% utilization is optimal for inference
        optimal_util = 80.0
        if utilization < optimal_util:
            waste_pct = (optimal_util - utilization) / optimal_util
            wasted_spend = hourly_cost * hours_running * waste_pct
            return round(wasted_spend, 2)
        return 0.0

    def generate_audit_report(self) -> pd.DataFrame:
        """Generate full cost waste report for all GPU instances"""
        instances = self.get_gpu_instances()
        report_data = []
        for inst in instances:
            launch_time = datetime.fromisoformat(inst["launch_time"])
            hours_running = (datetime.utcnow() - launch_time).total_seconds() / 3600
            util = self.get_gpu_utilization(inst["instance_id"])
            waste = self.calculate_waste(inst["instance_type"], util or 0, int(hours_running))
            report_data.append({
                "instance_id": inst["instance_id"],
                "instance_type": inst["instance_type"],
                "hours_running": round(hours_running, 2),
                "avg_gpu_util": util or "N/A",
                "estimated_waste_usd": waste,
                "team": inst["tags"].get("Team", "Unassigned"),
                "workload": inst["tags"].get("Workload", "Unknown")
            })
        df = pd.DataFrame(report_data)
        logger.info(f"Generated report with {len(df)} entries")
        return df

if __name__ == "__main__":
    try:
        monitor = GPUCostMonitor(region="us-east-1")
        report = monitor.generate_audit_report()
        total_waste = report["estimated_waste_usd"].sum()
        logger.info(f"Total estimated GPU waste: ${total_waste:.2f}")
        report.to_csv("gpu_waste_audit.csv", index=False)
        print(f"Report saved to gpu_waste_audit.csv. Total waste: ${total_waste:.2f}")
    except Exception as e:
        logger.error(f"Audit failed: {str(e)}")
        raise

import torch
from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
import time
import json
from typing import Dict, List, Tuple
import logging
from dataclasses import dataclass

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class BenchmarkResult:
    device: str
    instance_type: str
    avg_latency_ms: float
    throughput_qps: float
    cost_per_1k_inferences: float
    error_rate: float

class LLMInferenceBenchmark:
    def __init__(self, model_id: str = "meta-llama/Llama-2-7b-chat-hf"):
        self.model_id = model_id
        self.tokenizer = None
        self.model = None
        self.device = None
        logger.info(f"Initializing benchmark for model {model_id}")

    def load_model(self, device: str = "cuda", instance_type: str = "g5.12xlarge") -> None:
        """Load model to target device with error handling"""
        try:
            self.device = device
            self.tokenizer = AutoTokenizer.from_pretrained(self.model_id)
            if device == "cuda":
                # Check for CUDA availability
                if not torch.cuda.is_available():
                    raise RuntimeError("CUDA not available for GPU benchmark")
                # Use half precision for inference efficiency
                self.model = AutoModelForCausalLM.from_pretrained(
                    self.model_id,
                    torch_dtype=torch.float16,
                    device_map="auto"
                )
            else:
                # CPU fallback (for comparison)
                self.model = AutoModelForCausalLM.from_pretrained(self.model_id)
            logger.info(f"Loaded model to {device} on {instance_type}")
        except Exception as e:
            logger.error(f"Model loading failed: {str(e)}")
            raise

    def run_inference_batch(self, prompts: List[str], max_new_tokens: int = 128) -> Tuple[List[str], float]:
        """Run batch inference and return outputs + avg latency"""
        latencies = []
        outputs = []
        for prompt in prompts:
            try:
                start = time.perf_counter()
                inputs = self.tokenizer(prompt, return_tensors="pt").to(self.device)
                with torch.no_grad():
                    outputs_ids = self.model.generate(
                        **inputs,
                        max_new_tokens=max_new_tokens,
                        do_sample=True,
                        temperature=0.7
                    )
                output_text = self.tokenizer.decode(outputs_ids[0], skip_special_tokens=True)
                end = time.perf_counter()
                latencies.append((end - start) * 1000)  # ms
                outputs.append(output_text)
            except Exception as e:
                logger.error(f"Inference failed for prompt: {str(e)}")
                outputs.append("")
                latencies.append(0.0)
        avg_latency = sum(latencies) / len(latencies) if latencies else 0.0
        return outputs, avg_latency

    def calculate_throughput(self, avg_latency_ms: float, batch_size: int = 1) -> float:
        """Calculate queries per second"""
        if avg_latency_ms == 0:
            return 0.0
        latency_per_query = avg_latency_ms / 1000  # seconds
        qps = batch_size / latency_per_query
        return round(qps, 2)

    def get_instance_pricing(self, instance_type: str) -> float:
        """Hourly on-demand pricing (2026-09 rates)"""
        pricing = {
            "g5.12xlarge": 5.67,  # 4x A10G (similar to L4 performance)
            "p4d.24xlarge": 32.77,  # 8x A100
            "g6.24xlarge": 9.87  # 8x NVIDIA L4
        }
        return pricing.get(instance_type, 0.0)

    def run_benchmark(self, instance_type: str, num_prompts: int = 1000) -> BenchmarkResult:
        """Run full benchmark for a given instance type"""
        try:
            # Load test prompts (truncated for example)
            test_prompts = ["Explain quantum computing in 3 sentences."] * num_prompts
            device = "cuda" if "g" in instance_type or "p" in instance_type else "cpu"
            self.load_model(device, instance_type)
            # Warmup
            self.run_inference_batch(test_prompts[:10])
            # Full benchmark
            outputs, avg_latency = self.run_inference_batch(test_prompts)
            throughput = self.calculate_throughput(avg_latency)
            hourly_cost = self.get_instance_pricing(instance_type)
            # Assume 1 hour of continuous inference
            inferences_per_hour = throughput * 3600
            cost_per_1k = (hourly_cost / inferences_per_hour) * 1000 if inferences_per_hour > 0 else 0.0
            error_rate = sum(1 for o in outputs if o == "") / len(outputs) if outputs else 0.0
            return BenchmarkResult(
                device=device,
                instance_type=instance_type,
                avg_latency_ms=avg_latency,
                throughput_qps=throughput,
                cost_per_1k_inferences=round(cost_per_1k, 4),
                error_rate=round(error_rate, 4)
            )
        except Exception as e:
            logger.error(f"Benchmark failed for {instance_type}: {str(e)}")
            raise

if __name__ == "__main__":
    try:
        benchmark = LLMInferenceBenchmark(model_id="meta-llama/Llama-2-7b-chat-hf")
        # Run benchmarks for L4 (g6.24xlarge) and A100 (p4d.24xlarge)
        l4_result = benchmark.run_benchmark(instance_type="g6.24xlarge", num_prompts=500)
        a100_result = benchmark.run_benchmark(instance_type="p4d.24xlarge", num_prompts=500)
        # Print comparison
        print(json.dumps({
            "L4_g6.24xlarge": l4_result.__dict__,
            "A100_p4d.24xlarge": a100_result.__dict__
        }, indent=2))
    except Exception as e:
        logger.error(f"Benchmark suite failed: {str(e)}")
        raise

import torch
import neuronxcc
from transformers import AutoModelForCausalLM, AutoTokenizer
import boto3
import json
import logging
from pathlib import Path
from typing import Optional

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Inferentia3Deployer:
    def __init__(self, model_id: str = "meta-llama/Llama-2-7b-chat-hf", region: str = "us-east-1"):
        self.model_id = model_id
        self.region = region
        self.neuron_version = "2.18.0"  # Neuron SDK version for Inferentia 3
        self.s3 = boto3.client("s3", region_name=region)
        logger.info(f"Initialized Inferentia 3 deployer for {model_id}, Neuron v{self.neuron_version}")

    def compile_model(self, output_dir: str = "./neuron_compiled") -> Optional[str]:
        """Compile model for Inferentia 3 using Neuron SDK"""
        try:
            Path(output_dir).mkdir(parents=True, exist_ok=True)
            logger.info(f"Compiling {self.model_id} for Inferentia 3...")
            # Load model in float32 for compilation
            model = AutoModelForCausalLM.from_pretrained(self.model_id)
            tokenizer = AutoTokenizer.from_pretrained(self.model_id)
            # Save tokenizer for deployment
            tokenizer.save_pretrained(output_dir)
            # Compile model with Neuron CC
            # Specify Inferentia 3 as target (inferentia3)
            compiled_model_path = neuronxcc.compile(
                model,
                output_dir=output_dir,
                target="inferentia3",
                batch_size=4,  # Optimal batch size for 7B model on Inferentia 3
                precision="fp16",
                optimization_level=3
            )
            logger.info(f"Model compiled to {compiled_model_path}")
            return compiled_model_path
        except Exception as e:
            logger.error(f"Model compilation failed: {str(e)}")
            return None

    def upload_compiled_model(self, compiled_dir: str, bucket: str, prefix: str = "inferentia3-models") -> bool:
        """Upload compiled model to S3 for deployment"""
        try:
            compiled_path = Path(compiled_dir)
            if not compiled_path.exists():
                raise FileNotFoundError(f"Compiled directory {compiled_dir} not found")
            # Upload all files in compiled dir
            for file_path in compiled_path.rglob("*"):
                if file_path.is_file():
                    s3_key = f"{prefix}/{self.model_id.split('/')[-1]}/{file_path.name}"
                    self.s3.upload_file(str(file_path), bucket, s3_key)
                    logger.info(f"Uploaded {file_path} to s3://{bucket}/{s3_key}")
            return True
        except Exception as e:
            logger.error(f"S3 upload failed: {str(e)}")
            return False

    def create_neuron_endpoint(self, bucket: str, prefix: str, endpoint_name: str) -> Optional[str]:
        """Create SageMaker endpoint for Inferentia 3 deployed model"""
        try:
            sagemaker = boto3.client("sagemaker", region_name=self.region)
            # Define model URL
            model_url = f"s3://{bucket}/{prefix}/{self.model_id.split('/')[-1]}/"
            # Create model
            sagemaker.create_model(
                ModelName=endpoint_name,
                PrimaryContainer={
                    "Image": f"763104351884.dkr.ecr.{self.region}.amazonaws.com/neuron-inference:pytorch-2.1-neuronx-sdk-2.18.0",
                    "ModelDataUrl": model_url
                },
                ExecutionRoleArn="arn:aws:iam::123456789012:role/SageMakerNeuronRole"  # Replace with real role
            )
            # Create endpoint config with Inferentia 3 instance
            sagemaker.create_endpoint_config(
                EndpointConfigName=f"{endpoint_name}-config",
                ProductionVariants=[{
                    "VariantName": "inferentia3-variant",
                    "ModelName": endpoint_name,
                    "InstanceType": "inferentia3.2xlarge",  # 2 Inferentia 3 chips
                    "InitialInstanceCount": 1
                }]
            )
            # Create endpoint
            sagemaker.create_endpoint(
                EndpointName=endpoint_name,
                EndpointConfigName=f"{endpoint_name}-config"
            )
            logger.info(f"Created endpoint {endpoint_name}")
            return endpoint_name
        except Exception as e:
            logger.error(f"Endpoint creation failed: {str(e)}")
            return None

    def run_inference_test(self, endpoint_name: str, prompt: str = "Explain LLM inference optimization") -> Optional[str]:
        """Test inference on deployed Inferentia 3 endpoint"""
        try:
            sagemaker_runtime = boto3.client("sagemaker-runtime", region_name=self.region)
            # Prepare payload
            payload = json.dumps({"inputs": prompt, "max_new_tokens": 128})
            response = sagemaker_runtime.invoke_endpoint(
                EndpointName=endpoint_name,
                ContentType="application/json",
                Body=payload
            )
            result = json.loads(response["Body"].read().decode())
            output = result.get("generated_text", "")
            logger.info(f"Inference test result: {output[:100]}...")
            return output
        except Exception as e:
            logger.error(f"Inference test failed: {str(e)}")
            return None

if __name__ == "__main__":
    try:
        deployer = Inferentia3Deployer(model_id="meta-llama/Llama-2-7b-chat-hf")
        # Step 1: Compile model
        compiled_dir = deployer.compile_model(output_dir="./llama2-7b-neuron")
        if not compiled_dir:
            raise RuntimeError("Compilation failed")
        # Step 2: Upload to S3
        upload_success = deployer.upload_compiled_model(
            compiled_dir=compiled_dir,
            bucket="our-ml-models-2026",
            prefix="inferentia3"
        )
        if not upload_success:
            raise RuntimeError("S3 upload failed")
        # Step 3: Create endpoint
        endpoint_name = deployer.create_neuron_endpoint(
            bucket="our-ml-models-2026",
            prefix="inferentia3",
            endpoint_name="llama2-7b-inferentia3"
        )
        if not endpoint_name:
            raise RuntimeError("Endpoint creation failed")
        # Step 4: Test inference
        deployer.run_inference_test(endpoint_name=endpoint_name)
        print(f"Successfully deployed {deployer.model_id} to Inferentia 3 endpoint {endpoint_name}")
    except Exception as e:
        logger.error(f"Deployment failed: {str(e)}")
        raise

import boto3

# CloudWatch alarm to trigger scale-out when GPU util >70%
cloudwatch = boto3.client("cloudwatch", region_name="us-east-1")
cloudwatch.put_metric_alarm(
    AlarmName="gpu-scale-out",
    MetricName="GPUUtilization",
    Namespace="AWS/EC2",
    Statistic="Average",
    Period=300,
    Threshold=70.0,
    ComparisonOperator="GreaterThanThreshold",
    EvaluationPeriods=2,
    AlarmActions=["arn:aws:autoscaling:us-east-1:123456789012:scalingPolicy:123:asg:gpu-asg:scale-out"],
    Dimensions=[{"Name": "AutoScalingGroupName", "Value": "gpu-inference-asg"}]
)

# Calculate throughput from benchmark results
def print_throughput_comparison(l4_qps: float, a100_qps: float, l4_cost: float, a100_cost: float):
    l4_cost_per_qps = l4_cost / l4_qps
    a100_cost_per_qps = a100_cost / a100_qps
    print(f"L4 cost per QPS: ${l4_cost_per_qps:.4f}")
    print(f"A100 cost per QPS: ${a100_cost_per_qps:.4f}")
    print(f"L4 is {a100_cost_per_qps / l4_cost_per_qps:.1f}x more cost efficient")

# Neuron compilation snippet from Code Example 3
compiled_model_path = neuronxcc.compile(
    model,
    output_dir=output_dir,
    target="inferentia3",
    batch_size=4,
    precision="fp16",
    optimization_level=3
)