Documentation
¶
Overview ¶
Package transform runs user-defined transformations over OTLP payloads at the exporter seam: every batch a pipeline exports passes through the active program before buffering, so spooled bytes are final and replays are deterministic across reloads.
Programs are Starlark (pure-Go, hermetic by construction — no I/O, no imports; see engine.go). The transforms file is SEPARATE from the agent config and hot-reloads: edits compile-then-commit atomically (a broken edit keeps the last good program running, counted and warned), so transformation logic changes without a pod restart.
Cost model: one Starlark invocation per exported BATCH per signal, PLUS a post-script prune sweep, PLUS — for a payload NOT marked with Handoff — one deep copy of that batch, and the copy, not the script, is what the copying path costs. Measured on a 1024-record log batch with a `def transform(batch): return` script: ~249µs total, of which ld.CopyTo is ~226µs (91%) and the whole Starlark call is 245ns; a 10,000-point metrics payload (one promscrape chunk) is ~2.05ms and 2.3 MB for the same no-op. A payload marked Handoff(ctx) — the producer's promise that it rebuilds from source on failure rather than re-offering the object (see handoff.go for the roster) — runs the script IN PLACE and skips the copy, but NOT the prune: that sweep visits every record, span or data point and looks the drop marker up in its attributes whether or not the script called drop(), so it costs O(elements x attributes) and, on the in-place path, dwarfs the invocation (a 10k-point, 8-attribute chunk: 0.2-0.38 ms of prune against ~0.4µs of call, some 5-8% of that chunk's proto marshal). It cannot be skipped when no drop() ran: the prune is PRESENCE-based, and a marker that reached the payload through a door other than drop() (a logAttributes lift, for one) must be pruned all the same — gating the sweep would make that record's fate depend on whether another record in the batch was dropped. The copy remains the CORRECTNESS requirement for every unmarked producer, which re-exports the SAME object on retry (see the comment above ExportLogs). The tailer takes a third shape: it transforms its just-built batch once via TransformLogs and retries through Inner(), so its retry loop neither re-copies nor re-runs the script.
Within a batch, records ARE lazy: the iterators walk positions rather than materializing host objects, and each object resolves only the fields the script touches (~1µs per touched record), so a script that breaks out early pays only for what it visited. The per-line/per-sample hot paths never see any of this — pipelines without transforms don't even get the wrapper installed.
Index ¶
- Constants
- func Consumed(ctx context.Context) context.Context
- func HandedOff(ctx context.Context) bool
- func Handoff(ctx context.Context) context.Context
- func IsConsumed(ctx context.Context) bool
- func Reload(ctx context.Context, w *Wrapper, path string, poll time.Duration, ...)
- type Config
- type Exporter
- type MetricEmitter
- type Parsed
- type Program
- type TracesExporter
- type Wrapper
- func (w *Wrapper) Active() *Program
- func (w *Wrapper) AdmitResource(attrs pcommon.Map) bool
- func (w *Wrapper) ExportLogs(ctx context.Context, ld plog.Logs) error
- func (w *Wrapper) ExportMetrics(ctx context.Context, md pmetric.Metrics) error
- func (w *Wrapper) ExportTraces(ctx context.Context, td ptrace.Traces) error
- func (w *Wrapper) Fork(next Exporter, nextTraces TracesExporter) *Wrapper
- func (w *Wrapper) Inner() Exporter
- func (w *Wrapper) ParseLine(line string) (Parsed, bool)
- func (w *Wrapper) SampleDecider() func(tailsample.Trace) (sample, abstain bool)
- func (w *Wrapper) SetMetricEmitter(m MetricEmitter)
- func (w *Wrapper) Swap(p *Program)
- func (w *Wrapper) TransformLogs(ld plog.Logs) (dropped int, err error)
- func (w *Wrapper) TransformTargets(ts []kubemeta.ScrapeTarget) []kubemeta.ScrapeTarget
Constants ¶
const DropMarker = "__kubescrape_drop__"
DropMarker flags an element for post-run pruning. Logs and spans carry it as a record/span attribute; metrics carry it in the pdata Metadata map, which keeps it OFF the data-point attributes a rename could touch, so a drop-then-rename cannot un-drop it. The marker never survives into the export regardless — a marked metric is pruned whole and a script error discards the transform's whole working copy — so it never reaches the wire (the Metadata field IS an OTLP wire field, unlike an earlier claim here).
Exported because the key is RESERVED plumbing, like route.ScriptMarker: the prune is presence-only and cannot tell a script's mark from a sender's, so a marked element arriving ON THE WIRE would be deleted — and counted as an operator-intended kubescrape_transform_dropped_total — whenever any script for its signal is active. The ingest receivers strip it at first receipt (otlpingest.ServerConfig.ReservedAttrs, wired in cmd/kubescrape-agent), and they take the spelling from here so the two cannot drift.
Variables ¶
This section is empty.
Functions ¶
func Consumed ¶
Consumed is Handoff plus a promise about the RECORDS: a failed attempt is final for them — no retry offers them to the script again, because the attempt consumed their source (a take()n chunk, a window reset as it was read) or because the next attempt renders NEW points rather than these. The Wrapper then counts a failed forward's drops, which would otherwise be counted nowhere. A producer whose retry brings the same records back — an ingest sender's retransmission, a retained sample re-rendered at its own timestamp — marks Handoff alone, or its drops count once per attempt.
func Handoff ¶
Handoff marks ctx as carrying a payload its caller relinquishes: after the export call returns — success or failure — the caller neither re-sends the object's contents nor reads them back, and any retry rebuilds the payload from source data. The Wrapper then runs the transform script IN PLACE instead of on a deep copy. See the package's handoff.go doc for who may mark and why; a producer that re-offers the same object on retry must not.
The marker is a context VALUE, so it survives context.WithoutCancel and deadline wrappers — the same contract otlpexport.Own rides on.
func IsConsumed ¶
IsConsumed reports whether ctx was marked by Consumed.
Types ¶
type Config ¶
type Config struct {
// Logs/Metrics/Traces each hold one Starlark script defining
// transform(batch). Empty = passthrough for that signal.
Logs string `json:"logs,omitempty"`
Metrics string `json:"metrics,omitempty"`
Traces string `json:"traces,omitempty"`
// The hook sections (hooks.go), each optional and each defining its own
// function; all fail open on script errors:
//
// Ingest defines admit(resource) — per pushed resource, before
// enrichment; False removes it (the operator's per-sender policy).
Ingest string `json:"ingest,omitempty"`
// Targets defines target(t) — per fetched scrape target; t.drop()
// removes it, t.path is writable.
Targets string `json:"targets,omitempty"`
// Sample defines decide(trace) — the tail-sampling `script` policy body
// (True samples, False drops, None abstains).
Sample string `json:"sample,omitempty"`
// Parse defines parse(line) — plain log sources flagged parseScript run
// it per line; a dict return may set body/severity_text/time_unix_nano.
Parse string `json:"parse,omitempty"`
}
Config is the transforms file: one optional program per signal.
type Exporter ¶
type Exporter interface {
ExportLogs(ctx context.Context, ld plog.Logs) error
ExportMetrics(ctx context.Context, md pmetric.Metrics) error
}
Exporter is the downstream the wrapper forwards to (otlpexport.Client and Buffered both satisfy it; traces are optional at wrap time).
type MetricEmitter ¶
type MetricEmitter interface {
EmitDirect(name string, value float64, labels map[string]string, resource pcommon.Map) error
}
MetricEmitter is the emit_metric bridge target: the logMetrics set's EmitDirect. An interface so transform needs no metrics import in its API.
type Program ¶
type Program struct {
// Hash identifies the compiled config (content hash of the file), served on
// /debug/transforms so per-node convergence after a reload is observable.
Hash string
// contains filtered or unexported fields
}
Program is a compiled, immutable set of per-signal transforms. Swapped atomically on reload; in-flight batches finish on the program they started with.
func Compile ¶
Compile parses and compiles the whole config; any error rejects the WHOLE config (never "half the signals applied").
func CompileFile ¶
CompileFile reads and compiles path.
type TracesExporter ¶
TracesExporter is the optional traces downstream.
type Wrapper ¶
type Wrapper struct {
// contains filtered or unexported fields
}
Wrapper applies the active program to every batch, then forwards. The program pointer is swapped by the reloader; each Export loads it once.
func Wrap ¶
func Wrap(next Exporter, nextTraces TracesExporter, initial *Program) *Wrapper
Wrap builds a Wrapper forwarding to next (nextTraces may be nil when the exporter cannot ship traces).
func (*Wrapper) AdmitResource ¶
AdmitResource runs the ingest hook's admit(resource) for one pushed resource: false means the receiver removes the resource before enrichment (the operator's per-sender policy — the honest mitigation for senders a built-in bound can only slow down). No hook, a non-False return, or a script error all admit.
The resource is READ-ONLY to the script (roAttrsView; an assignment is a script error, which admits). admit is a predicate, and a writable view broke the fail-open contract every other hook keeps: a script that wrote and then erred forwarded its write, where the targets hook restores the pre-script target on error precisely because "the hook did nothing" excludes a half-applied write. It would also have let a script put back, AFTER the receipt-time strip, the identity keys that strip exists to remove.
func (*Wrapper) ExportLogs ¶
ExportLogs transforms then forwards.
func (*Wrapper) ExportMetrics ¶
ExportMetrics transforms then forwards.
func (*Wrapper) ExportTraces ¶
ExportTraces transforms then forwards.
func (*Wrapper) Fork ¶
func (w *Wrapper) Fork(next Exporter, nextTraces TracesExporter) *Wrapper
Fork returns a wrapper over a DIFFERENT downstream that shares this one's program AND its emit_metric target: one reloader keeps both current, and a SetMetricEmitter on either reaches both — they can never diverge. The agent uses it for the chain carrying its own metrics, which skips the namespace router but must still see the operator's transforms.
func (*Wrapper) Inner ¶
Inner is the exporter below the transform layer, for a producer that applies the transform itself (TransformLogs) and must not pay it again on the send path.
func (*Wrapper) ParseLine ¶
ParseLine runs the parse hook on one plain-source line. ok=false — no hook, a None return, or a script error — leaves the line untouched.
func (*Wrapper) SampleDecider ¶
func (w *Wrapper) SampleDecider() func(tailsample.Trace) (sample, abstain bool)
SampleDecider adapts the sample hook to tailsample's injected script policy: decide(trace) -> True samples, False drops, None (or an error) abstains to the next policy. nil when the section is absent at STARTUP — the policy compiler then refuses `type: script`, at config time.
A hot reload can remove the section after that check, and it cannot be refused at reload time: a sectionless file still compiles, so it commits as an APPLIED reload (last-good retention covers compile failures only), and this package cannot know whether tailsample is configured with a `type: script` policy that depends on the hook — the UsesScript/HasSample cross-check is the agent's, at startup, and never re-runs. The closure then keeps failing open (abstain, the hooks' contract) but as loudly as the error arm — every decision counted into kubescrape_transform_errors_total{signal="sample"} plus a throttled warn — because a silent abstain turns a load-bearing sampling policy off with zero signal: traces fall through to the next policy or the default drop while every reload metric reads green.
func (*Wrapper) SetMetricEmitter ¶
func (w *Wrapper) SetMetricEmitter(m MetricEmitter)
SetMetricEmitter wires the emit_metric bridge. It may be called after Fork — forks alias the same slot — and the caller keeps the typed-value guard: a nil *DynamicMetricSet boxed into the interface would defeat the builtin's own nil check, so main only calls this with a non-nil concrete value.
func (*Wrapper) Swap ¶
Swap installs a new program (compile-then-commit: callers only pass programs that compiled whole).
func (*Wrapper) TransformLogs ¶
TransformLogs runs the active logs program on ld IN PLACE (drop marks swept, empty groups pruned) and reports how many records the script dropped, or its error; no program means no-op. It is the tailer's seam: the tailer's retry loop re-sends the SAME batch, so it can neither mark Handoff (the contract forbids the re-send) nor export through the wrapper (every attempt would re-copy and re-run the script) — instead it transforms the batch it just built exactly once, here, and retries through Inner(). The caller decides what an emptied payload means (the tailer commits its offsets without a send) and treats an error like a failed export, so a re-read re-runs the — possibly hot-reloaded — program.
The drops are REPORTED, not counted: the caller counts them into kubescrape_transform_dropped_total{signal="logs"} once the batch's records are settled (its offsets commit, or it is dropped as permanently rejected). Counting here counted every rewind: a failed flush rewinds the files, the next sweep rebuilds the batch from the same bytes and the script drops the same records again, so one intended drop read as one per failed flush of the outage.
func (*Wrapper) TransformTargets ¶
func (w *Wrapper) TransformTargets(ts []kubemeta.ScrapeTarget) []kubemeta.ScrapeTarget
TransformTargets runs target(t) over each fetched scrape target: t.drop() removes it, t.path is writable (the URL re-renders). Errors keep the target untouched.
The input slice is metaclient's CACHED value (shallow-copied out under the treat-as-immutable contract, served again on every cache hit and 304 revalidation), so neither the slice nor its elements may be written: compacting into ts[:0] shifted survivors over the cached backing array (leaving a stale tail a later cycle scraped twice), and a path rewrite through &ts[i] compounded across cycles. Each target is copied BY VALUE before the script may touch it — sufficient because targetObj's only writable surface is the top-level Path string (SetField), plus the URL re-render here — and survivors go into a fresh slice. Once per scrape cycle, not a hot path.
Source Files