From 292fbd284b1d81ab0ecf1945beed4cd012c78a13 Mon Sep 17 00:00:00 2001 From: Josh Wilson Date: Tue, 7 Jul 2026 16:03:14 -0500 Subject: [PATCH] Move JSON ingestion into elasticgraph-json_ingestion via an indexer extension The JSON ingestion adapter and JSON-schema-based record preparer factory move from elasticgraph-indexer to elasticgraph-json_ingestion, and the JSON ingestion schema definition extension registers an indexer extension in runtime metadata that makes the adapter available at indexing time. Any schema defined with JSON ingestion support automatically gets JSON event ingestion--no settings needed--and elasticgraph-indexer no longer contains (or refers to) any JSON-specific ingestion logic. The be_a_valid_elastic_graph_event matcher moves to elastic_graph/json_ingestion/spec_support/event_matcher, and suites that build indexers from the shared test schema artifacts gain a development dependency on elasticgraph-json_ingestion (whose indexer extension those artifacts now reference). --- Steepfile | 4 +- config/schema/artifacts/runtime_metadata.yaml | 4 + .../runtime_metadata.yaml | 4 + .../elasticgraph-apollo.gemspec | 3 + .../lib/elastic_graph/indexer.rb | 26 +- .../lib/elastic_graph/indexer/event_id.rb | 2 +- .../indexer/ingestion_adapter/json_events.rb | 138 ---------- .../elastic_graph/indexer/record_preparer.rb | 52 ---- .../sig/elastic_graph/indexer.rbs | 6 +- .../indexer/ingestion_adapter/json_events.rbs | 31 --- .../elastic_graph/indexer/record_preparer.rbs | 14 - .../spec/support/indexing_preparer.rb | 2 +- .../indexer/datastore_indexing_router_spec.rb | 2 +- .../operation/count_accumulator_spec.rb | 2 +- .../indexer/operation/factory_spec.rb | 235 +---------------- .../indexer/operation/update_spec.rb | 2 +- .../indexer/record_preparer_spec.rb | 137 +--------- .../spec/unit/elastic_graph/indexer_spec.rb | 23 +- .../elasticgraph-indexer_lambda.gemspec | 4 + elasticgraph-json_ingestion/README.md | 13 + .../elasticgraph-json_ingestion.gemspec | 9 + .../json_ingestion/indexer_extension.rb | 26 ++ .../json_ingestion/ingestion_adapter.rb | 149 +++++++++++ .../json_ingestion/record_preparer_factory.rb | 73 +++++ .../schema_definition/api_extension.rb | 5 + .../spec_support/event_matcher.rb | 8 +- .../json_ingestion/indexer_extension.rbs | 9 + .../json_ingestion/ingestion_adapter.rbs | 31 +++ .../record_preparer_factory.rbs | 17 ++ .../spec/spec_helper.rb | 1 + .../spec/support/multiple_version_support.rb | 2 +- .../json_ingestion/indexer_extension_spec.rb | 27 ++ .../json_ingestion/ingestion_adapter_spec.rb | 249 ++++++++++++++++++ .../record_preparer_factory_spec.rb | 145 ++++++++++ .../spec_support/event_matcher_spec.rb | 2 +- .../elasticgraph-lambda_support.gemspec | 3 + .../spec_support/common_project_specs.rb | 2 +- .../elasticgraph-warehouse_lambda.gemspec | 3 + .../spec_support/builds_indexer_operation.rb | 2 +- .../spec_support/uses_datastore.rb | 2 +- spec_support/spec_helper.rb | 7 + 41 files changed, 838 insertions(+), 638 deletions(-) delete mode 100644 elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter/json_events.rb delete mode 100644 elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter/json_events.rbs create mode 100644 elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexer_extension.rb create mode 100644 elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/ingestion_adapter.rb create mode 100644 elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/record_preparer_factory.rb rename {elasticgraph-indexer/lib/elastic_graph/indexer => elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion}/spec_support/event_matcher.rb (84%) create mode 100644 elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/indexer_extension.rbs create mode 100644 elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/ingestion_adapter.rbs create mode 100644 elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/record_preparer_factory.rbs rename {elasticgraph-indexer => elasticgraph-json_ingestion}/spec/support/multiple_version_support.rb (98%) create mode 100644 elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/indexer_extension_spec.rb create mode 100644 elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/ingestion_adapter_spec.rb create mode 100644 elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/record_preparer_factory_spec.rb rename {elasticgraph-indexer/spec/unit/elastic_graph/indexer => elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion}/spec_support/event_matcher_spec.rb (97%) diff --git a/Steepfile b/Steepfile index 2d90d5498..d347ec10a 100644 --- a/Steepfile +++ b/Steepfile @@ -41,9 +41,9 @@ target :elasticgraph_gems do elasticgraph-graphql/lib/elastic_graph/graphql/schema/type.rb ]) - # elasticgraph-indexer: existing files that don't type check yet. + # elasticgraph-json_ingestion: RSpec matcher DSL that doesn't type check. ignore(*%w[ - elasticgraph-indexer/lib/elastic_graph/indexer/spec_support/event_matcher.rb + elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/spec_support/event_matcher.rb ]) # elasticgraph-local: existing files that don't type check yet. diff --git a/config/schema/artifacts/runtime_metadata.yaml b/config/schema/artifacts/runtime_metadata.yaml index 2d4fdb104..1420be209 100644 --- a/config/schema/artifacts/runtime_metadata.yaml +++ b/config/schema/artifacts/runtime_metadata.yaml @@ -3219,6 +3219,10 @@ index_definitions_by_name: frequency: yearly timestamp_field_path: created_at route_with: workspace_id2 +indexer_extension_modules: +- extension_ref: + name: ElasticGraph::JSONIngestion::IndexerExtension + require_path: elastic_graph/json_ingestion/indexer_extension object_types_by_name: Address: graphql_fields_by_name: diff --git a/config/schema/artifacts_with_apollo/runtime_metadata.yaml b/config/schema/artifacts_with_apollo/runtime_metadata.yaml index 7f890421c..59c679ac6 100644 --- a/config/schema/artifacts_with_apollo/runtime_metadata.yaml +++ b/config/schema/artifacts_with_apollo/runtime_metadata.yaml @@ -3248,6 +3248,10 @@ index_definitions_by_name: frequency: yearly timestamp_field_path: created_at route_with: workspace_id2 +indexer_extension_modules: +- extension_ref: + name: ElasticGraph::JSONIngestion::IndexerExtension + require_path: elastic_graph/json_ingestion/indexer_extension object_types_by_name: Address: graphql_fields_by_name: diff --git a/elasticgraph-apollo/elasticgraph-apollo.gemspec b/elasticgraph-apollo/elasticgraph-apollo.gemspec index b74256aa6..b948bb6fe 100644 --- a/elasticgraph-apollo/elasticgraph-apollo.gemspec +++ b/elasticgraph-apollo/elasticgraph-apollo.gemspec @@ -52,6 +52,9 @@ Gem::Specification.new do |spec| spec.add_development_dependency "elasticgraph-schema_definition", ElasticGraph::VERSION spec.add_development_dependency "elasticgraph-admin", ElasticGraph::VERSION 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`. + spec.add_development_dependency "elasticgraph-json_ingestion", ElasticGraph::VERSION spec.add_development_dependency "elasticgraph-opensearch", ElasticGraph::VERSION spec.add_development_dependency "elasticgraph-indexer", ElasticGraph::VERSION end diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer.rb b/elasticgraph-indexer/lib/elastic_graph/indexer.rb index f66fc63c3..9721a291b 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" @@ -56,21 +57,11 @@ def datastore_router end end - def record_preparer_factory - @record_preparer_factory ||= begin - require "elastic_graph/indexer/record_preparer" - RecordPreparer::Factory.new(schema_artifacts) - end - end - - # The ingestion adapters available for processing events. For now, only the JSON events - # adapter is available; indexer extension modules will be able to contribute additional - # adapters as alternate ingestion formats are supported. + # The ingestion adapters available for processing events. Returns an empty list by default; + # ingestion format gems contribute adapters via indexer extension modules that override this + # method. def ingestion_adapters - @ingestion_adapters ||= begin - require "elastic_graph/indexer/ingestion_adapter/json_events" - [IngestionAdapter::JSONEvents.new(schema_artifacts: schema_artifacts, logger: logger)] - end + [] # : ::Array[IngestionAdapter::_Adapter] end def processor @@ -88,6 +79,13 @@ def processor def operation_factory @operation_factory ||= begin + if ingestion_adapters.empty? + raise Errors::ConfigError, "No ingestion adapters are available to process events. Ingestion format gems " \ + "make an adapter available by registering an indexer extension (via `register_indexer_extension`) during " \ + "schema definition; ensure your schema definition uses an ingestion format extension and regenerate your " \ + "schema artifacts (or configure an extension via the `indexer.extension_modules` setting)." + end + require "elastic_graph/indexer/operation/factory" Operation::Factory.new( schema_artifacts: schema_artifacts, diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/event_id.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/event_id.rb index 5453553db..00c03cf49 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 record_preparer_factory, processor, operation_factory, ingestion_adapters, logger + # @dynamic processor, operation_factory, ingestion_adapters, logger # @dynamic self.from_parsed_yaml end end diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter/json_events.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter/json_events.rb deleted file mode 100644 index fef2ba26b..000000000 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter/json_events.rb +++ /dev/null @@ -1,138 +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/constants" -require "elastic_graph/indexer/event_id" -require "elastic_graph/indexer/ingestion_adapter" -require "elastic_graph/indexer/record_preparer" -require "elastic_graph/support/json_schema/validator_factory" - -module ElasticGraph - class Indexer - module IngestionAdapter - # Ingestion adapter for events in ElasticGraph's versioned JSON format: it validates events - # against the JSON schema identified by the event's `json_schema_version`, and prepares - # records using that version's view of the schema. - class JSONEvents - # @param schema_artifacts [SchemaArtifacts::FromDisk] the schema artifacts - # @param logger [Logger] the ElasticGraph logger - # @param configure_record_validator [Proc, nil] optional callback to further configure the record validator - def initialize(schema_artifacts:, logger:, configure_record_validator: nil) - @schema_artifacts = schema_artifacts - @logger = logger - @configure_record_validator = configure_record_validator - @record_preparer_factory = RecordPreparer::Factory.new(schema_artifacts) - end - - # (see Interface#handles_event?) - def handles_event?(event) - event.key?(JSON_SCHEMA_VERSION_KEY) - end - - # (see Interface#validate_event) - def validate_event(event) - selected_json_schema_version = select_json_schema_version(event) { |failure| return failure } - - # Because the `select_json_schema_version` picks the closest-matching json schema version, the incoming - # event might not match the expected json_schema_version value in the json schema (which is a `const` field). - # This is by design, since we're picking a schema based on best-effort, so to avoid that by-design validation error, - # performing the envelope validation on a "patched" version of the event. - event_with_patched_envelope = event.merge({JSON_SCHEMA_VERSION_KEY => selected_json_schema_version}) - - if (error_message = validator(EVENT_ENVELOPE_JSON_SCHEMA_NAME, selected_json_schema_version).validate_with_error_message(event_with_patched_envelope)) - return ValidationResult.invalid(payload_description: "event payload", message: error_message) - end - - record = event.fetch("record") - graphql_type_name = event.fetch("type") - - if (error_message = validator(graphql_type_name, selected_json_schema_version).validate_with_error_message(record)) - return ValidationResult.invalid(payload_description: "#{graphql_type_name} record", message: error_message) - end - - ValidationResult.valid(@record_preparer_factory.for_json_schema_version(selected_json_schema_version)) - end - - private - - def select_json_schema_version(event) - available_json_schema_versions = @schema_artifacts.available_json_schema_versions - - requested_json_schema_version = event[JSON_SCHEMA_VERSION_KEY] - - # First check that a valid value has been requested (a positive integer) - if !event.key?(JSON_SCHEMA_VERSION_KEY) - yield ValidationResult.invalid(payload_description: JSON_SCHEMA_VERSION_KEY, message: "Event lacks a `#{JSON_SCHEMA_VERSION_KEY}`") - elsif !requested_json_schema_version.is_a?(Integer) || requested_json_schema_version < 1 - yield ValidationResult.invalid(payload_description: JSON_SCHEMA_VERSION_KEY, message: "#{JSON_SCHEMA_VERSION_KEY} (#{requested_json_schema_version}) must be a positive integer.") - end - - # The requested version might not necessarily be available (if the publisher is deployed ahead of the indexer, or an old schema - # version is removed prematurely, or an indexer deployment is rolled back). So the behavior is to always pick the closest-available - # version. If there's an exact match, great. Even if not an exact match, if the incoming event payload conforms to the closest match, - # the event can still be indexed. - # - # This min_by block will take the closest version in the list. If a tie occurs, the first value in the list wins. The desired - # behavior is in the event of a tie (highly unlikely, there shouldn't be a gap in available json schema versions), the higher version - # should be selected. So to get that behavior, the list is sorted in descending order. - # - selected_json_schema_version = available_json_schema_versions.sort.reverse.min_by { |version| (requested_json_schema_version - version).abs } - - if selected_json_schema_version != requested_json_schema_version - @logger.info({ - "message_type" => "ElasticGraphMissingJSONSchemaVersion", - "message_id" => event["message_id"], - "event_id" => EventID.from_event(event), - "event_type" => event["type"], - "requested_json_schema_version" => requested_json_schema_version, - "selected_json_schema_version" => selected_json_schema_version - }) - end - - if selected_json_schema_version.nil? - yield ValidationResult.invalid( - payload_description: JSON_SCHEMA_VERSION_KEY, - message: "Failed to select json schema version. Requested version: #{event[JSON_SCHEMA_VERSION_KEY]}. \ - Available json schema versions: #{available_json_schema_versions.sort.join(", ")}" - ) - end - - selected_json_schema_version - end - - def validator(type, selected_json_schema_version) - factory = validator_factories_by_version[selected_json_schema_version] # : Support::JSONSchema::ValidatorFactory - factory.validator_for(type) - end - - def validator_factories_by_version - @validator_factories_by_version ||= ::Hash.new do |hash, json_schema_version| - factory = Support::JSONSchema::ValidatorFactory.new( - schema: @schema_artifacts.json_schemas_for(json_schema_version), - sanitize_pii: true - ) - - if (configure_record_validator = @configure_record_validator) - factory = configure_record_validator.call(factory) - end - - hash[json_schema_version] = factory - end - end - - # :nocov: -- this should not be called. Instead, it exists to guard against wrongly raising an error from this class. - def raise(*args) - super("`raise` was called on `IngestionAdapter::JSONEvents`, but should not. Instead, use " \ - "`yield ValidationResult.invalid(...)` so that we can accumulate all invalid events and allow " \ - "the valid events to still be processed.") - end - # :nocov: - end - end - end -end diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/record_preparer.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/record_preparer.rb index 6e0b04e44..f21b73e27 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/record_preparer.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/record_preparer.rb @@ -11,58 +11,6 @@ module ElasticGraph class Indexer class RecordPreparer - # Provides the ability to get a `RecordPreparer` for a specific JSON schema version. - class Factory - def initialize(schema_artifacts) - @schema_artifacts = schema_artifacts - - scalar_types_by_name = schema_artifacts.runtime_metadata.scalar_types_by_name - indexing_preparer_by_scalar_type_name = ::Hash.new do |hash, type_name| - hash[type_name] = scalar_types_by_name[type_name]&.load_indexing_preparer&.extension_class - end # : ::Hash[::String, SchemaArtifacts::RuntimeMetadata::extensionClass?] - - @preparers_by_json_schema_version = ::Hash.new do |hash, version| - hash[version] = RecordPreparer.new( - indexing_preparer_by_scalar_type_name, - build_type_metas_from(@schema_artifacts.json_schemas_for(version)) - ) - end - end - - # Gets the `RecordPreparer` for the given JSON schema version. - def for_json_schema_version(json_schema_version) - @preparers_by_json_schema_version[json_schema_version] # : RecordPreparer - end - - # Gets the `RecordPreparer` for the latest JSON schema version. Intended primarily - # for use in tests for convenience. - def for_latest_json_schema_version - for_json_schema_version(@schema_artifacts.latest_json_schema_version) - end - - private - - def build_type_metas_from(json_schemas) - json_schemas.fetch("$defs").filter_map do |type, type_def| - next if type == EVENT_ENVELOPE_JSON_SCHEMA_NAME - - properties = type_def.fetch("properties") do - {} # : ::Hash[::String, untyped] - end # : ::Hash[::String, untyped] - - eg_meta_by_field_name = properties.filter_map do |prop_name, prop| - eg_meta = prop["ElasticGraph"] - [prop_name, eg_meta] if eg_meta - end.to_h - - TypeMetadata.new( - name: type, - eg_meta_by_field_name: eg_meta_by_field_name - ) - end - end - end - # An alternate `RecordPreparer` implementation that implements the identity function: # it just echoes back the record it is given. # diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer.rbs index c876f01d2..5b80d8f36 100644 --- a/elasticgraph-indexer/sig/elastic_graph/indexer.rbs +++ b/elasticgraph-indexer/sig/elastic_graph/indexer.rbs @@ -19,8 +19,7 @@ module ElasticGraph @datastore_router: Indexer::_DatastoreRouter? def datastore_router: () -> Indexer::_DatastoreRouter - @record_preparer_factory: RecordPreparer::Factory? - def record_preparer_factory: () -> RecordPreparer::Factory + def ingestion_adapters: () -> ::Array[IngestionAdapter::_Adapter] @processor: Processor? def processor: () -> Processor @@ -28,9 +27,6 @@ module ElasticGraph @operation_factory: Operation::Factory? def operation_factory: () -> Operation::Factory - @ingestion_adapters: ::Array[IngestionAdapter::_Adapter]? - def ingestion_adapters: () -> ::Array[IngestionAdapter::_Adapter] - @monotonic_clock: Support::MonotonicClock? def monotonic_clock: () -> Support::MonotonicClock diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter/json_events.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter/json_events.rbs deleted file mode 100644 index a10b69923..000000000 --- a/elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter/json_events.rbs +++ /dev/null @@ -1,31 +0,0 @@ -module ElasticGraph - class Indexer - module IngestionAdapter - class JSONEvents - def initialize: ( - schema_artifacts: schemaArtifacts, - logger: ::Logger, - ?configure_record_validator: (^(Support::JSONSchema::ValidatorFactory) -> Support::JSONSchema::ValidatorFactory)? - ) -> void - - @schema_artifacts: schemaArtifacts - @logger: ::Logger - @configure_record_validator: (^(Support::JSONSchema::ValidatorFactory) -> Support::JSONSchema::ValidatorFactory)? - @record_preparer_factory: RecordPreparer::Factory - - def handles_event?: (event) -> bool - def validate_event: (event) -> ValidationResult - - private - - def select_json_schema_version: (event) { (ValidationResult) -> bot } -> (::Integer | bot) - def validator: (::String, ::Integer) -> Support::JSONSchema::Validator - - @validator_factories_by_version: ::Hash[::Integer, Support::JSONSchema::ValidatorFactory]? - def validator_factories_by_version: () -> ::Hash[::Integer, Support::JSONSchema::ValidatorFactory] - - def raise: (*untyped) -> bot - end - end - end -end diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer/record_preparer.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer/record_preparer.rbs index 1c4efcefc..694a5f307 100644 --- a/elasticgraph-indexer/sig/elastic_graph/indexer/record_preparer.rbs +++ b/elasticgraph-indexer/sig/elastic_graph/indexer/record_preparer.rbs @@ -8,20 +8,6 @@ module ElasticGraph type egMetaByFieldHash = ::Hash[::String, {"type" => ::String, "nameInIndex" => ::String}] type egMetaByFieldByTypeHash = ::Hash[::String, egMetaByFieldHash] - class Factory - def initialize: (schemaArtifacts) -> void - - def for_json_schema_version: (::Integer) -> RecordPreparer - def for_latest_json_schema_version: () -> RecordPreparer - - private - - @schema_artifacts: schemaArtifacts - @preparers_by_json_schema_version: ::Hash[::Integer, RecordPreparer] - - def build_type_metas_from: (::Hash[::String, untyped]) -> ::Array[TypeMetadata] - end - include _RecordPreparer module Identity diff --git a/elasticgraph-indexer/spec/support/indexing_preparer.rb b/elasticgraph-indexer/spec/support/indexing_preparer.rb index df85ca869..847cd8cfc 100644 --- a/elasticgraph-indexer/spec/support/indexing_preparer.rb +++ b/elasticgraph-indexer/spec/support/indexing_preparer.rb @@ -29,7 +29,7 @@ schema.object_type "Object" do |t| t.field "scalar", scalar_type end - end).record_preparer_factory.for_latest_json_schema_version + end).then { |indexer| latest_json_record_preparer_for(indexer) } end def prepare_scalar_value(value) diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/datastore_indexing_router_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/datastore_indexing_router_spec.rb index 7702a195a..dcb963ebb 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/datastore_indexing_router_spec.rb +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/datastore_indexing_router_spec.rb @@ -601,7 +601,7 @@ def new_operation(event, update_target: nil, **overrides) arguments = { event: event, - prepared_record: indexer.record_preparer_factory.for_latest_json_schema_version.prepare_for_index( + prepared_record: latest_json_record_preparer_for(indexer).prepare_for_index( event.fetch("type"), event.fetch("record"), destination_index_mapping.fetch("properties") diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/count_accumulator_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/count_accumulator_spec.rb index d420a40a3..eb16e86c2 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/count_accumulator_spec.rb +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/count_accumulator_spec.rb @@ -300,7 +300,7 @@ def script_params_for(data:, source_type:, destination_type:, indexer: build_ind update = Update.new( event: {"type" => source_type, "record" => data}, destination_index_def: destination_index_def, - prepared_record: indexer.record_preparer_factory.for_latest_json_schema_version.prepare_for_index( + prepared_record: latest_json_record_preparer_for(indexer).prepare_for_index( source_type, data, destination_index_mapping.fetch("properties") diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb index 80608dbce..62da01a24 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/factory_spec.rb @@ -8,8 +8,8 @@ require "elastic_graph/constants" require "elastic_graph/indexer" -require "elastic_graph/indexer/ingestion_adapter/json_events" require "elastic_graph/indexer/operation/factory" +require "elastic_graph/json_ingestion/record_preparer_factory" require "elastic_graph/spec_support/builds_indexer_operation" require "json" @@ -90,43 +90,6 @@ module Operation }.merge(latency_timestamps))]) end - it 'notifies an error when latency metrics contain keys that violate regex "^\\w+_at$"' do - valid_event = build_upsert_event(:component, id: "1", __version: 1) - invalid_event = valid_event.merge({ - "latency_timestamps" => { - "created_in_esperanto_at" => "2012-04-23T18:25:43.511Z", - "bad metric with spaces _at" => "2012-04-20T18:25:43.511Z", - "bad_metric" => "2012-04-20T18:25:43.511Z" - } - }) - - expect_failed_event_error(invalid_event, "/latency_timestamps/bad_metric", "bad metric with spaces _at") - end - - it "notifies an error when latency metrics contain values that are not ISO8601 date-time" do - valid_event = build_upsert_event(:component, id: "1", __version: 1) - invalid_event = valid_event.merge({ - "latency_timestamps" => { - "created_in_esperanto_at" => "2012-04-23T18:25:43.511Z", - "bad_metric_at" => "malformed datetime" - } - }) - - expect_failed_event_error(invalid_event, "/latency_timestamps/bad_metric") - end - - it "notifies an error on version number less than 1" do - event = build_upsert_event(:widget, __version: -1) - - expect_failed_event_error(event, "/properties/version") - end - - it "notifies an error on version number greater than 2^63 - 1" do - event = build_upsert_event(:widget, __version: 2**64) - - expect_failed_event_error(event, "/properties/version") - end - it "notifies an error on unknown graphql type" do event = { "op" => "upsert", @@ -157,30 +120,6 @@ module Operation expect_failed_event_error(event, "/properties/type", expect_no_ops: true) end - it "notifies an error on invalid operation" do - event = build_upsert_event(:widget).merge("op" => "invalid_op") - - expect_failed_event_error(event, "/properties/op") - end - - it "notifies an error on missing operation" do - event = build_upsert_event(:widget).except("op") - - expect_failed_event_error(event, "missing_keys", "op") - end - - it "notifies an error on missing record for upsert" do - event = build_upsert_event(:component).except("record") - - expect_failed_event_error(event, "/then") - end - - it "notifies an error on missing id" do - event = build_upsert_event(:component).except("id") - - expect_failed_event_error(event, "missing_keys", "id") - end - it "notifies an error on missing type" do event = build_upsert_event(:component).except("type") @@ -188,12 +127,6 @@ module Operation expect_failed_event_error(event, "missing_keys", "type", expect_no_ops: true) end - it "notifies an error on missing version" do - event = build_upsert_event(:component).except("version") - - expect_failed_event_error(event, "missing_keys", "version") - end - it "notifies an error on missing `#{JSON_SCHEMA_VERSION_KEY}`" do event = build_upsert_event(:component).except(JSON_SCHEMA_VERSION_KEY) @@ -235,152 +168,6 @@ module Operation expect_failed_event_error(bad_widget3, "/workspace_id") end - it "allows the validator to be configured with a block" do - event_with_extra_field = build_upsert_event(:widget, extra_field: 17) - - expect { - build_expecting_success(event_with_extra_field) - }.not_to raise_error - - expect { - build_expecting_success(event_with_extra_field) { |v| v.with_unknown_properties_disallowed } - }.to raise_error FailedEventError, a_string_including("extra_field") - end - - context "when the indexer has json schemas v2 and v4 (v4 adds yellow color)" do - before do - # With the "real" version one as a baseline, create a separate version with a small schema change. - # Tests will then specify the desired json_schema_version in the event payload to test the schema-choosing - # behavior of the `factory` class. - schemas = { - 2 => indexer.schema_artifacts.json_schemas_for(1), - 4 => ::Marshal.load(::Marshal.dump(indexer.schema_artifacts.json_schemas_for(1))).tap do |schema| - schema["$defs"]["Color"]["enum"] << "YELLOW" - end - } - - allow(indexer.schema_artifacts).to receive(:available_json_schema_versions).and_return(schemas.keys.to_set) - allow(indexer.schema_artifacts).to receive(:latest_json_schema_version).and_return(schemas.keys.max) - allow(indexer.schema_artifacts).to receive(:json_schemas_for) do |version| - ::Marshal.load(::Marshal.dump(schemas.fetch(version))).tap do |schema| - schema[JSON_SCHEMA_VERSION_KEY] = version - schema["$defs"]["ElasticGraphEventEnvelope"]["properties"][JSON_SCHEMA_VERSION_KEY]["const"] = version - end - end - end - - it "validates against an older version of a json schema if specified" do - # YELLOW doesn't exist in schema version 2. So expect an error when json_schema_version is set to 2. - event = build_upsert_event(:widget, id: "1", __version: 1, __json_schema_version: 2) - event["record"]["options"]["color"] = "YELLOW" - - expect_failed_event_error(event, "/options/color") - end - - it "validates against the latest version of a json schema if specified" do - event = build_upsert_event(:widget, id: "1", __version: 1, __json_schema_version: 4) - event["record"]["options"]["color"] = "YELLOW" - - expect(build_expecting_success(event)).to include(new_primary_indexing_operation({ - "op" => "upsert", - "id" => "1", - "type" => "Widget", - "version" => 1, - "record" => event["record"], - JSON_SCHEMA_VERSION_KEY => 4 - }, index_def: index_def_named("widgets"))) - end - - it "validates against the closest version if the requested version is newer than what's available" do - # 5 is closest to "4", validation should match behavior from version "4" - YELLOW should pass validation. - event = build_upsert_event(:widget, id: "1", __version: 1, __json_schema_version: 5) - event["record"]["options"]["color"] = "YELLOW" - - expect(build_expecting_success(event)).to include(new_primary_indexing_operation({ - "op" => "upsert", - "id" => "1", - "type" => "Widget", - "version" => 1, - "record" => event["record"], - JSON_SCHEMA_VERSION_KEY => 5 # Originally-specified version. - }, index_def: index_def_named("widgets"))) - - expect(logged_jsons_of_type("ElasticGraphMissingJSONSchemaVersion").last).to include( - "event_id" => "Widget:1@v1", - "event_type" => "Widget", - "requested_json_schema_version" => 5, - "selected_json_schema_version" => 4 - ) - end - - it "validates against the closest version if the requested version older than what's available" do - # 1 is closest to "2", validation should match behavior from version "2" - YELLOW should fail validation. - event = build_upsert_event(:widget, id: "1", __version: 1, __json_schema_version: 1).merge("message_id" => "m123") - event["record"]["options"]["color"] = "YELLOW" - - # Should fail, but should still log the version mismatch as well. - expect_failed_event_error( - event, - "Malformed Widget record", - "1", - "/options/color" - ) - - expect(logged_jsons_of_type("ElasticGraphMissingJSONSchemaVersion").last).to include( - "event_id" => "Widget:1@v1", - "message_id" => "m123", - "event_type" => "Widget", - "requested_json_schema_version" => 1, - "selected_json_schema_version" => 2 - ) - end - - it "validates against a version newer than what's requested, if the requested version is equidistant from two available versions" do - event = build_upsert_event(:widget, id: "1", __version: 1, __json_schema_version: 3) - event["record"]["options"]["color"] = "YELLOW" - - expect(build_expecting_success(event)).to include(new_primary_indexing_operation({ - "op" => "upsert", - "id" => "1", - "type" => "Widget", - "version" => 1, - "record" => event["record"], - JSON_SCHEMA_VERSION_KEY => 3 # Originally-specified version. - }, index_def: index_def_named("widgets"))) - - expect(logged_jsons_of_type("ElasticGraphMissingJSONSchemaVersion").last).to include( - "event_id" => "Widget:1@v1", - "event_type" => "Widget", - "requested_json_schema_version" => 3, - "selected_json_schema_version" => 4 - ) - end - - it "notifies an error if an invalid (e.g. negative) json_schema_version is specified" do - event = build_upsert_event(:widget, id: "1", __version: 1, __json_schema_version: -1) - - expect_failed_event_error(event, "must be a positive integer", "(-1)") - end - - it "notifies an error if it's unable to select a json_schema_version" do - event = build_upsert_event(:component, id: "1", __version: 1) - event["record"]["name"] = 123 - - fake_empty_schema_artifacts = instance_double( - "ElasticGraph::SchemaArtifacts::FromDisk", - available_json_schema_versions: Set[], - runtime_metadata: indexer.schema_artifacts.runtime_metadata, - indices: indexer.schema_artifacts.indices, - index_templates: indexer.schema_artifacts.index_templates, - index_mappings_by_index_def_name: indexer.schema_artifacts.index_mappings_by_index_def_name - ) - - operation_factory = build_indexer(schema_artifacts: fake_empty_schema_artifacts).operation_factory - - expect_failed_event_error(event, "Failed to select json schema version", factory: operation_factory) - end - end - it "also generates an update operation for related types that have fields `sourced_from` this event type" do event = build_upsert_event(:widget, id: "1", __version: 1, component_ids: ["c1", "c2", "c3"]) @@ -484,22 +271,10 @@ def event_id_from(event) end end - def build_expecting_success(event, **options, &configure_record_validator) - factory = indexer.operation_factory - - if configure_record_validator - json_events_adapter = IngestionAdapter::JSONEvents.new( - schema_artifacts: indexer.schema_artifacts, - logger: indexer.logger, - configure_record_validator: configure_record_validator - ) - - factory = factory.with(ingestion_adapters: [json_events_adapter]) - end - - result = factory.build(event, **options) + def build_expecting_success(event, **options) + result = indexer.operation_factory.build(event, **options) - raise result.failed_event_error if result.failed_event_error + expect(result.failed_event_error).to be nil result.operations end @@ -507,7 +282,7 @@ def widget_currency_derived_update_operation_for(event) operations = Update.operations_for( event: event, destination_index_def: index_def_named("widget_currencies"), - record_preparer: indexer.record_preparer_factory.for_latest_json_schema_version, + record_preparer: JSONIngestion::RecordPreparerFactory.new(indexer.schema_artifacts).for_latest_json_schema_version, update_target: indexer.schema_artifacts.runtime_metadata.object_types_by_name.fetch("Widget").update_targets.first, destination_index_mapping: indexer.schema_artifacts.index_mappings_by_index_def_name.fetch("widget_currencies") ) diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/update_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/update_spec.rb index 686611ceb..6c07c3e69 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/update_spec.rb +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/operation/update_spec.rb @@ -460,7 +460,7 @@ def operations_for_indexer(indexer, event: self.event, source_type: "Widget", de Update.operations_for( event: event, destination_index_def: index_defs_by_name.fetch(destination_index), - record_preparer: indexer.record_preparer_factory.for_latest_json_schema_version, + record_preparer: latest_json_record_preparer_for(indexer), update_target: update_target, destination_index_mapping: indexer.schema_artifacts.index_mappings_by_index_def_name.fetch(destination_index) ) diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/record_preparer_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/record_preparer_spec.rb index 78022efdb..d2586c6a6 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/record_preparer_spec.rb +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/record_preparer_spec.rb @@ -7,53 +7,10 @@ # frozen_string_literal: true require "elastic_graph/indexer/record_preparer" -require "elastic_graph/spec_support/schema_definition_helpers" -require "support/multiple_version_support" +require "elastic_graph/json_ingestion/record_preparer_factory" module ElasticGraph class Indexer - RSpec.describe RecordPreparer::Factory, :json_ingestion_schema_definition do - include_context "MultipleVersionSupport" - - let(:factory_with_multiple_versions) do - build_indexer_with_multiple_schema_versions(schema_versions: { - 1 => lambda do |schema| - schema.object_type "MyType" do |t| - t.field "id", "ID!" - t.field "name", "String" - t.index "my_type" - end - end, - - 2 => lambda do |schema| - schema.object_type "MyType" do |t| - t.field "id", "ID!" - t.field "name", "String!" - t.index "my_type" - end - end - }).record_preparer_factory - end - - describe "#for_json_schema_version" do - it "memoizes `RecordPreparer` since they are immutable and that saves on memory" do - for_v1 = factory_with_multiple_versions.for_json_schema_version(1) - for_v2 = factory_with_multiple_versions.for_json_schema_version(2) - - expect(for_v1).not_to eq(for_v2) - expect(factory_with_multiple_versions.for_json_schema_version(1)).to be for_v1 - end - end - - describe "#for_latest_json_schema_version" do - it "returns the record preparer for the latest JSON schema version" do - for_v2 = factory_with_multiple_versions.for_json_schema_version(2) - - expect(factory_with_multiple_versions.for_latest_json_schema_version).to be for_v2 - end - end - end - RSpec.describe RecordPreparer, :json_ingestion_schema_definition do describe "#prepare_for_index" do it "tolerates a `nil` value where an object would usually be" do @@ -494,101 +451,13 @@ class Indexer end end - context "when working with events for an old JSON schema version", :json_ingestion_schema_definition do - include_context "SchemaDefinitionHelpers" - - it "handles events for old versions before a field was deleted" do - preparer = build_preparer_for_old_json_schema_version( - v1_def: ->(schema) { - schema.object_type "MyType" do |t| - t.field "id", "ID!" - t.field "name", "String" - t.index "my_type" - end - }, - - v2_def: ->(schema) { - schema.object_type "MyType" do |t| - t.field "id", "ID!" - t.deleted_field "name" - t.index "my_type" - end - } - ) - - record = preparer.prepare_for_index("MyType", {"id" => "1", "name" => "Winston"}, {}) - - expect(record).to eq({"id" => "1"}) - end - - it "properly omits `__typename` under an embedded field for a non-abstract type, even when the type has been renamed" do - preparer = build_preparer_for_old_json_schema_version( - v1_def: ->(schema) { - schema.object_type "MyType" do |t| - t.field "id", "ID!" - t.field "cost", "Money" - t.index "my_type" - end - - schema.object_type "Money" do |t| - t.field "amount", "Int" - end - }, - - v2_def: ->(schema) { - schema.object_type "MyType" do |t| - t.field "id", "ID!" - t.field "cost", "Money2" - t.index "my_type" - end - - schema.object_type "Money2" do |t| - t.field "amount", "Int" - t.renamed_from "Money" - end - } - ) - - record = preparer.prepare_for_index("MyType", {"id" => "1", "cost" => {"amount" => 10, "__typename" => "Money"}}, {}) - - expect(record).to eq({"id" => "1", "cost" => {"amount" => 10}}) - end - - def build_preparer_for_old_json_schema_version(v1_def:, v2_def:) - v1_results = define_schema do |schema| - schema.json_schema_version 1 - v1_def.call(schema) - end - - v2_results = define_schema do |schema| - schema.json_schema_version 2 - v2_def.call(schema) - end - - v1_merge_result = v2_results.merge_field_metadata_into_json_schema(v1_results.current_public_json_schema) - - expect(v1_merge_result.missing_fields).to be_empty - expect(v1_merge_result.missing_types).to be_empty - - allow(v2_results).to receive(:json_schemas_for).with(1).and_return(v1_merge_result.json_schema) - - RecordPreparer::Factory.new(v2_results).for_json_schema_version(1) - end - - def define_schema(&schema_definition) - super(schema_element_name_form: "snake_case", &schema_definition) - end - end - def build_preparer(**config_overrides, &schema_definition) - build_indexer(schema_definition: schema_definition, **config_overrides) - .record_preparer_factory - .for_latest_json_schema_version + build_preparer_with_artifacts(**config_overrides, &schema_definition).first end def build_preparer_with_artifacts(**config_overrides, &schema_definition) indexer = build_indexer(schema_definition: schema_definition, **config_overrides) - preparer = indexer.record_preparer_factory.for_latest_json_schema_version + preparer = JSONIngestion::RecordPreparerFactory.new(indexer.schema_artifacts).for_latest_json_schema_version [preparer, indexer.schema_artifacts] 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 6830fd64a..31f440196 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer_spec.rb +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer_spec.rb @@ -30,6 +30,21 @@ module ElasticGraph end end + describe "#operation_factory" do + it "raises a clear error when no ingestion adapters are available (e.g. for a schema defined without an ingestion format extension)" do + indexer = build_indexer(schema_definition: lambda do |schema| + schema.object_type "Widget" do |t| + t.field "id", "ID!" + t.index "widgets" + end + end) + + expect { + indexer.operation_factory + }.to raise_error Errors::ConfigError, a_string_including("No ingestion adapters are available") + end + end + context "when `config.extension_modules` or runtime metadata indexer extension modules are configured" do it "applies the extensions when the Indexer instance is instantiated without impacting any other instances" do config_extension_module = Module.new do @@ -40,8 +55,8 @@ def datastore_router stub_const("ConfigExtensionModule", config_extension_module) runtime_metadata_extension_module = Module.new do - def processor - :runtime_metadata_processor + def monotonic_clock + :runtime_metadata_clock end end stub_const("RuntimeMetadataExtensionModule", runtime_metadata_extension_module) @@ -61,10 +76,10 @@ def processor ) expect(extended_indexer.datastore_router).to eq :config_router - expect(extended_indexer.processor).to eq :runtime_metadata_processor + expect(extended_indexer.monotonic_clock).to eq :runtime_metadata_clock expect(normal_indexer.datastore_router).to be_a(Indexer::DatastoreIndexingRouter) - expect(normal_indexer.processor).to be_a(Indexer::Processor) + expect(normal_indexer.monotonic_clock).to be_a(Support::MonotonicClock) end def define_schema_elements(schema) diff --git a/elasticgraph-indexer_lambda/elasticgraph-indexer_lambda.gemspec b/elasticgraph-indexer_lambda/elasticgraph-indexer_lambda.gemspec index 74252c0f1..7ce6ba076 100644 --- a/elasticgraph-indexer_lambda/elasticgraph-indexer_lambda.gemspec +++ b/elasticgraph-indexer_lambda/elasticgraph-indexer_lambda.gemspec @@ -44,4 +44,8 @@ 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 builds indexers from the shared test schema artifacts, whose runtime metadata + # registers the JSON ingestion indexer extension provided by `elasticgraph-json_ingestion`. + spec.add_development_dependency "elasticgraph-json_ingestion", ElasticGraph::VERSION end diff --git a/elasticgraph-json_ingestion/README.md b/elasticgraph-json_ingestion/README.md index 4b32d9cf1..06668f88f 100644 --- a/elasticgraph-json_ingestion/README.md +++ b/elasticgraph-json_ingestion/README.md @@ -102,6 +102,19 @@ ElasticGraph.define_schema do |schema| end ``` +## Indexing Support + +Beyond schema definition, this gem teaches `elasticgraph-indexer` how to ingest JSON events: it provides an +ingestion adapter that validates each event against the JSON schema identified by the event's +`json_schema_version` and prepares its record for indexing using that version's view of the schema. + +No configuration is needed: defining your schema with this gem's `SchemaDefinition::APIExtension` registers +an indexer extension in your schema artifacts' runtime metadata, which the indexer applies when it boots. + +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. + ## Dependency Diagram ```mermaid diff --git a/elasticgraph-json_ingestion/elasticgraph-json_ingestion.gemspec b/elasticgraph-json_ingestion/elasticgraph-json_ingestion.gemspec index cdade27c0..81d541535 100644 --- a/elasticgraph-json_ingestion/elasticgraph-json_ingestion.gemspec +++ b/elasticgraph-json_ingestion/elasticgraph-json_ingestion.gemspec @@ -36,6 +36,15 @@ Gem::Specification.new do |spec| spec.add_dependency "elasticgraph-support", ElasticGraph::VERSION + # 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. + 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 + # 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/indexer_extension.rb b/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexer_extension.rb new file mode 100644 index 000000000..e02615bce --- /dev/null +++ b/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexer_extension.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 + +require "elastic_graph/json_ingestion/ingestion_adapter" + +module ElasticGraph + module JSONIngestion + # Indexer extension module that makes the JSON {IngestionAdapter} available to the indexer. + # {SchemaDefinition::APIExtension} registers this extension during schema definition, so any + # schema defined with JSON ingestion support automatically gets JSON event ingestion at + # indexing time--no configuration needed. + module IndexerExtension + # Adds the JSON {IngestionAdapter} to the indexer's available ingestion adapters. + # + # @return [Array] the available ingestion adapters + def ingestion_adapters + @ingestion_adapters ||= super + [IngestionAdapter.new(schema_artifacts: schema_artifacts, logger: logger)] + end + end + end +end diff --git a/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/ingestion_adapter.rb b/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/ingestion_adapter.rb new file mode 100644 index 000000000..a3cce2d00 --- /dev/null +++ b/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/ingestion_adapter.rb @@ -0,0 +1,149 @@ +# 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/constants" +require "elastic_graph/indexer/event_id" +require "elastic_graph/indexer/ingestion_adapter" +require "elastic_graph/json_ingestion/record_preparer_factory" +require "elastic_graph/support/json_schema/validator_factory" + +module ElasticGraph + module JSONIngestion + # Ingestion adapter for events in ElasticGraph's versioned JSON format: it validates events + # against the JSON schema identified by the event's `json_schema_version`, and prepares + # records using that version's view of the schema. Made available to the indexer by the + # {IndexerExtension} that {SchemaDefinition::APIExtension} registers. + class IngestionAdapter + # Shorthand for the result type defined by the indexer's ingestion adapter interface. + ValidationResult = Indexer::IngestionAdapter::ValidationResult + private_constant :ValidationResult + + # @param schema_artifacts [SchemaArtifacts::FromDisk] the schema artifacts + # @param logger [Logger] the ElasticGraph logger + # @param configure_record_validator [Proc, nil] optional callback to further configure the record validator + def initialize(schema_artifacts:, logger:, configure_record_validator: nil) + @schema_artifacts = schema_artifacts + @logger = logger + @configure_record_validator = configure_record_validator + @record_preparer_factory = RecordPreparerFactory.new(schema_artifacts) + end + + # Indicates whether this adapter recognizes the given event as one of its own, based on the + # presence of the `json_schema_version` field in the event envelope. + # + # @param event [Hash] an ElasticGraph indexing event + # @return [Boolean] whether this adapter handles the event + def handles_event?(event) + event.key?(JSON_SCHEMA_VERSION_KEY) + end + + # Validates the given event and resolves the record preparer appropriate for the event's + # JSON schema version. + # + # @param event [Hash] an ElasticGraph indexing event + # @return [Indexer::IngestionAdapter::ValidationResult] the result of validating the event + def validate_event(event) + selected_json_schema_version = select_json_schema_version(event) { |failure| return failure } + + # Because the `select_json_schema_version` picks the closest-matching json schema version, the incoming + # event might not match the expected json_schema_version value in the json schema (which is a `const` field). + # This is by design, since we're picking a schema based on best-effort, so to avoid that by-design validation error, + # performing the envelope validation on a "patched" version of the event. + event_with_patched_envelope = event.merge({JSON_SCHEMA_VERSION_KEY => selected_json_schema_version}) + + if (error_message = validator(EVENT_ENVELOPE_JSON_SCHEMA_NAME, selected_json_schema_version).validate_with_error_message(event_with_patched_envelope)) + return ValidationResult.invalid(payload_description: "event payload", message: error_message) + end + + record = event.fetch("record") + graphql_type_name = event.fetch("type") + + if (error_message = validator(graphql_type_name, selected_json_schema_version).validate_with_error_message(record)) + return ValidationResult.invalid(payload_description: "#{graphql_type_name} record", message: error_message) + end + + ValidationResult.valid(@record_preparer_factory.for_json_schema_version(selected_json_schema_version)) + end + + private + + def select_json_schema_version(event) + available_json_schema_versions = @schema_artifacts.available_json_schema_versions + + requested_json_schema_version = event[JSON_SCHEMA_VERSION_KEY] + + # First check that a valid value has been requested (a positive integer) + if !event.key?(JSON_SCHEMA_VERSION_KEY) + yield ValidationResult.invalid(payload_description: JSON_SCHEMA_VERSION_KEY, message: "Event lacks a `#{JSON_SCHEMA_VERSION_KEY}`") + elsif !requested_json_schema_version.is_a?(Integer) || requested_json_schema_version < 1 + yield ValidationResult.invalid(payload_description: JSON_SCHEMA_VERSION_KEY, message: "#{JSON_SCHEMA_VERSION_KEY} (#{requested_json_schema_version}) must be a positive integer.") + end + + # The requested version might not necessarily be available (if the publisher is deployed ahead of the indexer, or an old schema + # version is removed prematurely, or an indexer deployment is rolled back). So the behavior is to always pick the closest-available + # version. If there's an exact match, great. Even if not an exact match, if the incoming event payload conforms to the closest match, + # the event can still be indexed. + # + # This min_by block will take the closest version in the list. If a tie occurs, the first value in the list wins. The desired + # behavior is in the event of a tie (highly unlikely, there shouldn't be a gap in available json schema versions), the higher version + # should be selected. So to get that behavior, the list is sorted in descending order. + # + selected_json_schema_version = available_json_schema_versions.sort.reverse.min_by { |version| (requested_json_schema_version - version).abs } + + if selected_json_schema_version != requested_json_schema_version + @logger.info({ + "message_type" => "ElasticGraphMissingJSONSchemaVersion", + "message_id" => event["message_id"], + "event_id" => Indexer::EventID.from_event(event), + "event_type" => event["type"], + "requested_json_schema_version" => requested_json_schema_version, + "selected_json_schema_version" => selected_json_schema_version + }) + end + + if selected_json_schema_version.nil? + yield ValidationResult.invalid( + payload_description: JSON_SCHEMA_VERSION_KEY, + message: "Failed to select json schema version. Requested version: #{event[JSON_SCHEMA_VERSION_KEY]}. \ + Available json schema versions: #{available_json_schema_versions.sort.join(", ")}" + ) + end + + selected_json_schema_version + end + + def validator(type, selected_json_schema_version) + factory = validator_factories_by_version[selected_json_schema_version] # : Support::JSONSchema::ValidatorFactory + factory.validator_for(type) + end + + def validator_factories_by_version + @validator_factories_by_version ||= ::Hash.new do |hash, json_schema_version| + factory = Support::JSONSchema::ValidatorFactory.new( + schema: @schema_artifacts.json_schemas_for(json_schema_version), + sanitize_pii: true + ) + + if (configure_record_validator = @configure_record_validator) + factory = configure_record_validator.call(factory) + end + + hash[json_schema_version] = factory + end + end + + # :nocov: -- this should not be called. Instead, it exists to guard against wrongly raising an error from this class. + def raise(*args) + super("`raise` was called on `JSONIngestion::IngestionAdapter`, but should not. Instead, use " \ + "`yield ValidationResult.invalid(...)` so that we can accumulate all invalid events and allow " \ + "the valid events to still be processed.") + end + # :nocov: + end + end +end diff --git a/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/record_preparer_factory.rb b/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/record_preparer_factory.rb new file mode 100644 index 000000000..b1cfa2ab0 --- /dev/null +++ b/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/record_preparer_factory.rb @@ -0,0 +1,73 @@ +# 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/constants" +require "elastic_graph/indexer/record_preparer" + +module ElasticGraph + module JSONIngestion + # Provides the ability to get an `Indexer::RecordPreparer` for a specific JSON schema version, + # deriving each version's per-type field metadata from that version's JSON schemas. + class RecordPreparerFactory + # @param schema_artifacts [SchemaArtifacts::FromDisk] the schema artifacts + def initialize(schema_artifacts) + @schema_artifacts = schema_artifacts + + scalar_types_by_name = schema_artifacts.runtime_metadata.scalar_types_by_name + indexing_preparer_by_scalar_type_name = ::Hash.new do |hash, type_name| + hash[type_name] = scalar_types_by_name[type_name]&.load_indexing_preparer&.extension_class + end # : ::Hash[::String, SchemaArtifacts::RuntimeMetadata::extensionClass?] + + @preparers_by_json_schema_version = ::Hash.new do |hash, version| + hash[version] = Indexer::RecordPreparer.new( + indexing_preparer_by_scalar_type_name, + build_type_metas_from(@schema_artifacts.json_schemas_for(version)) + ) + end + end + + # Gets the `Indexer::RecordPreparer` for the given JSON schema version. + # + # @param json_schema_version [Integer] the JSON schema version + # @return [Indexer::RecordPreparer] the record preparer for the given version + def for_json_schema_version(json_schema_version) + @preparers_by_json_schema_version[json_schema_version] # : Indexer::RecordPreparer + end + + # Gets the `Indexer::RecordPreparer` for the latest JSON schema version. Intended primarily + # for use in tests for convenience. + # + # @return [Indexer::RecordPreparer] the record preparer for the latest version + def for_latest_json_schema_version + for_json_schema_version(@schema_artifacts.latest_json_schema_version) + end + + private + + def build_type_metas_from(json_schemas) + json_schemas.fetch("$defs").filter_map do |type, type_def| + next if type == EVENT_ENVELOPE_JSON_SCHEMA_NAME + + properties = type_def.fetch("properties") do + {} # : ::Hash[::String, untyped] + end # : ::Hash[::String, untyped] + + eg_meta_by_field_name = properties.filter_map do |prop_name, prop| + eg_meta = prop["ElasticGraph"] + [prop_name, eg_meta] if eg_meta + end.to_h + + Indexer::RecordPreparer::TypeMetadata.new( + name: type, + eg_meta_by_field_name: eg_meta_by_field_name + ) + end + end + end + end +end diff --git a/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/schema_definition/api_extension.rb b/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/schema_definition/api_extension.rb index a8b8ef645..30418feec 100644 --- a/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/schema_definition/api_extension.rb +++ b/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/schema_definition/api_extension.rb @@ -29,6 +29,11 @@ def self.extended(api) state.reserved_type_names << EVENT_ENVELOPE_JSON_SCHEMA_NAME api.factory.extend(FactoryExtension) + # Register the indexer extension so that any schema defined with JSON ingestion support + # automatically gets JSON event ingestion at indexing time. + require(indexer_extension_require_path = "elastic_graph/json_ingestion/indexer_extension") + api.register_indexer_extension(IndexerExtension, defined_at: indexer_extension_require_path) + api.on_built_in_types do |type| if type.name == api.state.type_ref("GeoLocation").to_final_form.name # @type var geo_location_type: ElasticGraph::SchemaDefinition::SchemaElements::TypeWithSubfields & SchemaElements::TypeWithSubfieldsExtension diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/spec_support/event_matcher.rb b/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/spec_support/event_matcher.rb similarity index 84% rename from elasticgraph-indexer/lib/elastic_graph/indexer/spec_support/event_matcher.rb rename to elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/spec_support/event_matcher.rb index d1a705e59..394c36986 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/spec_support/event_matcher.rb +++ b/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/spec_support/event_matcher.rb @@ -6,13 +6,13 @@ # # frozen_string_literal: true -require "elastic_graph/indexer/ingestion_adapter/json_events" +require "elastic_graph/json_ingestion/ingestion_adapter" require "json" -# Defines an RSpec matcher that can be used to validate ElasticGraph events. +# Defines an RSpec matcher that can be used to validate ElasticGraph JSON events. ::RSpec::Matchers.define :be_a_valid_elastic_graph_event do |for_indexer:| match do |event| - json_events_adapter = ElasticGraph::Indexer::IngestionAdapter::JSONEvents.new( + ingestion_adapter = ElasticGraph::JSONIngestion::IngestionAdapter.new( schema_artifacts: for_indexer.schema_artifacts, logger: for_indexer.logger, configure_record_validator: block_arg @@ -20,7 +20,7 @@ result = for_indexer .operation_factory - .with(ingestion_adapters: [json_events_adapter]) + .with(ingestion_adapters: [ingestion_adapter]) .build(event) @validation_failure = result.failed_event_error diff --git a/elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/indexer_extension.rbs b/elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/indexer_extension.rbs new file mode 100644 index 000000000..d9b0eafec --- /dev/null +++ b/elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/indexer_extension.rbs @@ -0,0 +1,9 @@ +module ElasticGraph + module JSONIngestion + module IndexerExtension : ::ElasticGraph::Indexer + @ingestion_adapters: ::Array[Indexer::IngestionAdapter::_Adapter]? + + def ingestion_adapters: () -> ::Array[Indexer::IngestionAdapter::_Adapter] + end + end +end diff --git a/elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/ingestion_adapter.rbs b/elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/ingestion_adapter.rbs new file mode 100644 index 000000000..ef02311ba --- /dev/null +++ b/elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/ingestion_adapter.rbs @@ -0,0 +1,31 @@ +module ElasticGraph + module JSONIngestion + class IngestionAdapter + ValidationResult: singleton(Indexer::IngestionAdapter::ValidationResult) + + def initialize: ( + schema_artifacts: schemaArtifacts, + logger: ::Logger, + ?configure_record_validator: (^(Support::JSONSchema::ValidatorFactory) -> Support::JSONSchema::ValidatorFactory)? + ) -> void + + @schema_artifacts: schemaArtifacts + @logger: ::Logger + @configure_record_validator: (^(Support::JSONSchema::ValidatorFactory) -> Support::JSONSchema::ValidatorFactory)? + @record_preparer_factory: RecordPreparerFactory + + def handles_event?: (Indexer::event) -> bool + def validate_event: (Indexer::event) -> Indexer::IngestionAdapter::ValidationResult + + private + + def select_json_schema_version: (Indexer::event) { (Indexer::IngestionAdapter::ValidationResult) -> bot } -> (::Integer | bot) + def validator: (::String, ::Integer) -> Support::JSONSchema::Validator + + @validator_factories_by_version: ::Hash[::Integer, Support::JSONSchema::ValidatorFactory]? + def validator_factories_by_version: () -> ::Hash[::Integer, Support::JSONSchema::ValidatorFactory] + + def raise: (*untyped) -> bot + end + end +end diff --git a/elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/record_preparer_factory.rbs b/elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/record_preparer_factory.rbs new file mode 100644 index 000000000..6bdb207de --- /dev/null +++ b/elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/record_preparer_factory.rbs @@ -0,0 +1,17 @@ +module ElasticGraph + module JSONIngestion + class RecordPreparerFactory + def initialize: (schemaArtifacts) -> void + + def for_json_schema_version: (::Integer) -> Indexer::RecordPreparer + def for_latest_json_schema_version: () -> Indexer::RecordPreparer + + private + + @schema_artifacts: schemaArtifacts + @preparers_by_json_schema_version: ::Hash[::Integer, Indexer::RecordPreparer] + + def build_type_metas_from: (::Hash[::String, untyped]) -> ::Array[Indexer::RecordPreparer::TypeMetadata] + end + end +end diff --git a/elasticgraph-json_ingestion/spec/spec_helper.rb b/elasticgraph-json_ingestion/spec/spec_helper.rb index 1bbf10366..f09d716a7 100644 --- a/elasticgraph-json_ingestion/spec/spec_helper.rb +++ b/elasticgraph-json_ingestion/spec/spec_helper.rb @@ -14,6 +14,7 @@ RSpec.configure do |config| config.define_derived_metadata(absolute_file_path: %r{/elasticgraph-json_ingestion/}) do |meta| + meta[:builds_indexer] = true meta[:json_ingestion_schema_definition] = true end diff --git a/elasticgraph-indexer/spec/support/multiple_version_support.rb b/elasticgraph-json_ingestion/spec/support/multiple_version_support.rb similarity index 98% rename from elasticgraph-indexer/spec/support/multiple_version_support.rb rename to elasticgraph-json_ingestion/spec/support/multiple_version_support.rb index e3b044aab..9fbcc6705 100644 --- a/elasticgraph-indexer/spec/support/multiple_version_support.rb +++ b/elasticgraph-json_ingestion/spec/support/multiple_version_support.rb @@ -9,7 +9,7 @@ require "elastic_graph/spec_support/schema_definition_helpers" module ElasticGraph - class Indexer + module JSONIngestion ::RSpec.shared_context "MultipleVersionSupport", :json_ingestion_schema_definition do include_context "SchemaDefinitionHelpers" diff --git a/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/indexer_extension_spec.rb b/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/indexer_extension_spec.rb new file mode 100644 index 000000000..da539d4ae --- /dev/null +++ b/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/indexer_extension_spec.rb @@ -0,0 +1,27 @@ +# 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/json_ingestion/indexer_extension" + +module ElasticGraph + module JSONIngestion + RSpec.describe IndexerExtension, :factories do + it "makes the JSON ingestion adapter available to indexers built for schemas defined with JSON ingestion support, with no configuration needed" do + indexer = build_indexer + + expect(indexer.ingestion_adapters).to contain_exactly(an_instance_of(IngestionAdapter)) + expect(indexer.ingestion_adapters).to be(indexer.ingestion_adapters), "expected the adapters to be memoized" + + result = indexer.operation_factory.build(build_upsert_event(:component)) + + expect(result.failed_event_error).to be nil + expect(result.operations).not_to be_empty + end + end + end +end diff --git a/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/ingestion_adapter_spec.rb b/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/ingestion_adapter_spec.rb new file mode 100644 index 000000000..f0783a55d --- /dev/null +++ b/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/ingestion_adapter_spec.rb @@ -0,0 +1,249 @@ +# 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/constants" +require "elastic_graph/json_ingestion/ingestion_adapter" + +module ElasticGraph + module JSONIngestion + RSpec.describe IngestionAdapter, :capture_logs, :factories do + let(:schema_artifacts) { stock_schema_artifacts } + let(:adapter) { build_adapter } + + describe "#handles_event?" do + it "recognizes events that have a `#{JSON_SCHEMA_VERSION_KEY}` in their envelope" do + event = build_upsert_event(:component) + + expect(adapter.handles_event?(event)).to be true + expect(adapter.handles_event?(event.except(JSON_SCHEMA_VERSION_KEY))).to be false + end + end + + describe "#validate_event" do + it "returns a valid result with a record preparer for the event's JSON schema version" do + event = build_upsert_event(:component) + + result = adapter.validate_event(event) + + expect(result.failure).to be nil + expect(result.record_preparer.prepare_for_index("Component", {"id" => "1", "unknown_field" => 3}, {})) + .to eq({"id" => "1"}) + end + + it "notifies an error when latency metrics contain keys that violate regex \"^\\w+_at$\"" do + event = build_upsert_event(:component, id: "1", __version: 1).merge({ + "latency_timestamps" => { + "created_in_esperanto_at" => "2012-04-23T18:25:43.511Z", + "bad metric with spaces _at" => "2012-04-20T18:25:43.511Z", + "bad_metric" => "2012-04-20T18:25:43.511Z" + } + }) + + expect_invalid(event, payload_description: "event payload", message_including: ["/latency_timestamps/bad_metric", "bad metric with spaces _at"]) + end + + it "notifies an error when latency metrics contain values that are not ISO8601 date-time" do + event = build_upsert_event(:component, id: "1", __version: 1).merge({ + "latency_timestamps" => { + "created_in_esperanto_at" => "2012-04-23T18:25:43.511Z", + "bad_metric_at" => "malformed datetime" + } + }) + + expect_invalid(event, payload_description: "event payload", message_including: ["/latency_timestamps/bad_metric"]) + end + + it "notifies an error on version number less than 1" do + event = build_upsert_event(:widget, __version: -1) + + expect_invalid(event, payload_description: "event payload", message_including: ["/properties/version"]) + end + + it "notifies an error on version number greater than 2^63 - 1" do + event = build_upsert_event(:widget, __version: 2**64) + + expect_invalid(event, payload_description: "event payload", message_including: ["/properties/version"]) + end + + it "notifies an error on invalid operation" do + event = build_upsert_event(:widget).merge("op" => "invalid_op") + + expect_invalid(event, payload_description: "event payload", message_including: ["/properties/op"]) + end + + it "notifies an error on missing operation" do + event = build_upsert_event(:widget).except("op") + + expect_invalid(event, payload_description: "event payload", message_including: ["missing_keys", "op"]) + end + + it "notifies an error on missing record for upsert" do + event = build_upsert_event(:component).except("record") + + expect_invalid(event, payload_description: "event payload", message_including: ["/then"]) + end + + it "notifies an error on missing id" do + event = build_upsert_event(:component).except("id") + + expect_invalid(event, payload_description: "event payload", message_including: ["missing_keys", "id"]) + end + + it "notifies an error on missing version" do + event = build_upsert_event(:component).except("version") + + expect_invalid(event, payload_description: "event payload", message_including: ["missing_keys", "version"]) + end + + it "notifies an error on missing `#{JSON_SCHEMA_VERSION_KEY}`" do + event = build_upsert_event(:component).except(JSON_SCHEMA_VERSION_KEY) + + expect_invalid(event, payload_description: JSON_SCHEMA_VERSION_KEY, message_including: ["Event lacks a `#{JSON_SCHEMA_VERSION_KEY}`"]) + end + + it "notifies an error when given a record that does not satisfy the type's JSON schema, while avoiding revealing PII" do + event = build_upsert_event(:component, id: "1", __version: 1) + event["record"]["name"] = 123 + + message = expect_invalid(event, payload_description: "Component record", message_including: ["name"]) + expect(message).to exclude("123") + end + + it "allows the record validator to be configured with a block" do + event_with_extra_field = build_upsert_event(:widget, extra_field: 17) + event_with_extra_field["record"]["extra_field"] = 17 + + expect(adapter.validate_event(event_with_extra_field).failure).to be nil + + configured_adapter = build_adapter { |v| v.with_unknown_properties_disallowed } + failure = configured_adapter.validate_event(event_with_extra_field).failure + + expect(failure.message).to include("extra_field") + end + + context "when the adapter has json schemas v2 and v4 (v4 adds yellow color)" do + before do + # With the "real" version one as a baseline, create a separate version with a small schema change. + # Tests will then specify the desired json_schema_version in the event payload to test the schema-choosing + # behavior of the adapter. + schemas = { + 2 => schema_artifacts.json_schemas_for(1), + 4 => ::Marshal.load(::Marshal.dump(schema_artifacts.json_schemas_for(1))).tap do |schema| + schema["$defs"]["Color"]["enum"] << "YELLOW" + end + } + + allow(schema_artifacts).to receive(:available_json_schema_versions).and_return(schemas.keys.to_set) + allow(schema_artifacts).to receive(:latest_json_schema_version).and_return(schemas.keys.max) + allow(schema_artifacts).to receive(:json_schemas_for) do |version| + ::Marshal.load(::Marshal.dump(schemas.fetch(version))).tap do |schema| + schema[JSON_SCHEMA_VERSION_KEY] = version + schema["$defs"]["ElasticGraphEventEnvelope"]["properties"][JSON_SCHEMA_VERSION_KEY]["const"] = version + end + end + end + + it "validates against an older version of a json schema if specified" do + # YELLOW doesn't exist in schema version 2. So expect an error when json_schema_version is set to 2. + event = build_upsert_event(:widget, id: "1", __version: 1, __json_schema_version: 2) + event["record"]["options"]["color"] = "YELLOW" + + expect_invalid(event, payload_description: "Widget record", message_including: ["/options/color"]) + end + + it "validates against the latest version of a json schema if specified" do + event = build_upsert_event(:widget, id: "1", __version: 1, __json_schema_version: 4) + event["record"]["options"]["color"] = "YELLOW" + + expect(adapter.validate_event(event).failure).to be nil + end + + it "validates against the closest version if the requested version is newer than what's available" do + # 5 is closest to "4", validation should match behavior from version "4" - YELLOW should pass validation. + event = build_upsert_event(:widget, id: "1", __version: 1, __json_schema_version: 5) + event["record"]["options"]["color"] = "YELLOW" + + expect(adapter.validate_event(event).failure).to be nil + + expect(logged_jsons_of_type("ElasticGraphMissingJSONSchemaVersion").last).to include( + "event_id" => "Widget:1@v1", + "event_type" => "Widget", + "requested_json_schema_version" => 5, + "selected_json_schema_version" => 4 + ) + end + + it "validates against the closest version if the requested version older than what's available" do + # 1 is closest to "2", validation should match behavior from version "2" - YELLOW should fail validation. + event = build_upsert_event(:widget, id: "1", __version: 1, __json_schema_version: 1).merge("message_id" => "m123") + event["record"]["options"]["color"] = "YELLOW" + + # Should fail, but should still log the version mismatch as well. + expect_invalid(event, payload_description: "Widget record", message_including: ["/options/color"]) + + expect(logged_jsons_of_type("ElasticGraphMissingJSONSchemaVersion").last).to include( + "event_id" => "Widget:1@v1", + "message_id" => "m123", + "event_type" => "Widget", + "requested_json_schema_version" => 1, + "selected_json_schema_version" => 2 + ) + end + + it "validates against a version newer than what's requested, if the requested version is equidistant from two available versions" do + event = build_upsert_event(:widget, id: "1", __version: 1, __json_schema_version: 3) + event["record"]["options"]["color"] = "YELLOW" + + expect(adapter.validate_event(event).failure).to be nil + + expect(logged_jsons_of_type("ElasticGraphMissingJSONSchemaVersion").last).to include( + "event_id" => "Widget:1@v1", + "event_type" => "Widget", + "requested_json_schema_version" => 3, + "selected_json_schema_version" => 4 + ) + end + + it "notifies an error if an invalid (e.g. negative) json_schema_version is specified" do + event = build_upsert_event(:widget, id: "1", __version: 1, __json_schema_version: -1) + + expect_invalid(event, payload_description: JSON_SCHEMA_VERSION_KEY, message_including: ["must be a positive integer", "(-1)"]) + end + end + + it "notifies an error if it's unable to select a json_schema_version" do + allow(schema_artifacts).to receive(:available_json_schema_versions).and_return(Set[]) + + event = build_upsert_event(:component, id: "1", __version: 1) + + expect_invalid(event, payload_description: JSON_SCHEMA_VERSION_KEY, message_including: ["Failed to select json schema version"]) + end + + def expect_invalid(event, payload_description:, message_including:) + result = adapter.validate_event(event) + + expect(result.record_preparer).to be nil + + failure = result.failure + expect(failure.payload_description).to eq(payload_description) + expect(failure.message).to include(*message_including) + + failure.message + end + end + + def build_adapter(&configure_record_validator) + IngestionAdapter.new( + schema_artifacts: schema_artifacts, + logger: logger, + configure_record_validator: configure_record_validator + ) + end + end + end +end diff --git a/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/record_preparer_factory_spec.rb b/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/record_preparer_factory_spec.rb new file mode 100644 index 000000000..1eb478281 --- /dev/null +++ b/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/record_preparer_factory_spec.rb @@ -0,0 +1,145 @@ +# 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/json_ingestion/record_preparer_factory" +require "elastic_graph/spec_support/schema_definition_helpers" +require "support/multiple_version_support" + +module ElasticGraph + module JSONIngestion + RSpec.describe RecordPreparerFactory, :json_ingestion_schema_definition do + include_context "MultipleVersionSupport" + + let(:factory_with_multiple_versions) do + build_indexer_with_multiple_schema_versions(schema_versions: { + 1 => lambda do |schema| + schema.object_type "MyType" do |t| + t.field "id", "ID!" + t.field "name", "String" + t.index "my_type" + end + end, + + 2 => lambda do |schema| + schema.object_type "MyType" do |t| + t.field "id", "ID!" + t.field "name", "String!" + t.index "my_type" + end + end + }).then { |indexer| RecordPreparerFactory.new(indexer.schema_artifacts) } + end + + describe "#for_json_schema_version" do + it "memoizes `RecordPreparer` since they are immutable and that saves on memory" do + for_v1 = factory_with_multiple_versions.for_json_schema_version(1) + for_v2 = factory_with_multiple_versions.for_json_schema_version(2) + + expect(for_v1).not_to eq(for_v2) + expect(factory_with_multiple_versions.for_json_schema_version(1)).to be for_v1 + end + end + + describe "#for_latest_json_schema_version" do + it "returns the record preparer for the latest JSON schema version" do + for_v2 = factory_with_multiple_versions.for_json_schema_version(2) + + expect(factory_with_multiple_versions.for_latest_json_schema_version).to be for_v2 + end + end + end + + RSpec.describe RecordPreparerFactory, "record preparation for old JSON schema versions", :json_ingestion_schema_definition do + context "when working with events for an old JSON schema version" do + include_context "SchemaDefinitionHelpers" + + it "handles events for old versions before a field was deleted" do + preparer = build_preparer_for_old_json_schema_version( + v1_def: ->(schema) { + schema.object_type "MyType" do |t| + t.field "id", "ID!" + t.field "name", "String" + t.index "my_type" + end + }, + + v2_def: ->(schema) { + schema.object_type "MyType" do |t| + t.field "id", "ID!" + t.deleted_field "name" + t.index "my_type" + end + } + ) + + record = preparer.prepare_for_index("MyType", {"id" => "1", "name" => "Winston"}, {}) + + expect(record).to eq({"id" => "1"}) + end + + it "properly omits `__typename` under an embedded field for a non-abstract type, even when the type has been renamed" do + preparer = build_preparer_for_old_json_schema_version( + v1_def: ->(schema) { + schema.object_type "MyType" do |t| + t.field "id", "ID!" + t.field "cost", "Money" + t.index "my_type" + end + + schema.object_type "Money" do |t| + t.field "amount", "Int" + end + }, + + v2_def: ->(schema) { + schema.object_type "MyType" do |t| + t.field "id", "ID!" + t.field "cost", "Money2" + t.index "my_type" + end + + schema.object_type "Money2" do |t| + t.field "amount", "Int" + t.renamed_from "Money" + end + } + ) + + record = preparer.prepare_for_index("MyType", {"id" => "1", "cost" => {"amount" => 10, "__typename" => "Money"}}, {}) + + expect(record).to eq({"id" => "1", "cost" => {"amount" => 10}}) + end + + def build_preparer_for_old_json_schema_version(v1_def:, v2_def:) + v1_results = define_schema do |schema| + schema.json_schema_version 1 + v1_def.call(schema) + end + + v2_results = define_schema do |schema| + schema.json_schema_version 2 + v2_def.call(schema) + end + + v1_merge_result = v2_results.merge_field_metadata_into_json_schema(v1_results.current_public_json_schema) + + expect(v1_merge_result.missing_fields).to be_empty + expect(v1_merge_result.missing_types).to be_empty + + allow(v2_results).to receive(:json_schemas_for).with(1).and_return(v1_merge_result.json_schema) + + RecordPreparerFactory.new(v2_results).for_json_schema_version(1) + end + + def define_schema(&schema_definition) + super(schema_element_name_form: "snake_case", &schema_definition) + end + end + end + end +end diff --git a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/spec_support/event_matcher_spec.rb b/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/spec_support/event_matcher_spec.rb similarity index 97% rename from elasticgraph-indexer/spec/unit/elastic_graph/indexer/spec_support/event_matcher_spec.rb rename to elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/spec_support/event_matcher_spec.rb index 22ef35b28..afb594e19 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/spec_support/event_matcher_spec.rb +++ b/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/spec_support/event_matcher_spec.rb @@ -6,7 +6,7 @@ # # frozen_string_literal: true -require "elastic_graph/indexer/spec_support/event_matcher" +require "elastic_graph/json_ingestion/spec_support/event_matcher" require "rspec/matchers/fail_matchers" RSpec.describe "The `be_a_valid_elastic_graph_event` matcher", :builds_indexer, :factories, aggregate_failures: false do diff --git a/elasticgraph-lambda_support/elasticgraph-lambda_support.gemspec b/elasticgraph-lambda_support/elasticgraph-lambda_support.gemspec index bcfab1dc9..6bfd3b52e 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 indexers from the shared test schema artifacts, whose runtime metadata + # registers the JSON ingestion indexer extension 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/common_project_specs.rb b/elasticgraph-local/lib/elastic_graph/local/spec_support/common_project_specs.rb index 65d49ea55..f96f747aa 100644 --- a/elasticgraph-local/lib/elastic_graph/local/spec_support/common_project_specs.rb +++ b/elasticgraph-local/lib/elastic_graph/local/spec_support/common_project_specs.rb @@ -10,7 +10,7 @@ require "elastic_graph/graphql" require "elastic_graph/indexer" require "elastic_graph/indexer/test_support/converters" -require "elastic_graph/indexer/spec_support/event_matcher" +require "elastic_graph/json_ingestion/spec_support/event_matcher" require "elastic_graph/support/json_schema/validator" require "factory_bot" require "json_schemer" diff --git a/elasticgraph-warehouse_lambda/elasticgraph-warehouse_lambda.gemspec b/elasticgraph-warehouse_lambda/elasticgraph-warehouse_lambda.gemspec index 2c9b7f706..cfaf33432 100644 --- a/elasticgraph-warehouse_lambda/elasticgraph-warehouse_lambda.gemspec +++ b/elasticgraph-warehouse_lambda/elasticgraph-warehouse_lambda.gemspec @@ -48,5 +48,8 @@ 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`. + spec.add_development_dependency "elasticgraph-json_ingestion", ElasticGraph::VERSION spec.add_development_dependency "elasticgraph-opensearch", ElasticGraph::VERSION end diff --git a/spec_support/lib/elastic_graph/spec_support/builds_indexer_operation.rb b/spec_support/lib/elastic_graph/spec_support/builds_indexer_operation.rb index b9e7356b9..563621e26 100644 --- a/spec_support/lib/elastic_graph/spec_support/builds_indexer_operation.rb +++ b/spec_support/lib/elastic_graph/spec_support/builds_indexer_operation.rb @@ -32,7 +32,7 @@ def new_primary_indexing_operation(event, index_def: nil, idxr: indexer) Indexer::Operation::Update.new( event: event, - prepared_record: idxr.record_preparer_factory.for_latest_json_schema_version.prepare_for_index( + prepared_record: latest_json_record_preparer_for(idxr).prepare_for_index( event.fetch("type"), event.fetch("record"), idxr.schema_artifacts.index_mappings_by_index_def_name.fetch(index_def.name).fetch("properties") diff --git a/spec_support/lib/elastic_graph/spec_support/uses_datastore.rb b/spec_support/lib/elastic_graph/spec_support/uses_datastore.rb index a2f0ba795..aecccfcf4 100644 --- a/spec_support/lib/elastic_graph/spec_support/uses_datastore.rb +++ b/spec_support/lib/elastic_graph/spec_support/uses_datastore.rb @@ -331,7 +331,7 @@ def index_into(indexer, *records) destination_index_mapping = indexer.schema_artifacts.index_mappings_by_index_def_name.fetch(index_def.name) ElasticGraph::Indexer::Operation::Update.new( event: event, - prepared_record: indexer.record_preparer_factory.for_latest_json_schema_version.prepare_for_index( + prepared_record: latest_json_record_preparer_for(indexer).prepare_for_index( event.fetch("type"), event.fetch("record"), destination_index_mapping.fetch("properties") diff --git a/spec_support/spec_helper.rb b/spec_support/spec_helper.rb index c9ccda223..8d9dc4139 100644 --- a/spec_support/spec_helper.rb +++ b/spec_support/spec_helper.rb @@ -334,6 +334,13 @@ def parsed_test_settings_yaml CommonSpecHelpers.parsed_test_settings_yaml end + # Provides the record preparer an indexer would use for JSON events at the latest JSON schema + # version, for tests that need to build expected operations "as the indexer would". + def latest_json_record_preparer_for(indexer) + require "elastic_graph/json_ingestion/record_preparer_factory" + JSONIngestion::RecordPreparerFactory.new(indexer.schema_artifacts).for_latest_json_schema_version + end + module_function def expect_to_return_non_nil_values_from_all_attributes(object)