Argo EventBus消息丢失问题：如何配置传感器实现可靠触发与容错？

针对你遇到的消息丢失、触发失败无告警的问题，可通过以下几个关键配置实现流程容错：

1. 配置传感器延迟确认EventBus消息 #

通过设置传感器的ackMode为OnSuccess，确保只有当工作流成功触发（且进入预期运行状态）后，传感器才会向NATS发送确认信号，删除对应消息。同时配置重试策略，触发失败时自动重试，期间消息会保留在EventBus中。

示例传感器配置：

apiVersion: argoproj.io/v1alpha1
kind: Sensor
metadata:
  name: sqs-trigger-sensor
spec:
  template:
    serviceAccountName: argo-events-sa
  dependencies:
    - name: nats-event-dep
      eventSourceName: nats-event-source
      eventName: sqs-message-event
  # 消息确认与重试配置
  subscription:
    ackMode: "OnSuccess"  # 仅成功触发工作流后才删除EventBus消息
    maxRetry: 3  # 触发失败后的最大重试次数
    backoff:
      duration: "5s"       # 初始重试间隔
      factor: 2            # 间隔倍增因子
      maxDuration: "30s"   # 最大重试间隔
  triggers:
    - template:
        name: main-workflow-trigger
        k8s:
          operation: create
          source:
            resource:
              apiVersion: argoproj.io/v1alpha1
              kind: Workflow
              metadata:
                generateName: main-workflow-
              spec:
                entrypoint: main
                templates:
                  - name: main
                    container:
                      image: your-main-workflow-image:latest
                      command: ["/bin/sh", "-c"]
                      args: ["echo '执行主工作流'"]

2. 触发失败时自动运行轻量化备选工作流 #

在传感器的触发器中配置onError规则，当主工作流触发失败（如K8s Pod无法创建、调度失败）时，自动启动预先定义的轻量化容错工作流，用于日志记录、告警或简单恢复操作。

示例触发器错误处理配置：

triggers:
  - template:
      name: main-workflow-trigger
      k8s:
        operation: create
        source:
          resource:
            # 主工作流配置同上
        # 触发失败时启动备选工作流
        onError:
          trigger:
            name: fallback-workflow-trigger
            k8s:
              operation: create
              source:
                resource:
                  apiVersion: argoproj.io/v1alpha1
                  kind: Workflow
                  metadata:
                    generateName: fallback-workflow-
                  spec:
                    entrypoint: fallback
                    templates:
                      - name: fallback
                        container:
                          image: your-fallback-image:latest
                          command: ["/bin/sh", "-c"]
                          args: ["echo '执行容错工作流，处理失败触发' && echo '原始消息内容: {{ .Body }}' >> /data/fallback-logs.txt"]

3. 优化SQS事件源的消息删除逻辑 #

当前流程中SQS事件源放入EventBus后立即删除消息，存在消息丢失风险。调整SQS事件源配置，延迟删除消息，等待传感器的确认信号：

apiVersion: argoproj.io/v1alpha1
kind: EventSource
metadata:
  name: sqs-event-source
spec:
  sqs:
    sqs-message-reader:
      queueURL: "https://sqs.region.amazonaws.com/1234567890/your-queue"
      visibilityTimeout: "300s"  # 拉取消息后，消息在SQS中暂时不可见的时长
      waitTimeSeconds: 20
      batchSize: 1
      deleteOnSuccess: true  # 仅当传感器成功确认消息处理后，才删除SQS中的消息
      ackWaitTimeout: "300s" # 等待传感器确认的超时时间，超时后消息会回到SQS队列重新处理

4. 添加监控与告警，避免无感知失败 #

给传感器和工作流配置日志采集，同时利用Argo Events的内置监控指标设置告警：

• 在传感器Pod模板中添加Prometheus采集注解，暴露指标
• 基于argo_events_sensor_trigger_failed_total指标配置告警规则，当触发失败次数超过阈值时触发告警

示例传感器监控配置：

spec:
  template:
    metadata:
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "9000"
    serviceAccountName: argo-events-sa

内容的提问来源于stack exchange，提问作者Kumar-Sandeep