Add latest changes from gitlab-org/gitlab@13-12-stable-eev13.12.0-rc42

4 files changed, 122 insertions, 11 deletions

diff --git a/lib/gitlab/sidekiq_config/dummy_worker.rb b/lib/gitlab/sidekiq_config/dummy_worker.rb
index 7568840410b..ef0dce0cf84 100644
--- a/lib/gitlab/sidekiq_config/dummy_worker.rb
+++ b/lib/gitlab/sidekiq_config/dummy_worker.rb
@@ -4,9 +4,9 @@ module Gitlab
   module SidekiqConfig
     # For queues that don't have explicit workers - default and mailers
     class DummyWorker
-      attr_accessor :queue
-
       ATTRIBUTE_METHODS = {
+        queue: :queue,
+        name: :name,
         feature_category: :get_feature_category,
         has_external_dependencies: :worker_has_external_dependencies?,
         urgency: :get_urgency,
@@ -16,8 +16,7 @@ module Gitlab
         tags: :get_tags
       }.freeze
 
-      def initialize(queue, attributes = {})
-        @queue = queue
+      def initialize(attributes = {})
         @attributes = attributes
       end
 
diff --git a/lib/gitlab/sidekiq_config/worker.rb b/lib/gitlab/sidekiq_config/worker.rb
index 46fa0aa5be1..aea4209f631 100644
--- a/lib/gitlab/sidekiq_config/worker.rb
+++ b/lib/gitlab/sidekiq_config/worker.rb
@@ -6,10 +6,9 @@ module Gitlab
       include Comparable
 
       attr_reader :klass
-      delegate :feature_category_not_owned?, :get_feature_category, :get_tags,
-               :get_urgency, :get_weight, :get_worker_resource_boundary,
-               :idempotent?, :queue, :queue_namespace,
-               :worker_has_external_dependencies?,
+      delegate :feature_category_not_owned?, :get_feature_category, :get_sidekiq_options,
+               :get_tags, :get_urgency, :get_weight, :get_worker_resource_boundary,
+               :idempotent?, :queue, :queue_namespace, :worker_has_external_dependencies?,
                to: :klass
 
       def initialize(klass, ee:)
