构建基于AWS SNS的事件驱动模型部署状态通知管道

MLOps

文章字数: 3.5k

阅读时长: 14 分

模型部署流程的混乱始于一次深夜的紧急线上回滚。一个新版本的欺诈检测模型被推送上线，但由于数据预处理逻辑的一个微小偏差，导致误判率飙升。问题在于，我们是在收到大量用户投诉后才意识到问题的严重性。部署脚本跑完后，终端上只留下一个绿色的 “SUCCESS”，但这个“成功”背后到底发生了什么，对整个系统产生了何种影响，却无人知晓。部署过程完全是一个黑盒。

最初的想法很直接：在部署脚本的关键步骤里，加入向Slack频道发送通知的逻辑。这能解决一部分问题，但很快就引入了新的麻烦。如果Slack API变更怎么办？如果我们需要同时通知邮件组和更新一个内部监控仪表盘呢？部署脚本会迅速膨胀成一个耦合了各种通知渠道的怪物，极难维护。每一次通知需求的变化都意味着要冒着风险去修改核心的部署逻辑。

这显然不是一个可持续的方案。在真实项目中，我们需要的是解耦。部署流程本身只应该负责部署，并以某种标准化的方式“宣告”其状态。至于谁关心这些状态，以及如何处理这些状态，应该由其他系统去决定。这正是事件驱动架构（EDA）的用武之地。

我们决定引入AWS Simple Notification Service (SNS) 作为这个架构的核心。选择SNS而非SQS，是因为部署状态天然适合发布/订阅模式——一个部署事件（生产者）可能会有多个关注方（消费者）：告警系统、日志服务、仪表盘更新器，甚至自动回滚触发器。选择SNS而非功能更强大的EventBridge，是因为在当前阶段，我们的需求是纯粹的事件扇出（fan-out），不涉及复杂的事件内容路由，SNS的简单性和更低的延迟完全足够。

第一步：定义事件契约，一切规范的基石

在编写任何代码之前，首要任务是定义事件的结构。这是一个极其重要的“代码规范”环节，如果前期不统一，后期消费者各自为政，整个系统将陷入混乱。我们约定所有模型部署相关的事件都遵循一个固定的JSON Schema。

{
  "eventId": "string",       // UUID, 用于幂等性处理
  "eventType": "string",     // 事件类型, e.g., DEPLOYMENT_STARTED, VALIDATION_FAILED
  "timestamp": "string",     // ISO 8601 格式的时间戳
  "source": "string",        // 事件来源, e.g., 'ci-pipeline-prod'
  "model": {
    "name": "string",        // 模型名称
    "version": "string"      // 模型版本, e.g., 'v2.1.3'
  },
  "deployment": {
    "targetEnvironment": "string", // 'staging' or 'production'
    "correlationId": "string"    // 关联一次完整部署流程的ID
  },
  "payload": "object"        // 事件特有的数据
}

例如，一个“验证成功”事件可能长这样：

{
  "eventId": "a1b2c3d4-e5f6-7890-1234-567890abcdef",
  "eventType": "VALIDATION_SUCCEEDED",
  "timestamp": "2023-10-27T10:30:00.123Z",
  "source": "ci-pipeline-prod",
  "model": {
    "name": "fraud-detection-model",
    "version": "v2.1.4"
  },
  "deployment": {
    "targetEnvironment": "production",
    "correlationId": "deploy-run-98765"
  },
  "payload": {
    "validationMetrics": {
      "accuracy": 0.987,
      "precision": 0.965,
      "recall": 0.971
    },
    "validatorHost": "validator-worker-03"
  }
}

这个契约一旦确立，就成为了生产者和消费者之间唯一的沟通语言。

第二步：构建一个生产级的事件发布器

部署脚本（无论是Jenkins、GitLab CI还是其他工具）需要一个可靠的方式来发布这些事件。我们使用Node.js和AWS SDK v3编写一个健壮的发布器模块。直接在shell脚本里用AWS CLI也可以，但封装成一个Node.js模块能提供更好的错误处理、日志记录和代码复用性。

这里的代码规范要求我们不能将AWS凭证、SNS Topic ARN等配置硬编码。它们必须通过环境变量或配置文件注入。

config.js

// config.js
// 在生产环境中，这些值应该通过环境变量或Secrets Manager注入。
// 严禁将敏感信息硬编码到代码中。

