create dynamic index key using record_transformer

[yakir-shriker]

Hi All,
Im trying create dynamic index name based on k8s label, and use default index name while the specific label not found instead of null.
my configuration:

<filter kubernetes.**>
  @type record_transformer
  enable_ruby true
  <record>
    index_name ${record["kuberenetes"]["labels"]["app"]}
  </record>
</filter>

Now i got for example: "index_name":"cost-analyzer" which is good.
but application without this label get: "index_name":null
I would like to change null to "Default".
tried different variations

index_name ${record["kuberenetes"]["labels"]["app"] || "Default"}
index_name ${record["kuberenetes"]["labels"]["app"] ? "Default"}

but it doesn't work, anyone got an idea?

[yakir-shriker]

Managed to do it

<filter kubernetes.**>
  @type record_transformer
  enable_ruby true
  <record>
    index_name ${if record["kubernetes"]["labels"].include? "serviceName"; record["index_name"] = record["kubernetes"]["labels"]["serviceName"] ; else; "Default" end;}
  </record>
</filter>

Thanks anyway.

[a-masoumian]

hi @yakir-shriker

i'm using below config that is looks same as yours :

<filter kubernetes.**>
  @type record_transformer
  enable_ruby true
  <record>
    index_label ${if record["kubernetes"]["labels"].include? "serviceName"; record["index_name"] = record["kubernetes"]["labels"]["serviceName"] ; else; "Default" end;}
  </record>
</filter>



<match **>
  @type elasticsearch
  include_tag_key true
  tag_key @log_name
  @log_level debug
  log_es_400_reason true
  hosts "#{ENV['ELASTICSEARCH_HOSTS']}"
  user "#{ENV['ELASTICSEARCH_USER']}"
  password "#{ENV['ELASTICSEARCH_PASSWORD']}"

  suppress_type_name true
  include_timestamp true
  request_timeout 45s

  index_name k8s-${index_label}

  template_overwrite true
  template_name ""
  template_file /opt/bitnami/fluentd/config/index_template.json

  customize_template {"<<TAG>>": "k8s-${index_label}"}
  rollover_index true

  enable_ilm true
  ilm_policy_id fluentd-policy
  ilm_policy { "policy": { "phases": { "hot": { "min_age": "0ms", "actions": { "rollover": { "max_age": "1d", "max_size": "200mb" } } }, "delete": { "min_age": "1h", "actions": { "delete": {} } } } } }
  ilm_policy_overwrite true
  application_name log

  <buffer tag, $.kubernetes.namespace_name, $.kubernetes.labels.app, $.kubernetes.pod_name>
    @type file
    path /opt/bitnami/fluentd/logs/buffers/logs.buffer
    flush_thread_count 2
    flush_interval 5s
  </buffer>
</match>

but im getting below error :

2024-09-10 11:34:41 +0000 [error]: #0 Error while index creation - <k8s-${index_label}-log-{now/d}-000001> error_class=Elastic::Transport::Transport::Errors::BadRequest error="[400] {\"error\":{\"root_cause\":[{\"type\":\"parse_exception\",\"reason\":\"failed to parse date field [index_label] with format [uuuu.MM.dd]: [failed to parse date field [index_label] with format [uuuu.MM.dd]]\"}],\"type\":\"parse_exception\",\"reason\":\"failed to parse date field [index_label] with format [uuuu.MM.dd]: [failed to parse date field [index_label] with format [uuuu.MM.dd]]\",\"caused_by\":{\"type\":\"illegal_argument_exception\",\"reason\":\"failed to parse date field [index_label] with format [uuuu.MM.dd]\",\"caused_by\":{\"type\":\"date_time_parse_exception\",\"reason\":\"Text 'index_label' could not be parsed at index 0\"}}},\"status\":400}"

do you have any suggestions ?
fluentd version : 1.17.1
bitnami chart: fluentd-6.5.12
elastic : v8.8.0

[yakir-shriker]

Hi @a-masoumian ,
Try putting the variables you want to use in the elasticsearch output in the buffer, so it will look like this:

<filter kubernetes.**>
  @type record_transformer
  enable_ruby true
  <record>
    index_label ${if record["kubernetes"]["labels"].include? "serviceName"; record["index_label"] = record["kubernetes"]["labels"]["serviceName"] ; else; "Default" end;}
  </record>
</filter>



<match **>
  @type elasticsearch
  include_tag_key true
  tag_key @log_name
  @log_level debug
  log_es_400_reason true
  hosts "#{ENV['ELASTICSEARCH_HOSTS']}"
  user "#{ENV['ELASTICSEARCH_USER']}"
  password "#{ENV['ELASTICSEARCH_PASSWORD']}"

  suppress_type_name true
  include_timestamp true
  request_timeout 45s

  index_name k8s-${index_label}

  template_overwrite true
  template_name ""
  template_file /opt/bitnami/fluentd/config/index_template.json

  customize_template {"<<TAG>>": "k8s-${index_label}"}
  rollover_index true

  enable_ilm true
  ilm_policy_id fluentd-policy
  ilm_policy { "policy": { "phases": { "hot": { "min_age": "0ms", "actions": { "rollover": { "max_age": "1d", "max_size": "200mb" } } }, "delete": { "min_age": "1h", "actions": { "delete": {} } } } } }
  ilm_policy_overwrite true
  application_name log

  <buffer tag, $.kubernetes.namespace_name, $.kubernetes.labels.app, $.kubernetes.pod_name, index_label>
    @type file
    path /opt/bitnami/fluentd/logs/buffers/logs.buffer
    flush_thread_count 2
    flush_interval 5s
  </buffer>
