Allow sidekiq to react to becoming a Geo primary or secondary without a restart

Sidekiq runs a set of cron jobs. This set differs according to whether the node
is a Geo primary or secondary, or if Geo is disabled. Previously, the state was
determined once, when sidekiq was started. Now, we add a cron job that runs the
same logic each minute.

To reduce churn, the enable/disable logic now only changes jobs if they are in
the wrong state.

app/workers/all_queues.yml

@@ -103,6 +103,7 @@
 - cronjob:geo_metrics_update
 - cronjob:geo_prune_event_log
 - cronjob:geo_repository_sync
+- cronjob:geo_sidekiq_cron_config
 - cronjob:historical_data
 - cronjob:ldap_all_groups_sync
 - cronjob:ldap_sync

changelogs/unreleased-ee/4048-sidekiq-cron-geo-2.yml

+---
+title: Allow sidekiq to react to becoming a Geo primary or secondary without a restart
+merge_request: 3878
+author:
+type: changed

ee/app/workers/geo/sidekiq_cron_config_worker.rb

+module Geo
+  class SidekiqCronConfigWorker
+    include ApplicationWorker
+    include CronjobQueue
+
+    def perform
+      Gitlab::Geo::CronManager.new.execute
+    end
+  end
+end

ee/lib/gitlab/geo.rb

@@ -12,9 +12,6 @@ module Gitlab
       geo_oauth_application
     ).freeze
 
-    COMMON_CRON_JOBS = %i(geo_metrics_update_worker).freeze
-    SECONDARY_CRON_JOBS = %i(geo_repository_sync_worker geo_file_download_dispatch_worker).freeze
-
     FDW_SCHEMA = 'gitlab_secondary'.freeze
 
     def self.current_node
@@ -80,41 +77,10 @@ module Gitlab
       FDW_SCHEMA + ".#{table_name}"
     end
 
-    def self.configure_primary_jobs!
-      self.enable_all_cron_jobs!
-      SECONDARY_CRON_JOBS.each { |job_name| Sidekiq::Cron::Job.find(job_name).try(:disable!) }
-    end
-
-    def self.configure_secondary_jobs!
-      self.disable_all_cron_jobs!
-      (COMMON_CRON_JOBS + SECONDARY_CRON_JOBS).each { |job_name| Sidekiq::Cron::Job.find(job_name).try(:enable!) }
-    end
-
-    def self.disable_all_geo_jobs!
-      (COMMON_CRON_JOBS + SECONDARY_CRON_JOBS).each { |job_name| Sidekiq::Cron::Job.find(job_name).try(:disable!) }
-    end
-
-    def self.disable_all_cron_jobs!
-      self.cron_jobs.select(&:enabled?).each { |job| job.disable! }
-    end
-
-    def self.enable_all_cron_jobs!
-      self.cron_jobs.reject(&:enabled?).each { |job| job.enable! }
-    end
-
-    def self.cron_jobs
-      Sidekiq::Cron::Job.all
-    end
-
     def self.configure_cron_jobs!
-      if self.connected? && self.primary?
-        self.configure_primary_jobs!
-      elsif self.connected? && self.secondary?
-        self.configure_secondary_jobs!
-      else
-        self.enable_all_cron_jobs!
-        self.disable_all_geo_jobs!
-      end
+      manager = CronManager.new
+      manager.create_watcher!
+      manager.execute
     end
 
     def self.oauth_authentication

ee/lib/gitlab/geo/cron_manager.rb

