feat(active-job): Add performance metrics to process span

instrumentation/active_job/README.md

@@ -42,6 +42,10 @@ The instrumentation supports the following configuration options:
   the end of each job execution. This is recommended for job systems that fork
   worker processes, such as Resque.
   - Default: `false`
+- **enable_performance_metrics:** If enabled, process spans include job
+  execution performance attributes for allocations, CPU time, CPU utilization,
+  and GC time.
+  - Default: `false`
 - **propagation_style:** Controls how job execution traces are related to the
   trace where the job was enqueued.
   - `:link` – The job runs in a separate trace, with its initial span linked to
@@ -96,6 +100,15 @@ For jobs including the `ActiveJob::Continuable` module, the following attributes
 | `messaging.active_job.step.result` | String | Static value set to `interrupted` if the job was interrupted |
 | `messaging.active_job.step.cursor` | String | The persisted value after calling `step.set!` or `step.advance!` |
 
+When `enable_performance_metrics` is enabled, process spans also include:
+
+| Attribute Name | Type | Notes |
+| - | - | - |
+| `rails.memory.allocations` | Number | Object allocations during job execution |
+| `rails.cpu.time` | Number | CPU time during job execution, in milliseconds |
+| `rails.cpu.utilization` | Number | CPU time divided by wall-clock time during job execution |
+| `rails.gc.time` | Number | GC time during job execution, in milliseconds |
+
 ## Differences between ActiveJob versions
 
 ### ActiveJob 6.1

instrumentation/active_job/lib/opentelemetry/instrumentation/active_job/handlers/perform.rb

@@ -37,6 +37,39 @@ def start_span(name, _id, payload)
             { span: span, ctx_token: attach_consumer_context(span, parent_context) }
           end
 
+          # Overrides `Default#start` to optionally snapshot performance metrics at job start
+          #
+          # @param name [String] of the Event
+          # @param id [String] of the event
+          # @param payload [Hash] containing job run information
+          # @return [Hash] the payload passed as a method argument
+          def start(name, id, payload)
+            payload[:__otel] = start_span(name, id, payload)
+            payload[:__otel_metrics] = snapshot_metrics if @config[:enable_performance_metrics]
+            payload
+          rescue StandardError => e
+            OpenTelemetry.handle_error(exception: e)
+          end
+
+          # Overrides `Default#finish` to optionally record performance metrics on the span
+          #
+          # @param _name [String] of the Event (unused)
+          # @param _id [String] of the event (unused)
+          # @param payload [Hash] containing job run information
+          def finish(_name, _id, payload)
+            otel = payload.delete(:__otel)
+            metrics_start = payload.delete(:__otel_metrics)
+            span = otel&.fetch(:span)
+            token = otel&.fetch(:ctx_token)
+
+            record_metrics(span, metrics_start) if span && metrics_start
+            on_exception(payload[:error] || payload[:exception_object], span)
+          rescue StandardError => e
+            OpenTelemetry.handle_error(exception: e)
+          ensure
+            finish_span(span, token)
+          end
+
           # This method attaches a span to multiple contexts:
           # 1. Registers the ingress span as the top level ActiveJob span.
           #    This is used later to enrich the ingress span in children, e.g. setting span status to error when a child event like `discard` terminates due to an error
@@ -50,6 +83,70 @@ def attach_consumer_context(span, parent_context)
 
             OpenTelemetry::Context.attach(internal_context)
           end
