06 - Mapping and outputs¶

Mapping language: CEL¶

Mappings use CEL (Common Expression Language, cel.dev/cel-go). Rationale in ADR 0002. Key properties: non Turing-complete (termination guaranteed), type-checked at compile time (errors surface in tapyr validate), cost-estimable, fast once compiled, extensible with custom functions.

CEL environment¶

Variables available in extract expressions:

Variables are declared only where they are meaningful, so validate rejects, for example, item in extract.items:

Custom functions:

The has() macro and optional types (?. syntax) are enabled to handle missing fields.

Compilation limits, configured in settings.mapping: max expression cost (maxCost, default 1000000), max items per run (maxItemsPerRun, default 100000), max samples per run (maxSamplesPerRun, default 200000).

A metric value expression may return int, uint, double or bool (true is 1); the result is converted to float64.

Extract block¶

extract:
  items: "response.body.items"          # CEL -> list; default: response.body if it is a list
  filter: "item.status != 'unknown'"    # optional CEL -> bool
  metrics:
    - name: aruba_ap_clients
      type: gauge                       # gauge | counter
      help: Number of clients connected to the AP
      unit: ""                          # optional, Prometheus base units preferred
      value: "toFloat(item.client_count)"
      labels:
        serial: "item.serial"
        site: "default(item.site, '')"
      timestamp: "parseTimestamp(item.last_modified)"   # optional
      skipIf: "!has(item.client_count)"                 # optional CEL -> bool
  info:
    - name: aruba_ap_info
      labels:
        serial: "item.serial"
        model: "item.model"
        firmware: "item.firmware_version"
  inventory:                            # for role: inventory | both
    key: "item.serial"
    fields:
      serial: "item.serial"
      site: "item.site"
  post: []                              # optional metrics computed over all items

• info metrics are gauges with value 1 and descriptive labels (standard _info pattern). Descriptive attributes belong there, not on every metric.
• post entries can aggregate across items, e.g. value: "double(size(responses[0].body))" for a count.
• Account labels, target labels and job labels are added to every sample. Conflicts: job > target > account > mapping (settings.mapping.labelConflict: error turns conflicts into run errors).
• Empty-string label values are dropped (Prometheus semantics).
• Label expressions return strings; ints, doubles and bools are converted to their string form, null gives an empty (dropped) label. Other types are mapping errors.
• A mapping error on one item or one metric skips that sample and is counted (tapyr_mapping_errors_total); an error in extract.items fails the run.

Window values and counters¶

Many APIs return values aggregated over a window ("bytes over the last 5 minutes") rather than cumulative counters.

• Default: expose them as gauges named explicitly (meraki_uplink_sent_bytes_5m). This is honest and simple.
• Optional reconstruction: type: counter with accumulate: true sums successive window values into a monotonic counter persisted in the state store. Requires non-overlapping windows (windowField or job interval aligned). Counter resets on restart are avoided by persistence; gaps are documented.
• Native cumulative counters from APIs map directly to type: counter.

Internal metric model¶

Each sample: metric name, type, unit, help, label set (sorted), value (float64), timestamp (source timestamp or collection time), plus provenance (job instance, run id). Converted to each output format at the edge.

Series registry: staleness and deduplication¶

Per job instance, the registry keeps the set of series emitted by the last successful run and the last timestamp sent per series and output.

• Staleness: a series present in the previous successful run but absent from the current one is stale.
• Prometheus pull: removed from the exposition.
• Remote write: a stale marker sample (the Prometheus special NaN value 0x7ff0000000000002) is sent once.
• OTLP: one data point with the FLAG_NO_RECORDED_VALUE flag is sent once.
• When a job instance is removed (inventory change, config reload), all its series become stale.
• A failed run does not cause staleness (series stay as-is until staleAfter, default 3x effective interval, after which they are removed from pull exposition and marked stale in push outputs).
• Deduplication: when a sample carries a source timestamp lower than or equal to the last timestamp sent for that series and output, it is dropped (overlapping API windows). Counted in tapyr_samples_deduplicated_total.

Implementation notes¶

• The registry is keyed by job instance. Pull exposition follows it directly: a successful run replaces the series of its instance; an instance whose last success is older than staleAfter is withdrawn; a removed instance is withdrawn at once.
• Push outputs receive, per output, the samples that pass deduplication and the series to mark stale; the registry computes both.
• Deduplication only concerns samples with a source timestamp; samples stamped with the collection time are never deduplicated.
• The registry is snapshotted with the ledger (every 5 seconds and on shutdown): the series of every instance, and the last source timestamp sent per series and output.
• The source timestamps of a series are forgotten when it goes stale, which bounds the registry to the live series. After a restart, instances of the snapshot that are no longer scheduled (removed from the configuration, or children waiting for the first run of their parent without a cached inventory) are withdrawn at the first check.

Timestamps¶

