Parsing Ingress-Nginx Logs to Elasticsearch with Individual Key-Value Pairs for Enhanced Kibana Filtering

[cherrycharan]

I am encountering challenges in effectively filtering Ingress-Nginx logs in Kibana. Currently, these logs are sent as a JSON array to Elasticsearch, making individual key filtering difficult. My goal is to parse these logs such that each field in the log is treated as a separate key-value pair, facilitating more efficient filtering in Kibana.

Current Configuration:

Ingress-Nginx-Controller Log Format:
I have configured the log format of the Ingress-Nginx-Controller as follows:

{ "@timestamp": "$time_local", "client": "$remote_addr", "method":"$request_method", "URL": "$host", "request": "$request", "request_id": "$req_id", "request_length": "$request_length", "bytes_sent": "$bytes_sent", "status": "$status", "body_bytes_sent": "$body_bytes_sent", "referer": "$http_referer", "user_agent": "$http_user_agent", "upstream_addr": "$upstream_addr", "upstream_status": "$upstream_status", "request_time": "$request_time", "upstream_response_time": "$upstream_response_time", "upstream_connect_time": "$upstream_connect_time", "upstream_header_time": "$upstream_header_time" }

Fluentd Configuration:
My Fluentd configuration is set with the following source and destination parameters:

<match fluent.>
@type null


@type tail
path /var/log/containers/.log
exclude_path ["/var/log/containers/prometheus.log", "/var/log/containers/logging.log"]
pos_file /var/log/fluentd-containers.log.pos
tag kubernetes.
format json
read_from_head false

@type "#{ENV['FLUENT_CONTAINER_TAIL_PARSER_TYPE'] || 'json'}"
time_format %Y-%m-%dT%H:%M:%S.%NZ


<filter kubernetes.>
@type kubernetes_metadata
verify_ssl false


@type regexp
expression /^(?.+) (?stdout|stderr)( (?.))? (?.*)$/

<match kubernetes.**>
@type elasticsearch_dynamic
include_tag_key true
logstash_format true
logstash_prefix kubernetes-${record['kubernetes']['namespace_name']}
host "#{ENV['FLUENT_ELASTICSEARCH_HOST']}"
port "#{ENV['FLUENT_ELASTICSEARCH_PORT']}"
scheme "#{ENV['FLUENT_ELASTICSEARCH_SCHEME'] || 'http'}"
user "#{ENV['FLUENT_ELASTICSEARCH_USER']}"
password "#{ENV['FLUENT_ELASTICSEARCH_PASSWORD']}"
reload_connections false
reconnect_on_error true
reload_on_failure true

@type file
path /fluentd/log/elastic-buffer
flush_thread_count 16
flush_interval 60s
chunk_limit_size 10M
queue_limit_length 16
flush_mode interval
retry_max_interval 30
retry_forever true

Issue:
Despite this configuration, I am unable to parse each log field into separate key-value pairs for Elasticsearch. This is limiting my ability to filter logs effectively in Kibana.

Request for Assistance:
I am seeking guidance or suggestions on how to modify my configuration to achieve the desired log parsing and filtering capabilities. Any insights or recommendations from the community would be greatly appreciated.