Module: Bitfab::Otel

Defined in:
lib/bitfab/otel.rb

Overview

OpenTelemetry delivery for Bitfab spans and trace completions.

Bitfab payloads ride as OTLP span attributes (bitfab.operation + bitfab.payload), so a batch travels to Bitfab as OTLP/JSON without the payload itself changing shape.

Defined Under Namespace

Modules: Encoder Classes: BatchTransport, CarrierMeta, CarrierRef, DeliveryError, DeliveryTrackingExporter, DirectExporter, EncodedSpan, RequestBatch, RequestEnvelope

Constant Summary collapse

OPERATION_ATTRIBUTE = 
"bitfab.operation"
PAYLOAD_ATTRIBUTE = 
"bitfab.payload"
MAX_REQUEST_BYTES_ENV = 
"BITFAB_OTEL_MAX_REQUEST_BYTES"
EXPORT_CONCURRENCY_ENV = 
"BITFAB_OTEL_EXPORT_CONCURRENCY"
MAX_EXPORT_REQUEST_BYTES = 
3_000_000
MAX_DECOMPRESSED_REQUEST_BYTES = 
8_000_000
MAX_QUEUE_SIZE = 
8_192
DIRECT_MAX_EXPORT_BATCH_SIZE = 
512
DIRECT_MAX_REQUEST_BATCH_SIZE = 
8
DEFAULT_EXPORT_CONCURRENCY = 
32
MAX_EXPORT_CONCURRENCY = 
64
SCHEDULE_DELAY_MILLIS = 
5_000
EXPORT_TIMEOUT_MILLIS = 
30_000
EXPORT_RETRIES = 
3
RETRY_BASE_DELAY_MILLIS = 
100
RETRY_BACKOFF_CEILING_MILLIS =

Ceiling on the exponential growth of our OWN backoff. It does not bound a wait the server asked for: OTLP says to honor Retry-After, and calls data dropped while throttled the outcome to avoid. What bounds an honored wait is the export budget, since the processor kills a longer export.

5_000
SPAN_SEPARATOR_BYTES =

The comma that joins adjacent spans in the request's span list.

1
SUCCESS = 
OpenTelemetry::SDK::Trace::Export::SUCCESS
FAILURE = 
OpenTelemetry::SDK::Trace::Export::FAILURE
SPAN_KINDS = 
{
  internal: 1,
  server: 2,
  client: 3,
  producer: 4,
  consumer: 5
}.freeze
INVALID_SPAN_ID = 
("\0" * 8).b

Class Method Summary collapse

Class Method Details

.create_transport(direct_sender:, on_delivered: nil) ⇒ Object 



132
133
134
135
136
137
138
139
 
# File 'lib/bitfab/otel.rb', line 132

def create_transport(direct_sender:, on_delivered: nil)
  BatchTransport.new(
    direct_sender:,
    on_delivered:,
    export_concurrency: export_concurrency_from_env,
    max_request_bytes: max_request_bytes_from_env
  )
end

.ensure_process_state ⇒ Object

