otlp

type Decorator ¶ added in v1.14.0

type Decorator struct {
	// contains filtered or unexported fields
}

Decorator upserts user-provided resource attributes, scope attributes, scope name, and scope version onto every pdata batch flowing through the proxy before it is forwarded upstream. The mirror flags (`--resource-attribute`, `--scope-attribute`, `--scope-name`, `--scope-version`) on the proxy command match the same flags on `dash0 logs send` and `dash0 spans send` so the user experience is consistent across signal-authoring and signal-forwarding workflows.

All upserts are no-ops when the corresponding flag was not provided — the proxy does NOT default scope name/version to `dash0-cli` / CLI version the way the send commands do, because the inbound pdata already carries the SDK's own instrumentation-library identity and silently overwriting it would erase useful diagnostic context.

Decoration happens in the worker pool, after the consumer has returned 200 to the SDK. The pdata is therefore mutated only by the proxy's own goroutines; the receiver-side contract reported by ProxyConsumer.Capabilities is unaffected.

func NewDecorator(
	resourceAttrs, scopeAttrs map[string]string,
	scopeName, scopeVersion string,
	logAttrs, spanAttrs, metricAttrs map[string]string,
) *Decorator

func (d *Decorator) DecorateLogs(ld plog.Logs)

func (d *Decorator) DecorateMetrics(md pmetric.Metrics)

func (d *Decorator) DecorateTraces(td ptrace.Traces)

func (d *Decorator) IsEmpty() bool

type Emitter ¶ added in v1.14.0

type Emitter struct {
	// contains filtered or unexported fields
}

Emitter builds OTLP/JSON event records about the proxy's own lifecycle and sends them on a channel. The actual stdout write is performed by a separate writer goroutine (StdoutWriter) so the hot path (consumer + workers calling EmitForwarded / EmitError) never blocks on I/O.

When the channel is nil (the no-agent-mode case), every Emit method is a silent no-op — the caller constructs one Emitter shape regardless of mode and the events naturally drop when no one is listening.

func NewEmitter(instanceID string, ch chan<- plog.Logs) *Emitter

func (e *Emitter) EmitError(kind ErrorKind, reason string, code int)

func (e *Emitter) EmitForwarded(sig Signal, count, bytes int64)

func (e *Emitter) EmitShutdown(reason string, finalTotals [signalCount]int64)

func (e *Emitter) EmitStarted(httpEndpoint, grpcEndpoint, dataset, profileName string)

func (e *Emitter) EmitStats(snap SnapshotWithRate)

type ErrorKind ¶ added in v1.14.0

type ErrorKind string

ErrorKind classifies a worker outbound failure (KTD14). Agents reading the event stream can use this to differentiate transient upstream errors (the SDK should keep retrying) from terminal ones (the proxy's own credentials are bad and retry won't help).

const (
	ErrorKindUpstreamUnreachable ErrorKind = "upstream_unreachable"
	ErrorKindUpstream5xx         ErrorKind = "upstream_5xx"
	ErrorKindUpstream4xxAuth     ErrorKind = "upstream_4xx_auth"
	ErrorKindUpstream4xxOther    ErrorKind = "upstream_4xx_other"
	ErrorKindInternalPanic       ErrorKind = "internal_panic"
)

type Forwarder ¶ added in v1.14.0

type Forwarder interface {
	SendLogs(ctx context.Context, ld plog.Logs, dataset *string) error
	SendTraces(ctx context.Context, td ptrace.Traces, dataset *string) error
	SendMetrics(ctx context.Context, md pmetric.Metrics, dataset *string) error
}

Forwarder abstracts the dash0api.Client OTLP send surface so the worker pool can be exercised with a fake in unit tests without standing up the full API client. dash0api.Client satisfies this implicitly.

type LifecycleEvent ¶ added in v1.14.0

type LifecycleEvent struct {
	Kind    LifecycleKind
	Message string
}

LifecycleEvent is the typed channel input for one-shot stderr messages that should print above the live stats block. The writer renders Message verbatim followed by a newline; Kind is reserved for future color or prefix customization and currently informational only.

type LifecycleKind ¶ added in v1.14.0

type LifecycleKind int

LifecycleKind classifies lifecycle messages. v1 doesn't differentiate visually; the kind is recorded so a future polish pass can color or prefix the rendering without changing the channel signature.

