capture job_attempts on Worker

Author: alachaum

app/controllers/cloudtasker/worker_controller.rb

@@ -63,7 +63,9 @@ def json_payload
     #
     def payload
       # Return content parsed as JSON and add job retries count
-      @payload ||= JSON.parse(json_payload).merge(job_retries: job_retries, task_id: task_id)
+      @payload ||= JSON.parse(json_payload).merge(
+        job_retries: job_retries, job_attempts: job_attempts, task_id: task_id
+      )
     end
 
     #
@@ -75,6 +77,16 @@ def job_retries
       request.headers[Cloudtasker::Config::RETRY_HEADER].to_i
     end
 
+    #
+    # Extract the number of times this task was attempted at runtime.
+    # This includes all attempts (including 50x errors).
+    #
+    # @return [Integer] The number of attempts.
+    #
+    def job_attempts
+      request.headers[Cloudtasker::Config::ATTEMPT_HEADER].to_i
+    end
+
     #
     # Return the Google Cloud Task ID from headers.
     #

lib/cloudtasker/config.rb

@@ -19,10 +19,11 @@ class Config
     # Retry header in Cloud Task responses
     #
     # Definitions:
-    #   X-CloudTasks-TaskRetryCount: total number of retries (including 504 "instance unreachable")
-    #   X-CloudTasks-TaskExecutionCount: number of non-503 retries (= actual number of job failures)
+    #   X-CloudTasks-TaskRetryCount: total number of retries (including 50x errors)
+    #   X-CloudTasks-TaskExecutionCount: number of non-50x retries (= actual number of job failures)
     #
     RETRY_HEADER = 'X-Cloudtasks-Taskexecutioncount'
+    ATTEMPT_HEADER = 'X-CloudTasks-TaskRetryCount'
 
     # Cloud Task ID header
     TASK_ID_HEADER = 'X-CloudTasks-TaskName'

lib/cloudtasker/worker.rb

@@ -8,7 +8,7 @@ def self.included(base)
       base.extend(ClassMethods)
       base.attr_writer :job_queue
       base.attr_accessor :job_args, :job_id, :job_meta, :job_reenqueued, :job_retries,
-                         :perform_started_at, :perform_ended_at, :task_id
+                         :job_attempts, :perform_started_at, :perform_ended_at, :task_id
     end
 
     #
@@ -47,7 +47,10 @@ def self.from_hash(hash)
       return nil unless worker_klass.include?(self)
 
       # Return instantiated worker
-      worker_klass.new(**payload.slice(:job_queue, :job_args, :job_id, :job_meta, :job_retries, :task_id))
+      worker_klass.new(**payload.slice(
+        :job_queue, :job_args, :job_id, :job_meta,
+        :job_retries, :job_attempts, :task_id
+      ))
     rescue NameError
       nil
     end
@@ -176,11 +179,13 @@ def max_retries
     # @param [Array<any>] job_args The list of perform args.
     # @param [String] job_id A unique ID identifying this job.
     #
-    def initialize(job_queue: nil, job_args: nil, job_id: nil, job_meta: {}, job_retries: 0, task_id: nil)
+    def initialize(job_queue: nil, job_args: nil, job_id: nil, job_meta: {}, job_retries: 0, job_attempts: 0,
+                   task_id: nil)
       @job_args = job_args || []
       @job_id = job_id || SecureRandom.uuid
       @job_meta = MetaStore.new(job_meta)
       @job_retries = job_retries || 0
+      @job_attempts = job_attempts || 0
       @job_queue = job_queue
       @task_id = task_id
     end
@@ -331,6 +336,7 @@ def to_h
         job_args: job_args,
         job_meta: job_meta.to_h,
         job_retries: job_retries,
+        job_attempts: job_attempts,
         job_queue: job_queue,
         task_id: task_id
       }

spec/cloudtasker/worker_controller_spec.rb

@@ -21,22 +21,25 @@
       end
       let(:mime_type) { :json }
       let(:request_body) { payload.to_json }
-      let(:expected_payload) { payload.merge(job_retries: retries, task_id: task_id) }
+      let(:expected_payload) { payload.merge(job_retries: retries, job_attempts: attempts, task_id: task_id) }
       let(:task_id) { 'ab2341f' }
       let(:id) { '111' }
       let(:worker_class_name) { 'TestWorker' }
       let(:args) { [1, 2] }
       let(:meta) { { 'foo' => 'bar' } }
       let(:retries) { 3 }
+      let(:attempts) { 5 }
       let(:queue) { 'some-queue' }
       let(:signature) { Cloudtasker::Authenticator.sign_payload(payload.to_json) }
 
       let(:signature_header) { "HTTP_#{Cloudtasker::Config::CT_SIGNATURE_HEADER.tr('-', '_').upcase}" }
       let(:env_retries_header) { "HTTP_#{Cloudtasker::Config::RETRY_HEADER.tr('-', '_').upcase}" }
+      let(:env_attempts_header) { "HTTP_#{Cloudtasker::Config::ATTEMPT_HEADER.tr('-', '_').upcase}" }
       let(:env_task_id_header) { "HTTP_#{Cloudtasker::Config::TASK_ID_HEADER.tr('-', '_').upcase}" }
 
       before do
         request.env[env_retries_header] = retries
+        request.env[env_attempts_header] = attempts
         request.env[env_task_id_header] = task_id
       end
 

spec/cloudtasker/worker_spec.rb

@@ -33,13 +33,15 @@
     let(:job_retries) { 3 }
     let(:job_queue) { 'critical' }
     let(:worker_class_name) { worker_class.to_s }
+    let(:job_attempts) { 5 }
     let(:worker_hash) do
       {
         'worker' => worker_class_name,
         'job_id' => job_id,
         'job_args' => job_args,
         'job_meta' => job_meta,
         'job_retries' => job_retries,
+        'job_attempts' => job_attempts,
         'job_queue' => job_queue,
         'task_id' => task_id
       }
@@ -53,6 +55,7 @@
           job_args: job_args,
           job_meta: eq(job_meta),
           job_retries: job_retries,
+          job_attempts: job_attempts,
           task_id: task_id
         }
       end
@@ -233,6 +236,7 @@
     let(:args) { [1, 2] }
     let(:meta) { { foo: 'bar' } }
     let(:retries) { 3 }
+    let(:attempts) { 5 }
     let(:queue) { 'critical' }
 
     context 'without args' do
@@ -243,6 +247,7 @@
           job_args: [],
           job_id: be_a(String),
           job_retries: 0,
+          job_attempts: 0,
           task_id: nil
         }
       end
@@ -258,6 +263,7 @@
           job_id: id,
           job_meta: meta,
           job_retries: retries,
+          job_attempts: attempts,
           task_id: task_id
         }
       end
@@ -268,6 +274,7 @@
           job_id: id,
           job_meta: eq(meta),
           job_retries: retries,
+          job_attempts: attempts,
           task_id: task_id
         }
       end
@@ -597,6 +604,7 @@
         job_args: worker.job_args,
         job_meta: worker.job_meta.to_h,
         job_retries: worker.job_retries,
+        job_attempts: worker.job_attempts,
         job_queue: worker.job_queue,
         task_id: task_id
       }