Merge pull request #1682 from newrelic/concurrent_ruby_promises

Add instrumentation for Concurrent Ruby

CHANGELOG.md

@@ -6,6 +6,14 @@
 
   Following the 3.2.0 release of Ruby, the New Relic Ruby Agent has confirmed compatibility with and now supports the official release of Ruby 3.2.0. [PR#1715](https://github.com/newrelic/newrelic-ruby-agent/pull/1715)
 
+- **Add instrumentation for concurrent-ruby**
+
+  Instrumentation for the [concurrent-ruby](https://github.com/ruby-concurrency/concurrent-ruby) gem has been added to the agent. for versions 1.1.5 and above. When a transaction is already in progress and a call to a `Concurrent::` method that routes through `Concurrent::ThreadPoolExecutor#post` is made, a segment will be added to the transaction. Any content within the block passed to the `Concurrent::` method that is instrumented by the agent, such as a call to `Net::HTTP.get`, will have a nested segment created. [PR#1682](https://github.com/newrelic/newrelic-ruby-agent/pull/1682)
+
+  | Configuration name                | Default | Behavior                                                                                                                      |
+  | --------------------------------- | ------- | ----------------------------------------------------------------------------------------------------------------------------- |
+  | `instrumentation.concurrent_ruby` | auto    | Controls auto-instrumentation of the concurrent-ruby library at start up. May be one of `auto`, `prepend`, `chain`, `disabled`. |
+
 - **Infinite Tracing: Use batching and compression**
 
   For [Infinite Tracing](https://docs.newrelic.com/docs/distributed-tracing/infinite-tracing/introduction-infinite-tracing/) which Ruby applications can leverage with the `newrelic-infinite_tracing` gem, payloads will now be batched and compressed to signficantly decrease the amount of outbound network traffic. [PR#1723](https://github.com/newrelic/newrelic-ruby-agent/pull/1723)

lib/new_relic/agent/configuration/default_source.rb

@@ -1569,6 +1569,14 @@ def self.enforce_fallback(allowed_values: nil, fallback: nil)
           :allowed_from_server => false,
           :description => 'Controls auto-instrumentation of bunny at start up.  May be one of [auto|prepend|chain|disabled].'
         },
+        :'instrumentation.concurrent_ruby' => {
+          :default => 'auto',
+          :public => true,
+          :type => String,
+          :dynamic_name => true,
+          :allowed_from_server => false,
+          :description => 'Controls auto-instrumentation of the concurrent-ruby library at start up. May be one of [auto|prepend|chain|disabled].'
+        },
         :'instrumentation.curb' => {
           :default => instrumentation_value_of(:disable_curb),
           :documentation_default => 'auto',
@@ -1802,6 +1810,13 @@ def self.enforce_fallback(allowed_values: nil, fallback: nil)
           :allowed_from_server => false,
           :description => "Controls auto-instrumentation of the Thread class at start up to automatically add tracing to all Threads created in the application."
         },
+        :'thread_ids_enabled' => {
+          :default => false,
+          :public => false,
+          :type => Boolean,
+          :allowed_from_server => false,
+          :description => "If enabled, will append the current Thread and Fiber object ids onto the segment names of segments created in Threads and concurrent-ruby"
+        },
         :'instrumentation.tilt' => {
           :default => "auto",
           :public => true,

lib/new_relic/agent/instrumentation/concurrent_ruby.rb

@@ -0,0 +1,31 @@
+# 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 'concurrent_ruby/instrumentation'
+require_relative 'concurrent_ruby/chain'
+require_relative 'concurrent_ruby/prepend'
+
+DependencyDetection.defer do
+  named :'concurrent_ruby'
+
+  depends_on do
+    defined?(Concurrent) &&
+      Gem::Version.new(Concurrent::VERSION) >= Gem::Version.new('1.1.5')
+  end
+
+  executes do
+    NewRelic::Agent.logger.info('Installing concurrent-ruby instrumentation')
+
+    if use_prepend?
+      prepend_instrument(Concurrent::ThreadPoolExecutor, NewRelic::Agent::Instrumentation::ConcurrentRuby::Prepend)
+
+      [Concurrent::Promises.const_get(:'InternalStates')::Rejected,
+        Concurrent::Promises.const_get(:'InternalStates')::PartiallyRejected].each do |klass|
+          klass.prepend(NewRelic::Agent::Instrumentation::ConcurrentRuby::ErrorPrepend)
+        end
+    else
+      chain_instrument NewRelic::Agent::Instrumentation::ConcurrentRuby::Chain
+    end
+  end
+end

lib/new_relic/agent/instrumentation/concurrent_ruby/chain.rb

@@ -0,0 +1,36 @@
+# 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
+
+module NewRelic::Agent::Instrumentation
+  module ConcurrentRuby::Chain
+    def self.instrument!
+      ::Concurrent::ThreadPoolExecutor.class_eval do
+        include NewRelic::Agent::Instrumentation::ConcurrentRuby
+
+        alias_method(:post_without_new_relic, :post)
+
+        def post(*args, &task)
+          return post_without_new_relic(*args, &task) unless NewRelic::Agent::Tracer.tracing_enabled?
+
+          traced_task = add_task_tracing(*args, &task)
+          post_without_new_relic(*args, &traced_task)
+        end
+      end
+
+      [::Concurrent::Promises.const_get(:'InternalStates')::Rejected,
+        ::Concurrent::Promises.const_get(:'InternalStates')::PartiallyRejected].each do |klass|
+        klass.class_eval do
+          alias_method(:initialize_without_new_relic, :initialize)
+
+          # Uses args.last to record the error becuase the methods that this will monkey patch
+          # look like: initialize(reason) & initialize(value, reason)
+          def initialize(*args)
+            NewRelic::Agent.notice_error(args.last) if args.last.is_a?(Exception)
+            initialize_without_new_relic(*args)
+          end
+        end
+      end
+    end
+  end
+end

lib/new_relic/agent/instrumentation/concurrent_ruby/instrumentation.rb

@@ -0,0 +1,21 @@
+# 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
+
+module NewRelic::Agent::Instrumentation
+  module ConcurrentRuby
+    SEGMENT_NAME = 'Concurrent/Task'
+    SUPPORTABILITY_METRIC = 'Supportability/ConcurrentRuby/Invoked'
+
+    def add_task_tracing(*args, &task)
+      NewRelic::Agent.record_metric_once(SUPPORTABILITY_METRIC)
+
+      NewRelic::Agent::Tracer.thread_block_with_current_transaction(
+        *args,
+        segment_name: SEGMENT_NAME,
+        parent: NewRelic::Agent::Tracer.current_segment,
+        &task
+      )
+    end
+  end
+end

lib/new_relic/agent/instrumentation/concurrent_ruby/prepend.rb

@@ -0,0 +1,27 @@
+# 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
+
+module NewRelic::Agent::Instrumentation
+  module ConcurrentRuby
+    module Prepend
+      include NewRelic::Agent::Instrumentation::ConcurrentRuby
+
+      def post(*args, &task)
+        return super(*args, &task) unless NewRelic::Agent::Tracer.tracing_enabled?
+
+        traced_task = add_task_tracing(*args, &task)
+        super(*args, &traced_task)
+      end
+    end
+
+    module ErrorPrepend
+      # Uses args.last to record the error because the methods that this will be prepended to
+      # look like: initialize(reason) & initialize(value, reason)
+      def initialize(*args)
+        NewRelic::Agent.notice_error(args.last) if args.last.is_a?(Exception)
+        super
+      end
+    end
+  end
+end

lib/new_relic/agent/instrumentation/controller_instrumentation.rb

@@ -340,7 +340,7 @@ def self.class_name(traced_obj, options = {})
         #
         # Seldomly used options:
         #
-        # * <tt>:class_name => aClass.name</tt> is used to override the name
+        # * <tt>:class_name => Class.name</tt> is used to override the name
         #   of the class when used inside the metric name.  Default is the
         #   current class.
         # * <tt>:path => metric_path</tt> is *deprecated* in the public API.  It

lib/new_relic/agent/instrumentation/thread/instrumentation.rb

@@ -16,7 +16,11 @@ def initialize_with_newrelic_tracing
         def add_thread_tracing(*args, &block)
           return block if skip_tracing?
 
-          NewRelic::Agent::Tracer.thread_block_with_current_transaction(*args, &block)
+          NewRelic::Agent::Tracer.thread_block_with_current_transaction(
+            *args,
+            segment_name: 'Ruby/Thread',
+            &block
+          )
         end
 
         def skip_tracing?

lib/new_relic/agent/tracer.rb

@@ -2,6 +2,7 @@
 # See https://github.com/newrelic/newrelic-ruby-agent/blob/main/LICENSE for complete details.
 # frozen_string_literal: true
 
+require 'fiber'
 require 'new_relic/agent/transaction'
 require 'new_relic/agent/transaction/segment'
 require 'new_relic/agent/transaction/datastore_segment'
@@ -409,15 +410,18 @@ def clear_state
 
         alias_method :tl_clear, :clear_state
 
-        def thread_block_with_current_transaction(*args, &block)
+        def thread_block_with_current_transaction(*args, segment_name:, parent: nil, &block)
           current_txn = ::Thread.current[:newrelic_tracer_state].current_transaction if ::Thread.current[:newrelic_tracer_state] && ::Thread.current[:newrelic_tracer_state].is_execution_traced?
           proc do
             begin
               if current_txn
                 NewRelic::Agent::Tracer.state.current_transaction = current_txn
-                segment = NewRelic::Agent::Tracer.start_segment(name: "Ruby/Thread/#{::Thread.current.object_id}")
+                segment_name += "/Thread#{::Thread.current.object_id}/Fiber#{::Fiber.current.object_id}" if NewRelic::Agent.config[:'thread_ids_enabled']
+                segment = NewRelic::Agent::Tracer.start_segment(name: segment_name, parent: parent)
+              end
+              NewRelic::Agent::Tracer.capture_segment_error(segment) do
+                yield(*args)
               end
-              yield(*args) if block.respond_to?(:call)
             ensure
               ::NewRelic::Agent::Transaction::Segment.finish(segment)
             end

lib/new_relic/helper.rb

@@ -63,7 +63,7 @@ def run_command(command)
         raise NewRelic::CommandRunFailedError.new("Failed to run command '#{command}': #{message}")
       end
 
-      output.chomp
+      output.chomp if output
     end
 
     # TODO: Open3 defers the actual execution of a binary to Process.spawn,

lib/new_relic/traced_thread.rb

@@ -29,7 +29,11 @@ def initialize(*args, &block)
     def create_traced_block(*args, &block)
       return block if NewRelic::Agent.config[:'instrumentation.thread.tracing'] # if this is on, don't double trace
 
-      NewRelic::Agent::Tracer.thread_block_with_current_transaction(*args, &block)
+      NewRelic::Agent::Tracer.thread_block_with_current_transaction(
+        *args,
+        segment_name: 'Ruby/TracedThread',
+        &block
+      )
     end
   end
 end

newrelic.yml

@@ -328,6 +328,10 @@ common: &default_settings
   # May be one of [auto|prepend|chain|disabled].
   # instrumentation.bunny: auto
 
+  # Controls auto-instrumentation of concurrent_ruby at start up.
+  # May be one of [auto|prepend|chain|disabled]
+  # instrumentation.concurrent_ruby: auto
+
   # Controls auto-instrumentation of Curb at start up.
   # May be one of [auto|prepend|chain|disabled].
   # instrumentation.curb: auto

test/multiverse/suites/concurrent_ruby/Envfile

@@ -0,0 +1,21 @@
+# 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
+
+instrumentation_methods :chain, :prepend
+
+# The 1.1.x series of Concurrent Ruby starts to use the error classes
+# we reference in our logic to notice errors after 1.1.5
+# 1.1.4 and below do not use these classes.
+CONCURRENT_RUBY_VERSIONS = [
+  [nil, 2.2],
+  ['1.1.5', 2.2]
+]
+
+def gem_list(concurrent_version = nil)
+  <<-RB
+    gem 'concurrent-ruby'#{concurrent_version}
+  RB
+end
+
+create_gemfiles(CONCURRENT_RUBY_VERSIONS, gem_list)

test/multiverse/suites/concurrent_ruby/concurrent_ruby_instrumentation_test.rb

@@ -0,0 +1,121 @@
+# 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
+
+class ConcurrentRubyInstrumentationTest < Minitest::Test
+  EXPECTED_SEGMENTS_FOR_NESTED_CALLS = [
+    'Concurrent/Task',
+    'External/www.example.com/Net::HTTP/GET'
+  ]
+
+  # Helper methods
+  def future_in_transaction(&block)
+    in_transaction do
+      future = Concurrent::Promises.future { yield }
+      future.wait!
+    end
+  end
+
+  def concurrent_promises_calls_net_http_in_block
+    future_in_transaction { Net::HTTP.get(URI('http://www.example.com')) }
+  end
+
+  def simulate_error
+    future = Concurrent::Promises.future { raise 'hi' }
+    future.wait!
+  end
+
+  def assert_segment_noticed_simulated_error(txn)
+    assert_segment_noticed_error txn, /Concurrent\/Task$/, /RuntimeError/, /hi/i
+  end
+
+  def assert_expected_segments_in_transaction(txn)
+    assert_predicate (txn.segments.map(&:name) & EXPECTED_SEGMENTS_FOR_NESTED_CALLS), :any?
+  end
+
+  # Tests
+  def test_promises_future_creates_segment_with_default_name
+    txn = future_in_transaction { 'time keeps on slipping' }
+    expected_segment = 'Concurrent/Task'
+
+    assert_equal(2, txn.segments.length)
+    assert_includes txn.segments.map(&:name), expected_segment
+  end
+
+  def test_promises_future_creates_segments_for_nested_instrumented_calls
+    with_config(:'instrumentation.thread.tracing' => false) do
+      txn = concurrent_promises_calls_net_http_in_block
+
+      assert_equal(3, txn.segments.length)
+      assert_expected_segments_in_transaction(txn)
+    end
+  end
+
+  def test_promises_future_creates_segments_for_nested_instrumented_calls_with_thread_tracing_enabled
+    with_config(:'instrumentation.thread.tracing' => true) do
+      txn = concurrent_promises_calls_net_http_in_block
+
+      # We can't check the number of segments when thread tracing is enabled because we cannot rely on concurrent-ruby
+      # creating threads during this transaction, as it can reuse threads that were created previously.
+      # Instead, we check to make sure the segments that should be present are.
+      assert_expected_segments_in_transaction(txn)
+    end
+  end
+
+  def test_promises_future_captures_segment_error
+    txn = in_transaction do
+      # TODO: OLD RUBIES - RUBY_VERSION 2.2
+      # specific "begin" in block can be removed once we drop support for 2.2
+      begin
+        simulate_error
+      rescue StandardError => e
+        # NOOP -- allowing span to notice error
+      end
+    end
+
+    assert_segment_noticed_simulated_error(txn)
+  end
+
+  def test_noticed_error_at_segment_and_txn_on_error
+    txn = nil
+    begin
+      in_transaction do |test_txn|
+        txn = test_txn
+        simulate_error
+      end
+    rescue StandardError => e
+      # NOOP -- allowing span and transaction to notice error
+    end
+
+    assert_segment_noticed_simulated_error(txn)
+    assert_transaction_noticed_error txn, /RuntimeError/
+  end
+
+  def test_task_segment_has_correct_parent
+    txn = future_in_transaction { 'are you my mother?' }
+    task_segment = txn.segments.find { |n| n.name == 'Concurrent/Task' }
+
+    assert_equal task_segment.parent.name, txn.best_name
+  end
+
+  def test_segment_not_created_if_tracing_disabled
+    NewRelic::Agent::Tracer.stub :tracing_enabled?, false do
+      txn = future_in_transaction { 'the revolution will not be televised' }
+
+      assert_predicate txn.segments, :one?
+      assert_equal txn.segments.first.name, txn.best_name
+    end
+  end
+
+  def test_supportability_metric_recorded_once
+    in_transaction do
+      Concurrent::Promises.future { 'one-banana' }
+    end
+
+    in_transaction do
+      Concurrent::Promises.future { 'two-banana' }
+    end
+
+    assert_metrics_recorded(NewRelic::Agent::Instrumentation::ConcurrentRuby::SUPPORTABILITY_METRIC)
+  end
+end