在OpenTelemetry Collector中解析Map类型data字段并导出至Loki

在Kubernetes环境部署OpenTelemetry Collector，已配置FluentBit转发日志到OpenTelemetry的fluentforward端点，需求是解析日志并添加标签后导出到Loki。

当前遇到的问题：日志Attributes中的data字段是Map类型，需要将该Map内的字段提取为新的Attributes字段，但尝试的OTEL Collector配置均未生效，具体配置如下：

receivers:
  fluentforward:
    endpoint: 0.0.0.0:8006
exporters:
  logging:
    verbosity: detailed
  loki:
    endpoint: "http://loki.monitoring.svc.cluster.local:3100/loki/api/v1/push"
    headers:
      "X-Scope-OrgID": otel
processors:
  batch:
  transform:
    log_statements:
      - context: log
        statements:
          - merge_maps(attributes, attributes["data"]["kubernetes"], "upsert") #Fails with an error saying unexpected token [
          - merge_maps(attributes, ParseJSON(attributes["data"]), "upsert") # Fails saying the expected value is a String but received pcommons.Map
          - set(attributes["container_name"], attributes["data"].kubernetes.container_name) # Sets foo to data instead of data.kubernetes

  attributes:
    actions:
    - action: insert
      key: loki.attribute.labels
      value: namespace_name, container_name
    - action: insert
      key: container_name
      from_attribute: data.kubernetes.container_name # Just container_name also does not work 
    - action: insert
      key: namespace_name
      from_attribute: namespace_name #Sets it to attributes["data"]
  resource:
    attributes:
    - action: insert
      key: loki.resource.labels
      value: host_name, pod_name
    - action: insert
      key: host_name
      value: guarana
    - action: insert
      key: pod_name
      value: guarana-pod-01
  cumulativetodelta:
  k8sattributes:
    auth_type: "serviceAccount"
    passthrough: false
    filter:
      node_from_env_var: KUBE_NODE_NAME
    pod_association:
      - sources:
        - from: resource_attribute
          name: k8s.pod.ip
    extract:
      metadata:
        - k8s.pod.name
        - k8s.pod.uid
        - k8s.deployment.name
        - k8s.namespace.name
        - k8s.node.name
        - k8s.pod.start_time
  resourcedetection:
    detectors: [env, aks, azure]
    timeout: 2s
    override: false
extensions:
  memory_ballast: 
  health_check:
  zpages:
    endpoint: :55679
service:
  telemetry:
    logs:
      level: "debug"
  extensions: [zpages, health_check, memory_ballast]
  pipelines:
    logs:
      receivers: [fluentforward]
      processors: [ attributes, resource, transform]
      exporters: [loki, logging/debug]

以下是OTEL Collector的部分输出日志：

Timestamp: 2023-04-06 05:02:48.73177434 +0000 UTC
SeverityText:
SeverityNumber: Unspecified(0)
Body: Empty()
Attributes:
     -> data: Map({"kubernetes":{"annotations":{"checksum/config":"fff"},"container_hash":"fff","container_image":"image_name","container_name":"ingester","docker_id":"234","host":"host1","namespace_name":"monitoring","pod_id":"654","pod_name":"some_pod"},"logtag":"F","message":"info log","stream":"stderr","time":"2023-04-06T05:02:48.730945334Z"})
     -> type: Str(logs)
     -> agg: Str(logs)
     -> fluent.tag: Str(sometag.log)
     -> loki.attribute.labels: Str(http_status_code, container_name, namespace_name)

问题根源在于transform处理器语法错误，以及pipeline中处理器执行顺序颠倒，导致data字段未被解析就执行了属性提取。

1. 修正Transform处理器配置 #

使用正确的Map访问语法（.而非[]），直接通过merge_maps合并data字段到根属性：

transform:
  log_statements:
    - context: log
      statements:
        # 将data下所有字段合并到根Attributes
        - merge_maps(attributes, attributes.data, "upsert")
        # 可选：仅合并kubernetes子字段（按需选择）
        # - merge_maps(attributes, attributes.data.kubernetes, "upsert")
        # 可选：合并完成后删除原data字段
        - delete_key(attributes, "data")

2. 调整Pipeline执行顺序 #

将transform处理器移到最前面，确保先解析data字段，再执行属性提取和资源配置：

service:
  pipelines:
    logs:
      receivers: [fluentforward]
      processors: [transform, attributes, resource, batch] # transform前置，添加batch提升导出性能
      exporters: [loki, logging]

3. 修正Attributes处理器配置 #

此时data中的字段已合并到根属性，可直接引用目标字段：

attributes:
  actions:
  - action: insert
    key: loki.attribute.labels
    value: namespace_name, container_name
  - action: insert
    key: container_name
    from_attribute: container_name
  - action: insert
    key: namespace_name
    from_attribute: namespace_name

4. 验证效果 #

修改配置后，日志Attributes会包含container_name、namespace_name等字段，Loki exporter会根据loki.attribute.labels配置将这些字段作为标签导出到Loki。

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