From 8bccd91b9b38ad379dd6288cf77258fa8e97341c Mon Sep 17 00:00:00 2001 From: Josh Wilson Date: Sat, 12 Sep 2026 16:41:02 -0500 Subject: [PATCH] Move JSON ingestion into elasticgraph-json_ingestion via an indexer extension Move JSON validation and record preparation out of the indexer and into the JSON ingestion extension. Register the adapter through schema runtime metadata while keeping the indexer core ingestion-format neutral. --- CODEBASE_OVERVIEW.md | 6 + Gemfile.lock | 2 + Steepfile | 4 +- config/schema/artifacts/runtime_metadata.yaml | 4 + .../runtime_metadata.yaml | 4 + .../elasticgraph-apollo.gemspec | 1 + ...elasticgraph_graphql_acceptance_support.rb | 4 +- .../datastore_query_integration_support.rb | 4 +- elasticgraph-indexer/README.md | 3 + .../lib/elastic_graph/indexer.rb | 25 +- .../lib/elastic_graph/indexer/event_id.rb | 2 +- .../indexer/ingestion_adapter.rb | 4 +- .../indexer/ingestion_adapter/json_events.rb | 134 -------- .../elastic_graph/indexer/record_preparer.rb | 53 ---- .../sig/elastic_graph/indexer.rbs | 6 +- .../indexer/ingestion_adapter/json_events.rbs | 30 -- .../elastic_graph/indexer/record_preparer.rbs | 14 - .../spec/support/indexing_preparer.rb | 4 +- .../indexer/datastore_indexing_router_spec.rb | 4 +- .../operation/count_accumulator_spec.rb | 4 +- .../indexer/operation/factory_spec.rb | 240 +------------- .../indexer/operation/update_spec.rb | 4 +- .../elastic_graph/indexer/processor_spec.rb | 2 +- .../indexer/record_preparer_spec.rb | 138 +------- .../spec/unit/elastic_graph/indexer_spec.rb | 31 +- .../elasticgraph-indexer_lambda.gemspec | 2 + elasticgraph-json_ingestion/README.md | 19 ++ .../elasticgraph-json_ingestion.gemspec | 10 +- .../json_ingestion/indexer_extension.rb | 25 ++ .../json_ingestion/ingestion_adapter.rb | 142 +++++++++ .../json_ingestion/record_preparer_factory.rb | 74 +++++ .../schema_definition/api_extension.rb | 6 + .../spec_support/event_matcher.rb | 8 +- .../json_ingestion/indexer_extension.rbs | 7 + .../json_ingestion/ingestion_adapter.rbs | 30 ++ .../record_preparer_factory.rbs | 17 + .../spec/spec_helper.rb | 4 +- .../spec/support/multiple_version_support.rb | 8 +- .../json_ingestion/indexer_extension_spec.rb | 59 ++++ .../json_ingestion/ingestion_adapter_spec.rb | 295 ++++++++++++++++++ .../record_preparer_factory_spec.rb | 145 +++++++++ .../schema_definition/api_extension_spec.rb | 32 ++ .../spec_support/event_matcher_spec.rb | 2 +- .../elasticgraph-lambda_support.gemspec | 1 + .../spec_support/common_project_specs.rb | 2 +- elasticgraph-schema_definition/README.md | 3 + .../elasticgraph-warehouse_lambda.gemspec | 1 + .../warehouse_lambda/warehouse_dumper_spec.rb | 2 +- .../spec_support/builds_indexer.rb | 2 +- .../spec_support/builds_indexer_operation.rb | 2 +- .../spec_support/ingests_json_data.rb | 2 +- ...efinition.rb => json_ingestion_support.rb} | 19 +- .../spec_support/uses_datastore.rb | 4 +- spec_support/spec_helper.rb | 4 +- 54 files changed, 980 insertions(+), 674 deletions(-) delete mode 100644 elasticgraph-indexer/lib/elastic_graph/indexer/ingestion_adapter/json_events.rb delete mode 100644 elasticgraph-indexer/sig/elastic_graph/indexer/ingestion_adapter/json_events.rbs create mode 100644 elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/indexer_extension.rb create mode 100644 elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/ingestion_adapter.rb create mode 100644 elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion/record_preparer_factory.rb rename {elasticgraph-indexer/lib/elastic_graph/indexer => elasticgraph-json_ingestion/lib/elastic_graph/json_ingestion}/spec_support/event_matcher.rb (83%) create mode 100644 elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/indexer_extension.rbs create mode 100644 elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/ingestion_adapter.rbs create mode 100644 elasticgraph-json_ingestion/sig/elastic_graph/json_ingestion/record_preparer_factory.rbs rename {elasticgraph-indexer => elasticgraph-json_ingestion}/spec/support/multiple_version_support.rb (88%) create mode 100644 elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/indexer_extension_spec.rb create mode 100644 elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/ingestion_adapter_spec.rb create mode 100644 elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/record_preparer_factory_spec.rb create mode 100644 elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion/schema_definition/api_extension_spec.rb rename {elasticgraph-indexer/spec/unit/elastic_graph/indexer => elasticgraph-json_ingestion/spec/unit/elastic_graph/json_ingestion}/spec_support/event_matcher_spec.rb (97%) rename spec_support/lib/elastic_graph/spec_support/{json_ingestion_schema_definition.rb => json_ingestion_support.rb} (65%) 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