I cannot use tag_parts or tag_prefix in the output [Q&A]

[jtamayo-appgate]

What is a problem?

The tag_parts or tag prefix in the output does not work.
I have the output configuration to send it to opensearch and an S3 bucket and I am setting the path with tag_parts but it does not print the value but the string.
The output for example for the index is:

{"index":{"_index":"dtp-${tag_parts[1]}-2024.06.25"}}

Describe the configuration of Fluentd

01_sources.conf: |-

@type tail
@log_level debug
@Label @kubernetes
read_from_head true
tag kubernetes.*
path /var/log/containers/.log
pos_file /var/log/fluentd-containers.log.pos
exclude_path ["/var/log/containers/fluent"]

@type multi_format

format json
time_key time
time_type string
time_format "%Y-%m-%dT%H:%M:%S.%NZ"
keep_time_key false


format regexp
expression /^(?.+) (?stdout|stderr)( (.))? (?.*)$/
time_format '%Y-%m-%dT%H:%M:%S.%NZ'
keep_time_key false


emit_unmatched_lines true

# expose metrics in prometheus format
<source>
  @type prometheus
  bind 0.0.0.0
  port 24231
  metrics_path /metrics
</source>

02_filters.conf: |-
<label @kubernetes>
<match kubernetes.var.log.containers.fluentd**>
@type relabel
@Label @FLUENT_LOG

<filter kubernetes.**>
  @type kubernetes_metadata
  @log_level debug
  @id filter_kube_metadata
  kubernetes_url "#{ENV['FLUENT_FILTER_KUBERNETES_URL'] || 'https://' + ENV.fetch('KUBERNETES_SERVICE_HOST') + ':' + ENV.fetch('KUBERNETES_SERVICE_PORT') + '/api'}"
  verify_ssl "#{ENV['KUBERNETES_VERIFY_SSL'] || true}"
  ca_file "#{ENV['KUBERNETES_CA_FILE']}"
  skip_labels "#{ENV['FLUENT_KUBERNETES_METADATA_SKIP_LABELS'] || 'false'}"
  skip_container_metadata "#{ENV['FLUENT_KUBERNETES_METADATA_SKIP_CONTAINER_METADATA'] || 'false'}"
  skip_master_url "#{ENV['FLUENT_KUBERNETES_METADATA_SKIP_MASTER_URL'] || 'false'}"
  skip_namespace_metadata "#{ENV['FLUENT_KUBERNETES_METADATA_SKIP_NAMESPACE_METADATA'] || 'true'}"
  skip_filter_kube_metadata "#{ENV['FLUENT_KUBERNETES_METADATA_SKIP_FILTER_KUBE_METADATA'] || 'false'}"
  emit_invalid_record_to_error false
</filter>

<filter kubernetes.**>
  @type parser
  @log_level debug
  key_name "$.log"
  hash_value_field "log"
  reserve_data true
  <parse>
    @type json
  </parse>
  emit_invalid_record_to_error false
</filter>

<filter kubernetes.**>
  @type grep
  @log_level debug
  <regexp>
    key $.kubernetes.labels.logging
    pattern /opensearch/
  </regexp>
  <record>
    prueba "funciona"
  </record>
</filter>

<filter **>
  @type record_transformer
  enable_ruby true
  <record>
  type ${record.delete('_type')}
  </record>
</filter>

<filter **>
  @type record_transformer
  @log_level debug
  enable_ruby
  <record>
    kubernetes_tag ${"k8s.%s.%s.%s.%s" % [record["kubernetes"]["labels"]["env"], record["kubernetes"]["labels"]["type"], record["kubernetes"]["labels"]["app"], record["kubernetes"]["pod_id"]]}
  </record>
</filter>

<match kubernetes.**>
  @type rewrite_tag_filter
  @log_level debug
  <rule>
    key     $.kubernetes_tag
    pattern /^k8s\.([^.]+)\..+$/
    tag     $1
  </rule>
</match>

<filter **>
  @type record_transformer
  @log_level debug
  enable_ruby
  <record>
    context.namespace ${record["kubernetes"]["namespace_name"]}
    context.cluster "#{ENV['KUBERNETES_CLUSTER']}"
    context.hostname ${record["kubernetes"]["host"]}
    context.created ${time.strftime('%Y-%m-%dT%H:%M:%S.%NZ')}
    context.environment ${record["kubernetes"]["labels"]["env"]}
    service.name ${record["kubernetes"]["labels"]["app"]}
    service.node ${record["kubernetes"]["host"]}
    service.type ${record["kubernetes"]["labels"]["type"]}
  </record>
  remove_keys $["kubernetes"]["master_url"],$["kubernetes"]["labels"],$["docker"]["container_id"],$["kubernetes"]["namespace_id"],$["kubernetes"]["namespace_name"],$["kubernetes"]["host"],$["kubernetes"]["pod_name"],$["kubernetes"]["container_name"],$["kubernetes"]["pod_name"],$["kubernetes"]["container_image_id"],$["kubernetes"]["pod_id"],$["stream"]
</filter>

  <match **>
    @type relabel
    @log_level debug
    @label @DISPATCH
  </match>
</label>

03_dispatch.conf: |-
<label @dispatch>
<filter **>
@type prometheus
@log_level debug

name fluentd_input_status_num_records_total
type counter
desc The total number of incoming records

tag ${tag}
hostname ${hostname}

  <match **>
    @type relabel
    @log_level debug
    @label @OUTPUT
  </match>
</label>

04_outputs.conf: |-
<label @output>
<match **>
@type copy

@type opensearch
@log_level debug
host "#{ENV['FLUENT_ELASTICSEARCH_HOST'] || 'elasticsearch.elastic-kibana'}"
port "#{ENV['FLUENT_ELASTICSEARCH_PORT'] || '9200'}"
scheme "#{ENV['FLUENT_ELASTICSEARCH_SCHEME'] || 'http'}"
ssl_verify "#{ENV['FLUENT_ELASTICSEARCH_SSL_VERIFY']}"
ssl_version "#{ENV['FLUENT_ELASTICSEARCH_SSL_VERSION']}"
logstash_prefix "#{ENV['KUBERNETES_CLUSTER']}-${tag_parts[1]}"
include_tag_key false
logstash_format true
request_timeout 60s
with_transporter_log true
time_key_exclude_timestamp true
verify_es_version_at_startup false
<buffer tag,time>
@type file
path /var/log/fluentd-buffers/opensearch
timekey 10m
timekey_wait 2m
timekey_use_utc true
flush_interval 10s
chunk_limit_size 1g
total_limit_size 5g



@type s3
s3_bucket my_bucket
s3_region us-east-1
path ${tag_parts[1]}/${tag_parts[2]}/${tag_parts[3]}/
time_slice_format %Y/%m/%d/%H/%Hh-%Mm-${tag_parts[4]}
buffer_chunk_limit 50m
buffer_queue_limit 2048

timekey 10m
flush_interval 10m

Describe the logs of Fluentd

No response

Environment

- Fluentd version: 1.16.2
- TD Agent version:
- Fluent Package version:
- Docker image (tag): fluent/fluentd-kubernetes-daemonset
- Operating system:
- Kernel version: 6.1.92-99.174.amzn2023.x86_64