Add latest changes from gitlab-org/gitlab@13-11-stable-eev13.11.0-rc43

5 files changed, 46 insertions, 7 deletions

diff --git a/app/workers/concerns/application_worker.rb b/app/workers/concerns/application_worker.rb
index d101ef100d8..0de26e27631 100644
--- a/app/workers/concerns/application_worker.rb
+++ b/app/workers/concerns/application_worker.rb
@@ -18,7 +18,7 @@ module ApplicationWorker
     set_queue
 
     def structured_payload(payload = {})
-      context = Labkit::Context.current.to_h.merge(
+      context = Gitlab::ApplicationContext.current.merge(
         'class' => self.class.name,
         'job_status' => 'running',
         'queue' => self.class.queue,
diff --git a/app/workers/concerns/cronjob_queue.rb b/app/workers/concerns/cronjob_queue.rb
index 955387b5ad4..b89d6bba72c 100644
--- a/app/workers/concerns/cronjob_queue.rb
+++ b/app/workers/concerns/cronjob_queue.rb
@@ -15,7 +15,7 @@ module CronjobQueue
     # Cronjobs never get scheduled with arguments, so this is safe to
     # override
     def context_for_arguments(_args)
-      return if Gitlab::ApplicationContext.current_context_include?('meta.caller_id')
+      return if Gitlab::ApplicationContext.current_context_include?(:caller_id)
 
       Gitlab::ApplicationContext.new(caller_id: "Cronjob")
     end
diff --git a/app/workers/concerns/each_shard_worker.rb b/app/workers/concerns/each_shard_worker.rb
index 00f589f957e..d1d558f55fe 100644
--- a/app/workers/concerns/each_shard_worker.rb
+++ b/app/workers/concerns/each_shard_worker.rb
@@ -24,7 +24,13 @@ module EachShardWorker
   end
 
   def healthy_ready_shards
-    ready_shards.select(&:success)
+    success_checks, failed_checks = ready_shards.partition(&:success)
+
+    if failed_checks.any?
+      ::Gitlab::AppLogger.error(message: 'Excluding unhealthy shards', failed_checks: failed_checks.map(&:payload), class: self.class.name)
+    end
+
+    success_checks
   end
 
   def ready_shards
diff --git a/app/workers/concerns/reactive_cacheable_worker.rb b/app/workers/concerns/reactive_cacheable_worker.rb
index 189b0607605..9e882c8ac7a 100644
--- a/app/workers/concerns/reactive_cacheable_worker.rb
+++ b/app/workers/concerns/reactive_cacheable_worker.rb
@@ -17,10 +17,10 @@ module ReactiveCacheableWorker
 
   def perform(class_name, id, *args)
     klass = begin
-              class_name.constantize
-            rescue NameError
-              nil
-            end
+      class_name.constantize
+    rescue NameError
+      nil
+    end
 
     return unless klass
 
diff --git a/app/workers/concerns/worker_attributes.rb b/app/workers/concerns/worker_attributes.rb
index 042508d08f2..6f99fd089ac 100644
--- a/app/workers/concerns/worker_attributes.rb
+++ b/app/workers/concerns/worker_attributes.rb
@@ -11,6 +11,8 @@ module WorkerAttributes
   # Urgencies that workers can declare through the `urgencies` attribute
   VALID_URGENCIES = [:high, :low, :throttled].freeze
 
+  VALID_DATA_CONSISTENCIES = [:always, :sticky, :delayed].freeze
+
   NAMESPACE_WEIGHTS = {
     auto_devops: 2,
     auto_merge: 3,
@@ -69,6 +71,35 @@ module WorkerAttributes
       class_attributes[:urgency] || :low
     end
 
+    def data_consistency(data_consistency, feature_flag: nil)
+      raise ArgumentError, "Invalid data consistency: #{data_consistency}" unless VALID_DATA_CONSISTENCIES.include?(data_consistency)
+      raise ArgumentError, 'Data consistency is already set' if class_attributes[:data_consistency]
+
+      class_attributes[:data_consistency_feature_flag] = feature_flag if feature_flag
+      class_attributes[:data_consistency] = data_consistency
+
+      validate_worker_attributes!
+    end
+
+    def validate_worker_attributes!
+      # Since the deduplication should always take into account the latest binary replication pointer into account,
+      # not the first one, the deduplication will not work with sticky or delayed.
+      # Follow up issue to improve this: https://gitlab.com/gitlab-org/gitlab/-/issues/325291
+      if idempotent? && get_data_consistency != :always
+        raise ArgumentError, "Class can't be marked as idempotent if data_consistency is not set to :always"
+      end
+    end
+
+    def get_data_consistency
+      class_attributes[:data_consistency] || :always
+    end
+
+    def get_data_consistency_feature_flag_enabled?
+      return true unless class_attributes[:data_consistency_feature_flag]
+
+      Feature.enabled?(class_attributes[:data_consistency_feature_flag], default_enabled: :yaml)
+    end
+
     # Set this attribute on a job when it will call to services outside of the
     # application, such as 3rd party applications, other k8s clusters etc See
     # doc/development/sidekiq_style_guide.md#jobs-with-external-dependencies for
@@ -96,6 +127,8 @@ module WorkerAttributes
 
     def idempotent!
       class_attributes[:idempotent] = true
+
+      validate_worker_attributes!
     end
 
     def idempotent?