Class: Bitfab::Otel::BatchTransport

Inherits:

Object

• Object
• Bitfab::Otel::BatchTransport

Defined in:

lib/bitfab/otel.rb

Overview

Owns one private TracerProvider + BatchSpanProcessor per client. The provider is never installed globally, so the SDK cannot disturb an application's own OpenTelemetry setup.

Instance Attribute Summary

• #owner_pid ⇒ Object readonly

  Returns the value of attribute owner_pid.

Instance Method Summary

• #flush(timeout = 30.0) ⇒ Object
• #initialize(api_key:, direct_sender:, collector_endpoint: nil, max_export_batch_size: nil, max_request_batch_size: DIRECT_MAX_REQUEST_BATCH_SIZE, max_queue_size: MAX_QUEUE_SIZE, export_concurrency: DEFAULT_EXPORT_CONCURRENCY, max_request_bytes: nil) ⇒ BatchTransport constructor

  A new instance of BatchTransport.

• #shutdown(timeout = 30.0) ⇒ Object
• #submit(operation, payload) ⇒ Object

Constructor Details

#initialize(api_key:, direct_sender:, collector_endpoint: nil, max_export_batch_size: nil, max_request_batch_size: DIRECT_MAX_REQUEST_BATCH_SIZE, max_queue_size: MAX_QUEUE_SIZE, export_concurrency: DEFAULT_EXPORT_CONCURRENCY, max_request_bytes: nil) ⇒ BatchTransport

Returns a new instance of BatchTransport.

Raises:

• (ArgumentError)

# File 'lib/bitfab/otel.rb', line 497

def initialize(api_key:, direct_sender:, collector_endpoint: nil, max_export_batch_size: nil,
  max_request_batch_size: DIRECT_MAX_REQUEST_BATCH_SIZE, max_queue_size: MAX_QUEUE_SIZE,
  export_concurrency: DEFAULT_EXPORT_CONCURRENCY, max_request_bytes: nil)
  raise ArgumentError, "max_request_batch_size must be a positive integer" unless max_request_batch_size.positive?

  @api_key = api_key
  @direct_sender = direct_sender
  @collector_endpoint = collector_endpoint
  @max_export_batch_size = max_export_batch_size || default_export_batch_size
  @max_request_batch_size = max_request_batch_size
  @max_queue_size = max_queue_size
  @export_concurrency = export_concurrency
  @max_request_bytes = max_request_bytes || MAX_EXPORT_REQUEST_BYTES
  @owner_pid = Process.pid
  @state_mutex = Mutex.new
  @flush_mutex = Mutex.new
  @closed = false
  create_pipeline
  Otel.register_transport(self)
end

Instance Attribute Details

#owner_pid ⇒ Object (readonly)

Returns the value of attribute owner_pid.

# File 'lib/bitfab/otel.rb', line 495

def owner_pid
  @owner_pid
end

Instance Method Details

#flush(timeout = 30.0) ⇒ Object

# File 'lib/bitfab/otel.rb', line 559

def flush(timeout = 30.0)
  processor, delivery_tracker = @state_mutex.synchronize do
    ensure_process
    [@processor, @delivery_tracker]
  end
  return true if processor.nil?

  flushed = false
  flush_thread = Thread.new do
    @flush_mutex.synchronize do
      drained = begin
        processor.force_flush(timeout:)
      rescue => e
        Bitfab.warn_once(
          "otel-flush-failed",
          "failed to flush OpenTelemetry spans (further occurrences " \
          "suppressed): #{e.message}"
        )
        FAILURE
      end
      failed_exports = delivery_tracker.nil? ? 0 : delivery_tracker.take_failed_exports
      flushed = drained == SUCCESS && failed_exports.zero?
    end
  end
  return false unless flush_thread.join([timeout, 0].max)

  flushed
end

#shutdown(timeout = 30.0) ⇒ Object

# File 'lib/bitfab/otel.rb', line 588

def shutdown(timeout = 30.0)
  deadline = Otel.monotonic_now + [timeout, 0].max
  @state_mutex.synchronize { @closed = true }
  flushed = flush([deadline - Otel.monotonic_now, 0].max)

  processor = @state_mutex.synchronize do
    current = @processor
    @processor = nil
    @provider = nil
    @delivery_tracker = nil
    @tracer = nil
    current
  end
  return finish_shutdown(flushed) if processor.nil?

  shutdown_thread = Thread.new { processor.shutdown(timeout: [deadline - Otel.monotonic_now, 0].max) }
  joined = shutdown_thread.join([deadline - Otel.monotonic_now, 0].max)
  finish_shutdown(flushed && !joined.nil?)
end

#submit(operation, payload) ⇒ Object

# File 'lib/bitfab/otel.rb', line 518

def submit(operation, payload)
  Otel.record_submission(operation, payload)
  # Encoding is the expensive part of a submit and needs no mutual
  # exclusion, so it stays outside the lock.
  name = Encoder.span_name(operation, payload)
  encoded_payload = Serialize.safe_generate(payload)
  started_at = Encoder.timestamp(payload, "started_at")
  ended_at = Encoder.timestamp(payload, "ended_at")
  errored = Encoder.error?(payload)

  # Pipeline construction and the closed check are serialized: concurrent
  # first submits in a forked child would otherwise each build a pipeline,
  # orphaning one batch worker along with whatever it had queued.
  @state_mutex.synchronize do
    return warn_closed if @closed

    ensure_process
    tracer = @tracer
    return warn_closed if tracer.nil?

    span = tracer.start_root_span(
      name,
      attributes: {
        OPERATION_ATTRIBUTE => operation,
        PAYLOAD_ATTRIBUTE => encoded_payload
      },
      start_timestamp: started_at
    )
    span.status = OpenTelemetry::Trace::Status.error if errored
    span.finish(end_timestamp: ended_at)
  end
  nil
rescue => e
  Bitfab.warn_once(
    "otel-submit-failed",
    "failed to queue an OpenTelemetry span (further occurrences " \
    "suppressed): #{e.message}"
  )
  nil
end