From 2bdd8f3de86a7f92414325ea4d0476ed30162184 Mon Sep 17 00:00:00 2001 From: Joshua Wilson Date: Sat, 17 Jan 2026 14:31:08 -0600 Subject: [PATCH 1/5] Add WarehouseLambda core with S3 WarehouseDumper MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This commit implements the core S3 data export functionality for the elasticgraph-warehouse_lambda gem, including the main WarehouseLambda class and WarehouseDumper implementation. Components added: - WarehouseLambda: Main class that orchestrates indexer, S3 client, and warehouse dumper with from_parsed_yaml factory method - WarehouseDumper: Implements DatastoreIndexingRouter interface to dump indexing operations to S3 as gzipped JSONL files - Configures indexer with WarehouseDumper as the datastore_router to redirect all indexing operations to S3 instead of the datastore Key features: - Groups operations by GraphQL type and writes separate JSONL files per type - Generates deterministic S3 keys with versioning, date partitioning, and UUIDs: s3://bucket/dumped-data///v//.jsonl.gz - Structured logging for observability (batch counts, file sizes, S3 keys) - Extracts data from operation payloads including id, __eg_version, and record fields - Filters operations to only include primary index updates - Skips operations with no datastore payload (e.g., delete operations) - Returns BulkResult with success status for all operations - Leverages aws-sdk-s3 with SHA256 checksum validation and conditional writes - S3 client initialization with optional region override from config Testing: - 37 comprehensive test examples covering all components - Tests verify S3 operations, JSONL formatting, compression, and logging - Uses stubbed AWS S3 client for fast, deterministic tests - Full RBS type signatures for type safety - 100% test coverage (from_env marked :nocov: until LambdaFunction added in next commit) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude --- Steepfile | 3 +- .../sig/elastic_graph/indexer.rbs | 6 +- .../indexer/datastore_indexing_router.rbs | 12 +- .../sig/elastic_graph/indexer/processor.rbs | 4 +- .../lib/elastic_graph/warehouse_lambda.rb | 105 +++++++ .../warehouse_lambda/warehouse_dumper.rb | 146 ++++++++++ .../sig/elastic_graph/warehouse_lambda.rbs | 30 ++ .../warehouse_lambda/warehouse_dumper.rbs | 36 +++ .../warehouse_lambda/warehouse_dumper_spec.rb | 274 ++++++++++++++++++ .../elastic_graph/warehouse_lambda_spec.rb | 102 +++++++ .../spec_support/parallel_spec_runner.rb | 1 + 11 files changed, 710 insertions(+), 9 deletions(-) create mode 100644 elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda.rb create mode 100644 elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/warehouse_dumper.rb create mode 100644 elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda.rbs create mode 100644 elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda/warehouse_dumper.rbs create mode 100644 elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda/warehouse_dumper_spec.rb create mode 100644 elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda_spec.rb 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-warehouse_lambda/lib/elastic_graph/warehouse_lambda.rb b/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda.rb new file mode 100644 index 000000000..52cdd32ec --- /dev/null +++ b/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda.rb @@ -0,0 +1,105 @@ +# 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" +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 ::ElasticGraph::Support::FromYamlFile + + # @dynamic logger, indexer, clock, warehouse_dumper + + # @return [Logger] logger for recording diagnostic information + # @return [Indexer] indexer instance that processes events + # @return [WarehouseDumper] dumper that writes data to S3 + # @return [Module] clock module used for time generation (typically {::Time}) + attr_reader :logger, :indexer, :warehouse_dumper, :clock + + # Builds an `ElasticGraph::WarehouseLambda` instance from our lambda ENV vars. + def self.from_env + # :nocov: covered by lambda_function_spec in commit 3 + ::ElasticGraph::LambdaSupport.build_from_env(self) + # :nocov: + end + + # 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) + config = Config.from_parsed_yaml!(parsed_yaml) + + # Build datastore_core first so we can access schema_artifacts + datastore_core = ::ElasticGraph::DatastoreCore.from_parsed_yaml(parsed_yaml, &datastore_client_customization_block) + + # Create warehouse dumper with schema_artifacts from datastore_core + warehouse_dumper = build_warehouse_dumper( + config: config, + logger: datastore_core.logger, + latest_json_schema_version: datastore_core.schema_artifacts.latest_json_schema_version + ) + + # Create indexer with warehouse_dumper as the datastore_router + indexer = ::ElasticGraph::Indexer.new( + config: ::ElasticGraph::Indexer::Config.from_parsed_yaml(parsed_yaml) || ::ElasticGraph::Indexer::Config.new, + datastore_core: datastore_core, + datastore_router: warehouse_dumper + ) + + new( + indexer: indexer, + warehouse_dumper: warehouse_dumper, + clock: ::Time + ) + end + + def initialize(indexer:, warehouse_dumper:, clock: ::Time) + @logger = indexer.logger + @indexer = indexer + @warehouse_dumper = warehouse_dumper + @clock = clock + end + + # Returns the processor from the indexer for event processing. + # + # @return [Processor] the processor that handles incoming events + def processor + indexer.processor + end + + private_class_method def self.build_warehouse_dumper(config:, logger:, latest_json_schema_version:) + require "elastic_graph/warehouse_lambda/warehouse_dumper" + require "aws-sdk-s3" + + s3_client_options = {} #: ::Hash[::Symbol, ::String] + if (region = config.aws_region) + s3_client_options[:region] = region + end + s3_client = ::Aws::S3::Client.new(**s3_client_options) + + WarehouseDumper.new( + logger: logger, + s3_client: s3_client, + s3_bucket_name: config.s3_bucket_name, + s3_file_prefix: config.s3_path_prefix, + latest_json_schema_version: latest_json_schema_version, + clock: ::Time + ) + 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..fb5120850 --- /dev/null +++ b/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/warehouse_dumper.rb @@ -0,0 +1,146 @@ +# 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/datastore_indexing_router" +require "elastic_graph/indexer/operation/result" +require "json" +require "time" +require "securerandom" +require "zlib" + +module ElasticGraph + class WarehouseLambda + # Responsible for dumping data into the 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:, latest_json_schema_version:, clock:) + @logger = logger + @s3_client = s3_client + @s3_bucket_name = s3_bucket_name + @s3_file_prefix = s3_file_prefix + @latest_json_schema_version = latest_json_schema_version + @clock = clock + end + + # Processes a batch of indexing operations by dumping them to S3 as gzipped JSONL files. + # Operations are grouped by GraphQL type, with each type 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 = operations.group_by { |op| op.event.fetch("type") } + + @logger.info({ + "message_type" => LOG_MSG_RECEIVED_BATCH, + "record_counts_by_type" => operations_by_type.transform_values(&:size) + }) + + operations_by_type.each do |type, 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) + + # 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, + "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) + date = @clock.now.utc.strftime("%Y-%m-%d") + uuid = ::SecureRandom.uuid + + [ + @s3_file_prefix, + type, + "v#{@latest_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") + # Skip operations with no datastore payload (e.g., delete operations or invalid data) + next nil if op.to_datastore_bulk.empty? + + payload_body = op.to_datastore_bulk[1] + script = payload_body.fetch(:script) + params = 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..7ab7e1986 --- /dev/null +++ b/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda.rbs @@ -0,0 +1,30 @@ +module ElasticGraph + class WarehouseLambda + extend Support::FromYamlFile[WarehouseLambda] + + attr_reader logger: ::Logger + attr_reader indexer: Indexer + attr_reader warehouse_dumper: WarehouseDumper + attr_reader clock: singleton(::Time) + + def self.from_env: () -> WarehouseLambda + + def self.from_parsed_yaml: (parsedYamlSettings) ?{ (untyped) -> void } -> WarehouseLambda + + def initialize: ( + indexer: Indexer, + warehouse_dumper: WarehouseDumper, + ?clock: singleton(::Time) + ) -> void + + def processor: () -> Indexer::Processor + + private + + def self.build_warehouse_dumper: ( + config: WarehouseLambda::Config, + logger: ::Logger, + latest_json_schema_version: ::Integer + ) -> WarehouseDumper + 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..518a06dbe --- /dev/null +++ b/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda/warehouse_dumper.rbs @@ -0,0 +1,36 @@ +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, + latest_json_schema_version: ::Integer, + clock: singleton(::Time) + ) -> void + + private + + @logger: ::Logger + @s3_client: Aws::S3::Client + @s3_bucket_name: ::String + @s3_file_prefix: ::String + @latest_json_schema_version: ::Integer + @clock: singleton(::Time) + + def generate_s3_key_for: (::String) -> ::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/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..67342a274 --- /dev/null +++ b/elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda/warehouse_dumper_spec.rb @@ -0,0 +1,274 @@ +# 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/spec_support/lambda_function" + +require "aws-sdk-s3" +require "elastic_graph/indexer" +require "elastic_graph/indexer/operation/update" +require "elastic_graph/warehouse_lambda" +require "elastic_graph/warehouse_lambda/warehouse_dumper" +require "elastic_graph/warehouse_lambda/config" + +module ElasticGraph + class WarehouseLambda + RSpec.describe WarehouseDumper do + include_context "lambda function", config_overrides_in_yaml: {"warehouse" => {"s3_path_prefix" => "Data001", "s3_bucket_name" => "test-bucket"}} + + let(:parsed_yaml) { CommonSpecHelpers.parsed_test_settings_yaml.merge("warehouse" => {"s3_path_prefix" => "Data001", "s3_bucket_name" => "warehouse-bucket"}) } + 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_dumper) do + datastore_core = ::ElasticGraph::DatastoreCore.from_parsed_yaml(parsed_yaml) + WarehouseDumper.new( + logger: datastore_core.logger, + s3_client: s3_client, + s3_bucket_name: s3_bucket_name, + s3_file_prefix: "Data001", + latest_json_schema_version: datastore_core.schema_artifacts.latest_json_schema_version, + clock: clock + ) + end + + let(:indexer) do + datastore_core = ::ElasticGraph::DatastoreCore.from_parsed_yaml(parsed_yaml) + ::ElasticGraph::Indexer.new( + config: ::ElasticGraph::Indexer::Config.new, + datastore_core: datastore_core, + datastore_router: warehouse_dumper + ) + end + + let(:widget_primary_indexing_op) do + new_primary_indexing_op({ + "type" => "Widget", + "id" => "1", + "version" => 3, + "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_op({"type" => "Widget", "id" => "1", "version" => 3, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}}) + op2 = new_primary_indexing_op({"type" => "Widget", "id" => "2", "version" => 5, "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 upload + expect(s3_client.api_requests.map { |req| req[:operation_name] }).to eq [:put_object] + + # Verify S3 upload parameters + params = s3_client.api_requests.first.fetch(:params) + expect(params[:bucket]).to eq s3_bucket_name + expect(params[:key]).to match %r{Data001/Widget/v1/2024-09-15/[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}\.jsonl\.gz} + expect(params[:checksum_algorithm]).to eq "sha256" + expect(params[:if_none_match]).to eq "*" + + # Verify compression (gzip actually reduces size) + compressed_body = params[:body] + jsonl_content = ::Zlib::GzipReader.new(StringIO.new(compressed_body)).read + expect(compressed_body.bytesize).to be < jsonl_content.bytesize + + # Verify JSONL content has both records + lines = jsonl_content.split("\n") + expect(lines.size).to eq 2 + + 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 ::ElasticGraph::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_op({"type" => "Widget", "id" => "1", "version" => 3, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}}) + component_op = new_primary_indexing_op({"type" => "Component", "id" => "c1", "version" => 2, "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{Data001/Widget/v1/2024-09-15/} + expect(keys[1]).to match %r{Data001/Component/v1/2024-09-15/} + end + + it "logs structured information about received batch and dumped files" do + widget_op = new_primary_indexing_op({"type" => "Widget", "id" => "1", "version" => 3, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}}) + component_op = new_primary_indexing_op({"type" => "Component", "id" => "c1", "version" => 2, "record" => {"id" => "c1", "created_at" => "2024-09-15T12:30:12Z"}}) + operations = [widget_op, component_op] + + logger = instance_double(::Logger) + datastore_core = ::ElasticGraph::DatastoreCore.from_parsed_yaml(parsed_yaml) + custom_warehouse_dumper = WarehouseDumper.new( + logger: logger, + s3_client: s3_client, + s3_bucket_name: s3_bucket_name, + s3_file_prefix: "Data001", + latest_json_schema_version: datastore_core.schema_artifacts.latest_json_schema_version, + clock: clock + ) + + expect(logger).to receive(:info).with(hash_including( + "message_type" => WarehouseDumper::LOG_MSG_RECEIVED_BATCH, + "record_counts_by_type" => {"Widget" => 1, "Component" => 1} + )) + + expect(logger).to receive(:info).with(hash_including( + "message_type" => WarehouseDumper::LOG_MSG_DUMPED_FILE, + "s3_bucket" => s3_bucket_name, + "type" => "Widget", + "record_count" => 1 + )).once + + expect(logger).to receive(:info).with(hash_including( + "message_type" => WarehouseDumper::LOG_MSG_DUMPED_FILE, + "s3_bucket" => s3_bucket_name, + "type" => "Component", + "record_count" => 1 + )).once + + custom_warehouse_dumper.bulk(operations) + 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 "skips operations with empty to_datastore_bulk" do + # Create two Widget operations - one valid, one with empty datastore_bulk + valid_widget = new_primary_indexing_op({ + "type" => "Widget", + "id" => "1", + "version" => 3, + "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"} + }) + + empty_widget = new_primary_indexing_op({ + "type" => "Widget", + "id" => "2", + "version" => 4, + "record" => {"id" => "2", "dayOfWeek" => "TUE", "created_at" => "2024-09-15T13:30:12Z", "workspace_id" => "ws-2"} + }) + + # Stub to_datastore_bulk to return empty on empty_widget + allow(empty_widget).to receive(:to_datastore_bulk).and_return([]) + + warehouse_dumper.bulk([valid_widget, empty_widget]) + + # Verify only one record in the JSONL + params = s3_client.api_requests.first.fetch(:params) + jsonl_content = ::Zlib::GzipReader.new(StringIO.new(params[:body])).read + lines = jsonl_content.split("\n") + + expect(lines.size).to eq 1 + record = ::JSON.parse(lines[0]) + expect(record["id"]).to eq "1" + 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_op({ + "type" => "Widget", + "id" => "1", + "version" => 3, + "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 + + def new_primary_indexing_op(event) + update_targets = indexer + .schema_artifacts + .runtime_metadata + .object_types_by_name + .fetch(event.fetch("type")) + .update_targets + .select { |ut| ut.type == event.fetch("type") } + + expect(update_targets.size).to eq(1) + index_def = indexer.datastore_core.index_definitions_by_graphql_type.fetch(event.fetch("type")).first + + ::ElasticGraph::Indexer::Operation::Update.new( + event: event, + prepared_record: indexer.record_preparer_factory.for_latest_json_schema_version.prepare_for_index( + event.fetch("type"), + event.fetch("record") + ), + destination_index_def: index_def, + update_target: update_targets.first, + doc_id: event.fetch("id"), + destination_index_mapping: indexer.schema_artifacts.index_mappings_by_index_def_name.fetch(index_def.name) + ) + 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..c713e7b1a --- /dev/null +++ b/elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda_spec.rb @@ -0,0 +1,102 @@ +# 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/spec_support/lambda_function" + +require "aws-sdk-s3" +require "elastic_graph/indexer" +require "elastic_graph/warehouse_lambda" +require "elastic_graph/warehouse_lambda/config" +require "elastic_graph/warehouse_lambda/warehouse_dumper" + +module ElasticGraph + RSpec.describe WarehouseLambda do + include_context "lambda function", config_overrides_in_yaml: {"warehouse" => {"s3_path_prefix" => "Data001", "s3_bucket_name" => "test-bucket"}} + + let(:parsed_yaml) { CommonSpecHelpers.parsed_test_settings_yaml.merge("warehouse" => {"s3_path_prefix" => "Data001", "s3_bucket_name" => "test-bucket"}) } + let(:s3_client) { ::Aws::S3::Client.new(stub_responses: true) } + let(:warehouse_dumper) do + datastore_core = ::ElasticGraph::DatastoreCore.from_parsed_yaml(parsed_yaml) + WarehouseLambda::WarehouseDumper.new( + logger: datastore_core.logger, + s3_client: s3_client, + s3_bucket_name: "test-bucket", + s3_file_prefix: "Data001", + latest_json_schema_version: datastore_core.schema_artifacts.latest_json_schema_version, + clock: ::Time + ) + end + let(:indexer) do + datastore_core = ::ElasticGraph::DatastoreCore.from_parsed_yaml(parsed_yaml) + ::ElasticGraph::Indexer.new( + config: ::ElasticGraph::Indexer::Config.new, + datastore_core: datastore_core, + datastore_router: warehouse_dumper + ) + end + + describe ".from_parsed_yaml" do + it "builds a WarehouseLambda instance from parsed YAML" do + warehouse_lambda = WarehouseLambda.from_parsed_yaml(parsed_yaml) + + expect(warehouse_lambda).to be_a WarehouseLambda + expect(warehouse_lambda.indexer).to be_a ::ElasticGraph::Indexer + expect(warehouse_lambda.warehouse_dumper).to be_a WarehouseLambda::WarehouseDumper + expect(warehouse_lambda.logger).to be_a ::Logger + end + + it "configures the indexer to use the warehouse_dumper as its datastore_router" do + warehouse_lambda = WarehouseLambda.from_parsed_yaml(parsed_yaml) + + expect(warehouse_lambda.indexer.datastore_router).to eq warehouse_lambda.warehouse_dumper + end + + it "configures the S3 client with aws_region when provided" do + yaml_with_region = parsed_yaml.merge("warehouse" => parsed_yaml["warehouse"].merge("aws_region" => "us-west-2")) + warehouse_lambda = WarehouseLambda.from_parsed_yaml(yaml_with_region) + + # We can't easily inspect the S3 client's region since it's private, but we can verify + # the warehouse_dumper was created successfully (which means the S3 client was configured) + expect(warehouse_lambda.warehouse_dumper).to be_a WarehouseLambda::WarehouseDumper + end + + it "raises an error when warehouse config is missing" do + yaml_without_warehouse = CommonSpecHelpers.parsed_test_settings_yaml + + expect { + WarehouseLambda.from_parsed_yaml(yaml_without_warehouse) + }.to raise_error ::ElasticGraph::Errors::ConfigError, a_string_including("warehouse") + end + end + + describe "#initialize" do + it "initializes with provided components and sets logger from indexer" do + warehouse_lambda = WarehouseLambda.new(indexer: indexer, warehouse_dumper: warehouse_dumper) + + expect(warehouse_lambda.logger).to eq indexer.logger + expect(warehouse_lambda.indexer).to eq indexer + expect(warehouse_lambda.warehouse_dumper).to eq warehouse_dumper + end + + it "accepts an optional clock parameter" do + custom_clock = class_double(::Time, now: ::Time.utc(2024, 1, 1)) + warehouse_lambda = WarehouseLambda.new(indexer: indexer, warehouse_dumper: warehouse_dumper, clock: custom_clock) + + expect(warehouse_lambda.clock).to eq custom_clock + end + end + + describe "#processor" do + it "returns the indexer's processor" do + warehouse_lambda = WarehouseLambda.new(indexer: indexer, warehouse_dumper: warehouse_dumper) + + expect(warehouse_lambda.processor).to eq indexer.processor + end + end + end +end 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" } From 844bb8bc2e4bb6e445360cd19291d12c35b579fe Mon Sep 17 00:00:00 2001 From: Josh Wilson Date: Sun, 18 Jan 2026 20:55:14 -0600 Subject: [PATCH 2/5] Apply suggestions from code review Co-authored-by: Myron Marston --- .../warehouse_lambda/warehouse_dumper.rb | 9 +- .../warehouse_lambda/warehouse_dumper_spec.rb | 80 ++++---------- .../elastic_graph/warehouse_lambda_spec.rb | 101 +++++++----------- 3 files changed, 66 insertions(+), 124 deletions(-) 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 index fb5120850..a036ac6c7 100644 --- a/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/warehouse_dumper.rb +++ b/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/warehouse_dumper.rb @@ -15,7 +15,7 @@ module ElasticGraph class WarehouseLambda - # Responsible for dumping data into the data warehouse. Implements the same interface as `DatastoreIndexingRouter` from + # 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 @@ -49,7 +49,7 @@ def bulk(operations, refresh: false) operations_by_type.each do |type, operations| # Operations coming from the indexer are always Update operations for warehouse dumping - update_operations = operations #: ::Array[::ElasticGraph::Indexer::Operation::Update] + 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) @@ -80,7 +80,7 @@ def bulk(operations, refresh: false) ops_and_results = operations.map do |op| [op, ::ElasticGraph::Indexer::Operation::Result.success_of(op)] - end #: ::Array[[::ElasticGraph::Indexer::_Operation, ::ElasticGraph::Indexer::Operation::Result]] + end # : ::Array[[::ElasticGraph::Indexer::_Operation, ::ElasticGraph::Indexer::Operation::Result]] ::ElasticGraph::Indexer::DatastoreIndexingRouter::BulkResult.new({"warehouse" => ops_and_results}) end @@ -117,8 +117,7 @@ def build_jsonl_file_from(operations) next nil if op.to_datastore_bulk.empty? payload_body = op.to_datastore_bulk[1] - script = payload_body.fetch(:script) - params = script.fetch(:params) + params = payload_body.fetch(:script).fetch(:params) data = params.fetch("data").merge({ "id" => params.fetch("id"), "__eg_version" => params.fetch("version") 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 67342a274..997cac0ea 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 @@ -6,45 +6,22 @@ # # frozen_string_literal: true -require "elastic_graph/spec_support/lambda_function" - require "aws-sdk-s3" -require "elastic_graph/indexer" require "elastic_graph/indexer/operation/update" -require "elastic_graph/warehouse_lambda" require "elastic_graph/warehouse_lambda/warehouse_dumper" -require "elastic_graph/warehouse_lambda/config" +require "support/builds_warehouse_lambda" module ElasticGraph class WarehouseLambda - RSpec.describe WarehouseDumper do - include_context "lambda function", config_overrides_in_yaml: {"warehouse" => {"s3_path_prefix" => "Data001", "s3_bucket_name" => "test-bucket"}} + RSpec.describe WarehouseDumper, :capture_logs do + include BuildsWarehouseLambda - let(:parsed_yaml) { CommonSpecHelpers.parsed_test_settings_yaml.merge("warehouse" => {"s3_path_prefix" => "Data001", "s3_bucket_name" => "warehouse-bucket"}) } 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_dumper) do - datastore_core = ::ElasticGraph::DatastoreCore.from_parsed_yaml(parsed_yaml) - WarehouseDumper.new( - logger: datastore_core.logger, - s3_client: s3_client, - s3_bucket_name: s3_bucket_name, - s3_file_prefix: "Data001", - latest_json_schema_version: datastore_core.schema_artifacts.latest_json_schema_version, - clock: clock - ) - end - - let(:indexer) do - datastore_core = ::ElasticGraph::DatastoreCore.from_parsed_yaml(parsed_yaml) - ::ElasticGraph::Indexer.new( - config: ::ElasticGraph::Indexer::Config.new, - datastore_core: datastore_core, - datastore_router: warehouse_dumper - ) - end + 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_op({ @@ -124,37 +101,24 @@ class WarehouseLambda component_op = new_primary_indexing_op({"type" => "Component", "id" => "c1", "version" => 2, "record" => {"id" => "c1", "created_at" => "2024-09-15T12:30:12Z"}}) operations = [widget_op, component_op] - logger = instance_double(::Logger) - datastore_core = ::ElasticGraph::DatastoreCore.from_parsed_yaml(parsed_yaml) - custom_warehouse_dumper = WarehouseDumper.new( - logger: logger, - s3_client: s3_client, - s3_bucket_name: s3_bucket_name, - s3_file_prefix: "Data001", - latest_json_schema_version: datastore_core.schema_artifacts.latest_json_schema_version, - clock: clock - ) + warehouse_dumper.bulk(operations) - expect(logger).to receive(:info).with(hash_including( - "message_type" => WarehouseDumper::LOG_MSG_RECEIVED_BATCH, + expect(logged_jsons_of_type(WarehouseDumper::LOG_MSG_RECEIVED_BATCH)).to match [a_hash_including({ "record_counts_by_type" => {"Widget" => 1, "Component" => 1} - )) - - expect(logger).to receive(:info).with(hash_including( - "message_type" => WarehouseDumper::LOG_MSG_DUMPED_FILE, - "s3_bucket" => s3_bucket_name, - "type" => "Widget", - "record_count" => 1 - )).once - - expect(logger).to receive(:info).with(hash_including( - "message_type" => WarehouseDumper::LOG_MSG_DUMPED_FILE, - "s3_bucket" => s3_bucket_name, - "type" => "Component", - "record_count" => 1 - )).once - - custom_warehouse_dumper.bulk(operations) + })] + + expect(logged_jsons_of_type(WarehouseDumper::LOG_MSG_DUMPED_FILE)).to match [ + a_hash_including({ + "s3_bucket" => s3_bucket_name, + "type" => "Widget", + "record_count" => 1 + }), + a_hash_including({ + "s3_bucket" => s3_bucket_name, + "type" => "Component", + "record_count" => 1 + }) + ] end it "generates unique S3 keys using UUIDs" do 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 index c713e7b1a..d4af0f4fe 100644 --- a/elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda_spec.rb +++ b/elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda_spec.rb @@ -6,96 +6,75 @@ # # frozen_string_literal: true -require "elastic_graph/spec_support/lambda_function" - require "aws-sdk-s3" -require "elastic_graph/indexer" -require "elastic_graph/warehouse_lambda" -require "elastic_graph/warehouse_lambda/config" -require "elastic_graph/warehouse_lambda/warehouse_dumper" +require "support/builds_warehouse_lambda" module ElasticGraph RSpec.describe WarehouseLambda do - include_context "lambda function", config_overrides_in_yaml: {"warehouse" => {"s3_path_prefix" => "Data001", "s3_bucket_name" => "test-bucket"}} - - let(:parsed_yaml) { CommonSpecHelpers.parsed_test_settings_yaml.merge("warehouse" => {"s3_path_prefix" => "Data001", "s3_bucket_name" => "test-bucket"}) } - let(:s3_client) { ::Aws::S3::Client.new(stub_responses: true) } - let(:warehouse_dumper) do - datastore_core = ::ElasticGraph::DatastoreCore.from_parsed_yaml(parsed_yaml) - WarehouseLambda::WarehouseDumper.new( - logger: datastore_core.logger, - s3_client: s3_client, - s3_bucket_name: "test-bucket", - s3_file_prefix: "Data001", - latest_json_schema_version: datastore_core.schema_artifacts.latest_json_schema_version, - clock: ::Time - ) + include BuildsWarehouseLambda + + # Without these ENV vars, instanting 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 - let(:indexer) do - datastore_core = ::ElasticGraph::DatastoreCore.from_parsed_yaml(parsed_yaml) - ::ElasticGraph::Indexer.new( - config: ::ElasticGraph::Indexer::Config.new, - datastore_core: datastore_core, - datastore_router: warehouse_dumper - ) + + 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 ::ElasticGraph::Indexer expect(warehouse_lambda.warehouse_dumper).to be_a WarehouseLambda::WarehouseDumper - expect(warehouse_lambda.logger).to be_a ::Logger - end - - it "configures the indexer to use the warehouse_dumper as its datastore_router" do - warehouse_lambda = WarehouseLambda.from_parsed_yaml(parsed_yaml) - - expect(warehouse_lambda.indexer.datastore_router).to eq warehouse_lambda.warehouse_dumper - end - - it "configures the S3 client with aws_region when provided" do - yaml_with_region = parsed_yaml.merge("warehouse" => parsed_yaml["warehouse"].merge("aws_region" => "us-west-2")) - warehouse_lambda = WarehouseLambda.from_parsed_yaml(yaml_with_region) - - # We can't easily inspect the S3 client's region since it's private, but we can verify - # the warehouse_dumper was created successfully (which means the S3 client was configured) - expect(warehouse_lambda.warehouse_dumper).to be_a WarehouseLambda::WarehouseDumper end it "raises an error when warehouse config is missing" do - yaml_without_warehouse = CommonSpecHelpers.parsed_test_settings_yaml - expect { - WarehouseLambda.from_parsed_yaml(yaml_without_warehouse) + WarehouseLambda.from_parsed_yaml(CommonSpecHelpers.parsed_test_settings_yaml) }.to raise_error ::ElasticGraph::Errors::ConfigError, a_string_including("warehouse") end end - describe "#initialize" do - it "initializes with provided components and sets logger from indexer" do - warehouse_lambda = WarehouseLambda.new(indexer: indexer, warehouse_dumper: warehouse_dumper) + describe "#indexer" do + it "uses the `warehouse_dumper` as its `datastore_router`" do + warehouse_lambda = build_warehouse_lambda - expect(warehouse_lambda.logger).to eq indexer.logger - expect(warehouse_lambda.indexer).to eq indexer - expect(warehouse_lambda.warehouse_dumper).to eq warehouse_dumper + expect(warehouse_lambda.indexer.datastore_router).to be warehouse_lambda.warehouse_dumper end + end - it "accepts an optional clock parameter" do - custom_clock = class_double(::Time, now: ::Time.utc(2024, 1, 1)) - warehouse_lambda = WarehouseLambda.new(indexer: indexer, warehouse_dumper: warehouse_dumper, clock: custom_clock) + describe "#s3_client" do + it "uses the provided `aws_region`" do + warehouse_lambda = build_warehouse_lambda(aws_region: "ap-east-1") - expect(warehouse_lambda.clock).to eq custom_clock + expect(warehouse_lambda.s3_client.config.region).to eq "ap-east-1" end - end - describe "#processor" do - it "returns the indexer's processor" do - warehouse_lambda = WarehouseLambda.new(indexer: indexer, warehouse_dumper: warehouse_dumper) + 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.processor).to eq indexer.processor + 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 From 8b1d3ea14c117fbf83b7d66708331618fe8205e7 Mon Sep 17 00:00:00 2001 From: Joshua Wilson Date: Sun, 18 Jan 2026 21:14:54 -0600 Subject: [PATCH 3/5] Address comments. --- CLAUDE.md | 7 + elasticgraph-warehouse_lambda/README.md | 6 +- .../lib/elastic_graph/warehouse_lambda.rb | 122 ++++++++++-------- .../warehouse_lambda/warehouse_dumper.rb | 26 ++-- .../sig/elastic_graph/warehouse_lambda.rbs | 35 +++-- .../warehouse_lambda/warehouse_dumper.rbs | 4 +- .../spec/support/builds_warehouse_lambda.rb | 39 ++++++ .../warehouse_lambda/warehouse_dumper_spec.rb | 94 ++++++-------- .../elastic_graph/warehouse_lambda_spec.rb | 6 +- 9 files changed, 193 insertions(+), 146 deletions(-) create mode 100644 elasticgraph-warehouse_lambda/spec/support/builds_warehouse_lambda.rb 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/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 index 52cdd32ec..d01ad1eda 100644 --- a/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda.rb +++ b/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda.rb @@ -6,7 +6,6 @@ # # frozen_string_literal: true -require "elastic_graph/indexer" require "elastic_graph/lambda_support" require "elastic_graph/support/from_yaml_file" require "elastic_graph/warehouse_lambda/config" @@ -19,22 +18,15 @@ module ElasticGraph # # @private class WarehouseLambda - extend ::ElasticGraph::Support::FromYamlFile + extend Support::FromYamlFile - # @dynamic logger, indexer, clock, warehouse_dumper - - # @return [Logger] logger for recording diagnostic information - # @return [Indexer] indexer instance that processes events - # @return [WarehouseDumper] dumper that writes data to S3 - # @return [Module] clock module used for time generation (typically {::Time}) - attr_reader :logger, :indexer, :warehouse_dumper, :clock - - # Builds an `ElasticGraph::WarehouseLambda` instance from our lambda ENV vars. - def self.from_env - # :nocov: covered by lambda_function_spec in commit 3 - ::ElasticGraph::LambdaSupport.build_from_env(self) - # :nocov: - end + # @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. # @@ -42,37 +34,27 @@ def self.from_env # @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) - config = Config.from_parsed_yaml!(parsed_yaml) - - # Build datastore_core first so we can access schema_artifacts - datastore_core = ::ElasticGraph::DatastoreCore.from_parsed_yaml(parsed_yaml, &datastore_client_customization_block) - - # Create warehouse dumper with schema_artifacts from datastore_core - warehouse_dumper = build_warehouse_dumper( - config: config, - logger: datastore_core.logger, - latest_json_schema_version: datastore_core.schema_artifacts.latest_json_schema_version - ) - - # Create indexer with warehouse_dumper as the datastore_router - indexer = ::ElasticGraph::Indexer.new( - config: ::ElasticGraph::Indexer::Config.from_parsed_yaml(parsed_yaml) || ::ElasticGraph::Indexer::Config.new, - datastore_core: datastore_core, - datastore_router: warehouse_dumper - ) - new( - indexer: indexer, - warehouse_dumper: warehouse_dumper, - clock: ::Time + 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 - def initialize(indexer:, warehouse_dumper:, clock: ::Time) - @logger = indexer.logger - @indexer = indexer - @warehouse_dumper = warehouse_dumper + # 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. @@ -82,24 +64,50 @@ def processor indexer.processor end - private_class_method def self.build_warehouse_dumper(config:, logger:, latest_json_schema_version:) - require "elastic_graph/warehouse_lambda/warehouse_dumper" - require "aws-sdk-s3" + # 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 - s3_client_options = {} #: ::Hash[::Symbol, ::String] - if (region = config.aws_region) - s3_client_options[:region] = region + # 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 - s3_client = ::Aws::S3::Client.new(**s3_client_options) + end - WarehouseDumper.new( - logger: logger, - s3_client: s3_client, - s3_bucket_name: config.s3_bucket_name, - s3_file_prefix: config.s3_path_prefix, - latest_json_schema_version: latest_json_schema_version, - clock: ::Time - ) + # 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 index a036ac6c7..4fe462741 100644 --- a/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/warehouse_dumper.rb +++ b/elasticgraph-warehouse_lambda/lib/elastic_graph/warehouse_lambda/warehouse_dumper.rb @@ -6,11 +6,12 @@ # # frozen_string_literal: true +require "elastic_graph/constants" require "elastic_graph/indexer/datastore_indexing_router" require "elastic_graph/indexer/operation/result" require "json" -require "time" require "securerandom" +require "time" require "zlib" module ElasticGraph @@ -24,30 +25,29 @@ class WarehouseDumper # @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:, latest_json_schema_version:, clock:) + 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 - @latest_json_schema_version = latest_json_schema_version @clock = clock end # Processes a batch of indexing operations by dumping them to S3 as gzipped JSONL files. - # Operations are grouped by GraphQL type, with each type written to a separate file. + # 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 = operations.group_by { |op| op.event.fetch("type") } + 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.transform_values(&:size) + "record_counts_by_type" => operations_by_type_and_json_schema_version.transform_keys { |(type, _json_schema_version)| type }.transform_values(&:size) }) - operations_by_type.each do |type, operations| + 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) @@ -56,7 +56,7 @@ def bulk(operations, refresh: false) next if jsonl_data.empty? gzip_data = compress(jsonl_data) - s3_key = generate_s3_key_for(type) + 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( @@ -72,6 +72,7 @@ def bulk(operations, refresh: false) "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 @@ -96,14 +97,14 @@ def source_event_versions_in_index(operations) private - def generate_s3_key_for(type) + 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#{@latest_json_schema_version}", + "v#{json_schema_version}", date, "#{uuid}.jsonl.gz" ].join("/") @@ -113,11 +114,8 @@ 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") - # Skip operations with no datastore payload (e.g., delete operations or invalid data) - next nil if op.to_datastore_bulk.empty? - payload_body = op.to_datastore_bulk[1] - params = payload_body.fetch(:script).fetch(:params) + params = op.to_datastore_bulk[1].fetch(:script).fetch(:params) data = params.fetch("data").merge({ "id" => params.fetch("id"), "__eg_version" => params.fetch("version") diff --git a/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda.rbs b/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda.rbs index 7ab7e1986..0afd4e863 100644 --- a/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda.rbs +++ b/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda.rbs @@ -1,30 +1,41 @@ 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 indexer: Indexer - attr_reader warehouse_dumper: WarehouseDumper attr_reader clock: singleton(::Time) - def self.from_env: () -> WarehouseLambda - def self.from_parsed_yaml: (parsedYamlSettings) ?{ (untyped) -> void } -> WarehouseLambda def initialize: ( - indexer: Indexer, - warehouse_dumper: WarehouseDumper, - ?clock: singleton(::Time) + 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 - def self.build_warehouse_dumper: ( - config: WarehouseLambda::Config, - logger: ::Logger, - latest_json_schema_version: ::Integer - ) -> WarehouseDumper + @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 index 518a06dbe..63a71a5de 100644 --- a/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda/warehouse_dumper.rbs +++ b/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda/warehouse_dumper.rbs @@ -11,7 +11,6 @@ module ElasticGraph s3_client: Aws::S3::Client, s3_bucket_name: ::String, s3_file_prefix: ::String, - latest_json_schema_version: ::Integer, clock: singleton(::Time) ) -> void @@ -21,10 +20,9 @@ module ElasticGraph @s3_client: Aws::S3::Client @s3_bucket_name: ::String @s3_file_prefix: ::String - @latest_json_schema_version: ::Integer @clock: singleton(::Time) - def generate_s3_key_for: (::String) -> ::String + def generate_s3_key_for: (::String, ::Integer) -> ::String def build_jsonl_file_from: ( ::Array[Indexer::Operation::Update] 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 index 997cac0ea..56e8ada8f 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 @@ -28,35 +28,49 @@ class WarehouseLambda "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_op({"type" => "Widget", "id" => "1", "version" => 3, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}}) - op2 = new_primary_indexing_op({"type" => "Widget", "id" => "2", "version" => 5, "record" => {"id" => "2", "dayOfWeek" => "TUE", "created_at" => "2024-09-15T13:30:12Z", "workspace_id" => "ws-2"}}) + op1 = new_primary_indexing_op({"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_op({"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 upload - expect(s3_client.api_requests.map { |req| req[:operation_name] }).to eq [:put_object] + # 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 S3 upload parameters - params = s3_client.api_requests.first.fetch(:params) - expect(params[:bucket]).to eq s3_bucket_name - expect(params[:key]).to match %r{Data001/Widget/v1/2024-09-15/[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}\.jsonl\.gz} - expect(params[:checksum_algorithm]).to eq "sha256" - expect(params[:if_none_match]).to eq "*" + # 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_body = params[:body] - jsonl_content = ::Zlib::GzipReader.new(StringIO.new(compressed_body)).read - expect(compressed_body.bytesize).to be < jsonl_content.bytesize + compressed_body1 = params1[:body] + jsonl_content1 = ::Zlib::GzipReader.new(StringIO.new(compressed_body1)).read + expect(compressed_body1.bytesize).to be < jsonl_content1.bytesize - # Verify JSONL content has both records - lines = jsonl_content.split("\n") - expect(lines.size).to eq 2 + # 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]) @@ -77,14 +91,14 @@ class WarehouseLambda ops_and_results.each do |op, result| expect(operations).to include(op) - expect(result).to be_a ::ElasticGraph::Indexer::Operation::Result + 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_op({"type" => "Widget", "id" => "1", "version" => 3, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}}) - component_op = new_primary_indexing_op({"type" => "Component", "id" => "c1", "version" => 2, "record" => {"id" => "c1", "created_at" => "2024-09-15T12:30:12Z"}}) + widget_op = new_primary_indexing_op({"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_op({"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) @@ -92,13 +106,13 @@ class WarehouseLambda 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{Data001/Widget/v1/2024-09-15/} - expect(keys[1]).to match %r{Data001/Component/v1/2024-09-15/} + 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_op({"type" => "Widget", "id" => "1", "version" => 3, "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"}}) - component_op = new_primary_indexing_op({"type" => "Component", "id" => "c1", "version" => 2, "record" => {"id" => "c1", "created_at" => "2024-09-15T12:30:12Z"}}) + widget_op = new_primary_indexing_op({"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_op({"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) @@ -111,11 +125,13 @@ class WarehouseLambda 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 }) ] @@ -153,37 +169,6 @@ class WarehouseLambda }.to raise_error(Aws::S3::Errors::ServiceUnavailable) end - it "skips operations with empty to_datastore_bulk" do - # Create two Widget operations - one valid, one with empty datastore_bulk - valid_widget = new_primary_indexing_op({ - "type" => "Widget", - "id" => "1", - "version" => 3, - "record" => {"id" => "1", "dayOfWeek" => "MON", "created_at" => "2024-09-15T12:30:12Z", "workspace_id" => "ws-1"} - }) - - empty_widget = new_primary_indexing_op({ - "type" => "Widget", - "id" => "2", - "version" => 4, - "record" => {"id" => "2", "dayOfWeek" => "TUE", "created_at" => "2024-09-15T13:30:12Z", "workspace_id" => "ws-2"} - }) - - # Stub to_datastore_bulk to return empty on empty_widget - allow(empty_widget).to receive(:to_datastore_bulk).and_return([]) - - warehouse_dumper.bulk([valid_widget, empty_widget]) - - # Verify only one record in the JSONL - params = s3_client.api_requests.first.fetch(:params) - jsonl_content = ::Zlib::GzipReader.new(StringIO.new(params[:body])).read - lines = jsonl_content.split("\n") - - expect(lines.size).to eq 1 - record = ::JSON.parse(lines[0]) - expect(record["id"]).to eq "1" - end - it "handles empty operations list without creating S3 files" do warehouse_dumper.bulk([]) @@ -196,6 +181,7 @@ class WarehouseLambda "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"} }) 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 index d4af0f4fe..6e1ab898c 100644 --- a/elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda_spec.rb +++ b/elasticgraph-warehouse_lambda/spec/unit/elastic_graph/warehouse_lambda_spec.rb @@ -13,7 +13,7 @@ module ElasticGraph RSpec.describe WarehouseLambda do include BuildsWarehouseLambda - # Without these ENV vars, instanting the S3 client causes it to try to fetch instance profile creds, which significantly + # 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| @@ -35,14 +35,14 @@ module ElasticGraph warehouse_lambda = WarehouseLambda.from_parsed_yaml(parsed_yaml) expect(warehouse_lambda).to be_a WarehouseLambda - expect(warehouse_lambda.indexer).to be_a ::ElasticGraph::Indexer + 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 ::ElasticGraph::Errors::ConfigError, a_string_including("warehouse") + }.to raise_error Errors::ConfigError, a_string_including("warehouse") end end From b716416a68836b6289c9e1dd72240700e7f0a620 Mon Sep 17 00:00:00 2001 From: Joshua Wilson Date: Mon, 19 Jan 2026 19:08:45 -0600 Subject: [PATCH 4/5] move shared method to spec support. --- .../indexer/datastore_indexing_router_spec.rb | 4 +- .../indexer/operation/factory_spec.rb | 4 +- .../elastic_graph/indexer/processor_spec.rb | 4 +- .../warehouse_lambda/warehouse_dumper_spec.rb | 43 +++++-------------- .../spec_support/builds_indexer_operation.rb | 10 ++--- 5 files changed, 21 insertions(+), 44 deletions(-) rename elasticgraph-indexer/spec/support/primary_indexing_operation_support.rb => spec_support/lib/elastic_graph/spec_support/builds_indexer_operation.rb (86%) 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/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 56e8ada8f..47109d580 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 @@ -10,11 +10,13 @@ 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" } @@ -24,7 +26,7 @@ class WarehouseLambda let(:indexer) { warehouse_lambda.indexer } let(:widget_primary_indexing_op) do - new_primary_indexing_op({ + new_primary_indexing_operation({ "type" => "Widget", "id" => "1", "version" => 3, @@ -34,8 +36,8 @@ class WarehouseLambda end it "writes operations to S3 as gzipped JSONL files and returns success results" do - op1 = new_primary_indexing_op({"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_op({"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"}}) + 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) @@ -97,8 +99,8 @@ class WarehouseLambda end it "writes operations of different types to separate S3 files" do - widget_op = new_primary_indexing_op({"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_op({"type" => "Component", "id" => "c1", "version" => 2, "json_schema_version" => 1, "record" => {"id" => "c1", "created_at" => "2024-09-15T12:30:12Z"}}) + 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) @@ -111,8 +113,8 @@ class WarehouseLambda end it "logs structured information about received batch and dumped files" do - widget_op = new_primary_indexing_op({"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_op({"type" => "Component", "id" => "c1", "version" => 2, "json_schema_version" => 1, "record" => {"id" => "c1", "created_at" => "2024-09-15T12:30:12Z"}}) + 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) @@ -177,7 +179,7 @@ class WarehouseLambda 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_op({ + widget_op = new_primary_indexing_operation({ "type" => "Widget", "id" => "1", "version" => 3, @@ -194,31 +196,6 @@ class WarehouseLambda # Should not create any S3 files when all operations are filtered expect(s3_client.api_requests).to be_empty end - - def new_primary_indexing_op(event) - update_targets = indexer - .schema_artifacts - .runtime_metadata - .object_types_by_name - .fetch(event.fetch("type")) - .update_targets - .select { |ut| ut.type == event.fetch("type") } - - expect(update_targets.size).to eq(1) - index_def = indexer.datastore_core.index_definitions_by_graphql_type.fetch(event.fetch("type")).first - - ::ElasticGraph::Indexer::Operation::Update.new( - event: event, - prepared_record: indexer.record_preparer_factory.for_latest_json_schema_version.prepare_for_index( - event.fetch("type"), - event.fetch("record") - ), - destination_index_def: index_def, - update_target: update_targets.first, - doc_id: event.fetch("id"), - destination_index_mapping: indexer.schema_artifacts.index_mappings_by_index_def_name.fetch(index_def.name) - ) - 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"), From c2f23d25ebd8aa43d447145f7367eb86ac20237f Mon Sep 17 00:00:00 2001 From: Josh Wilson Date: Mon, 19 Jan 2026 17:24:35 -0600 Subject: [PATCH 5/5] Update elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda.rbs Co-authored-by: Myron Marston --- .../sig/elastic_graph/warehouse_lambda.rbs | 2 -- 1 file changed, 2 deletions(-) diff --git a/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda.rbs b/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda.rbs index 0afd4e863..279721c85 100644 --- a/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda.rbs +++ b/elasticgraph-warehouse_lambda/sig/elastic_graph/warehouse_lambda.rbs @@ -9,8 +9,6 @@ module ElasticGraph attr_reader logger: ::Logger attr_reader clock: singleton(::Time) - def self.from_parsed_yaml: (parsedYamlSettings) ?{ (untyped) -> void } -> WarehouseLambda - def initialize: ( config: Config, indexer_config: Indexer::Config,