const config = {
  aws: {
    region: process.env.AWS_REGION || 'us-east-1',
    sns: {
      topicArn: process.env.MODEL_DEPLOYMENT_TOPIC_ARN,
    },
  },
  app: {
    sourceName: process.env.DEPLOYMENT_SOURCE_NAME || 'unknown-ci-pipeline',
  }
};

// 启动时进行严格校验，确保关键配置存在
if (!config.aws.sns.topicArn) {
  console.error("FATAL: MODEL_DEPLOYMENT_TOPIC_ARN environment variable is not set.");
  process.exit(1);
}

export default config;

NotificationService.js

// services/NotificationService.js

import { SNSClient, PublishCommand } from "@aws-sdk/client-sns";
import { randomUUID } from 'crypto';
import config from '../config.js';

// 初始化SNS客户端，最佳实践是复用这个客户端实例
const snsClient = new SNSClient({ region: config.aws.region });

/**
 * 封装了向模型部署SNS Topic发布事件的逻辑
 */
class NotificationService {
  /**
   * 发布一个标准化的模型部署事件
   * @param {object} params
   * @param {string} params.eventType - 事件类型, e.g., 'DEPLOYMENT_STARTED'
   * @param {string} params.modelName - 模型名称
   * @param {string} params.modelVersion - 模型版本
   * @param {string} params.targetEnvironment - 部署目标环境
   * @param {string} params.correlationId - 部署流程关联ID
   * @param {object} [params.payload={}] - 事件相关的附加数据
   * @returns {Promise<string>} - 返回发布消息的 MessageId
   */
  async publishEvent({
    eventType,
    modelName,
    modelVersion,
    targetEnvironment,
    correlationId,
    payload = {}
  }) {
    // 1. 构建严格遵循事件契约的消息体
    const eventBody = {
      eventId: randomUUID(),
      eventType,
      timestamp: new Date().toISOString(),
      source: config.app.sourceName,
      model: {
        name: modelName,
        version: modelVersion,
      },
      deployment: {
        targetEnvironment,
        correlationId,
      },
      payload,
    };

    // 2. 构造PublishCommand
    // 这里的关键是使用 MessageAttributes。
    // 它允许消费者基于这些属性进行过滤，而无需解析完整的JSON消息体。
    // 这是一个非常重要的性能和成本优化点。
    const command = new PublishCommand({
      TopicArn: config.aws.sns.topicArn,
      Message: JSON.stringify(eventBody),
      MessageAttributes: {
        eventType: {
          DataType: 'String',
          StringValue: eventType,
        },
        modelName: {
          DataType: 'String',
          StringValue: modelName,
        },
        targetEnvironment: {
          DataType: 'String',
          StringValue: targetEnvironment,
        },
      },
    });

    try {
      // 3. 发送命令并处理响应
      console.log(`[NotificationService] Publishing event: ${eventType} for model ${modelName}:${modelVersion}`);
      const response = await snsClient.send(command);
      console.log(`[NotificationService] Event published successfully. MessageId: ${response.MessageId}`);
      return response.MessageId;
    } catch (error) {
      // 4. 详尽的错误处理和日志记录
      console.error(`[NotificationService] Failed to publish event type ${eventType}. Error:`, error);
      // 在真实的CI/CD流程中，这里可能需要决定是让整个部署失败，还是仅仅记录错误后继续。
      // 这取决于业务需求。例如，一个启动事件的发布失败可能应该中止部署。
      throw new Error(`SNS_PUBLISH_FAILED: ${error.message}`);
    }
  }
}

// 导出单例，避免不必要的重复实例化
export const notificationService = new NotificationService();

在CI脚本中，就可以这样调用：

publish-deployment-status.js

import { notificationService } from './services/NotificationService.js';

// 这个脚本由CI/CD工具调用，参数通过命令行传入
const [ eventType, modelName, modelVersion, targetEnvironment, correlationId, payloadJson ] = process.argv.slice(2);

if (!eventType || !modelName || !modelVersion || !targetEnvironment || !correlationId) {
    console.error("Usage: node publish-deployment-status.js <eventType> <modelName> <modelVersion> <targetEnvironment> <correlationId> [payloadJson]");
    process.exit(1);
}

async function main() {
    try {
        let payload = {};
        if (payloadJson) {
            try {
                payload = JSON.parse(payloadJson);
            } catch (e) {
                console.error("Invalid payload JSON provided.");
                process.exit(1);
            }
        }

        await notificationService.publishEvent({
            eventType,
            modelName,
            modelVersion,
            targetEnvironment,
            correlationId,
            payload
        });
    } catch (error) {
        // 向上层CI工具抛出错误
        process.exit(1);
    }
}

