diff --git a/CLAUDE.md b/CLAUDE.md index 3848cd4b6..05c2ebc4a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -42,6 +42,7 @@ bundle exec rake opensearch:test:boot ### Documentation - API documentation uses YARD +- 100% documentation coverage is required for all public methods and classes. - Website source: `config/site/` - Example queries: `config/site/examples/*/queries/` - When writing links in documentation, use permalinks (links to a specific commit/version) @@ -152,6 +153,9 @@ Custom gems can be added via `Gemfile-custom` (see `Gemfile-custom.example`), th - Prefer `::Data.define` over `::Struct.new` for immutable data classes. Use `Struct` only when mutability is required. - Don't rely on exceptions for control flow (exceptions are slow). Handle edge cases explicitly instead (e.g., check for `nil` before calling a method that would raise `ArgumentError`). - For constants accessed from multiple EG gems, define them in `elasticgraph-support/lib/elastic_graph/constants.rb`. +- Always put a blank space after `#` in comments. This applies to all comments, including RBS type annotation comments (e.g., `# : String` not `#: String`). +- Avoid defensive code for impossible cases. Don't add checks, tests, or handling for scenarios that should never occur due to the design of the system. If something can only happen due to a bug in the implementation, it's better to fail fast than to silently handle it. +- Drop unnecessary namespace prefixes. When code is already inside `module ElasticGraph`, use `Indexer` instead of `::ElasticGraph::Indexer`, and `Errors::ConfigError` instead of `::ElasticGraph::Errors::ConfigError`. Use fully qualified names only when necessary to avoid ambiguity. ### RBS Type Signatures @@ -160,6 +164,9 @@ Custom gems can be added via `Gemfile-custom` (see `Gemfile-custom.example`), th ### Testing - Avoid duplicate tests. If two tests will always pass/fail together, keep only one. +- Use `expect_to_return_non_nil_values_from_all_attributes` to test wrapper classes (like `WarehouseLambda`, `GraphQL`, `Indexer`, etc.). This automatically exercises every zero-argument method and verifies all dependencies are built successfully. +- Use the `:capture_logs` RSpec tag instead of logger test doubles for verifying log output. Access logs with `logged_jsons_of_type(message_type)`. +- Use `build_*` helper methods from `spec/support/builds_*.rb` to construct test objects. These helpers provide sensible defaults while allowing selective overrides for testing specific scenarios. ## Important Patterns diff --git a/Steepfile b/Steepfile index 73d8ab14e..e7ba1cfcc 100644 --- a/Steepfile +++ b/Steepfile @@ -80,7 +80,8 @@ target :elasticgraph_gems do "tmpdir", "tempfile", "time", - "uri" + "uri", + "zlib" configure_code_diagnostics(::Steep::Diagnostic::Ruby.all_error) do |config| # Setting these to :hint for now, as some branches are unreachable by steep diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer.rbs index 880b484ed..6f1d1bc40 100644 --- a/elasticgraph-indexer/sig/elastic_graph/indexer.rbs +++ b/elasticgraph-indexer/sig/elastic_graph/indexer.rbs @@ -11,13 +11,13 @@ module ElasticGraph def initialize: ( config: Config, datastore_core: DatastoreCore, - ?datastore_router: DatastoreIndexingRouter?, + ?datastore_router: Indexer::_DatastoreRouter?, ?monotonic_clock: Support::MonotonicClock?, ?clock: singleton(::Time)? ) -> void - @datastore_router: DatastoreIndexingRouter? - def datastore_router: () -> DatastoreIndexingRouter + @datastore_router: Indexer::_DatastoreRouter? + def datastore_router: () -> Indexer::_DatastoreRouter @record_preparer_factory: RecordPreparer::Factory? def record_preparer_factory: () -> RecordPreparer::Factory diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer/datastore_indexing_router.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer/datastore_indexing_router.rbs index 96fdb7f01..93b3125a7 100644 --- a/elasticgraph-indexer/sig/elastic_graph/indexer/datastore_indexing_router.rbs +++ b/elasticgraph-indexer/sig/elastic_graph/indexer/datastore_indexing_router.rbs @@ -1,14 +1,20 @@ module ElasticGraph class Indexer + # Interface for routing indexing operations to a datastore (or alternative destination like S3). + # Implemented by DatastoreIndexingRouter and WarehouseDumper. + interface _DatastoreRouter + def bulk: (::Array[_Operation], ?refresh: bool) -> DatastoreIndexingRouter::BulkResult + def source_event_versions_in_index: (::Array[_Operation]) -> ::Hash[_Operation, ::Hash[::String, ::Array[::Integer]]] + end + class DatastoreIndexingRouter + include _DatastoreRouter + def initialize: ( datastore_clients_by_name: ::Hash[::String, DatastoreCore::_Client], logger: ::Logger ) -> void - def bulk: (::Array[_Operation], ?refresh: bool) -> BulkResult - def source_event_versions_in_index: (::Array[_Operation]) -> ::Hash[_Operation, ::Hash[::String, ::Array[::Integer]]] - private @datastore_clients_by_name: ::Hash[::String, DatastoreCore::_Client] diff --git a/elasticgraph-indexer/sig/elastic_graph/indexer/processor.rbs b/elasticgraph-indexer/sig/elastic_graph/indexer/processor.rbs index feedbe135..ac450f43a 100644 --- a/elasticgraph-indexer/sig/elastic_graph/indexer/processor.rbs +++ b/elasticgraph-indexer/sig/elastic_graph/indexer/processor.rbs @@ -2,7 +2,7 @@ module ElasticGraph class Indexer class Processor def initialize: ( - datastore_router: DatastoreIndexingRouter, + datastore_router: _DatastoreRouter, operation_factory: Operation::Factory, logger: ::Logger, indexing_latency_slo_thresholds_by_timestamp_in_ms: ::Hash[::String, ::Integer], @@ -14,7 +14,7 @@ module ElasticGraph private - @datastore_router: DatastoreIndexingRouter + @datastore_router: _DatastoreRouter @operation_factory: Operation::Factory @logger: ::Logger @indexing_latency_slo_thresholds_by_timestamp_in_ms: ::Hash[::String, ::Integer] 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 d364216a4..0f78f4581 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 @@ -11,12 +11,12 @@ require "elastic_graph/elasticsearch/client" require "elastic_graph/indexer/datastore_indexing_router" require "elastic_graph/indexer/operation/factory" -require "support/primary_indexing_operation_support" +require "elastic_graph/spec_support/builds_indexer_operation" module ElasticGraph class Indexer RSpec.describe DatastoreIndexingRouter, :capture_logs do - include PrimaryIndexingOperationSupport + include SpecSupport::BuildsIndexerOperation let(:main_datastore_client) { instance_spy(Elasticsearch::Client, cluster_name: "main") } let(:other_datastore_client) { instance_spy(Elasticsearch::Client, cluster_name: "other") } 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 d2dae16b2..1fed90d64 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 @@ -9,7 +9,7 @@ require "elastic_graph/indexer" require "elastic_graph/constants" require "elastic_graph/indexer/operation/factory" -require "support/primary_indexing_operation_support" +require "elastic_graph/spec_support/builds_indexer_operation" require "json" module ElasticGraph @@ -17,7 +17,7 @@ class Indexer module Operation RSpec.describe Factory, :capture_logs do describe "#build", :factories do - include PrimaryIndexingOperationSupport + include SpecSupport::BuildsIndexerOperation let(:indexer) { build_indexer } let(:component_index_definition) { index_def_named("components") } 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 3be9e6ace..c469cd8b4 100644 --- a/elasticgraph-indexer/spec/unit/elastic_graph/indexer/processor_spec.rb +++ b/elasticgraph-indexer/spec/unit/elastic_graph/indexer/processor_spec.rb @@ -11,14 +11,14 @@ require "elastic_graph/indexer/processor" require "elastic_graph/indexer/datastore_indexing_router" require "elastic_graph/support/hash_util" -require "support/primary_indexing_operation_support" +require "elastic_graph/spec_support/builds_indexer_operation" require "json" module ElasticGraph class Indexer RSpec.describe Processor do describe ".process", :factories, :capture_logs do - include PrimaryIndexingOperationSupport + include SpecSupport::BuildsIndexerOperation let(:clock) { class_double(Time, now: Time.iso8601("2020-09-15T12:30:00Z")) } let(:datastore_router) { instance_spy(ElasticGraph::Indexer::DatastoreIndexingRouter) } diff --git a/elasticgraph-warehouse_lambda/README.md b/elasticgraph-warehouse_lambda/README.md index c57b5ca90..cf277a16b 100644 --- a/elasticgraph-warehouse_lambda/README.md +++ b/elasticgraph-warehouse_lambda/README.md @@ -4,7 +4,7 @@ Write ElasticGraph-shaped JSONL files to S3, packaged for AWS Lambda. This gem adapts ElasticGraph's indexing pipeline so that, instead of writing to the datastore, it writes batched, gzipped [JSON Lines](https://jsonlines.org/) (JSONL) files to Amazon S3. Each line in the file -conforms to your ElasticGraph schema's latest JSON Schema for the corresponding object type. +conforms to a specific JSON Schema version for the corresponding object type, with files partitioned by schema version. **Note:** This code does not deduplicate when writing to S3, so the data will contain all events and versions published, plus any Lambda retries. Consumers of the S3 bucket are responsible for @@ -37,9 +37,9 @@ graph LR; ## What it does -- Consumes ElasticGraph indexing operations and groups them by GraphQL type +- Consumes ElasticGraph indexing operations and groups them by GraphQL type and JSON schema version - Transforms each operation into a flattened JSON document that matches your ElasticGraph schema -- Writes one gzipped JSONL file per type per batch to S3 with deterministic keys: +- Writes one gzipped JSONL file per type per JSON schema version per batch to S3 with deterministic keys: - `s3://///v//.jsonl.gz` - Emits structured logs for observability (counts, sizes, S3 key, etc.) diff --git a/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda.rb b/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda.rb new file mode 100644 index 000000000..d01ad1eda --- /dev/null +++ b/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda.rb @@ -0,0 +1,113 @@ +# 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/lambda_support" +require "elastic_graph/support/from_yaml_file" +require "elastic_graph/warehouse_lambda/config" + +module ElasticGraph + # Wraps an {Indexer} to dump data to S3 instead of indexing to a datastore. + # This is a stateful wrapper class (unlike {IndexerLambda} and {GraphQLLambda}, + # which are namespace modules), as it manages the relationship between the + # indexer, S3 client, and warehouse dumper. + # + # @private + class WarehouseLambda + extend Support::FromYamlFile + + # @return [Config] warehouse configuration + # @return [Indexer::Config] indexer configuration + # @return [DatastoreCore] datastore core for accessing schema artifacts + # @return [Logger] logger instance from datastore core + # @return [Module] clock module for time generation + # @dynamic config, indexer_config, datastore_core, logger, clock, indexer + attr_reader :config, :indexer_config, :datastore_core, :logger, :clock + + # Builds an `ElasticGraph::WarehouseLambda` instance from parsed YAML configuration. + # + # @param parsed_yaml [Hash] parsed YAML configuration + # @yield [Datastore::Client] optional block to customize the datastore client + # @return [WarehouseLambda] configured warehouse lambda instance + def self.from_parsed_yaml(parsed_yaml, &datastore_client_customization_block) + new( + config: Config.from_parsed_yaml!(parsed_yaml), + indexer_config: Indexer::Config.from_parsed_yaml(parsed_yaml) || Indexer::Config.new, + datastore_core: DatastoreCore.from_parsed_yaml(parsed_yaml, &datastore_client_customization_block) + ) + end + + # Initializes a WarehouseLambda instance. + # + # @param config [Config] warehouse configuration + # @param indexer_config [Indexer::Config] indexer configuration + # @param datastore_core [DatastoreCore] datastore core for accessing schema artifacts + # @param clock [Module] clock module for time generation (defaults to {::Time}) + # @param s3_client [Aws::S3::Client, nil] optional S3 client (for testing) + def initialize(config:, indexer_config:, datastore_core:, clock: ::Time, s3_client: nil) + @config = config + @indexer_config = indexer_config + @datastore_core = datastore_core + @logger = datastore_core.logger + @clock = clock + @s3_client = s3_client + end + + # Returns the processor from the indexer for event processing. + # + # @return [Processor] the processor that handles incoming events + def processor + indexer.processor + end + + # Returns the indexer instance, lazily building it on first access. + # + # @return [Indexer] the indexer that processes events + def indexer + @indexer ||= begin + require "elastic_graph/indexer" + Indexer.new( + config: indexer_config, + datastore_core: datastore_core, + datastore_router: warehouse_dumper, + clock: clock + ) + end + end + + # Returns the warehouse dumper instance, lazily building it on first access. + # + # @return [WarehouseDumper] the dumper that writes data to S3 + def warehouse_dumper + @warehouse_dumper ||= begin + require "elastic_graph/warehouse_lambda/warehouse_dumper" + WarehouseDumper.new( + logger: logger, + s3_client: s3_client, + s3_bucket_name: config.s3_bucket_name, + s3_file_prefix: config.s3_path_prefix, + clock: clock + ) + end + end + + # Returns the S3 client instance, lazily building it on first access. + # + # @return [Aws::S3::Client] the S3 client for uploading data + def s3_client + @s3_client ||= begin + require "aws-sdk-s3" + + if (region = config.aws_region) + ::Aws::S3::Client.new(region: region) + else + ::Aws::S3::Client.new + end + end + end + end +end diff --git a/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/warehouse_dumper.rb b/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/warehouse_dumper.rb new file mode 100644 index 000000000..4fe462741 --- /dev/null +++ b/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/warehouse_dumper.rb @@ -0,0 +1,143 @@ +# 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/datastore_indexing_router" +require "elastic_graph/indexer/operation/result" +require "json" +require "securerandom" +require "time" +require "zlib" + +module ElasticGraph + class WarehouseLambda + # Responsible for dumping data into a data warehouse. Implements the same interface as `DatastoreIndexingRouter` from + # `elasticgraph-indexer` so that it can be used in place of the standard datastore indexing router. + class WarehouseDumper + # @return [String] message type for logging when a batch is received + LOG_MSG_RECEIVED_BATCH = "WarehouseLambdaReceivedBatch" + + # @return [String] message type for logging when a file is dumped to S3 + LOG_MSG_DUMPED_FILE = "DumpedToWarehouseFile" + + def initialize(logger:, s3_client:, s3_bucket_name:, s3_file_prefix:, clock:) + @logger = logger + @s3_client = s3_client + @s3_bucket_name = s3_bucket_name + @s3_file_prefix = s3_file_prefix + @clock = clock + end + + # Processes a batch of indexing operations by dumping them to S3 as gzipped JSONL files. + # Operations are grouped by GraphQL type and JSON schema version, with each group written to a separate file. + # + # @param operations [Array] the indexing operations to process + # @param refresh [Boolean] ignored (included for interface compatibility with DatastoreIndexingRouter) + # @return [BulkResult] result containing success status for all operations + def bulk(operations, refresh: false) + operations_by_type_and_json_schema_version = operations.group_by { |op| [op.event.fetch("type"), op.event.fetch(JSON_SCHEMA_VERSION_KEY)] } + + @logger.info({ + "message_type" => LOG_MSG_RECEIVED_BATCH, + "record_counts_by_type" => operations_by_type_and_json_schema_version.transform_keys { |(type, _json_schema_version)| type }.transform_values(&:size) + }) + + operations_by_type_and_json_schema_version.each do |(type, json_schema_version), operations| + # Operations coming from the indexer are always Update operations for warehouse dumping + update_operations = operations # : ::Array[::ElasticGraph::Indexer::Operation::Update] + jsonl_data = build_jsonl_file_from(update_operations) + + # Skip S3 upload if all operations were filtered out (no data to write) + next if jsonl_data.empty? + + gzip_data = compress(jsonl_data) + s3_key = generate_s3_key_for(type, json_schema_version) + + # Use if_none_match: "*" to prevent overwrites (defense-in-depth, though UUIDs make collisions impossible) + @s3_client.put_object( + bucket: @s3_bucket_name, + key: s3_key, + body: gzip_data, + checksum_algorithm: :sha256, + if_none_match: "*" + ) + + @logger.info({ + "message_type" => LOG_MSG_DUMPED_FILE, + "s3_bucket" => @s3_bucket_name, + "s3_key" => s3_key, + "type" => type, + JSON_SCHEMA_VERSION_KEY => json_schema_version, + "record_count" => operations.size, + "json_size" => jsonl_data.bytesize, + "gzip_size" => gzip_data.bytesize + }) + end + + ops_and_results = operations.map do |op| + [op, ::ElasticGraph::Indexer::Operation::Result.success_of(op)] + end # : ::Array[[::ElasticGraph::Indexer::_Operation, ::ElasticGraph::Indexer::Operation::Result]] + + ::ElasticGraph::Indexer::DatastoreIndexingRouter::BulkResult.new({"warehouse" => ops_and_results}) + end + + # Returns existing event versions for the given operations. + # Always returns an empty hash since the warehouse doesn't maintain version state. + # + # @param operations [Array] the operations to check (unused) + # @return [Hash] empty hash (warehouse doesn't track versions) + def source_event_versions_in_index(operations) + {} + end + + private + + def generate_s3_key_for(type, json_schema_version) + date = @clock.now.utc.strftime("%Y-%m-%d") + uuid = ::SecureRandom.uuid + + [ + @s3_file_prefix, + type, + "v#{json_schema_version}", + date, + "#{uuid}.jsonl.gz" + ].join("/") + end + + def build_jsonl_file_from(operations) + operation_payloads = operations.filter_map do |op| + # Only include operations where the update target matches the event type (excludes derived indices) + next nil if op.update_target.type != op.event.fetch("type") + + params = op.to_datastore_bulk[1].fetch(:script).fetch(:params) + data = params.fetch("data").merge({ + "id" => params.fetch("id"), + "__eg_version" => params.fetch("version") + }) + + ::JSON.generate(data) + end + + operation_payloads.join("\n") + end + + def compress(jsonl_data) + io = ::StringIO.new + gz = ::Zlib::GzipWriter.new(io, ::Zlib::DEFAULT_COMPRESSION, ::Zlib::DEFAULT_STRATEGY) + + begin + gz << jsonl_data + ensure + gz.close + end + io.string + end + end + end +end diff --git a/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda.rbs b/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda.rbs new file mode 100644 index 000000000..279721c85 --- /dev/null +++ b/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda.rbs @@ -0,0 +1,39 @@ +module ElasticGraph + class WarehouseLambda + extend Support::FromYamlFile[WarehouseLambda] + extend _BuildableFromParsedYaml[WarehouseLambda] + + attr_reader config: Config + attr_reader indexer_config: Indexer::Config + attr_reader datastore_core: DatastoreCore + attr_reader logger: ::Logger + attr_reader clock: singleton(::Time) + + def initialize: ( + config: Config, + indexer_config: Indexer::Config, + datastore_core: DatastoreCore, + ?clock: singleton(::Time), + ?s3_client: Aws::S3::Client? + ) -> void + + def processor: () -> Indexer::Processor + + def indexer: () -> Indexer + + def warehouse_dumper: () -> WarehouseDumper + + def s3_client: () -> Aws::S3::Client + + private + + @config: Config + @indexer_config: Indexer::Config + @datastore_core: DatastoreCore + @logger: ::Logger + @clock: singleton(::Time) + @indexer: Indexer? + @warehouse_dumper: WarehouseDumper? + @s3_client: Aws::S3::Client? + end +end diff --git a/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda/warehouse_dumper.rbs b/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda/warehouse_dumper.rbs new file mode 100644 index 000000000..63a71a5de --- /dev/null +++ b/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda/warehouse_dumper.rbs @@ -0,0 +1,34 @@ +module ElasticGraph + class WarehouseLambda + class WarehouseDumper + include Indexer::_DatastoreRouter + + LOG_MSG_RECEIVED_BATCH: ::String + LOG_MSG_DUMPED_FILE: ::String + + def initialize: ( + logger: ::Logger, + s3_client: Aws::S3::Client, + s3_bucket_name: ::String, + s3_file_prefix: ::String, + clock: singleton(::Time) + ) -> void + + private + + @logger: ::Logger + @s3_client: Aws::S3::Client + @s3_bucket_name: ::String + @s3_file_prefix: ::String + @clock: singleton(::Time) + + def generate_s3_key_for: (::String, ::Integer) -> ::String + + def build_jsonl_file_from: ( + ::Array[Indexer::Operation::Update] + ) -> ::String + + def compress: (::String) -> ::String + end + end +end diff --git a/elasticgraph-warehouse_lambda/spec/support/builds_warehouse_lambda.rb b/elasticgraph-warehouse_lambda/spec/support/builds_warehouse_lambda.rb new file mode 100644 index 000000000..77060cdb7 --- /dev/null +++ b/elasticgraph-warehouse_lambda/spec/support/builds_warehouse_lambda.rb @@ -0,0 +1,39 @@ +# Copyright 2024 - 2026 Block, Inc. +# +# Use of this source code is governed by an MIT-style +# license that can be found in the LICENSE file or at +# https://opensource.org/licenses/MIT. +# +# frozen_string_literal: true + +require "elastic_graph/indexer/config" +require "elastic_graph/spec_support/builds_datastore_core" +require "elastic_graph/warehouse_lambda" + +module ElasticGraph + module BuildsWarehouseLambda + include BuildsDatastoreCore + + def build_warehouse_lambda( + s3_path_prefix: "Data0001", + s3_bucket_name: "warehouse-bucket", + aws_region: "us-west-2", + s3_client: nil, + clock: ::Time, + **datastore_core_options, + &customize_datastore_config + ) + WarehouseLambda.new( + config: WarehouseLambda::Config.new( + s3_path_prefix: s3_path_prefix, + s3_bucket_name: s3_bucket_name, + aws_region: aws_region + ), + indexer_config: Indexer::Config.from_parsed_yaml(CommonSpecHelpers.parsed_test_settings_yaml), + datastore_core: build_datastore_core(**datastore_core_options, &customize_datastore_config), + clock: clock, + s3_client: s3_client + ) + end + end +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 new file mode 100644 index 000000000..47109d580 --- /dev/null +++ b/elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda/warehouse_dumper_spec.rb @@ -0,0 +1,201 @@ +# 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 "aws-sdk-s3" +require "elastic_graph/indexer/operation/update" +require "elastic_graph/warehouse_lambda/warehouse_dumper" +require "support/builds_warehouse_lambda" +require "elastic_graph/spec_support/builds_indexer_operation" + +module ElasticGraph + class WarehouseLambda + RSpec.describe WarehouseDumper, :capture_logs do + include BuildsWarehouseLambda + include SpecSupport::BuildsIndexerOperation + + let(:s3_client) { ::Aws::S3::Client.new(stub_responses: true) } + let(:s3_bucket_name) { "warehouse-bucket" } + let(:clock) { class_double(::Time, now: ::Time.utc(2024, 9, 15, 12, 30, 12.123454)) } + let(:warehouse_lambda) { build_warehouse_lambda(s3_client: s3_client, clock: clock, s3_bucket_name: s3_bucket_name) } + let(:warehouse_dumper) { warehouse_lambda.warehouse_dumper } + let(:indexer) { warehouse_lambda.indexer } + + let(:widget_primary_indexing_op) do + new_primary_indexing_operation({ + "type" => "Widget", + "id" => "1", + "version" => 3, + "json_schema_version" => 1, + "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"} + }) + end + + it "writes operations to S3 as gzipped JSONL files and returns success results" do + op1 = new_primary_indexing_operation({"type" => "Widget", "id" => "1", "version" => 3, "json_schema_version" => 1, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}}) + op2 = new_primary_indexing_operation({"type" => "Widget", "id" => "2", "version" => 5, "json_schema_version" => 2, "record" => {"id" => "2", "dayOfWeek" => "TUE", "created_at" => "2024-09-15T13:30:12Z", "workspace_id" => "ws-2"}}) + operations = [op1, op2] + + results = warehouse_dumper.bulk(operations) + + # Verify S3 uploads - should have 2 files (one for json_schema_version 1, one for json_schema_version 2) + expect(s3_client.api_requests.map { |req| req[:operation_name] }).to eq [:put_object, :put_object] + + # Verify first file (json_schema_version 1) + params1 = s3_client.api_requests[0].fetch(:params) + expect(params1[:bucket]).to eq s3_bucket_name + expect(params1[:key]).to match %r{Data0001/Widget/v1/2024-09-15/[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}\.jsonl\.gz} + expect(params1[:checksum_algorithm]).to eq "sha256" + expect(params1[:if_none_match]).to eq "*" + + # Verify compression (gzip actually reduces size) + compressed_body1 = params1[:body] + jsonl_content1 = ::Zlib::GzipReader.new(StringIO.new(compressed_body1)).read + expect(compressed_body1.bytesize).to be < jsonl_content1.bytesize + + # Verify first file has one record + lines1 = jsonl_content1.split("\n") + expect(lines1.size).to eq 1 + + # Verify second file (json_schema_version 2) + params2 = s3_client.api_requests[1].fetch(:params) + expect(params2[:key]).to match %r{Data0001/Widget/v2/2024-09-15/[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}\.jsonl\.gz} + + compressed_body2 = params2[:body] + jsonl_content2 = ::Zlib::GzipReader.new(StringIO.new(compressed_body2)).read + + lines2 = jsonl_content2.split("\n") + expect(lines2.size).to eq 1 + + # Verify both records (combine from both files) + lines = lines1 + lines2 + + record1 = ::JSON.parse(lines[0]) + record2 = ::JSON.parse(lines[1]) + + expect(record1).to include("id" => "1", "__eg_version" => 3) + expect(record1["created_at"]).to eq "2024-09-15T12:30:12.000Z" + # Verify that name_in_index is used (workspace_id2) not the GraphQL field name (workspace_id) + expect(record1.keys).to include("workspace_id2") + expect(record1.keys).not_to include("workspace_id") + + expect(record2).to include("id" => "2", "__eg_version" => 5) + expect(record2["created_at"]).to eq "2024-09-15T13:30:12.000Z" + + # Verify success results + expect(results.ops_and_results_by_cluster.keys).to eq ["warehouse"] + ops_and_results = results.ops_and_results_by_cluster.fetch("warehouse") + expect(ops_and_results.size).to eq 2 + + ops_and_results.each do |op, result| + expect(operations).to include(op) + expect(result).to be_a Indexer::Operation::Result + expect(result.category).to eq :success + end + end + + it "writes operations of different types to separate S3 files" do + widget_op = new_primary_indexing_operation({"type" => "Widget", "id" => "1", "version" => 3, "json_schema_version" => 1, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}}) + component_op = new_primary_indexing_operation({"type" => "Component", "id" => "c1", "version" => 2, "json_schema_version" => 1, "record" => {"id" => "c1", "created_at" => "2024-09-15T12:30:12Z"}}) + operations = [widget_op, component_op] + + warehouse_dumper.bulk(operations) + + expect(s3_client.api_requests.size).to eq 2 + keys = s3_client.api_requests.map { |req| req[:params][:key] } + + expect(keys[0]).to match %r{Data0001/Widget/v1/2024-09-15/} + expect(keys[1]).to match %r{Data0001/Component/v1/2024-09-15/} + end + + it "logs structured information about received batch and dumped files" do + widget_op = new_primary_indexing_operation({"type" => "Widget", "id" => "1", "version" => 3, "json_schema_version" => 1, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}}) + component_op = new_primary_indexing_operation({"type" => "Component", "id" => "c1", "version" => 2, "json_schema_version" => 1, "record" => {"id" => "c1", "created_at" => "2024-09-15T12:30:12Z"}}) + operations = [widget_op, component_op] + + warehouse_dumper.bulk(operations) + + expect(logged_jsons_of_type(WarehouseDumper::LOG_MSG_RECEIVED_BATCH)).to match [a_hash_including({ + "record_counts_by_type" => {"Widget" => 1, "Component" => 1} + })] + + expect(logged_jsons_of_type(WarehouseDumper::LOG_MSG_DUMPED_FILE)).to match [ + a_hash_including({ + "s3_bucket" => s3_bucket_name, + "type" => "Widget", + "json_schema_version" => 1, + "record_count" => 1 + }), + a_hash_including({ + "s3_bucket" => s3_bucket_name, + "type" => "Component", + "json_schema_version" => 1, + "record_count" => 1 + }) + ] + end + + it "generates unique S3 keys using UUIDs" do + operations1 = [widget_primary_indexing_op] + operations2 = [widget_primary_indexing_op] + + warehouse_dumper.bulk(operations1) + warehouse_dumper.bulk(operations2) + + keys = s3_client.api_requests.map { |req| req[:params][:key] } + expect(keys.size).to eq 2 + expect(keys[0]).not_to eq keys[1] + + # Both should be valid UUIDs in the filename + keys.each do |key| + expect(key).to match %r{[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}\.jsonl\.gz$} + end + end + + it "returns an empty hash from source_event_versions_in_index" do + operations = [widget_primary_indexing_op] + result = warehouse_dumper.source_event_versions_in_index(operations) + expect(result).to eq({}) + end + + it "propagates S3 errors when upload fails" do + s3_client.stub_responses(:put_object, "ServiceUnavailable") + operations = [widget_primary_indexing_op] + + expect { + warehouse_dumper.bulk(operations) + }.to raise_error(Aws::S3::Errors::ServiceUnavailable) + end + + it "handles empty operations list without creating S3 files" do + warehouse_dumper.bulk([]) + + expect(s3_client.api_requests).to be_empty + end + + it "skips S3 upload when all operations are filtered out (derived index operations)" do + # Create an operation where update_target.type != event type (simulates derived index) + widget_op = new_primary_indexing_operation({ + "type" => "Widget", + "id" => "1", + "version" => 3, + "json_schema_version" => 1, + "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"} + }) + + # Stub the update_target to return a different type + derived_update_target = instance_double("ElasticGraph::SchemaArtifacts::RuntimeMetadata::UpdateTarget", type: "WidgetDerived") + allow(widget_op).to receive(:update_target).and_return(derived_update_target) + + warehouse_dumper.bulk([widget_op]) + + # Should not create any S3 files when all operations are filtered + expect(s3_client.api_requests).to be_empty + end + end + end +end diff --git a/elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda_spec.rb b/elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda_spec.rb new file mode 100644 index 000000000..6e1ab898c --- /dev/null +++ b/elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda_spec.rb @@ -0,0 +1,81 @@ +# 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 "aws-sdk-s3" +require "support/builds_warehouse_lambda" + +module ElasticGraph + RSpec.describe WarehouseLambda do + include BuildsWarehouseLambda + + # Without these ENV vars, instantiating the S3 client causes it to try to fetch instance profile creds, which significantly + # slows these tests down and produces warnings: + # > Error retrieving instance profile credentials: Failed to open TCP connection to 169.254.169.254:80 (execution expired) + around do |ex| + with_env("AWS_ACCESS_KEY_ID" => "AWS_AKI", "AWS_SECRET_ACCESS_KEY" => "AWS_SAK", &ex) + end + + it "returns non-nil values from each attribute" do + expect_to_return_non_nil_values_from_all_attributes(build_warehouse_lambda) + end + + describe ".from_parsed_yaml" do + it "builds a WarehouseLambda instance from parsed YAML" do + parsed_yaml = CommonSpecHelpers.parsed_test_settings_yaml.merge("warehouse" => { + "s3_path_prefix" => "Data001", + "s3_bucket_name" => "test-bucket", + "aws_region" => "us-west-2" + }) + + warehouse_lambda = WarehouseLambda.from_parsed_yaml(parsed_yaml) + + expect(warehouse_lambda).to be_a WarehouseLambda + expect(warehouse_lambda.indexer).to be_a Indexer + expect(warehouse_lambda.warehouse_dumper).to be_a WarehouseLambda::WarehouseDumper + end + + it "raises an error when warehouse config is missing" do + expect { + WarehouseLambda.from_parsed_yaml(CommonSpecHelpers.parsed_test_settings_yaml) + }.to raise_error Errors::ConfigError, a_string_including("warehouse") + end + end + + describe "#indexer" do + it "uses the `warehouse_dumper` as its `datastore_router`" do + warehouse_lambda = build_warehouse_lambda + + expect(warehouse_lambda.indexer.datastore_router).to be warehouse_lambda.warehouse_dumper + end + end + + describe "#s3_client" do + it "uses the provided `aws_region`" do + warehouse_lambda = build_warehouse_lambda(aws_region: "ap-east-1") + + expect(warehouse_lambda.s3_client.config.region).to eq "ap-east-1" + end + + it "falls back to AWS_REGION env var when `aws_region` is not configured" do + with_env "AWS_REGION" => "ap-south-1" do + warehouse_lambda = build_warehouse_lambda(aws_region: nil) + + expect(warehouse_lambda.s3_client.config.region).to eq "ap-south-1" + end + end + + it "raises an error if `aws_region` is not configured and `AWS_REGION` env var is not set" do + warehouse_lambda = build_warehouse_lambda(aws_region: nil) + + expect { + warehouse_lambda.s3_client + }.to raise_error ::Aws::Errors::MissingRegionError + end + end + end +end diff --git a/elasticgraph-indexer/spec/support/primary_indexing_operation_support.rb b/spec_support/lib/elastic_graph/spec_support/builds_indexer_operation.rb similarity index 86% rename from elasticgraph-indexer/spec/support/primary_indexing_operation_support.rb rename to spec_support/lib/elastic_graph/spec_support/builds_indexer_operation.rb index cea294cbf..917ab69e0 100644 --- a/elasticgraph-indexer/spec/support/primary_indexing_operation_support.rb +++ b/spec_support/lib/elastic_graph/spec_support/builds_indexer_operation.rb @@ -7,16 +7,16 @@ # frozen_string_literal: true module ElasticGraph - class Indexer + module SpecSupport # Provides test support for building primary indexing operations. - module PrimaryIndexingOperationSupport - # Builds a primary indexing operation (Operation::Update) for the given event. + module BuildsIndexerOperation + # Builds a primary indexing operation (Indexer::Operation::Update) for the given event. # # @param event [Hash] The event hash containing "type", "id", and "record" # @param index_def [DatastoreCore::IndexDefinition, nil] The index definition to use. # If not provided, it will be looked up automatically from the indexer. # @param idxr [Indexer, nil] The indexer instance to use. Defaults to `indexer` method. - # @return [Operation::Update] The primary indexing operation + # @return [Indexer::Operation::Update] The primary indexing operation def new_primary_indexing_operation(event, index_def: nil, idxr: indexer) update_targets = idxr .schema_artifacts @@ -30,7 +30,7 @@ def new_primary_indexing_operation(event, index_def: nil, idxr: indexer) index_def ||= idxr.datastore_core.index_definitions_by_graphql_type.fetch(event.fetch("type")).first - Operation::Update.new( + Indexer::Operation::Update.new( event: event, prepared_record: idxr.record_preparer_factory.for_latest_json_schema_version.prepare_for_index( event.fetch("type"), diff --git a/spec_support/lib/elastic_graph/spec_support/parallel_spec_runner.rb b/spec_support/lib/elastic_graph/spec_support/parallel_spec_runner.rb index 26639d4e8..7ff4c4f4f 100644 --- a/spec_support/lib/elastic_graph/spec_support/parallel_spec_runner.rb +++ b/spec_support/lib/elastic_graph/spec_support/parallel_spec_runner.rb @@ -63,6 +63,7 @@ module Overrides "ElasticGraph::GraphQL::DatastoreQuery::Paginator" => "elastic_graph/graphql/datastore_query", "ElasticGraph::GraphQL::DatastoreSearchRouter" => "elastic_graph/graphql/datastore_search_router", "ElasticGraph::SchemaArtifacts::FromDisk" => "elastic_graph/schema_artifacts/from_disk", + "ElasticGraph::SchemaArtifacts::RuntimeMetadata::UpdateTarget" => "elastic_graph/schema_artifacts/runtime_metadata/update_target", "ElasticGraph::Support::MonotonicClock" => "elastic_graph/support/monotonic_clock", "GraphQL::Execution::Lookahead" => "graphql/execution/lookahead" }