diff --git a/app/workers/all_queues.yml b/app/workers/all_queues.yml
index 5b1a25396afa894290150a4921460656c0a5d19a..7e186baf13401f9c3d7f3dbb3186c1d38b23f829 100644
--- a/app/workers/all_queues.yml
+++ b/app/workers/all_queues.yml
@@ -282,6 +282,15 @@
:weight: 1
:idempotent: true
:tags: []
+- :name: cronjob:container_registry_migration_observer
+ :worker_name: ContainerRegistry::Migration::ObserverWorker
+ :feature_category: :container_registry
+ :has_external_dependencies:
+ :urgency: :low
+ :resource_boundary: :unknown
+ :weight: 1
+ :idempotent: true
+ :tags: []
- :name: cronjob:database_batched_background_migration
:worker_name: Database::BatchedBackgroundMigrationWorker
:feature_category: :database
diff --git a/app/workers/container_registry/migration/observer_worker.rb b/app/workers/container_registry/migration/observer_worker.rb
new file mode 100644
index 0000000000000000000000000000000000000000..757c4fd11a5e308cbc5efaae54fc8e46f7f46440
--- /dev/null
+++ b/app/workers/container_registry/migration/observer_worker.rb
@@ -0,0 +1,40 @@
+# frozen_string_literal: true
+
+module ContainerRegistry
+ module Migration
+ class ObserverWorker
+ include ApplicationWorker
+ # This worker does not perform work scoped to a context
+ include CronjobQueue # rubocop:disable Scalability/CronWorkerContext
+
+ COUNT_BATCH_SIZE = 50000
+
+ data_consistency :sticky
+ feature_category :container_registry
+ urgency :low
+ deduplicate :until_executed, including_scheduled: true
+ idempotent!
+
+ def perform
+ return unless ::ContainerRegistry::Migration.enabled?
+
+ use_replica_if_available do
+ ContainerRepository::MIGRATION_STATES.each do |state|
+ relation = ContainerRepository.with_migration_state(state)
+ count = ::Gitlab::Database::BatchCount.batch_count(
+ relation, batch_size: COUNT_BATCH_SIZE
+ )
+ name = "#{state}_count".to_sym
+ log_extra_metadata_on_done(name, count)
+ end
+ end
+ end
+
+ private
+
+ def use_replica_if_available(&block)
+ ::Gitlab::Database::LoadBalancing::Session.current.use_replicas_for_read_queries(&block)
+ end
+ end
+ end
+end
diff --git a/config/initializers/1_settings.rb b/config/initializers/1_settings.rb
index 692488226580e5517786a9c119c95bf065758438..c634dbafe836d1060a5baa1aac93259f223d08f4 100644
--- a/config/initializers/1_settings.rb
+++ b/config/initializers/1_settings.rb
@@ -542,6 +542,9 @@
Settings.cron_jobs['container_registry_migration_guard_worker'] ||= Settingslogic.new({})
Settings.cron_jobs['container_registry_migration_guard_worker']['cron'] ||= '*/10 * * * *'
Settings.cron_jobs['container_registry_migration_guard_worker']['job_class'] = 'ContainerRegistry::Migration::GuardWorker'
+Settings.cron_jobs['container_registry_migration_observer_worker'] ||= Settingslogic.new({})
+Settings.cron_jobs['container_registry_migration_observer_worker']['cron'] ||= '*/30 * * * *'
+Settings.cron_jobs['container_registry_migration_observer_worker']['job_class'] = 'ContainerRegistry::Migration::ObserverWorker'
Settings.cron_jobs['image_ttl_group_policy_worker'] ||= Settingslogic.new({})
Settings.cron_jobs['image_ttl_group_policy_worker']['cron'] ||= '40 0 * * *'
Settings.cron_jobs['image_ttl_group_policy_worker']['job_class'] = 'DependencyProxy::ImageTtlGroupPolicyWorker'
diff --git a/db/migrate/20220202034409_add_tmp_index_on_id_and_migration_state_to_container_repositories.rb b/db/migrate/20220202034409_add_tmp_index_on_id_and_migration_state_to_container_repositories.rb
new file mode 100644
index 0000000000000000000000000000000000000000..b999c871a3ef7875b046fb72bbc072e5ec6b184e
--- /dev/null
+++ b/db/migrate/20220202034409_add_tmp_index_on_id_and_migration_state_to_container_repositories.rb
@@ -0,0 +1,16 @@
+# frozen_string_literal: true
+
+class AddTmpIndexOnIdAndMigrationStateToContainerRepositories < Gitlab::Database::Migration[1.0]
+ INDEX_NAME = 'tmp_index_container_repositories_on_id_migration_state'
+
+ disable_ddl_transaction!
+
+ # Temporary index to be removed https://gitlab.com/gitlab-org/gitlab/-/issues/351783
+ def up
+ add_concurrent_index :container_repositories, [:id, :migration_state], name: INDEX_NAME
+ end
+
+ def down
+ remove_concurrent_index_by_name :container_repositories, INDEX_NAME
+ end
+end
diff --git a/db/schema_migrations/20220202034409 b/db/schema_migrations/20220202034409
new file mode 100644
index 0000000000000000000000000000000000000000..4eb359f45e6d3c16130fed325d8538cb2259a451
--- /dev/null
+++ b/db/schema_migrations/20220202034409
@@ -0,0 +1 @@
+0efe482aa626cf80912feaa1176837253b094fc434f273bee35b5fe3e8ce4243
\ No newline at end of file
diff --git a/db/structure.sql b/db/structure.sql
index 8d3a7607a1f7828aefe7a6cd4be80c620534c75d..c1645c2169f55e9726be7c03ac51d8fe3079bbc8 100644
--- a/db/structure.sql
+++ b/db/structure.sql
@@ -28320,6 +28320,8 @@ CREATE INDEX tmp_idx_deduplicate_vulnerability_occurrences ON vulnerability_occu
CREATE INDEX tmp_idx_vulnerability_occurrences_on_id_where_report_type_7_99 ON vulnerability_occurrences USING btree (id) WHERE (report_type = ANY (ARRAY[7, 99]));
+CREATE INDEX tmp_index_container_repositories_on_id_migration_state ON container_repositories USING btree (id, migration_state);
+
CREATE INDEX tmp_index_for_namespace_id_migration_on_routes ON routes USING btree (id) WHERE ((namespace_id IS NULL) AND ((source_type)::text = 'Namespace'::text));
CREATE INDEX tmp_index_members_on_state ON members USING btree (state) WHERE (state = 2);
diff --git a/spec/workers/container_registry/migration/observer_worker_spec.rb b/spec/workers/container_registry/migration/observer_worker_spec.rb
new file mode 100644
index 0000000000000000000000000000000000000000..fec6640d7ec88478682bf38e25dae8d333b8ba75
--- /dev/null
+++ b/spec/workers/container_registry/migration/observer_worker_spec.rb
@@ -0,0 +1,57 @@
+# frozen_string_literal: true
+
+require 'spec_helper'
+
+RSpec.describe ContainerRegistry::Migration::ObserverWorker, :aggregate_failures do
+ let(:worker) { described_class.new }
+
+ describe '#perform' do
+ subject { worker.perform }
+
+ context 'when the migration feature flag is disabled' do
+ before do
+ stub_feature_flags(container_registry_migration_phase2_enabled: false)
+ end
+
+ it 'does nothing' do
+ expect(worker).not_to receive(:log_extra_metadata_on_done)
+
+ subject
+ end
+ end
+
+ context 'when the migration is enabled' do
+ before do
+ create_list(:container_repository, 3)
+ create(:container_repository, :pre_importing)
+ create(:container_repository, :pre_import_done)
+ create_list(:container_repository, 2, :importing)
+ create(:container_repository, :import_aborted)
+ # batch_count is not allowed within a transaction but
+ # all rspec tests run inside of a transaction.
+ # This mocks the false positive.
+ allow(ActiveRecord::Base.connection).to receive(:transaction_open?).and_return(false) # rubocop:disable Database/MultipleDatabases
+ end
+
+ it 'logs all the counts' do
+ expect(worker).to receive(:log_extra_metadata_on_done).with(:default_count, 3)
+ expect(worker).to receive(:log_extra_metadata_on_done).with(:pre_importing_count, 1)
+ expect(worker).to receive(:log_extra_metadata_on_done).with(:pre_import_done_count, 1)
+ expect(worker).to receive(:log_extra_metadata_on_done).with(:importing_count, 2)
+ expect(worker).to receive(:log_extra_metadata_on_done).with(:import_done_count, 0)
+ expect(worker).to receive(:log_extra_metadata_on_done).with(:import_aborted_count, 1)
+ expect(worker).to receive(:log_extra_metadata_on_done).with(:import_skipped_count, 0)
+
+ subject
+ end
+
+ context 'with load balancing enabled', :db_load_balancing do
+ it 'uses the replica' do
+ expect(Gitlab::Database::LoadBalancing::Session.current).to receive(:use_replicas_for_read_queries).and_call_original
+
+ subject
+ end
+ end
+ end
+ end
+end