Use a dedicated queue for each worker

app/workers/all_queues.yml

+---
+- cronjob:admin_email
+- cronjob:expire_build_artifacts
+- cronjob:gitlab_usage_ping
+- cronjob:import_export_project_cleanup
+- cronjob:pipeline_schedule
+- cronjob:prune_old_events
+- cronjob:remove_expired_group_links
+- cronjob:remove_expired_members
+- cronjob:remove_old_web_hook_logs
+- cronjob:remove_unreferenced_lfs_objects
+- cronjob:repository_archive_cache
+- cronjob:repository_check_batch
+- cronjob:requests_profiles
+- cronjob:schedule_update_user_activity
+- cronjob:stuck_ci_jobs
+- cronjob:stuck_import_jobs
+- cronjob:stuck_merge_jobs
+- cronjob:trending_projects
+
+- gcp_cluster:cluster_install_app
+- gcp_cluster:cluster_provision
+- gcp_cluster:cluster_wait_for_app_installation
+- gcp_cluster:wait_for_cluster_creation
+
+- github_import_advance_stage
+- github_importer:github_import_import_diff_note
+- github_importer:github_import_import_issue
+- github_importer:github_import_import_note
+- github_importer:github_import_import_pull_request
+- github_importer:github_import_refresh_import_jid
+- github_importer:github_import_stage_finish_import
+- github_importer:github_import_stage_import_base_data
+- github_importer:github_import_stage_import_issues_and_diff_notes
+- github_importer:github_import_stage_import_notes
+- github_importer:github_import_stage_import_pull_requests
+- github_importer:github_import_stage_import_repository
+
+- pipeline_cache:expire_job_cache
+- pipeline_cache:expire_pipeline_cache
+- pipeline_creation:create_pipeline
+- pipeline_default:build_coverage
+- pipeline_default:build_trace_sections
+- pipeline_default:pipeline_metrics
+- pipeline_default:pipeline_notification
+- pipeline_default:update_head_pipeline_for_merge_request
+- pipeline_hooks:build_hooks
+- pipeline_hooks:pipeline_hooks
+- pipeline_processing:build_finished
+- pipeline_processing:build_queue
+- pipeline_processing:build_success
+- pipeline_processing:pipeline_process
+- pipeline_processing:pipeline_success
+- pipeline_processing:pipeline_update
+- pipeline_processing:stage_update
+
+- repository_check:repository_check_clear
+- repository_check:repository_check_single_repository
+
+- default
+- mailers # ActionMailer::DeliveryJob.queue_name
+
+- authorized_projects
+- background_migration
+- create_gpg_signature
+- delete_merged_branches
+- delete_user
+- email_receiver
+- emails_on_push
+- expire_build_instance_artifacts
+- git_garbage_collect
+- gitlab_shell
+- group_destroy
+- invalid_gpg_signature_update
+- irker
+- merge
+- namespaceless_project_destroy
+- new_issue
+- new_merge_request
+- new_note
+- pages
+- post_receive
+- process_commit
+- project_cache
+- project_destroy
+- project_export
+- project_migrate_hashed_storage
+- project_service
+- propagate_service_template
+- reactive_caching
+- repository_fork
+- repository_import
+- storage_migrator
+- system_hook_push
+- update_merge_requests
+- update_user_activity
+- upload_checksum
+- web_hook
+
+# EE-specific queues
+- cronjob:clear_shared_runners_minutes
+- cronjob:geo_file_download_dispatch
+- cronjob:geo_metrics_update
+- cronjob:geo_prune_event_log
+- cronjob:geo_repository_sync
+- cronjob:historical_data
+- cronjob:ldap_all_groups_sync
+- cronjob:ldap_sync
+- cronjob:update_all_mirrors
+
+- geo:geo_file_removal
+- geo:geo_hashed_storage_attachments_migration
+- geo:geo_hashed_storage_migration
+- geo:geo_rename_repository
+- geo:geo_repositories_clean_up
+- geo:geo_repository_destroy
+
+- admin_emails
+- elastic_batch_project_indexer
+- elastic_commit_indexer
+- elastic_indexer
+- export_csv
+- geo_base_scheduler
+- geo_file_download
+- geo_project_sync
+- geo_repository_shard_sync
+- ldap_group_sync
+- object_storage_upload
+- project_update_repository_storage
+- rebase
+- repository_update_mirror
+- repository_update_remote_mirror

app/workers/build_finished_worker.rb

@@ -2,7 +2,7 @@ class BuildFinishedWorker
   include ApplicationWorker
   include PipelineQueue
-  enqueue_in group: :processing
+  queue_namespace :pipeline_processing
   def perform(build_id)
     Ci::Build.find_by(id: build_id).try do |build|

app/workers/build_hooks_worker.rb

@@ -2,7 +2,7 @@ class BuildHooksWorker
   include ApplicationWorker
   include PipelineQueue
-  enqueue_in group: :hooks
+  queue_namespace :pipeline_hooks
   def perform(build_id)
     Ci::Build.find_by(id: build_id)

app/workers/build_queue_worker.rb

@@ -2,7 +2,7 @@ class BuildQueueWorker
   include ApplicationWorker
   include PipelineQueue
-  enqueue_in group: :processing
+  queue_namespace :pipeline_processing
   def perform(build_id)
     Ci::Build.find_by(id: build_id).try do |build|

app/workers/build_success_worker.rb

@@ -2,7 +2,7 @@ class BuildSuccessWorker
   include ApplicationWorker
   include PipelineQueue
-  enqueue_in group: :processing
+  queue_namespace :pipeline_processing
   def perform(build_id)
     Ci::Build.find_by(id: build_id).try do |build|

app/workers/concerns/application_worker.rb