+module Gitlab
+  module Geo
+    class CronManager
+      COMMON_JOBS = %w[geo_metrics_update_worker].freeze
+      SECONDARY_JOBS = %w[
+        geo_repository_sync_worker
+        geo_file_download_dispatch_worker
+      ].freeze
+      GEO_JOBS = (COMMON_JOBS + SECONDARY_JOBS).freeze
+
+      CONFIG_WATCHER = 'geo_sidekiq_cron_config_worker'.freeze
+      CONFIG_WATCHER_CLASS = 'Geo::SidekiqCronConfigWorker'.freeze
+
+      def execute
+        if Geo.connected? && Geo.primary?
+          configure_primary
+        elsif Geo.connected? && Geo.secondary?
+          configure_secondary
+        else
+          enable!(all_jobs(except: GEO_JOBS))
+          disable!(jobs(GEO_JOBS))
+        end
+      end
+
+      def create_watcher!
+        job(CONFIG_WATCHER)&.destroy
+
+        Sidekiq::Cron::Job.create(
+          name: CONFIG_WATCHER,
+          cron: '*/1 * * * *',
+          class: CONFIG_WATCHER_CLASS
+        )
+      end
+
+      private
+
+      def configure_primary
+        disable!(jobs(SECONDARY_JOBS))
+        enable!(all_jobs(except: SECONDARY_JOBS))
+      end
+
+      def configure_secondary
+        names = [CONFIG_WATCHER, COMMON_JOBS, SECONDARY_JOBS].flatten
+
+        disable!(all_jobs(except: names))
+        enable!(jobs(names))
+      end
+
+      def enable!(jobs)
+        jobs.compact.each { |job| job.enable! unless job.enabled? }
+      end
+
+      def disable!(jobs)
+        jobs.compact.each { |job| job.disable! unless job.disabled? }
+      end
+
+      def all_jobs(except: [])
+        Sidekiq::Cron::Job.all.reject { |job| except.include?(job.name) }
+      end
+
+      def jobs(names)
+        names.map { |name| job(name) }
+      end
+
+      def job(name)
+        Sidekiq::Cron::Job.find(name)
+      end
+    end
+  end
+end

spec/ee/spec/lib/gitlab/geo/cron_manager_spec.rb

+require 'spec_helper'
+
+describe Gitlab::Geo::CronManager, :geo do
+  include ::EE::GeoHelpers
+
+  def job(name)
+    Sidekiq::Cron::Job.find(name)
+  end
+
+  subject(:manager) { described_class.new }
+
+  describe '#execute' do
+    set(:primary) { create(:geo_node, :primary) }
+    set(:secondary) { create(:geo_node) }
+
+    def init_cron_job(job_name, class_name)
+      job = Sidekiq::Cron::Job.new(
+        name: job_name,
+        cron: '0 * * * *',
+        class: class_name
+      )
+
+      job.enable!
+    end
+
+    JOBS = %w(ldap_test geo_repository_sync_worker geo_file_download_dispatch_worker geo_metrics_update_worker).freeze
+
+    before(:all) do
+      JOBS.each { |name| init_cron_job(name, name.camelize) }
+    end
+
+    after(:all) do
+      JOBS.each { |name| job(name)&.destroy }
+    end
+
+    let(:common_jobs) { [job('geo_metrics_update_worker')] }
+    let(:ldap_test_job) { job('ldap_test') }
+
+    let(:secondary_jobs) do
+      [
+        job('geo_file_download_dispatch_worker'),
+        job('geo_repository_sync_worker')
+      ]
+    end
+
+    context 'on a Geo primary' do
+      before do
+        stub_current_geo_node(primary)
+
+        manager.execute
+      end
+
+      it 'disables secondary-only jobs' do
+        secondary_jobs.each { |job| expect(job).not_to be_enabled }
+      end
+
+      it 'enables common jobs' do
+        common_jobs.each { |job| expect(job).to be_enabled }
+      end
+
+      it 'enables non-geo jobs' do
+        expect(ldap_test_job).to be_enabled
+      end
+    end
+
+    context 'on a Geo secondary' do
+      before do
+        stub_current_geo_node(secondary)
+
+        manager.execute
+      end
+
+      it 'enables secondary-only jobs' do
+        secondary_jobs.each { |job| expect(job).to be_enabled }
+      end
+
+      it 'enables common jobs' do
+        common_jobs.each { |job| expect(job).to be_enabled }
+      end
+
+      it 'disables non-geo jobs' do
+        expect(ldap_test_job).not_to be_enabled
+      end
+    end
+
+    context 'on a non-Geo node' do
+      before do
+        stub_current_geo_node(nil)
+
+        manager.execute
+      end
+
+      it 'disables secondary-only jobs' do
+        secondary_jobs.each { |job| expect(job).not_to be_enabled }
+      end
+
+      it 'disables common jobs' do
+        common_jobs.each { |job| expect(job).not_to be_enabled }
+      end
+
+      it 'enables non-geo jobs' do
+        expect(ldap_test_job).to be_enabled
+      end
+    end
+  end
+
+  describe '#create_watcher!' do
+    it 'creates a Geo::SidekiqCronConfigWorker sidekiq-cron job' do
+      manager.create_watcher!
+
+      created = job('geo_sidekiq_cron_config_worker')
+
+      expect(created).not_to be_nil
+      expect(created.klass).to eq('Geo::SidekiqCronConfigWorker')
+      expect(created.cron).to eq('*/1 * * * *')
+      expect(created.name).to eq('geo_sidekiq_cron_config_worker')
+    end
+  end
+end

