From 8ea57121f0aa64d59d1254f7e204ef7dd200f8f7 Mon Sep 17 00:00:00 2001 From: Josh Wilson Date: Sat, 30 May 2026 15:26:14 -0500 Subject: [PATCH 1/5] Add configurable indexing event decoder --- elasticgraph-indexer/README.md | 26 ++++++++ .../lib/elastic_graph/indexer.rb | 11 ++++ .../lib/elastic_graph/indexer/config.rb | 60 ++++++++++++++++++- .../lib/elastic_graph/indexer/event_id.rb | 4 +- .../indexer/indexing_event_decoder.rb | 41 +++++++++++++ .../sig/elastic_graph/indexer.rbs | 3 + .../sig/elastic_graph/indexer/config.rbs | 14 ++++- .../indexer/indexing_event_decoder.rbs | 20 +++++++ .../indexing_event_decoder.rb | 26 ++++++++ .../unit/elastic_graph/indexer/config_spec.rb | 28 +++++++++ .../indexer/indexing_event_decoder_spec.rb | 28 +++++++++ .../spec/unit/elastic_graph/indexer_spec.rb | 22 +++++++ .../indexer_lambda/lambda_function.rb | 1 + .../indexer_lambda/sqs_processor.rb | 50 +++++++++------- .../indexer_lambda/sqs_processor.rbs | 5 +- .../indexer_lambda/sqs_processor_spec.rb | 20 +++++++ .../local/spec_support/config_schema.yaml | 39 ++++++++++++ 17 files changed, 366 insertions(+), 32 deletions(-) create mode 100644 elasticgraph-indexer/lib/elastic_graph/indexer/indexing_event_decoder.rb create mode 100644 elasticgraph-indexer/sig/elastic_graph/indexer/indexing_event_decoder.rbs create mode 100644 elasticgraph-indexer/spec/support/example_extensions/indexing_event_decoder.rb create mode 100644 elasticgraph-indexer/spec/unit/elastic_graph/indexer/indexing_event_decoder_spec.rb diff --git a/elasticgraph-indexer/README.md b/elasticgraph-indexer/README.md index e4f82753b..52a3be8c9 100644 --- a/elasticgraph-indexer/README.md +++ b/elasticgraph-indexer/README.md @@ -48,3 +48,29 @@ indexer = ElasticGraph::Indexer.from_yaml_file("config/settings/local.yaml") events = [] # JSON events read from an async datastream indexer.processor.process(events) ``` + +## Custom Payload Decoding + +`ElasticGraph::Indexer` can be configured with an indexing event decoder extension. Decoders turn raw payload strings +from a transport into ElasticGraph indexing event hashes before the normal validation and indexing pipeline runs. The +default decoder expects JSON Lines. + +```yaml +indexer: + indexing_event_decoder: + name: MyCompany::ElasticGraph::CSVIndexingEventDecoder + require_path: ./lib/my_company/elastic_graph/csv_indexing_event_decoder + config: + delimiter: "," +``` + +Decoder extensions must implement: + +```ruby +def initialize(config:, schema_artifacts:, logger:) +end + +def decode(payload) + # return an array of ElasticGraph indexing event hashes +end +``` diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer.rb b/elasticgraph-indexer/lib/elastic_graph/indexer.rb index 2d64978bd..a5fbea39a 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer.rb @@ -86,6 +86,17 @@ def operation_factory end end + def indexing_event_decoder + @indexing_event_decoder ||= begin + extension = config.indexing_event_decoder + (_ = extension.extension_class).new( + config: extension.config, + schema_artifacts: schema_artifacts, + logger: logger + ) + end + end + def monotonic_clock @monotonic_clock ||= begin require "elastic_graph/support/monotonic_clock" diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb index 2b760f0ca..ad9fa354e 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb @@ -8,10 +8,17 @@ require "elastic_graph/support/config" require "elastic_graph/errors" +require "elastic_graph/indexer/indexing_event_decoder" +require "elastic_graph/schema_artifacts/runtime_metadata/extension_loader" module ElasticGraph class Indexer - class Config < Support::Config.define(:latency_slo_thresholds_by_timestamp_in_ms, :skip_derived_indexing_type_updates) + class Config < Support::Config.define(:latency_slo_thresholds_by_timestamp_in_ms, :skip_derived_indexing_type_updates, :indexing_event_decoder) + DEFAULT_INDEXING_EVENT_DECODER = { + "name" => "ElasticGraph::Indexer::IndexingEventDecoder::JSONLines", + "require_path" => "elastic_graph/indexer/indexing_event_decoder" + } + json_schema at: "indexer", optional: false, description: "Configuration for indexing operations and metrics used by `elasticgraph-indexer`.", @@ -42,17 +49,64 @@ class Config < Support::Config.define(:latency_slo_thresholds_by_timestamp_in_ms {}, # : untyped {"WidgetWorkspace" => ["ABC12345678"]} ] + }, + indexing_event_decoder: { + description: "Extension object used to decode raw indexing payloads into ElasticGraph indexing event hashes. The default decoder expects JSON Lines.", + type: "object", + properties: { + name: { + description: "The name of the indexing event decoder extension class.", + type: "string", + minLength: 1, + examples: ["MyCompany::ElasticGraph::CSVIndexingEventDecoder"] + }, + require_path: { + description: "The path to require to load the indexing event decoder extension.", + type: "string", + minLength: 1, + examples: ["./lib/my_company/elastic_graph/csv_indexing_event_decoder"] + }, + config: { + description: "Configuration for the indexing event decoder. Will be passed into the decoder's `#initialize` method.", + type: "object", + default: {}, # : untyped + examples: [ + {}, # : untyped + {"delimiter" => ","} + ] + } + }, + required: ["name", "require_path"], + default: DEFAULT_INDEXING_EVENT_DECODER, + examples: [ + DEFAULT_INDEXING_EVENT_DECODER, + { + "name" => "MyCompany::ElasticGraph::CSVIndexingEventDecoder", + "require_path" => "./lib/my_company/elastic_graph/csv_indexing_event_decoder", + "config" => {"delimiter" => ","} + } + ] } } private - def convert_values(skip_derived_indexing_type_updates:, latency_slo_thresholds_by_timestamp_in_ms:) + def convert_values(skip_derived_indexing_type_updates:, latency_slo_thresholds_by_timestamp_in_ms:, indexing_event_decoder:) { skip_derived_indexing_type_updates: skip_derived_indexing_type_updates.transform_values(&:to_set), - latency_slo_thresholds_by_timestamp_in_ms: latency_slo_thresholds_by_timestamp_in_ms + latency_slo_thresholds_by_timestamp_in_ms: latency_slo_thresholds_by_timestamp_in_ms, + indexing_event_decoder: load_indexing_event_decoder(indexing_event_decoder) } end + + def load_indexing_event_decoder(config) + loader = SchemaArtifacts::RuntimeMetadata::ExtensionLoader.new(IndexingEventDecoder::Interface) + loader.load( + config.fetch("name"), + from: config.fetch("require_path"), + config: config["config"] || {} + ) + end end end end diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/event_id.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/event_id.rb index 133ff5d7d..a3e3e1e9c 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/event_id.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/event_id.rb @@ -12,7 +12,7 @@ module ElasticGraph class Indexer # A unique identifier for an event ingested by the indexer. As a string, takes the form of # "[type]:[id]@v[version]", such as "Widget:123abc@v7". This format was designed to make it - # easy to put these ids in a comma-seperated list. + # easy to put these ids in a comma-separated list. EventID = ::Data.define(:type, :id, :version) do # @implements EventID def self.from_event(event) @@ -26,7 +26,7 @@ def to_s # Steep weirdly expects them here... # @dynamic initialize, config, datastore_core, schema_artifacts, datastore_router, monotonic_clock - # @dynamic record_preparer_factory, processor, operation_factory, logger + # @dynamic record_preparer_factory, processor, operation_factory, indexing_event_decoder, logger # @dynamic self.from_parsed_yaml end end diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/indexing_event_decoder.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/indexing_event_decoder.rb new file mode 100644 index 000000000..bfa888808 --- /dev/null +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/indexing_event_decoder.rb @@ -0,0 +1,41 @@ +# Copyright 2024 - 2026 Block, Inc. +# +# Use of this source code is governed by an MIT-style +# license that can be found in the LICENSE file or at +# https://opensource.org/licenses/MIT. +# +# frozen_string_literal: true + +require "json" + +module ElasticGraph + class Indexer + module IndexingEventDecoder + # Defines the extension interface implemented by indexing event decoders. + # + # @api private + class Interface + # :nocov: + def initialize(config:, schema_artifacts:, logger:) + end + + def decode(payload) + [] + end + # :nocov: + end + + # The default indexing event decoder, which expects newline-delimited JSON objects. + # + # @api private + class JSONLines + def initialize(config:, schema_artifacts:, logger:) + end + + def decode(payload) + payload.split("\n").map { |event| JSON.parse(event) } + end + end + end + end +end diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer.rbs index 6f1d1bc40..e516c6eba 100644 --- a/elasticgraph-indexer/sig/elastic_graph/indexer.rbs +++ b/elasticgraph-indexer/sig/elastic_graph/indexer.rbs @@ -28,6 +28,9 @@ module ElasticGraph @operation_factory: Operation::Factory? def operation_factory: () -> Operation::Factory + @indexing_event_decoder: indexingEventDecoder? + def indexing_event_decoder: () -> indexingEventDecoder + @monotonic_clock: Support::MonotonicClock? def monotonic_clock: () -> Support::MonotonicClock diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer/config.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer/config.rbs index 1228afd4d..a33c4b567 100644 --- a/elasticgraph-indexer/sig/elastic_graph/indexer/config.rbs +++ b/elasticgraph-indexer/sig/elastic_graph/indexer/config.rbs @@ -5,25 +5,33 @@ module ElasticGraph attr_reader latency_slo_thresholds_by_timestamp_in_ms: ::Hash[::String, ::Integer] attr_reader skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]] + attr_reader indexing_event_decoder: SchemaArtifacts::RuntimeMetadata::Extension def initialize: ( ?latency_slo_thresholds_by_timestamp_in_ms: ::Hash[::String, ::Integer], - ?skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]]) -> void + ?skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]], + ?indexing_event_decoder: ::Hash[::String, untyped]) -> void def with: ( ?latency_slo_thresholds_by_timestamp_in_ms: ::Hash[::String, ::Integer], - ?skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]]) -> Config + ?skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]], + ?indexing_event_decoder: SchemaArtifacts::RuntimeMetadata::Extension) -> Config def self.members: () -> ::Array[::Symbol] end class Config < ConfigSupertype + DEFAULT_INDEXING_EVENT_DECODER: ::Hash[::String, untyped] + private def convert_values: ( latency_slo_thresholds_by_timestamp_in_ms: untyped, - skip_derived_indexing_type_updates: untyped + skip_derived_indexing_type_updates: untyped, + indexing_event_decoder: untyped ) -> ::Hash[::Symbol, untyped] + + def load_indexing_event_decoder: (::Hash[::String, untyped]) -> SchemaArtifacts::RuntimeMetadata::Extension end end end diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer/indexing_event_decoder.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer/indexing_event_decoder.rbs new file mode 100644 index 000000000..7c586b8ef --- /dev/null +++ b/elasticgraph-indexer/sig/elastic_graph/indexer/indexing_event_decoder.rbs @@ -0,0 +1,20 @@ +module ElasticGraph + class Indexer + type indexingEventDecoder = IndexingEventDecoder::Interface + + module IndexingEventDecoder + class Interface + def initialize: ( + config: ::Hash[::Symbol | ::String, untyped], + schema_artifacts: schemaArtifacts?, + logger: ::Logger + ) -> void + + def decode: (::String) -> ::Array[event] + end + + class JSONLines < Interface + end + end + end +end diff --git a/elasticgraph-indexer/spec/support/example_extensions/indexing_event_decoder.rb b/elasticgraph-indexer/spec/support/example_extensions/indexing_event_decoder.rb new file mode 100644 index 000000000..b9c777ce6 --- /dev/null +++ b/elasticgraph-indexer/spec/support/example_extensions/indexing_event_decoder.rb @@ -0,0 +1,26 @@ +# Copyright 2024 - 2026 Block, Inc. +# +# Use of this source code is governed by an MIT-style +# license that can be found in the LICENSE file or at +# https://opensource.org/licenses/MIT. +# +# frozen_string_literal: true + +class ExampleIndexingEventDecoder + attr_reader :config, :schema_artifacts, :logger + + def initialize(config:, schema_artifacts:, logger:) + @config = config + @schema_artifacts = schema_artifacts + @logger = logger + end + + def decode(payload) + payload.split(config.fetch("delimiter")).map { |value| {"value" => value} } + end +end + +class InvalidIndexingEventDecoder + def initialize(config:, schema_artifacts:, logger:) + end +end diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/config_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/config_spec.rb index 072d43707..72f591ec6 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/config_spec.rb +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/config_spec.rb @@ -31,6 +31,34 @@ class Indexer expect(config.skip_derived_indexing_type_updates).to eq("WidgetCurrency" => ["USD"].to_set) end + + it "uses the JSON Lines indexing event decoder by default" do + expect(Config.new.indexing_event_decoder.extension_class).to be(IndexingEventDecoder::JSONLines) + end + + it "loads a configured indexing event decoder" do + config = Config.from_parsed_yaml("indexer" => { + "indexing_event_decoder" => { + "name" => "ExampleIndexingEventDecoder", + "require_path" => "support/example_extensions/indexing_event_decoder", + "config" => {"delimiter" => "|"} + } + }) + + expect(config.indexing_event_decoder.extension_class).to be(ExampleIndexingEventDecoder) + expect(config.indexing_event_decoder.config).to eq({"delimiter" => "|"}) + end + + it "verifies that a configured indexing event decoder implements the expected interface" do + expect { + Config.from_parsed_yaml("indexer" => { + "indexing_event_decoder" => { + "name" => "InvalidIndexingEventDecoder", + "require_path" => "support/example_extensions/indexing_event_decoder" + } + }) + }.to raise_error Errors::InvalidExtensionError, a_string_including("InvalidIndexingEventDecoder", "Missing instance methods: `decode`") + end end end end diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/indexing_event_decoder_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/indexing_event_decoder_spec.rb new file mode 100644 index 000000000..13e0ad93a --- /dev/null +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/indexing_event_decoder_spec.rb @@ -0,0 +1,28 @@ +# Copyright 2024 - 2026 Block, Inc. +# +# Use of this source code is governed by an MIT-style +# license that can be found in the LICENSE file or at +# https://opensource.org/licenses/MIT. +# +# frozen_string_literal: true + +require "elastic_graph/indexer/indexing_event_decoder" + +module ElasticGraph + class Indexer + RSpec.describe IndexingEventDecoder::JSONLines do + it "decodes newline-delimited JSON objects" do + decoder = described_class.new(config: {}, schema_artifacts: nil, logger: nil) + payload = <<~JSONL + {"op":"upsert","id":"1"} + {"op":"upsert","id":"2"} + JSONL + + expect(decoder.decode(payload)).to eq([ + {"op" => "upsert", "id" => "1"}, + {"op" => "upsert", "id" => "2"} + ]) + end + end + end +end diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer_spec.rb index abd32bc88..42e5b2f95 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer_spec.rb +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer_spec.rb @@ -28,6 +28,28 @@ module ElasticGraph expect(indexer).to be_a(Indexer) end + + it "builds a configured indexing event decoder" do + config = Indexer::Config.from_parsed_yaml("indexer" => { + "indexing_event_decoder" => { + "name" => "ExampleIndexingEventDecoder", + "require_path" => "support/example_extensions/indexing_event_decoder", + "config" => {"delimiter" => "|"} + } + }) + indexer = Indexer.new(config: config, datastore_core: build_datastore_core) + + decoder = indexer.indexing_event_decoder + + expect(decoder).to be_a(ExampleIndexingEventDecoder) + expect(decoder.config).to eq({"delimiter" => "|"}) + expect(decoder.schema_artifacts).to be(indexer.schema_artifacts) + expect(decoder.logger).to be(indexer.logger) + expect(decoder.decode("one|two")).to eq([ + {"value" => "one"}, + {"value" => "two"} + ]) + end end end end diff --git a/elasticgraph-indexer_lambda/lib/elastic_graph/indexer_lambda/lambda_function.rb b/elasticgraph-indexer_lambda/lib/elastic_graph/indexer_lambda/lambda_function.rb index b5ea6b4d0..b90f48d6a 100644 --- a/elasticgraph-indexer_lambda/lib/elastic_graph/indexer_lambda/lambda_function.rb +++ b/elasticgraph-indexer_lambda/lib/elastic_graph/indexer_lambda/lambda_function.rb @@ -27,6 +27,7 @@ def initialize @sqs_processor = ElasticGraph::IndexerLambda::SqsProcessor.new( indexer.processor, + indexing_event_decoder: indexer.indexing_event_decoder, ignore_sqs_latency_timestamps_from_arns: ignore_sqs_latency_timestamps_from_arns, logger: indexer.logger ) 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 635b6dd0c..f360ab0f2 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 @@ -19,9 +19,10 @@ class SqsProcessor # @dynamic ignore_sqs_latency_timestamps_from_arns attr_reader :ignore_sqs_latency_timestamps_from_arns - def initialize(indexer_processor, logger:, ignore_sqs_latency_timestamps_from_arns:, s3_client: nil) + def initialize(indexer_processor, logger:, ignore_sqs_latency_timestamps_from_arns:, indexing_event_decoder: nil, s3_client: nil) @indexer_processor = indexer_processor @logger = logger + @indexing_event_decoder = indexing_event_decoder || default_indexing_event_decoder @s3_client = s3_client @ignore_sqs_latency_timestamps_from_arns = ignore_sqs_latency_timestamps_from_arns end @@ -41,32 +42,24 @@ def process(lambda_event, refresh_indices: false) private - # Given a lambda event payload, returns an array of raw operations in JSON format. + # Given a lambda event payload, returns an array of raw ElasticGraph indexing events. # # The SQS payload is wrapped in the following format already: # See https://docs.aws.amazon.com/lambda/latest/dg/with-sqs.html#example-standard-queue-message-event for more details # { - # Records: { - # [ - # { body: }, - # { body: }, - # ... - # ] - # } + # "Records" => [ + # {"body" => }, + # {"body" => }, + # ... + # ] # } # - # Each entry in "Records" is a SQS entry. Since lambda handles some batching + # Each entry in "Records" is a SQS entry. Since lambda handles some batching # for you (with some limits), you can get multiple. # # We also want to do our own batching in order to cram more into a given payload - # and issue fewer SQS entries and Lambda invocations when possible. As such, we - # encoded multiple JSON with JSON Lines (http://jsonlines.org/) in record body. - # Each JSON Lines object under 'body' should be of the form: - # - # {op: 'upsert', __typename: 'Payment', id: "123", version: "1", record: {...} } \n - # {op: 'upsert', __typename: 'Payment', id: "123", version: "2", record: {...} } \n - # ... - # Note: "\n" at the end of each line is a single byte newline control character, instead of a string sequence + # and issue fewer SQS entries and Lambda invocations when possible. The default decoder expects + # JSON Lines (http://jsonlines.org/), but custom decoders can support other payload formats. def events_from(lambda_event) sqs_received_at_by_message_id = {} # : Hash[String, String] lambda_event.fetch("Records").flat_map do |record| @@ -80,7 +73,7 @@ def events_from(lambda_event) sqs_metadata = sqs_metadata.except("latency_timestamps") end - parse_jsonl(record.fetch("body")).map do |event| + decoded_events_from(record.fetch("body")).map do |event| ElasticGraph::Support::HashUtil.deep_merge(event, sqs_metadata) end end.tap do @@ -93,11 +86,22 @@ def events_from(lambda_event) S3_OFFLOADING_INDICATOR = '["software.amazon.payloadoffloading.PayloadS3Pointer"' - def parse_jsonl(jsonl_string) - if jsonl_string.start_with?(S3_OFFLOADING_INDICATOR) - jsonl_string = get_payload_from_s3(jsonl_string) + def decoded_events_from(payload) + if payload.start_with?(S3_OFFLOADING_INDICATOR) + payload = get_payload_from_s3(payload) end - jsonl_string.split("\n").map { |event| JSON.parse(event) } + + @indexing_event_decoder.decode(payload) + end + + def default_indexing_event_decoder + require "elastic_graph/indexer/indexing_event_decoder" + + Indexer::IndexingEventDecoder::JSONLines.new( + config: {}, + schema_artifacts: nil, + logger: @logger + ) 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 65df06668..fc1661a74 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 @@ -5,6 +5,7 @@ module ElasticGraph Indexer::Processor, logger: ::Logger, ignore_sqs_latency_timestamps_from_arns: ::Set[::String], + ?indexing_event_decoder: Indexer::indexingEventDecoder?, ?s3_client: Aws::S3::Client?, ) -> void @@ -14,6 +15,7 @@ module ElasticGraph @indexer_processor: Indexer::Processor @logger: ::Logger + @indexing_event_decoder: Indexer::indexingEventDecoder @s3_client: Aws::S3::Client? attr_reader ignore_sqs_latency_timestamps_from_arns: ::Set[::String] @@ -22,7 +24,8 @@ module ElasticGraph S3_OFFLOADING_INDICATOR: String def extract_sqs_metadata: (::Hash[String, untyped]) -> ::Hash[::String, untyped] def millis_to_iso8601: (::String) -> ::String? - def parse_jsonl: (::String) -> ::Array[::Hash[::String, untyped]] + def decoded_events_from: (::String) -> ::Array[::Hash[::String, untyped]] + def default_indexing_event_decoder: () -> Indexer::indexingEventDecoder 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 eb1343e5d..33e48a9e2 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 @@ -8,6 +8,7 @@ require "elastic_graph/errors" require "elastic_graph/indexer/failed_event_error" +require "elastic_graph/indexer/indexing_event_decoder" require "elastic_graph/indexer/processor" require "elastic_graph/indexer_lambda/sqs_processor" require "elastic_graph/spec_support/lambda_function" @@ -75,6 +76,22 @@ module IndexerLambda ], refresh_indices: false) end + it "uses the configured indexing event decoder" do + custom_decoder = instance_spy(Indexer::IndexingEventDecoder::Interface, decode: [{"field1" => {}}]) + lambda_event = { + "Records" => [ + sqs_message("a", "not-json") + ] + } + + build_sqs_processor(indexing_event_decoder: custom_decoder).process(lambda_event) + + expect(custom_decoder).to have_received(:decode).with("not-json") + expect(indexer_processor).to have_received(:process_returning_failures).with([ + {"field1" => {}, "message_id" => "a"} + ], 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" @@ -377,9 +394,12 @@ def jsonl(*items) end def build_sqs_processor(**options) + event_decoder = options.delete(:indexing_event_decoder) { nil } + SqsProcessor.new( indexer_processor, logger: logger, + indexing_event_decoder: event_decoder, ignore_sqs_latency_timestamps_from_arns: ignore_sqs_latency_timestamps_from_arns, **options ) diff --git a/elasticgraph-local/lib/elastic_graph/local/spec_support/config_schema.yaml b/elasticgraph-local/lib/elastic_graph/local/spec_support/config_schema.yaml index b42bfc9f5..2239638fd 100644 --- a/elasticgraph-local/lib/elastic_graph/local/spec_support/config_schema.yaml +++ b/elasticgraph-local/lib/elastic_graph/local/spec_support/config_schema.yaml @@ -508,6 +508,45 @@ properties: - {} - WidgetWorkspace: - ABC12345678 + indexing_event_decoder: + description: Extension object used to decode raw indexing payloads into ElasticGraph + indexing event hashes. The default decoder expects JSON Lines. + type: object + properties: + name: + description: The name of the indexing event decoder extension class. + type: string + minLength: 1 + examples: + - MyCompany::ElasticGraph::CSVIndexingEventDecoder + require_path: + description: The path to require to load the indexing event decoder extension. + type: string + minLength: 1 + examples: + - "./lib/my_company/elastic_graph/csv_indexing_event_decoder" + config: + description: Configuration for the indexing event decoder. Will be passed + into the decoder's `#initialize` method. + type: object + default: {} + examples: + - {} + - delimiter: "," + required: + - name + - require_path + default: + name: ElasticGraph::Indexer::IndexingEventDecoder::JSONLines + require_path: elastic_graph/indexer/indexing_event_decoder + examples: + - name: ElasticGraph::Indexer::IndexingEventDecoder::JSONLines + require_path: elastic_graph/indexer/indexing_event_decoder + - name: MyCompany::ElasticGraph::CSVIndexingEventDecoder + require_path: "./lib/my_company/elastic_graph/csv_indexing_event_decoder" + config: + delimiter: "," + additionalProperties: false additionalProperties: false logger: description: Configuration for logging used by all parts of ElasticGraph. From cda0bae057281fa6bf2be6c45fe6334c7027c991 Mon Sep 17 00:00:00 2001 From: Josh Wilson Date: Tue, 9 Jun 2026 14:11:34 -0500 Subject: [PATCH 2/5] Address review feedback on indexing event decoder PR - Make `indexing_event_decoder` a required `SqsProcessor` kwarg instead of a nilable one with a second, duplicate default-construction path. The warehouse lambda now passes its indexer's configured decoder, so there is a single source of truth for the default and decoders can rely on receiving non-nil `schema_artifacts`. - Make `JSONLines` inherit from `Interface` so its RBS superclass declaration matches the runtime class. - Validate the decoder's `name` config with the same class-name pattern used for query interceptors, and regenerate the config schema artifact. - Match the established extension-interface style (explanatory comments, `:nocov:` only around the body that must return a value) and document the interface publicly since it is the contract decoder authors implement. - Replace the `_ =` cast with an inline type annotation, alphabetize requires, and simplify the SQS processor spec helper. --- .../lib/elastic_graph/indexer.rb | 3 ++- .../lib/elastic_graph/indexer/config.rb | 4 ++-- .../indexer/indexing_event_decoder.rb | 24 ++++++++++++------- .../indexer/indexing_event_decoder.rbs | 2 +- .../indexer/indexing_event_decoder_spec.rb | 4 ++-- .../indexer_lambda/sqs_processor.rb | 14 ++--------- .../indexer_lambda/sqs_processor.rbs | 3 +-- .../indexer_lambda/sqs_processor_spec.rb | 8 +++++-- .../local/spec_support/config_schema.yaml | 2 +- .../warehouse_lambda/lambda_function.rb | 1 + 10 files changed, 34 insertions(+), 31 deletions(-) diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer.rb b/elasticgraph-indexer/lib/elastic_graph/indexer.rb index a5fbea39a..da4d0ad6c 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer.rb @@ -89,7 +89,8 @@ def operation_factory def indexing_event_decoder @indexing_event_decoder ||= begin extension = config.indexing_event_decoder - (_ = extension.extension_class).new( + decoder_class = extension.extension_class # : untyped + decoder_class.new( config: extension.config, schema_artifacts: schema_artifacts, logger: logger diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb index ad9fa354e..2b0146b03 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb @@ -6,10 +6,10 @@ # # frozen_string_literal: true -require "elastic_graph/support/config" require "elastic_graph/errors" require "elastic_graph/indexer/indexing_event_decoder" require "elastic_graph/schema_artifacts/runtime_metadata/extension_loader" +require "elastic_graph/support/config" module ElasticGraph class Indexer @@ -57,7 +57,7 @@ class Config < Support::Config.define(:latency_slo_thresholds_by_timestamp_in_ms name: { description: "The name of the indexing event decoder extension class.", type: "string", - minLength: 1, + pattern: /^[A-Z]\w+(::[A-Z]\w+)*$/.source, # https://rubular.com/r/UuqAz4fR3kdMip examples: ["MyCompany::ElasticGraph::CSVIndexingEventDecoder"] }, require_path: { 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 bfa888808..2fe8b9162 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/indexing_event_decoder.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/indexing_event_decoder.rb @@ -10,28 +10,36 @@ 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. module IndexingEventDecoder - # Defines the extension interface implemented by indexing event decoders. - # - # @api private + # Defines the indexing event decoder interface, which our extension loader will validate against. class Interface - # :nocov: + # @param config [Hash] configuration from the `indexing_event_decoder.config` setting + # @param schema_artifacts [SchemaArtifacts::FromDisk] the schema artifacts + # @param logger [Logger] the ElasticGraph logger def initialize(config:, schema_artifacts:, logger:) + # must be defined, but nothing to do end + # @param payload [String] a raw payload from the transport + # @return [Array>] the decoded ElasticGraph indexing events def decode(payload) + # :nocov: -- must return an array to satisfy Steep type checking but never called [] + # :nocov: end - # :nocov: end # The default indexing event decoder, which expects newline-delimited JSON objects. - # - # @api private - class JSONLines + class JSONLines < Interface + # (see Interface#initialize) def initialize(config:, schema_artifacts:, logger:) + # must be defined for extension interface verification, but nothing to do end + # (see Interface#decode) def decode(payload) payload.split("\n").map { |event| JSON.parse(event) } end 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 7c586b8ef..2aba3cc05 100644 --- a/elasticgraph-indexer/sig/elastic_graph/indexer/indexing_event_decoder.rbs +++ b/elasticgraph-indexer/sig/elastic_graph/indexer/indexing_event_decoder.rbs @@ -6,7 +6,7 @@ module ElasticGraph class Interface def initialize: ( config: ::Hash[::Symbol | ::String, untyped], - schema_artifacts: schemaArtifacts?, + schema_artifacts: schemaArtifacts, logger: ::Logger ) -> void diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/indexing_event_decoder_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/indexing_event_decoder_spec.rb index 13e0ad93a..14b3674a5 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/indexing_event_decoder_spec.rb +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/indexing_event_decoder_spec.rb @@ -10,9 +10,9 @@ module ElasticGraph class Indexer - RSpec.describe IndexingEventDecoder::JSONLines do + RSpec.describe IndexingEventDecoder::JSONLines, :capture_logs do it "decodes newline-delimited JSON objects" do - decoder = described_class.new(config: {}, schema_artifacts: nil, logger: nil) + decoder = described_class.new(config: {}, schema_artifacts: nil, logger: logger) payload = <<~JSONL {"op":"upsert","id":"1"} {"op":"upsert","id":"2"} 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 f360ab0f2..45144fede 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 @@ -19,10 +19,10 @@ class SqsProcessor # @dynamic ignore_sqs_latency_timestamps_from_arns attr_reader :ignore_sqs_latency_timestamps_from_arns - def initialize(indexer_processor, logger:, ignore_sqs_latency_timestamps_from_arns:, indexing_event_decoder: nil, s3_client: nil) + def initialize(indexer_processor, logger:, ignore_sqs_latency_timestamps_from_arns:, indexing_event_decoder:, s3_client: nil) @indexer_processor = indexer_processor @logger = logger - @indexing_event_decoder = indexing_event_decoder || default_indexing_event_decoder + @indexing_event_decoder = indexing_event_decoder @s3_client = s3_client @ignore_sqs_latency_timestamps_from_arns = ignore_sqs_latency_timestamps_from_arns end @@ -94,16 +94,6 @@ def decoded_events_from(payload) @indexing_event_decoder.decode(payload) end - def default_indexing_event_decoder - require "elastic_graph/indexer/indexing_event_decoder" - - Indexer::IndexingEventDecoder::JSONLines.new( - config: {}, - schema_artifacts: nil, - logger: @logger - ) - end - def extract_sqs_metadata(record) sqs_timestamps = { "processing_first_attempted_at" => millis_to_iso8601(record.dig("attributes", "ApproximateFirstReceiveTimestamp")), 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 fc1661a74..9c34781fe 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 @@ -5,7 +5,7 @@ module ElasticGraph Indexer::Processor, logger: ::Logger, ignore_sqs_latency_timestamps_from_arns: ::Set[::String], - ?indexing_event_decoder: Indexer::indexingEventDecoder?, + indexing_event_decoder: Indexer::indexingEventDecoder, ?s3_client: Aws::S3::Client?, ) -> void @@ -25,7 +25,6 @@ module ElasticGraph 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 default_indexing_event_decoder: () -> Indexer::indexingEventDecoder 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 33e48a9e2..9e34eedfd 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 @@ -394,12 +394,16 @@ def jsonl(*items) end def build_sqs_processor(**options) - event_decoder = options.delete(:indexing_event_decoder) { nil } + json_lines_decoder = Indexer::IndexingEventDecoder::JSONLines.new( + config: {}, + schema_artifacts: nil, # not used by `JSONLines` + logger: logger + ) SqsProcessor.new( indexer_processor, logger: logger, - indexing_event_decoder: event_decoder, + indexing_event_decoder: json_lines_decoder, ignore_sqs_latency_timestamps_from_arns: ignore_sqs_latency_timestamps_from_arns, **options ) diff --git a/elasticgraph-local/lib/elastic_graph/local/spec_support/config_schema.yaml b/elasticgraph-local/lib/elastic_graph/local/spec_support/config_schema.yaml index 2239638fd..139ca1cf5 100644 --- a/elasticgraph-local/lib/elastic_graph/local/spec_support/config_schema.yaml +++ b/elasticgraph-local/lib/elastic_graph/local/spec_support/config_schema.yaml @@ -516,7 +516,7 @@ properties: name: description: The name of the indexing event decoder extension class. type: string - minLength: 1 + pattern: "^[A-Z]\\w+(::[A-Z]\\w+)*$" examples: - MyCompany::ElasticGraph::CSVIndexingEventDecoder require_path: diff --git a/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/lambda_function.rb b/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/lambda_function.rb index 40e81863d..d58479f39 100644 --- a/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/lambda_function.rb +++ b/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/lambda_function.rb @@ -28,6 +28,7 @@ def initialize @sqs_processor = IndexerLambda::SqsProcessor.new( warehouse_lambda.processor, ignore_sqs_latency_timestamps_from_arns: ignore_sqs_latency_timestamps_from_arns, + indexing_event_decoder: warehouse_lambda.indexer.indexing_event_decoder, logger: warehouse_lambda.logger ) end From aee63156ca39b550540f151a92e64bb592eb6383 Mon Sep 17 00:00:00 2001 From: Josh Wilson Date: Tue, 7 Jul 2026 10:51:46 -0500 Subject: [PATCH 3/5] Polish indexing event decoder changes --- elasticgraph-indexer/README.md | 24 +++++++++++++++---- .../indexer/indexing_event_decoder_spec.rb | 4 ++-- .../spec/unit/elastic_graph/indexer_spec.rb | 4 +++- .../warehouse_lambda/lambda_function.rb | 2 +- 4 files changed, 25 insertions(+), 9 deletions(-) diff --git a/elasticgraph-indexer/README.md b/elasticgraph-indexer/README.md index 52a3be8c9..3b3a66a56 100644 --- a/elasticgraph-indexer/README.md +++ b/elasticgraph-indexer/README.md @@ -64,13 +64,27 @@ indexer: delimiter: "," ``` -Decoder extensions must implement: +Decoder extensions must implement this interface: ```ruby -def initialize(config:, schema_artifacts:, logger:) -end +# lib/my_company/elastic_graph/csv_indexing_event_decoder.rb +module MyCompany + module ElasticGraph + class CSVIndexingEventDecoder + def initialize(config:, schema_artifacts:, logger:) + # `config` is a hash containing parameterized configuration values from the + # `indexing_event_decoder.config` setting (see above for an example). + # + # `schema_artifacts` provides access to the schema artifacts, in case decoding + # depends on the schema. + # + # `logger` is the ElasticGraph logger. + end -def decode(payload) - # return an array of ElasticGraph indexing event hashes + def decode(payload) + # Must return an array of ElasticGraph indexing event hashes. + end + end + end end ``` diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/indexing_event_decoder_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/indexing_event_decoder_spec.rb index 14b3674a5..6c3aee792 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/indexing_event_decoder_spec.rb +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/indexing_event_decoder_spec.rb @@ -10,9 +10,9 @@ module ElasticGraph class Indexer - RSpec.describe IndexingEventDecoder::JSONLines, :capture_logs do + RSpec.describe IndexingEventDecoder::JSONLines do it "decodes newline-delimited JSON objects" do - decoder = described_class.new(config: {}, schema_artifacts: nil, logger: logger) + decoder = described_class.new(config: {}, schema_artifacts: nil, logger: nil) # args are not used by `JSONLines` payload = <<~JSONL {"op":"upsert","id":"1"} {"op":"upsert","id":"2"} diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer_spec.rb index 42e5b2f95..330855c0e 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer_spec.rb +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer_spec.rb @@ -28,8 +28,10 @@ module ElasticGraph expect(indexer).to be_a(Indexer) end + end - it "builds a configured indexing event decoder" do + describe "#indexing_event_decoder" do + it "builds the configured indexing event decoder" do config = Indexer::Config.from_parsed_yaml("indexer" => { "indexing_event_decoder" => { "name" => "ExampleIndexingEventDecoder", diff --git a/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/lambda_function.rb b/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/lambda_function.rb index d58479f39..4c4919a41 100644 --- a/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/lambda_function.rb +++ b/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/lambda_function.rb @@ -27,8 +27,8 @@ def initialize @sqs_processor = IndexerLambda::SqsProcessor.new( warehouse_lambda.processor, - ignore_sqs_latency_timestamps_from_arns: ignore_sqs_latency_timestamps_from_arns, indexing_event_decoder: warehouse_lambda.indexer.indexing_event_decoder, + ignore_sqs_latency_timestamps_from_arns: ignore_sqs_latency_timestamps_from_arns, logger: warehouse_lambda.logger ) end From 4416d0959960822b15b2dd7bc2227c5be558a437 Mon Sep 17 00:00:00 2001 From: Josh Wilson Date: Tue, 7 Jul 2026 13:28:00 -0500 Subject: [PATCH 4/5] Move JSON Lines decoding to elasticgraph-json_ingestion with no indexer default --- config/settings/test.yaml.template | 3 ++ elasticgraph-indexer/README.md | 9 ++-- .../lib/elastic_graph/indexer.rb | 8 ++++ .../lib/elastic_graph/indexer/config.rb | 16 ++++--- .../indexer/indexing_event_decoder.rb | 15 ------- .../sig/elastic_graph/indexer/config.rbs | 10 ++--- .../indexer/indexing_event_decoder.rbs | 3 -- .../unit/elastic_graph/indexer/config_spec.rb | 4 +- .../indexer/indexing_event_decoder_spec.rb | 28 ------------- .../spec/unit/elastic_graph/indexer_spec.rb | 26 ++++++++---- elasticgraph-indexer_lambda/README.md | 6 ++- .../elasticgraph-indexer_lambda.gemspec | 5 +++ .../indexer_lambda/sqs_processor.rb | 4 +- .../indexer_lambda/sqs_processor_spec.rb | 5 ++- elasticgraph-json_ingestion/README.md | 15 +++++++ .../elasticgraph-json_ingestion.gemspec | 3 ++ .../json_ingestion/indexing_event_decoder.rb | 31 ++++++++++++++ .../json_ingestion/indexing_event_decoder.rbs | 13 ++++++ .../indexing_event_decoder_spec.rb | 42 +++++++++++++++++++ .../local/spec_support/config_schema.yaml | 12 +++--- .../elasticgraph-warehouse_lambda.gemspec | 4 ++ .../config/settings/local.yaml.tt | 3 ++ .../spec_support/builds_indexer.rb | 4 +- 23 files changed, 181 insertions(+), 88 deletions(-) delete mode 100644 elasticgraph-indexer/spec/unit/elastic_graph/indexer/indexing_event_decoder_spec.rb create mode 100644 elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexing_event_decoder.rb create mode 100644 elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/indexing_event_decoder.rbs create mode 100644 elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/indexing_event_decoder_spec.rb diff --git a/config/settings/test.yaml.template b/config/settings/test.yaml.template index a405b75a9..f07a86539 100644 --- a/config/settings/test.yaml.template +++ b/config/settings/test.yaml.template @@ -87,5 +87,8 @@ logger: device: log/test.log indexer: latency_slo_thresholds_by_timestamp_in_ms: {} + indexing_event_decoder: + name: ElasticGraph::JSONIngestion::IndexingEventDecoder + require_path: elastic_graph/json_ingestion/indexing_event_decoder schema_artifacts: directory: config/schema/artifacts diff --git a/elasticgraph-indexer/README.md b/elasticgraph-indexer/README.md index 3b3a66a56..407441965 100644 --- a/elasticgraph-indexer/README.md +++ b/elasticgraph-indexer/README.md @@ -49,11 +49,12 @@ events = [] # JSON events read from an async datastream indexer.processor.process(events) ``` -## Custom Payload Decoding +## Payload Decoding -`ElasticGraph::Indexer` can be configured with an indexing event decoder extension. Decoders turn raw payload strings -from a transport into ElasticGraph indexing event hashes before the normal validation and indexing pipeline runs. The -default decoder expects JSON Lines. +Ingesting encoded payloads from a transport (such as the SQS lambdas) requires an indexing event decoder extension, +configured via the `indexer.indexing_event_decoder` setting. Decoders turn raw payload strings into ElasticGraph +indexing event hashes before the normal validation and indexing pipeline runs. Ingestion format gems provide decoder +implementations, or you can define your own: ```yaml indexer: diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer.rb b/elasticgraph-indexer/lib/elastic_graph/indexer.rb index da4d0ad6c..8055595d6 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer.rb @@ -7,6 +7,7 @@ # frozen_string_literal: true require "elastic_graph/datastore_core" +require "elastic_graph/errors" require "elastic_graph/indexer/config" require "elastic_graph/support/from_yaml_file" @@ -89,6 +90,13 @@ def operation_factory def indexing_event_decoder @indexing_event_decoder ||= begin extension = config.indexing_event_decoder + + if extension.nil? + raise Errors::ConfigError, "`indexer.indexing_event_decoder` is not configured, but is required to " \ + "decode indexing event payloads. Configure it with a decoder extension provided by an ingestion " \ + "format gem (or one of your own) that understands your transport's payload format." + end + decoder_class = extension.extension_class # : untyped decoder_class.new( config: extension.config, diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb index 2b0146b03..d359963f0 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb @@ -14,11 +14,6 @@ module ElasticGraph class Indexer class Config < Support::Config.define(:latency_slo_thresholds_by_timestamp_in_ms, :skip_derived_indexing_type_updates, :indexing_event_decoder) - DEFAULT_INDEXING_EVENT_DECODER = { - "name" => "ElasticGraph::Indexer::IndexingEventDecoder::JSONLines", - "require_path" => "elastic_graph/indexer/indexing_event_decoder" - } - json_schema at: "indexer", optional: false, description: "Configuration for indexing operations and metrics used by `elasticgraph-indexer`.", @@ -51,8 +46,10 @@ class Config < Support::Config.define(:latency_slo_thresholds_by_timestamp_in_ms ] }, indexing_event_decoder: { - description: "Extension object used to decode raw indexing payloads into ElasticGraph indexing event hashes. The default decoder expects JSON Lines.", - type: "object", + description: "Extension object used to decode raw indexing payloads into ElasticGraph indexing event hashes. " \ + "Required when using a transport that delivers encoded payloads (such as the SQS lambdas). Ingestion " \ + "format gems provide decoder implementations.", + type: ["object", "null"], properties: { name: { description: "The name of the indexing event decoder extension class.", @@ -77,9 +74,8 @@ class Config < Support::Config.define(:latency_slo_thresholds_by_timestamp_in_ms } }, required: ["name", "require_path"], - default: DEFAULT_INDEXING_EVENT_DECODER, + default: nil, examples: [ - DEFAULT_INDEXING_EVENT_DECODER, { "name" => "MyCompany::ElasticGraph::CSVIndexingEventDecoder", "require_path" => "./lib/my_company/elastic_graph/csv_indexing_event_decoder", @@ -100,6 +96,8 @@ def convert_values(skip_derived_indexing_type_updates:, latency_slo_thresholds_b end def load_indexing_event_decoder(config) + return nil if config.nil? + loader = SchemaArtifacts::RuntimeMetadata::ExtensionLoader.new(IndexingEventDecoder::Interface) loader.load( config.fetch("name"), 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 2fe8b9162..4f0bbb018 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/indexing_event_decoder.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/indexing_event_decoder.rb @@ -6,8 +6,6 @@ # # frozen_string_literal: true -require "json" - module ElasticGraph class Indexer # Namespace for indexing event decoders, which turn raw payload strings from a transport into @@ -31,19 +29,6 @@ def decode(payload) # :nocov: end end - - # The default indexing event decoder, which expects newline-delimited JSON objects. - class JSONLines < Interface - # (see Interface#initialize) - def initialize(config:, schema_artifacts:, logger:) - # must be defined for extension interface verification, but nothing to do - end - - # (see Interface#decode) - def decode(payload) - payload.split("\n").map { |event| JSON.parse(event) } - end - end end end end diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer/config.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer/config.rbs index a33c4b567..4d946c684 100644 --- a/elasticgraph-indexer/sig/elastic_graph/indexer/config.rbs +++ b/elasticgraph-indexer/sig/elastic_graph/indexer/config.rbs @@ -5,24 +5,22 @@ module ElasticGraph attr_reader latency_slo_thresholds_by_timestamp_in_ms: ::Hash[::String, ::Integer] attr_reader skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]] - attr_reader indexing_event_decoder: SchemaArtifacts::RuntimeMetadata::Extension + attr_reader indexing_event_decoder: SchemaArtifacts::RuntimeMetadata::Extension? def initialize: ( ?latency_slo_thresholds_by_timestamp_in_ms: ::Hash[::String, ::Integer], ?skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]], - ?indexing_event_decoder: ::Hash[::String, untyped]) -> void + ?indexing_event_decoder: ::Hash[::String, untyped]?) -> void def with: ( ?latency_slo_thresholds_by_timestamp_in_ms: ::Hash[::String, ::Integer], ?skip_derived_indexing_type_updates: ::Hash[::String, ::Set[::String]], - ?indexing_event_decoder: SchemaArtifacts::RuntimeMetadata::Extension) -> Config + ?indexing_event_decoder: SchemaArtifacts::RuntimeMetadata::Extension?) -> Config def self.members: () -> ::Array[::Symbol] end class Config < ConfigSupertype - DEFAULT_INDEXING_EVENT_DECODER: ::Hash[::String, untyped] - private def convert_values: ( @@ -31,7 +29,7 @@ module ElasticGraph indexing_event_decoder: untyped ) -> ::Hash[::Symbol, untyped] - def load_indexing_event_decoder: (::Hash[::String, untyped]) -> SchemaArtifacts::RuntimeMetadata::Extension + def load_indexing_event_decoder: (::Hash[::String, untyped]?) -> SchemaArtifacts::RuntimeMetadata::Extension? end end end 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 2aba3cc05..2ada53193 100644 --- a/elasticgraph-indexer/sig/elastic_graph/indexer/indexing_event_decoder.rbs +++ b/elasticgraph-indexer/sig/elastic_graph/indexer/indexing_event_decoder.rbs @@ -12,9 +12,6 @@ module ElasticGraph def decode: (::String) -> ::Array[event] end - - class JSONLines < Interface - end end end end diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/config_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/config_spec.rb index 72f591ec6..6a8ed58bc 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/config_spec.rb +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/config_spec.rb @@ -32,8 +32,8 @@ class Indexer expect(config.skip_derived_indexing_type_updates).to eq("WidgetCurrency" => ["USD"].to_set) end - it "uses the JSON Lines indexing event decoder by default" do - expect(Config.new.indexing_event_decoder.extension_class).to be(IndexingEventDecoder::JSONLines) + it "leaves the indexing event decoder unconfigured by default" do + expect(Config.new.indexing_event_decoder).to be nil end it "loads a configured indexing event decoder" do diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/indexing_event_decoder_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/indexing_event_decoder_spec.rb deleted file mode 100644 index 6c3aee792..000000000 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/indexing_event_decoder_spec.rb +++ /dev/null @@ -1,28 +0,0 @@ -# Copyright 2024 - 2026 Block, Inc. -# -# Use of this source code is governed by an MIT-style -# license that can be found in the LICENSE file or at -# https://opensource.org/licenses/MIT. -# -# frozen_string_literal: true - -require "elastic_graph/indexer/indexing_event_decoder" - -module ElasticGraph - class Indexer - RSpec.describe IndexingEventDecoder::JSONLines do - it "decodes newline-delimited JSON objects" do - decoder = described_class.new(config: {}, schema_artifacts: nil, logger: nil) # args are not used by `JSONLines` - payload = <<~JSONL - {"op":"upsert","id":"1"} - {"op":"upsert","id":"2"} - JSONL - - expect(decoder.decode(payload)).to eq([ - {"op" => "upsert", "id" => "1"}, - {"op" => "upsert", "id" => "2"} - ]) - end - end - end -end diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer_spec.rb index 330855c0e..90d5d7384 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer_spec.rb +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer_spec.rb @@ -11,7 +11,12 @@ module ElasticGraph RSpec.describe Indexer do it "returns non-nil values from each attribute" do - expect_to_return_non_nil_values_from_all_attributes(build_indexer) + indexer = build_indexer(indexing_event_decoder: { + "name" => "ExampleIndexingEventDecoder", + "require_path" => "support/example_extensions/indexing_event_decoder" + }) + + expect_to_return_non_nil_values_from_all_attributes(indexer) end describe ".from_parsed_yaml" do @@ -32,14 +37,11 @@ module ElasticGraph describe "#indexing_event_decoder" do it "builds the configured indexing event decoder" do - config = Indexer::Config.from_parsed_yaml("indexer" => { - "indexing_event_decoder" => { - "name" => "ExampleIndexingEventDecoder", - "require_path" => "support/example_extensions/indexing_event_decoder", - "config" => {"delimiter" => "|"} - } + indexer = build_indexer(indexing_event_decoder: { + "name" => "ExampleIndexingEventDecoder", + "require_path" => "support/example_extensions/indexing_event_decoder", + "config" => {"delimiter" => "|"} }) - indexer = Indexer.new(config: config, datastore_core: build_datastore_core) decoder = indexer.indexing_event_decoder @@ -52,6 +54,14 @@ module ElasticGraph {"value" => "two"} ]) end + + it "raises a clear error when no indexing event decoder is configured" do + indexer = build_indexer + + expect { + indexer.indexing_event_decoder + }.to raise_error Errors::ConfigError, a_string_including("indexer.indexing_event_decoder") + end end end end diff --git a/elasticgraph-indexer_lambda/README.md b/elasticgraph-indexer_lambda/README.md index b9c6a069c..3b6dd5dea 100644 --- a/elasticgraph-indexer_lambda/README.md +++ b/elasticgraph-indexer_lambda/README.md @@ -28,8 +28,10 @@ graph LR; ## SQS Message Payload Format -This gem is designed to run in an AWS lambda that consumes from an SQS queue. Messages in the SQS queue should use -[JSON Lines](https://jsonlines.org/) format to encode indexing events. +This gem is designed to run in an AWS lambda that consumes from an SQS queue. Message payloads in the SQS queue are +decoded by the decoder configured via the `indexer.indexing_event_decoder` setting (see the `elasticgraph-indexer` +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. 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/elasticgraph-indexer_lambda.gemspec b/elasticgraph-indexer_lambda/elasticgraph-indexer_lambda.gemspec index 74252c0f1..c3f3cfa9c 100644 --- a/elasticgraph-indexer_lambda/elasticgraph-indexer_lambda.gemspec +++ b/elasticgraph-indexer_lambda/elasticgraph-indexer_lambda.gemspec @@ -44,4 +44,9 @@ Gem::Specification.new do |spec| spec.add_dependency "elasticgraph-indexer", ElasticGraph::VERSION spec.add_dependency "elasticgraph-lambda_support", ElasticGraph::VERSION spec.add_dependency "aws-sdk-s3", "~> 1.226" + + # The test suite exercises JSON Lines payload decoding via the decoder provided by `elasticgraph-json_ingestion`. + # This is a development dependency (rather than a runtime one) so that applications using a different + # ingestion format can configure their own decoder without bundling `elasticgraph-json_ingestion`. + spec.add_development_dependency "elasticgraph-json_ingestion", ElasticGraph::VERSION end 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 45144fede..a9a910ca2 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 @@ -58,8 +58,8 @@ def process(lambda_event, refresh_indices: false) # for you (with some limits), you can get multiple. # # We also want to do our own batching in order to cram more into a given payload - # and issue fewer SQS entries and Lambda invocations when possible. The default decoder expects - # JSON Lines (http://jsonlines.org/), but custom decoders can support other payload formats. + # and issue fewer SQS entries and Lambda invocations when possible. The configured indexing + # event decoder determines the payload format (e.g. JSON Lines: http://jsonlines.org/). def events_from(lambda_event) sqs_received_at_by_message_id = {} # : Hash[String, String] lambda_event.fetch("Records").flat_map do |record| 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 9e34eedfd..7d341ff91 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 @@ -11,6 +11,7 @@ require "elastic_graph/indexer/indexing_event_decoder" require "elastic_graph/indexer/processor" require "elastic_graph/indexer_lambda/sqs_processor" +require "elastic_graph/json_ingestion/indexing_event_decoder" require "elastic_graph/spec_support/lambda_function" require "json" require "aws-sdk-s3" @@ -394,9 +395,9 @@ def jsonl(*items) end def build_sqs_processor(**options) - json_lines_decoder = Indexer::IndexingEventDecoder::JSONLines.new( + json_lines_decoder = JSONIngestion::IndexingEventDecoder.new( config: {}, - schema_artifacts: nil, # not used by `JSONLines` + schema_artifacts: nil, # not used by the JSON Lines decoder logger: logger ) diff --git a/elasticgraph-json_ingestion/README.md b/elasticgraph-json_ingestion/README.md index 4b32d9cf1..05a135078 100644 --- a/elasticgraph-json_ingestion/README.md +++ b/elasticgraph-json_ingestion/README.md @@ -102,6 +102,21 @@ ElasticGraph.define_schema do |schema| end ``` +## Indexing Event Decoding + +This gem also provides `ElasticGraph::JSONIngestion::IndexingEventDecoder`, which decodes JSON Lines payloads +into ElasticGraph indexing event hashes. Configure it as your indexer's indexing event decoder (generated +ElasticGraph projects include this configuration): + +```yaml +indexer: + indexing_event_decoder: + name: ElasticGraph::JSONIngestion::IndexingEventDecoder + require_path: elastic_graph/json_ingestion/indexing_event_decoder +``` + +See the `elasticgraph-indexer` README for the decoder extension interface. + ## Dependency Diagram ```mermaid diff --git a/elasticgraph-json_ingestion/elasticgraph-json_ingestion.gemspec b/elasticgraph-json_ingestion/elasticgraph-json_ingestion.gemspec index cdade27c0..a6b8ab1b7 100644 --- a/elasticgraph-json_ingestion/elasticgraph-json_ingestion.gemspec +++ b/elasticgraph-json_ingestion/elasticgraph-json_ingestion.gemspec @@ -36,6 +36,9 @@ Gem::Specification.new do |spec| spec.add_dependency "elasticgraph-support", ElasticGraph::VERSION + # The test suite verifies `IndexingEventDecoder` against the decoder interface defined by + # `elasticgraph-indexer`. Keeping this as a development dependency avoids a runtime dependency cycle. + spec.add_development_dependency "elasticgraph-indexer", ElasticGraph::VERSION # This gem's schema-definition extension code references `elasticgraph-schema_definition`, but # applications load it through schema-definition tasks after `elasticgraph-schema_definition` is already # available. Keeping this as a development dependency avoids a runtime dependency cycle. diff --git a/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexing_event_decoder.rb b/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexing_event_decoder.rb new file mode 100644 index 000000000..40dac5253 --- /dev/null +++ b/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexing_event_decoder.rb @@ -0,0 +1,31 @@ +# Copyright 2024 - 2026 Block, Inc. +# +# Use of this source code is governed by an MIT-style +# license that can be found in the LICENSE file or at +# https://opensource.org/licenses/MIT. +# +# frozen_string_literal: true + +require "json" + +module ElasticGraph + module JSONIngestion + # An indexing event decoder for payloads encoded as newline-delimited JSON objects + # ([JSON Lines](https://jsonlines.org/)). Configure it via the `indexer.indexing_event_decoder` + # setting of `elasticgraph-indexer`. + class IndexingEventDecoder + # @param config [Hash] configuration from the `indexing_event_decoder.config` setting + # @param schema_artifacts [SchemaArtifacts::FromDisk] the schema artifacts + # @param logger [Logger] the ElasticGraph logger + def initialize(config:, schema_artifacts:, logger:) + # must be defined for extension interface verification, but nothing to do + end + + # @param payload [String] a raw payload from the transport + # @return [Array>] the decoded ElasticGraph indexing events + def decode(payload) + payload.split("\n").map { |event| JSON.parse(event) } + end + end + end +end diff --git a/elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/indexing_event_decoder.rbs b/elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/indexing_event_decoder.rbs new file mode 100644 index 000000000..9b2700abc --- /dev/null +++ b/elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/indexing_event_decoder.rbs @@ -0,0 +1,13 @@ +module ElasticGraph + module JSONIngestion + class IndexingEventDecoder + def initialize: ( + config: ::Hash[::Symbol | ::String, untyped], + schema_artifacts: schemaArtifacts, + logger: ::Logger + ) -> void + + def decode: (::String) -> ::Array[::Hash[::String, untyped]] + end + end +end diff --git a/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/indexing_event_decoder_spec.rb b/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/indexing_event_decoder_spec.rb new file mode 100644 index 000000000..93c74cba6 --- /dev/null +++ b/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/indexing_event_decoder_spec.rb @@ -0,0 +1,42 @@ +# Copyright 2024 - 2026 Block, Inc. +# +# Use of this source code is governed by an MIT-style +# license that can be found in the LICENSE file or at +# https://opensource.org/licenses/MIT. +# +# frozen_string_literal: true + +require "elastic_graph/indexer/indexing_event_decoder" +require "elastic_graph/json_ingestion/indexing_event_decoder" +require "elastic_graph/schema_artifacts/runtime_metadata/extension_loader" + +module ElasticGraph + module JSONIngestion + RSpec.describe IndexingEventDecoder do + it "decodes newline-delimited JSON objects" do + decoder = IndexingEventDecoder.new(config: {}, schema_artifacts: nil, logger: nil) # args are not used + payload = <<~JSONL + {"op":"upsert","id":"1"} + {"op":"upsert","id":"2"} + JSONL + + expect(decoder.decode(payload)).to eq([ + {"op" => "upsert", "id" => "1"}, + {"op" => "upsert", "id" => "2"} + ]) + end + + it "implements the indexing event decoder interface defined by `elasticgraph-indexer`" do + loader = SchemaArtifacts::RuntimeMetadata::ExtensionLoader.new(Indexer::IndexingEventDecoder::Interface) + + extension = loader.load( + "ElasticGraph::JSONIngestion::IndexingEventDecoder", + from: "elastic_graph/json_ingestion/indexing_event_decoder", + config: {} + ) + + expect(extension.extension_class).to be(IndexingEventDecoder) + end + end + end +end diff --git a/elasticgraph-local/lib/elastic_graph/local/spec_support/config_schema.yaml b/elasticgraph-local/lib/elastic_graph/local/spec_support/config_schema.yaml index 139ca1cf5..82d650d2f 100644 --- a/elasticgraph-local/lib/elastic_graph/local/spec_support/config_schema.yaml +++ b/elasticgraph-local/lib/elastic_graph/local/spec_support/config_schema.yaml @@ -510,8 +510,12 @@ properties: - ABC12345678 indexing_event_decoder: description: Extension object used to decode raw indexing payloads into ElasticGraph - indexing event hashes. The default decoder expects JSON Lines. - type: object + indexing event hashes. Required when using a transport that delivers encoded + payloads (such as the SQS lambdas). Ingestion format gems provide decoder + implementations. + type: + - object + - 'null' properties: name: description: The name of the indexing event decoder extension class. @@ -537,11 +541,7 @@ properties: - name - require_path default: - name: ElasticGraph::Indexer::IndexingEventDecoder::JSONLines - require_path: elastic_graph/indexer/indexing_event_decoder examples: - - name: ElasticGraph::Indexer::IndexingEventDecoder::JSONLines - require_path: elastic_graph/indexer/indexing_event_decoder - name: MyCompany::ElasticGraph::CSVIndexingEventDecoder require_path: "./lib/my_company/elastic_graph/csv_indexing_event_decoder" config: diff --git a/elasticgraph-warehouse_lambda/elasticgraph-warehouse_lambda.gemspec b/elasticgraph-warehouse_lambda/elasticgraph-warehouse_lambda.gemspec index 2c9b7f706..0f12b3d41 100644 --- a/elasticgraph-warehouse_lambda/elasticgraph-warehouse_lambda.gemspec +++ b/elasticgraph-warehouse_lambda/elasticgraph-warehouse_lambda.gemspec @@ -48,5 +48,9 @@ Gem::Specification.new do |spec| spec.add_development_dependency "aws_lambda_ric", "~> 3.2" spec.add_development_dependency "elasticgraph-elasticsearch", ElasticGraph::VERSION + # The test suite exercises JSON Lines payload decoding via the decoder provided by `elasticgraph-json_ingestion`. + # This is a development dependency (rather than a runtime one) so that applications using a different + # ingestion format can configure their own decoder without bundling `elasticgraph-json_ingestion`. + spec.add_development_dependency "elasticgraph-json_ingestion", ElasticGraph::VERSION spec.add_development_dependency "elasticgraph-opensearch", ElasticGraph::VERSION end diff --git a/elasticgraph/lib/elastic_graph/project_template/config/settings/local.yaml.tt b/elasticgraph/lib/elastic_graph/project_template/config/settings/local.yaml.tt index 1bce1f169..014277ead 100644 --- a/elasticgraph/lib/elastic_graph/project_template/config/settings/local.yaml.tt +++ b/elasticgraph/lib/elastic_graph/project_template/config/settings/local.yaml.tt @@ -27,6 +27,9 @@ logger: device: stderr indexer: latency_slo_thresholds_by_timestamp_in_ms: {} + indexing_event_decoder: + name: ElasticGraph::JSONIngestion::IndexingEventDecoder + require_path: elastic_graph/json_ingestion/indexing_event_decoder schema_artifacts: directory: config/schema/artifacts query_registry: diff --git a/spec_support/lib/elastic_graph/spec_support/builds_indexer.rb b/spec_support/lib/elastic_graph/spec_support/builds_indexer.rb index 23fa72092..6cc53a8da 100644 --- a/spec_support/lib/elastic_graph/spec_support/builds_indexer.rb +++ b/spec_support/lib/elastic_graph/spec_support/builds_indexer.rb @@ -18,6 +18,7 @@ def build_indexer( datastore_core: nil, latency_slo_thresholds_by_timestamp_in_ms: {}, skip_derived_indexing_type_updates: {}, + indexing_event_decoder: nil, datastore_router: nil, clock: nil, monotonic_clock: nil, @@ -28,7 +29,8 @@ def build_indexer( datastore_core: datastore_core || build_datastore_core(**datastore_core_options, &customize_datastore_config), config: Indexer::Config.new( latency_slo_thresholds_by_timestamp_in_ms: latency_slo_thresholds_by_timestamp_in_ms, - skip_derived_indexing_type_updates: skip_derived_indexing_type_updates + skip_derived_indexing_type_updates: skip_derived_indexing_type_updates, + indexing_event_decoder: indexing_event_decoder ), datastore_router: datastore_router, clock: clock, From 5f50bb53203f00b016c64082653ed997c3ef2ba6 Mon Sep 17 00:00:00 2001 From: Josh Wilson Date: Tue, 7 Jul 2026 14:02:13 -0500 Subject: [PATCH 5/5] Add elasticgraph-json_ingestion dev dependency to elasticgraph-lambda_support Its test suite builds an ElasticGraph::Indexer from the shared test settings, which now configure the JSON Lines indexing event decoder, so the per-gem CI bundle needs the gem that provides it. --- .../elasticgraph-lambda_support.gemspec | 3 +++ 1 file changed, 3 insertions(+) diff --git a/elasticgraph-lambda_support/elasticgraph-lambda_support.gemspec b/elasticgraph-lambda_support/elasticgraph-lambda_support.gemspec index bcfab1dc9..d9741ec7f 100644 --- a/elasticgraph-lambda_support/elasticgraph-lambda_support.gemspec +++ b/elasticgraph-lambda_support/elasticgraph-lambda_support.gemspec @@ -48,4 +48,7 @@ Gem::Specification.new do |spec| spec.add_development_dependency "elasticgraph-graphql", ElasticGraph::VERSION spec.add_development_dependency "elasticgraph-indexer", ElasticGraph::VERSION spec.add_development_dependency "elasticgraph-indexer_autoscaler_lambda", ElasticGraph::VERSION + # The test suite builds an `ElasticGraph::Indexer` from the shared test settings, which configure + # the indexing event decoder provided by `elasticgraph-json_ingestion`. + spec.add_development_dependency "elasticgraph-json_ingestion", ElasticGraph::VERSION end