Add migration for backfilling traversal_ids in blobs and wiki blobs

What does this MR do and why?

Describe in detail what your merge request does and why.

Backfills the traversal_ids for blobs and wiki blobs in the main index, more details in the issue #351381 (closed)

Time for completion estimate calculation (internal link): 278 hours (may take a little longer due to having to work through each project). Indexing will not be paused during the migration.

Screenshots or screen recordings

Screenshots are required for UI changes, and strongly recommended for all other merge requests.

How to set up and validate locally

Numbered steps to set up and validate the change are strongly suggested.

1. Run the following query to get the blobs and wiki_blobs with missing traversal_ids

{
  "size": 0,
  "query": {
    "bool": {
      "must_not": {
        "exists": {
          "field": "traversal_ids"
        }
      },
      "must": {
        "terms": {
          "type": [
            "blob",
            "wiki_blob"
          ]
        }
      }
    }
  },
  "aggs": {
    "my-agg-name": {
      "terms": {
        "size": 1000,
        "field": "project_id"
      }
    }
  }
}

2. Make sure advanced search is enabled and you run the migration from rails console by entering the following lines:

require File.expand_path('ee/elastic/migrate/20221221110300_add_traversal_ids_in_blobs_and_wiki_blobs.rb')
BackfillTraversalIdsToBlobsAndWikiBlobs.new(20221221110300).migrate

3. Run the query again in ES to verify that there are no records with misssing traversal_ids

MR acceptance checklist

This checklist encourages us to confirm any changes have been analyzed to reduce risks in quality, performance, reliability, security, and maintainability.

• I have evaluated the MR acceptance checklist for this MR.

Related to #351381 (closed)

ee/elastic/migrate/20221221110300_backfill_traversal_ids_to_blobs_and_wiki_blobs.rb

@@ -3,47 +3,70 @@
 class BackfillTraversalIdsToBlobsAndWikiBlobs < Elastic::Migration
   include Elastic::MigrationHelper
 
-  BATCH_SIZE = 1000
+  batch_size 100_000
   ELASTIC_TIMEOUT = '5m'
+  BLOB_AND_WIKI_BLOB = %w[blob wiki_blob].freeze
   batched!
   throttle_delay 1.minute
   retry_on_failure
 
   def migrate
+    task_id = migration_state[:task_id]
+
+    if task_id
+      task_status = helper.task_status(task_id: task_id)
+
+      if task_status['failures'].present?
+        set_migration_state(task_id: nil)
+        log_raise "Failed to update projects : #{task_status['failures']}"
+      end
+
+      if task_status['completed']
+        log "Updating traversal_ids in original index is completed for task_id: #{task_id}"
+        set_migration_state(task_id: nil)
+      else
+        log "Updating traversal_ids in original index is still in progress for task_id: #{task_id}"
+      end
+
+      return
+    end
+
     if completed?
-      log "Migration Completed There are no projects left to add traversal_ids"
+      log "Migration Completed: There are no projects left to add traversal_ids"
       return
     end
 
     log "Searching for the projects with missing traversal_ids"
-    project_ids = projects_with_missing_traversal_ids_for_blobs_or_wiki_blobs
+    project_ids = projects_with_missing_traversal_ids
     log "Found #{project_ids.size} projects with missing traversal_ids"
-    project_ids.each_with_index do |project_id, index|
-      update_by_query(Project.find(project_id))
+    begin
+      update_by_query(Project.find(project_ids[0])) unless project_ids.empty?
+    rescue ActiveRecord::RecordNotFound
+      log "Project not found: #{project_ids[0]}"
     end
   end
 
   def completed?
-    helper.refresh_index
+    helper.refresh_index(index_name: helper.target_name)
     log "Running the count_items_missing_traversal_ids query"
     total_remaining = count_items_missing_traversal_ids
 
-    log "Checking to see if migration is completed based on index counts remaining:#{total_remaining}"
-
+    log "Checking to see if migration is completed based on index counts remaining: #{total_remaining}"
     total_remaining == 0
   end
 
   private
 
   def update_by_query(project)
-    client.update_by_query(
+    log "Launching update query for project #{project.id}"
+    response = client.update_by_query(
       index: helper.target_name,
       body: {
         query: {
           bool: {
             filter: [
               { term: { project_id: project.id.to_s } },
-              { terms: { type: %w[blob wiki_blob] } }
+              { terms: { type: BLOB_AND_WIKI_BLOB } }
             ]
           }
         },
@@ -52,10 +75,25 @@ def update_by_query(project)
           source: "ctx._source.traversal_ids = '#{project.namespace_ancestry}'"
         }
       },
-      wait_for_completion: true,
+      wait_for_completion: false,
       timeout: ELASTIC_TIMEOUT,
       conflicts: 'proceed'
     )
+
+    if response['failures'].present?
+      set_migration_state(task_id: nil)
+      log_raise "Failed to update project with project_id: #{project.id} - #{response['failures']}"
+    end
+
+    task_id = response['task']
+    log "Adding traversal_ids to original index is started with task_id: #{task_id}"
+
+    set_migration_state(
+      task_id: task_id
+    )
+  rescue StandardError => e
+    set_migration_state(task_id: nil)
+    raise e
   end
 
   def count_items_missing_traversal_ids
@@ -65,14 +103,14 @@ def count_items_missing_traversal_ids
         query: {
           bool: {
             must_not: { exists: { field: "traversal_ids" } },
-            must: { terms: { type: %w[blob wiki_blob] } }
+            must: { terms: { type: BLOB_AND_WIKI_BLOB } }
           }
         }
       }
     )['count']
   end
 
-  def projects_with_missing_traversal_ids_for_blobs_or_wiki_blobs
+  def projects_with_missing_traversal_ids
     results = client.search(
       index: helper.target_name,
       body: {
@@ -80,12 +118,12 @@ def projects_with_missing_traversal_ids_for_blobs_or_wiki_blobs
         query: {
           bool: {
             must_not: { exists: { field: "traversal_ids" } },
-            must: { terms: { type: %w[blob wiki_blob] } }
+            must: { terms: { type: BLOB_AND_WIKI_BLOB } }
           }
         },
         aggs: {
           project_ids: {
-            terms: { size: BATCH_SIZE, field: "project_id" }
+            terms: { size: batch_size, field: "project_id" }
           }
         }
       }