diff options
Diffstat (limited to 'spec/support/shared_examples/workers')
4 files changed, 354 insertions, 82 deletions
diff --git a/spec/support/shared_examples/workers/batched_background_migration_execution_worker_shared_example.rb b/spec/support/shared_examples/workers/batched_background_migration_execution_worker_shared_example.rb new file mode 100644 index 00000000000..ae29b76ee87 --- /dev/null +++ b/spec/support/shared_examples/workers/batched_background_migration_execution_worker_shared_example.rb @@ -0,0 +1,203 @@ +# frozen_string_literal: true + +RSpec.shared_examples 'batched background migrations execution worker' do + include ExclusiveLeaseHelpers + + it 'is a limited capacity worker' do + expect(described_class.new).to be_a(LimitedCapacity::Worker) + end + + describe 'defining the job attributes' do + it 'defines the data_consistency as always' do + expect(described_class.get_data_consistency).to eq(:always) + end + + it 'defines the feature_category as database' do + expect(described_class.get_feature_category).to eq(:database) + end + + it 'defines the idempotency as false' do + expect(described_class).not_to be_idempotent + end + + it 'does not retry failed jobs' do + expect(described_class.sidekiq_options['retry']).to eq(0) + end + + it 'does not deduplicate jobs' do + expect(described_class.get_deduplicate_strategy).to eq(:none) + end + + it 'defines the queue namespace' do + expect(described_class.queue_namespace).to eq('batched_background_migrations') + end + end + + describe '.perform_with_capacity' do + it 'enqueues jobs without modifying provided arguments' do + expect_next_instance_of(described_class) do |instance| + expect(instance).to receive(:remove_failed_jobs) + end + + args = [['main', 123]] + + expect(described_class) + .to receive(:bulk_perform_async) + .with(args) + + described_class.perform_with_capacity(args) + end + end + + describe '.max_running_jobs' do + it 'returns MAX_RUNNING_MIGRATIONS' do + expect(described_class.max_running_jobs).to eq(described_class::MAX_RUNNING_MIGRATIONS) + end + end + + describe '#max_running_jobs' do + it 'returns MAX_RUNNING_MIGRATIONS' do + expect(described_class.new.max_running_jobs).to eq(described_class::MAX_RUNNING_MIGRATIONS) + end + end + + describe '#remaining_work_count' do + it 'returns 0' do + expect(described_class.new.remaining_work_count).to eq(0) + end + end + + describe '#perform_work' do + let(:database_name) { Gitlab::Database::MAIN_DATABASE_NAME.to_sym } + let(:base_model) { Gitlab::Database.database_base_models[database_name] } + let(:table_name) { :events } + let(:job_interval) { 5.minutes } + let(:lease_timeout) { job_interval * described_class::LEASE_TIMEOUT_MULTIPLIER } + let(:interval_variance) { described_class::INTERVAL_VARIANCE } + + subject(:worker) { described_class.new } + + context 'when the feature flag is disabled' do + let(:migration) do + create(:batched_background_migration, :active, interval: job_interval, table_name: table_name) + end + + before do + stub_feature_flags(execute_batched_migrations_on_schedule: false) + end + + it 'does nothing' do + expect(Gitlab::Database::BackgroundMigration::BatchedMigration).not_to receive(:find_executable) + expect(worker).not_to receive(:run_migration_job) + + worker.perform_work(database_name, migration.id) + end + end + + context 'when the feature flag is enabled' do + before do + stub_feature_flags(execute_batched_migrations_on_schedule: true) + end + + context 'when the provided database is sharing config' do + before do + skip_if_multiple_databases_not_setup + end + + it 'does nothing' do + ci_model = Gitlab::Database.database_base_models['ci'] + expect(Gitlab::Database).to receive(:db_config_share_with) + .with(ci_model.connection_db_config).and_return('main') + + expect(Gitlab::Database::BackgroundMigration::BatchedMigration).not_to receive(:find_executable) + expect(worker).not_to receive(:run_migration_job) + + worker.perform_work(:ci, 123) + end + end + + context 'when migration does not exist' do + it 'does nothing' do + expect(worker).not_to receive(:run_migration_job) + + worker.perform_work(database_name, non_existing_record_id) + end + end + + context 'when migration exist' do + let(:migration) do + create(:batched_background_migration, :active, interval: job_interval, table_name: table_name) + end + + before do + allow(Gitlab::Database::BackgroundMigration::BatchedMigration).to receive(:find_executable) + .with(migration.id, connection: base_model.connection) + .and_return(migration) + end + + context 'when the migration is no longer active' do + it 'does not run the migration' do + expect(Gitlab::Database::SharedModel).to receive(:using_connection).with(base_model.connection).and_yield + + expect(migration).to receive(:active?).and_return(false) + + expect(worker).not_to receive(:run_migration_job) + + worker.perform_work(database_name, migration.id) + end + end + + context 'when the interval has not elapsed' do + it 'does not run the migration' do + expect(Gitlab::Database::SharedModel).to receive(:using_connection).with(base_model.connection).and_yield + expect(migration).to receive(:interval_elapsed?).with(variance: interval_variance).and_return(false) + expect(worker).not_to receive(:run_migration_job) + + worker.perform_work(database_name, migration.id) + end + end + + context 'when the migration is still active and the interval has elapsed' do + let(:table_name_lease_key) do + "#{described_class.name.underscore}:database_name:#{database_name}:" \ + "table_name:#{table_name}" + end + + context 'when can not obtain lease on the table name' do + it 'does nothing' do + stub_exclusive_lease_taken(table_name_lease_key, timeout: lease_timeout) + + expect(worker).not_to receive(:run_migration_job) + + worker.perform_work(database_name, migration.id) + end + end + + it 'always cleans up the exclusive lease' do + expect_to_obtain_exclusive_lease(table_name_lease_key, 'uuid-table-name', timeout: lease_timeout) + expect_to_cancel_exclusive_lease(table_name_lease_key, 'uuid-table-name') + + expect(worker).to receive(:run_migration_job).and_raise(RuntimeError, 'I broke') + + expect { worker.perform_work(database_name, migration.id) }.to raise_error(RuntimeError, 'I broke') + end + + it 'runs the migration' do + expect(Gitlab::Database::SharedModel).to receive(:using_connection).with(base_model.connection).and_yield + + expect_next_instance_of(Gitlab::Database::BackgroundMigration::BatchedMigrationRunner) do |instance| + expect(instance).to receive(:run_migration_job).with(migration) + end + + expect_to_obtain_exclusive_lease(table_name_lease_key, 'uuid-table-name', timeout: lease_timeout) + expect_to_cancel_exclusive_lease(table_name_lease_key, 'uuid-table-name') + + expect(worker).to receive(:run_migration_job).and_call_original + + worker.perform_work(database_name, migration.id) + end + end + end + end + end +end diff --git a/spec/support/shared_examples/workers/batched_background_migration_worker_shared_examples.rb b/spec/support/shared_examples/workers/batched_background_migration_worker_shared_examples.rb index 0be55fd2a3e..09ebc495e61 100644 --- a/spec/support/shared_examples/workers/batched_background_migration_worker_shared_examples.rb +++ b/spec/support/shared_examples/workers/batched_background_migration_worker_shared_examples.rb @@ -125,10 +125,28 @@ RSpec.shared_examples 'it runs batched background migration jobs' do |tracking_d end context 'when no active migrations exist' do - it 'does nothing' do - expect(worker).not_to receive(:run_active_migration) + context 'when parallel execution is disabled' do + before do + stub_feature_flags(batched_migrations_parallel_execution: false) + end - worker.perform + it 'does nothing' do + expect(worker).not_to receive(:run_active_migration) + + worker.perform + end + end + + context 'when parallel execution is enabled' do + before do + stub_feature_flags(batched_migrations_parallel_execution: true) + end + + it 'does nothing' do + expect(worker).not_to receive(:queue_migrations_for_execution) + + worker.perform + end end end @@ -136,7 +154,6 @@ RSpec.shared_examples 'it runs batched background migration jobs' do |tracking_d let(:job_interval) { 5.minutes } let(:lease_timeout) { 15.minutes } let(:lease_key) { described_class.name.demodulize.underscore } - let(:interval_variance) { described_class::INTERVAL_VARIANCE } let(:migration_id) { 123 } let(:migration) do build( @@ -145,52 +162,86 @@ RSpec.shared_examples 'it runs batched background migration jobs' do |tracking_d ) end + let(:execution_worker_class) do + case tracking_database + when :main + Database::BatchedBackgroundMigration::MainExecutionWorker + when :ci + Database::BatchedBackgroundMigration::CiExecutionWorker + end + end + before do allow(Gitlab::Database::BackgroundMigration::BatchedMigration).to receive(:active_migration) .with(connection: base_model.connection) .and_return(migration) - - allow(migration).to receive(:interval_elapsed?).with(variance: interval_variance).and_return(true) - allow(migration).to receive(:reload) end - context 'when the calculated timeout is less than the minimum allowed' do - let(:minimum_timeout) { described_class::MINIMUM_LEASE_TIMEOUT } - let(:job_interval) { 2.minutes } + context 'when parallel execution is disabled' do + before do + stub_feature_flags(batched_migrations_parallel_execution: false) + end + + let(:execution_worker) { instance_double(execution_worker_class) } - it 'sets the lease timeout to the minimum value' do - expect_to_obtain_exclusive_lease(lease_key, timeout: minimum_timeout) + context 'when the calculated timeout is less than the minimum allowed' do + let(:minimum_timeout) { described_class::MINIMUM_LEASE_TIMEOUT } + let(:job_interval) { 2.minutes } - expect_next_instance_of(Database::BatchedBackgroundMigration::ExecutionWorker) do |worker| - expect(worker).to receive(:perform).with(tracking_database, migration_id) + it 'sets the lease timeout to the minimum value' do + expect_to_obtain_exclusive_lease(lease_key, timeout: minimum_timeout) + + expect(execution_worker_class).to receive(:new).and_return(execution_worker) + expect(execution_worker).to receive(:perform_work).with(tracking_database, migration_id) + + expect(worker).to receive(:run_active_migration).and_call_original + + worker.perform end + end - expect(worker).to receive(:run_active_migration).and_call_original + it 'always cleans up the exclusive lease' do + lease = stub_exclusive_lease_taken(lease_key, timeout: lease_timeout) + + expect(lease).to receive(:try_obtain).and_return(true) + + expect(worker).to receive(:run_active_migration).and_raise(RuntimeError, 'I broke') + expect(lease).to receive(:cancel) + + expect { worker.perform }.to raise_error(RuntimeError, 'I broke') + end + + it 'delegetes the execution to ExecutionWorker' do + base_model = Gitlab::Database.database_base_models[tracking_database] + + expect(Gitlab::Database::SharedModel).to receive(:using_connection).with(base_model.connection).and_yield + expect(execution_worker_class).to receive(:new).and_return(execution_worker) + expect(execution_worker).to receive(:perform_work).with(tracking_database, migration_id) worker.perform end end - it 'always cleans up the exclusive lease' do - lease = stub_exclusive_lease_taken(lease_key, timeout: lease_timeout) + context 'when parallel execution is enabled' do + before do + stub_feature_flags(batched_migrations_parallel_execution: true) + end - expect(lease).to receive(:try_obtain).and_return(true) + it 'delegetes the execution to ExecutionWorker' do + expect(Gitlab::Database::BackgroundMigration::BatchedMigration) + .to receive(:active_migrations_distinct_on_table).with( + connection: base_model.connection, + limit: execution_worker_class.max_running_jobs + ).and_return([migration]) - expect(worker).to receive(:run_active_migration).and_raise(RuntimeError, 'I broke') - expect(lease).to receive(:cancel) + expected_arguments = [ + [tracking_database.to_s, migration_id] + ] - expect { worker.perform }.to raise_error(RuntimeError, 'I broke') - end + expect(execution_worker_class).to receive(:perform_with_capacity).with(expected_arguments) - it 'delegetes the execution to ExecutionWorker' do - base_model = Gitlab::Database.database_base_models[tracking_database] - - expect(Gitlab::Database::SharedModel).to receive(:using_connection).with(base_model.connection).and_yield - expect_next_instance_of(Database::BatchedBackgroundMigration::ExecutionWorker) do |worker| - expect(worker).to receive(:perform).with(tracking_database, migration_id) + worker.perform end - - worker.perform end end end @@ -249,6 +300,8 @@ RSpec.shared_examples 'it runs batched background migration jobs' do |tracking_d end before do + stub_feature_flags(execute_batched_migrations_on_schedule: true) + # Create example table populated with test data to migrate. # # Test data should have two records that won't be updated: @@ -269,80 +322,96 @@ RSpec.shared_examples 'it runs batched background migration jobs' do |tracking_d WHERE some_column = #{migration_records - 5}; SQL - stub_feature_flags(execute_batched_migrations_on_schedule: true) - stub_const('Gitlab::BackgroundMigration::ExampleDataMigration', migration_class) end - subject(:full_migration_run) do - # process all batches, then do an extra execution to mark the job as finished - (number_of_batches + 1).times do - described_class.new.perform + shared_examples 'batched background migration execution' do + subject(:full_migration_run) do + # process all batches, then do an extra execution to mark the job as finished + (number_of_batches + 1).times do + described_class.new.perform - travel_to((migration.interval + described_class::INTERVAL_VARIANCE).seconds.from_now) + travel_to((migration.interval + described_class::INTERVAL_VARIANCE).seconds.from_now) + end end - end - it 'marks the migration record as finished' do - expect { full_migration_run }.to change { migration.reload.status }.from(1).to(3) # active -> finished - end + it 'marks the migration record as finished' do + expect { full_migration_run }.to change { migration.reload.status }.from(1).to(3) # active -> finished + end - it 'creates job records for each processed batch', :aggregate_failures do - expect { full_migration_run }.to change { migration.reload.batched_jobs.count }.from(0) + it 'creates job records for each processed batch', :aggregate_failures do + expect { full_migration_run }.to change { migration.reload.batched_jobs.count }.from(0) - final_min_value = migration.batched_jobs.reduce(1) do |next_min_value, batched_job| - expect(batched_job.min_value).to eq(next_min_value) + final_min_value = migration.batched_jobs.order(id: :asc).reduce(1) do |next_min_value, batched_job| + expect(batched_job.min_value).to eq(next_min_value) - batched_job.max_value + 1 + batched_job.max_value + 1 + end + + final_max_value = final_min_value - 1 + expect(final_max_value).to eq(migration_records) end - final_max_value = final_min_value - 1 - expect(final_max_value).to eq(migration_records) - end + it 'marks all job records as succeeded', :aggregate_failures do + expect { full_migration_run }.to change { migration.reload.batched_jobs.count }.from(0) - it 'marks all job records as succeeded', :aggregate_failures do - expect { full_migration_run }.to change { migration.reload.batched_jobs.count }.from(0) + expect(migration.batched_jobs).to all(be_succeeded) + end - expect(migration.batched_jobs).to all(be_succeeded) - end + it 'updates matching records in the range', :aggregate_failures do + expect { full_migration_run } + .to change { example_data.where('status = 1 AND some_column <> 0').count } + .from(migration_records).to(1) - it 'updates matching records in the range', :aggregate_failures do - expect { full_migration_run } - .to change { example_data.where('status = 1 AND some_column <> 0').count } - .from(migration_records).to(1) + record_outside_range = example_data.last - record_outside_range = example_data.last + expect(record_outside_range.status).to eq(1) + expect(record_outside_range.some_column).not_to eq(0) + end - expect(record_outside_range.status).to eq(1) - expect(record_outside_range.some_column).not_to eq(0) - end + it 'does not update non-matching records in the range' do + expect { full_migration_run }.not_to change { example_data.where('status <> 1 AND some_column <> 0').count } + end - it 'does not update non-matching records in the range' do - expect { full_migration_run }.not_to change { example_data.where('status <> 1 AND some_column <> 0').count } - end + context 'health status' do + subject(:migration_run) { described_class.new.perform } + + it 'puts migration on hold when there is autovaccum activity on related tables' do + swapout_view_for_table(:postgres_autovacuum_activity, connection: connection) + create( + :postgres_autovacuum_activity, + table: migration.table_name, + table_identifier: "public.#{migration.table_name}" + ) + + expect { migration_run }.to change { migration.reload.on_hold? }.from(false).to(true) + end - context 'health status' do - subject(:migration_run) { described_class.new.perform } + it 'puts migration on hold when the pending WAL count is above the limit' do + sql = Gitlab::Database::BackgroundMigration::HealthStatus::Indicators::WriteAheadLog::PENDING_WAL_COUNT_SQL + limit = Gitlab::Database::BackgroundMigration::HealthStatus::Indicators::WriteAheadLog::LIMIT - it 'puts migration on hold when there is autovaccum activity on related tables' do - swapout_view_for_table(:postgres_autovacuum_activity, connection: connection) - create( - :postgres_autovacuum_activity, - table: migration.table_name, - table_identifier: "public.#{migration.table_name}" - ) + expect(connection).to receive(:execute).with(sql).and_return([{ 'pending_wal_count' => limit + 1 }]) - expect { migration_run }.to change { migration.reload.on_hold? }.from(false).to(true) + expect { migration_run }.to change { migration.reload.on_hold? }.from(false).to(true) + end end + end - it 'puts migration on hold when the pending WAL count is above the limit' do - sql = Gitlab::Database::BackgroundMigration::HealthStatus::Indicators::WriteAheadLog::PENDING_WAL_COUNT_SQL - limit = Gitlab::Database::BackgroundMigration::HealthStatus::Indicators::WriteAheadLog::LIMIT + context 'when parallel execution is disabled' do + before do + stub_feature_flags(batched_migrations_parallel_execution: false) + end - expect(connection).to receive(:execute).with(sql).and_return([{ 'pending_wal_count' => limit + 1 }]) + it_behaves_like 'batched background migration execution' + end - expect { migration_run }.to change { migration.reload.on_hold? }.from(false).to(true) + context 'when parallel execution is enabled', :sidekiq_inline do + before do + stub_feature_flags(batched_migrations_parallel_execution: true) end + + it_behaves_like 'batched background migration execution' end end end diff --git a/spec/support/shared_examples/workers/schedule_bulk_repository_shard_moves_shared_examples.rb b/spec/support/shared_examples/workers/schedule_bulk_repository_shard_moves_shared_examples.rb index 465aca63148..6707f65eb69 100644 --- a/spec/support/shared_examples/workers/schedule_bulk_repository_shard_moves_shared_examples.rb +++ b/spec/support/shared_examples/workers/schedule_bulk_repository_shard_moves_shared_examples.rb @@ -15,7 +15,7 @@ RSpec.shared_examples 'schedules bulk repository shard moves' do let(:job_args) { [source_storage_name, destination_storage_name] } it 'schedules container repository storage moves' do - expect { subject }.to change(move_service_klass, :count).by(1) + expect { subject }.to change { move_service_klass.count }.by(1) storage_move = container.repository_storage_moves.last! diff --git a/spec/support/shared_examples/workers/update_repository_move_shared_examples.rb b/spec/support/shared_examples/workers/update_repository_move_shared_examples.rb index babd7cfbbeb..c50dc6d5372 100644 --- a/spec/support/shared_examples/workers/update_repository_move_shared_examples.rb +++ b/spec/support/shared_examples/workers/update_repository_move_shared_examples.rb @@ -15,7 +15,7 @@ RSpec.shared_examples 'an update storage move worker' do expect do subject.perform(container.id, 'test_second_storage') - end.to change(repository_storage_move_klass, :count).by(1) + end.to change { repository_storage_move_klass.count }.by(1) storage_move = container.repository_storage_moves.last expect(storage_move).to have_attributes( @@ -32,7 +32,7 @@ RSpec.shared_examples 'an update storage move worker' do expect do subject.perform(nil, nil, repository_storage_move.id) - end.not_to change(repository_storage_move_klass, :count) + end.not_to change { repository_storage_move_klass.count } end end end |