Module: Datadog::Tracing::Contrib::ActiveJob::DataStreams

Defined in:
lib/datadog/tracing/contrib/active_job/data_streams.rb

Overview

Propagates Data Streams Monitoring pathway context through the serialized job payload, allowing produce (enqueue) and consume (perform) checkpoints to be connected across the process boundary.

The DSM calls are rescued in isolation so a checkpoint failure can never break a job, while a genuine error raised by the underlying serialization still propagates.

Only prepended on ActiveJob 5.0+, where serialize/deserialize are instance methods the framework routes through. ActiveJob 4.2 deserializes via a class method, so the pathway cannot be propagated there.

Instance Method Summary collapse

Instance Method Details

#deserialize(job_data) ⇒ Object 



41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/datadog/tracing/contrib/active_job/data_streams.rb', line 41

def deserialize(job_data)
  super

  return unless Datadog::DataStreams.enabled?

  begin
    Datadog::DataStreams.set_consume_checkpoint(
      type: Ext::TAG_COMPONENT,
      source: queue_name,
      auto_instrumentation: true
    ) do |key|
      job_data[key]
    end
  rescue => e
    Datadog.logger.debug { "Error setting DSM consume checkpoint: #{e.class}: #{e.message}" }
  end
end

#serialize ⇒ Object 



22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
 
# File 'lib/datadog/tracing/contrib/active_job/data_streams.rb', line 22

def serialize
  job_data = super
  return job_data unless Datadog::DataStreams.enabled?

  begin
    Datadog::DataStreams.set_produce_checkpoint(
      type: Ext::TAG_COMPONENT,
      destination: queue_name,
      auto_instrumentation: true
    ) do |key, value|
      job_data[key] = value
    end
  rescue => e
    Datadog.logger.debug { "Error setting DSM produce checkpoint: #{e.class}: #{e.message}" }
  end

  job_data
end