🚀 CI/CD 自动部署模块运维手册

# 需求输入模板 (requirement_input.md)
## 需求基本信息
- 需求标题: [简洁描述需求]
- 需求类型: [新功能 | 功能优化 | Bug 修复 | 技术债务]
- 优先级: [P0 | P1 | P2 | P3]
- 期望上线时间: YYYY-MM-DD

## 业务背景
[描述需求的业务背景、解决的问题、目标用户]

## 功能描述
[详细描述需要实现的功能点]

## 验收标准
- [ ] 标准 1: ...
- [ ] 标准 2: ...
- [ ] 标准 3: ...

## 约束条件
[技术约束、时间约束、资源约束等]

# 技术方案设计 Prompt 模板
ROLE: 你是一位资深系统架构师
TASK: 根据 PRD 文档设计完整的技术方案

INPUT:
- PRD 文档内容
- 现有技术栈约束
- 性能指标要求

OUTPUT REQUIREMENTS:
1. 系统架构图 (使用 Mermaid 语法)
2. 技术栈详细说明及选型理由
3. 数据库 ER 图设计
4. 核心模块划分与职责说明
5. 接口设计原则
6. 安全设计方案
7. 性能优化策略
8. 部署架构说明
9. 技术风险评估与应对方案

FORMAT: Markdown 文档，包含图表和代码示例

# OpenAPI 3.0 规范示例
openapi: 3.0.0
info:
  title: 用户管理系统 API
  version: 1.0.0
  description: 提供用户管理的完整 API 接口

servers:
  - url: https://api.example.com/v1

paths:
  /users:
    get:
      summary: 获取用户列表
      parameters:
        - name: page
          in: query
          schema:
            type: integer
            default: 1
        - name: limit
          in: query
          schema:
            type: integer
            default: 20
      responses:
        '200':
          description: 成功返回用户列表
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/UserList'
    
    post:
      summary: 创建新用户
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/CreateUserRequest'
      responses:
        '201':
          description: 用户创建成功

components:
  schemas:
    User:
      type: object
      properties:
        id:
          type: string
          format: uuid
        username:
          type: string
        email:
          type: string
          format: email
        createdAt:
          type: string
          format: date-time

# 使用 Prism 启动 Mock 服务
$ prism mock openapi.yaml

# 或使用 Postman Mock Server
$ newman run collection.json --environment mock-env.json

# 后端代码生成 Prompt
ROLE: 你是一位资深后端开发工程师
TASK: 根据 API 规范和技术方案生成后端代码

CONTEXT:
- 技术栈：Python 3.12 + FastAPI + SQLAlchemy + PostgreSQL
- 编码规范：PEP 8, Google Python Style Guide
- 项目结构：遵循 DDD 分层架构

INPUT:
- OpenAPI 规范文档
- 数据库 Schema 设计
- 业务逻辑说明

REQUIREMENTS:
1. 生成完整的 CRUD 操作代码
2. 包含输入验证和数据校验
3. 实现异常处理和日志记录
4. 添加类型注解 (Type Hints)
5. 编写 Docstring 文档字符串
6. 遵循依赖
7. 生成对应的单元测试代码

OUTPUT: 完整的 Python 源文件，可直接运行

# Python Pytest 单元测试示例
import pytest
from app.services.user_service import UserService
from app.models.user import User

class TestUserService:
    @pytest.fixture
    def user_service(self, mock_db):
        return UserService(mock_db)

    def test_create_user_success(self, user_service):
        """测试成功创建用户"""
        user_data = {
            "username": "test_user",
            "email": "test@example.com",
            "password": "SecurePass123!"
        }
        
        result = user_service.create_user(user_data)
        
        assert result.success is True
        assert result.user.username == "test_user"
        assert result.user.email == "test@example.com"

    def test_create_user_duplicate_email(self, user_service, existing_user):
        """测试创建重复邮箱的用户"""
        user_data = {
            "username": "another_user",
            "email": existing_user.email,
            "password": "SecurePass123!"
        }
        
        result = user_service.create_user(user_data)
        
        assert result.success is False
        assert "email already exists" in result.error.message

    @pytest.mark.parametrize("invalid_email", [
        "invalid-email",
        "missing@domain",
        "@missing-local.com",
    ])
    def test_create_user_invalid_email(self, user_service, invalid_email):
        """参数化测试：无效邮箱格式"""
        user_data = {
            "username": "test_user",
            "email": invalid_email,
            "password": "SecurePass123!"
        }
        
        result = user_service.create_user(user_data)
        
        assert result.success is False
        assert "invalid email format" in result.error.message

