Class: Datadog::Core::Telemetry::Worker Private

Inherits:
Object
  • Object
show all
Includes:
Workers::Polling, Workers::Queue
Defined in:
lib/datadog/core/telemetry/worker.rb

Overview

This class is part of a private API. You should avoid using this class if possible, as it may be removed or be changed in the future.

Accumulates events and sends them to the API at a regular interval, including heartbeat event.

Constant Summary collapse

DEFAULT_BUFFER_MAX_SIZE =

This constant is part of a private API. You should avoid using this constant if possible, as it may be removed or be changed in the future. 

1000
APP_STARTED_EVENT_RETRIES =

This constant is part of a private API. You should avoid using this constant if possible, as it may be removed or be changed in the future. 

10

Constants included from Workers::Polling

Workers::Polling::DEFAULT_SHUTDOWN_TIMEOUT

Instance Attribute Summary collapse

Attributes included from Workers::Queue

#buffer

Instance Method Summary collapse

Methods included from Workers::Polling

#enabled=, #enabled?, included

Methods included from Workers::Queue

included

Constructor Details

#initialize(heartbeat_interval_seconds:, metrics_aggregation_interval_seconds:, emitter:, metrics_manager:, dependency_collection:, logger:, enabled: true, shutdown_timeout: Workers::Polling::DEFAULT_SHUTDOWN_TIMEOUT, buffer_size: DEFAULT_BUFFER_MAX_SIZE) ⇒ Worker

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a new instance of Worker.



23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
 
# File 'lib/datadog/core/telemetry/worker.rb', line 23

def initialize(
  heartbeat_interval_seconds:,
  metrics_aggregation_interval_seconds:,
  emitter:,
  metrics_manager:,
  dependency_collection:,
  logger:,
  enabled: true,
  shutdown_timeout: Workers::Polling::DEFAULT_SHUTDOWN_TIMEOUT,
  buffer_size: DEFAULT_BUFFER_MAX_SIZE
)
  @emitter = emitter
  @metrics_manager = metrics_manager
  @dependency_collection = dependency_collection
  @logger = logger

  @ticks_per_heartbeat = (heartbeat_interval_seconds / metrics_aggregation_interval_seconds).to_i
  @current_ticks = 0

  # Workers::Polling settings
  self.enabled = enabled
  # Workers::IntervalLoop settings
  self.loop_base_interval = metrics_aggregation_interval_seconds
  self.fork_policy = Core::Workers::Async::Thread::FORK_POLICY_RESTART

  @shutdown_timeout = shutdown_timeout
  @buffer_size = buffer_size

  initialize_state
end

Instance Attribute Details

#emitter ⇒ Object (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



63
64
65
 
# File 'lib/datadog/core/telemetry/worker.rb', line 63

def emitter
  @emitter
end

#initial_event ⇒ Object (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



62
63
64
 
# File 'lib/datadog/core/telemetry/worker.rb', line 62

def initial_event
  @initial_event
end

#initial_event_once ⇒ Object (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



61
62
63
 
# File 'lib/datadog/core/telemetry/worker.rb', line 61

def initial_event_once
  @initial_event_once
end

#logger ⇒ Object (readonly)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



60
61
62
 
# File 'lib/datadog/core/telemetry/worker.rb', line 60

def logger
  @logger
end

Instance Method Details

#enqueue(event) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns true if event was enqueued, nil if not. While returning false may seem more reasonable, the only reason for not enqueueing event (presently) is that telemetry is disabled altogether, and in this case other methods return nil.



89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
 
# File 'lib/datadog/core/telemetry/worker.rb', line 89

def enqueue(event)
  return unless enabled?

  # Start the worker if needed, including in forked children.
  # Needs to be done before pushing to buffer since perform
  # may invoke after_fork handler which resets the buffer.
  #
  # Telemetry is special in that it permits events to be submitted
  # to the worker with the worker not running, and the worker is
  # explicitly started later (to maintain proper initialization order).
  # Thus here we can't just call perform unconditionally and must
  # check if the worker is supposed to be running, and only call
  # perform in that case.
  if worker && !worker.alive?
    perform
  end

  buffer.push(event)
  true
end

#failed_initial_event? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns:

  • (Boolean) 


114
115
116
 
# File 'lib/datadog/core/telemetry/worker.rb', line 114

def failed_initial_event?
  initial_event_once.failed?
end

#flush ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Wait for the worker to send out all events that have already been queued, up to 15 seconds. Returns whether all events have been flushed.



127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
 
# File 'lib/datadog/core/telemetry/worker.rb', line 127

def flush
  return true unless enabled? || !run_loop?

  started = Utils::Time.get_time
  loop do
    # The AppStarted event is triggered by the worker itself,
    # from the worker thread. As such the main thread has no way
    # to delay itself until that event is queued and we need some
    # way to wait until that event is sent out to assert on it in
    # the test suite. Check the run once flag which *should*
    # indicate the event has been queued (at which point our queue
    # depth check should waint until it's sent).
    # This is still a hack because the flag can be overridden
    # either way with or without the event being sent out.
    # Note that if the AppStarted sending fails, this check
    # will return false and flushing will be blocked until the
    # 15 second timeout.
    # Note that the first wait interval between telemetry event
    # sending is 10 seconds, the timeout needs to be strictly
    # greater than that.
    return true if buffer.empty? && !in_iteration? && sent_initial_event?

    sleep 0.5

    return false if Utils::Time.get_time - started > 15
  end
end

#need_initial_event? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns:

  • (Boolean) 


118
119
120
 
# File 'lib/datadog/core/telemetry/worker.rb', line 118

def need_initial_event?
  !sent_initial_event? && !failed_initial_event?
end

#sent_initial_event? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns:

  • (Boolean) 


110
111
112
 
# File 'lib/datadog/core/telemetry/worker.rb', line 110

def sent_initial_event?
  initial_event_once.success?
end

#start(initial_event) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns true if worker thread is successfully started, false if worker thread was not started but telemetry is enabled, nil if telemetry is disabled.



68
69
70
71
72
73
74
75
76
77
 
# File 'lib/datadog/core/telemetry/worker.rb', line 68

def start(initial_event)
  return unless enabled?

  @initial_event = initial_event

  # starts async worker
  # perform should return true if thread was actually started,
  # false otherwise
  perform
end

#stop(force_stop = false, timeout = @shutdown_timeout) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future. 



79
80
81
82
83
 
# File 'lib/datadog/core/telemetry/worker.rb', line 79

def stop(force_stop = false, timeout = @shutdown_timeout)
  buffer.close if running?

  super
end