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.



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

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.



85
86
87
 
# File 'lib/datadog/tracing/workers/trace_writer.rb', line 85

def async=(value)
  @async = value
end

Instance Method Details

#after_fork ⇒ Object 



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

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) 


144
145
146
 
# File 'lib/datadog/tracing/workers/trace_writer.rb', line 144

def async?
  @async == true
end

#dequeue ⇒ Object 



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

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

#enqueue(trace) ⇒ Object 



129
130
131
 
# File 'lib/datadog/tracing/workers/trace_writer.rb', line 129

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

#fork_policy=(policy) ⇒ Object 



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

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.



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

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 



124
125
126
127
 
# File 'lib/datadog/tracing/workers/trace_writer.rb', line 124

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) 


140
141
142
 
# File 'lib/datadog/tracing/workers/trace_writer.rb', line 140

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.



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

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