1. class
  2. Hypervel
  3. \
  4. OpenTelemetry
  5. \
  6. Instrumentation
  7. \
  8. QueueInstrumentation

QueueInstrumentation

class QueueInstrumentation extends AbstractInstrumentation

Traits

LogsMessagesTrait

Constants

protected SENT_MESSAGES_METRIC

protected SEND_DURATION_METRIC

protected CONSUMED_MESSAGES_METRIC

protected PROCESS_DURATION_METRIC

protected DEPTH_METRIC

protected CONNECTION_ATTRIBUTE

protected DURATION_BOUNDARIES

Properties

protected InstrumentationOptions $options from  AbstractInstrumentation
protected TracerInterface|null $tracer
protected CounterInterface|null $sentMessages
protected HistogramInterface|null $sendDuration
protected CounterInterface|null $consumedMessages
protected HistogramInterface|null $processDuration
protected bool $propagation
protected array<string, string> $connectionDrivers
protected array<string, true> $syncConnections
protected QueueConsumerState> $consumerStates

Methods

void
register(array $options)

Register the instrumentation with normalized options.

from  AbstractInstrumentation
void
registerInstrumentation()

Register queue listeners, propagation, and instruments.

bool
tracesEnabled()

Determine whether trace output is enabled.

from  AbstractInstrumentation
bool
metricEnabled(string $name)

Determine whether a metric is enabled.

from  AbstractInstrumentation
array
indexObservers(array $names, array $observers)

Index callback observers by their matching instrument name.

from  AbstractInstrumentation
__construct(Dispatcher $events, Repository $config, QueueManager $queues, TracerProviderInterface $tracerProvider, MeterProviderInterface $meterProvider, ClockInterface $clock, TextMapPropagatorInterface $propagator, ExceptionContextRegistry $exceptionContexts, OperationOrigin $origins, ProcessIdentity $identity, LogContextScopeFactory $logContextScopes)

Create queue instrumentation.

void
registerMetrics()

Create enabled queue metric instruments.

void
finalizePayload(JobPayloadFinalizing $event)

Inject propagation and start a persistent producer span.

void
finishProducer(string $payload, Throwable|null $exception = null)

Complete persistent producer telemetry.

string|null
payloadUuid(string $payload)

Return a framework UUID from a rewritten terminal payload.

void
registerLocalPropagation()

Register flat propagation for local asynchronous queue drivers.

void
startConsumer(JobProcessing $event)

Start consumer telemetry when a job reaches application processing.

void
recordConsumerError(Job $job, Throwable $exception)

Retain the first ordinary consumer failure.

void
recordConsumerTimeout(Job $job)

Retain a timeout when no earlier failure explains the attempt.

void
finishConsumer(JobAttempted $event)

Complete consumer telemetry at the canonical attempt boundary.

void
registerDepthMetric()

Register collection-only queue depth in event worker zero.

array
attributes(string $connection, string|null $queue, string $operation, string $operationType)

Return low-cardinality attributes shared by queue spans and metrics.

array
connectionDrivers()

Return configured connection drivers by connection name.

string
spanName(string $operation, string|null $queue)

Return a queue operation span name.

Details

in AbstractInstrumentation at line 18
final void register(array $options)

Register the instrumentation with normalized options.

Parameters

array $options

Return Value

void

at line 125
protected void registerInstrumentation()

Register queue listeners, propagation, and instruments.

Return Value

void

in AbstractInstrumentation at line 32
protected bool tracesEnabled()

Determine whether trace output is enabled.

Return Value

bool

in AbstractInstrumentation at line 40
protected bool metricEnabled(string $name)

Determine whether a metric is enabled.

Parameters

string $name

Return Value

bool

in AbstractInstrumentation at line 55
protected array indexObservers(array $names, array $observers)

Index callback observers by their matching instrument name.

The SDK supplies observers in the same order as the instruments passed to batchObserve(), so the unchanged name order is the mapping contract.

Parameters

array $names
array $observers

Return Value

array

at line 106
__construct(Dispatcher $events, Repository $config, QueueManager $queues, TracerProviderInterface $tracerProvider, MeterProviderInterface $meterProvider, ClockInterface $clock, TextMapPropagatorInterface $propagator, ExceptionContextRegistry $exceptionContexts, OperationOrigin $origins, ProcessIdentity $identity, LogContextScopeFactory $logContextScopes)

Create queue instrumentation.

Parameters

Dispatcher $events
Repository $config
QueueManager $queues
TracerProviderInterface $tracerProvider
MeterProviderInterface $meterProvider
ClockInterface $clock
TextMapPropagatorInterface $propagator
ExceptionContextRegistry $exceptionContexts
OperationOrigin $origins
ProcessIdentity $identity
LogContextScopeFactory $logContextScopes

at line 201
protected void registerMetrics()

Create enabled queue metric instruments.

Return Value

void

at line 247
protected void finalizePayload(JobPayloadFinalizing $event)

Inject propagation and start a persistent producer span.

Parameters

JobPayloadFinalizing $event

Return Value

void

at line 329
protected void finishProducer(string $payload, Throwable|null $exception = null)

Complete persistent producer telemetry.

Parameters

string $payload
Throwable|null $exception

Return Value

void

at line 376
protected string|null payloadUuid(string $payload)

Return a framework UUID from a rewritten terminal payload.

Parameters

string $payload

Return Value

string|null

at line 392
protected void registerLocalPropagation()

Register flat propagation for local asynchronous queue drivers.

Return Value

void

at line 422
protected void startConsumer(JobProcessing $event)

Start consumer telemetry when a job reaches application processing.

Parameters

JobProcessing $event

Return Value

void

at line 511
protected void recordConsumerError(Job $job, Throwable $exception)

Retain the first ordinary consumer failure.

Parameters

Job $job
Throwable $exception

Return Value

void

at line 526
protected void recordConsumerTimeout(Job $job)

Retain a timeout when no earlier failure explains the attempt.

Parameters

Job $job

Return Value

void

at line 540
protected void finishConsumer(JobAttempted $event)

Complete consumer telemetry at the canonical attempt boundary.

Parameters

JobAttempted $event

Return Value

void

at line 598
protected void registerDepthMetric()

Register collection-only queue depth in event worker zero.

Return Value

void

at line 648
protected array attributes(string $connection, string|null $queue, string $operation, string $operationType)

Return low-cardinality attributes shared by queue spans and metrics.

Parameters

string $connection
string|null $queue
string $operation
string $operationType

Return Value

array

at line 668
protected array connectionDrivers()

Return configured connection drivers by connection name.

Return Value

array

at line 690
protected string spanName(string $operation, string|null $queue)

Return a queue operation span name.

Parameters

string $operation
string|null $queue

Return Value

string