# 执行单元测试并生成报告
$ pytest tests/unit/ --cov=app --cov-report=html --cov-report=xml

# 生成 Allure 测试报告
$ pytest tests/ --alluredir=./allure-results
$ allure serve ./allure-results

# CI 环境中的测试执行
$ pytest tests/unit/ -v --junitxml=test-results.xml --cov=app --cov-fail-under=80

# Docker Compose 测试环境配置
version: '3.8'
services:
  app:
    build: .
    environment:
      - DATABASE_URL=postgresql://test:test@db:5432/testdb
      - REDIS_URL=redis://redis:6379
    depends_on:
      - db
      - redis
  
  db:
    image: postgres:15
    environment:
      - POSTGRES_DB=testdb
      - POSTGRES_USER=test
      - POSTGRES_PASSWORD=test
    volumes:
      - ./tests/fixtures/init.sql:/docker-entrypoint-initdb.d/init.sql
  
  redis:
    image: redis:7-alpine
  
  test-runner:
    build: .
    command: pytest tests/integration/ -v
    environment:
      - DATABASE_URL=postgresql://test:test@db:5432/testdb
      - REDIS_URL=redis://redis:6379
    depends_on:
      - db
      - redis

# Postman Collection + Newman 执行
const pm = require('postman-collection');
const collection = new pm.Collection(require('./api-tests.json'));

# Newman CLI 执行
$ newman run api-tests.json \
  --environment test-env.json \
  --reporters cli,junit,htmlextra \
  --reporter-junit-export test-results.xml \
  --reporter-htmlextra-export test-report.html

# Python Requests 集成测试
import requests
import pytest

class TestUserAPI:
    BASE_URL = "http://localhost:8000/api/v1"
    
    def test_user_crud_flow(self):
        """测试用户 CRUD 完整流程"""
        # 1. 创建用户
        create_resp = requests.post(
            f"{self.BASE_URL}/users",
            json={"username": "test", "email": "test@example.com"}
        )
        assert create_resp.status_code == 201
        user_id = create_resp.json()["data"]["id"]
        
        # 2. 查询用户
        get_resp = requests.get(f"{self.BASE_URL}/users/{user_id}")
        assert get_resp.status_code == 200
        
        # 3. 更新用户
        update_resp = requests.put(
            f"{self.BASE_URL}/users/{user_id}",
            json={"username": "updated"}
        )
        assert update_resp.status_code == 200
        
        # 4. 删除用户
        delete_resp = requests.delete(f"{self.BASE_URL}/users/{user_id}")
        assert delete_resp.status_code == 204

