From a14e55da699cbf94f9ee46e0cf5f36a23bcd17cd Mon Sep 17 00:00:00 2001 From: Josh Wilson Date: Mon, 7 Sep 2026 21:33:28 -0500 Subject: [PATCH] Pass transport metadata to event decoders --- .../indexer/indexing_event_decoder.rb | 5 ++- .../indexer/indexing_event_decoder.rbs | 4 +++ elasticgraph-indexer_lambda/README.md | 5 +++ .../indexer_lambda/sqs_processor.rb | 12 +++++-- .../indexer_lambda/sqs_processor.rbs | 2 +- .../indexer_lambda/sqs_processor_spec.rb | 32 +++++++++++++++++++ 6 files changed, 55 insertions(+), 5 deletions(-) diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/indexing_event_decoder.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/indexing_event_decoder.rb index 647e9c91d..0fe4a55a6 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/indexing_event_decoder.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/indexing_event_decoder.rb @@ -10,7 +10,10 @@ module ElasticGraph class Indexer # Namespace for indexing event decoders, which turn raw payload strings from a transport into # ElasticGraph indexing event hashes. The decoder to use is configured via the - # `indexer.indexing_event_decoder` setting. + # `indexer.indexing_event_decoder` setting. A decoder may additionally implement + # `decode_with_metadata(payload, metadata:)` to receive transport headers/properties. + # The SQS integration uses this optional capability to pass message attribute string values; + # decoders implementing only `decode(payload)` remain supported. module IndexingEventDecoder # Defines the indexing event decoder interface, which our extension loader will validate against. class Interface diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer/indexing_event_decoder.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer/indexing_event_decoder.rbs index 2ada53193..90405245c 100644 --- a/elasticgraph-indexer/sig/elastic_graph/indexer/indexing_event_decoder.rbs +++ b/elasticgraph-indexer/sig/elastic_graph/indexer/indexing_event_decoder.rbs @@ -3,6 +3,10 @@ module ElasticGraph type indexingEventDecoder = IndexingEventDecoder::Interface module IndexingEventDecoder + interface _MetadataAwareDecoder + def decode_with_metadata: (::String, metadata: ::Hash[::String, untyped]) -> ::Array[event] + end + class Interface def initialize: ( config: ::Hash[::Symbol | ::String, untyped], diff --git a/elasticgraph-indexer_lambda/README.md b/elasticgraph-indexer_lambda/README.md index c3134ce5f..c62255f1b 100644 --- a/elasticgraph-indexer_lambda/README.md +++ b/elasticgraph-indexer_lambda/README.md @@ -33,6 +33,11 @@ decoded by the decoder configured via the `indexer.indexing_event_decoder` setti README for details). For example, with `ElasticGraph::JSONIngestion::IndexingEventDecoder` (provided by `elasticgraph-json_ingestion`), messages use [JSON Lines](https://jsonlines.org/) format to encode indexing events. +A decoder can also implement `decode_with_metadata(payload, metadata:)` to receive SQS message attributes. +The `metadata` hash maps attribute names to their `stringValue` values, including number attributes encoded as strings. +It is empty when a message has no attributes. Attributes are passed alongside the fetched body for S3-offloaded messages. +Decoders that only implement `decode(payload)` continue to receive the payload as before. + JSON lines format contains individual JSON objects delimited by a newline control character (not the `\n` string sequence), such as: diff --git a/elasticgraph-indexer_lambda/lib/elastic_graph/indexer_lambda/sqs_processor.rb b/elasticgraph-indexer_lambda/lib/elastic_graph/indexer_lambda/sqs_processor.rb index 13fc41be1..6640ac349 100644 --- a/elasticgraph-indexer_lambda/lib/elastic_graph/indexer_lambda/sqs_processor.rb +++ b/elasticgraph-indexer_lambda/lib/elastic_graph/indexer_lambda/sqs_processor.rb @@ -73,7 +73,8 @@ def events_from(lambda_event) sqs_metadata = sqs_metadata.except("latency_timestamps") end - decoded_events_from(record.fetch("body")).map do |event| + metadata = (record["messageAttributes"] || {}).transform_values { |attribute| attribute["stringValue"] } + decoded_events_from(record.fetch("body"), metadata: metadata).map do |event| ElasticGraph::Support::HashUtil.deep_merge(event, sqs_metadata) end end.tap do @@ -86,12 +87,17 @@ def events_from(lambda_event) S3_OFFLOADING_INDICATOR = '["software.amazon.payloadoffloading.PayloadS3Pointer"' - def decoded_events_from(payload) + def decoded_events_from(payload, metadata:) if payload.start_with?(S3_OFFLOADING_INDICATOR) payload = get_payload_from_s3(payload) end - @indexing_event_decoder.decode(payload) + if @indexing_event_decoder.respond_to?(:decode_with_metadata) + decoder = @indexing_event_decoder # : Indexer::indexingEventDecoder & Indexer::IndexingEventDecoder::_MetadataAwareDecoder + decoder.decode_with_metadata(payload, metadata: metadata) + else + @indexing_event_decoder.decode(payload) + end end def extract_sqs_metadata(record) diff --git a/elasticgraph-indexer_lambda/sig/elastic_graph/indexer_lambda/sqs_processor.rbs b/elasticgraph-indexer_lambda/sig/elastic_graph/indexer_lambda/sqs_processor.rbs index 9c34781fe..3003d1f59 100644 --- a/elasticgraph-indexer_lambda/sig/elastic_graph/indexer_lambda/sqs_processor.rbs +++ b/elasticgraph-indexer_lambda/sig/elastic_graph/indexer_lambda/sqs_processor.rbs @@ -24,7 +24,7 @@ module ElasticGraph S3_OFFLOADING_INDICATOR: String def extract_sqs_metadata: (::Hash[String, untyped]) -> ::Hash[::String, untyped] def millis_to_iso8601: (::String) -> ::String? - def decoded_events_from: (::String) -> ::Array[::Hash[::String, untyped]] + def decoded_events_from: (::String, metadata: ::Hash[::String, untyped]) -> ::Array[::Hash[::String, untyped]] def get_payload_from_s3: (::String) -> ::String def s3_client: () -> Aws::S3::Client def format_response: ( diff --git a/elasticgraph-indexer_lambda/spec/unit/elastic_graph/indexer_lambda/sqs_processor_spec.rb b/elasticgraph-indexer_lambda/spec/unit/elastic_graph/indexer_lambda/sqs_processor_spec.rb index cd6e1f19d..a6bdfbf35 100644 --- a/elasticgraph-indexer_lambda/spec/unit/elastic_graph/indexer_lambda/sqs_processor_spec.rb +++ b/elasticgraph-indexer_lambda/spec/unit/elastic_graph/indexer_lambda/sqs_processor_spec.rb @@ -93,6 +93,38 @@ module IndexerLambda ], refresh_indices: false) end + it "passes transport attributes to metadata-aware decoders for direct and S3-offloaded payloads" do + decoder_class = Class.new do + def decode_with_metadata(payload, metadata:) + end + end + custom_decoder = instance_spy(decoder_class, decode_with_metadata: [{"field1" => {}}]) + message = sqs_message("a", "protobuf-payload").merge("messageAttributes" => { + "eg_type" => {"stringValue" => "Widget", "dataType" => "String"}, + "eg_version" => {"stringValue" => "7", "dataType" => "Number"} + }) + + offloaded_message = message.merge("messageId" => "b", "body" => JSON.generate([ + "software.amazon.payloadoffloading.PayloadS3Pointer", {"s3BucketName" => "events", "s3Key" => "payload"} + ])) + s3_client.stub_responses(:get_object, body: "protobuf-payload") + message_without_attributes = message.except("messageAttributes").merge("messageId" => "c") + + build_sqs_processor(indexing_event_decoder: custom_decoder).process({ + "Records" => [message, offloaded_message, message_without_attributes] + }) + + expect(custom_decoder).to have_received(:decode_with_metadata).with( + "protobuf-payload", metadata: {"eg_type" => "Widget", "eg_version" => "7"} + ).twice + expect(custom_decoder).to have_received(:decode_with_metadata).with("protobuf-payload", metadata: {}) + expect(indexer_processor).to have_received(:process_returning_failures).with([ + {"field1" => {}, "message_id" => "a"}, + {"field1" => {}, "message_id" => "b"}, + {"field1" => {}, "message_id" => "c"} + ], refresh_indices: false) + end + it "logs the SQS message ids received in the lambda event and the `sqs_received_at` if available" do sent_timestamp_millis = "796010423456" sent_timestamp_iso8601 = "1995-03-24T02:00:23.456Z"