How to capture Date/timestamp with systemd plugin and Kafka as Output

[nareshku]

Hello, I am trying to get fluent-bit systemd plugin with Kafka as output working.

I am having issues retrieving the timestamp. The problem I see is the date field missing. I do see a discrepancy in the way date is retrieved when using stdout msgpack vs json. Below is a sample stdout as output with msgpack and json formats.

Having the output as stdout with type as msgpack, the log looks something like below.

[1631] kube-systemd-kubelet: [1674851353.744348000, {"transport"=>"stdout", "stream_id"=>"6c715dd867f24d72a7799902feda221f", "priority"=>"6", "syslog_facility"=>"3", "syslog_identifier"=>"kubelet", "pid"=>"21573", "uid"=>"0", "gid"=>"0", "comm"=>"kubelet", "exe"=>"/bin/kubelet", "cmdline"=>"/bin/kubelet --config=kubelet_config.yaml --kubeconfig=kubeconfig --container-runtime=remote --network-plugin=cni --root-dir=kubelet-root --container-runtime-endpoint=unix:///run/containerd/containerd.sock --v=3", "cap_effective"=>"3fffffffff", "systemd_cgroup"=>"", "systemd_unit"=>"kubelet.service", "systemd_slice"=>"", "boot_id"=>"", "machine_id"=>"", "hostname"=>"host", "message"=>"I0127 20:29:13.744128   21573 httplog.go:89] "HTTP" verb="GET" URI="/metrics" latency="61.171583ms" userAgent="Go-http-client/1.1" resp=200"}]

Having the output as stdout with type as json does produce a data field that looks something like below.

{"date":1675095061.459447,"transport":"stdout","priority":"6","syslog_facility":"3","uid":"0","gid":"0","cap_effective":"3fffffffff","boot_id":"","machine_id":"","hostname":"host","systemd_slice":"system.slice","syslog_identifier":"containerd","comm":"containerd","exe":"/bin/containerd","cmdline":"/bin/containerd --config=config.toml --log-level info","systemd_cgroup":"containerd.service","systemd_unit":"containerd.service","stream_id":"","pid":"26982","message":"time=\"2023-01-30T16:11:01.459304626Z\" level=info msg=\"loading plugin \\\"io.containerd.grpc.v1.introspection\\\"...\" type=io.containerd.grpc.v1"}

Now, how do I create a avro schema to parse the date field?

This is how the [INPUT] is configured.

[INPUT]
    Name            systemd
    Tag             kube-systemd-kubelet
    Systemd_Filter  _SYSTEMD_UNIT=kubelet.service
    DB              /var/log/kube-systemd-kubelet.db
    Strip_Underscores on
    Lowercase on

[nareshku]

Used a Lua script to append the timestamp filed to the record. Below are the details.

[INPUT]
    Name            systemd
    Tag             kube-systemd-kubelet
    Systemd_Filter  _SYSTEMD_UNIT=kubelet.service
    DB              /var/log/systemd-kubelet.db
    Max_Entries     10
    Read_From_Tail  on
    Strip_Underscores on
    Lowercase on

[FILTER]
    Name lua
    Match kube-systemd-*
    call append_tag
    script append_tag.lua

[OUTPUT]
    Name Kafka
....

Lua Script

function append_tag(tag, timestamp, record)
    new_record = record
    new_record["timestamp"] = timestamp
    return 1, timestamp, new_record