Pass transport metadata to event decoders

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

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],

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:
 

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)

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: (

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"