@@ -47,6 +46,7 @@ module Gitlab
       def to_yaml
         {
           name: queue,
+          worker_name: klass.name,
           feature_category: get_feature_category,
           has_external_dependencies: worker_has_external_dependencies?,
           urgency: get_urgency,
@@ -64,6 +64,10 @@ module Gitlab
       def queue_and_weight
         [queue, get_weight]
       end
+
+      def retries
+        get_sidekiq_options['retry']
+      end
     end
   end
 end
diff --git a/lib/gitlab/sidekiq_config/worker_matcher.rb b/lib/gitlab/sidekiq_config/worker_matcher.rb
index fe5ac10c65a..d615d5ecba4 100644
--- a/lib/gitlab/sidekiq_config/worker_matcher.rb
+++ b/lib/gitlab/sidekiq_config/worker_matcher.rb
@@ -10,6 +10,7 @@ module Gitlab
       QUERY_TERM_REGEX = %r{^(\w+)(!?=)([\w:#{QUERY_CONCATENATE_OPERATOR}]+)}.freeze
 
       QUERY_PREDICATES = {
+        worker_name: :to_s,
         feature_category: :to_sym,
         has_external_dependencies: lambda { |value| value == 'true' },
         name: :to_s,
@@ -50,7 +51,7 @@ module Gitlab
       def predicate_for_term(term)
         match = term.match(QUERY_TERM_REGEX)
 
-        raise InvalidTerm.new("Invalid term: #{term}") unless match
+        raise InvalidTerm, "Invalid term: #{term}" unless match
 
         _, lhs, op, rhs = *match
 
@@ -66,14 +67,14 @@ module Gitlab
         else
           # This is unreachable because InvalidTerm will be raised instead, but
           # keeping it allows to guard against that changing in future.
-          raise UnknownOperator.new("Unknown operator: #{op}")
+          raise UnknownOperator, "Unknown operator: #{op}"
         end
       end
 
       def predicate_factory(lhs, values)
         values_block = QUERY_PREDICATES[lhs.to_sym]
 
-        raise UnknownPredicate.new("Unknown predicate: #{lhs}") unless values_block
+        raise UnknownPredicate, "Unknown predicate: #{lhs}" unless values_block
 
         lambda do |queue|
           comparator = Array(queue[lhs.to_sym]).to_set
diff --git a/lib/gitlab/sidekiq_config/worker_router.rb b/lib/gitlab/sidekiq_config/worker_router.rb
new file mode 100644
index 00000000000..946296a24d3
--- /dev/null
+++ b/lib/gitlab/sidekiq_config/worker_router.rb
@@ -0,0 +1,107 @@
+# frozen_string_literal: true
+
+module Gitlab
+  module SidekiqConfig
+    class WorkerRouter
+      InvalidRoutingRuleError = Class.new(StandardError)
+      RuleEvaluator = Struct.new(:matcher, :queue_name)
+
+      def self.queue_name_from_worker_name(worker_klass)
+        base_queue_name =
+          worker_klass.name
+            .delete_prefix('Gitlab::')
+            .delete_suffix('Worker')
+            .underscore
+            .tr('/', '_')
+        [worker_klass.queue_namespace, base_queue_name].compact.join(':')
+      end
+
+      def self.global
+        @global_worker_router ||= new(::Gitlab.config.sidekiq.routing_rules)
+      rescue InvalidRoutingRuleError, ::Gitlab::SidekiqConfig::WorkerMatcher::UnknownPredicate => e
+        ::Gitlab::ErrorTracking.track_and_raise_for_dev_exception(e)
+
+        @global_worker_router = new([])
+      end
+
+      # call-seq:
+      #   router = WorkerRouter.new([
+      #     ["resource_boundary=cpu", 'cpu_boundary'],
+      #     ["feature_category=pages", nil],
+      #     ["feature_category=source_code_management", ''],
+      #     ["*", "default"]
+      #   ])
+      #   router.route(ACpuBoundaryWorker) # Return "cpu_boundary"
+      #   router.route(JustAPagesWorker)   # Return "just_a_pages_worker"
+      #   router.route(PostReceive)        # Return "post_receive"
+      #   router.route(RandomWorker)       # Return "default"
+      #
+      # This class is responsible for routing a Sidekiq worker to a certain
+      # queue defined in the input routing rules. The input routing rules, as
+      # described above, is an order-matter array of tuples [query, queue_name].
+      #
+      # - The query syntax is the same as the "queue selector" detailedly
+      # denoted in doc/administration/operations/extra_sidekiq_processes.md.
+      #
+      # - The queue_name must be a valid Sidekiq queue name. If the queue name
+      # is nil, or an empty string, the worker is routed to the queue generated
+      # by the name of the worker instead.
+      #
+      # Rules are evaluated from first to last, and as soon as we find a match
+      # for a given worker we stop processing for that worker (first match
+      # wins). If the worker doesn't match any rule, it falls back the queue
+      # name generated from the worker name
+      #
+      # For further information, please visit:
+      #   https://gitlab.com/gitlab-com/gl-infra/scalability/-/issues/1016
+      #
+      def initialize(routing_rules)
+        @rule_evaluators = parse_routing_rules(routing_rules)
+      end
+
+      def route(worker_klass)
+        # A medium representation to ensure the backward-compatibility of
+        # WorkerMatcher
+        worker_metadata = generate_worker_metadata(worker_klass)
+        @rule_evaluators.each do |evaluator|
+          if evaluator.matcher.match?(worker_metadata)
+            return evaluator.queue_name.presence || queue_name_from_worker_name(worker_klass)
+          end
+        end
+
+        queue_name_from_worker_name(worker_klass)
+      end
+
+      private
+
+      def parse_routing_rules(routing_rules)
+        raise InvalidRoutingRuleError, 'The set of routing rule must be an array' unless routing_rules.is_a?(Array)
+
+        routing_rules.map do |rule_tuple|
+          raise InvalidRoutingRuleError, "Routing rule `#{rule_tuple.inspect}` is invalid" unless valid_routing_rule?(rule_tuple)
+
+          selector, destination_queue = rule_tuple
+          RuleEvaluator.new(
+            ::Gitlab::SidekiqConfig::WorkerMatcher.new(selector),
+            destination_queue
+          )
+        end
+      end
+
+      def valid_routing_rule?(rule_tuple)
+        rule_tuple.is_a?(Array) && rule_tuple.length == 2
+      end
+
+      def generate_worker_metadata(worker_klass)
+        # The ee indicator here is insignificant and irrelevant to the matcher.
+        # Plus, it's not easy to determine whether a worker is **only**
+        # available in EE.
+        ::Gitlab::SidekiqConfig::Worker.new(worker_klass, ee: false).to_yaml
+      end
+
+      def queue_name_from_worker_name(worker_klass)
+        self.class.queue_name_from_worker_name(worker_klass)
+      end
+    end
+  end
+end