// Jenkinsfile - 声明式 Pipeline
pipeline {
    agent {
        kubernetes {
            yaml '''
                apiVersion: v1
                kind: Pod
                spec:
                  containers:
                  - name: maven
                    image: maven:3.9-openjdk-17
                    command:
                    - cat
                    tty: true
                  - name: docker
                    image: docker:24-dind
                    securityContext:
                      privileged: true
                  - name: kubectl
                    image: bitnami/kubectl:latest
                    command:
                    - cat
                    tty: true
            '''
        }
    }
    
    environment {
        REGISTRY = 'harbor.company.com'
        REGISTRY_CREDENTIAL_ID = 'harbor-credentials'
        KUBECONFIG_CREDENTIAL_ID = 'kubeconfig'
        DOCKER_IMAGE = "${REGISTRY}/project/${APP_NAME}:${BUILD_NUMBER}"
    }
    
    stages {
        stage('Checkout') {
            steps {
                checkout scm
                script {
                    env.GIT_COMMIT_SHORT = sh(script: 'git rev-parse --short HEAD', returnStdout: true).trim()
                }
            }
        }
        
        stage('Code Quality Check') {
            parallel {
                stage('Lint') {
                    steps {
                        sh 'python -m flake8 app/'
                        sh 'python -m black --check app/'
                    }
                }
                stage('Security Scan') {
                    steps {
                        sh 'bandit -r app/ -f json -o bandit-report.json'
                        archiveArtifacts artifacts: 'bandit-report.json'
                    }
                }
            }
        }
        
        stage('Unit Test') {
            steps {
                sh 'pytest tests/unit/ --cov=app --cov-report=xml --junitxml=test-results.xml'
                junit 'test-results.xml'
                publishCoverage adapters: [coberturaAdapter('coverage.xml')]
            }
        }
        
        stage('Build Docker Image') {
            steps {
                script {
                    withDockerRegistry([credentialsId: REGISTRY_CREDENTIAL_ID, url: "https://${REGISTRY}"]) {
                        sh "docker build -t ${DOCKER_IMAGE} ."
                        sh "docker push ${DOCKER_IMAGE}"
                    }
                }
            }
        }
        
        stage('Deploy to Staging') {
            steps {
                withKubeConfig([credentialsId: KUBECONFIG_CREDENTIAL_ID]) {
                    sh """
                        kubectl set image deployment/${APP_NAME} \
                          ${APP_NAME}=${DOCKER_IMAGE} \
                          -n staging
                        kubectl rollout status deployment/${APP_NAME} -n staging --timeout=300s
                    """
                }
            }
        }
        
        stage('Integration Test') {
            steps {
                sh 'pytest tests/integration/ -v --junitxml=integration-results.xml'
                junit 'integration-results.xml'
            }
        }
        
        stage('UI Automation Test') {
            steps {
                sh 'playwright test --reporter=html'
                archiveArtifacts artifacts: 'playwright-report/**/*'
            }
        }
        
        stage('Manual Approval') {
            steps {
                timeout(time: 24, unit: 'HOURS') {
                    input message: '是否部署到生产环境？', 
                          ok: '部署',
                          submitter: 'admin,release-manager'
                }
            }
        }
        
        stage('Deploy to Production') {
            steps {
                withKubeConfig([credentialsId: KUBECONFIG_CREDENTIAL_ID]) {
                    sh """
                        kubectl set image deployment/${APP_NAME} \
                          ${APP_NAME}=${DOCKER_IMAGE} \
                          -n production
                        kubectl rollout status deployment/${APP_NAME} -n production --timeout=600s
                    """
                }
            }
            post {
                success {
                    slackSend channel: '#deployments',
                              color: 'good',
                              message: ":white_check_mark: ${APP_NAME} 已成功部署到生产环境 (版本：${BUILD_NUMBER})"
                }
                failure {
                    slackSend channel: '#deployments',
                              color: 'danger',
                              message: ":x: ${APP_NAME} 生产部署失败 (版本：${BUILD_NUMBER})"
                }
            }
        }
    }
    
    post {
        always {
            cleanWs()
        }
        failure {
            emailext subject: "构建失败：${JOB_NAME} [${BUILD_NUMBER}]",
                     body: "请查看：${BUILD_URL}",
                     to: 'team@company.com'
        }
    }
}

# 多阶段构建 Dockerfile
# Stage 1: 构建阶段
FROM python:3.12-slim as builder

WORKDIR /app

