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 e4f82753b..407441965 100644 --- a/elasticgraph-indexer/README.md +++ b/elasticgraph-indexer/README.md @@ -48,3 +48,44 @@ indexer = ElasticGraph::Indexer.from_yaml_file("config/settings/local.yaml") events = [] # JSON events read from an async datastream indexer.processor.process(events) ``` + +## Payload Decoding + +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: + indexing_event_decoder: + name: MyCompany::ElasticGraph::CSVIndexingEventDecoder + require_path: ./lib/my_company/elastic_graph/csv_indexing_event_decoder + config: + delimiter: "," +``` + +Decoder extensions must implement this interface: + +```ruby +# 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) + # Must return an array of ElasticGraph indexing event hashes. + end + end + end +end +``` diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer.rb b/elasticgraph-indexer/lib/elastic_graph/indexer.rb index 2d64978bd..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" @@ -86,6 +87,25 @@ def operation_factory end end + 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, + 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..d359963f0 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/config.rb @@ -6,12 +6,14 @@ # # 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 - 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) json_schema at: "indexer", optional: false, description: "Configuration for indexing operations and metrics used by `elasticgraph-indexer`.", @@ -42,17 +44,67 @@ 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. " \ + "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.", + type: "string", + pattern: /^[A-Z]\w+(::[A-Z]\w+)*$/.source, # https://rubular.com/r/UuqAz4fR3kdMip + 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: nil, + examples: [ + { + "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) + return nil if config.nil? + + 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..4f0bbb018 --- /dev/null +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/indexing_event_decoder.rb @@ -0,0 +1,34 @@ +# 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 + +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 indexing event decoder interface, which our extension loader will validate against. + class Interface + # @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 + 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..4d946c684 100644 --- a/elasticgraph-indexer/sig/elastic_graph/indexer/config.rbs +++ b/elasticgraph-indexer/sig/elastic_graph/indexer/config.rbs @@ -5,14 +5,17 @@ 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 @@ -22,8 +25,11 @@ module ElasticGraph 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..2ada53193 --- /dev/null +++ b/elasticgraph-indexer/sig/elastic_graph/indexer/indexing_event_decoder.rbs @@ -0,0 +1,17 @@ +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 + 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..6a8ed58bc 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 "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 + 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_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer_spec.rb index abd32bc88..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 @@ -29,5 +34,34 @@ module ElasticGraph expect(indexer).to be_a(Indexer) end end + + describe "#indexing_event_decoder" do + it "builds the configured indexing event decoder" do + indexer = build_indexer(indexing_event_decoder: { + "name" => "ExampleIndexingEventDecoder", + "require_path" => "support/example_extensions/indexing_event_decoder", + "config" => {"delimiter" => "|"} + }) + + 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 + + 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/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..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 @@ -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:, s3_client: nil) @indexer_processor = indexer_processor @logger = logger + @indexing_event_decoder = 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 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| @@ -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,12 @@ 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 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..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,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,7 @@ 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 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..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 @@ -8,8 +8,10 @@ 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/json_ingestion/indexing_event_decoder" require "elastic_graph/spec_support/lambda_function" require "json" require "aws-sdk-s3" @@ -75,6 +77,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 +395,16 @@ def jsonl(*items) end def build_sqs_processor(**options) + json_lines_decoder = JSONIngestion::IndexingEventDecoder.new( + config: {}, + schema_artifacts: nil, # not used by the JSON Lines decoder + logger: logger + ) + SqsProcessor.new( indexer_processor, logger: logger, + indexing_event_decoder: json_lines_decoder, ignore_sqs_latency_timestamps_from_arns: ignore_sqs_latency_timestamps_from_arns, **options ) 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-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 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..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 @@ -508,6 +508,45 @@ properties: - {} - WidgetWorkspace: - ABC12345678 + indexing_event_decoder: + 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. + type: string + pattern: "^[A-Z]\\w+(::[A-Z]\\w+)*$" + 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: + examples: + - 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. 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-warehouse_lambda/lib/elastic_graph/warehouse_lambda/lambda_function.rb b/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/lambda_function.rb index 40e81863d..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,6 +27,7 @@ def initialize @sqs_processor = IndexerLambda::SqsProcessor.new( warehouse_lambda.processor, + 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 ) 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,