Class: Datadog::Tracing::Workers::AsyncTraceWriter

Inherits:
TraceWriter show all
Includes:
Core::Workers::Polling, Core::Workers::Queue
Defined in:
lib/datadog/tracing/workers/trace_writer.rb

Overview

Writes traces to transport asynchronously, using a thread & buffer.

Constant Summary collapse

DEFAULT_BUFFER_MAX_SIZE = 
1000
FORK_POLICY_ASYNC = 
:async
FORK_POLICY_SYNC = 
:sync

Constants included from Core::Workers::Polling

Core::Workers::Polling::DEFAULT_SHUTDOWN_TIMEOUT

Instance Attribute Summary collapse

Attributes included from Core::Workers::Queue

#buffer

Attributes inherited from TraceWriter

#transport

Attributes inherited from Core::Worker

#task

Instance Method Summary collapse

Methods included from Core::Workers::Polling

#enabled=, #enabled?, included

Methods included from Core::Workers::Queue

included

Methods inherited from TraceWriter

#flush_completed, #flush_traces, #process_traces, #write_traces

Constructor Details

#initialize(options = {}) ⇒ AsyncTraceWriter

Returns a new instance of AsyncTraceWriter.



90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
 
# File 'lib/datadog/tracing/workers/trace_writer.rb', line 90

def initialize(options = {})
  # Workers::TraceWriter settings
  super

  # Workers::Polling settings
  self.enabled = options.fetch(:enabled, true)

  # Workers::Async::Thread settings
  @async = true
  self.fork_policy = options.fetch(:fork_policy, FORK_POLICY_ASYNC)

  # Workers::IntervalLoop settings
  self.loop_base_interval = options[:interval] if options.key?(:interval)
  self.loop_back_off_ratio = options[:back_off_ratio] if options.key?(:back_off_ratio)
  self.loop_back_off_max = options[:back_off_max] if options.key?(:back_off_max)

  # Workers::Queue settings
  @buffer_size = options.fetch(:buffer_size, DEFAULT_BUFFER_MAX_SIZE)
  self.buffer = TraceBuffer.new(@buffer_size)

  @shutdown_timeout = options.fetch(:shutdown_timeout, Core::Workers::Polling::DEFAULT_SHUTDOWN_TIMEOUT)
end

Instance Attribute Details

#async=(value) ⇒ Object (writeonly)

Sets the attribute async

Parameters:

  • value —

    the value to set the attribute async to.



87
88
89
 
# File 'lib/datadog/tracing/workers/trace_writer.rb', line 87

def async=(value)
  @async = value
end

Instance Method Details

#after_fork ⇒ Object 



170
171
172
173
174
175
176
177
178
179
 
# File 'lib/datadog/tracing/workers/trace_writer.rb', line 170

def after_fork
  # In multiprocess environments, forks will share the same buffer until its written to.
  # A.K.A. copy-on-write. We don't want forks to write traces generated from another process.
  # Instead, we reset it after the fork. (Make sure any enqueue operations happen after this.)
  self.buffer = TraceBuffer.new(@buffer_size)

  # Switch to synchronous mode if configured to do so.
  # In some cases synchronous writing is preferred because the fork will be short lived.
  @async = false if @writer_fork_policy == FORK_POLICY_SYNC
end

#async? ⇒ Boolean

Returns:

  • (Boolean) 


146
147
148
 
# File 'lib/datadog/tracing/workers/trace_writer.rb', line 146

def async?
  @async == true
end

#dequeue ⇒ Object 



135
136
137
138
139
 
# File 'lib/datadog/tracing/workers/trace_writer.rb', line 135

def dequeue
  # Wrap results in Array because they are
  # splatted as args against TraceWriter#perform.
  [buffer.pop]
end

#enqueue(trace) ⇒ Object 



131
132
133
 
# File 'lib/datadog/tracing/workers/trace_writer.rb', line 131

def enqueue(trace)
  buffer.push(trace)
end

#fork_policy=(policy) ⇒ Object 



150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
 
# File 'lib/datadog/tracing/workers/trace_writer.rb', line 150

def fork_policy=(policy)
  # Translate to Workers::Async::Thread policy
  thread_fork_policy =  case policy
                        when Core::Workers::Async::Thread::FORK_POLICY_STOP
                          policy
                        when FORK_POLICY_SYNC
                          # Stop the async thread because the writer
                          # will bypass and run synchronously.
                          Core::Workers::Async::Thread::FORK_POLICY_STOP
                        else
                          Core::Workers::Async::Thread::FORK_POLICY_RESTART
                        end

  # Update thread fork policy
  super(thread_fork_policy)

  # Update local policy
  @writer_fork_policy = policy
end

#perform(traces) ⇒ Object

NOTE: #perform is wrapped by other modules:

Polling --> Async --> IntervalLoop --> AsyncTraceWriter --> TraceWriter

WARNING: This method breaks the Liskov Substitution Principle – TraceWriter#perform is spec’d to return the result from the writer, whereas this method always returns nil.



118
119
120
121
122
123
124
 
# File 'lib/datadog/tracing/workers/trace_writer.rb', line 118

def perform(traces)
  super(traces).tap do |responses|
    loop_back_off! if responses.find(&:server_error?)
  end

  nil
end

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



126
127
128
129
 
# File 'lib/datadog/tracing/workers/trace_writer.rb', line 126

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

#work_pending? ⇒ Boolean

Are there more traces to be processed next?

Returns:

  • (Boolean) 


142
143
144
 
# File 'lib/datadog/tracing/workers/trace_writer.rb', line 142

def work_pending?
  !buffer.empty?
end

#write(trace) ⇒ Object

WARNING: This method breaks the Liskov Substitution Principle – TraceWriter#write is spec’d to return the result from the writer, whereas this method returns something else when running in async mode.



183
184
185
186
187
188
189
190
191
192
 
# File 'lib/datadog/tracing/workers/trace_writer.rb', line 183

def write(trace)
  # Start worker thread. If the process has forked, it will trigger #after_fork to
  # reconfigure the worker accordingly.
  # NOTE: It's important we do this before queuing or it will drop the current trace,
  #       because #after_fork resets the buffer.
  perform

  # Queue the trace if running asynchronously, otherwise short-circuit and write it directly.
  async? ? enqueue(trace) : write_traces([trace])
end