ML Model Deployment: Kubernetes and MLOps Best Practices

FROM python:3.11-slim

WORKDIR /app

# 先複製 requirements，利用 Docker layer cache
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 複製模型和程式碼
COPY model/ ./model/
COPY app.py .

# 非 root 用戶運行（安全性）
RUN useradd -m appuser
USER appuser

EXPOSE 8000
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

apiVersion: apps/v1
kind: Deployment
metadata:
  name: ml-model
  labels:
    app: ml-model
spec:
  replicas: 3
  selector:
    matchLabels:
      app: ml-model
  template:
    metadata:
      labels:
        app: ml-model
    spec:
      containers:
      - name: ml-model
        image: your-registry/ml-model:v1.0.0
        ports:
        - containerPort: 8000
        resources:
          requests:
            memory: "1Gi"
            cpu: "500m"
          limits:
            memory: "2Gi"
            cpu: "1000m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5

apiVersion: v1
kind: Service
metadata:
  name: ml-model-service
spec:
  selector:
    app: ml-model
  ports:
  - port: 80
    targetPort: 8000
  type: ClusterIP

from fastapi import FastAPI
import time

app = FastAPI()
model = None
model_loaded_at = None

@app.on_event("startup")
async def load_model():
    global model, model_loaded_at
    model = load_your_model()
    model_loaded_at = time.time()

@app.get("/health")
async def health():
    """Liveness probe - 程式還活著嗎？"""
    return {"status": "healthy"}

@app.get("/ready")
async def ready():
    """Readiness probe - 可以接收流量了嗎？"""
    if model is None:
        return {"status": "not ready"}, 503
    return {"status": "ready", "model_loaded_at": model_loaded_at}

import mlflow

mlflow.set_tracking_uri("http://mlflow-server:5000")

with mlflow.start_run():
    # 訓練
    model = train_model(X_train, y_train)

    # 記錄參數和指標
    mlflow.log_params({
        "n_estimators": 100,
        "max_depth": 10
    })
    mlflow.log_metrics({
        "accuracy": accuracy,
        "f1_score": f1
    })

    # 儲存模型
    mlflow.sklearn.log_model(model, "model")

import mlflow

model = mlflow.sklearn.load_model("models:/my-model/Production")

from prometheus_client import Counter, Histogram, generate_latest
from fastapi import Response

REQUEST_COUNT = Counter(
    'prediction_requests_total',
    'Total prediction requests'
)

PREDICTION_LATENCY = Histogram(
    'prediction_latency_seconds',
    'Prediction latency in seconds'
)

@app.post("/predict")
async def predict(data: PredictRequest):
    REQUEST_COUNT.inc()

    with PREDICTION_LATENCY.time():
        result = model.predict(data.features)

    return {"prediction": result}

@app.get("/metrics")
async def metrics():
    return Response(
        content=generate_latest(),
        media_type="text/plain"
    )

# .github/workflows/deploy.yml
name: Deploy ML Model

on:
  push:
    branches: [main]

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Build and push image
        run: |
          docker build -t $REGISTRY/ml-model:${{ github.sha }} .
          docker push $REGISTRY/ml-model:${{ github.sha }}

      - name: Deploy to staging
        run: |
          kubectl set image deployment/ml-model \
            ml-model=$REGISTRY/ml-model:${{ github.sha }} \
            --namespace=staging

      - name: Run smoke tests
        run: |
          ./scripts/smoke-test.sh staging

      - name: Deploy to production
        if: success()
        run: |
          kubectl set image deployment/ml-model \
            ml-model=$REGISTRY/ml-model:${{ github.sha }} \
            --namespace=production