const (
	LifecycleInfo LifecycleKind = iota
	LifecycleWarning
	LifecycleError
	LifecycleBanner
)

type OtlpLogSeverityRange ¶

type OtlpLogSeverityRange string

OtlpLogSeverityRange represents the OpenTelemetry log severity ranges. Each range covers a band of four severity numbers as defined by the OpenTelemetry specification.

const (
	Unknown OtlpLogSeverityRange = "UNKNOWN"
	Trace   OtlpLogSeverityRange = "TRACE"
	Debug   OtlpLogSeverityRange = "DEBUG"
	Info    OtlpLogSeverityRange = "INFO"
	Warn    OtlpLogSeverityRange = "WARN"
	Error   OtlpLogSeverityRange = "ERROR"
	Fatal   OtlpLogSeverityRange = "FATAL"
)

type Pipeline ¶ added in v1.14.0

type Pipeline struct {
	// contains filtered or unexported fields
}

Pipeline owns the otlpreceiver-based listener stack. Construction pre-binds ports (so the supervisor can decide collision behavior before the receiver internals fire), builds the receiver Config, and creates the three signal-receiver instances (logs/traces/metrics). Start transfers control to the receiver, which actually binds the HTTP and gRPC servers.

func BuildPipeline(ctx context.Context, httpPort, grpcPort int, consumer *ProxyConsumer) (*Pipeline, error)

func (p *Pipeline) Endpoints() PipelineEndpoints

func (p *Pipeline) Shutdown(ctx context.Context) error

func (p *Pipeline) Start(ctx context.Context) error

type PipelineEndpoints ¶ added in v1.14.0

type PipelineEndpoints struct {
	HTTPEndpoint string // e.g. "127.0.0.1:4318" or "127.0.0.1:53291"
	GRPCEndpoint string // e.g. "127.0.0.1:4317"
}

PipelineEndpoints reports the actual host:port each listener bound to, after pre-bind fallback to OS-assigned ports. The supervisor (U5) uses these to compose the banner and the agent-mode `started` event.

type ProxyConsumer ¶ added in v1.14.0

type ProxyConsumer struct {
	// contains filtered or unexported fields
}

ProxyConsumer implements consumer.Logs / consumer.Traces / consumer.Metrics and bridges the otlpreceiver's pdata delivery into the worker pool's per-signal channels.

Per KTD3a (async-forward), the consumer returns nil (mapping to HTTP 200 / gRPC OK) after enqueueing — the actual upstream forward happens asynchronously on the worker goroutines. The OTLP spec scopes acceptance to a single hop, so this is spec-compliant.

Lifecycle side-effects on each Consume call:

1. stats forwarded counter incremented (U6) — drives both the TTY sparkline (U7) and the agent-mode stats event (U8).
2. agent-mode `dash0.cli.otlp_proxy.forwarded` event emitted (U8).
3. --tail rendering pushed to TailCh when --tail is enabled (U7/U9).
4. Non-blocking enqueue to the per-signal channel.
5. Non-empty channel → return nil (200). Full channel → return retryable error (503 + SDK exponential backoff).

func NewProxyConsumer(stats *Stats, emitter *Emitter, tailCh chan<- string) *ProxyConsumer

func (c *ProxyConsumer) Capabilities() consumer.Capabilities

func (c *ProxyConsumer) ConsumeLogs(_ context.Context, ld plog.Logs) error

func (c *ProxyConsumer) ConsumeMetrics(_ context.Context, md pmetric.Metrics) error

func (c *ProxyConsumer) ConsumeTraces(_ context.Context, td ptrace.Traces) error

func (c *ProxyConsumer) LogsChannel() <-chan plog.Logs

func (c *ProxyConsumer) MetricsChannel() <-chan pmetric.Metrics

func (c *ProxyConsumer) TracesChannel() <-chan ptrace.Traces

type RateSampler ¶ added in v1.14.0

type RateSampler struct {
	// contains filtered or unexported fields
}

RateSampler tracks a rolling window of snapshots so consumers can render a sparkline timeline of forwarded rate per signal. The sampler owns the per-interval ticking and fans snapshots out to subscribed sink channels.

func NewRateSampler(stats *Stats, interval time.Duration, capacity int) *RateSampler