+
+          private
+
+          def snapshot_metrics
+            {
+              monotonic_time: Process.clock_gettime(Process::CLOCK_MONOTONIC, :float_millisecond),
+              cpu_time: now_cpu,
+              gc_time: now_gc,
+              allocations: now_allocations
+            }
+          end
+
+          def record_metrics(span, start)
+            finish = snapshot_metrics
+
+            duration = finish[:monotonic_time] - start[:monotonic_time]
+            cpu_time = finish[:cpu_time] - start[:cpu_time]
+            cpu_utilization = duration.positive? ? (cpu_time / duration).clamp(0.0, 1.0) : 0.0
+            gc_time = (finish[:gc_time] - start[:gc_time]) / 1_000_000.0
+            allocations = finish[:allocations] - start[:allocations]
+
+            span.set_attribute('rails.cpu.time', cpu_time)
+            span.set_attribute('rails.cpu.utilization', cpu_utilization)
+            span.set_attribute('rails.gc.time', gc_time)
+            span.set_attribute('rails.memory.allocations', allocations)
+          end
+
+          if GC.respond_to?(:total_time)
+            # Returns total GC time in nanoseconds.
+            def now_gc
+              GC.total_time
+            end
+          else
+            # Returns zero when total GC time is unavailable.
+            def now_gc
+              0
+            end
+          end
+
+          begin
+            Process.clock_gettime(Process::CLOCK_THREAD_CPUTIME_ID, :float_millisecond)
+
+            # Returns current thread CPU time in milliseconds.
+            def now_cpu
+              Process.clock_gettime(Process::CLOCK_THREAD_CPUTIME_ID, :float_millisecond)
+            end
+          rescue StandardError
+            # Returns zero when current thread CPU time is unavailable.
+            def now_cpu
+              0.0
+            end
+          end
+
+          if GC.stat.key?(:total_allocated_objects)
+            # Returns total allocated objects.
+            def now_allocations
+              GC.stat(:total_allocated_objects)
+            end
+          else
+            # Returns zero when total allocated objects is unavailable.
+            def now_allocations
+              0
+            end
+          end
         end
       end
     end

instrumentation/active_job/lib/opentelemetry/instrumentation/active_job/instrumentation.rb

@@ -52,9 +52,13 @@ class Instrumentation < OpenTelemetry::Instrumentation::Base
         # spans will always be children of the enqueueing spans. This is due to the way
         # ActiveJob immediately executes jobs during the process of "enqueueing" jobs when
         # using the `:inline` adapter.
+        #
+        # enable_performance_metrics: when `true`, process spans will include
+        #   allocations, CPU time, CPU utilization, and GC time attributes.
         option :propagation_style, default: :link, validate: %i[link child none]
         option :force_flush, default: false, validate: :boolean
         option :span_naming, default: :queue, validate: %i[job_class queue]
+        option :enable_performance_metrics, default: false, validate: :boolean
 
         private
 

instrumentation/active_job/test/opentelemetry/instrumentation/active_job/handlers/perform_test.rb

@@ -101,6 +101,55 @@
   end
 
   describe 'attributes' do
+    describe 'performance metrics' do
+      let(:performance_metric_names) { ['rails.cpu.time', 'rails.cpu.utilization', 'rails.gc.time', 'rails.memory.allocations'] }
+
+      it 'does not record performance metrics on the process span by default' do
+        TestJob.perform_now
+
+        performance_metric_names.each do |metric|
+          _(process_span.attributes[metric]).must_be_nil
+        end
+      end
+
+      describe 'when enable_performance_metrics is true' do
+        let(:config) { { propagation_style: :link, span_naming: :queue, enable_performance_metrics: true } }
+
+        it 'records allocations on the process span' do
+          TestJob.perform_now
+
+          _(process_span.attributes['rails.memory.allocations']).must_be :>=, 0
+        end
+
+        it 'records cpu_time on the process span' do
+          TestJob.perform_now
+
+          _(process_span.attributes['rails.cpu.time']).must_be :>=, 0.0
+        end
+
+        it 'records cpu_utilization on the process span' do
+          TestJob.perform_now
+
+          _(process_span.attributes['rails.cpu.utilization']).must_be :>=, 0.0
+          _(process_span.attributes['rails.cpu.utilization']).must_be :<=, 1.0
+        end
+
+        it 'records gc_time on the process span' do
+          TestJob.perform_now
+
+          _(process_span.attributes['rails.gc.time']).must_be :>=, 0.0
+        end
+
+        it 'does not record performance metrics on the publish span' do
+          TestJob.perform_later
+
+          performance_metric_names.each do |metric|
+            _(publish_span.attributes[metric]).must_be_nil
+          end
+        end
+      end
+    end
+
     describe 'active_job.priority' do
       it 'is unset for unprioritized jobs' do
         TestJob.perform_later