Module: Bitfab::Otel

Defined in:
lib/bitfab/otel.rb,
lib/bitfab/otel_collector.rb

Overview

OpenTelemetry delivery for Bitfab spans and trace completions.

Bitfab payloads ride as OTLP span attributes (bitfab.operation + bitfab.payload) so a batch can travel either straight to Bitfab as OTLP/JSON or through a customer's OTel Collector as protobuf, without the payload itself changing shape.

Defined Under Namespace

Modules: Encoder Classes: BatchTransport, CollectorExporter, DeliveryTrackingExporter, DirectExporter, SizeLimitedExporter

Constant Summary collapse

OPERATION_ATTRIBUTE = 
"bitfab.operation"
PAYLOAD_ATTRIBUTE = 
"bitfab.payload"
OTLP_TRACES_ENDPOINT = 
"/api/sdk/otel/v1/traces"
COLLECTOR_ENDPOINT_ENV = 
"BITFAB_OTEL_EXPORTER_ENDPOINT"
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
COLLECTOR_MAX_EXPORT_BATCH_SIZE = 
32
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
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(api_key:, direct_sender:) ⇒ Object 



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

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



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

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

  reset_state_after_fork
end

.export_concurrency_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 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 



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

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 



55
56
57
58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/bitfab/otel.rb', line 55

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 



155
156
157
 
# File 'lib/bitfab/otel.rb', line 155

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

.record_submission(operation, payload) ⇒ Object 



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

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 



95
96
97
98
 
# File 'lib/bitfab/otel.rb', line 95

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.



172
173
174
175
176
 
# File 'lib/bitfab/otel.rb', line 172

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 



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

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.



124
125
126
127
128
129
130
131
132
133
134
135
 
# File 'lib/bitfab/otel.rb', line 124

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 



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

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