# 安装构建依赖
RUN apt-get update && apt-get install -y \\
    gcc \\
    libpq-dev \\
    && rm -rf /var/lib/apt/lists/*

# 复制依赖文件并安装
COPY requirements.txt .
RUN pip install --no-cache-dir --user -r requirements.txt

# 复制源代码
COPY . .

# Stage 2: 运行阶段
FROM python:3.12-slim

WORKDIR /app

# 创建非 root 用户
RUN useradd --create-home --shell /bin/bash appuser

# 从构建阶段复制已安装的包和代码
COPY --from=builder /root/.local /home/appuser/.local
COPY --from=builder /app /app

# 设置环境变量
ENV PATH=/home/appuser/.local/bin:$PATH \\
    PYTHONUNBUFFERED=1 \\
    PYTHONDONTWRITEBYTECODE=1

# 切换用户
USER appuser

# 暴露端口
EXPOSE 8000

# 健康检查
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \\
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')" || exit 1

# 启动命令
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

# Kubernetes Deployment YAML
apiVersion: apps/v1
kind: Deployment
metadata:
  name: myapp
  namespace: production
  labels:
    app: myapp
    version: v1.0.0
spec:
  replicas: 3
  selector:
    matchLabels:
      app: myapp
  strategy:
    type: RollingUpdate
    rollingUpdate:
      maxSurge: 1
      maxUnavailable: 0
  template:
    metadata:
      labels:
        app: myapp
        version: v1.0.0
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "8000"
    spec:
      serviceAccountName: myapp-sa
      securityContext:
        runAsNonRoot: true
        runAsUser: 1000
      containers:
      - name: myapp
        image: harbor.company.com/project/myapp:latest
        imagePullPolicy: Always
        ports:
        - containerPort: 8000
          protocol: TCP
        env:
        - name: DATABASE_URL
          valueFrom:
            secretKeyRef:
              name: db-secret
              key: url
        - name: REDIS_URL
          valueFrom:
            configMapKeyRef:
              name: app-config
              key: redis-url
        resources:
          requests:
            memory: "256Mi"
            cpu: "250m"
          limits:
            memory: "512Mi"
            cpu: "500m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
          timeoutSeconds: 5
          failureThreshold: 3
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5
          timeoutSeconds: 3
          failureThreshold: 3
        volumeMounts:
        - name: config-volume
          mountPath: /app/config
      volumes:
      - name: config-volume
        configMap:
          name: app-config
      affinity:
        podAntiAffinity:
          preferredDuringSchedulingIgnoredDuringExecution:
          - weight: 100
            podAffinityTerm:
              labelSelector:
                matchLabels:
                  app: myapp
              topologyKey: kubernetes.io/hostname

---
# Kubernetes Service
apiVersion: v1
kind: Service
metadata:
  name: myapp-service
  namespace: production
spec:
  selector:
    app: myapp
  ports:
  - protocol: TCP
    port: 80
    targetPort: 8000
  type: ClusterIP

---
# Kubernetes Ingress
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
  name: myapp-ingress
  namespace: production
  annotations:
    nginx.ingress.kubernetes.io/ssl-redirect: "true"
    cert-manager.io/cluster-issuer: letsencrypt-prod
spec:
  ingressClassName: nginx
  tls:
  - hosts:
    - myapp.example.com
    secretName: myapp-tls
  rules:
  - host: myapp.example.com
    http:
      paths:
      - path: /
        pathType: Prefix
        backend:
          service:
            name: myapp-service
            port:
              number: 80

# Playwright 配置 (playwright.config.ts)
import { defineConfig, devices } from '@playwright/test';

export default defineConfig({
  testDir: './tests/e2e',
  timeout: 30 * 1000,
  expect: {
    timeout: 5000
  },
  fullyParallel: true,
  forbidOnly: !!process.env.CI,
  retries: process.env.CI ? 2 : 0,
  workers: process.env.CI ? 1 : undefined,
  reporter: [['html'], ['junit', { outputFile: 'test-results.xml' }]],
  
  use: {
    baseURL: 'http://localhost:3000',
    trace: 'on-first-retry',
    screenshot: 'only-on-failure',
    video: 'retain-on-failure',
  },

  projects: [
    {
      name: 'chromium',
      use: { ...devices['Desktop Chrome'] },
    },
    {
      name: 'firefox',
      use: { ...devices['Desktop Firefox'] },
    },
    {
      name: 'webkit',
      use: { ...devices['Desktop Safari'] },
    },
    {
      name: 'Mobile Chrome',
      use: { ...devices['Pixel 5'] },
    },
  ],

  webServer: {
    command: 'npm run start',
    url: 'http://localhost:3000',
    timeout: 120 * 1000,
  },
});

# E2E 测试用例示例
import { test, expect } from '@playwright/test';

test.describe('用户登录流程', () => {
  test.beforeEach(async ({ page }) => {
    await page.goto('/login');
  });

  test('成功登录并跳转到首页', async ({ page }) => {
    // 填写登录表单
    await page.fill('input[name="email"]', 'test@example.com');
    await page.fill('input[name="password"]', 'SecurePass123!');
    
    // 提交表单
    await page.click('button[type="submit"]');
    
    // 验证跳转
    await expect(page).toHaveURL('/dashboard');
    
    // 验证欢迎信息
    await expect(page.locator('.welcome-message'))
      .toContainText('Welcome, Test User');
  });

  test(async ({ page }) => {
    await page.fill('input[name="email"]', 'test@example.com');
    await page.fill('input[name="password"]', 'WrongPassword');
    
    await page.click('button[type="submit"]');
    
    // 验证错误提示
    await expect(page.locator('.error-message'))
      .toContainText('Invalid email or password');
    
    // 验证 URL 未变化
    await expect(page).toHaveURL('/login');
  });
});

test.describe('用户管理功能', () => {
  test('创建新用户完整流程', async ({ page }) => {
    // 登录
    await page.goto('/login');
    await page.fill('input[name="email"]', 'admin@example.com');
    await page.fill('input[name="password"]', 'AdminPass123!');
    await page.click('button[type="submit"]');
    
    // 导航到用户管理页面
    await page.click('a[href="/users"]');
    
    // 点击新建用户按钮
    await page.click('button:has-text("New User")');
    
    // 填写用户信息
    await page.fill('input[name="username"]', 'new_user');
    await page.fill('input[name="email"]', 'newuser@example.com');
    await page.selectOption('select[name="role"]', 'editor');
    
    // 保存
    await page.click('button:has-text("Save")');
    
    // 验证成功提示
    await expect(page.locator('.toast-success'))
      .toContainText('User created successfully');
    
    // 验证用户出现在列表中
    await expect(page.locator('tr:has-text("new_user")'))
      .toBeVisible();
  });
});

# 视觉回归测试示例
test('首页视觉回归测试', async ({ page }) => {
  await page.goto('/');
  
  // 截取全屏截图并与基准对比
  await expect(page).toHaveScreenshot('homepage.png', {
    fullPage: true,
    maxDiffPixels: 100, // 允许的最大差异像素数
  });
});

test('登录页面视觉回归测试', async ({ page }) => {
  await page.goto('/login');
  
  // 截取特定元素截图
  await expect(page.locator('.login-form'))
    .toHaveScreenshot('login-form.png');
});

# Jenkins 人工审批配置
stage('Production Deployment Approval') {
    steps {
        timeout(time: 24, unit: 'HOURS') {
            input message: '生产环境部署审批',
                  ok: '批准部署',
                  submitter: 'release-managers',
                  submitterParameter: 'APPROVER'
        }
    }
    post {
        success {
            script {
                currentBuild.description = "Approved by ${APPROVER} for production deployment"
            }
        }
    }
}

# GitLab MR 审批配置 (.gitlab-ci.yml)
deployment_prod:
  stage: deploy
  script:
    - echo "Deploying to production..."
  environment:
    name: production
    url: https://app.example.com
  when: manual
  only:
    - main
  allow_failure: false
  variables:
    DEPLOY_ENV: production

# Prometheus 配置 (prometheus.yml)
global:
  scrape_interval: 15s
  evaluation_interval: 15s

rule_files:
  - "alerts/*.yml"

scrape_configs:
  - job_name: 'kubernetes-nodes'
    kubernetes_sd_configs:
      - role: node
    relabel_configs:
      - action: labelmap
        regex: __meta_kubernetes_node_label_(.+)

  - job_name: 'kubernetes-pods'
    kubernetes_sd_configs:
      - role: pod
    relabel_configs:
      - source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_scrape]
        action: keep
        regex: true
      - source_labels: [__meta_kubernetes_pod_annotation_prometheus_io_path]
        action: replace
        target_label: __metrics_path__
        regex: (.+)

  - job_name: 'myapp'
    static_configs:
      - targets: ['myapp-service.production.svc:8000']

alerting:
  alertmanagers:
    - static_configs:
        - targets:
          - alertmanager:9093

# AlertManager 告警规则 (alerts.yml)
groups:
- name: myapp-alerts
  rules:
  - alert: HighErrorRate
    expr: sum(rate(http_requests_total{status=~"5.."}[5m])) / sum(rate(http_requests_total[5m])) > 0.05
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "高错误率告警"
      description: "应用 {{ $labels.instance }} 的错误率超过 5% (当前值：{{ $value | humanizePercentage }})"

  - alert: HighLatency
    expr: histogram_quantile(0.95, sum(rate(http_request_duration_seconds_bucket[5m])) by (le)) > 1
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "高延迟告警"
      description: "应用 {{ $labels.instance }} 的 P95 延迟超过 1 秒 (当前值：{{ $value | humanizeDuration }})"

  - alert: PodCrashLooping
    expr: rate(kube_pod_container_status_restarts_total[15m]) * 60 * 5 > 0
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "Pod 频繁重启"
      description: "Pod {{ $labels.pod }} 在 5 分钟内重启超过 {{ $value }} 次"

  - alert: HighMemoryUsage
    expr: container_memory_usage_bytes / container_spec_memory_limit_bytes > 0.9
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "内存使用率过高"
      description: "容器 {{ $labels.container }} 内存使用率超过 90% (当前值：{{ $value | humanizePercentage }})"

# Kubectl 常用命令
# 查看 Pod 状态
$ kubectl get pods -n production -o wide

# 查看 Pod 详情
$ kubectl describe pod myapp-xxx -n production

# 查看 Pod 日志
$ kubectl logs -f myapp-xxx -n production

# 进入 Pod 调试
$ kubectl exec -it myapp-xxx -n production -- /bin/bash

# 滚动重启 Deployment
$ kubectl rollout restart deployment/myapp -n production

# 回滚到上一版本
$ kubectl rollout undo deployment/myapp -n production

# 查看资源使用情况
$ kubectl top pods -n production
$ kubectl top nodes

# 端口转发本地调试
$ kubectl port-forward svc/myapp-service 8080:80 -n production

# Docker 常用命令
# 查看容器日志
$ docker logs -f myapp-container

# 进入容器
$ docker exec -it myapp-container /bin/bash

# 查看容器资源使用
$ docker stats myapp-container

# Jenkins 常用命令
# 通过 CLI 触发构建
$ java -jar jenkins-cli.jar -s http://jenkins.example.com build my-job

# 查看构建队列
$ java -jar jenkins-cli.jar -s http://jenkins.example.com list-changes my-job