oteljob

README

oteljob

Optional OpenTelemetry metrics for job. Two instruments measure completed attempts. No automatic logging, tracing, exporter setup or network calls by the handler.

Install

Requires Go 1.27 or later.

go get github.com/uchaloop/oteljob

Use

Given an application-configured metric.MeterProvider named provider:

observer, err := oteljob.MakeHandler(provider.Meter("example/orders-worker"))
if err != nil {
    return err
}
runner, err := job.MakeRunner(
    job.Config{Timeout: time.Minute},
    func(ctx context.Context) (int64, error) {
        return processBatch(ctx)
    },
)
if err != nil {
    return err
}
result := runner.Run(ctx)
observer.Handle(context.WithoutCancel(ctx), result)

Deliver each result once. job.Runner does not call an observer automatically. Omit the observer entirely, implement job.Handler yourself, or combine sinks with job.MultiHandler. Neither job nor beat imports OTel.

See the executable example for a complete SDK example.

flowchart LR
    Work[Func + middleware] --> Result[job.Result]
    Error[Optional ErrorHandler] --> Result
    Result --> Observer[Explicit Handler.Handle]
    Observer --> Duration[job.run.duration]
    Observer --> Processed[job.run.processed]
    Duration --> SDK[Application SDK and exporter]
    Processed --> SDK

Metrics

Instrument	Type / unit	Attributes
job.run.duration	Float64Histogram / s	phase, outcome
job.run.processed	Int64Histogram / {item}	outcome

Duration phases:

Phase	Measures	Outcome describes
total	Entire attempt, including error processing	Original work
work	Func and middleware	Original work
error_handler	Error processing, only when called	ErrorHandler itself

Outcomes are ok, error, panic, timeout, canceled; unrecognized values become unknown. Success is outcome="ok"; a separate numeric code would repeat the same information. These outcomes do not represent process exit codes. A successful DLQ callback does not change a failed work outcome. The callback classifier recognizes wrapped/joined panic, deadline and cancellation errors, then other errors; the original work uses Result.Outcome unchanged.

Count attempts using duration count with phase="total". Counting all phases would count the same attempt two or three times. Filter outcome="panic" for recovered panics; no separate panic instrument is necessary.

The processed histogram includes zero and partial progress on failed attempts. Its sum is reported processed items, its count is sampled attempts, and its buckets describe batch sizes. Negative counts are omitted rather than turned into empty batches. The library cannot verify application-reported counts. Integer aggregation is int64; cumulative sums must remain within its range.

Results with zero Start are ignored: work never ran. An unrecovered panic or blocked function has no completed result to observe. Timing excludes observation handlers, scheduling waits, backoff and application startup/shutdown. Aggregated metrics do not preserve a per-iteration history.

Histograms

Default advisory boundaries:

• Duration, seconds: 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60, 300, 600, 1800, 3600, 14400, 86400.
• Processed items: 0, 1, 10, 50, 100, 250, 500, 750, 1000, 1500, 2500, 5000, 10000.

The application can override these with SDK Views, including choosing a different aggregation. Values above the last boundary enter the overflow bucket. Attributes and histogram buckets both contribute to the number of exported time series.

Service identity and export

The application configures the SDK Resource; the library adds no service or infrastructure attributes. Suggested Resource fields are service.name and deployment.environment.name, plus k8s.cluster.name and k8s.namespace.name when applicable. Keep independent replicas distinguishable in the export pipeline. Do not add error text, message IDs or iteration numbers as metric attributes.

Prometheus resource promotion is pipeline-specific: Resource fields do not necessarily become labels on every series. Promote only the fields needed for filtering. These metrics require neither logs nor traces.

For short-lived jobs, a scrape endpoint may disappear before it is scraped. Use an application-configured OTLP exporter/Collector where appropriate, and shut down the provider after recording the result, with a fresh bounded context. A shutdown attempts to export buffered measurements; it does not guarantee backend persistence. A ManualReader only collects when asked and does not send telemetry.

rate/increase need suitable samples of a time series. A single sample from an ephemeral job does not provide a reliable execution count via increase. Choose the collection and aggregation strategy for one-shot jobs accordingly.

Environment configuration

Read telemetry identity from the standard OTel environment variables when the application constructs its MeterProvider. The adapters consume that provider; they do not read environment variables themselves. No separate configuration library is needed for these values.

Environment variable	Value
OTEL_SERVICE_NAME	Set to a stable service name, such as daemon-efiro.
OTEL_RESOURCE_ATTRIBUTES	Comma-separated key=value Resource attributes, as below.

Recommended attributes inside OTEL_RESOURCE_ATTRIBUTES:

Attribute	When to set it
deployment.environment.name	Local, staging or production environment.
service.instance.id	A unique identity for each concurrently running instance.
k8s.namespace.name	Kubernetes namespace; omit outside Kubernetes.
k8s.cluster.name	Kubernetes cluster name; omit outside Kubernetes.

