Merge branch 'add-cluster-health-to-preflight' into 'master'

Add cluster healthcheck to MigrationWorker

See merge request https://gitlab.com/gitlab-org/gitlab/-/merge_requests/135238



Merged-by: Terri Chu <tchu@gitlab.com>
Approved-by: Siddharth Dungarwal <sdungarwal@gitlab.com>
Approved-by: Terri Chu <tchu@gitlab.com>
Reviewed-by: Terri Chu <tchu@gitlab.com>
Co-authored-by: Dmitry Gruzd <dgruzd@gitlab.com>

ee/app/workers/elastic/migration_worker.rb

@@ -31,12 +31,12 @@ def perform
         migration = Elastic::MigrationRecord.current_migration
 
         unless migration
-          logger.info(structured_payload(message: 'MigrationWorker: no migration available'))
+          log message: 'MigrationWorker: no migration available'
           break false
         end
 
         if migration.halted?
-          logger.error(structured_payload(message: "MigrationWorker: migration[#{migration.name}] has been halted. All future migrations will be halted because of that. Exiting"))
+          error message: "MigrationWorker: migration[#{migration.name}] has been halted. All future migrations will be halted because of that. Exiting"
           unpause_indexing!(migration)
 
           break false
@@ -45,11 +45,11 @@ def perform
         if !migration.started? && migration.space_requirements?
           free_size_bytes = helper.cluster_free_size_bytes
           space_required_bytes = migration.space_required_bytes
-          logger.info(structured_payload(message: "MigrationWorker: migration[#{migration.name}] checking free space in cluster. Required space #{number_to_human_size(space_required_bytes)}. Free space #{number_to_human_size(free_size_bytes)}."))
+          log message: "MigrationWorker: migration[#{migration.name}] checking free space in cluster. Required space #{number_to_human_size(space_required_bytes)}. Free space #{number_to_human_size(free_size_bytes)}."
 
           if free_size_bytes < space_required_bytes
-            logger.warn(structured_payload(message: "MigrationWorker: migration[#{migration.name}] You should have at least #{number_to_human_size(space_required_bytes)} of free space in the cluster to run this migration. Please increase the storage in your Elasticsearch cluster."))
-            logger.info(structured_payload(message: "MigrationWorker: migration[#{migration.name}] updating with halted: true"))
+            warn message: "MigrationWorker: migration[#{migration.name}] You should have at least #{number_to_human_size(space_required_bytes)} of free space in the cluster to run this migration. Please increase the storage in your Elasticsearch cluster."
+            log message: "MigrationWorker: migration[#{migration.name}] updating with halted: true"
             migration.halt
 
             break false
@@ -67,11 +67,23 @@ def perform
         enqueue_next_batch(migration)
       end
     rescue StandardError => e
-      logger.error(structured_payload(message: "#{self.class.name}: #{e.class} #{e.message}", backtrace: e.backtrace.join("\n")))
+      error message: "#{self.class.name}: #{e.class} #{e.message}", backtrace: e.backtrace.join("\n")
     end
 
     private
 
+    def log(...)
+      logger.info(structured_payload(...))
+    end
+
+    def warn(...)
+      logger.warn(structured_payload(...))
+    end
+
+    def error(...)
+      logger.error(structured_payload(...))
+    end
+
     def preflight_check_successful?
       return false if Feature.disabled?(:elastic_migration_worker, type: :ops)
       return false unless Gitlab::CurrentSettings.elasticsearch_indexing?
@@ -79,25 +91,30 @@ def preflight_check_successful?
       return false if Elastic::ReindexingTask.current
 
       if helper.unsupported_version?
-        logger.info(structured_payload(message: 'MigrationWorker: You are using an unsupported version of Elasticsearch. Indexing will be paused to prevent data loss'))
+        log message: 'MigrationWorker: You are using an unsupported version of Elasticsearch. Indexing will be paused to prevent data loss'
         Gitlab::CurrentSettings.update!(elasticsearch_pause_indexing: true)
 
         return false
       end
 
+      unless Search::ClusterHealthCheck::Elastic.healthy?
+        error message: "Advanced search cluster is unhealthy. Execution is skipped."
+        return false
+      end
+
       true
     end
 
     def execute_migration(migration)
       if migration.started? && !migration.batched? && !migration.retry_on_failure?
