云原生架构设计：构建可扩展的现代化应用

# 十二要素应用示例
# 1. 代码库 - 单一代码库，多环境部署
# 2. 依赖 - 显式声明依赖
# 3. 配置 - 环境变量配置
# 4. 后端服务 - 服务即资源
# 5. 构建、发布、运行 - 严格分离
# 6. 进程 - 无状态进程
# 7. 端口绑定 - 通过端口提供服务
# 8. 并发 - 通过进程模型扩展
# 9. 易处理 - 快速启动和优雅关闭
# 10. 开发环境与生产环境等价
# 11. 日志 - 事件流
# 12. 管理进程 - 一次性管理任务

# 服务发现示例
import requests
from typing import List, Dict

class ServiceRegistry:
    def __init__(self, registry_url: str):
        self.registry_url = registry_url
        self.services = {}
    
    def register_service(self, service_name: str, service_url: str):
        """注册服务"""
        payload = {
            "name": service_name,
            "url": service_url,
            "health_check": f"{service_url}/health"
        }
        response = requests.post(f"{self.registry_url}/register", json=payload)
        return response.json()
    
    def discover_service(self, service_name: str) -> List[Dict]:
        """发现服务"""
        response = requests.get(f"{self.registry_url}/services/{service_name}")
        return response.json()
    
    def health_check(self, service_name: str):
        """健康检查"""
        services = self.discover_service(service_name)
        healthy_services = []
        
        for service in services:
            try:
                health_response = requests.get(service['health_check'], timeout=5)
                if health_response.status_code == 200:
                    healthy_services.append(service)
            except:
                continue
        
        return healthy_services

# 多阶段构建示例
FROM node:18-alpine AS builder

WORKDIR /app
COPY package*.json ./
RUN npm ci --only=production

COPY . .
RUN npm run build

# 生产阶段
FROM nginx:alpine
COPY --from=builder /app/dist /usr/share/nginx/html
COPY nginx.conf /etc/nginx/nginx.conf
EXPOSE 80

# 健康检查
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
  CMD curl -f http://localhost/health || exit 1

CMD ["nginx", "-g", "daemon off;"]

# Kubernetes 部署配置
apiVersion: apps/v1
kind: Deployment
metadata:
  name: user-service
  labels:
    app: user-service
spec:
  replicas: 3
  selector:
    matchLabels:
      app: user-service
  template:
    metadata:
      labels:
        app: user-service
    spec:
      containers:
      - name: user-service
        image: user-service:latest
        ports:
        - containerPort: 3000
        env:
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: db-secret
              key: url
        resources:
          requests:
            memory: "128Mi"
            cpu: "100m"
          limits:
            memory: "256Mi"
            cpu: "200m"
        livenessProbe:
          httpGet:
            path: /health
            port: 3000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 3000
          initialDelaySeconds: 5
          periodSeconds: 5

# .gitlab-ci.yml
stages:
  - test
  - build
  - deploy

variables:
  DOCKER_DRIVER: overlay2
  DOCKER_TLS_CERTDIR: "/certs"

test:
  stage: test
  image: node:18-alpine
  script:
    - npm ci
    - npm run lint
    - npm run test:unit
    - npm run test:integration
  coverage: '/Coverage: \d+\.\d+%/'
  artifacts:
    reports:
      coverage_report:
        coverage_format: cobertura
        path: coverage/cobertura-coverage.xml

build:
  stage: build
  image: docker:latest
  services:
    - docker:dind
  script:
    - docker build -t $CI_REGISTRY_IMAGE:$CI_COMMIT_SHA .
    - docker push $CI_REGISTRY_IMAGE:$CI_COMMIT_SHA
    - docker tag $CI_REGISTRY_IMAGE:$CI_COMMIT_SHA $CI_REGISTRY_IMAGE:latest
    - docker push $CI_REGISTRY_IMAGE:latest
  only:
    - main

deploy:
  stage: deploy
  image: bitnami/kubectl:latest
  script:
    - kubectl set image deployment/user-service user-service=$CI_REGISTRY_IMAGE:$CI_COMMIT_SHA
    - kubectl rollout status deployment/user-service
  environment:
    name: production
    url: https://app.example.com
  only:
    - main

# .github/workflows/deploy.yml
name: Deploy to Production

on:
  push:
    branches: [ main ]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v3
    - name: Use Node.js
      uses: actions/setup-node@v3
      with:
        node-version: '18'
        cache: 'npm'
    - run: npm ci
    - run: npm run test
    - run: npm run build

  deploy:
    needs: test
    runs-on: ubuntu-latest
    steps:
    - uses: actions/checkout@v3
    - name: Deploy to Kubernetes
      uses: steebchen/kubectl@v2
      with:
        config: ${{ secrets.KUBE_CONFIG_DATA }}
        command: set image deployment/user-service user-service=${{ secrets.REGISTRY }}/user-service:${{ github.sha }}

# OpenTelemetry 追踪示例
from opentelemetry import trace
from opentelemetry.exporter.jaeger.thrift import JaegerExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.instrumentation.flask import FlaskInstrumentor
from flask import Flask

# 设置追踪
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)

