Google Cloud Run Guide — Container-Based Serverless Deployment

# Dockerfile
# Multi-stage build for a Python web app on Cloud Run
FROM python:3.12-slim AS builder

WORKDIR /app
COPY requirements.txt .
RUN pip install --user -r requirements.txt

FROM python:3.12-slim
WORKDIR /app

# Copy only the installed packages from builder
COPY --from=builder /root/.local /root/.local
COPY app.py .

# Cloud Run provides the PORT environment variable
ENV PATH=/root/.local/bin:$PATH

CMD exec gunicorn --bind :$PORT --workers 1 --threads 8 --timeout 0 app:app

# app.py
# Simple web app for Cloud Run
import os
import json
from datetime import datetime
from flask import Flask, request, jsonify

app = Flask(__name__)

# Cloud Run injects the port via environment variable
PORT = int(os.environ.get('PORT', 8080))

@app.route('/')
def home():
    return jsonify({
        "service": "dodatech-cloud-run-demo",
        "version": "1.0.0",
        "status": "running",
        "timestamp": datetime.now().isoformat(),
    })

@app.route('/api/process', methods=['POST'])
def process():
    """Example API endpoint."""
    data = request.get_json()
    if not data:
        return jsonify({"error": "No data provided"}), 400

    result = {
        "received": data,
        "processed": True,
        "processed_at": datetime.now().isoformat(),
        "instance": os.environ.get('K_SERVICE', 'unknown'),
        "revision": os.environ.get('K_REVISION', 'unknown'),
    }
    return jsonify(result)

@app.route('/health')
def health():
    return jsonify({"status": "healthy"}), 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=PORT, debug=False)

# requirements.txt
flask>=3.0
gunicorn>=22.0

# cloud_run_simulator.py
# Simulate Cloud Run auto-scaling and concurrency
import time
import random
from datetime import datetime

class CloudRunSimulator:
    def __init__(self, max_concurrent=80, min_instances=0):
        self.max_concurrent = max_concurrent
        self.min_instances = min_instances
        self.instances = []
        self.total_requests = 0

    def handle_request(self, request_id):
        """Simulate a single request to a container instance."""
        # Find an available instance or create one
        instance = None
        for inst in self.instances:
            if inst["concurrent_requests"] < self.max_concurrent:
                instance = inst
                break

        if instance is None:
            instance = {
                "id": len(self.instances) + 1,
                "concurrent_requests": 0,
                "created": datetime.now().isoformat(),
                "cold_start": True,
            }
            self.instances.append(instance)
            time.sleep(0.2)  # Cold start penalty

        instance["concurrent_requests"] += 1
        self.total_requests += 1

        processing_time = random.uniform(0.01, 0.1)
        time.sleep(processing_time)

        instance["concurrent_requests"] -= 1
        return {"request_id": request_id, "instance": instance["id"], "cold_start": instance.get("cold_start", False)}

sim = CloudRunSimulator(max_concurrent=10)

print("=== Cloud Run Auto-Scaling Simulation ===\n")

# Simulate traffic spike
all_results = []
for i in range(30):
    result = sim.handle_request(f"req-{i:03d}")
    all_results.append(result)
    label = "[COLD]" if result["cold_start"] else "[WARM]"
    print(f"  {label} {result['request_id']} → Instance {result['instance']}")
    time.sleep(random.uniform(0, 0.05))

cold_starts = sum(1 for r in all_results if r["cold_start"])
instances_used = len(set(r["instance"] for r in all_results))
print(f"\nSummary: {len(all_results)} requests, {cold_starts} cold starts, {instances_used} instances")
print(f"Peak instances: {instances_used} (auto-scaled from demand)")

# concurrency_demo.py
# Demonstrate Cloud Run concurrency
import time
import threading

