Instrument Que job enqueues

Record an `enqueue.que` event when a Que job is enqueued, and a
`bulk_enqueue.que` event for the whole batch on a bulk enqueue, so
enqueues show up in the transaction timeline. The bulk wrapper records
one event and its inner enqueues pass through.

The bulk module is only prepended on Que 2+, which has `bulk_enqueue`,
so no phantom method is defined on Que 1.

Author: unflxw

.changesets/instrument-que-enqueues.md

@@ -0,0 +1,6 @@
+---
+bump: minor
+type: add
+---
+
+Instrument Que job enqueues. AppSignal now records an `enqueue.que` event when a Que job is enqueued, and a `bulk_enqueue.que` event for bulk enqueues on Que 2, so the enqueue shows up in the transaction timeline.

lib/appsignal/hooks/que.rb

@@ -13,6 +13,12 @@ def dependencies_present?
       def install
         require "appsignal/integrations/que"
         ::Que::Job.prepend Appsignal::Integrations::QuePlugin
+        ::Que::Job.singleton_class.prepend Appsignal::Integrations::QueClientPlugin
+
+        # `bulk_enqueue` exists only on Que 2+; don't define one where it's absent.
+        if ::Que::Job.respond_to?(:bulk_enqueue)
+          ::Que::Job.singleton_class.prepend Appsignal::Integrations::QueBulkClientPlugin
+        end
 
         ::Que.error_notifier = proc do |error, _job|
           Appsignal::Transaction.current.set_error(error)

lib/appsignal/integrations/que.rb

@@ -34,5 +34,41 @@ def _run(*args)
         end
       end
     end
+
+    # @!visibility private
+    #
+    # Prepended to `Que::Job`'s singleton so it records each enqueue as an
+    # `enqueue.que` event under the active transaction. Like all AppSignal
+    # events, it only records when there's an active transaction (e.g. enqueuing
+    # from within a web request or another job); otherwise it's a transparent
+    # pass-through.
+    module QueClientPlugin
+      def enqueue(*_args, **_rest)
+        # Inside a Que 2 `bulk_enqueue` block the batch is recorded once by the
+        # `bulk_enqueue` wrapper, so each inner enqueue is a pass-through to
+        # avoid recording an event per job.
+        return super if bulk_insert_in_progress?
+
+        Appsignal.instrument("enqueue.que") { super }
+      end
+
+      private
+
+      def bulk_insert_in_progress?
+        !Thread.current[:que_jobs_to_bulk_insert].nil?
+      end
+    end
+
+    # @!visibility private
+    #
+    # `bulk_enqueue` exists only on Que 2+, so this lives in its own module that
+    # the hook prepends only when Que has the method -- otherwise we'd define a
+    # `bulk_enqueue` on Que versions that have none. The whole batch records a
+    # single `bulk_enqueue.que` event; the inner enqueues are pass-throughs.
+    module QueBulkClientPlugin
+      def bulk_enqueue(*_args, **_rest)
+        Appsignal.instrument("bulk_enqueue.que") { super }
+      end
+    end
   end
 end

spec/lib/appsignal/integrations/que_spec.rb

@@ -184,4 +184,94 @@ def run(*_args)
       end
     end
   end
+
+  describe Appsignal::Integrations::QueClientPlugin do
+    let(:job) do
+      Class.new(::Que::Job) do
+        def self.name
+          "MyQueJob"
+        end
+      end
+    end
+
+    # Capture what Que would persist, without needing a database.
+    let(:captured) { {} }
+    before do
+      allow(Que).to receive(:execute) do |command, values|
+        captured[:values] = values if command == :insert_job
+        [{}]
+      end
+
+      start_agent
+    end
+    around { |example| keep_transactions { example.run } }
+
+    # `data` is the last value Que passes to its `:insert_job` query on both Que
+    # 1 and Que 2 (Que 2 inserts `kwargs` before it, shifting its index); the
+    # tags live under it as a JSON string.
+    def enqueued_tags
+      data = captured[:values]&.last
+      data ? JSON.parse(data)["tags"] : nil
+    end
+
+    def enqueue(tags: ["user:42"])
+      job.enqueue("post_id_123", :job_options => { :tags => tags })
+    end
+
+    context "with an active transaction" do
+      it "records an enqueue event and leaves the job's tags untouched" do
+        transaction = http_request_transaction
+        set_current_transaction(transaction)
+
+        enqueue
+
+        event_names = transaction.to_h["events"].map { |event| event["name"] }
+        expect(event_names).to include("enqueue.que")
+        expect(enqueued_tags).to eq(["user:42"])
+      end
+    end
+
+    context "without an active transaction" do
+      it "is a transparent pass-through" do
+        expect { enqueue }.to_not raise_error
+
+        expect(enqueued_tags).to eq(["user:42"])
+      end
+    end
+
+    # `bulk_enqueue` is Que 2 only. The whole batch records a single
+    # `bulk_enqueue.que` event; the inner enqueues are pass-throughs.
+    describe "#bulk_enqueue", :if => DependencyHelper.que2_present? do
+      before do
+        # Que's bulk path constantizes the job class by name, so it needs a real
+        # constant (the single-enqueue path uses `new` and doesn't).
+        stub_const("MyQueJob", job)
+        allow(Que).to receive(:transaction).and_yield
+        allow(Que).to receive(:execute) do |command, values|
+          captured[:values] = values if command == :bulk_insert_jobs
+          [{}]
+        end
+      end
+
+      def bulk_enqueue(tags: ["user:42"])
+        job.bulk_enqueue(:job_options => { :tags => tags }) do
+          job.enqueue("post_id_123")
+          job.enqueue("post_id_456")
+        end
+      end
+
+      it "records one bulk_enqueue event for the whole batch" do
+        transaction = http_request_transaction
+        set_current_transaction(transaction)
+
+        bulk_enqueue
+
+        event_names = transaction.to_h["events"].map { |event| event["name"] }
+        # One event for the whole batch -- the inner enqueues don't add their own.
+        expect(event_names.count { |name| name == "bulk_enqueue.que" }).to eq(1)
+        expect(event_names).to_not include("enqueue.que")
+        expect(enqueued_tags).to eq(["user:42"])
+      end
+    end
+  end
 end