elasticsearch - Istio fluentd 适配器以字节形式发送 IP 地址

Question

自从 https://istio.io/docs/tasks/telemetry/logs/fluentd/ 中的“使用 Fluentd 记录”很有限，我试过配置更完整的logentry包含要发送到 fluentd 的 IP 地址(就像这里定义的 https://github.com/istio/istio/blob/84f3b8ff576125a6c3cab667bb213c65768f68a3/install/kubernetes/helm/istio/charts/mixer/templates/config.yaml#L242 )。

当我这样做时，elasticsearch 会提示无效的 UTF-8 字符，因为 IP 实际上是以字节形式发送的。 (见这里:https://github.com/istio/istio/blob/84f3b8ff576125a6c3cab667bb213c65768f68a3/mixer/pkg/lang/cel/types.go#L291)。有没有办法使用 istio 的表达式语言或在预先存在的 fluentd 插件中转换为字符串？

下面是我发送到 stdout 的消息示例，其中包括带有 stdio 适配器的混音器和带有 stdout 插件的 fluentd。如您所见，字段 destinationIP和 sourceIP在fluentd的输出中确实是乱码。

Mixer stdio adapter

{"level":"info","time":"2019-09-10T17:06:09.847408Z","instance":"accesslog.instance.istio-system","apiClaims":"","apiKey":"","clientTraceId":"","connection_security_policy":"none","destinationApp":"telemetry","destinationIp":"10.1.2.64","destinationName":"istio-telemetry-cb4486d94-c8b8z","destinationNamespace":"istio-system","destinationOwner":"kubernetes://apis/apps/v1/namespaces/istio-system/deployments/istio-telemetry","destinationPrincipal":"","destinationServiceHost":"istio-telemetry.istio-system.svc.cluster.local","destinationWorkload":"istio-telemetry","grpcMessage":"","grpcStatus":"0","httpAuthority":"mixer","latency":"1.586447ms","method":"POST","permissiveResponseCode":"none","permissiveResponsePolicyID":"none","protocol":"http","receivedBytes":1251,"referer":"","reporter":"destination","requestId":"5c796abc-b379-41e6-9814-c6ef2d33dec2","requestSize":873,"requestedServerName":"","responseCode":200,"responseFlags":"-","responseSize":5,"sentBytes":141,"sourceApp":"reviews","sourceIp":"10.1.1.61","sourceName":"reviews-v1-59fd8b965b-5zp9f","sourceNamespace":"default","sourceOwner":"kubernetes://apis/apps/v1/namespaces/default/deployments/reviews-v1","sourcePrincipal":"","sourceWorkload":"reviews-v1","url":"/istio.mixer.v1.Mixer/Report","userAgent":"","xForwardedFor":"10.1.1.61"}


Fluentd stdout

2019-09-10 17:06:09.000000000 +0000 accesslog.instance.istio-system: {"sourcePrincipal":"","destinationNamespace":"istio-system","permissiveResponseCode":"none","permissiveResponsePolicyID":"none","latency":"2.250387ms","apiClaims":"","destinationOwner":"kubernetes://apis/apps/v1/namespaces/istio-system/deployments/istio-telemetry","requestedServerName":"","connection_security_policy":"none","destinationName":"istio-telemetry-cb4486d94-c8b8z","requestId":"486b70d4-85e8-4c34-aa1b-7c4a83accbb7","userAgent":"","responseCode":200,"xForwardedFor":"10.1.2.83","reporter":"destination","grpcStatus":"0","grpcMessage":"","apiKey":"","severity":"Info","destinationServiceHost":"istio-telemetry.istio-system.svc.cluster.local","sourceOwner":"kubernetes://apis/apps/v1/namespaces/default/deployments/reviews-v2","referer":"","responseFlags":"-","sourceWorkload":"reviews-v2","protocol":"http","httpAuthority":"mixer","sourceIp":"\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000��\n\u0001\u0002S","destinationApp":"telemetry","sourceApp":"reviews","method":"POST","requestSize":1902,"destinationPrincipal":"","destinationIp":"\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000\u0000��\n\u0001\u0002@","receivedBytes":2280,"destinationWorkload":"istio-telemetry","sentBytes":141,"url":"/istio.mixer.v1.Mixer/Report","sourceNamespace":"default","sourceName":"reviews-v2-d6cfdb7d6-hckpr","responseSize":5,"clientTraceId":""}

我还运行了一个 tcpdump 捕获来检查 fluentd 和字符串 destinationIP 之后的内容。有两个字节，0xC4 和 0x04，这意味着后面是一个长度为 4 的字节数组。 https://github.com/msgpack/msgpack/blob/master/spec.md#bin-format-family

有没有其他人遇到过这个问题，或者甚至是解决方案？

Answer 1

我最终编写了一个流利的过滤器插件来纠正错误的条目。

require 'fluent/plugin/filter'

module Fluent::Plugin
  class IstioCorrector < Filter
    Fluent::Plugin.register_filter('istiocorrect', self)

    def configure(conf)
      super
    end

    def filter(tag, time, record)
      if (record["destinationIp"] != nil)
        record["destinationIp"] = record["destinationIp"].unpack('CCCC').join('.')
      end
      if (record["sourceIp"] != nil)
        record["sourceIp"] = record["sourceIp"].unpack('CCCC').join('.')
      end
      record
    end
  end
end