-        logger.info(structured_payload(message: "MigrationWorker: migration[#{migration.name}] did not execute migrate method since it was already executed. Waiting for migration to complete"))
+        log message: "MigrationWorker: migration[#{migration.name}] did not execute migrate method since it was already executed. Waiting for migration to complete"
 
         return
       end
 
       pause_indexing!(migration)
 
-      logger.info(structured_payload(message: "MigrationWorker: migration[#{migration.name}] executing migrate method"))
+      log message: "MigrationWorker: migration[#{migration.name}] executing migrate method"
       migration.migrate
     rescue StandardError => e
       retry_migration(migration, e) if migration.retry_on_failure?
@@ -108,11 +125,11 @@ def execute_migration(migration)
     def retry_migration(migration, exception)
       if migration.current_attempt >= migration.max_attempts
         message = "MigrationWorker: migration has failed with #{exception.class}:#{exception.message}, no retries left"
-        logger.error(structured_payload(message: message, backtrace: exception.backtrace.join("\n")))
+        error message: message, backtrace: exception.backtrace.join("\n")
 
         migration.fail(message: message)
       else
-        logger.info(structured_payload(message: "MigrationWorker: increasing previous_attempts to #{migration.current_attempt}"))
+        log message: "MigrationWorker: increasing previous_attempts to #{migration.current_attempt}"
         migration.save_state!(previous_attempts: migration.current_attempt)
       end
     end
@@ -120,13 +137,13 @@ def retry_migration(migration, exception)
     def enqueue_next_batch(migration)
       return unless migration.batched? && !migration.completed?
 
-      logger.info(structured_payload(message: "MigrationWorker: migration[#{migration.name}] kicking off next migration batch"))
+      log message: "MigrationWorker: migration[#{migration.name}] kicking off next migration batch"
       Elastic::MigrationWorker.perform_in(migration.throttle_delay)
     end
 
     def check_and_save_completed_status(migration)
       migration.completed?.tap do |status|
-        logger.info(structured_payload(message: "MigrationWorker: migration[#{migration.name}] updating with completed: #{status}"))
+        log message: "MigrationWorker: migration[#{migration.name}] updating with completed: #{status}"
         migration.save!(completed: status)
       end
     end
@@ -139,7 +156,7 @@ def pause_indexing!(migration)
       migration.save_state!(pause_indexing: pause_indexing)
 
       if pause_indexing
-        logger.info(structured_payload(message: 'MigrationWorker: Pausing indexing'))
+        log message: 'MigrationWorker: Pausing indexing'
         Gitlab::CurrentSettings.update!(elasticsearch_pause_indexing: true)
       end
     end
@@ -149,7 +166,7 @@ def unpause_indexing!(migration)
       return unless migration.load_state[:pause_indexing]
       return if migration.load_state[:halted_indexing_unpaused]
 
-      logger.info(structured_payload(message: 'MigrationWorker: unpausing indexing'))
+      log message: 'MigrationWorker: unpausing indexing'
       Gitlab::CurrentSettings.update!(elasticsearch_pause_indexing: false)
 
       migration.save_state!(halted_indexing_unpaused: true) if migration.halted?
@@ -158,7 +175,7 @@ def unpause_indexing!(migration)
     def create_migrations_index_if_not_exists
       return if helper.migrations_index_exists?
 
-      logger.info(structured_payload(message: 'MigrationWorker: creating migrations index'))
+      log message: 'MigrationWorker: creating migrations index'
       helper.create_migrations_index
     end
 

ee/spec/workers/elastic/migration_worker_spec.rb

@@ -47,6 +47,20 @@
       end
     end
 
+    context 'cluster is unhealthy' do
+      before do
+        stub_ee_application_setting(elasticsearch_indexing: true)
+        allow(Search::ClusterHealthCheck::Elastic).to receive(:healthy?).and_return(false)
+        allow(::Gitlab::Elasticsearch::Logger).to receive(:build).and_return(logger)
+      end
+
+      it 'raises an error and does not execute migration' do
+        expect(subject).not_to receive(:execute_migration)
+        expect(logger).to receive(:error)
+        expect(subject.perform).to be_falsey
+      end
+    end
+
     context 'reindexing task is in progress' do
       let!(:task) { create(:elastic_reindexing_task) }