AI+ 云原生：构建弹性可扩展的智能应用架构

# 部署NVIDIA设备插件
kubectl apply -f https://raw.githubusercontent.com/NVIDIA/k8s-device-plugin/v0.14.0/nvidia-device-plugin.yaml
# 验证GPU资源
kubectl describe node gpu-node | grep nvidia.com/gpu

resources:
  limits:
    nvidia.com/gpu: 1   # 占用整卡
  requests:
    nvidia.com/gpu: 1

# train_component.py
from kfp import dsl
from kfp.components import create_component_from_func

@create_component_from_func
def train_model(data_path: str, epochs: int, output_model: str) -> str:
    import torch
    import torch.nn as nn
    from torch.utils.data import DataLoader, TensorDataset
    # 模拟简单MLP训练（实际可替换为ResNet/Transformer）
    X = torch.randn(10000, 128)
    y = torch.randint(0, 10, (10000,))
    model = nn.Sequential(nn.Linear(128, 256), nn.ReLU(), nn.Linear(256, 10))
    optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
    loader = DataLoader(TensorDataset(X, y), batch_size=64, shuffle=True)
    for epoch in range(epochs):
        for batch_x, batch_y in loader:
            optimizer.zero_grad()
            loss = nn.CrossEntropyLoss()(model(batch_x), batch_y)
            loss.backward()
            optimizer.step()
    torch.save(model.state_dict(), f"{output_model}/model.pth")
    return f"{output_model}/model.pth"

# 流水线定义
@dsl.pipeline(name='training-pipeline')
def pipeline(data_path: str = '/data', epochs: int = 5):
    train_task = train_model(data_path=data_path, epochs=epochs, output_model='/mnt/models')
    # 后续可接模型评估、推送至模型注册表等

dsl-compile --py train_component.py --output pipeline.yaml
kubectl apply -f pipeline.yaml

apiVersion: serving.kserve.io/v1beta1
kind: InferenceService
metadata:
  name: iris-classifier
spec:
  predictor:
    minReplicas: 1
    maxReplicas: 10
    scaleMetric: concurrency          # 基于并发数伸缩
    scaleTarget: 10                   # 每个Pod目标并发数
    containers:
    - name: kserve-container
      image: kserve/pytorchserver:latest
      resources:
        limits:
          cpu: "2"
          memory: "4Gi"
          nvidia.com/gpu: 1
      env:
      - name: MODEL_NAME
        value: iris-model
      - name: STORAGE_URI
        value: s3://my-bucket/iris/model.pth  # 模型路径
    # 金丝雀配置：将10%流量导向新版本
    canaryTrafficPercent: 10

import requests
import json

# 获取Ingress地址
service_url = "http://kserve-ingress.<namespace>.example.com/v1/models/iris-classifier:predict"
payload = {
    "instances": [[5.1, 3.5, 1.4, 0.2], [6.7, 3.1, 4.7, 1.5]]  # 鸢尾花特征
}
headers = {"Host": "iris-classifier.default.example.com", "Content-Type": "application/json"}
response = requests.post(service_url, json=payload, headers=headers)
print(response.json())  # 输出类别预测

# prometheus-adapter-config.yaml
rules:
- seriesQuery: 'kserve_requests_total{namespace!="",service_name!=""}'
  resources:
    overrides:
      namespace: {resource: "namespace"}
      service_name: {resource: "service"}
  name:
    as: "rps_per_pod"
  metricsQuery: 'sum(rate(kserve_requests_total{<<.LabelMatchers>>}[1m])) by (<<.GroupBy>>)'

apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: iris-classifier-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: iris-classifier-predictor-default
  minReplicas: 2
  maxReplicas: 20
  metrics:
  - type: Pods
    pods:
      metric:
        name: rps_per_pod
      target:
        type: AverageValue
        averageValue: "50"   # 每个Pod承载50 RPS

# 在推理代码中添加OpenTelemetry
from opentelemetry import trace
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
tracer = trace.get_tracer(__name__)
with tracer.start_as_current_span("predict") as span:
    span.set_attribute("model_name", "iris-classifier")
    result = model.predict(instances)
    span.set_attribute("prediction_count", len(result))