Active Job notifications subscriber

CHANGELOG.md

@@ -15,6 +15,7 @@ The upcoming release of the agent introduces additional Ruby on Rails instrument
   - Action Dispatch (for middleware) [PR#1745](https://github.com/newrelic/newrelic-ruby-agent/pull/1745)
   - Action Mailbox (for sending mail) [PR#1740](https://github.com/newrelic/newrelic-ruby-agent/pull/1740)
   - Action Mailer (for routing mail) [PR#1740](https://github.com/newrelic/newrelic-ruby-agent/pull/1740)
+  - Active Job (for background jobs) [PR#1742](https://github.com/newrelic/newrelic-ruby-agent/pull/1761)
   - Active Support (for caching operations) [PR#1742](https://github.com/newrelic/newrelic-ruby-agent/pull/1742)
 
   The instrumentations for each of these libaries are all enabled by default, but can be independently disabled via configuration by using the following parameters:
@@ -26,6 +27,7 @@ The upcoming release of the agent introduces additional Ruby on Rails instrument
   | `disable_action_dispatch` | `false` | If `true`, disables Action Dispatch instrumentation. |
   | `disable_action_mailbox` | `false` | If `true`, disables Action Mailbox instrumentation. |
   | `disable_action_mailer` | `false` | If `true`, disables Action Mailer instrumentation. |
+  | `disable_activejob` | `false` | If `true`, disables Active Job instrumentation. |
   | `disable_active_support` | `false` | If `true`, disables Active Support instrumentation. |
 
 ## 8.15.0

lib/new_relic/agent/instrumentation/active_job.rb

@@ -12,7 +12,7 @@
   end
 
   executes do
-    NewRelic::Agent.logger.info('Installing ActiveJob instrumentation')
+    NewRelic::Agent.logger.info('Installing base ActiveJob instrumentation')
 
     ActiveSupport.on_load(:active_job) do
       ActiveJob::Base.around_enqueue do |job, block|
@@ -26,6 +26,19 @@
       NewRelic::Agent::PrependSupportability.record_metrics_for(ActiveJob::Base)
     end
   end
+
+  executes do
+    if defined?(ActiveSupport) &&
+        ActiveJob.respond_to?(:gem_version) &&
+        ActiveJob.gem_version >= Gem::Version.new('6.0.0') &&
+        !NewRelic::Agent.config[:disable_activejob] &&
+        !NewRelic::Agent::Instrumentation::ActiveJobSubscriber.subscribed?
+      NewRelic::Agent.logger.info('Installing notifications based ActiveJob instrumentation')
+
+      ActiveSupport::Notifications.subscribe(/\A[^\.]+\.active_job\z/,
+        NewRelic::Agent::Instrumentation::ActiveJobSubscriber.new)
+    end
+  end
 end
 
 module NewRelic

lib/new_relic/agent/instrumentation/active_job_subscriber.rb

@@ -0,0 +1,41 @@
+# This file is distributed under New Relic's license terms.
+# See https://github.com/newrelic/newrelic-ruby-agent/blob/main/LICENSE for complete details.
+# frozen_string_literal: true
+
+require 'new_relic/agent/instrumentation/notifications_subscriber'
+
+module NewRelic
+  module Agent
+    module Instrumentation
+      class ActiveJobSubscriber < NotificationsSubscriber
+        PAYLOAD_KEYS = %i[adapter db_runtime error job wait]
+
+        def add_segment_params(segment, payload)
+          PAYLOAD_KEYS.each do |key|
+            segment.params[key] = payload[key] if payload.key?(key)
+          end
+        end
+
+        def metric_name(name, payload)
+          queue = payload[:job].queue_name
+          method = method_from_name(name)
+          "Ruby/ActiveJob/#{queue}/#{method}"
+        end
+
+        PATTERN = /\A([^\.]+)\.active_job\z/
+
+        METHOD_NAME_MAPPING = Hash.new do |h, k|
+          if PATTERN =~ k
+            h[k] = $1
+          else
+            h[k] = NewRelic::UNKNOWN
+          end
+        end
+
+        def method_from_name(name)
+          METHOD_NAME_MAPPING[name]
+        end
+      end
+    end
+  end
+end

test/new_relic/agent/instrumentation/active_job_subscriber_test.rb

@@ -0,0 +1,14 @@
+# This file is distributed under New Relic's license terms.
+# See https://github.com/newrelic/newrelic-ruby-agent/blob/main/LICENSE for complete details.
+# frozen_string_literal: true
+
+require_relative '../../../test_helper'
+require 'new_relic/agent/instrumentation/active_job_subscriber'
+
+if defined?(ActiveJob) &&
+    ActiveJob.respond_to?(:gem_version) &&
+    ActiveJob.gem_version >= Gem::Version.new('6.0.0')
+  require_relative 'rails/active_job_subscriber'
+else
+  puts "Skipping tests in #{__FILE__} because ActiveJob is unavailable or < 6.0"
+end

test/new_relic/agent/instrumentation/rails/active_job_subscriber.rb

@@ -0,0 +1,94 @@
+# This file is distributed under New Relic's license terms.
+# See https://github.com/newrelic/newrelic-ruby-agent/blob/main/LICENSE for complete details.
+# frozen_string_literal: true
+
+require_relative '../../../../test_helper'
+require 'new_relic/agent/instrumentation/active_job_subscriber'
+
+module NewRelic::Agent::Instrumentation
+  class RetryMe < StandardError; end
+  class DiscardMe < StandardError; end
+
+  class TestJob < ActiveJob::Base
+    retry_on RetryMe
+
+    discard_on DiscardMe
+
+    def perform(error = nil)
+      raise error if error
+
+      rand(1138)
+    end
+  end
+
+  class ActiveJobSubscriberTest < Minitest::Test
+    NAME = 'perform.active_job'
+    ID = 71741
+    SUBSCRIBER = NewRelic::Agent::Instrumentation::ActiveJobSubscriber.new
+
+    def test_segment_naming_with_unknown_method
+      assert_equal 'Ruby/ActiveJob/default/Unknown',
+        SUBSCRIBER.send(:metric_name, 'indecipherable', {job: TestJob.new})
+    end
+
+    # perform.active_job
+    def test_perform_active_job
+      job = TestJob.new
+      in_transaction do |txn|
+        job.perform_now
+        validate_transaction(txn, 'perform')
+      end
+    end
+
+    # enqueue_at.active_job
+    def test_enqueue_at_active_job
+      in_transaction do |txn|
+        TestJob.set(wait_until: 7.hours.from_now).perform_later
+        validate_transaction(txn, 'enqueue_at')
+      end
+    end
+
+    # enqueue.active_job
+    def test_enqueue_active_job
+      in_transaction do |txn|
+        TestJob.perform_later
+        validate_transaction(txn, 'enqueue')
+      end
+    end
+
+    # perform_start.active_job
+    # enqueue_retry.active_job
+    def test_perform_start_active_job_and_enqueue_retry_active_job
+      in_transaction do |txn|
+        TestJob.perform_now(RetryMe)
+        validate_transaction(txn, %w[enqueue_retry perform_start])
+      end
+    end
+
+    # discard.active_job
+    def test_discard_active_job
+      in_transaction do |txn|
+        TestJob.perform_now(DiscardMe)
+        validate_transaction(txn, 'discard')
+      end
+    end
+
+    # TODO: test for retry_stopped.active_job
+
+    private
+
+    def validate_transaction(txn, methods = [])
+      methods = Array(methods)
+      segments = txn.segments.select { |s| s.name.start_with?('Ruby/ActiveJob') }
+
+      refute_empty segments
+
+      methods.each do |method|
+        segment = segments.detect { |s| s.name == "Ruby/ActiveJob/default/#{method}" }
+
+        assert segment
+        assert_equal 'ActiveJob::QueueAdapters::AsyncAdapter', segment.params[:adapter].class.name
+      end
+    end
+  end
+end

test/simplecov_test_helper.rb

@@ -7,4 +7,11 @@
 # required Ruby >= 2.5.0 and Ruby 2.6.0 was marked for EOL
 SIMPLECOV_MIN_RUBY_VERSION = '2.7.0'
 
-require 'simplecov' if RUBY_VERSION >= SIMPLECOV_MIN_RUBY_VERSION
+begin
+  require 'simplecov' if RUBY_VERSION >= SIMPLECOV_MIN_RUBY_VERSION
+rescue LoadError => e
+  puts
+  puts "SimpleCov requested by Ruby #{RUBY_VERSION} which is >=#{SIMPLECOV_MIN_RUBY_VERSION} "
+  puts "but the gem is not available. #{e.class}: #{e.message}"
+  puts
+end