Ruby has no after-fork hook a library can register (Python's os.register_at_fork has no equivalent), so every entry point that touches module state checks the pid first. Inherited mutexes need no special handling: CRuby abandons a mutex held by a thread the child did not inherit, so the child sees it unlocked.



194
195
196
197
198
199
 
# File 'lib/bitfab/otel.rb', line 194

def ensure_process_state
  @state_pid ||= Process.pid
  return if @state_pid == Process.pid

  reset_state_after_fork
end

.export_concurrency_from_env ⇒ Object 



117
118
119
120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/bitfab/otel.rb', line 117

def export_concurrency_from_env
  raw = ENV[EXPORT_CONCURRENCY_ENV]
  return DEFAULT_EXPORT_CONCURRENCY if raw.nil?

  value = Integer(raw, exception: false) || 0
  return value if value.positive? && value <= MAX_EXPORT_CONCURRENCY

  Bitfab.warn_once(
    "otel-export-concurrency-invalid",
    "#{EXPORT_CONCURRENCY_ENV} must be a positive integer no greater than " \
    "#{MAX_EXPORT_CONCURRENCY}; using #{DEFAULT_EXPORT_CONCURRENCY}"
  )
  DEFAULT_EXPORT_CONCURRENCY
end

.flush_transports(timeout = 30.0) ⇒ Object 



151
152
153
154
155
156
157
 
# File 'lib/bitfab/otel.rb', line 151

def flush_transports(timeout = 30.0)
  ensure_process_state
  deadline = monotonic_now + [timeout, 0].max
  current_process_transports.reduce(true) do |succeeded, transport|
    transport.flush([deadline - monotonic_now, 0].max) && succeeded
  end
end

.max_request_bytes_from_env ⇒ Object 



102
103
104
105
106
107
108
109
110
111
112
113
114
115
 
# File 'lib/bitfab/otel.rb', line 102

def max_request_bytes_from_env
  raw = ENV[MAX_REQUEST_BYTES_ENV]
  return MAX_EXPORT_REQUEST_BYTES if raw.nil?

  value = Integer(raw, exception: false) || 0
  return value if value.positive? && value <= MAX_EXPORT_REQUEST_BYTES

  Bitfab.warn_once(
    "otel-max-request-bytes-invalid",
    "#{MAX_REQUEST_BYTES_ENV} must be a positive integer no greater than " \
    "#{MAX_EXPORT_REQUEST_BYTES}; using #{MAX_EXPORT_REQUEST_BYTES}"
  )
  MAX_EXPORT_REQUEST_BYTES
end

.monotonic_now ⇒ Object 



185
186
187
 
# File 'lib/bitfab/otel.rb', line 185

def monotonic_now
  Process.clock_gettime(Process::CLOCK_MONOTONIC)
end

.record_carrier_ref(span_id, ref) ⇒ Object

Bounded, because the only thing that removes an entry is the encode that consumes it: a span the processor drops under backpressure never reaches encode, and its ref would otherwise sit here for the life of the process. The queue is the ceiling on how many spans can be waiting at once, so anything older than that is already unreachable. Evicting a live one costs an ack, and the barrier falls back to asking the server.



173
174
175
176
177
178
179
 
# File 'lib/bitfab/otel.rb', line 173

def record_carrier_ref(span_id, ref)
  ensure_process_state
  submission_mutex.synchronize do
    carrier_refs[span_id] = ref
    carrier_refs.shift while carrier_refs.size > MAX_QUEUE_SIZE
  end
end

.register_transport(transport) ⇒ Object 



141
142
143
144
 
# File 'lib/bitfab/otel.rb', line 141

def register_transport(transport)
  ensure_process_state
  live_transports_mutex.synchronize { live_transports << transport }
end

.reset_state_after_fork ⇒ Object

Carrier refs belong to the parent's run, never the child's.



202
203
204
205
 
# File 'lib/bitfab/otel.rb', line 202

def reset_state_after_fork
  @state_pid = Process.pid
  @carrier_refs = {}
end

.shutdown_transports(timeout = 30.0) ⇒ Object 



159
160
161
162
163
164
165
 
# File 'lib/bitfab/otel.rb', line 159

def shutdown_transports(timeout = 30.0)
  ensure_process_state
  deadline = monotonic_now + [timeout, 0].max
  current_process_transports.reduce(true) do |succeeded, transport|
    transport.shutdown([deadline - monotonic_now, 0].max) && succeeded
  end
end

.take_carrier_ref(span_id) ⇒ Object 



181
182
183
 
# File 'lib/bitfab/otel.rb', line 181

def take_carrier_ref(span_id)
  submission_mutex.synchronize { carrier_refs.delete(span_id) }
end

.unregister_transport(transport) ⇒ Object 



146
147
148
149
 
# File 'lib/bitfab/otel.rb', line 146

def unregister_transport(transport)
  ensure_process_state
  live_transports_mutex.synchronize { live_transports.delete(transport) }
end