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(direct_sender:, 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, on_delivered: nil) ⇒ BatchTransport constructor

  A new instance of BatchTransport.

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

Constructor Details

#initialize(direct_sender:, 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, on_delivered: nil) ⇒ BatchTransport

Returns a new instance of BatchTransport.

Raises:

• (ArgumentError)

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

def initialize(direct_sender:, 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, on_delivered: nil)
  raise ArgumentError, "max_request_batch_size must be a positive integer" unless max_request_batch_size.positive?

  @direct_sender = direct_sender
  @on_delivered = on_delivered
  @max_export_batch_size = max_export_batch_size || DIRECT_MAX_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 523

def owner_pid
  @owner_pid
end

Instance Method Details

#flush(timeout = 30.0) ⇒ Object

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

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 619

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, meta = nil) ⇒ Object

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

def submit(operation, payload, meta = nil)
  meta ||= CarrierMeta.new
  # Encoding is the expensive part of a submit and needs no mutual
  # exclusion, so it stays outside the lock.
  name = meta.name || "bitfab.#{operation}"
  encoded_payload = Serialize.safe_generate(
    payload,
    max_carrier_bytes: PayloadBudget::MAX_COMPRESSIBLE_SPAN_CARRIER_BYTES
  )
  started_at = meta.started_at || Time.now
  ended_at = meta.ended_at || Time.now
  errored = meta.errored

  # 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
    )
    Otel.record_carrier_ref(span.context.hex_span_id, meta.ref) if meta.ref
    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