diff --git a/config/settings/test.yaml.template b/config/settings/test.yaml.template index de8557800..0360c0fe5 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 9721a291b..c408e5893 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer.rb @@ -97,6 +97,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 6eca6a902..a259b483b 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/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, :extension_modules) + class Config < Support::Config.define(:latency_slo_thresholds_by_timestamp_in_ms, :skip_derived_indexing_type_updates, :extension_modules, :indexing_event_decoder) json_schema at: "indexer", optional: false, description: "Configuration for indexing operations and metrics used by `elasticgraph-indexer`.", @@ -43,18 +45,68 @@ class Config < Support::Config.define(:latency_slo_thresholds_by_timestamp_in_ms {"WidgetWorkspace" => ["ABC12345678"]} ] }, - extension_modules: Support::Config::EXTENSION_MODULE_SCHEMA + extension_modules: Support::Config::EXTENSION_MODULE_SCHEMA, + 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:, extension_modules:) + def convert_values(skip_derived_indexing_type_updates:, latency_slo_thresholds_by_timestamp_in_ms:, extension_modules:, 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, - extension_modules: SchemaArtifacts::RuntimeMetadata::ExtensionLoader.load_component_extensions(extension_modules) + extension_modules: SchemaArtifacts::RuntimeMetadata::ExtensionLoader.load_component_extensions(extension_modules), + 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 15f78ce4d..f36bd2f86 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/event_id.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/event_id.rb @@ -26,7 +26,7 @@ def to_s # Steep weirdly expects them here... # @dynamic initialize, config, datastore_core, schema_artifacts, datastore_router, monotonic_clock - # @dynamic processor, operation_factory, ingestion_adapters, logger + # @dynamic processor, operation_factory, ingestion_adapters, 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 5b80d8f36..64a9b87b2 100644 --- a/elasticgraph-indexer/sig/elastic_graph/indexer.rbs +++ b/elasticgraph-indexer/sig/elastic_graph/indexer.rbs @@ -27,6 +27,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 a4f343d59..ce4ffface 100644 --- a/elasticgraph-indexer/sig/elastic_graph/indexer/config.rbs +++ b/elasticgraph-indexer/sig/elastic_graph/indexer/config.rbs @@ -6,16 +6,19 @@ 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 extension_modules: ::Array[::Module] + 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]], - ?extension_modules: ::Array[::Module]) -> void + ?extension_modules: ::Array[::Module], + ?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]], - ?extension_modules: ::Array[::Module]) -> Config + ?extension_modules: ::Array[::Module], + ?indexing_event_decoder: SchemaArtifacts::RuntimeMetadata::Extension?) -> Config def self.members: () -> ::Array[::Symbol] end @@ -26,9 +29,11 @@ module ElasticGraph def convert_values: ( latency_slo_thresholds_by_timestamp_in_ms: untyped, skip_derived_indexing_type_updates: untyped, - extension_modules: untyped + extension_modules: 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 2c157a8e9..b2856e51c 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/config_spec.rb +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/config_spec.rb @@ -82,6 +82,34 @@ def extension_modules_from(yaml) Config.from_parsed_yaml("indexer" => ::YAML.safe_load(yaml)).extension_modules end 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 31f440196..e7a67f26d 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 @@ -89,5 +94,34 @@ def define_schema_elements(schema) end 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 11d58ee0a..c3134ce5f 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 39e28e305..559a685ea 100644 --- a/elasticgraph-indexer_lambda/elasticgraph-indexer_lambda.gemspec +++ b/elasticgraph-indexer_lambda/elasticgraph-indexer_lambda.gemspec @@ -46,6 +46,9 @@ Gem::Specification.new do |spec| spec.add_dependency "aws-sdk-s3", "~> 1.228", ">= 1.228.1" # The test suite builds indexers from the shared test schema artifacts, whose runtime metadata - # registers the JSON ingestion indexer extension provided by `elasticgraph-json_ingestion`. + # registers the JSON ingestion indexer extension provided by `elasticgraph-json_ingestion`, and + # exercises JSON Lines payload decoding via that gem's decoder. 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 78cae93ed..13fc41be1 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 an 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 - # encode multiple JSON objects with JSON Lines (http://jsonlines.org/) in the 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 06668f88f..ddd12e2fc 100644 --- a/elasticgraph-json_ingestion/README.md +++ b/elasticgraph-json_ingestion/README.md @@ -115,6 +115,21 @@ This gem also provides the `be_a_valid_elastic_graph_event` RSpec matcher (via `require "elastic_graph/json_ingestion/spec_support/event_matcher"`) for testing that publisher events conform to your schema. +## 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 81d541535..f53b25662 100644 --- a/elasticgraph-json_ingestion/elasticgraph-json_ingestion.gemspec +++ b/elasticgraph-json_ingestion/elasticgraph-json_ingestion.gemspec @@ -38,9 +38,10 @@ Gem::Specification.new do |spec| # The test suite builds indexers (with real datastore client classes) to exercise the ingestion adapter. spec.add_development_dependency "elasticgraph-elasticsearch", ElasticGraph::VERSION - # This gem's ingestion adapter code references `elasticgraph-indexer`, but the indexer loads it - # through the indexer extension registered in runtime metadata, after `elasticgraph-indexer` is - # already available. Keeping this as a development dependency avoids a runtime dependency cycle. + # This gem's ingestion adapter and indexing event decoder code references `elasticgraph-indexer`, + # but the indexer loads them through the indexer extension registered in runtime metadata and + # through the `indexing_event_decoder` setting, after `elasticgraph-indexer` is already available. + # Keeping this as a development dependency avoids a runtime dependency cycle. spec.add_development_dependency "elasticgraph-indexer", ElasticGraph::VERSION # The test suite builds indexers (with real datastore client classes) to exercise the ingestion adapter. spec.add_development_dependency "elasticgraph-opensearch", ElasticGraph::VERSION 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 6bfd3b52e..b5f62d3fb 100644 --- a/elasticgraph-lambda_support/elasticgraph-lambda_support.gemspec +++ b/elasticgraph-lambda_support/elasticgraph-lambda_support.gemspec @@ -49,6 +49,7 @@ Gem::Specification.new do |spec| spec.add_development_dependency "elasticgraph-indexer", ElasticGraph::VERSION spec.add_development_dependency "elasticgraph-indexer_autoscaler_lambda", ElasticGraph::VERSION # The test suite builds indexers from the shared test schema artifacts, whose runtime metadata - # registers the JSON ingestion indexer extension provided by `elasticgraph-json_ingestion`. + # registers the JSON ingestion indexer extension provided by `elasticgraph-json_ingestion`, and + # from the shared test settings, which configure that gem's indexing event decoder. 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 638a74a12..48959b041 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 @@ -538,6 +538,45 @@ properties: - [] - - name: MyExtensionModule require_path: "./my_extension_module" + 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 81358ba78..d7cb5143e 100644 --- a/elasticgraph-warehouse_lambda/elasticgraph-warehouse_lambda.gemspec +++ b/elasticgraph-warehouse_lambda/elasticgraph-warehouse_lambda.gemspec @@ -49,7 +49,10 @@ 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 builds indexers from the shared test schema artifacts, whose runtime metadata - # registers the JSON ingestion indexer extension provided by `elasticgraph-json_ingestion`. + # registers the JSON ingestion indexer extension provided by `elasticgraph-json_ingestion`, and + # exercises JSON Lines payload decoding via that gem's decoder. 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 2e32b7cf3..4a49615bb 100644 --- a/spec_support/lib/elastic_graph/spec_support/builds_indexer.rb +++ b/spec_support/lib/elastic_graph/spec_support/builds_indexer.rb @@ -19,6 +19,7 @@ def build_indexer( latency_slo_thresholds_by_timestamp_in_ms: {}, skip_derived_indexing_type_updates: {}, extension_modules: [], + indexing_event_decoder: nil, datastore_router: nil, clock: nil, monotonic_clock: nil, @@ -27,7 +28,8 @@ def build_indexer( ) 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 ) # This config setting must bypass the JSON schema validation so we provide it via `with`.