Serverless Computing: Complete Guide

# lambda_handler.py
# AWS Lambda function for image processing

import json
import boto3
from PIL import Image
import io

s3 = boto3.client('s3')

def lambda_handler(event, context):
    """Process uploaded image: create thumbnail and metadata."""
    print(f"Received event: {json.dumps(event)}")

    # Extract S3 bucket and key from the event
    for record in event['Records']:
        bucket = record['s3']['bucket']['name']
        key = record['s3']['object']['key']

        print(f"Processing s3://{bucket}/{key}")

        # Download image from S3
        response = s3.get_object(Bucket=bucket, Key=key)
        image_data = response['Body'].read()

        # Create thumbnail
        img = Image.open(io.BytesIO(image_data))
        img.thumbnail((200, 200))

        # Save thumbnail to S3
        thumbnail_key = f"thumbnails/{key.split('/')[-1]}"
        buffer = io.BytesIO()
        img.save(buffer, 'JPEG')
        buffer.seek(0)

        s3.put_object(
            Bucket=bucket,
            Key=thumbnail_key,
            Body=buffer,
            ContentType='image/jpeg'
        )

        print(f"Created thumbnail: s3://{bucket}/{thumbnail_key}")

    return {
        'statusCode': 200,
        'body': json.dumps(f'Processed {len(event["Records"])} images')
    }

# azure_function.py
# Azure Function that processes queue messages

import azure.functions as func
import logging
import json
from azure.storage.blob import BlobServiceClient
import os

def main(msg: func.QueueMessage) -> None:
    """Process a queue message from Azure Queue Storage."""
    message_body = msg.get_body().decode('utf-8')
    logging.info(f'Processing queue message: {message_body}')

    data = json.loads(message_body)
    action = data.get('action')

    if action == 'generate_report':
        generate_report(data)
    elif action == 'send_notification':
        send_notification(data)
    else:
        logging.warning(f'Unknown action: {action}')

    logging.info(f'Successfully processed: {action}')

def generate_report(data):
    """Generate and store a report."""
    report_id = data.get('report_id')
    logging.info(f'Generating report {report_id}')
    # Business logic here
    logging.info(f'Report {report_id} generated')

def send_notification(data):
    """Send a push notification."""
    user_id = data.get('user_id')
    message = data.get('message', 'You have a new update')
    logging.info(f'Sending notification to user {user_id}: {message}')
    # Send notification via Firebase or similar

# gcf_main.py
# Google Cloud Function for HTTP endpoint

import functions_framework
import json

@functions_framework.http
def hello_http(request):
    """HTTP Cloud Function example.
    Responds to HTTP requests with user info.
    """

    # Parse request
    request_json = request.get_json(silent=True)
    request_args = request.args

    name = None
    if request_json and 'name' in request_json:
        name = request_json['name']
    elif request_args and 'name' in request_args:
        name = request_args['name']

    if name:
        message = f'Hello {name}! Welcome to serverless.'
    else:
        message = 'Hello! Welcome to serverless. Send a name parameter to personalize.'

    print(f"Processed request for: {name or 'anonymous'}")

    return json.dumps({
        'message': message,
        'timestamp': __import__('datetime').datetime.now().isoformat()
    }), 200, {'Content-Type': 'application/json'}

# cold_start_test.py
# Measure cold start latency in cloud functions

import time
import requests
import json

def measure_cold_start(function_url, num_calls=5):
    """Measure cold start vs warm start latency."""
    results = []

    for i in range(num_calls):
        start = time.time()
        response = requests.post(
            function_url,
            json={"name": f"Test {i}"},
            headers={"Content-Type": "application/json"}
        )
        elapsed = time.time() - start

        # Wait between calls to see cold/warm difference
        if i < num_calls - 1:
            time.sleep(2)  # Short wait between warm calls

        results.append({
            "call": i + 1,
            "latency_ms": round(elapsed * 1000, 2),
            "status": response.status_code,
        })

    # Print results
    print(f"{'Call':<6} {'Latency (ms)':<15} {'Status':<8} {'Type':<10}")
    print("-" * 40)
    for r in results:
        call_type = "Cold" if r['call'] == 1 else "Warm" if r['latency_ms'] < 500 else "Cold"
        print(f"{r['call']:<6} {r['latency_ms]:<15} {r['status']:<8} {call_type:<10}")

    print(f"\nAverage warm latency: {sum(r['latency_ms'] for r in results[1:]) / (num_calls - 1):.2f}ms")
    print(f"Cold start penalty: {results[0]['latency_ms'] - results[1]['latency_ms']:.2f}ms")

# Example usage (requires a deployed function URL)
# measure_cold_start("https://us-central1-project.cloudfunctions.net/hello_http")

# event_driven.py
# Serverless event-driven workflow pattern

import json
import boto3
import os

# AWS services
sns = boto3.client('sns')
dynamodb = boto3.resource('dynamodb')

