Class: Datadog::DataStreams::Processor

Inherits:
Core::Worker show all
Includes:
Core::Workers::Polling
Defined in:
lib/datadog/data_streams/processor.rb

Overview

Processor for Data Streams Monitoring This class is responsible for collecting and reporting pathway stats Periodically (every interval, 10 seconds by default) flushes stats to the Datadog agent.

Constant Summary collapse

PROPAGATION_KEY = 
"dd-pathway-ctx-base64"
DEFAULT_BUFFER_SIZE =

Default buffer size for lock-free event queue Set to handle high-throughput scenarios (e.g., 10k events/sec for 10s interval)

100_000

Constants included from Core::Workers::Polling

Core::Workers::Polling::DEFAULT_SHUTDOWN_TIMEOUT

Instance Attribute Summary collapse

Attributes inherited from Core::Worker

#task

Instance Method Summary collapse

Methods included from Core::Workers::Polling

#enabled=, #enabled?, included, #stop

Constructor Details

#initialize(interval:, logger:, settings:, agent_settings:, agent_info:, buffer_size: DEFAULT_BUFFER_SIZE) ⇒ Processor

Initialize the Data Streams Monitoring processor

Parameters:

  • Flush interval in seconds (e.g., 10.0 for 10 seconds)

  • Logger instance for debugging

  • Global configuration settings

  • Agent connection settings

  • Agent capability information

  • (defaults to: DEFAULT_BUFFER_SIZE)

    Size of the lock-free event buffer for async stat collection (default: DEFAULT_BUFFER_SIZE). Higher values support more throughput but use more memory.

Raises:

  • if DDSketch is not available on this platform



43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/datadog/data_streams/processor.rb', line 43

def initialize(interval:, logger:, settings:, agent_settings:, agent_info:, buffer_size: DEFAULT_BUFFER_SIZE)
  raise UnsupportedError, "DDSketch is not supported" unless Datadog::Core::DDSketch.supported?

  @settings = settings
  @agent_settings = agent_settings
  @agent_info = agent_info
  @logger = logger

  now = Core::Utils::Time.now
  @pathway_context = PathwayContext.new(
    hash_value: 0,
    pathway_start: now,
    current_edge_start: now
  )
  @bucket_size_ns = (interval * 1e9).to_i
  @buckets = {}
  @consumer_stats = []
  @stats_mutex = Mutex.new
  @event_buffer = Core::Buffer::CRuby.new(buffer_size)

  super()
  self.loop_base_interval = interval
  # Without this, a preload-then-fork deployment model (e.g. Puma cluster mode with
  # preload_app!, or Sidekiq Enterprise's sidekiqswarm with SIDEKIQ_PRELOAD_APP) leaves
  # every forked child with a dead flush thread: Ruby threads don't survive fork, and the
  # default FORK_POLICY_STOP means #perform would otherwise just mark the worker stopped
  # instead of restarting it. See #restart_flush_thread.
  self.fork_policy = Core::Workers::Async::Thread::FORK_POLICY_RESTART

  perform
end

Instance Attribute Details

#bucket_size_ns ⇒ Object (readonly)

Returns the value of attribute bucket_size_ns.



31
32
33
 
# File 'lib/datadog/data_streams/processor.rb', line 31

def bucket_size_ns
  @bucket_size_ns
end

#buckets ⇒ Object (readonly)

Returns the value of attribute buckets.



31
32
33
 
# File 'lib/datadog/data_streams/processor.rb', line 31

def buckets
  @buckets
end

#pathway_context ⇒ Object (readonly)

Returns the value of attribute pathway_context.



31
32
33
 
# File 'lib/datadog/data_streams/processor.rb', line 31

def pathway_context
  @pathway_context
end

Instance Method Details

#perform ⇒ Object

Called periodically by the worker to flush stats to the agent



169
170
171
172
173
 
# File 'lib/datadog/data_streams/processor.rb', line 169

def perform
  process_events
  flush_stats
  true
end

#restart_flush_thread ⇒ Object

Restarts the flush thread if a fork has been detected since this processor was created. Called by Components#after_fork. Safe to call even when no fork occurred: #perform (via Workers::Async::Thread) only restarts the worker when forked? is true. Deliberately not named after_fork: Workers::Async::Thread already defines a protected after_fork template method that #perform's internal restart path (+restart_after_fork+) calls on every restart -- overriding it here would recurse.