• If timestamp is set in the mapping, the sample carries the source timestamp. Otherwise, the collection time (response received).
• Push outputs always send timestamps.
• Prometheus pull exposition omits timestamps by default (exposeTimestamps: false), as recommended by Prometheus; the latest value per series is exposed.
• Out-of-order caveat: Prometheus remote write receivers may reject out-of-order samples unless out-of-order ingestion is enabled; VictoriaMetrics accepts them. See docs/outputs.md; rejected samples are counted in tapyr_output_samples_dropped_total{reason="out_of_order"}.

Outputs¶

All outputs implement:

type Output interface {
    Name() string
    Enqueue(batch model.Batch)           // non-blocking; drops with accounting when the queue is full
    Run(ctx context.Context) error       // sender loop
    Flush(ctx context.Context) error     // on shutdown
}

prometheus (pull)¶

• Served by the main HTTP server on /metrics (collected data only; Tapyr's own metrics are on /metrics/self, see 07-self-observability.md).
• Holds the samples of the last successful run of every job instance: a successful run replaces the series of its instance, a failed run keeps them (until staleAfter, see above). Content negotiation: Prometheus text format and OpenMetrics, gzip-compressed when the scraper accepts it (Accept-Encoding).
• HELP and TYPE from the mapping. Metric name conflicts (same name, different type) are rejected at validation.

remote_write¶

• Prometheus remote write 1.0 (protobuf + snappy), HTTP POST.
• Batching (maxBatch, flushInterval), bounded queue (capacity, drop-oldest with accounting), retries with backoff on 5xx and 429 (respecting Retry-After), no retry on 4xx other than 429.
• Metadata (HELP, TYPE) sent periodically.
• Compatible targets: Prometheus (with the remote write receiver enabled), VictoriaMetrics (/api/v1/write), Grafana Mimir, Thanos Receive, Cortex.
• Remote write 2.0 is not supported yet.

otlp¶

• OTLP metrics over HTTP (protobuf) or gRPC.
• Payloads are built directly with pmetric (go.opentelemetry.io/collector/pdata), not with the OpenTelemetry metrics SDK, which aggregates and cannot carry source timestamps (ADR 0008).
• Mapping: gauge to OTLP Gauge; counter to OTLP Sum (monotonic, cumulative, start time = series first seen or persisted start). Labels to data point attributes. Resource attributes: service.name=tapyr, service.instance.id, tapyr.provider, tapyr.account (configurable), plus resourceAttributes.
• Batching, bounded queue and retries as for remote write; honors OTLP partial success responses and throttling hints.

Implementation notes¶

• Push engine (internal/output/push), shared by remote_write and otlp: the queue counts samples, stale markers included; when full, the oldest samples are dropped (queue_full). One request is in flight per output, so samples of a series leave in order. A batch is sent when it reaches maxBatch or when its oldest sample waited flushInterval. Network errors, 5xx and 429 are retried with exponential backoff (1s doubling up to 1m, full jitter), honoring Retry-After (capped at 5 minutes); a batch is retried until it succeeds or is pushed out by newer samples, so a slow backend costs memory up to capacity, never collection. Other 4xx drop the batch: out_of_order when the response mentions out of order, too old or out of bounds samples, rejected otherwise. On shutdown, Flush sends what is queued within the shutdown deadline, without retry delays beyond it.
• External labels are added to every sample; a label set by the sample wins.
• Stale markers carry the time the series went stale (the run that no longer produced it, or the withdrawal). The output remembers the type and start time of the series it sent, so a marker is a gauge point when the series is unknown (after a restart).
• Remote write sends Content-Encoding: snappy, Content-Type: application/x-protobuf, X-Prometheus-Remote-Write-Version: 0.1.0 and User-Agent: tapyr/<version>; the time series of a request are sorted by labels with their samples in time order. Metadata (TYPE, HELP, unit) of every metric family seen is sent in a request of its own on the first send and every 1 minute.
• OTLP groups samples by resource (provider and account of the job instance) and metric name; the instrumentation scope is github.com/kentrow/tapyr with the Tapyr version. service.instance.id is a random identifier created on first start and kept in the state store (resourceAttributes may override it). HTTP: POST to the endpoint, with /v1/metrics appended when the endpoint has no path; gRPC: http:// endpoints are plaintext, https:// use TLS with the system roots. Retryable: HTTP 429, 502, 503, 504 and gRPC UNAVAILABLE, RESOURCE_EXHAUSTED (with RetryInfo); rejected data points of a partial success are counted as rejected. The start time of a cumulative sum is the first time the output saw the series in this process, or its earliest timestamp when a source timestamp is older; a restart starts the sums again, which receivers treat as a reset.
• Counter reconstruction (accumulate: true): per job instance and series, the store keeps the running total and the last window counted (the windowField value, else the sample timestamp). A sample whose window was already counted adds nothing; negative window values are mapping errors. The totals are persisted with the ledger (table accumulators) and dropped when the series goes stale or its instance is removed, a counter reset receivers handle.

Multiple outputs¶

A job sends to all outputs by default, or to the subset listed in spec.outputs. Each output has its own queue: a slow backend never blocks collection or other outputs.