Documentation
¶
Overview ¶
Package telemetry holds the OpenTelemetry spans and metric instruments Dispatch records for the work it does itself: the job pool and runner, cron, checkpoint workflows, the durable runtime, the outbox publisher and the artifact sweeper.
It is a leaf package. It imports the standard library, OpenTelemetry and go-utils metrics, and nothing else from Dispatch, so every subsystem can import it without a cycle. exec deliberately does not import it.
You rarely build a Telemetry yourself. engine.Build builds one from engine.WithMetricFactory and engine.WithTracerProvider and hands it to every subsystem it constructs, and the Forge extension passes the app's metrics. Build one with New only when you assemble a subsystem (a worker.Pool, a durable runtime Worker, a delivery Publisher) outside the engine.
A nil *Telemetry is valid everywhere: spans go to the global tracer provider and no metrics are recorded.
Index ¶
- Constants
- func Detach(ctx context.Context) context.Context
- func ErrorType(err error) string
- func Fail(span trace.Span, class, errType string)
- func StartChild(ctx context.Context, name string, opts ...trace.SpanStartOption) (context.Context, trace.Span)
- type ArtifactMetrics
- type CronMetrics
- type DurableMetrics
- type JobMetrics
- func (m *JobMetrics) Active(n int)
- func (m *JobMetrics) Attempt(queue, outcome string, d time.Duration)
- func (m *JobMetrics) DeadLettered(queue, reason string)
- func (m *JobMetrics) Enqueued(queue string)
- func (m *JobMetrics) Poll(result string)
- func (m *JobMetrics) QueueWait(queue string, d time.Duration)
- func (m *JobMetrics) Reclaimed(n int)
- func (m *JobMetrics) Renewal(result string)
- func (m *JobMetrics) Requeued(queue, reason string)
- type Option
- type OutboxMetrics
- type Telemetry
- func (t *Telemetry) Artifacts() *ArtifactMetrics
- func (t *Telemetry) Cron() *CronMetrics
- func (t *Telemetry) Durable() *DurableMetrics
- func (t *Telemetry) EnqueueLinks(jobID string) []trace.Link
- func (t *Telemetry) Jobs() *JobMetrics
- func (t *Telemetry) Outbox() *OutboxMetrics
- func (t *Telemetry) QueueLabel(queue string) string
- func (t *Telemetry) RememberEnqueue(jobID string, sc trace.SpanContext)
- func (t *Telemetry) Start(ctx context.Context, name string, opts ...trace.SpanStartOption) (context.Context, trace.Span)
- func (t *Telemetry) Tracer() trace.Tracer
- func (t *Telemetry) Workflows() *WorkflowMetrics
- type WorkflowMetrics
Constants ¶
const ( MetricJobsEnqueued = "dispatch_jobs_enqueued_total" MetricJobAttemptDuration = "dispatch_job_attempt_duration_seconds" MetricJobsDeadLettered = "dispatch_jobs_dead_lettered_total" MetricJobQueueWait = "dispatch_job_queue_wait_seconds" MetricJobPolls = "dispatch_job_polls_total" MetricJobsRequeued = "dispatch_jobs_requeued_total" MetricJobLeaseRenewals = "dispatch_job_lease_renewals_total" MetricJobsReclaimed = "dispatch_jobs_reclaimed_total" MetricJobsActive = "dispatch_jobs_active" MetricCronFires = "dispatch_cron_fires_total" MetricCronFireLag = "dispatch_cron_fire_lag_seconds" MetricCronLeader = "dispatch_cron_leader" MetricWorkflowRuns = "dispatch_workflow_run_duration_seconds" MetricWorkflowStep = "dispatch_workflow_step_duration_seconds" MetricDurableTaskDuration = "dispatch_durable_task_duration_seconds" MetricDurablePolls = "dispatch_durable_polls_total" MetricDurableReplayDuration = "dispatch_durable_replay_duration_seconds" MetricOutboxDeliveryDuration = "dispatch_outbox_delivery_duration_seconds" MetricOutboxDeliveryLag = "dispatch_outbox_delivery_lag_seconds" MetricArtifactsSwept = "dispatch_artifacts_swept_total" MetricArtifactsPurged = "dispatch_artifacts_purged_total" MetricArtifactBytesReclaimed = "dispatch_artifact_bytes_reclaimed_total" )
Metric names, as Dispatch registers them. A Forge app scrapes each one with its namespace in front: dispatch_jobs_enqueued_total is forge_dispatch_jobs_enqueued_total on /_/metrics.
const ( OutcomeCompleted = "completed" OutcomeRetry = "retry" OutcomeDeadLettered = "dead_lettered" OutcomeRequeued = "requeued" OutcomeLeaseLost = "lease_lost" OutcomeCancelled = "cancelled" OutcomeStoreError = "store_error" )
Job attempt outcomes, the outcome label on dispatch_job_attempt_duration_seconds and the dispatch.job.outcome span attribute. Cancelled covers an operator's cancel and a worker shutting down mid-attempt; store_error means the store refused the terminal write.
const ( DeadLetterRetriesExhausted = "retries_exhausted" DeadLetterPermanent = "permanent" DeadLetterLaunchFailures = "launch_failures" )
Dead letter reasons, the reason label on dispatch_jobs_dead_lettered_total.
const ( RequeueRateLimited = "rate_limited" RequeueNoCapacity = "no_capacity" RequeueShutdown = "shutdown" )
Requeue reasons, the reason label on dispatch_jobs_requeued_total: a job this worker claimed and handed back without running it.
const ( ResultOK = "ok" ResultError = "error" )
Shared result values.
const ( PollWork = "work" PollEmpty = "empty" PollError = "error" )
Poll results, the result label on dispatch_job_polls_total and dispatch_durable_polls_total.
const ( RenewalOK = "ok" RenewalLost = "lost" RenewalError = "error" )
Lease renewal results, the result label on dispatch_job_lease_renewals_total.
const ( CronFired = "fired" CronLocked = "locked" CronSkipped = "skipped" CronError = "error" )
Cron fire results, the result label on dispatch_cron_fires_total.
const ( DurableOK = "ok" DurableContention = "contention" DurableLeaseLost = "lease_lost" DurableNoHandler = "no_handler" DurableCancelled = "cancelled" DurableError = "error" )
Durable task results, the result label on dispatch_durable_task_duration_seconds.
const ( OutboxAcknowledged = "acknowledged" OutboxCancelled = "cancelled" OutboxRejected = "rejected" OutboxTimeout = "timeout" OutboxConflict = "conflict" OutboxInvalidReceipt = "invalid_receipt" )
Outbox delivery results, the result label on dispatch_outbox_delivery_duration_seconds. Rejected through invalid_receipt are the failure categories the publisher already persists on a retry; cancelled means Stop interrupted the attempt, which leaves the delivery to be claimed again.
const ( PurgeOK = "ok" PurgeSkipped = "skipped" )
Artifact purge results, the result label on dispatch_artifacts_purged_total.
const OtherQueue = "other"
OtherQueue is the queue label for any queue outside the configured set. Queue names are chosen by whoever enqueues, so only the names you configured (WithQueues) become label values. Everything else collapses into this one.
const ScopeName = "github.com/xraph/dispatch"
ScopeName is the instrumentation scope on every span Dispatch starts.
Variables ¶
This section is empty.
Functions ¶
func Detach ¶
Detach returns ctx without its span or baggage, keeping its values, deadline and cancellation. A long-lived loop calls it once on the context it was started with. forge v1.12.9 hands an extension's Start a context that carries the "start <ext>" boot span, and a request context carries the request's server span; without this, every query an idle poll makes would land in that long-finished trace.
func ErrorType ¶
ErrorType names err for an exception event: "context.Canceled" or "context.DeadlineExceeded" when the chain holds one, otherwise the Go type of the innermost error, such as "*net.OpError". It returns "" for nil.
func Fail ¶
Fail marks span as failed. class is a short, bounded description (a failure class such as "handler error", never err.Error()), and the exception event carries only exception.type: the error's Go type or a sentinel's name. Driver and handler messages quote values (payload fragments, keys, addresses), and none of that belongs on a span.
func StartChild ¶
func StartChild(ctx context.Context, name string, opts ...trace.SpanStartOption) (context.Context, trace.Span)
StartChild starts a span only when ctx already carries a local Dispatch span, using that span's own tracer provider. Code that has no Telemetry of its own (the staging middleware) uses it to nest under the job attempt it runs in without a span of its own outside one. The caller ends it.
Types ¶
type ArtifactMetrics ¶
type ArtifactMetrics struct {
// contains filtered or unexported fields
}
ArtifactMetrics are the artifact sweeper instruments.
func (*ArtifactMetrics) BytesReclaimed ¶
func (m *ArtifactMetrics) BytesReclaimed(n int64)
BytesReclaimed counts bytes the purge pass deleted.
func (*ArtifactMetrics) Purged ¶
func (m *ArtifactMetrics) Purged(result string, n int)
Purged counts artifacts the purge pass handled.
func (*ArtifactMetrics) Swept ¶
func (m *ArtifactMetrics) Swept(reason string, n int)
Swept counts artifacts marked for purge.
type CronMetrics ¶
type CronMetrics struct {
// contains filtered or unexported fields
}
CronMetrics are the cron scheduler instruments.
func (*CronMetrics) Fire ¶
func (m *CronMetrics) Fire(result string)
Fire counts one due cron entry handled.
func (*CronMetrics) Lag ¶
func (m *CronMetrics) Lag(d time.Duration)
Lag records how late a cron entry fired.
func (*CronMetrics) Leader ¶
func (m *CronMetrics) Leader(leading bool)
Leader records whether this scheduler leads.
type DurableMetrics ¶
type DurableMetrics struct {
// contains filtered or unexported fields
}
DurableMetrics are the durable runtime instruments.
func (*DurableMetrics) Poll ¶
func (m *DurableMetrics) Poll(kind, result string)
Poll counts one durable claim poll.
func (*DurableMetrics) Replay ¶
func (m *DurableMetrics) Replay(d time.Duration)
Replay records one history replay.
type JobMetrics ¶
type JobMetrics struct {
// contains filtered or unexported fields
}
JobMetrics are the job pool, runner and enqueue instruments.
func (*JobMetrics) Active ¶
func (m *JobMetrics) Active(n int)
Active sets the number of attempts running on this worker.
func (*JobMetrics) Attempt ¶
func (m *JobMetrics) Attempt(queue, outcome string, d time.Duration)
Attempt records one finished job attempt.
func (*JobMetrics) DeadLettered ¶
func (m *JobMetrics) DeadLettered(queue, reason string)
DeadLettered counts one job moved to the dead letter queue.
func (*JobMetrics) Enqueued ¶
func (m *JobMetrics) Enqueued(queue string)
Enqueued counts one job written to the store.
func (*JobMetrics) QueueWait ¶
func (m *JobMetrics) QueueWait(queue string, d time.Duration)
QueueWait records how long a claimed job waited past its due time.
func (*JobMetrics) Reclaimed ¶
func (m *JobMetrics) Reclaimed(n int)
Reclaimed counts jobs the reaper returned to pending.
func (*JobMetrics) Renewal ¶
func (m *JobMetrics) Renewal(result string)
Renewal counts one lease renewal or heartbeat.
func (*JobMetrics) Requeued ¶
func (m *JobMetrics) Requeued(queue, reason string)
Requeued counts one claimed job handed back without running.
type Option ¶
type Option func(*Telemetry)
Option configures a Telemetry.
func WithMetrics ¶
func WithMetrics(f gu.MetricFactory) Option
WithMetrics records Dispatch's metrics through f. In a Forge app that is app.Metrics(). The factory has to key series by name and labels, as Forge's does. A nil factory leaves metrics off.
func WithQueues ¶
WithQueues names the queues that may appear as a metric label. Pass the queues your workers poll and the ones you configured limits for. Any other queue name is recorded as OtherQueue.
func WithTracerProvider ¶
func WithTracerProvider(tp trace.TracerProvider) Option
WithTracerProvider starts Dispatch's spans on tp. A nil tp, the default, means the global provider, looked up on every span so a provider installed later (forge.New, a test recorder) still receives them.
type OutboxMetrics ¶
type OutboxMetrics struct {
// contains filtered or unexported fields
}
OutboxMetrics are the outbox publisher instruments.
type Telemetry ¶
type Telemetry struct {
// contains filtered or unexported fields
}
Telemetry carries the tracer provider and metric factory Dispatch records through. Build it once with New and share it; it is safe for concurrent use. A nil *Telemetry records no metrics and starts spans on the global tracer provider.
func (*Telemetry) Artifacts ¶
func (t *Telemetry) Artifacts() *ArtifactMetrics
Artifacts returns the artifact sweeper instruments, or nil when metrics are off.
func (*Telemetry) Cron ¶
func (t *Telemetry) Cron() *CronMetrics
Cron returns the cron instruments, or nil when metrics are off.
func (*Telemetry) Durable ¶
func (t *Telemetry) Durable() *DurableMetrics
Durable returns the durable runtime instruments, or nil when metrics are off.
func (*Telemetry) EnqueueLinks ¶
EnqueueLinks returns a link to the span that enqueued jobID, when this process enqueued it recently enough to remember.
func (*Telemetry) Jobs ¶
func (t *Telemetry) Jobs() *JobMetrics
Jobs returns the job instruments, building them on the first call. It returns nil, which records nothing, when metrics are off.
func (*Telemetry) Outbox ¶
func (t *Telemetry) Outbox() *OutboxMetrics
Outbox returns the outbox publisher instruments, or nil when metrics are off.
func (*Telemetry) QueueLabel ¶
QueueLabel maps a queue name to its metric label: the name itself when it is one of the configured queues, OtherQueue otherwise.
func (*Telemetry) RememberEnqueue ¶
func (t *Telemetry) RememberEnqueue(jobID string, sc trace.SpanContext)
RememberEnqueue records the span context that enqueued a job, so an attempt this process runs can link back to it. Nothing about it is stored with the job: the cache is in memory, bounded, and an attempt on another worker simply starts without a link.
func (*Telemetry) Start ¶
func (t *Telemetry) Start(ctx context.Context, name string, opts ...trace.SpanStartOption) (context.Context, trace.Span)
Start starts a span under Dispatch's instrumentation scope. The caller ends it.
func (*Telemetry) Workflows ¶
func (t *Telemetry) Workflows() *WorkflowMetrics
Workflows returns the checkpoint workflow instruments, or nil when metrics are off.
type WorkflowMetrics ¶
type WorkflowMetrics struct {
// contains filtered or unexported fields
}
WorkflowMetrics are the checkpoint workflow instruments.
Source Files