Perform repositories clean up on Geo::RepositoriesCleanUpWorker

app/workers/geo/repositories_clean_up_worker.rb

@@ -3,15 +3,41 @@ module Geo
     include Sidekiq::Worker
     include GeoQueue
 
+    BATCH_SIZE = 250
+
     def perform(geo_node_id)
       geo_node = GeoNode.find(geo_node_id)
+
+      restricted_project_ids = geo_node.project_ids
+      return unless restricted_project_ids
+
+      Project.where.not(id: restricted_project_ids).find_in_batches(batch_size: BATCH_SIZE) do |batch|
+        batch.each do |project|
+          clean_up_repositories(project)
+        end
+      end
     rescue ActiveRecord::RecordNotFound => e
-      Gitlab::Geo::Logger.error(
-        class: self.class.name,
-        message: 'Could not find Geo node, skipping repositories clean up',
-        geo_node_id: geo_node_id,
-        error: e
-      )
+      log_error('Could not find Geo node, skipping repositories clean up', geo_node_id: geo_node_id, error: e)
+    end
+
+    private
+
+    def clean_up_repositories(project)
+      job_id = ::GeoRepositoryDestroyWorker.perform_async(project.id, project.name, project.full_path)
+
+      if job_id
+        log_info('Repository cleaned up', project_id: project.id, full_path: project.full_path, job_id: job_id)
+      else
+        log_error('Could not clean up repository', project_id: project.id, full_path: project.full_path)
+      end
+    end
+
+    def log_info(message, params = {})
+      Gitlab::Geo::Logger.info({ class: self.class.name, message: message }.merge(params))
+    end
+
+    def log_error(message, params = {})
+      Gitlab::Geo::Logger.error({ class: self.class.name, message: message }.merge(params))
     end
   end
 end

spec/workers/geo/repositories_clean_up_worker_spec.rb

+require 'spec_helper'
+
+describe Geo::RepositoriesCleanUpWorker do
+  let!(:geo_node)  { create(:geo_node) }
+  let(:group)      { create(:group) }
+  let!(:project_1) { create(:empty_project, group: group) }
+  let!(:project_2) { create(:empty_project) }
+
+  describe '#perform' do
+    context 'when node have namespace restrictions' do
+      it 'performs GeoRepositoryDestroyWorker for each project that do not belong to selected namespaces to replicate' do
+        geo_node.update_attribute(:namespaces, [group])
+
+        expect(GeoRepositoryDestroyWorker).to receive(:perform_async)
+          .with(project_2.id, project_2.name, project_2.full_path)
+          .once.and_return(1)
+
+        subject.perform(geo_node.id)
+      end
+    end
+
+    context 'when does not node have namespace restrictions' do
+      it 'does not perform GeoRepositoryDestroyWorker' do
+        expect(GeoRepositoryDestroyWorker).not_to receive(:perform_async)
+
+        subject.perform(geo_node.id)
+      end
+    end
+
+    context 'when Geo node could not be found' do
+      it 'does not raise an error' do
+        expect { subject.perform(-1) }.not_to raise_error
+      end
+    end
+  end
+end