</match>

if it's not working, send the full config

[a-masoumian]

Thanks for your quick response.
still getting same error :

2024-09-10 12:29:25 +0000 [info]: #0 Template '${index_label}' overwritten with /opt/bitnami/fluentd/config/index_template.json.
2024-09-10 12:29:25 +0000 [error]: #0 Error while index creation - <${index_label}-log-{now/d}-000001> error_class=Elastic::Transport::Transport::Errors::BadRequest error="[400] {\"error\":{\"root_cause\":[{\"type\":\"parse_exception\",\"reason\":\"failed to parse date field [index_label] with format [uuuu.MM.dd]: [failed to parse date field [index_label] with format [uuuu.MM.dd]]\"}],\"type\":\"parse_exception\",\"reason\":\"failed to parse date field [index_label] with format [uuuu.MM.dd]: [failed to parse date field [index_label] with format [uuuu.MM.dd]]\",\"caused_by\":{\"type\":\"illegal_argument_exception\",\"reason\":\"failed to parse date field [index_label] with format [uuuu.MM.dd]\",\"caused_by\":{\"type\":\"date_time_parse_exception\",\"reason\":\"Text 'index_label' could not be parsed at index 0\"}}},\"status\":400}"
2024-09-10 12:29:28 +0000 [warn]: #0 Could not communicate to Elasticsearch, resetting connection and trying again. [403] {"error":{"root_cause":[{"type":"parse_exception","reason":"failed to parse date field [index_label] with format [uuuu.MM.dd]: [failed to parse date field [index_label] with format [uuuu.MM.dd]]"}],"type":"security_exception","reason":"action [indices:admin/aliases] is unauthorized for user [elastic] with effective roles [superuser], this action is granted by the index privileges [manage,all]","caused_by":{"type":"parse_exception","reason":"failed to parse date field [index_label] with format [uuuu.MM.dd]: [failed to parse date field [index_label] with format [uuuu.MM.dd]]","caused_by":{"type":"illegal_argument_exception","reason":"failed to parse date field [index_label] with format [uuuu.MM.dd]","caused_by":{"type":"date_time_parse_exception","reason":"Text 'index_label' could not be parsed at index 0"}}}},"status":403}

attached my full config file and also a working config file (only to proof that its not elastic issue),
thanx in advance
full-config.md
working-config.md

[yakir-shriker]

try editing your filter index_label should be also in the if statement

<filter kubernetes.**>
  @type record_transformer
  enable_ruby true
  <record>
    index_label ${if record["kubernetes"]["labels"].include? "serviceName"; record["index_label"] = record["kubernetes"]["labels"]["serviceName"] ; else; "Default" end;}
  </record>
</filter>

[a-masoumian]

not again :

# Prometheus Exporter Plugin
# input plugin that exports metrics
<source>
  @type prometheus
  port 24231
</source>

# input plugin that collects metrics from MonitorAgent
<source>
  @type prometheus_monitor
  <labels>
    host ${hostname}
  </labels>
</source>

# input plugin that collects metrics for output plugin
<source>
  @type prometheus_output_monitor
  <labels>
    host ${hostname}
  </labels>
</source>

# Ignore fluentd own events
<match fluent.**>
  @type null
</match>

# TCP input to receive logs from the forwarders
<source>
  @type forward
  bind 0.0.0.0
  port 24224
</source>

# HTTP input for the liveness and readiness probes
<source>
  @type http
  bind 0.0.0.0
  port 9880
</source>

# Throw the healthcheck to the standard output instead of forwarding it
<match fluentd.healthcheck>
  @type stdout
</match>


<filter kubernetes.**>
  @type record_transformer
  enable_ruby true
  <record>
    index_label ${if record["kubernetes"]["labels"].include? "serviceName"; record["index_label"] = record["kubernetes"]["labels"]["serviceName"] ; else; "Default" end;}
  </record>
</filter>



<match **>
  @type elasticsearch
  include_tag_key true
  tag_key @log_name
  #@log_level debug
  #log_es_400_reason true
  hosts "#{ENV['ELASTICSEARCH_HOSTS']}"
  user "#{ENV['ELASTICSEARCH_USER']}"
  password "#{ENV['ELASTICSEARCH_PASSWORD']}"

  suppress_type_name true
  include_timestamp true
  request_timeout 60s

  index_name k8s-${index_label}

  template_overwrite true
  template_name ""
  template_file /opt/bitnami/fluentd/config/index_template.json

  customize_template {"<<TAG>>": "k8s-${index_label}"}
  rollover_index true

  enable_ilm true
  ilm_policy_id fluentd-policy
  ilm_policy { "policy": { "phases": { "hot": { "min_age": "0ms", "actions": { "rollover": { "max_age": "1d", "max_size": "200mb" } } }, "delete": { "min_age": "1h", "actions": { "delete": {} } } } } }
  ilm_policy_overwrite true
  application_name log

  <buffer tag, $.kubernetes.namespace_name, $.kubernetes.labels.app, index_label>
    @type file
    path /opt/bitnami/fluentd/logs/buffers/logs.buffer
    flush_thread_count 2
    flush_interval 5s
  </buffer>
</match>

thanks for your reply.