Create Geo::PruneEventLogWorker

This worker runs regularly to check all secondary nodes what there
last handled Geo Log event was. All the Geo Log events that are
handled by all nodes are pruned from the database. If the primary
failed to get the status of one of the nodes, nothing is done.

Closes gitlab-org/gitlab-ee#3577.

app/workers/geo/prune_event_log_worker.rb

+module Geo
+  class PruneEventLogWorker
+    include Sidekiq::Worker
+    include CronjobQueue
+    include ExclusiveLeaseGuard
+
+    LEASE_TIMEOUT = 60.minutes
+
+    def lease_timeout
+      LEASE_TIMEOUT
+    end
+
+    def log_error(message, extra_args = {})
+      args = { class: self.class.name, message: message }.merge(extra_args)
+      Gitlab::Geo::Logger.error(args)
+    end
+
+    def perform
+      return unless Gitlab::Geo.primary?
+
+      try_obtain_lease do
+        if Gitlab::Geo.secondary_nodes.empty?
+          Geo::EventLog.delete_all
+          return
+        end
+
+        cursor_last_event_ids = Gitlab::Geo.secondary_nodes.map do |node|
+          node_status_service.call(node).cursor_last_event_id
+        end
+
+        # Abort when any of the nodes could not be contacted
+        return if cursor_last_event_ids.include?(nil)
+
+        Geo::EventLog.delete_all(['id < ?', cursor_last_event_ids.min])
+      end
+    end
+
+    private
+
+    def node_status_service
+      @node_status_service ||= Geo::NodeStatusService.new
+    end
+  end
+end

changelogs/unreleased-ee/tc-geo-prune-event-log.yml

+---
+title: Add worker to prune the Geo Event Log
+merge_request: 3172
+author:
+type: added

config/gitlab.yml.example

@@ -246,6 +246,11 @@ production: &base
     geo_metrics_update_worker:
       cron: "*/1 * * * *"
 
+    # GitLab Geo prune event log worker
+    # NOTE: This will only take effect if Geo is enabled (primary node only)
+    geo_prune_event_log_worker:
+      cron: "0 */6 * * *"
+
     # GitLab Geo repository sync worker
     # NOTE: This will only take effect if Geo is enabled (secondary nodes only)
     geo_repository_sync_worker:

config/initializers/1_settings.rb

@@ -452,6 +452,9 @@ Settings.cron_jobs['geo_repository_sync_worker']['job_class'] ||= 'Geo::Reposito
 Settings.cron_jobs['geo_file_download_dispatch_worker'] ||= Settingslogic.new({})
 Settings.cron_jobs['geo_file_download_dispatch_worker']['cron'] ||= '*/1 * * * *'
 Settings.cron_jobs['geo_file_download_dispatch_worker']['job_class'] ||= 'Geo::FileDownloadDispatchWorker'
+Settings.cron_jobs['geo_prune_event_log_worker'] ||= Settingslogic.new({})
+Settings.cron_jobs['geo_prune_event_log_worker']['cron'] ||= '0 */6 * * *'
+Settings.cron_jobs['geo_prune_event_log_worker']['job_class'] ||= 'Geo::PruneEventLogWorker'
 Settings.cron_jobs['import_export_project_cleanup_worker'] ||= Settingslogic.new({})
 Settings.cron_jobs['import_export_project_cleanup_worker']['cron'] ||= '0 * * * *'
 Settings.cron_jobs['import_export_project_cleanup_worker']['job_class'] = 'ImportExportProjectCleanupWorker'

spec/controllers/admin/geo_nodes_controller_spec.rb

@@ -259,25 +259,7 @@ describe Admin::GeoNodesController, :postgresql do
     end
 
     context 'with add-on license' do
-      let(:geo_node_status) do
-        GeoNodeStatus.new(
-          id: 1,
-          health: nil,
-          attachments_count: 329,
-          attachments_failed_count: 13,
-          attachments_synced_count: 141,
-          lfs_objects_count: 256,
-          lfs_objects_failed_count: 12,
-          lfs_objects_synced_count: 123,
-          repositories_count: 10,
-          repositories_synced_count: 5,
-          repositories_failed_count: 0,
-          last_event_id: 2,
-          last_event_timestamp: Time.now.to_i,
-          cursor_last_event_id: 1,
-          cursor_last_event_timestamp: Time.now.to_i
-        )
-      end
+      let(:geo_node_status) { build(:geo_node_status, :healthy) }
 
       before do
         allow(Gitlab::Geo).to receive(:license_allows?).and_return(true)