class ContainerInstance:
    def __init__(self, max_concurrency=80):
        self.max_concurrency = max_concurrency
        self.active_requests = 0

    def can_accept(self):
        return self.active_requests < self.max_concurrency

    def handle(self, request_id):
        self.active_requests += 1
        print(f"  [{request_id}] Active requests on instance: {self.active_requests}")
        time.sleep(0.05)
        self.active_requests -= 1
        return f"{request_id} done"

instance = ContainerInstance(max_concurrency=5)
threads = []

print("=== Concurrency Test (5 concurrent requests, 5 capacity) ===\n")
for i in range(5):
    t = threading.Thread(target=lambda: instance.handle(f"req-{i:03d}"))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(f"\n✓ All requests completed concurrently on one instance")

# Grant a service account invoker access
gcloud run services add-iam-policy-binding dodatech-api \
  --member="serviceAccount:api-consumer@project.iam.gserviceaccount.com" \
  --role="roles/run.invoker"

# Allow unauthenticated access (for public APIs)
gcloud run services add-iam-policy-binding dodatech-api \
  --member="allUsers" \
  --role="roles/run.invoker"

# Build and deploy in one command
gcloud builds submit --tag gcr.io/PROJECT/dodatech-api
gcloud run deploy dodatech-api \
  --image gcr.io/PROJECT/dodatech-api \
  --platform managed \
  --region us-central1 \
  --allow-unauthenticated \
  --concurrency=80 \
  --timeout=300 \
  --memory=512Mi \
  --cpu=1

# Deploy a new revision with different settings
gcloud run deploy dodatech-api \
  --image gcr.io/PROJECT/dodatech-api \
  --min-instances=2 \
  --max-instances=10 \
  --concurrency=50

# View logs
gcloud logging read "resource.type=cloud_run_revision AND resource.labels.service_name=dodatech-api" --limit=10

# cloud_run_check.py
# Check Cloud Run deployment configuration best practices
from datetime import datetime

class CloudRunConfig:
    def __init__(self, min_instances=0, max_instances=100, concurrency=80,
                 timeout=300, memory="512Mi", cpu=1):
        self.min_instances = min_instances
        self.max_instances = max_instances
        self.concurrency = concurrency
        self.timeout = timeout
        self.memory = memory
        self.cpu = cpu

    def check(self):
        issues = []
        warnings = []

        if self.min_instances == 0:
            warnings.append("min_instances=0: Requests will experience cold start delays")

        if self.concurrency > 100:
            warnings.append(f"concurrency={self.concurrency}: High concurrency may degrade performance for CPU-bound apps")

        if self.timeout < 60 and self.timeout > 1:
            if self.timeout < 10:
                issues.append(f"timeout={self.timeout}s: Very short timeout; requests may be cut off")

        memory_mb = int(self.memory.replace('Mi', '').replace('Gi', '000'))
        if memory_mb < 256:
            warnings.append(f"memory={self.memory}: Low memory may cause OOM errors for Python/Node apps")

        return {"issues": issues, "warnings": warnings, "pass": len(issues) == 0}

configs = [
    CloudRunConfig(min_instances=0, max_instances=100, concurrency=80, timeout=60, memory="128Mi"),
    CloudRunConfig(min_instances=2, max_instances=50, concurrency=40, timeout=300, memory="1Gi"),
    CloudRunConfig(min_instances=0, max_instances=10, concurrency=200, timeout=5, memory="256Mi"),
]

print("=== Cloud Run Config Checker ===\n")
for i, cfg in enumerate(configs):
    result = cfg.check()
    print(f"Config {i+1}: min={cfg.min_instances}, max={cfg.max_instances}, "
          f"conc={cfg.concurrency}, timeout={cfg.timeout}s, mem={cfg.memory}")
    for w in result["warnings"]:
        print(f"  ⚠ {w}")
    for e in result["issues"]:
        print(f"  ✗ {e}")
    if result["pass"] and not result["warnings"]:
        print(f"  ✓ Perfect configuration")
    print()