Argo Events Kafka触发器无法解析消息头以实现分布式追踪

摘要 #

Argo Events的Kafka事件源触发器目前无法解析所消费的Kafka消息头，而这是实现分布式追踪所必需的。我已提交功能请求，若您遇到相同问题请点赞，同时想了解是否有可行的解决办法。

背景 #

我们部署的Argo Workflows常见模式是基于Kafka事件驱动的异步分布式工作负载，例如：

• 服务"A"作为Kafka生产者向主题发送消息
• Argo Events的Kafka事件源触发器监听该主题
• 触发Argo Workflow并执行后处理...
• ...工作流结束时，服务"B"作为Kafka生产者发送完成通知

为监控整个系统的用户核心指标（耗时情况及瓶颈位置），我希望实现从服务"A"到服务"B"的分布式追踪。我们使用Datadog作为聚合器，搭配dd-trace。

常见实现模式是通过Kafka消息头手动传播追踪上下文——发送消息前将父追踪元数据注入消息头（类似HTTP头），消息消费者处理完成后基于上游父Span添加子Span。

问题 #

Argo-Events的Kafka事件源触发器不解析任何消息头，仅将消息体JSON传递给下游Workflow，供其通过eventData.Body使用。

我的Argo事件源->触发器->工作流简化视图：

# eventsource/my-kafka-eventsource.yaml
apiVersion: argoproj.io/v1alpha1
kind: EventSource
spec:
  kafka:
    my-kafka-eventsource:
      topic: <my-topic>
      version: "2.5.0"

# sensors/trigger-my-workflow.yaml
apiVersion: argoproj.io/v1alpha1
kind: Sensor
spec:
  dependencies:
    - name: my-kafka-eventsource-dep
      eventSourceName: my-kafka-eventsource
      eventName: my-kafka-eventsource
  triggers:
    - template:
        name: start-my-workflow
        k8s:
          operation: create
          source:
            resource:
              apiVersion: argoproj.io/v1alpha1
              kind: Workflow
              spec:
                entrypoint: my-sick-workflow
                arguments:
                  parameters:
                    - name: proto_message
                      value: needs to be overriden
                    # 我希望能添加这一项
                    - name: msg_headers
                      value: needs to be overriden
                templates:
                  - name: my-sick-workflow
                    dag:
                      tasks:
                        - name: my-sick-workflow
                          templateRef:
                            name: my-sick-workflow
                            template: my-sick-workflow
          parameters:
            # 消费消息的内容/体
            - src: 
                dependencyName: my-kafka-eventsource-dep
                dataKey: body  
              dest: spec.arguments.parameters.0.value
            # 我希望能实现这一点——获取msg.headers（如果存在）
            - src: 
                dependencyName: my-kafka-eventsource-dep
                dataKey: headers
              dest: spec.arguments.parameters.1.value

# templates/my-sick-workflow.yaml
apiVersion: argoproj.io/v1alpha1
kind: WorkflowTemplate
spec:
  templates:
    - name: my-sick-workflow
      container:
        image: <image>
        command: [ "python", "/main.py" ] 
        # 我希望在此添加第二个参数——msg_headers
        args: [ "{{workflow.parameters.proto_message}}", "{{workflow.parameters.msg_headers}}" ]

# 这样在我的Workflow Dag步骤源代码中，
# 我可以通过以下方式获取上游Kafka消息的头信息.... 
# body=sys.argv[1], headers=sys.argv[2]

Confluent-Kafka API支持通过对应方法访问消息头。

问题问询 #

1. 是否有人找到了解决方案，可在Kafka生产者与Argo Events之间传递追踪上下文至下游服务？
2. 我曾考虑将Argo-Workflows的传感器触发器改为接收负载的HTTP触发器，新增Kafka消费者监听当前触发Argo Workflow的消息，再转发含父追踪元数据的HTTP负载，但这与现有工作流模式相悖，若有更简单方案希望避免此方法。

内容的提问来源于stack exchange，提问作者Jae Ryu