spec/factories/geo_node_statuses.rb

+FactoryGirl.define do
+  factory :geo_node_status do
+    skip_create
+
+    sequence(:id)
+
+    trait :healthy do
+      health nil
+      attachments_count 329
+      attachments_failed_count 13
+      attachments_synced_count 141
+      lfs_objects_count 256
+      lfs_objects_failed_count 12
+      lfs_objects_synced_count 123
+      repositories_count 10
+      repositories_synced_count 5
+      repositories_failed_count 0
+      last_event_id 2
+      last_event_timestamp Time.now.to_i
+      cursor_last_event_id 1
+      cursor_last_event_timestamp Time.now.to_i
+    end
+
+    trait :unhealthy do
+      health "Could not connect to Geo node - HTTP Status Code: 401 Unauthorized\nTest"
+    end
+  end
+end

spec/workers/geo/prune_event_log_worker_spec.rb

+require 'spec_helper'
+
+describe Geo::PruneEventLogWorker, :geo do
+  include ::EE::GeoHelpers
+
+  subject(:worker) { described_class.new }
+  set(:primary) { create(:geo_node, :primary, host: 'primary-geo-node') }
+  set(:secondary) { create(:geo_node) }
+
+  before do
+    allow_any_instance_of(Gitlab::ExclusiveLease).to receive(:try_obtain).and_return(true)
+  end
+
+  describe '#perform' do
+    context 'current node secondary' do
+      before do
+        stub_current_geo_node(secondary)
+      end
+
+      it 'does nothing' do
+        expect(worker).not_to receive(:try_obtain_lease)
+
+        worker.perform
+      end
+    end
+
+    context 'current node primary' do
+      before do
+        stub_current_geo_node(primary)
+      end
+
+      it 'logs error when it cannot obtain lease' do
+        allow_any_instance_of(Gitlab::ExclusiveLease).to receive(:try_obtain) { nil }
+
+        expect(worker).to receive(:log_error).with('Cannot obtain an exclusive lease. There must be another worker already in execution.')
+
+        worker.perform
+      end
+
+      context 'no secondary nodes' do
+        before do
+          secondary.destroy
+        end
+
+        it 'deletes everything from the Geo event log' do
+          expect(Geo::EventLog).to receive(:delete_all)
+
+          worker.perform
+        end
+      end
+
+      context 'multiple secondary nodes' do
+        set(:secondary2) { create(:geo_node) }
+        let(:healthy_status) { build(:geo_node_status, :healthy) }
+        let(:unhealthy_status) { build(:geo_node_status, :unhealthy) }
+
+        it 'contacts all secondary nodes for their status' do
+          expect_any_instance_of(Geo::NodeStatusService).to receive(:call).twice { healthy_status }
+          expect(Geo::EventLog).to receive(:delete_all)
+
+          worker.perform
+        end
+
+        it 'aborts when there are unhealthy nodes' do
+          expect_any_instance_of(Geo::NodeStatusService).to receive(:call) { healthy_status }
+          expect_any_instance_of(Geo::NodeStatusService).to receive(:call) { unhealthy_status }
+          expect(Geo::EventLog).not_to receive(:delete_all)
+
+          worker.perform
+        end
+
+        it 'takes the integer-minimum value of all nodes' do
+          allow_any_instance_of(Geo::NodeStatusService).to receive(:call) { build(:geo_node_status, :healthy, cursor_last_event_id: 3) }
+          allow_any_instance_of(Geo::NodeStatusService).to receive(:call) { build(:geo_node_status, :healthy, cursor_last_event_id: 10) }
+          expect(Geo::EventLog).not_to receive(:delete_all).with(['id < ?', 3])
+
+          worker.perform
+        end
+      end
+    end
+  end
+
+  describe '#log_error' do
+    it 'calls the Geo logger' do
+      expect(Gitlab::Geo::Logger).to receive(:error)
+
+      worker.log_error('Something is wrong')
+    end
+  end
+end