func (rs *RateSampler) History() []SnapshotWithRate

func (rs *RateSampler) Run(ctx context.Context, sinks ...chan<- SnapshotWithRate)

func (rs *RateSampler) Sample() SnapshotWithRate

type Signal ¶ added in v1.14.0

type Signal int

Signal enumerates the three OTLP signal types the proxy tracks. Profiles is intentionally absent — the OTel Profiles signal is not stable and adding it would lock the proxy to a moving target.

const (
	SignalLogs Signal = iota
	SignalSpans
	SignalMetrics
)

func (s Signal) String() string

type Snapshot ¶ added in v1.14.0

type Snapshot struct {
	Forwarded [signalCount]int64
	Failed    [signalCount]int64
	Timestamp time.Time
}

Snapshot is a moment-in-time view of the counters. Captured atomically per signal but not transactionally across signals — a snapshot taken during a burst may see logs from after a span counter increment but spans from before. For visibility purposes this is fine.

type SnapshotWithRate ¶ added in v1.14.0

type SnapshotWithRate struct {
	Snapshot
	// Rate[i] is records-per-second for signal i, derived from the delta
	// against the previous snapshot. Rate is 0 on the first sample (no prior
	// reference) or when the time delta is zero.
	Rate [signalCount]float64
}

SnapshotWithRate annotates a Snapshot with per-signal forwarded rate (records per second) computed against the previous sample in the sampler's rolling window.

type Stats ¶ added in v1.14.0

type Stats struct {
	// contains filtered or unexported fields
}

Stats holds atomic counters per signal. Methods are concurrency-safe; the hot path (consumer.ConsumeX + worker outcome) calls them with no locking overhead beyond an atomic add.

func (s *Stats) Failed(sig Signal) int64

func (s *Stats) Forwarded(sig Signal) int64

func (s *Stats) RecordFailed(sig Signal, n int)

func (s *Stats) RecordForwarded(sig Signal, n int)

type StderrWriter ¶ added in v1.14.0

type StderrWriter struct {
	// contains filtered or unexported fields
}

StderrWriter owns os.Stderr in TTY mode and interleaves the live stats block with one-shot lifecycle messages.

The writer maintains its own rolling per-signal rate history independent of the RateSampler — the stats channel only carries the current snapshot, not the full history, so the writer accumulates samples as they arrive. This decouples the writer from the sampler beyond the channel.

func NewStderrWriter(out io.Writer, fd int) (*StderrWriter, chan SnapshotWithRate, chan LifecycleEvent)

func (w *StderrWriter) Run(ctx context.Context, statsCh <-chan SnapshotWithRate, lifecycleCh <-chan LifecycleEvent)

type StdoutWriter ¶ added in v1.14.0

type StdoutWriter struct {
	// contains filtered or unexported fields
}

StdoutWriter drains an event channel and writes one OTLP/JSON line per event to its underlying writer. The single-goroutine ownership of os.Stdout guarantees one valid JSON record per line: concurrent emitters produce no interleaving because everyone funnels through one channel and one writer.

Each event triggers a marshal → write → newline → flush cycle, so a downstream consumer reading the stream line-by-line sees complete records without head-of-line buffering delays.

func NewStdoutWriter(out io.Writer) (*StdoutWriter, chan plog.Logs)

func (w *StdoutWriter) Run(ctx context.Context, ch <-chan plog.Logs) error

type WorkerPool ¶ added in v1.14.0

type WorkerPool struct {
	// contains filtered or unexported fields
}

WorkerPool drains the per-signal channels populated by ProxyConsumer (U13), calls the upstream Forwarder, classifies outcomes per KTD14, and surfaces failures via the Stats counters, the agent-mode event channel, and an at-most-once-per-throttle-window stderr line for credential issues.

Concurrency is one goroutine per signal — the transport's max-concurrent-requests semaphore (KTD3b) provides the actual outbound concurrency cap, and per-signal goroutines keep batch ordering intact for downstream sanity (e.g., a metrics burst arriving in order goes upstream in order, modulo SDK retries).

func NewWorkerPool(forwarder Forwarder, dataset *string, stats *Stats, emitter *Emitter, consumer *ProxyConsumer, lifecycleCh chan<- LifecycleEvent, decorator *Decorator) *WorkerPool

func (p *WorkerPool) Run(ctx context.Context)