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)