feat(active_job): add interrupt.active_job span event for Continuation

Record a span event on the current span when a continuable job is
interrupted, with attributes for the interrupt reason, description,
and completed steps.

Author: bdewater-thatch

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

@@ -8,6 +8,7 @@
 require_relative 'handlers/default'
 require_relative 'handlers/enqueue'
 require_relative 'handlers/perform'
+require_relative 'handlers/interrupt'
 require_relative 'handlers/resume'
 require_relative 'handlers/step'
 require_relative 'handlers/step_skipped'
@@ -30,6 +31,12 @@ module Handlers
         # - perform
         # - retry_stopped
         # - discard
+        # - step (Continuations, Rails 8.1+)
+        #
+        # The following events are recorded as span events:
+        # - resume (Continuations, Rails 8.1+)
+        # - step_skipped (Continuations, Rails 8.1+)
+        # - interrupt (Continuations, Rails 8.1+)
         #
         # Ingress and Egress spans (perform, enqueue, enqueue_at) use Messaging semantic conventions for naming the span,
         # while internal spans keep their ActiveSupport event name.
@@ -52,6 +59,7 @@ def subscribe
           enqueue_handler = Handlers::Enqueue.new(parent_span_provider, mapper, config)
           perform_handler = Handlers::Perform.new(parent_span_provider, mapper, config)
           step_handler = Handlers::Step.new(parent_span_provider, mapper, config)
+          interrupt_handler = Handlers::Interrupt.new(parent_span_provider)
           resume_handler = Handlers::Resume.new(parent_span_provider)
           step_skipped_handler = Handlers::StepSkipped.new(parent_span_provider)
 
@@ -64,7 +72,8 @@ def subscribe
             'discard' => default_handler,
             'step' => step_handler,
             'resume' => resume_handler,
-            'step_skipped' => step_skipped_handler
+            'step_skipped' => step_skipped_handler,
+            'interrupt' => interrupt_handler
           }
 
           @subscriptions = handlers_by_pattern.map do |key, handler|

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

@@ -0,0 +1,38 @@
+# frozen_string_literal: true
+
+# Copyright The OpenTelemetry Authors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+module OpenTelemetry
+  module Instrumentation
+    module ActiveJob
+      module Handlers
+        # Handles interrupt.active_job to record a span event when a continuable
+        # job is interrupted for resumption
+        class Interrupt
+          def initialize(parent_span_provider)
+            @parent_span_provider = parent_span_provider
+          end
+
+          # @param _name [String] of the Event (unused)
+          # @param _id [String] of the event (unused)
+          # @param payload [Hash] containing job run information
+          def start(_name, _id, payload)
+            span = @parent_span_provider.current_span
+            return unless span.recording?
+
+            attributes = {
+              'messaging.active_job.continuation.reason' => payload[:reason].to_s,
+              'messaging.active_job.continuation.description' => payload[:description]
+            }
+
+            span.add_event('interrupt', attributes: attributes)
+          end
+
+          def finish(_name, _id, _payload); end
+        end
+      end
+    end
+  end
+end

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

@@ -0,0 +1,91 @@
+# frozen_string_literal: true
+
+# Copyright The OpenTelemetry Authors
+#
+# SPDX-License-Identifier: Apache-2.0
+
+require 'test_helper'
+
+require_relative '../../../../../lib/opentelemetry/instrumentation/active_job'
+
+require 'active_job/continuation/test_helper' if defined?(ActiveJob::Continuable)
+
+describe OpenTelemetry::Instrumentation::ActiveJob::Handlers::Interrupt do
+  let(:instrumentation) { OpenTelemetry::Instrumentation::ActiveJob::Instrumentation.instance }
+  let(:config) { { propagation_style: :link, span_naming: :queue } }
+  let(:exporter) { EXPORTER }
+  let(:spans) { exporter.finished_spans }
+  let(:process_span) { spans.find { |s| s.name == 'default process' } }
+
+  before do
+    skip 'Requires ActiveJob::Continuable (Rails 8.1+)' unless defined?(ActiveJob::Continuable)
+
+    singleton_class.include ActiveJob::Continuation::TestHelper
+
+    OpenTelemetry::Instrumentation::ActiveJob::Handlers.unsubscribe
+    instrumentation.instance_variable_set(:@config, config)
+    instrumentation.instance_variable_set(:@installed, false)
+
+    instrumentation.install(config)
+    ActiveJob::Base.queue_adapter = :test
+
+    exporter.reset
+  end
+
+  after do
+    ActiveJob::Base.queue_adapter = :inline
+  end
+
+  describe 'when a job is interrupted between steps' do
+    it 'adds an interrupt event to the process span' do
+      ContinuableJob.perform_later
+      interrupt_job_after_step(ContinuableJob, :first_step) { perform_enqueued_jobs }
+
+      interrupt_events = process_span.events&.select { |e| e.name == 'interrupt' } || []
+
+      _(interrupt_events.length).must_equal 1
+    end
+
+    it 'includes the interrupt reason' do
+      ContinuableJob.perform_later
+      interrupt_job_after_step(ContinuableJob, :first_step) { perform_enqueued_jobs }
+
+      interrupt_event = process_span.events.find { |e| e.name == 'interrupt' }
+
+      _(interrupt_event.attributes['messaging.active_job.continuation.reason']).must_equal 'stopping'
+    end
+
+    it 'includes the description' do
+      ContinuableJob.perform_later
+      interrupt_job_after_step(ContinuableJob, :first_step) { perform_enqueued_jobs }
+
+      interrupt_event = process_span.events.find { |e| e.name == 'interrupt' }
+
+      _(interrupt_event.attributes['messaging.active_job.continuation.description']).wont_be_nil
+    end
+  end
+
+  describe 'when a job is interrupted during a step' do
+    it 'adds an interrupt event to the process span, not the step span' do
+      ContinuableWithCursorJob.perform_later
+      interrupt_job_during_step(ContinuableWithCursorJob, :process_items, cursor: 2) { perform_enqueued_jobs }
+
+      step_span = spans.find { |s| s.name == 'default process_items' }
+      step_interrupt_events = step_span&.events&.select { |e| e.name == 'interrupt' } || []
+      process_interrupt_events = process_span.events&.select { |e| e.name == 'interrupt' } || []
+
+      _(step_interrupt_events).must_be(:empty?)
+      _(process_interrupt_events.length).must_equal 1
+    end
+  end
+
+  describe 'when a job completes without interruption' do
+    it 'does not add an interrupt event' do
+      ContinuableJob.perform_later
+      perform_enqueued_jobs
+
+      interrupt_events = process_span.events&.select { |e| e.name == 'interrupt' } || []
+      _(interrupt_events).must_be(:empty?)
+    end
+  end
+end