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_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 



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

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.



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

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

  reset_state_after_fork
end

.export_concurrency_from_env ⇒ Object 



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

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 



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

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 



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

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 



166
167
168
 
# File 'lib/bitfab/otel.rb', line 166

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

.record_submission(operation, payload) ⇒ Object 



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

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 



106
107
108
109
 
# File 'lib/bitfab/otel.rb', line 106

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.



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

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 



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

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.



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

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 



111
112
113
114
 
# File 'lib/bitfab/otel.rb', line 111

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