main();

第三步：设计一个不会丢失消息的消费者

发布事件只是第一半。另一半，也是更复杂的一半，是消费这些事件。一个常见的错误是直接用Lambda函数订阅SNS Topic。这很简单，但如果Lambda函数执行失败，或者出现并发问题，消息可能会丢失或被重复处理得一团糟。

一个更稳健的架构模式是：SNS Topic -> SQS Queue -> Consumer Service。

graph TD
    subgraph Deployment Pipeline
        A[CI/CD Script] -- calls --> B(Node.js Publisher);
    end
    B -- publishes event --> C{AWS SNS Topic};
    subgraph Resilient Consumer Architecture
        C -- fans out to --> D[SQS Queue for Logging];
        C -- fans out to --> E[SQS Queue for Alerting];
        D --> F[Logging Service];
        E --> G[Alerting Service];
        subgraph Failure Handling
            E -- after N failures --> H(SQS Dead-Letter Queue);
        end
    end

这种模式的好处是：

持久性: SQS队列作为缓冲区。如果消费者服务（比如 Alerting Service）宕机，消息会安全地保留在队列中，待服务恢复后继续处理。
解耦: 每个消费者都有自己的队列，可以独立扩展和失败，互不影响。
重试与死信: SQS原生支持消息重试和死信队列（DLQ）。如果一条消息处理失败多次，它会被自动移入DLQ，供人工排查，而不会阻塞主队列。

现在，我们来实现这个Alerting Service。它将轮询SQS队列，获取消息，并根据事件类型执行相应操作。

sqs-consumer.js

// sqs-consumer.js
import { SQSClient, ReceiveMessageCommand, DeleteMessageCommand } from "@aws-sdk/client-sqs";
import config from './config.js'; // 假设config也包含了SQS配置

const sqsClient = new SQSClient({ region: config.aws.region });
const queueUrl = config.aws.sqs.alertingQueueUrl; // 从配置中读取队列URL

/**
 * 处理单条消息的业务逻辑
 * @param {object} message - 从SQS接收到的原始消息
 */
async function processMessage(message) {
    console.log(`[Consumer] Received message with ID: ${message.MessageId}`);

    // SQS消息体是一个字符串，其内容是SNS发送的完整JSON结构
    const snsMessage = JSON.parse(message.Body);
    
    // 真正的事件内容在SNS消息的'Message'字段中，它也是一个JSON字符串
    const event = JSON.parse(snsMessage.Message);

    // 在这里实现幂等性检查是至关重要的
    // 一个简单的方法是使用Redis或DynamoDB记录处理过的eventId
    // const isProcessed = await redis.get(`processed_event:${event.eventId}`);
    // if (isProcessed) {
    //   console.log(`[Consumer] Event ${event.eventId} already processed. Skipping.`);
    //   return;
    // }

    console.log(`[Consumer] Processing event type: ${event.eventType} for model ${event.model.name}:${event.model.version}`);

    // 核心业务逻辑
    try {
        switch (event.eventType) {
            case 'DEPLOYMENT_STARTED':
                // postToSlack(`🚀 Deployment started for ${event.model.name}:${event.model.version} to ${event.deployment.targetEnvironment}`);
                break;
            case 'VALIDATION_FAILED':
                // postToSlack(`🔥 VALIDATION FAILED for ${event.model.name}:${event.model.version}. Reason: ${event.payload.failureReason}`);
                // pageOnCallEngineer();
                break;
            case 'ROLLOUT_SUCCEEDED':
                // postToSlack(`✅ Deployment successful for ${event.model.name}:${event.model.version}.`);
                break;
            default:
                console.warn(`[Consumer] Unknown event type received: ${event.eventType}`);
        }

        // 业务逻辑成功后，标记事件为已处理
        // await redis.set(`processed_event:${event.eventId}`, '1', 'EX', 3600 * 24); // 设置一天过期

    } catch (error) {
        console.error(`[Consumer] Error processing event ${event.eventId}:`, error);
        // 抛出错误，这样外层循环就不会删除消息，消息会根据队列策略自动重试
        throw error; 
    }
}

/**
 * 主轮询循环
 */
