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, DeliveryTrackingExporter, DirectExporter, EncodedSpan, RequestBatch, RequestEnvelope

Constant Summary collapse

OPERATION_ATTRIBUTE = 
"bitfab.operation"
PAYLOAD_ATTRIBUTE = 
"bitfab.payload"
OTLP_TRACES_ENDPOINT = 
"/api/sdk/otel/v1/traces"
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_DELAY_SECONDS = 
0.1
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
PayloadTooLargeError = 
Class.new(StandardError)
PartialSuccessError = 
Class.new(StandardError)

Class Method Summary collapse

Class Method Details

.create_transport(direct_sender:) ⇒ Object 



100
101
102
103
104
105
106
 
# File 'lib/bitfab/otel.rb', line 100

def create_transport(direct_sender:)
  BatchTransport.new(
    direct_sender:,
    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.



177
178
179
180
181
182
 
# File 'lib/bitfab/otel.rb', line 177

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

  reset_state_after_fork
end

.export_concurrency_from_env ⇒ Object 



85
86
87
88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/bitfab/otel.rb', line 85

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 



118
119
120
121
122
123
124
 
# File 'lib/bitfab/otel.rb', line 118

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 



70
71
72
73
74
75
76
77
78
79
80
81
82
83
 
# File 'lib/bitfab/otel.rb', line 70

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 



168
169
170
 
# File 'lib/bitfab/otel.rb', line 168

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

.record_submission(operation, payload) ⇒ Object 



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

def record_submission(operation, payload)
  ensure_process_state
  source_trace_id = payload["sourceTraceId"]
  unless source_trace_id.is_a?(String)
    raw_trace = payload["externalTrace"] || payload["rawTrace"]
    source_trace_id = raw_trace["id"] if raw_trace.is_a?(Hash)
  end
  return unless source_trace_id.is_a?(String)

  submission_mutex.synchronize do
    if operation == "external_span"
      record_span_submission(source_trace_id, payload)
    elsif payload["completed"] == true
      record_trace_completion(source_trace_id, payload)
    end
  end
end

.register_transport(transport) ⇒ Object 



108
109
110
111
 
# File 'lib/bitfab/otel.rb', line 108

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

.reset_state_after_fork ⇒ Object

Replay submission counts belong to the parent's run, never the child's.



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

def reset_state_after_fork
  @state_pid = Process.pid
  @trace_submission_span_ids = {}
  @replay_trace_submissions = Set.new
end

.shutdown_transports(timeout = 30.0) ⇒ Object 



126
127
128
129
130
131
132
 
# File 'lib/bitfab/otel.rb', line 126

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_replay_span_counts(trace_ids) ⇒ Object

Count the spans submitted for each replay trace and forget them, so the replay barrier knows how many spans the server must have persisted before the run can be finalized.



137
138
139
140
141
142
143
144
145
146
147
148
 
# File 'lib/bitfab/otel.rb', line 137

def take_replay_span_counts(trace_ids)
  ensure_process_state
  submission_mutex.synchronize do
    counts = trace_ids.each_with_object({}) do |trace_id, acc|
      next unless replay_trace_submissions.include?(trace_id)

      acc[trace_id] = (trace_submission_span_ids.delete(trace_id) || Set.new).size
    end
    replay_trace_submissions.subtract(trace_ids)
    counts
  end
end

.unregister_transport(transport) ⇒ Object 



113
114
115
116
 
# File 'lib/bitfab/otel.rb', line 113

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