debian-mirror-gitlab/lib/gitlab/database/background_migration/batched_migration_wrapper.rb

Ignoring revisions in .git-blame-ignore-revs. Click here to bypass and see the normal blame view.

92 lines
2.8 KiB
Ruby
Raw Normal View History

2021-04-17 20:07:23 +05:30
# frozen_string_literal: true
module Gitlab
module Database
module BackgroundMigration
class BatchedMigrationWrapper
2022-06-21 17:19:12 +05:30
def initialize(connection:, metrics: PrometheusMetrics.new)
2022-04-04 11:22:00 +05:30
@connection = connection
2022-06-21 17:19:12 +05:30
@metrics = metrics
2022-04-04 11:22:00 +05:30
end
2021-04-29 21:17:54 +05:30
# Wraps the execution of a batched_background_migration.
#
# Updates the job's tracking records with the status of the migration
# when starting and finishing execution, and optionally saves batch_metrics
# the migration provides, if any are given.
#
# The job's batch_metrics are serialized to JSON for storage.
2021-04-17 20:07:23 +05:30
def perform(batch_tracking_record)
start_tracking_execution(batch_tracking_record)
execute_batch(batch_tracking_record)
2022-04-04 11:22:00 +05:30
batch_tracking_record.succeed!
rescue Exception => error # rubocop:disable Lint/RescueException
batch_tracking_record.failure!(error: error)
2021-04-17 20:07:23 +05:30
2021-06-08 01:23:25 +05:30
raise
2021-04-17 20:07:23 +05:30
ensure
2022-06-21 17:19:12 +05:30
metrics.track(batch_tracking_record)
2021-04-17 20:07:23 +05:30
end
private
2022-06-21 17:19:12 +05:30
attr_reader :connection, :metrics
2022-04-04 11:22:00 +05:30
2021-04-17 20:07:23 +05:30
def start_tracking_execution(tracking_record)
2022-04-04 11:22:00 +05:30
tracking_record.run!
2021-04-17 20:07:23 +05:30
end
def execute_batch(tracking_record)
2022-07-16 23:28:13 +05:30
job_instance = execute_job(tracking_record)
if job_instance.respond_to?(:batch_metrics)
tracking_record.metrics = job_instance.batch_metrics
end
end
def execute_job(tracking_record)
job_class = tracking_record.migration_job_class
if job_class < Gitlab::BackgroundMigration::BatchedMigrationJob
execute_batched_migration_job(job_class, tracking_record)
else
execute_legacy_job(job_class, tracking_record)
end
end
def execute_batched_migration_job(job_class, tracking_record)
job_instance = job_class.new(
start_id: tracking_record.min_value,
end_id: tracking_record.max_value,
batch_table: tracking_record.migration_table_name,
batch_column: tracking_record.migration_column_name,
sub_batch_size: tracking_record.sub_batch_size,
pause_ms: tracking_record.pause_ms,
connection: connection)
job_instance.perform(*tracking_record.migration_job_arguments)
job_instance
end
def execute_legacy_job(job_class, tracking_record)
job_instance = job_class.new
2021-04-17 20:07:23 +05:30
job_instance.perform(
tracking_record.min_value,
tracking_record.max_value,
tracking_record.migration_table_name,
tracking_record.migration_column_name,
tracking_record.sub_batch_size,
2021-06-08 01:23:25 +05:30
tracking_record.pause_ms,
2021-04-17 20:07:23 +05:30
*tracking_record.migration_job_arguments)
2021-04-29 21:17:54 +05:30
2022-07-16 23:28:13 +05:30
job_instance
2021-04-17 20:07:23 +05:30
end
end
end
end
end