These are application identity conventions, not required configuration fields of the handler. OTEL_SERVICE_NAME takes precedence over service.name inside OTEL_RESOURCE_ATTRIBUTES. Set identity before constructing the provider; changing the environment afterward does not refresh its Resource.

Local example (use a distinct instance ID for each concurrent process):

export OTEL_SERVICE_NAME=worker
export OTEL_RESOURCE_ATTRIBUTES='deployment.environment.name=local,service.instance.id=worker-local-1'

In Kubernetes, obtain the namespace (metadata.namespace) and Pod identity (metadata.uid) through the Downward API. Supply the service name, deployment environment and cluster name through your deployment configuration. To compose OTEL_RESOURCE_ATTRIBUTES from these values, follow Kubernetes' official guide to dependent environment variables.

resource.WithFromEnv() reads these values. It does not configure an exporter. Select a reader/exporter separately; exporter-specific environment variables (such as an OTLP endpoint) only apply when that exporter is constructed. Resource attributes also do not automatically become Prometheus labels: configure selected Resource-to-label promotion in your exporter or Collector and preserve distinct replica identity.

Read the Resource while constructing the provider in your application:

import (
    "context"

    "go.opentelemetry.io/otel/sdk/resource"
    sdkmetric "go.opentelemetry.io/otel/sdk/metric"
)

func makeMeterProvider(reader sdkmetric.Reader) (*sdkmetric.MeterProvider, error) {
    res, err := resource.New(context.Background(), resource.WithFromEnv())
    if err != nil {
        return nil, err
    }
    return sdkmetric.NewMeterProvider(
        sdkmetric.WithResource(res),
        sdkmetric.WithReader(reader),
    ), nil
}

Pass a Meter from this provider to MakeHandler. The application owns provider shutdown with a fresh, bounded context after work and observation finish.

Grafana dashboard

Job / CronJob dashboard · Setup and metric contract

Last-run Pushgateway snapshots and an optional Kubernetes lifecycle block. This example requires a separate snapshot publisher; MakeHandler alone does not produce its gauges.

Uber Fx

Use oteljobfx to provide a job.Handler from your metric.MeterProvider. Call the handler explicitly after Runner.Run; SDK configuration and shutdown remain application-owned.

Acknowledgements

Built on OpenTelemetry Go. Thank you to its authors and maintainers for the instrumentation API and SDK.

Documentation

Overview

Package oteljob implements optional job.Handler instrumentation with OTel. The caller delivers each completed Result exactly once. The application owns the MeterProvider, Resource, Views, exporters and shutdown. This package does not export telemetry, log errors or create traces itself.

job.run.duration (seconds) has phase=total|work|error_handler and outcome. Count completed attempts using phase=total only. The error_handler phase is recorded only when called and describes the callback's own outcome; total and work retain the original work outcome. job.run.processed ({item}) records nonnegative reported counts, including zero and partial progress on failure, with the work outcome. Its sum is the total and its count the sample count.

Results with zero Start are ignored because no work ran. Panics are visible only when recovered into Results; blocked work has no completion measurement. Unknown outcomes are normalized to "unknown" to bound attribute cardinality. Histogram boundaries are advisory and can be overridden with SDK Views.

Index

• func MakeHandler(meter metric.Meter) (job.Handler, error)

Examples

• MakeHandler

Functions

func MakeHandler

func MakeHandler(meter metric.Meter) (job.Handler, error)

MakeHandler builds optional instrumentation on an application-provided meter. It does not install itself on a runner or change the runner's behavior.

Example

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/uchaloop/job"
	"github.com/uchaloop/oteljob"

	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
	"go.opentelemetry.io/otel/sdk/metric/metricdata"
)

func main() {
	// Use a configured exporter with a PeriodicReader in a one-shot application.
	// This example uses a ManualReader to inspect measurements without a server.
	reader := sdkmetric.NewManualReader()
	provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader))
	observer, err := oteljob.MakeHandler(provider.Meter("example/worker"))
	if err != nil {
		panic(err)
	}
	runner, err := job.MakeRunner(job.Config{Timeout: time.Minute},
		func(context.Context) (int64, error) { return 1000, nil })
	if err != nil {
		panic(err)
	}
	result := runner.Run(context.Background())
	observer.Handle(context.Background(), result) // Explicit, exactly once.
	var data metricdata.ResourceMetrics
	if err := reader.Collect(context.Background(), &data); err != nil {
		panic(err)
	}
	fmt.Println("instruments:", len(data.ScopeMetrics[0].Metrics))

	// A production provider flushes its configured reader/exporter on shutdown.
	// Use a fresh budget even if the work context was canceled.
	flushCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := provider.Shutdown(flushCtx); err != nil {
		panic(err)
	}
}

Output:
instruments: 2