spec/ee/spec/lib/gitlab/geo_spec.rb

@@ -176,67 +176,22 @@ describe Gitlab::Geo, :geo do
   end
 
   describe '.configure_cron_jobs!' do
-    JOBS = %w(ldap_test geo_repository_sync_worker geo_file_download_dispatch_worker geo_metrics_update_worker).freeze
+    let(:manager) { double('cron_manager').as_null_object }
 
-    def init_cron_job(job_name, class_name)
-      job = Sidekiq::Cron::Job.new(
-        name: job_name,
-        cron: '0 * * * *',
-        class: class_name
-      )
-
-      job.enable!
-    end
-
-    before(:all) do
-      JOBS.each { |job| init_cron_job(job, job.camelize) }
-    end
-
-    after(:all) do
-      JOBS.each { |job| Sidekiq::Cron::Job.find(job)&.destroy }
-    end
-
-    it 'activates cron jobs for primary' do
-      described_class.configure_cron_jobs!
-
-      expect(Sidekiq::Cron::Job.find('geo_repository_sync_worker')).not_to be_enabled
-      expect(Sidekiq::Cron::Job.find('geo_file_download_dispatch_worker')).not_to be_enabled
-      expect(Sidekiq::Cron::Job.find('geo_metrics_update_worker')).to be_enabled
-      expect(Sidekiq::Cron::Job.find('ldap_test')).to be_enabled
-    end
-
-    it 'does not activate cron jobs for secondary' do
-      stub_current_geo_node(secondary_node)
-
-      described_class.configure_cron_jobs!
-
-      expect(Sidekiq::Cron::Job.find('ldap_test')).not_to be_enabled
-      expect(Sidekiq::Cron::Job.find('geo_metrics_update_worker')).to be_enabled
-      expect(Sidekiq::Cron::Job.find('geo_repository_sync_worker')).to be_enabled
-      expect(Sidekiq::Cron::Job.find('geo_file_download_dispatch_worker')).to be_enabled
+    before do
+      allow(Gitlab::Geo::CronManager).to receive(:new) { manager }
     end
 
-    it 'deactivates all jobs when Geo is not active' do
-      stub_current_geo_node(nil)
+    it 'creates a cron watcher' do
+      expect(manager).to receive(:create_watcher!)
 
       described_class.configure_cron_jobs!
-
-      expect(Sidekiq::Cron::Job.find('geo_repository_sync_worker')).not_to be_enabled
-      expect(Sidekiq::Cron::Job.find('geo_file_download_dispatch_worker')).not_to be_enabled
-      expect(Sidekiq::Cron::Job.find('geo_metrics_update_worker')).not_to be_enabled
-      expect(Sidekiq::Cron::Job.find('ldap_test')).to be_enabled
     end
 
-    it 'reactivates cron jobs when node turns off Geo' do
-      stub_current_geo_node(secondary_node)
-
-      described_class.configure_cron_jobs!
-      expect(Sidekiq::Cron::Job.find('ldap_test')).not_to be_enabled
-
-      allow(described_class).to receive(:secondary?).and_return(false)
+    it 'runs the cron manager' do
+      expect(manager).to receive(:execute)
 
       described_class.configure_cron_jobs!
-      expect(Sidekiq::Cron::Job.find('ldap_test')).to be_enabled
     end
   end
 end

spec/ee/spec/workers/geo/sidekiq_cron_config_worker_spec.rb

+require 'spec_helper'
+
+describe Geo::SidekiqCronConfigWorker do
+  describe '#perform' do
+    it 'runs the cron manager' do
+      manager = double('cron_manager')
+      allow(Gitlab::Geo::CronManager).to receive(:new) { manager }
+
+      expect(manager).to receive(:execute)
+
+      described_class.new.perform
+    end
+  end
+end