Migrate Sidekiq jobs based on routing rules

What does this MR do and why?

As described in gitlab-com/gl-infra/scalability#1930, we are defaulting self-managed installations' Sidekiq (omnibus-gitlab!6289 (closed), gitlab-org/charts/gitlab!2789 (closed)) to listen to default and mailers queue only. To prevent jobs from being left out after user upgrades to the new version, we're leveraging post-deployment migration to automatically migrate jobs based on the queues that are outside of the current routing rules defined in gitlab.yml.

This is what a job looks like in redis:

> lrange resque:gitlab:queue:email_receiver 0 -1
1) "{\"retry\":3,\"queue\":\"email_receiver\",\"backtrace\":true,\"version\":0,\"args\":[],\"class\":\"EmailReceiverWorker\",\"jid\":\"a2cf84850e1506f5e5acb47d\",\"created_at\":1664962337.9325042,\"meta.feature_category\":\"team_planning\",\"correlation_id\":\"7bb10dedd488b4b8c8a6eadea2ad6df9\",\"worker_data_consistency\":\"always\",\"idempotency_key\":\"resque:gitlab:duplicate:email_receiver:e24f8243f3bdc71e54ba932255525eb20e3c5416e63890942eb16b1b38ed3f2f\",\"duplicate-of\":\"460f87d4dcdea1758213403f\",\"size_limiter\":\"validated\",\"enqueued_at\":1664962337.934328}"

This MR has to be included with below Related MRs in the same release at the very least.

Related MRs

These MRs are changing Sidekiq to listen to default and mailers queue by default, if no custom routing rules are defined. Otherwise, the queues are generated from the routing rules.

• omnibus-gitlab!6289 (closed)
• gitlab-org/charts/gitlab!2789 (closed)

Below MR has the same copy-pasted code as a Rake task:

• !101348 (merged)

How to set up and validate locally

1. Start GitLab.
2. Change gitlab.yml sidekiq.routing_rules to simulate queue per-worker configuration (default routing rules prior to !97908 (merged))

    routing_rules:
      - ["*", null]

3. Stop Sidekiq (so no jobs get processed).

4. Check queues before migration

redis /Users/gregoriusmarco/Documents/workspace/gdk-10-22/redis/redis.socket[1]> keys resque:gitlab:queue:*
(empty array)

5. Schedule some jobs either from the console or the UI.

In rails console:
[2] pry(main)> EmailReceiverWorker.perform_async
=> "3b153d570496aa1f82a4af9f"
[3] pry(main)> ExportCsvWorker.perform_async
=> "e563fcdaa5bc13fd35b204a5"

6. Check the Sidekiq queues.

redis /Users/gregoriusmarco/Documents/workspace/gdk-10-22/redis/redis.socket[1]> keys resque:gitlab:queue:*
1) "resque:gitlab:queue:email_receiver"
2) "resque:gitlab:queue:export_csv"
redis /Users/gregoriusmarco/Documents/workspace/gdk-10-22/redis/redis.socket[1]> lrange resque:gitlab:queue:email_receiver 0 -1
1) "{\"retry\":3,\"queue\":\"email_receiver\",\"backtrace\":true,\"version\":0,\"args\":[],\"class\":\"EmailReceiverWorker\",\"jid\":\"3b153d570496aa1f82a4af9f\",\"created_at\":1667224064.14315,\"meta.feature_category\":\"team_planning\",\"correlation_id\":\"aeeaf1d8714cfbfe6b58ef74f3943e42\",\"worker_data_consistency\":\"always\",\"idempotency_key\":\"resque:gitlab:duplicate:email_receiver:e24f8243f3bdc71e54ba932255525eb20e3c5416e63890942eb16b1b38ed3f2f\",\"duplicate-of\":\"2faba86933661b0c02bf33c0\",\"size_limiter\":\"validated\",\"enqueued_at\":1667224064.151187}"
redis /Users/gregoriusmarco/Documents/workspace/gdk-10-22/redis/redis.socket[1]> lrange resque:gitlab:queue:export_csv 0 -1
1) "{\"retry\":3,\"queue\":\"export_csv\",\"backtrace\":true,\"version\":0,\"args\":[],\"class\":\"ExportCsvWorker\",\"jid\":\"e563fcdaa5bc13fd35b204a5\",\"created_at\":1667224069.729505,\"meta.feature_category\":\"team_planning\",\"correlation_id\":\"97075e24957f895557707a225d7559ab\",\"worker_data_consistency\":\"always\",\"idempotency_key\":\"resque:gitlab:duplicate:export_csv:5a7377c023323988e6c86830ca9cc09d9276e04b7efb5dab402c84f319474907\",\"duplicate-of\":\"e8cf11cc821bfae428bd812e\",\"size_limiter\":\"validated\",\"enqueued_at\":1667224069.732599}"