81
82
83
84
 
# File 'lib/datadog/data_streams/processor.rb', line 81

def restart_flush_thread
  discard_inherited_state_after_fork! if forked?
  perform
end

#set_consume_checkpoint(type:, source:, manual_checkpoint: true, tags: {}) {|key| ... } ⇒ String

Set a consume checkpoint

Parameters:

  • The type of the checkpoint (e.g., 'kafka', 'kinesis', 'sns')

  • The source (e.g., topic, exchange, stream name)

  • (defaults to: true)

    Whether this checkpoint was manually set (default: true)

  • (defaults to: {})

    Additional tags to include

Yields:

  • (key) —

    Block to extract context from carrier

Returns:

  • Base64 encoded pathway context



151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
 
# File 'lib/datadog/data_streams/processor.rb', line 151

def set_consume_checkpoint(type:, source:, manual_checkpoint: true, tags: {}, &block)
  if block
    pathway_ctx = yield(PROPAGATION_KEY)
    if pathway_ctx
      decoded_ctx = decode_pathway_b64(pathway_ctx)
      set_pathway_context(decoded_ctx)
    end
  end

  checkpoint_tags = ["type:#{type}", "topic:#{source}", "direction:in"]
  checkpoint_tags << "manual_checkpoint:true" if manual_checkpoint
  checkpoint_tags.concat(tags.map { |k, v| "#{k}:#{v}" }) unless tags.empty?

  span = Datadog::Tracing.active_span
  set_checkpoint(tags: checkpoint_tags, span: span)
end

#set_produce_checkpoint(type:, destination:, manual_checkpoint: true, tags: {}) {|key, value| ... } ⇒ String

Set a produce checkpoint

Parameters:

  • The type of the checkpoint (e.g., 'kafka', 'kinesis', 'sns')

  • The destination (e.g., topic, exchange, stream name)

  • (defaults to: true)

    Whether this checkpoint was manually set (default: true)

  • (defaults to: {})

    Additional tags to include

Yields:

  • (key, value) —

    Block to inject context into carrier

Returns:

  • Base64 encoded pathway context



131
132
133
134
135
136
137
138
139
140
141
142
 
# File 'lib/datadog/data_streams/processor.rb', line 131

def set_produce_checkpoint(type:, destination:, manual_checkpoint: true, tags: {}, &block)
  checkpoint_tags = ["type:#{type}", "topic:#{destination}", "direction:out"]
  checkpoint_tags << "manual_checkpoint:true" if manual_checkpoint
  checkpoint_tags.concat(tags.map { |k, v| "#{k}:#{v}" }) unless tags.empty?

  span = Datadog::Tracing.active_span
  pathway = set_checkpoint(tags: checkpoint_tags, span: span)

  yield(PROPAGATION_KEY, pathway) if pathway && block

  pathway
end

#track_kafka_consume(topic, partition, offset, now) ⇒ Boolean

Track Kafka message consumption for consumer lag monitoring

Parameters:

  • The Kafka topic name

  • The partition number

  • The offset of the consumed message

  • Timestamp

Returns:

  • true if tracking succeeded



111
112
113
114
115
116
117
118
119
120
121
122
 
# File 'lib/datadog/data_streams/processor.rb', line 111

def track_kafka_consume(topic, partition, offset, now)
  @event_buffer.push(
    {
      type: :kafka_consume,
      topic: topic,
      partition: partition,
      offset: offset,
      timestamp: now,
    }
  )
  true
end

#track_kafka_produce(topic, partition, offset, now) ⇒ Boolean

Track Kafka produce offset for lag monitoring

Parameters:

  • The Kafka topic name

  • The partition number

  • The offset of the produced message

  • Timestamp

Returns:

  • true if tracking succeeded



92
93
94
95
96
97
98
99
100
101
102
103
 
# File 'lib/datadog/data_streams/processor.rb', line 92

def track_kafka_produce(topic, partition, offset, now)
  @event_buffer.push(
    {
      type: :kafka_produce,
      topic: topic,
      partition: partition,
      offset: offset,
      timestamp_ns: (now.to_f * 1e9).to_i,
    }
  )
  true
end