Extract a streamname value and replace in kinesis stream_name - extraction didn't works out- Need suggestion

[ak-gamer86]

Hello folks,'m trying to achieve to send the syslog logs to splunk as well to kinesis from fluentd using . when the fluentd receives a syslog logs, I do not want to send the whole logs as such to kinesis, hence I'm trying to do below stuffs when sending to kinesis alone.

• filter to extract only event attribute
• use transformer to create a new record with only the transformed fields, keep only event key drop all others.
• Removes the 'event' key using remove_key_name_field after parsing
• Send transformed content to Kinesis

Here the kinesis streamname where I want to push the data is part in final record, problem I face is partition key which is available in tat content is successfully substituted for partition_key under @type kinesis_streams where as the stream_name is not getting substituted. kindly see my configuration

#Label section for Kinesis (transformed content)
<label @kinesis_output>

  # Filter to extract only event attribute 
  # Use transformer, create a new record with only the transformed fields, keep only event key drop all others

  <filter syslog.*.*.*.*>
    @type record_transformer
    renew_record true
    keep_keys event
    enable_ruby true  
  </filter>

  # Parser removes the message key & expands its string value as JSON.
  # keep original record alongside the parsed result
  # Removes the 'event' key using remove_key_name_field after parsing, 
  # final record will not contain the original JSON string.
  <filter syslog.*.*.*.*>
    @type parser
    key_name event
    reserve_data true
    remove_key_name_field true
    <parse>
      @type json
    </parse>
  </filter>

 Note: After the above piece is executed, my data looks like this 
 {"saccade"=>nil, "accid"=>"1769992098", "timestamp"=>"2024-10-07T06:47:45.606Z", "userId"=>test_user, "personaId"=>test_persona, "cid"=>"7-oct-1220PM-utc-pr-pb-streamnamefromapp", "drmScheme"=>"false", "status"=>200, "release"=>"abc-rls", "licenseType"=>"STREAMING", "contentId"=>"abc-content", "keyIds"=>"7jhgj87hv7t", "keyType"=>nil, "keyAlgorithm"=>yes, "kinesisStream"=>"my_test_stream", "partitionKey"=>"my_test_partition"}


  # Send transformed content to Kinesis
  <match syslog.*.*.*.*>
    @type kinesis_streams
    region us-west-2
    role_arn arn:aws:iam::3040876341599:role/Custom-Fluent-Kinesis
    partition_key partitionKey
    stream_name kinesisStream
    <buffer>
      @type memory
      flush_interval 5s  # How often to send buffered data
    </buffer>
  </match>
</label>

Here as we can see in data "kinesisStream"=>"my_test_stream", "partitionKey"=>"my_test_partition", - my_test_stream value is not getting substituted to stream_name while send transformed content to Kinesis.

where as partitionKey value (my_test_partition) in data is properly getting substituted- this I have confirmed by hard coding the stream name and verified the partition name used in kinesis. During this verification when hardcoding streamname in fluentd, "my_test_partition" value is considered properly for partition_key.

If partition_key is able to properly gets considered & substituted ,not sure why stream name is not considered, so I end up with issue Aws::Kinesis::Errors::ResourceNotFoundException error="Stream kinesisStream under account 3040876341599 not found.".

Please suggest what's the issue with stream name substitution under <match syslog....> @type kinesis_streams ?

[ak-gamer86]

Fluentd experts, any suggestion on this clarification would help unblock me.