7. Change the routing rules configuration. Set gitlab.yml sidekiq.routing_rules to null (Routing rules will be defaulted to [['*', 'default']] here).
8. Run the migration.

9. Check the Sidekiq queues again: they should match the new routing rules.

redis /Users/gregoriusmarco/Documents/workspace/gdk-10-22/redis/redis.socket[1]> keys resque:gitlab:queue:*
1) "resque:gitlab:queue:default"
redis /Users/gregoriusmarco/Documents/workspace/gdk-10-22/redis/redis.socket[1]> lrange resque:gitlab:queue:default 0 -1
1) "{\"retry\":3,\"queue\":\"default\",\"backtrace\":true,\"version\":0,\"args\":[],\"class\":\"ExportCsvWorker\",\"jid\":\"e563fcdaa5bc13fd35b204a5\",\"created_at\":1667224069.729505,\"meta.feature_category\":\"team_planning\",\"correlation_id\":\"97075e24957f895557707a225d7559ab\",\"worker_data_consistency\":\"always\",\"idempotency_key\":\"resque:gitlab:duplicate:export_csv:5a7377c023323988e6c86830ca9cc09d9276e04b7efb5dab402c84f319474907\",\"duplicate-of\":\"e8cf11cc821bfae428bd812e\",\"size_limiter\":\"validated\",\"enqueued_at\":1667224069.732599}"
2) "{\"retry\":3,\"queue\":\"default\",\"backtrace\":true,\"version\":0,\"args\":[],\"class\":\"EmailReceiverWorker\",\"jid\":\"3b153d570496aa1f82a4af9f\",\"created_at\":1667224064.14315,\"meta.feature_category\":\"team_planning\",\"correlation_id\":\"aeeaf1d8714cfbfe6b58ef74f3943e42\",\"worker_data_consistency\":\"always\",\"idempotency_key\":\"resque:gitlab:duplicate:email_receiver:e24f8243f3bdc71e54ba932255525eb20e3c5416e63890942eb16b1b38ed3f2f\",\"duplicate-of\":\"2faba86933661b0c02bf33c0\",\"size_limiter\":\"validated\",\"enqueued_at\":1667224064.151187}"

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.

db/post_migrate/20220928093644_migrate_sidekiq_jobs.rb

 # frozen_string_literal: true
 
 class MigrateSidekiqJobs < Gitlab::Database::Migration[2.0]
+  class SidekiqMigrateJobs
+    LOG_FREQUENCY_QUEUES = 10
+
+    attr_reader :logger, :mappings
+
+    # mappings is a hash of WorkerClassName => target_queue_name
+    def initialize(mappings, logger: nil)
+      @mappings = mappings
+      @logger = logger
+    end
+
+    # Migrates jobs from queues that are outside the mappings
+    # rubocop: disable Cop/SidekiqRedisCall
+    def migrate_queues
+      routing_rules_queues = mappings.values.uniq
+      logger&.info("List of queues based on routing rules: #{routing_rules_queues}")
+      Sidekiq.redis do |conn|
+        conn.scan_each(match: "queue:*", type: 'list') do |key|
+          queue_from = key.split(':', 2).last
+          next if routing_rules_queues.include?(queue_from)
+
+          logger&.info("Migrating #{queue_from} queue")
+
+          migrated = 0
+          while queue_length(queue_from) > 0
+            begin
+              if migrated >= 0 && migrated % LOG_FREQUENCY_QUEUES == 0
+                logger&.info("Migrating from #{queue_from}. Total: #{queue_length(queue_from)}. Migrated: #{migrated}.")
+              end
+
+              job = conn.rpop "queue:#{queue_from}"
+              job_hash = Sidekiq.load_json job
+              next unless mappings.has_key?(job_hash['class'])
+
+              destination_queue = mappings[job_hash['class']]
+              job_hash['queue'] = destination_queue
+              conn.lpush("queue:#{destination_queue}", Sidekiq.dump_json(job_hash))
+              migrated += 1
+            rescue JSON::ParserError
+              logger&.error("Unmarshal JSON payload from SidekiqMigrateJobs failed. Job: #{job}")
+              next
+            end
+          end
+          logger&.info("Finished migrating #{queue_from} queue")
+        end
+      end
+    end
+
+    private
+
+    def queue_length(queue_name)
+      Sidekiq.redis do |conn|
+        conn.llen("queue:#{queue_name}")
+      end
+    end
+    # rubocop: enable Cop/SidekiqRedisCall
+  end
+
   def up
     mappings = Gitlab::SidekiqConfig.worker_queue_mappings
     logger = ::Gitlab::BackgroundMigration::Logger.build