Merge pull request #208 from rails/log-and-instrumentation

Rely on `ActiveSupport::Notifications` and `ActiveSupport::LogSubscriber` for logging and instrumentation

Author: rosa

app/models/solid_queue/blocked_execution.rb

@@ -10,21 +10,25 @@ class BlockedExecution < Execution
     scope :expired, -> { where(expires_at: ...Time.current) }
 
     class << self
-      def unblock(count)
-        expired.distinct.limit(count).pluck(:concurrency_key).then do |concurrency_keys|
-          release_many releasable(concurrency_keys)
+      def unblock(limit)
+        SolidQueue.instrument(:release_many_blocked, limit: limit) do |payload|
+          expired.distinct.limit(limit).pluck(:concurrency_key).then do |concurrency_keys|
+            payload[:size] = release_many releasable(concurrency_keys)
+          end
         end
       end
 
       def release_many(concurrency_keys)
         # We want to release exactly one blocked execution for each concurrency key, and we need to do it
         # one by one, locking each record and acquiring the semaphore individually for each of them:
-        Array(concurrency_keys).each { |concurrency_key| release_one(concurrency_key) }
+        Array(concurrency_keys).count { |concurrency_key| release_one(concurrency_key) }
       end
 
       def release_one(concurrency_key)
         transaction do
-          ordered.where(concurrency_key: concurrency_key).limit(1).non_blocking_lock.each(&:release)
+          if execution = ordered.where(concurrency_key: concurrency_key).limit(1).non_blocking_lock.first
+            execution.release
+          end
         end
       end
 
@@ -38,12 +42,14 @@ def releasable(concurrency_keys)
     end
 
     def release
-      transaction do
-        if acquire_concurrency_lock
-          promote_to_ready
-          destroy!
+      SolidQueue.instrument(:release_blocked, job_id: job.id, concurrency_key: concurrency_key, released: false) do |payload|
+        transaction do
+          if acquire_concurrency_lock
+            promote_to_ready
+            destroy!
 
-          SolidQueue.logger.debug("[SolidQueue] Unblocked job #{job.id} under #{concurrency_key}")
+            payload[:released] = true
+          end
         end
       end
     end

app/models/solid_queue/claimed_execution.rb

@@ -20,7 +20,12 @@ def claiming(job_ids, process_id, &block)
     end
 
     def release_all
-      includes(:job).each(&:release)
+      SolidQueue.instrument(:release_many_claimed) do |payload|
+        includes(:job).tap do |executions|
+          payload[:size] = executions.size
+          executions.each(&:release)
+        end
+      end
     end
 
     def discard_all_in_batches(*)
@@ -45,9 +50,11 @@ def perform
   end
 
   def release
-    transaction do
-      job.dispatch_bypassing_concurrency_limits
-      destroy!
+    SolidQueue.instrument(:release_claimed, job_id: job.id, process_id: process_id) do
+      transaction do
+        job.dispatch_bypassing_concurrency_limits
+        destroy!
+      end
     end
   end
 

app/models/solid_queue/execution.rb

@@ -13,6 +13,10 @@ class UndiscardableError < StandardError; end
     belongs_to :job
 
     class << self
+      def type
+        model_name.element.sub("_execution", "").to_sym
+      end
+
       def create_all_from_jobs(jobs)
         insert_all execution_data_from_jobs(jobs)
       end
@@ -27,25 +31,32 @@ def discard_all_in_batches(batch_size: 500)
         pending = count
         discarded = 0
 
-        loop do
-          transaction do
-            job_ids = limit(batch_size).order(:job_id).lock.pluck(:job_id)
+        SolidQueue.instrument(:discard_all, batch_size: batch_size, status: type, batches: 0, size: 0) do |payload|
+          loop do
+            transaction do
+              job_ids = limit(batch_size).order(:job_id).lock.pluck(:job_id)
+              discarded = discard_jobs job_ids
 
-            discard_jobs job_ids
-            discarded = where(job_id: job_ids).delete_all
-            pending -= discarded
-          end
+              where(job_id: job_ids).delete_all
+              pending -= discarded
+
+              payload[:size] += discarded
+              payload[:batches] += 1
+            end
 
-          break if pending <= 0 || discarded == 0
+            break if pending <= 0 || discarded == 0
+          end
         end
       end
 
       def discard_all_from_jobs(jobs)
-        transaction do
-          job_ids = lock_all_from_jobs(jobs)
+        SolidQueue.instrument(:discard_all, jobs_size: jobs.size, status: type) do |payload|
+          transaction do
+            job_ids = lock_all_from_jobs(jobs)
 
-          discard_jobs job_ids
-          where(job_id: job_ids).delete_all
+            payload[:size] = discard_jobs job_ids
+            where(job_id: job_ids).delete_all
+          end
         end
       end
 
@@ -59,10 +70,16 @@ def discard_jobs(job_ids)
         end
     end
 
+    def type
+      self.class.type
+    end
+
     def discard