def user_signup_handler(event, context):
    """Handle new user signups — fan-out to multiple downstream services."""
    for record in event['Records']:
        user = json.loads(record['body'])

        print(f"Processing signup: {user['email']}")

        # Persist user to database
        table = dynamodb.Table(os.environ['USERS_TABLE'])
        table.put_item(Item={
            'userId': user['id'],
            'email': user['email'],
            'name': user['name'],
            'status': 'active',
            'createdAt': context.aws_request_id,
        })

        # Fan-out events via SNS
        sns.publish(
            TopicArn=os.environ['SIGNUP_TOPIC_ARN'],
            Message=json.dumps({
                'event': 'user.signup',
                'userId': user['id'],
                'email': user['email'],
            }),
            Subject=f"New user: {user['email']}"
        )

        print(f"User {user['id']} created. Events published to SNS.")

    return {'statusCode': 200, 'body': f'Processed {len(event["Records"])} signups'}

# pricing_calculator.py
# Compare serverless vs traditional hosting costs

def calculate_serverless_cost(invocations_per_month, avg_duration_ms, memory_mb):
    """Estimate AWS Lambda monthly cost."""
    gb_seconds = (invocations_per_month * avg_duration_ms / 1000) * (memory_mb / 1024)
    free_tier_gb_s = 400000  # Free tier: 400,000 GB-seconds
    cost_per_gb_s = 0.0000166667  # $0.0000166667 per GB-second

    billable_gb_s = max(0, gb_seconds - free_tier_gb_s)
    cost = billable_gb_s * cost_per_gb_s

    # Requests cost
    free_requests = 1000000  # 1 million requests free
    billable_requests = max(0, invocations_per_month - free_requests)
    request_cost = billable_requests * 0.0000002  # $0.20 per million

    total = cost + request_cost
    print(f"Serverless Cost Estimate:")
    print(f"  Invocations: {invocations_per_month:,}")
    print(f"  Avg duration: {avg_duration_ms}ms")
    print(f"  Memory: {memory_mb}MB")
    print(f"  Compute cost: ${cost:.4f}")
    print(f"  Request cost: ${request_cost:.4f}")
    print(f"  Total: ${total:.4f}/month")
    return total

def calculate_server_cost(instances, monthly_per_instance):
    """Estimate traditional server monthly cost."""
    total = instances * monthly_per_instance
    print(f"Server Cost Estimate:")
    print(f"  Instances: {instances}")
    print(f"  Per instance: ${monthly_per_instance}/month")
    print(f"  Total: ${total:.2f}/month")
    return total

# Compare costs
print("=" * 40)
print("Cost Comparison: Serverless vs Traditional")
print("=" * 40)

serverless = calculate_serverless_cost(
    invocations_per_month=5000000,
    avg_duration_ms=200,
    memory_mb=512,
)

server = calculate_server_cost(
    instances=2,
    monthly_per_instance=50
)

print(f"\nSavings with serverless: ${server - serverless:.2f}/month")

# url_shortener.py
# Simulated serverless URL shortener (can be deployed to Lambda)

import json
import hashlib
import time
import os

# Simulated storage (use DynamoDB in production)
url_store = {}

def generate_short_code(long_url):
    """Generate a short code from a URL hash."""
    hash_input = f"{long_url}{time.time()}".encode()
    return hashlib.md5(hash_input).hexdigest()[:8]

def create_short_url(event, context=None):
    """Handler: POST /shorten with {"url": "https://..."}."""
    body = json.loads(event.get('body', '{}'))
    long_url = body.get('url')

    if not long_url:
        return {
            'statusCode': 400,
            'body': json.dumps({'error': 'Missing url parameter'})
        }

    short_code = generate_short_code(long_url)
    url_store[short_code] = {
        'long_url': long_url,
        'created_at': time.time(),
        'clicks': 0
    }

    short_url = f"https://short.link/{short_code}"
    print(f"Created: {short_url} → {long_url}")

    return {
        'statusCode': 200,
        'body': json.dumps({
            'short_url': short_url,
            'short_code': short_code,
            'long_url': long_url
        })
    }

def redirect_url(event, context=None):
    """Handler: GET /{short_code} — redirect to original URL."""
    short_code = event.get('pathParameters', {}).get('short_code')

    if short_code not in url_store:
        return {
            'statusCode': 404,
            'body': json.dumps({'error': 'URL not found'})
        }

    url_store[short_code]['clicks'] += 1
    print(f"Redirect {short_code} ({url_store[short_code]['clicks']} clicks)")

    return {
        'statusCode': 301,
        'headers': {'Location': url_store[short_code]['long_url']},
        'body': ''
    }

# Test
test_event = {'body': json.dumps({'url': 'https://example.com/very/long/url'})}
result = create_short_url(test_event)
print(json.dumps(result, indent=2))