Placeholders don't work in data_stream_name

[voronin-ilya]

(check apply)

• [x] read the contribution guideline
• [ ] (optional) already reported 3rd party upstream repository or mailing list if you use k8s addon or helm charts.

Steps to replicate

Configure Fluentd to write to a data stream using a placeholder for dynamic stream name resolution:

<match datastream.test**>
  @type opensearch_data_stream
  data_stream_name ${indexname}
...
  <endpoint>
    url https://xxxxxxxxx.xxxxxxxxxxx.es.amazonaws.com
    region xxxxxxx
    assume_role_arn "#{ENV['AWS_ROLE_ARN']}"
    assume_role_web_identity_token_file "#{ENV['AWS_WEB_IDENTITY_TOKEN_FILE']}"
  </endpoint>
...
</match>

Expected Behavior or What you need to ask

Expected: The stream name is dynamically looked up from the event, just as it works for a regular @type opensearch output. Actual: ${indexname} is passed literally as data stream name to OpenSearch.

Using Fluentd and OpenSearch plugin versions

• OS version
• Linux logs-aggregator-2 6.1.119-129.201.amzn2023.x86_64 #1 SMP PREEMPT_DYNAMIC Tue Dec 3 21:07:35 UTC 2024 x86_64 GNU/Linux
• Bare Metal or within Docker or Kubernetes or others?
• Kubernetes, EKS
• Fluentd v1.0 or later
  • paste result of fluentd --version or td-agent --version
  • fluentd 1.16.2
• OpenSearch plugin version
  • paste boot log of fluentd or td-agent
  • paste result of fluent-gem list, td-agent-gem list or your Gemfile.lock

opensearch-api (2.2.0)
opensearch-ruby (2.1.0)
opensearch-transport (2.1.0)

• OpenSearch version (optional)
• OpenSearch template(s) (optional)

[daipom]

@voronin-ilya Do you set indexname as one of the buffer keys? If you use an arbitrary field as a placeholder, you need to include that field in the buffer keys.

If this is the cause, you can see the following debug log when you set the log level to DEBUG.

• ... is tested built-in placeholder(s) but there is no valid placeholder(s). ...

[voronin-ilya]

Yes, indexname was included in the buffer keys. Here is the full match snippet:

##############
<filter datastream.**>
  @type record_transformer
  <record>
    indexname datastream-test
  </record>
</filter>
#################
<match datastream.test**>
  @type opensearch_data_stream
  data_stream_name ${indexname}
  flush_interval 1s
  include_tag_key true
  tag_key @tags
  logstash_format true
  reconnect_on_error true
  reload_connections false
  reload_on_failure false
  request_timeout 15s
  bulk_message_request_threshold 5m
  log_os_400_reason true

  <endpoint>
    url https://xxxxxxxxx.xxxxxxxxxxx.es.amazonaws.com
    region xxxxxxx
    assume_role_arn "#{ENV['AWS_ROLE_ARN']}"
    assume_role_web_identity_token_file "#{ENV['AWS_WEB_IDENTITY_TOKEN_FILE']}"
  </endpoint>

  <buffer tag,indexname,indexformat>
    @type file

    flush_at_shutdown true
    flush_mode interval
    chunk_limit_size 10m
    total_limit_size 8g
    path /var/log/fluent/itprod-logs-ds

    retry_wait 30s
    retry_max_interval 10m
    flush_interval 10s

    overflow_action drop_oldest_chunk
  </buffer>
</match>