-      with_lock do
-        job.destroy
-        destroy
+      SolidQueue.instrument(:discard, job_id: job_id, status: type) do
+        with_lock do
+          job.destroy
+          destroy
+        end
       end
     end
   end

app/models/solid_queue/execution/dispatching.rb

@@ -9,9 +9,11 @@ module Dispatching
         def dispatch_jobs(job_ids)
           jobs = Job.where(id: job_ids)
 
-          Job.dispatch_all(jobs).map(&:id).tap do |dispatched_job_ids|
-            where(job_id: dispatched_job_ids).order(:job_id).delete_all
-            SolidQueue.logger.info("[SolidQueue] Dispatched #{dispatched_job_ids.size} jobs")
+          Job.dispatch_all(jobs).map(&:id).then do |dispatched_job_ids|
+            if dispatched_job_ids.none? then 0
+            else
+              where(job_id: dispatched_job_ids).order(:job_id).delete_all
+            end
           end
         end
       end

app/models/solid_queue/failed_execution.rb

@@ -11,15 +11,19 @@ class FailedExecution < Execution
     attr_accessor :exception
 
     def self.retry_all(jobs)
-      transaction do
-        dispatch_jobs lock_all_from_jobs(jobs)
+      SolidQueue.instrument(:retry_all, jobs_size: jobs.size) do |payload|
+        transaction do
+          payload[:size] = dispatch_jobs lock_all_from_jobs(jobs)
+        end
       end
     end
 
     def retry
-      with_lock do
-        job.prepare_for_execution
-        destroy!
+      SolidQueue.instrument(:retry, job_id: job.id) do
+        with_lock do
+          job.prepare_for_execution
+          destroy!
+        end
       end
     end
 

app/models/solid_queue/job/executable.rb

@@ -93,7 +93,7 @@ def status
         if finished?
           :finished
         elsif execution.present?
-          execution.model_name.element.sub("_execution", "").to_sym
+          execution.type
         end
       end
 

app/models/solid_queue/process.rb

@@ -12,17 +12,24 @@ class SolidQueue::Process < SolidQueue::Record
   after_destroy -> { claimed_executions.release_all }
 
   def self.register(**attributes)
-    create!(attributes.merge(last_heartbeat_at: Time.current))
+    SolidQueue.instrument :register_process, **attributes do
+      create!(attributes.merge(last_heartbeat_at: Time.current))
+    end
+  rescue Exception => error
+    SolidQueue.instrument :register_process, **attributes.merge(error: error)
+    raise
   end
 
   def heartbeat
     touch(:last_heartbeat_at)
   end
 
-  def deregister
-    destroy!
-  rescue Exception
-    SolidQueue.logger.error("[SolidQueue] Error deregistering process #{id} - #{metadata}")
-    raise
+  def deregister(pruned: false)
+    SolidQueue.instrument :deregister_process, process: self, pruned: pruned, claimed_size: claimed_executions.size do |payload|
+      destroy!
+    rescue Exception => error
+      payload[:error] = error
+      raise
+    end
   end
 end

app/models/solid_queue/process/prunable.rb

@@ -9,10 +9,11 @@ module SolidQueue::Process::Prunable
 
   class_methods do
     def prune
-      prunable.non_blocking_lock.find_in_batches(batch_size: 50) do |batch|
-        batch.each do |process|
-          SolidQueue.logger.info("[SolidQueue] Pruning dead process #{process.id} - #{process.metadata}")
-          process.deregister
+      SolidQueue.instrument :prune_processes, size: 0 do |payload|
+        prunable.non_blocking_lock.find_in_batches(batch_size: 50) do |batch|
+          payload[:size] += batch.size
+
+          batch.each { |process| process.deregister(pruned: true) }
         end
       end
     end

app/models/solid_queue/recurring_execution.rb

@@ -7,12 +7,12 @@ class RecurringExecution < Execution
     class << self
       def record(task_key, run_at, &block)
         transaction do
-          if job_id = block.call
-            create!(job_id: job_id, task_key: task_key, run_at: run_at)
+          block.call.tap do |active_job|
+            create!(job_id: active_job.provider_job_id, task_key: task_key, run_at: run_at)
           end
         end
       rescue ActiveRecord::RecordNotUnique
-        SolidQueue.logger.info("[SolidQueue] Skipped recurring task #{task_key} at #{run_at} — already dispatched")
+        # Task already dispatched
       end
 
       def clear_in_batches(batch_size: 500)

app/models/solid_queue/scheduled_execution.rb

@@ -16,7 +16,9 @@ def dispatch_next_batch(batch_size)
           job_ids = next_batch(batch_size).non_blocking_lock.pluck(:job_id)
           if job_ids.empty? then []
           else
-            dispatch_jobs(job_ids)
+            SolidQueue.instrument(:dispatch_scheduled, batch_size: batch_size) do |payload|
+              payload[:size] = dispatch_jobs(job_ids)
+            end
           end
         end
       end