Documentation
¶
Overview ¶
Package messaging traces and times message publishing and processing the way the OpenTelemetry messaging semantic conventions describe. Forge's queue, events, conduit and kafka extensions use it, and you can use it in producers and consumers you write yourself.
mt := messaging.New("kafka", messaging.WithMetrics(app.Metrics()))
err := mt.Process(ctx, carrier, "orders", "process", func(ctx context.Context) error {
return handle(ctx, msg)
})
Index ¶
- func Extract(ctx context.Context, carrier propagation.TextMapCarrier) context.Context
- func Inject(ctx context.Context, carrier propagation.TextMapCarrier)
- func WithTraceContext(ctx context.Context, headers map[string]string) map[string]string
- type Operation
- type Option
- type PanicError
- type StartOption
- type Telemetry
- func (t *Telemetry) Process(ctx context.Context, carrier propagation.TextMapCarrier, ...) (err error)
- func (t *Telemetry) StartCreate(ctx context.Context, destination string, opts ...StartOption) (context.Context, *Operation)
- func (t *Telemetry) StartProcess(ctx context.Context, carrier propagation.TextMapCarrier, ...) (context.Context, *Operation)
- func (t *Telemetry) StartSend(ctx context.Context, destination, operation string, opts ...StartOption) (context.Context, *Operation)
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func Extract ¶
func Extract(ctx context.Context, carrier propagation.TextMapCarrier) context.Context
Extract returns ctx with the remote span context found in carrier, or ctx unchanged when there is none.
func Inject ¶
func Inject(ctx context.Context, carrier propagation.TextMapCarrier)
Inject writes ctx's trace context into carrier with the global propagator. It works with tracing off, so a pass-through hop keeps the caller's trace. Injecting into a nil map panics, as propagation.MapCarrier does; use WithTraceContext for maps.
Types ¶
type Operation ¶
type Operation struct {
// contains filtered or unexported fields
}
Operation is one started create, send or process. End it once; a second End does nothing. A nil *Operation is valid and End on it does nothing.
type Option ¶
type Option func(*config)
Option configures New.
func WithMetrics ¶
WithMetrics records messaging.client.operation.duration and messaging.process.duration into m. Without it, or with a nil m, nothing is recorded. forge.Metrics is this type.
func WithTracing ¶
WithTracing turns spans on or off. It defaults to true. With tracing off, trace context still passes through.
type PanicError ¶
type PanicError struct{ Value any }
PanicError is the error End records when Process recovers a panic, before re-raising it.
func (*PanicError) Error ¶
func (e *PanicError) Error() string
func (*PanicError) Unwrap ¶
func (e *PanicError) Unwrap() error
Unwrap returns the panic value when it is an error, so errors.Is and errors.As can reach it through the error End records. It returns nil for any other value.
type StartOption ¶
type StartOption func(*startConfig)
StartOption adjusts one operation.
func WithBatchSize ¶
func WithBatchSize(n int) StartOption
WithBatchSize sets messaging.batch.message_count. An n of zero or less is ignored and leaves the attribute off.
func WithLinks ¶
func WithLinks(links ...trace.Link) StartOption
WithLinks adds span links, such as a batch send linking each message's create span.
func WithSpanKind ¶
func WithSpanKind(kind trace.SpanKind) StartOption
WithSpanKind overrides the span kind. Request and reply use client and server.
type Telemetry ¶
type Telemetry struct {
// contains filtered or unexported fields
}
Telemetry instruments one messaging system. A nil *Telemetry is valid: it starts no spans and records nothing, but StartProcess and Process still extract the caller's trace context.
func New ¶
New builds the instrumenter for one messaging system, such as "kafka", "rabbitmq", "redis", "nats" or "memory". The value becomes messaging.system.
func (*Telemetry) Process ¶
func (t *Telemetry) Process(ctx context.Context, carrier propagation.TextMapCarrier, destination, operation string, fn func(context.Context) error, opts ...StartOption) (err error)
Process runs fn inside a process operation and returns fn's error. A panic in fn ends the operation with a *PanicError and is re-raised.
func (*Telemetry) StartCreate ¶
func (t *Telemetry) StartCreate(ctx context.Context, destination string, opts ...StartOption) (context.Context, *Operation)
StartCreate starts a producer span for creating one message, to inject into that message. It records no metric.
func (*Telemetry) StartProcess ¶
func (t *Telemetry) StartProcess(ctx context.Context, carrier propagation.TextMapCarrier, destination, operation string, opts ...StartOption) (context.Context, *Operation)
StartProcess extracts the trace context from carrier, starts a consumer span as its child, puts the trace and span IDs on ctx for log correlation, and times messaging.process.duration.
Source Files