I'm using Fluent Bit (1.3.11) to collect logs from containers running on k8s. What I would like to do is some basic processing using Fluent Bit Modify filters (i.e. to standardize log level names).
I added two modify filters (see below). The first one is just to verify that modify filters work. It checks if log key is present and sets a property if found. That works. The second modify filter is what I'm actually trying to do. It supposedly rewrites log_processed.Level value from ERR to Error but I couldn't make it actually work. After some experimenting I think that none of the filters that have a condition depending on either log_processed.* or kubernetes.* properties (added by the kubernetes filter) work.
Is there a recommended/working way to modify logs comming from Kubernetes?
[FILTER]
Name kubernetes
Match kube.*
Kube_URL https://kubernetes.default.svc:443
Kube_CA_File /var/run/secrets/kubernetes.io/serviceaccount/ca.crt
Kube_Token_File /var/run/secrets/kubernetes.io/serviceaccount/token
Kube_Tag_Prefix kube.var.log.containers.
Merge_Log On
Merge_Log_Key log_processed
K8S-Logging.Parser On
K8S-Logging.Exclude Off
[FILTER]
Name modify
Match kube.*
Condition Key_exists log
Set my.custom.prop modify-filter-applied
[FILTER]
NAME modify
Match kube.*
Condition Key_value_equals log_processed.Level ERR
Set log_processed.Level ErrorI managed to utilize a Lua filter to modify the records.
[FILTER]
Name lua
Match kube.*
script levelRewrite.lua
call level_rewrite
function level_rewrite(tag, timestamp, record)
for key, val in pairs(record) do
if key == "log_processed" then
if val["level"] == "I" then
val["level"] = "Info"
-- return code 2 is supported in Fluent Bit v.1.4.3+
return 2, timestamp, record
end
end
end
return 0, 0, 0
end