@@ -6,10 +6,20 @@ module ApplicationWorker
   include Sidekiq::Worker
   included do
-    sidekiq_options queue: base_queue_name
+    set_queue
   end
   module ClassMethods
+    def inherited(subclass)
+      subclass.set_queue
+    end
+
+    def set_queue
+      queue_name = [queue_namespace, base_queue_name].compact.join(':')
+
+      sidekiq_options queue: queue_name
+    end
+
     def base_queue_name
       name
         .sub(/\AGitlab::/, '')
@@ -18,6 +28,16 @@ def base_queue_name
         .tr('/', '_')
     end
+    def queue_namespace(new_namespace = nil)
+      if new_namespace
+        sidekiq_options queue_namespace: new_namespace
+
+        set_queue
+      else
+        get_sidekiq_options['queue_namespace']&.to_s
+      end
+    end
+
     def queue
       get_sidekiq_options['queue'].to_s
     end

app/workers/concerns/cluster_queue.rb

@@ -5,6 +5,6 @@ module ClusterQueue
   extend ActiveSupport::Concern
   included do
-    sidekiq_options queue: :gcp_cluster
+    queue_namespace :gcp_cluster
   end
 end

app/workers/concerns/cronjob_queue.rb

@@ -4,6 +4,7 @@ module CronjobQueue
   extend ActiveSupport::Concern
   included do
-    sidekiq_options queue: :cronjob, retry: false
+    queue_namespace :cronjob
+    sidekiq_options retry: false
   end
 end

app/workers/concerns/geo_queue.rb

@@ -3,6 +3,6 @@ module GeoQueue
   extend ActiveSupport::Concern
   included do
-    sidekiq_options queue: :geo
+    queue_namespace :geo
   end
 end

app/workers/concerns/gitlab/github_import/queue.rb

@@ -4,12 +4,14 @@ module Queue
       extend ActiveSupport::Concern
       included do
+        queue_namespace :github_importer
+
         # If a job produces an error it may block a stage from advancing
         # forever. To prevent this from happening we prevent jobs from going to
         # the dead queue. This does mean some resources may not be imported, but
         # this is better than a project being stuck in the "import" state
         # forever.
-        sidekiq_options queue: 'github_importer', dead: false, retry: 5
+        sidekiq_options dead: false, retry: 5
       end
     end
   end

app/workers/concerns/pipeline_queue.rb

@@ -5,14 +5,6 @@ module PipelineQueue
   extend ActiveSupport::Concern
   included do
-    sidekiq_options queue: 'pipeline_default'
-  end
-
-  class_methods do
-    def enqueue_in(group:)
-      raise ArgumentError, 'Unspecified queue group!' if group.empty?
-
-      sidekiq_options queue: "pipeline_#{group}"
-    end
+    queue_namespace :pipeline_default
   end
 end

app/workers/concerns/repository_check_queue.rb

@@ -3,6 +3,8 @@ module RepositoryCheckQueue
   extend ActiveSupport::Concern
   included do
-    sidekiq_options queue: :repository_check, retry: false
+    queue_namespace :repository_check
+
+    sidekiq_options retry: false
   end
 end

app/workers/create_pipeline_worker.rb

@@ -2,7 +2,7 @@ class CreatePipelineWorker
   include ApplicationWorker
   include PipelineQueue
-  enqueue_in group: :creation
+  queue_namespace :pipeline_creation
   def perform(project_id, user_id, ref, source, params = {})
     project = Project.find(project_id)

app/workers/expire_job_cache_worker.rb

@@ -2,7 +2,7 @@ class ExpireJobCacheWorker
   include ApplicationWorker
   include PipelineQueue
-  enqueue_in group: :cache
+  queue_namespace :pipeline_cache
   def perform(job_id)
     job = CommitStatus.joins(:pipeline, :project).find_by(id: job_id)

app/workers/expire_pipeline_cache_worker.rb

@@ -2,7 +2,7 @@ class ExpirePipelineCacheWorker
   include ApplicationWorker
   include PipelineQueue
-  enqueue_in group: :cache
+  queue_namespace :pipeline_cache
   def perform(pipeline_id)
     pipeline = Ci::Pipeline.find_by(id: pipeline_id)

app/workers/geo/base_scheduler_worker.rb

 module Geo
   class BaseSchedulerWorker
     include ApplicationWorker
-    include CronjobQueue
     include ExclusiveLeaseGuard
     DB_RETRIEVE_BATCH_SIZE = 1000

app/workers/geo/file_download_dispatch_worker.rb

 module Geo
   class FileDownloadDispatchWorker < Geo::BaseSchedulerWorker
+    include CronjobQueue
+
     private
     def max_capacity

app/workers/geo/repository_shard_sync_worker.rb

 module Geo
   class RepositoryShardSyncWorker < Geo::BaseSchedulerWorker
-    # We may have many long-running threads, so split them out
-    # into their own queue to make it possible for other jobs to run.
-    sidekiq_options queue: :geo_repository_shard_sync, retry: false
+    sidekiq_options retry: false
     attr_accessor :shard_name

app/workers/gitlab/github_import/advance_stage_worker.rb

@@ -9,7 +9,7 @@ module GithubImport
     class AdvanceStageWorker
       include ApplicationWorker
-      sidekiq_options queue: 'github_importer_advance_stage', dead: false
+      sidekiq_options dead: false
       INTERVAL = 30.seconds.to_i

app/workers/pages_worker.rb

 class PagesWorker
   include ApplicationWorker
-  sidekiq_options queue: :pages, retry: false
+  sidekiq_options retry: false
   def perform(action, *arg)
     send(action, *arg) # rubocop:disable GitlabSecurity/PublicSend