# 配置 Jaeger 导出器
jaeger_exporter = JaegerExporter(
    agent_host_name="localhost",
    agent_port=6831,
)
span_processor = BatchSpanProcessor(jaeger_exporter)
trace.get_tracer_provider().add_span_processor(span_processor)

app = Flask(__name__)
FlaskInstrumentor().instrument_app(app)

@app.route('/api/users/<user_id>')
def get_user(user_id):
    with tracer.start_as_current_span("get_user") as span:
        span.set_attribute("user.id", user_id)
        
        # 业务逻辑
        user = fetch_user_from_database(user_id)
        
        span.set_attribute("user.found", user is not None)
        return {"user": user}

# Prometheus 指标示例
from prometheus_client import Counter, Histogram, generate_latest
from flask import Flask, Response
import time

app = Flask(__name__)

# 定义指标
REQUEST_COUNT = Counter('http_requests_total', 'Total HTTP requests', ['method', 'endpoint', 'status'])
REQUEST_LATENCY = Histogram('http_request_duration_seconds', 'HTTP request latency')

@app.route('/metrics')
def metrics():
    return Response(generate_latest(), mimetype='text/plain')

@app.route('/api/users/<user_id>')
@REQUEST_LATENCY.time()
def get_user(user_id):
    start_time = time.time()
    
    try:
        user = fetch_user_from_database(user_id)
        status = '200'
        response = {"user": user}
    except Exception as e:
        status = '500'
        response = {"error": str(e)}
    
    # 记录指标
    REQUEST_COUNT.labels(method='GET', endpoint='/api/users', status=status).inc()
    
    return response, int(status)

# Istio 安全策略
apiVersion: security.istio.io/v1beta1
kind: AuthorizationPolicy
metadata:
  name: user-service-policy
  namespace: default
spec:
  selector:
    matchLabels:
      app: user-service
  rules:
  - from:
    - source:
        principals: ["cluster.local/ns/default/sa/frontend"]
    to:
    - operation:
        methods: ["GET"]
        paths: ["/api/users/*"]
  - from:
    - source:
        namespaces: ["admin"]
    to:
    - operation:
        methods: ["*"]
        paths: ["/api/*"]

# Vault 集成示例
import hvac
import os

class SecretManager:
    def __init__(self):
        self.client = hvac.Client(
            url=os.getenv('VAULT_URL'),
            token=os.getenv('VAULT_TOKEN')
        )
    
    def get_secret(self, path: str) -> dict:
        """获取密钥"""
        try:
            response = self.client.secrets.kv.v2.read_secret_version(
                path=path
            )
            return response['data']['data']
        except Exception as e:
            print(f"获取密钥失败: {e}")
            return {}
    
    def set_secret(self, path: str, data: dict):
        """设置密钥"""
        try:
            self.client.secrets.kv.v2.create_or_update_secret(
                path=path,
                secret_data=data
            )
            return True
        except Exception as e:
            print(f"设置密钥失败: {e}")
            return False

# 使用示例
secret_manager = SecretManager()
db_credentials = secret_manager.get_secret('database/credentials')

# Redis 缓存示例
import redis
import json
from functools import wraps

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def cache_result(expire_time=300):
    """缓存装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 生成缓存键
            cache_key = f"{func.__name__}:{hash(str(args) + str(kwargs))}"
            
            # 尝试从缓存获取
            cached_result = redis_client.get(cache_key)
            if cached_result:
                return json.loads(cached_result)
            
            # 执行函数
            result = func(*args, **kwargs)
            
            # 缓存结果
            redis_client.setex(
                cache_key,
                expire_time,
                json.dumps(result)
            )
            
            return result
        return wrapper
    return decorator

@cache_result(expire_time=600)
def get_user_profile(user_id: int):
    """获取用户资料（带缓存）"""
    # 模拟数据库查询
    return {"id": user_id, "name": "John Doe", "email": "john@example.com"}

# 负载均衡器示例
import random
from typing import List, Dict

class LoadBalancer:
    def __init__(self, strategy: str = "round_robin"):
        self.strategy = strategy
        self.servers = []
        self.current_index = 0
    
    def add_server(self, server: Dict):
        """添加服务器"""
        self.servers.append(server)
    
    def remove_server(self, server_id: str):
        """移除服务器"""
        self.servers = [s for s in self.servers if s['id'] != server_id]
    
    def get_next_server(self) -> Dict:
        """获取下一个服务器"""
        if not self.servers:
            raise Exception("没有可用的服务器")
        
        if self.strategy == "round_robin":
            server = self.servers[self.current_index]
            self.current_index = (self.current_index + 1) % len(self.servers)
            return server
        
        elif self.strategy == "random":
            return random.choice(self.servers)
        
        elif self.strategy == "least_connections":
            return min(self.servers, key=lambda x: x['connections'])
        
        else:
            raise ValueError(f"不支持的负载均衡策略: {self.strategy}")

# 使用示例
lb = LoadBalancer(strategy="round_robin")
lb.add_server({"id": "server1", "url": "http://server1:3000", "connections": 0})
lb.add_server({"id": "server2", "url": "http://server2:3000", "connections": 0})

server = lb.get_next_server()
print(f"选择服务器: {server['url']}")