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]
+  # Migrate jobs that don't belong to current routing rules
+  # https://gitlab.com/gitlab-com/gl-infra/scalability/-/issues/1930
+
+  def up
+    queues = scan_queues
+    migrate_jobs(queues)
+  end
+
+  def down
+    # no-op
+  end
+
+  private
+
+  def scan_queues
+    queues = []
+    routing_rules_queues = Settings.sidekiq.routing_rules.map { |query, rule| rule }
+    Sidekiq.redis do |conn| # rubocop:disable Cop/SidekiqRedisCall
+      conn.scan_each(match: "queue:*", type: 'list') do |key|
+        queue = key.split(':').last
+        queues << queue unless routing_rules_queues.include? queue
+      end
+    end
+    queues.uniq
+  end
+
+  def migrate_jobs(queues_from)
+    queues_from.each do |queue_from|
+      while sidekiq_queue_length(queue_from) > 0
+        Sidekiq.redis do |conn| # rubocop:disable Cop/SidekiqRedisCall
+          payload = conn.rpop "queue:#{queue_from}"
+          parsed = Gitlab::Json.parse payload
+          next if parsed['class'].nil?
+
+          worker_class = Object.const_get(parsed['class'], false)
+          conn.lpush("queue:#{worker_class.queue}", payload) unless worker_class.queue == queue_from
+        rescue JSON::ParserError
+          next
+        end
+      end
+    end
+  end
+end