[lbarrosby]

opensearch_data_stream inherits the pipeline parameter from opensearch, but its write builds the bulk request with only index and body, so the parameter is ignored and records skip the ingest pipeline with no warning. OpenSearch applies an ingest pipeline given on a _bulk request to data streams too, so nothing on the server side stops this from working.

This change adds pipeline to the bulk parameters and fills in chunk placeholders (pipeline-${tag}) after the records are checked and before any invalid record is sent to @ERROR. It is only expanded when the chunk has a valid record. If pipeline is not set, or it is empty after expansion (OpenSearch rejects ?pipeline= for the whole request), the request is unchanged. A short pipeline entry was added to the Data Stream section of the README, like the id_key one added in #200.

Tests added to test/plugin/test_out_opensearch_data_stream.rb:

• test_record_with_pipeline, with a fixed value and a ${tag} placeholder. The bulk request must carry ?pipeline=mypipeline or ?pipeline=pipeline-test. Both fail on main because the request goes out without the query parameter.
• test_record_with_empty_pipeline and test_record_with_pipeline_placeholder_expanded_to_empty: no pipeline query parameter is sent.
• test_invalid_records_go_to_error_when_pipeline_cannot_be_built and test_chunk_with_valid_records_is_not_sent_to_error_when_pipeline_cannot_be_built: with a tag that cannot be used in a placeholder, a chunk with only an invalid record sends it to @ERROR, a chunk with an invalid and a valid record sends nothing to @ERROR, and no bulk request is made in either case.

bundle exec rake test: 466 tests, 0 failures, 0 errors.

(check all that apply)

• tests added
• tests passing
• README updated (if needed)
• README Table of Contents updated (if needed)
• History.md and version in gemspec are untouched
• backward compatible

[Watson1978]

Thanks for the fix. I checked it with OpenSearch 3.9.0. A pipeline given on the _bulk request is applied to the data stream, and the ${tag} placeholder works too.

Two changes are needed before merging. Details are in the inline comments.

1. Do not send the pipeline parameter when the expanded value is empty. OpenSearch rejects ?pipeline= for the whole request.
2. Expand the pipeline at the top of write, before the records are processed.

[Watson1978]

Thanks for the update. Both requested changes are implemented in 81d86dc. Empty pipeline values are no longer sent, and the three new tests fail on 1cef9dd and pass now.

However, my second request introduced a regression, and I am sorry about that. When the pipeline cannot be built, a chunk with only invalid records no longer reaches @ERROR. Details, a possible fix, and tests are in the inline comment.

[Watson1978]

MUST: My earlier request in #205 (comment) caused a regression here.

Use pipeline pipe-${tag} and id_key req_id. Use <buffer tag> and the tag /badtag. Keep emit_error_label_event at its default, and use no secondary output. Feed one record whose req_id is an array.

On main and on 1cef9dd, the record goes to @ERROR. On 81d86dc, the expansion fails before the record is checked, so the record does not reach @ERROR. Fluentd backs up the chunk instead, or drops it with disable_chunk_backup true.

Please keep both of these:

• A chunk with only invalid records sends them to @ERROR, as on main.
• When the pipeline cannot be built, a chunk with both invalid and valid records does not send the invalid records to @ERROR. The whole chunk is backed up instead, as in 81d86dc.

One way is to collect the invalid records in the loop, and send them to @ERROR after the pipeline is expanded. Expand the pipeline only when records is not empty. Do not check bulk_message here. When a record fails to serialize, its action line is already in bulk_message.

      records = []
      invalid_records = []
      chunk.msgpack_each do |time, record|
        # ...
        rescue => e
          invalid_records << [time, record, e]
        end
      end

      pipeline = extract_placeholders(@pipeline, chunk) if @pipeline && !records.empty?
      if emit_error_label_event?
        invalid_records.each { |time, record, e| router.emit_error_event(tag, time, record, e) }
      end
      return if bulk_message.to_s.empty?

Remove the expansion at the top of write (this line). Keep the params[:pipeline] line as it is.

Please replace test_pipeline_is_expanded_before_records_are_processed with two tests. Both use its configuration (pipeline pipe-${tag}, id_key req_id, <buffer tag> with disable_chunk_backup true, the tag /badtag), and assert that no bulk request is made.

• Feed one record whose req_id is an array. driver.error_events has one event.
• Feed the same record and one valid record. driver.error_events is empty.

[lbarrosby]

No problem, thanks for the follow-up. Pushed 6aeed9c:

• The expansion at the top of write is gone. Invalid records are collected in the loop, the pipeline is expanded only when records is not empty, and then the invalid records go to @ERROR. The params[:pipeline] line is unchanged.
• test_pipeline_is_expanded_before_records_are_processed is replaced by two tests with the same configuration and the tag /badtag, both asserting no bulk request:
  • test_invalid_records_go_to_error_when_pipeline_cannot_be_built: one record with an array req_id, one event in driver.error_events. It fails on 81d86dc.
  • test_chunk_with_valid_records_is_not_sent_to_error_when_pipeline_cannot_be_built: the same record and a valid one, driver.error_events is empty.

TZ=UTC bundle exec rake test: 466 tests, 0 failures, 0 errors. Without TZ=UTC, test_time_placeholder fails here on 81d86dc too, because the local date is not the UTC date.

[Watson1978]

The Ruby 3.1 jobs on macOS and Windows fail in the new test test_chunk_with_valid_records_is_not_sent_to_error_when_pipeline_cannot_be_built:

<[]> expected but was
<[["/badtag",
  ...,
  {"@timestamp"=>"...", "message"=>"bad", "req_id"=>[1]},
  #<ArgumentError: 'req_id' must not be an object or an array to be used as _id>]]>

The other jobs pass.

The test expects pipe-${tag} to fail to expand for the tag /badtag. That failure comes from the path check that Fluentd added in 1.19.3. With Ruby 3.1, CI installs Fluentd 1.18.0, so the pipeline expands without an error, and the invalid record goes to @ERROR.

Could you make this test pass with Fluentd versions before 1.19.3 too? For example, skip it there, or make the expansion fail in a way that does not depend on that check. The change in out_opensearch_data_stream.rb itself looks good.