diff --git a/CODEBASE_OVERVIEW.md b/CODEBASE_OVERVIEW.md index 543711c4c..c91a5abe9 100644 --- a/CODEBASE_OVERVIEW.md +++ b/CODEBASE_OVERVIEW.md @@ -219,6 +219,8 @@ graph LR; elasticgraph-health_check["eg-health_check"]; elasticgraph-datastore_core["eg-datastore_core"]; elasticgraph-json_ingestion["eg-json_ingestion"]; + elasticgraph-indexer["eg-indexer"]; + elasticgraph-schema_definition["eg-schema_definition"]; elasticgraph-proto_ingestion["eg-proto_ingestion"]; elasticgraph-query_interceptor["eg-query_interceptor"]; elasticgraph-schema_artifacts["eg-schema_artifacts"]; @@ -232,7 +234,9 @@ graph LR; elasticgraph-health_check --> elasticgraph-datastore_core; elasticgraph-health_check --> elasticgraph-graphql; elasticgraph-health_check --> elasticgraph-support; + elasticgraph-json_ingestion --> elasticgraph-indexer; elasticgraph-json_ingestion --> elasticgraph-support; + elasticgraph-json_ingestion --> elasticgraph-schema_definition; elasticgraph-proto_ingestion --> elasticgraph-support; elasticgraph-query_interceptor --> elasticgraph-graphql; elasticgraph-query_interceptor --> elasticgraph-schema_artifacts; @@ -249,6 +253,8 @@ graph LR; class elasticgraph-health_check targetGemStyle; class elasticgraph-datastore_core otherEgGemStyle; class elasticgraph-json_ingestion targetGemStyle; + class elasticgraph-indexer otherEgGemStyle; + class elasticgraph-schema_definition otherEgGemStyle; class elasticgraph-proto_ingestion targetGemStyle; class elasticgraph-query_interceptor targetGemStyle; class elasticgraph-schema_artifacts otherEgGemStyle; diff --git a/Gemfile.lock b/Gemfile.lock index 54a76bf63..d91dc8f22 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -109,6 +109,8 @@ PATH remote: elasticgraph-json_ingestion specs: elasticgraph-json_ingestion (1.2.1.pre) + elasticgraph-indexer (= 1.2.1.pre) + elasticgraph-schema_definition (= 1.2.1.pre) elasticgraph-support (= 1.2.1.pre) PATH 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 6b418ee01..64f9e1c7b 100644 --- a/config/schema/artifacts/runtime_metadata.yaml +++ b/config/schema/artifacts/runtime_metadata.yaml @@ -3114,6 +3114,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 825492908..bc19b0a29 100644 --- a/config/schema/artifacts_with_apollo/runtime_metadata.yaml +++ b/config/schema/artifacts_with_apollo/runtime_metadata.yaml @@ -3143,6 +3143,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 e883d482d..5e750254b 100644 --- a/elasticgraph-apollo/elasticgraph-apollo.gemspec +++ b/elasticgraph-apollo/elasticgraph-apollo.gemspec @@ -52,6 +52,7 @@ 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 + 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-graphql/spec/acceptance/elasticgraph_graphql_acceptance_support.rb b/elasticgraph-graphql/spec/acceptance/elasticgraph_graphql_acceptance_support.rb index 1d5d0b4e6..f141f6226 100644 --- a/elasticgraph-graphql/spec/acceptance/elasticgraph_graphql_acceptance_support.rb +++ b/elasticgraph-graphql/spec/acceptance/elasticgraph_graphql_acceptance_support.rb @@ -20,13 +20,13 @@ class << self self.camel_case_cluster_configured = false - # We list the `:ingests_json_data` component tags (`:json_ingestion_schema_definition`, + # We list the `:ingests_json_data` component tags (`:json_ingestion_support`, # `:uses_datastore`, `:builds_indexer`) individually rather than using `:ingests_json_data` # itself. Applying the derived `:ingests_json_data` tag to a `shared_context` and then combining # it with `include_examples` crashes RSpec (`undefined method 'key?' for nil`), whereas literal # tags are inherited correctly. Once https://github.com/rspec/rspec/issues/333 is fixed we can # collapse these back to `:ingests_json_data`. - RSpec.shared_context "ElasticGraph GraphQL acceptance support", :factories, :json_ingestion_schema_definition, :uses_datastore, :builds_indexer, :capture_logs, :builds_admin do + RSpec.shared_context "ElasticGraph GraphQL acceptance support", :factories, :json_ingestion_support, :uses_datastore, :builds_indexer, :capture_logs, :builds_admin do include GraphQLSupport include PreventSearchesFromUsingWriteRequests diff --git a/elasticgraph-graphql/spec/integration/elastic_graph/graphql/datastore_query/datastore_query_integration_support.rb b/elasticgraph-graphql/spec/integration/elastic_graph/graphql/datastore_query/datastore_query_integration_support.rb index 1ead5c1af..6154b4dc2 100644 --- a/elasticgraph-graphql/spec/integration/elastic_graph/graphql/datastore_query/datastore_query_integration_support.rb +++ b/elasticgraph-graphql/spec/integration/elastic_graph/graphql/datastore_query/datastore_query_integration_support.rb @@ -19,13 +19,13 @@ class GraphQL # return a partial response that our router handled by logging a warning. The # `:capture_logs` here caused the tests to fail until we fixed that issue. # - # We list the `:ingests_json_data` component tags (`:json_ingestion_schema_definition`, + # We list the `:ingests_json_data` component tags (`:json_ingestion_support`, # `:uses_datastore`, `:builds_indexer`) individually rather than using `:ingests_json_data` # itself. Applying the derived `:ingests_json_data` tag to a `shared_context` and then combining # it with `include_examples` crashes RSpec (`undefined method 'key?' for nil`), whereas literal # tags are inherited correctly. Once https://github.com/rspec/rspec/issues/333 is fixed we can # collapse these back to `:ingests_json_data`. - RSpec.shared_context "DatastoreQueryIntegrationSupport", :json_ingestion_schema_definition, :uses_datastore, :builds_indexer, :factories, :capture_logs do + RSpec.shared_context "DatastoreQueryIntegrationSupport", :json_ingestion_support, :uses_datastore, :builds_indexer, :factories, :capture_logs do include AggregationsHelpers include SortSupport diff --git a/elasticgraph-indexer/README.md b/elasticgraph-indexer/README.md index e4f82753b..0d7eed349 100644 --- a/elasticgraph-indexer/README.md +++ b/elasticgraph-indexer/README.md @@ -29,6 +29,9 @@ graph LR; elasticgraph-indexer_lambda["elasticgraph-indexer_lambda"]; elasticgraph-indexer_lambda --> elasticgraph-indexer; class elasticgraph-indexer_lambda otherEgGemStyle; + elasticgraph-json_ingestion["elasticgraph-json_ingestion"]; + elasticgraph-json_ingestion --> elasticgraph-indexer; + class elasticgraph-json_ingestion otherEgGemStyle; elasticgraph-local["elasticgraph-local"]; elasticgraph-local --> elasticgraph-indexer; class elasticgraph-local otherEgGemStyle; diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer.rb b/elasticgraph-indexer/lib/elastic_graph/indexer.rb index ca552edc8..f2073f7c2 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" @@ -38,7 +39,7 @@ def initialize( @datastore_core = datastore_core @logger = datastore_core.logger @datastore_router = datastore_router - @ingestion_adapters_by_format = ingestion_adapters_by_format + @injected_ingestion_adapters_by_format = ingestion_adapters_by_format @schema_artifacts = @datastore_core.schema_artifacts @monotonic_clock = monotonic_clock @clock = clock || ::Time @@ -58,21 +59,9 @@ 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, keyed by the format tag used by - # their 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, keyed by format. def ingestion_adapters_by_format - @ingestion_adapters_by_format ||= begin - require "elastic_graph/indexer/ingestion_adapter/json_events" - {"json" => IngestionAdapter::JSONEvents.new(schema_artifacts: schema_artifacts, logger: logger)} - end + @ingestion_adapters_by_format ||= @injected_ingestion_adapters_by_format || {} end def processor @@ -90,6 +79,12 @@ def processor def operation_factory @operation_factory ||= begin + if ingestion_adapters_by_format.empty? + raise Errors::ConfigError, "No ingestion adapters are available. Enable an ingestion format extension " \ + "in your schema definition and regenerate the schema artifacts, or configure `indexer.extension_modules` " \ + "with an extension module that registers an ingestion adapter." + 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 483f1e38c..4e5e600c5 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_by_format, logger + # @dynamic processor, operation_factory, ingestion_adapters_by_format, logger # @dynamic self.from_parsed_yaml end end diff --git a/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter.rb b/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter.rb index ba16b7cba..cdbcf5198 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter.rb @@ -8,8 +8,8 @@ module ElasticGraph class Indexer - # Namespace for ingestion adapters. An ingestion adapter teaches the indexer how to handle - # events of a particular ingestion format: it validates each event and provides the + # Namespace for ingestion adapters. An ingestion adapter handles events of a particular + # ingestion format: it validates each event and provides the # version-appropriate machinery to prepare the event's record for indexing. module IngestionAdapter # Defines the ingestion adapter interface. Adapter classes are not required to subclass this, 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 35dba1deb..000000000 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter/json_events.rb +++ /dev/null @@ -1,134 +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#validate_event) - def validate_event(event, skip_record_validation: false) - 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(validation_target: "event payload", message: error_message) - end - - record = event.fetch("record") - graphql_type_name = event.fetch("type") - - if !skip_record_validation && (error_message = validator(graphql_type_name, selected_json_schema_version).validate_with_error_message(record)) - return ValidationResult.invalid(validation_target: "#{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(validation_target: 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(validation_target: 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( - validation_target: 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, raw_json_schema_version| - json_schema_version = raw_json_schema_version # : Integer - 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 - - # simplecov:disable -- 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, return or yield " \ - "`ValidationResult.invalid(...)` so that we can accumulate all invalid events and allow " \ - "the valid events to still be processed.") - end - # simplecov:enable - 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 397edce64..f21b73e27 100644 --- a/elasticgraph-indexer/lib/elastic_graph/indexer/record_preparer.rb +++ b/elasticgraph-indexer/lib/elastic_graph/indexer/record_preparer.rb @@ -11,59 +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, raw_version| - version = raw_version # : Integer - 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 b7d0d5159..852fe2499 100644 --- a/elasticgraph-indexer/sig/elastic_graph/indexer.rbs +++ b/elasticgraph-indexer/sig/elastic_graph/indexer.rbs @@ -20,23 +20,19 @@ module ElasticGraph @datastore_router: Indexer::_DatastoreRouter? def datastore_router: () -> Indexer::_DatastoreRouter - @record_preparer_factory: RecordPreparer::Factory? - def record_preparer_factory: () -> RecordPreparer::Factory - @processor: Processor? def processor: () -> Processor @operation_factory: Operation::Factory? def operation_factory: () -> Operation::Factory + @injected_ingestion_adapters_by_format: ::Hash[::String, IngestionAdapter::_Adapter]? @ingestion_adapters_by_format: ::Hash[::String, IngestionAdapter::_Adapter]? def ingestion_adapters_by_format: () -> ::Hash[::String, IngestionAdapter::_Adapter] @monotonic_clock: Support::MonotonicClock? def monotonic_clock: () -> Support::MonotonicClock - private - @clock: singleton(::Time) end end 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 9921e55a5..000000000 --- a/elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter/json_events.rbs +++ /dev/null @@ -1,30 +0,0 @@ -module ElasticGraph - class Indexer - module IngestionAdapter - class JSONEvents - include _Adapter - - 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 - - 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..b79626d59 100644 --- a/elasticgraph-indexer/spec/support/indexing_preparer.rb +++ b/elasticgraph-indexer/spec/support/indexing_preparer.rb @@ -12,7 +12,7 @@ # For example, to see your indexing preparer being used, this requires that it is registered # on your scalar type correctly, whereas if this directly called your indexing preparer # it wouldn't require it to be correctly registered. -RSpec.shared_context "indexing preparer support", :json_ingestion_schema_definition do |scalar_type| +RSpec.shared_context "indexing preparer support", :json_ingestion_support do |scalar_type| before(:context) do @record_preparer = build_indexer(clients_by_name: {}, schema_definition: lambda do |schema| schema.object_type "MyType" do |t| @@ -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..194428250 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 @@ -15,7 +15,7 @@ module ElasticGraph class Indexer - RSpec.describe DatastoreIndexingRouter, :json_ingestion_schema_definition, :capture_logs do + RSpec.describe DatastoreIndexingRouter, :json_ingestion_support, :capture_logs do include SpecSupport::BuildsIndexerOperation let(:main_datastore_client) { instance_spy(Elasticsearch::Client, cluster_name: "main") } @@ -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..2545871a6 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 @@ -12,7 +12,7 @@ module ElasticGraph class Indexer module Operation - RSpec.describe CountAccumulator do + RSpec.describe CountAccumulator, :json_ingestion_support do describe "counts indexed on the root document" do it "includes counts for each list field" do counts = root_counts_for_team_event({"past_names" => ["a", "b", "c"], "seasons_object" => []}) @@ -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 7d5f3defc..a7f5cd30b 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,7 +8,7 @@ require "elastic_graph/constants" require "elastic_graph/indexer" -require "elastic_graph/indexer/ingestion_adapter/json_events" +require "elastic_graph/indexer/ingestion_adapter" require "elastic_graph/indexer/operation/factory" require "elastic_graph/spec_support/builds_indexer_operation" require "json" @@ -16,7 +16,7 @@ module ElasticGraph class Indexer module Operation - RSpec.describe Factory, :capture_logs do + RSpec.describe Factory, :json_ingestion_support, :capture_logs do describe "#build", :factories do include SpecSupport::BuildsIndexerOperation @@ -339,43 +339,6 @@ def factory_whose_record_preparation_is_broken }.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", @@ -406,30 +369,6 @@ def factory_whose_record_preparation_is_broken 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") @@ -437,12 +376,6 @@ def factory_whose_record_preparation_is_broken 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) @@ -484,152 +417,6 @@ def factory_whose_record_preparation_is_broken 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 is 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"]) @@ -641,7 +428,7 @@ def factory_whose_record_preparation_is_broken expect(operations.map(&:doc_id)).to contain_exactly("c1", "c2", "c3") end - context "for an event of a non-indexed `sourced_from` source type", :json_ingestion_schema_definition do + context "for an event of a non-indexed `sourced_from` source type", :json_ingestion_support do let(:indexer) do build_indexer(schema_definition: lambda do |s| s.object_type "Widget" do |t| @@ -762,23 +549,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_by_format: {"json" => json_events_adapter}) - end - - result = factory.build(event, **options) + def build_expecting_success(event, **options) + result = indexer.operation_factory.build(event, **options) - failure = result.failed_event_error - raise failure if failure + expect(result.failed_event_error).to be nil result.operations end @@ -786,7 +560,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: latest_json_record_preparer_for(indexer), 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..38d33c7be 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 @@ -15,7 +15,7 @@ module ElasticGraph class Indexer module Operation - RSpec.describe Update, :json_ingestion_schema_definition do + RSpec.describe Update, :json_ingestion_support do include SchemaArtifacts::RuntimeMetadata::RuntimeMetadataSupport let(:indexer) { build_indexer } @@ -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/processor_spec.rb b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/processor_spec.rb index 83e29b392..244f78e80 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/processor_spec.rb +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/processor_spec.rb @@ -16,7 +16,7 @@ module ElasticGraph class Indexer - RSpec.describe Processor do + RSpec.describe Processor, :json_ingestion_support do describe ".process", :factories, :capture_logs do include SpecSupport::BuildsIndexerOperation 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..2489c4b82 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,54 +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" 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 + RSpec.describe RecordPreparer, :json_ingestion_support do describe "#prepare_for_index" do it "tolerates a `nil` value where an object would usually be" do preparer = build_preparer do |s| @@ -494,101 +450,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 = latest_json_record_preparer_for(indexer) [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 8e491abd6..2551ea084 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer_spec.rb +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer_spec.rb @@ -14,10 +14,10 @@ module ElasticGraph expect_to_return_non_nil_values_from_all_attributes(build_indexer) end - it "accepts injected ingestion adapters by format" do + it "preserves injected ingestion adapters by format" do adapters = {"test" => Object.new} - expect(build_indexer(ingestion_adapters_by_format: adapters).ingestion_adapters_by_format).to be(adapters) + expect(build_indexer(ingestion_adapters_by_format: adapters).ingestion_adapters_by_format).to include(adapters) end describe ".from_parsed_yaml" do @@ -36,6 +36,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 @@ -43,14 +58,14 @@ def datastore_router :config_router end end - stub_const("ConfigExtensionModule", config_extension_module) + stub_const("IndexerConfigExtensionModule", 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) + stub_const("IndexerRuntimeMetadataExtensionModule", runtime_metadata_extension_module) extended_indexer = build_indexer( extension_modules: [config_extension_module], @@ -67,10 +82,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 c71593753..47ff8305f 100644 --- a/elasticgraph-indexer_lambda/elasticgraph-indexer_lambda.gemspec +++ b/elasticgraph-indexer_lambda/elasticgraph-indexer_lambda.gemspec @@ -44,4 +44,6 @@ 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.229" + + 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..6ae21b124 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 provides an adapter used by `elasticgraph-indexer` to ingest JSON events. The +adapter 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. + +JSON ingestion is enabled automatically for schemas defined with this gem's `SchemaDefinition::APIExtension`; +no indexer configuration is needed. + +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 @@ -111,7 +124,13 @@ graph LR; classDef externalGemStyle fill:#E0EFFF,stroke:#70A1D7,color:#2980B9; elasticgraph-json_ingestion["elasticgraph-json_ingestion"]; class elasticgraph-json_ingestion targetGemStyle; + elasticgraph-indexer["elasticgraph-indexer"]; + elasticgraph-json_ingestion --> elasticgraph-indexer; + class elasticgraph-indexer otherEgGemStyle; elasticgraph-support["elasticgraph-support"]; elasticgraph-json_ingestion --> elasticgraph-support; class elasticgraph-support otherEgGemStyle; + elasticgraph-schema_definition["elasticgraph-schema_definition"]; + elasticgraph-json_ingestion --> elasticgraph-schema_definition; + class elasticgraph-schema_definition otherEgGemStyle; ``` diff --git a/elasticgraph-json_ingestion/elasticgraph-json_ingestion.gemspec b/elasticgraph-json_ingestion/elasticgraph-json_ingestion.gemspec index cdade27c0..46039cd77 100644 --- a/elasticgraph-json_ingestion/elasticgraph-json_ingestion.gemspec +++ b/elasticgraph-json_ingestion/elasticgraph-json_ingestion.gemspec @@ -34,10 +34,12 @@ Gem::Specification.new do |spec| spec.required_ruby_version = [">= 3.4", "< 4.1"] + spec.add_dependency "elasticgraph-indexer", ElasticGraph::VERSION spec.add_dependency "elasticgraph-support", 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. - spec.add_development_dependency "elasticgraph-schema_definition", ElasticGraph::VERSION + spec.add_dependency "elasticgraph-schema_definition", ElasticGraph::VERSION + + # Indexer specs use real datastore client classes. + spec.add_development_dependency "elasticgraph-elasticsearch", ElasticGraph::VERSION + spec.add_development_dependency "elasticgraph-opensearch", ElasticGraph::VERSION end 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..a10c9a86a --- /dev/null +++ b/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexer_extension.rb @@ -0,0 +1,25 @@ +# 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 + # Provides the default JSON adapter. Registered by {SchemaDefinition::APIExtension}. + module IndexerExtension + # Returns the configured ingestion adapters with the built-in JSON adapter registered. + # + # @return [Hash] + def ingestion_adapters_by_format + @ingestion_adapters_by_format ||= super.merge( + "json" => 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..120db2a45 --- /dev/null +++ b/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/ingestion_adapter.rb @@ -0,0 +1,142 @@ +# 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 + + # Validates the given event and resolves the record preparer appropriate for the event's + # JSON schema version. + # + # @param event [Hash] an ElasticGraph indexing event + # @param skip_record_validation [Boolean] whether to skip record validation; the event envelope must still be validated + # @return [Indexer::IngestionAdapter::ValidationResult] the result of validating the event + def validate_event(event, skip_record_validation: false) + 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(validation_target: "event payload", message: error_message) + end + + record = event.fetch("record") + graphql_type_name = event.fetch("type") + + if !skip_record_validation && (error_message = validator(graphql_type_name, selected_json_schema_version).validate_with_error_message(record)) + return ValidationResult.invalid(validation_target: "#{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(validation_target: 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(validation_target: 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( + validation_target: 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, raw_json_schema_version| + json_schema_version = raw_json_schema_version # : Integer + 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 + + # simplecov:disable -- 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, return or yield " \ + "`ValidationResult.invalid(...)` so that we can accumulate all invalid events and allow " \ + "the valid events to still be processed.") + end + # simplecov:enable + 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..4f3a879fc --- /dev/null +++ b/elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/record_preparer_factory.rb @@ -0,0 +1,74 @@ +# 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, raw_version| + version = raw_version # : Integer + 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..f611aa2f6 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,12 @@ 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. + indexer_extension_require_path = "elastic_graph/json_ingestion/indexer_extension" + require indexer_extension_require_path + 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 83% 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 7a982a8f7..10ff713ff 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_by_format: {"json" => json_events_adapter}) + .with(ingestion_adapters_by_format: {"json" => 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..034182297 --- /dev/null +++ b/elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/indexer_extension.rbs @@ -0,0 +1,7 @@ +module ElasticGraph + module JSONIngestion + module IndexerExtension : ::ElasticGraph::Indexer + def ingestion_adapters_by_format: () -> ::Hash[::String, 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..15d391c48 --- /dev/null +++ b/elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/ingestion_adapter.rbs @@ -0,0 +1,30 @@ +module ElasticGraph + module JSONIngestion + class IngestionAdapter + include Indexer::IngestionAdapter::_Adapter + + 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 + + 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..9dacfd79b 100644 --- a/elasticgraph-json_ingestion/spec/spec_helper.rb +++ b/elasticgraph-json_ingestion/spec/spec_helper.rb @@ -14,8 +14,8 @@ RSpec.configure do |config| config.define_derived_metadata(absolute_file_path: %r{/elasticgraph-json_ingestion/}) do |meta| - meta[:json_ingestion_schema_definition] = true + meta[:json_ingestion_support] = true end - config.include_context "SchemaDefinitionHelpers", :json_ingestion_schema_definition + config.include_context "SchemaDefinitionHelpers", :json_ingestion_support end diff --git a/elasticgraph-indexer/spec/support/multiple_version_support.rb b/elasticgraph-json_ingestion/spec/support/multiple_version_support.rb similarity index 88% rename from elasticgraph-indexer/spec/support/multiple_version_support.rb rename to elasticgraph-json_ingestion/spec/support/multiple_version_support.rb index e3b044aab..7bdb9e0a2 100644 --- a/elasticgraph-indexer/spec/support/multiple_version_support.rb +++ b/elasticgraph-json_ingestion/spec/support/multiple_version_support.rb @@ -9,11 +9,11 @@ require "elastic_graph/spec_support/schema_definition_helpers" module ElasticGraph - class Indexer - ::RSpec.shared_context "MultipleVersionSupport", :json_ingestion_schema_definition do + module JSONIngestion + ::RSpec.shared_context "MultipleVersionSupport", :json_ingestion_support do include_context "SchemaDefinitionHelpers" - def build_indexer_with_multiple_schema_versions(schema_versions:) + def build_schema_artifacts_with_multiple_versions(schema_versions:) results_by_version = schema_versions.to_h do |json_schema_version, prior_def| results = define_schema(schema_element_name_form: :snake_case, json_schema_version: json_schema_version, &prior_def) [json_schema_version, results] @@ -31,7 +31,7 @@ def build_indexer_with_multiple_schema_versions(schema_versions:) json_schemas_by_version.fetch(version) end - build_indexer(schema_artifacts: artifacts) + artifacts end end end 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..86af65cf9 --- /dev/null +++ b/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/indexer_extension_spec.rb @@ -0,0 +1,59 @@ +# 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, :builds_indexer, :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_by_format).to match("json" => an_instance_of(IngestionAdapter)) + expect(indexer.ingestion_adapters_by_format).to be(indexer.ingestion_adapters_by_format), "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 + + it "preserves adapters for other formats contributed by a configured extension" do + custom_adapter = Object.new + extension = Module.new do + define_method(:ingestion_adapters_by_format) do + super().merge("custom" => custom_adapter) + end + end + + indexer = build_indexer(extension_modules: [extension]) + + expect(indexer.ingestion_adapters_by_format).to match( + "json" => an_instance_of(IngestionAdapter), + "custom" => custom_adapter + ) + end + + it "adds the JSON adapter to an explicitly empty ingestion adapter registry" do + indexer = build_indexer(ingestion_adapters_by_format: {}) + + expect(indexer.ingestion_adapters_by_format).to match("json" => an_instance_of(IngestionAdapter)) + end + + it "adds the JSON adapter to an explicitly injected ingestion adapter registry" do + adapters = {"custom" => Object.new} + indexer = build_indexer(ingestion_adapters_by_format: adapters) + + expect(indexer.ingestion_adapters_by_format).to match( + "json" => an_instance_of(IngestionAdapter), + "custom" => adapters.fetch("custom") + ) + 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..5dffdd480 --- /dev/null +++ b/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/ingestion_adapter_spec.rb @@ -0,0 +1,295 @@ +# 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" +require "support/multiple_version_support" + +module ElasticGraph + module JSONIngestion + RSpec.describe IngestionAdapter, :capture_logs, :factories, :builds_indexer do + let(:schema_artifacts) { stock_schema_artifacts } + let(:indexer) { build_indexer(schema_artifacts: schema_artifacts) } + describe "#validate_event" do + it "accepts events explicitly tagged as JSON" do + event = build_upsert_event(:component, id: "1").merge(INGESTION_FORMAT_KEY => "json") + + result = indexer.operation_factory.build(event) + + expect(result.failed_event_error).to be nil + expect(result.operations.map(&:doc_id)).to include("1") + end + + # The factory replaces an adapter's missing preparer with an identity preparer while + # building its failed result, so this adapter contract must be asserted directly. + it "returns no record preparer, plus the part of the event at fault, for an event it rejects" do + event = build_upsert_event(:component, id: "1", __version: 1) + event["record"]["name"] = 123 + + adapter = indexer.ingestion_adapters_by_format.fetch("json") + result = adapter.validate_event(event) + + expect(result.record_preparer).to be nil + expect(result.failure.validation_target).to eq("Component record") + end + + context "when record preparation differs between schema versions" do + include_context "MultipleVersionSupport" + + let(:schema_artifacts) do + build_schema_artifacts_with_multiple_versions(schema_versions: [2, 4].to_h do |version| + [version, lambda do |schema| + schema.object_type "Component" do |t| + t.field "id", "ID!" + t.field "name", "String", name_in_index: "name_v#{version}" + t.index "components" + end + end] + end) + end + + { + "an exact match on the oldest available version" => [2, 2], + "an exact match on the newest available version" => [4, 4], + "equidistant between the available versions (newer wins)" => [3, 4], + "above the available range" => [5, 4], + "below the available range" => [1, 2] + }.each do |case_description, (requested_version, selected_version)| + it "prepares an event requesting a version that is #{case_description} using schema version #{selected_version}, while leaving the requested version on the event" do + event = component_event(requested_version) + original_event = Marshal.load(Marshal.dump(event)) + + result = indexer.operation_factory.build(event) + + expect(result.failed_event_error).to be nil + expect(result.operations.map(&:prepared_record)).to eq [{"id" => "1", "name_v#{selected_version}" => "example"}] + expect(result.operations.map { |op| op.event[JSON_SCHEMA_VERSION_KEY] }).to eq [requested_version] + expect(event).to eq(original_event) + end + end + + def component_event(json_schema_version) + { + "op" => "upsert", + "id" => "1", + "type" => "Component", + "version" => 1, + JSON_SCHEMA_VERSION_KEY => json_schema_version, + "record" => {"id" => "1", "name" => "example"} + } + end + 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, validation_target: "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, validation_target: "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, validation_target: "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, validation_target: "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, validation_target: "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, validation_target: "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, validation_target: "event payload", message_including: ["/then"]) + end + + it "notifies an error on missing id" do + event = build_upsert_event(:component).except("id") + + expect_invalid(event, validation_target: "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, validation_target: "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, validation_target: 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, validation_target: "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(build_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, validation_target: "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(indexer.operation_factory.build(event).failed_event_error).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(indexer.operation_factory.build(event).failed_event_error).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 is 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, validation_target: "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(indexer.operation_factory.build(event).failed_event_error).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, validation_target: 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, validation_target: JSON_SCHEMA_VERSION_KEY, message_including: ["Failed to select json schema version"]) + end + + def expect_invalid(event, validation_target:, message_including:) + error = indexer.operation_factory.build(event).failed_event_error + + expect(error.message).to include("Malformed #{validation_target}.", *message_including) + + error.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..fb450371b --- /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_support do + include_context "MultipleVersionSupport" + + let(:factory_with_multiple_versions) do + build_schema_artifacts_with_multiple_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 { |artifacts| RecordPreparerFactory.new(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_support 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-json_ingestion/spec/unit/elastic_graph/json_ingestion/schema_definition/api_extension_spec.rb b/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/schema_definition/api_extension_spec.rb new file mode 100644 index 000000000..518c2a161 --- /dev/null +++ b/elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/schema_definition/api_extension_spec.rb @@ -0,0 +1,32 @@ +# Copyright 2024 - 2026 Block, Inc. +# +# Use of this source code is governed by an MIT-style +# license that can be found in the LICENSE file or at +# https://opensource.org/licenses/MIT. +# +# frozen_string_literal: true + +module ElasticGraph + module JSONIngestion + module SchemaDefinition + RSpec.describe APIExtension do + it "registers the JSON ingestion indexer extension" do + runtime_metadata = define_schema(schema_element_name_form: "snake_case") do |schema| + schema.object_type "Widget" do |type| + type.field "id", "ID!" + type.index "widgets" + end + end.runtime_metadata + + expect(runtime_metadata.indexer_extension_modules.map(&:extension_ref)).to include( + SchemaArtifacts::RuntimeMetadata::Extension.new( + IndexerExtension, + "elastic_graph/json_ingestion/indexer_extension", + {} + ).to_dumpable_hash + ) + 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..105b8b025 100644 --- a/elasticgraph-lambda_support/elasticgraph-lambda_support.gemspec +++ b/elasticgraph-lambda_support/elasticgraph-lambda_support.gemspec @@ -48,4 +48,5 @@ 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 + 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 e1cca9bb3..c2da32b9b 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-schema_definition/README.md b/elasticgraph-schema_definition/README.md index b85f79b9e..ed56e80c5 100644 --- a/elasticgraph-schema_definition/README.md +++ b/elasticgraph-schema_definition/README.md @@ -33,6 +33,9 @@ graph LR; rake["rake"]; elasticgraph-schema_definition --> rake; class rake externalGemStyle; + elasticgraph-json_ingestion["elasticgraph-json_ingestion"]; + elasticgraph-json_ingestion --> elasticgraph-schema_definition; + class elasticgraph-json_ingestion otherEgGemStyle; elasticgraph-local["elasticgraph-local"]; elasticgraph-local --> elasticgraph-schema_definition; class elasticgraph-local otherEgGemStyle; diff --git a/elasticgraph-warehouse_lambda/elasticgraph-warehouse_lambda.gemspec b/elasticgraph-warehouse_lambda/elasticgraph-warehouse_lambda.gemspec index fba1c603a..26c57883c 100644 --- a/elasticgraph-warehouse_lambda/elasticgraph-warehouse_lambda.gemspec +++ b/elasticgraph-warehouse_lambda/elasticgraph-warehouse_lambda.gemspec @@ -48,5 +48,6 @@ Gem::Specification.new do |spec| spec.add_development_dependency "aws_lambda_ric", "~> 3.2" spec.add_development_dependency "elasticgraph-elasticsearch", ElasticGraph::VERSION + spec.add_development_dependency "elasticgraph-json_ingestion", ElasticGraph::VERSION spec.add_development_dependency "elasticgraph-opensearch", ElasticGraph::VERSION end diff --git a/elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda/warehouse_dumper_spec.rb b/elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda/warehouse_dumper_spec.rb index 47109d580..53851f097 100644 --- a/elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda/warehouse_dumper_spec.rb +++ b/elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda/warehouse_dumper_spec.rb @@ -14,7 +14,7 @@ module ElasticGraph class WarehouseLambda - RSpec.describe WarehouseDumper, :capture_logs do + RSpec.describe WarehouseDumper, :json_ingestion_support, :capture_logs do include BuildsWarehouseLambda include SpecSupport::BuildsIndexerOperation diff --git a/spec_support/lib/elastic_graph/spec_support/builds_indexer.rb b/spec_support/lib/elastic_graph/spec_support/builds_indexer.rb index 670d9548a..3fb07ae2f 100644 --- a/spec_support/lib/elastic_graph/spec_support/builds_indexer.rb +++ b/spec_support/lib/elastic_graph/spec_support/builds_indexer.rb @@ -6,9 +6,9 @@ # # frozen_string_literal: true -require "elastic_graph/spec_support/builds_datastore_core" require "elastic_graph/indexer" require "elastic_graph/indexer/config" +require "elastic_graph/spec_support/builds_datastore_core" module ElasticGraph module BuildsIndexer 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/ingests_json_data.rb b/spec_support/lib/elastic_graph/spec_support/ingests_json_data.rb index 49ea13157..26c129382 100644 --- a/spec_support/lib/elastic_graph/spec_support/ingests_json_data.rb +++ b/spec_support/lib/elastic_graph/spec_support/ingests_json_data.rb @@ -7,5 +7,5 @@ # frozen_string_literal: true require "elastic_graph/spec_support/builds_indexer" -require "elastic_graph/spec_support/json_ingestion_schema_definition" +require "elastic_graph/spec_support/json_ingestion_support" require "elastic_graph/spec_support/uses_datastore" diff --git a/spec_support/lib/elastic_graph/spec_support/json_ingestion_schema_definition.rb b/spec_support/lib/elastic_graph/spec_support/json_ingestion_support.rb similarity index 65% rename from spec_support/lib/elastic_graph/spec_support/json_ingestion_schema_definition.rb rename to spec_support/lib/elastic_graph/spec_support/json_ingestion_support.rb index 5cb3bf0e8..3419e3ccf 100644 --- a/spec_support/lib/elastic_graph/spec_support/json_ingestion_schema_definition.rb +++ b/spec_support/lib/elastic_graph/spec_support/json_ingestion_support.rb @@ -6,19 +6,28 @@ # # frozen_string_literal: true +require "elastic_graph/json_ingestion/record_preparer_factory" require "elastic_graph/json_ingestion/schema_definition/api_extension" require "elastic_graph/json_ingestion/schema_definition/test_support" module ElasticGraph - # Wires the JSON ingestion schema definition support into specs tagged - # `:json_ingestion_schema_definition`. Mixing in - # {JSONIngestion::SchemaDefinition::TestSupport} overrides `define_schema` / + # Wires the JSON ingestion support a spec needs into examples tagged `:json_ingestion_support`. + # Everything here is specific to the JSON ingestion format, so a spec gets it by declaring that + # dependency rather than from a format-agnostic helper like `BuildsIndexer`. + # + # Mixing in {JSONIngestion::SchemaDefinition::TestSupport} overrides `define_schema` / # `define_schema_with_schema_elements` to inject `APIExtension` and default the JSON schema version. # `generate_schema_artifacts` is handled separately below because it calls `TestSupport.define_schema` # as a module method (which the instance-level mixin doesn't reach). - module JSONIngestionSchemaDefinition + module JSONIngestionSupport include JSONIngestion::SchemaDefinition::TestSupport + # Reuses JSON record preparers within an example, keyed by schema artifacts. + def latest_json_record_preparer_for(indexer) + preparers = @latest_json_record_preparers_by_schema_artifacts ||= {} + preparers[indexer.schema_artifacts] ||= JSONIngestion::RecordPreparerFactory.new(indexer.schema_artifacts).for_latest_json_schema_version + end + def generate_schema_artifacts(extension_modules: [], **options) super(extension_modules: json_ingestion_schema_definition_extension_modules(extension_modules), **options) do |schema| yield schema @@ -40,6 +49,6 @@ def json_ingestion_schema_definition_extension_modules(extension_modules = []) end RSpec.configure do |config| - config.include JSONIngestionSchemaDefinition, :json_ingestion_schema_definition + config.include JSONIngestionSupport, :json_ingestion_support end end 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 1a8270a03..98695abe0 100644 --- a/spec_support/lib/elastic_graph/spec_support/uses_datastore.rb +++ b/spec_support/lib/elastic_graph/spec_support/uses_datastore.rb @@ -8,8 +8,8 @@ require "digest/md5" require "elastic_graph/elasticsearch/client" -require "elastic_graph/indexer/test_support/converters" require "elastic_graph/indexer/operation/update" +require "elastic_graph/indexer/test_support/converters" require "elastic_graph/support/hash_util" require "logger" require "yaml" @@ -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 025f78c89..402aeb92e 100644 --- a/spec_support/spec_helper.rb +++ b/spec_support/spec_helper.rb @@ -210,7 +210,7 @@ config.raise_on_warning = true config.raise_errors_for_deprecations! - config.when_first_matching_example_defined(:json_ingestion_schema_definition) { require "elastic_graph/spec_support/json_ingestion_schema_definition" } + config.when_first_matching_example_defined(:json_ingestion_support) { require "elastic_graph/spec_support/json_ingestion_support" } config.when_first_matching_example_defined(:uses_datastore) { require "elastic_graph/spec_support/uses_datastore" } config.when_first_matching_example_defined(:ingests_json_data) { require "elastic_graph/spec_support/ingests_json_data" } config.when_first_matching_example_defined(:builds_admin) { require "elastic_graph/spec_support/builds_admin" } @@ -443,7 +443,7 @@ def with_env(overrides) RSpec.configure do |c| c.define_derived_metadata(:ingests_json_data) do |m| - m[:json_ingestion_schema_definition] = true + m[:json_ingestion_support] = true m[:uses_datastore] = true m[:builds_indexer] = true end