async function pollMessages() {
    console.log('[Consumer] Starting to poll SQS queue...');
    while (true) {
        try {
            const command = new ReceiveMessageCommand({
                QueueUrl: queueUrl,
                MaxNumberOfMessages: 10, // 一次最多拉取10条
                WaitTimeSeconds: 20, // 启用长轮询，降低成本和API调用次数
                MessageAttributeNames: ['All'], // 获取所有消息属性
            });

            const { Messages } = await sqsClient.send(command);

            if (Messages && Messages.length > 0) {
                // 并行处理拉取到的消息
                const processingPromises = Messages.map(async (message) => {
                    try {
                        await processMessage(message);
                        // 只有在processMessage成功执行后才删除消息
                        const deleteCommand = new DeleteMessageCommand({
                            QueueUrl: queueUrl,
                            ReceiptHandle: message.ReceiptHandle,
                        });
                        await sqsClient.send(deleteCommand);
                        console.log(`[Consumer] Message ${message.MessageId} processed and deleted.`);
                    } catch (processingError) {
                        // 如果processMessage失败，我们什么都不做。
                        // 消息的VisibilityTimeout过期后，它会自动在队列中重新可见。
                        // SQS会根据RedrivePolicy配置决定重试次数，超过后送入DLQ。
                        console.error(`[Consumer] Failed to process message ${message.MessageId}. It will be retried.`, processingError);
                    }
                });
                await Promise.all(processingPromises);
            }
        } catch (error) {
            console.error('[Consumer] Error polling SQS:', error);
            // 在生产环境中，这里应该有一个退避策略，比如等待几秒再重试
            await new Promise(resolve => setTimeout(resolve, 5000));
        }
    }
}

pollMessages();

这段消费者代码体现了几个生产级的关键点：

长轮询 (Long Polling): WaitTimeSeconds: 20 开启长轮询，可以显著减少空轮询的次数，从而降低成本和网络流量。
原子性操作: 消息只有在被成功处理后才会被删除。如果处理过程中抛出异常，消息不会被删除，会在“可见性超时”（Visibility Timeout）后重新出现在队列中，等待下一次处理。
死信队列 (DLQ) 的隐式依赖: 代码本身不直接与DLQ交互。这个机制完全由SQS的配置（RedrivePolicy）来保证。这是基础设施即代码（IaC）应该负责的部分。
幂等性考量: 注释中提到了幂等性检查的必要性。在分布式系统中，“至少一次”（at-least-once）的消息传递语义很常见，这意味着消费者必须有能力处理重复的消息而不产生副作用。

局限性与未来迭代方向

我们现在拥有一个解耦、可靠的事件通知管道。部署脚本变得干净，只负责发布事件。下游系统可以根据需要订阅这些事件，进行各自的处理，整个系统的可维护性和扩展性得到了极大的提升。

但这个方案并非银弹。首先，SNS的消息体大小限制为256KB。如果模型验证的报告或部署日志非常大，就不能直接放在payload里。一个常见的做法是将大的数据对象上传到S3，然后在事件payload中只包含S3对象的URI。

其次，当前的消费者是一个长时间运行的进程，需要我们自己管理其部署、伸缩和健康检查。对于某些事件类型，如果处理逻辑简单且无状态，完全可以采用更轻量级的AWS Lambda。利用SNS的订阅过滤策略，我们可以让VALIDATION_FAILED事件直接触发一个专门处理失败告警的Lambda函数，而其他事件则继续走向SQS队列。这可以进一步优化成本和运维复杂度。

最后，随着业务变得更复杂，我们可能需要一个真正的流程编排引擎。例如，一个完整的模型上线流程可能包括：部署到预发环境 -> 运行自动化测试 -> 请求人工审批 -> 部署到生产环境（金丝雀） -> 监控核心指标 -> 全量发布。这种有状态、长周期的工作流，使用AWS Step Functions来编排会比纯粹的事件驱动模式更加清晰和可控。届时，我们当前的SNS管道可以作为Step Functions状态机中的一个通知步骤，无缝集成进去。

AWS SNS JavaScript 模型部署代码规范 Node.js

构建基于 Ktor WebSocket 网关与 Recoil 的高并发实时 CV 分析架构

2023-10-27 分布式架构

Kafka Ktor CV Recoil WebSocket

运用图数据库与Turbopack为内部开发者平台构建实时Monorepo依赖分析器

2023-10-27 前端工程化

NoSQL (图) Prettier 状态管理 Turbopack