I’m using fluent-bit Kubernetes Events input plugin to collect k8s events from Kubernetes API server without any parsers or filters. The event log contains fields like involvedObject.kind, involvedObject.name and involvedObject.namespace. I tried using the kubernetes filter to enrich the event log just like its used with other container logs. But the published event logs did not contain any pod labels or annotations. Following is the fluent-bit config I used.
<code>inputs: |
[INPUT]
name kubernetes_events
tag k8s_events
kube_url https://kubernetes.default.svc
interval_sec 30
kube_request_limit 500
filters: |
[FILTER]
Name kubernetes
K8S-Logging.Parser On
K8S-Logging.Exclude On
Match k8s_events
Keep_Log Off
Merge_Log On
outputs: |
[OUTPUT]
name stdout
Match k8s_events
</code>
<code>inputs: |
[INPUT]
name kubernetes_events
tag k8s_events
kube_url https://kubernetes.default.svc
interval_sec 30
kube_request_limit 500
filters: |
[FILTER]
Name kubernetes
K8S-Logging.Parser On
K8S-Logging.Exclude On
Match k8s_events
Keep_Log Off
Merge_Log On
outputs: |
[OUTPUT]
name stdout
Match k8s_events
</code>
inputs: |
[INPUT]
name kubernetes_events
tag k8s_events
kube_url https://kubernetes.default.svc
interval_sec 30
kube_request_limit 500
filters: |
[FILTER]
Name kubernetes
K8S-Logging.Parser On
K8S-Logging.Exclude On
Match k8s_events
Keep_Log Off
Merge_Log On
outputs: |
[OUTPUT]
name stdout
Match k8s_events
How can I enrich the k8s event log in fluent-bit with extra fields like labels and annotations of the pod (involvedObject) using any specific filter or parser?
You can use LUA plugin and write custom LUA filter for this purposes
Here is an example of LUA filter:
<code>function enrich(tag, timestamp, record)
local kind = record["involvedObject"]["kind"]
local name = record["involvedObject"]["name"]
local namespace = record["involvedObject"]["namespace"]
if kind == "Pod" then
local handle = io.popen("kubectl get pod " .. name .. " -n " .. namespace .. " -o json")
local pod_info = handle:read("*a")
handle:close()
local json = require('json')
local pod = json.decode(pod_info)
if pod and pod["metadata"] then
record["pod_labels"] = pod["metadata"]["labels"]
record["pod_annotations"] = pod["metadata"]["annotations"]
end
end
return 1, timestamp, record
end
</code>
<code>function enrich(tag, timestamp, record)
local kind = record["involvedObject"]["kind"]
local name = record["involvedObject"]["name"]
local namespace = record["involvedObject"]["namespace"]
if kind == "Pod" then
local handle = io.popen("kubectl get pod " .. name .. " -n " .. namespace .. " -o json")
local pod_info = handle:read("*a")
handle:close()
local json = require('json')
local pod = json.decode(pod_info)
if pod and pod["metadata"] then
record["pod_labels"] = pod["metadata"]["labels"]
record["pod_annotations"] = pod["metadata"]["annotations"]
end
end
return 1, timestamp, record
end
</code>
function enrich(tag, timestamp, record)
local kind = record["involvedObject"]["kind"]
local name = record["involvedObject"]["name"]
local namespace = record["involvedObject"]["namespace"]
if kind == "Pod" then
local handle = io.popen("kubectl get pod " .. name .. " -n " .. namespace .. " -o json")
local pod_info = handle:read("*a")
handle:close()
local json = require('json')
local pod = json.decode(pod_info)
if pod and pod["metadata"] then
record["pod_labels"] = pod["metadata"]["labels"]
record["pod_annotations"] = pod["metadata"]["annotations"]
end
end
return 1, timestamp, record
end
After that you should edit fluentbit config and add your filter:
<code>[INPUT]
Name kube_events
Tag kube.events
[FILTER]
Name lua
Match kube.events
script /path/to/enrich.lua
call enrich
[OUTPUT]
Name stdout
Match *
</code>
<code>[INPUT]
Name kube_events
Tag kube.events
[FILTER]
Name lua
Match kube.events
script /path/to/enrich.lua
call enrich
[OUTPUT]
Name stdout
Match *
</code>
[INPUT]
Name kube_events
Tag kube.events
[FILTER]
Name lua
Match kube.events
script /path/to/enrich.lua
call enrich
[OUTPUT]
Name stdout
Match *
Or you can use fluend which is much more flexible
4