A telemetry pipeline is the layer between the systems that emit logs, metrics, traces and events and the tools that store or analyse them. It collects that data, cleans it into one shape, drops what nobody needs, and routes each record to every destination that does. Built well, it lets you choose what reaches your observability platform and shows you where each record went.
This guide builds a small telemetry pipeline with Expanso Edge, runs it, and counts the result. If you want the definitions first, What Is Telemetry Data? covers the data itself and What Is an Observability Pipeline? covers the product category. This post is about building one you can check.
What a telemetry pipeline does
Every telemetry pipeline, whatever product runs it, does five jobs. Each one has a characteristic way of failing.
Collect. Read telemetry from where it is produced: log files, message brokers such as MQTT or Kafka, HTTP endpoints, cloud services. The failure is a source nobody connected, which looks exactly like a quiet system.
Normalise. Different services write the same thing differently. One logs ERROR, another error, a third FATAL. Until severity, timestamps and service names mean the same thing everywhere, every rule downstream is a guess.
Filter and reduce. Drop debug output, health checks and duplicates before anyone pays to move or store them. The failure is a rule that silently drops records it does not understand, which turns a parsing problem into a quieter dashboard. We covered that trap in Filter Noisy Logs Without Losing the Errors.
Route and fan out. Send each record where it is useful. Errors go to alerting, everything worth keeping goes to search, and the same stream can go to an archive as well. Fan-out gives each destination its own copy, which raises a design question: when one destination is down, should the others wait for it or carry on without it?
Deliver and account for failure. Records that cannot be processed need somewhere to go with the reason attached. A pipeline that cannot tell you what it rejected cannot tell you what it lost.
Telemetry pipeline or observability pipeline?
The terms overlap almost completely. “Observability pipeline” is usually the name of a product category, such as Cribl Stream or Datadog Observability Pipelines. “Telemetry pipeline” more often describes the data path itself, from the first log line to its last destination. The OpenTelemetry Collector uses the same structure under different names: receivers, processors and exporters.
Where a telemetry pipeline should run
Most telemetry pipelines run centrally: agents ship everything to a cluster, and the cluster filters it on the way into the observability platform. That reduces what you index, but you have already paid to move every byte, and the data has already left the network where it was produced.
Running the pipeline next to the source changes both. A health check dropped on the host that produced it never crosses a network. A record redacted at the edge never exists in plaintext anywhere else. Expanso runs pipeline jobs on nodes you choose, selected by label, so the same job can run on a server, a gateway or a cloud VM without changing its configuration. The telemetry control-plane architecture goes further into separating collection from routing decisions.
Build a telemetry pipeline you can verify
The example below takes twelve synthetic telemetry records from three services and does all five jobs. It runs on one machine with no account, using a local-mode Expanso Edge node. We ran it with expanso-edge and expanso-cli v2.1.21 on macOS.
Install both tools from the installation guide, then create a working directory:
mkdir -p telemetry-demo/out
cd telemetry-demoThe input
Save these twelve lines as telemetry.jsonl. They are deliberately untidy: mixed-case severities, two health checks, a record with no service, a line that is not JSON, and a severity (notice) the pipeline has not been told about.
{"ts":"2026-09-27T12:00:00Z","service":"checkout","level":"INFO","msg":"order placed","order_id":"o-1001"}
{"ts":"2026-09-27T12:00:01Z","service":"checkout","level":"DEBUG","msg":"cache lookup","key":"cart:88"}
{"ts":"2026-09-27T12:00:02Z","service":"gateway","level":"info","msg":"GET /healthz 200","path":"/healthz"}
{"ts":"2026-09-27T12:00:03Z","service":"auth","level":"Warning","msg":"token refresh slow","latency_ms":1840}
{"ts":"2026-09-27T12:00:04Z","service":"checkout","level":"ERROR","msg":"payment provider timeout","order_id":"o-1002"}
{"ts":"2026-09-27T12:00:05Z","service":"gateway","level":"info","msg":"GET /healthz 200","path":"/healthz"}
{"ts":"2026-09-27T12:00:06Z","service":"gateway","level":"INFO","msg":"GET /api/orders 200","path":"/api/orders","latency_ms":38}
{"ts":"2026-09-27T12:00:07Z","level":"ERROR","msg":"disk full"}
checkout: connection reset by peer
{"ts":"2026-09-27T12:00:09Z","service":"auth","level":"debug","msg":"session lookup","session":"s-17"}
{"ts":"2026-09-27T12:00:10Z","service":"checkout","level":"FATAL","msg":"worker crashed","order_id":"o-1003"}
{"ts":"2026-09-27T12:00:11Z","service":"auth","level":"notice","msg":"password policy updated"}Before running anything, write down what should happen. Two debug records and two health checks should be dropped. Three records should be rejected with a reason. The remaining five should reach both destinations, and the two errors among them (ERROR and FATAL) should also raise alerts.
The pipeline
Save this as telemetry-pipeline.yaml:
name: telemetry-pipeline
type: pipeline
restart_policy: never
config:
input:
file:
paths: ["./telemetry.jsonl"]
scanner:
lines: {}
pipeline:
processors:
- mapping: |
let rec = content().string().parse_json().catch(null)
let raw_sev = $rec.level.or("").string().lowercase()
let sev = match $raw_sev {
"warning" => "warn"
"fatal" => "error"
"critical" => "error"
_ => $raw_sev
}
let known = ["debug", "info", "warn", "error"]
let reason = if $rec == null {
"not valid JSON"
} else if $rec.service.or("") == "" {
"missing service"
} else if !$known.contains($sev) {
"unknown severity: " + $raw_sev
} else {
null
}
root = if $reason != null {
{"rejected": true, "reason": $reason,
"raw": content().string()}
} else if $sev == "debug" || $rec.path.or("") == "/healthz" {
deleted()
} else {
$rec.without("level").merge({"severity": $sev})
}
output:
switch:
cases:
- check: this.rejected.or(false)
output:
file:
path: "./out/rejects.jsonl"
codec: lines
- check: this.severity == "error"
continue: true
output:
file:
path: "./out/alerts.jsonl"
codec: lines
- output:
broker:
pattern: fan_out
outputs:
- file:
path: "./out/observability.jsonl"
codec: lines
- file:
path: "./out/archive.jsonl"
codec: linesThe single Bloblang mapping normalises and filters. It parses each line, normalises severity so Warning becomes warn and FATAL becomes error, and decides whether the record is rejected, dropped or kept. The switch output does the routing: rejects go to their own file, errors go to alerts and, because that case sets continue: true, on to the next case as well. The last case is a broker output with pattern: fan_out, which sends every kept record to both destinations. restart_policy: never makes this one-shot job stop after one failed execution instead of retrying.
Run it
Check the configuration offline first. This contacts nothing:
expanso-edge validate telemetry-pipeline.yaml[OK] telemetry-pipeline.yaml: validStart a local-mode node from the same directory. It serves its API on port 9010 and keeps its state in ./edge-data:
expanso-edge run --local --data-dir ./edge-dataIn a second terminal, in the same directory, deploy the job and check it:
expanso-cli job deploy telemetry-pipeline.yaml \
--endpoint http://localhost:9010
expanso-cli job list --endpoint http://localhost:9010
wc -l out/*.jsonlThe relative paths in the job resolve against the directory the node was started in, which is why both terminals need to be in telemetry-demo. Stop the node with Ctrl+C when you are done; deleting the directory removes everything the example created.
What the run produced
The job finished in the completed state and the files held exactly what we wrote down in advance:
2 out/alerts.jsonl
5 out/archive.jsonl
5 out/observability.jsonl
3 out/rejects.jsonl
15 totalTwelve records went in. Four were dropped as noise, three were rejected, and five were kept and delivered to both destinations. Two of those five were also written to alerts. The line order inside each file can differ from the input, so compare contents rather than positions.
The rejects file is the part worth reading:
{"raw":"{\"ts\":\"2026-09-27T12:00:07Z\",\"level\":\"ERROR\",\"msg\":\"disk full\"}","reason":"missing service","rejected":true}
{"raw":"{\"ts\":\"2026-09-27T12:00:11Z\",\"service\":\"auth\",\"level\":\"notice\",\"msg\":\"password policy updated\"}","reason":"unknown severity: notice","rejected":true}
{"raw":"checkout: connection reset by peer","reason":"not valid JSON","rejected":true}The first reject is an ERROR saying the disk is full. A pipeline that quietly dropped malformed records would have lost it, and the alerting destination would have looked healthy. Here it sits in a file with its reason and its original text, where a person can find it.
What building this taught us
The version above is the fourth one we ran. Each earlier version failed in a way a production telemetry pipeline can.
A job can route every record correctly and still report failure. The first version rejected records by raising an error in the mapping (throw) and routed errored records to the rejects file. The files were right, but on v2.1.21 the node marked the run unhealthy and the job ended failed. A second version recovered the error with a catch processor, and the job still ended failed. Computing the rejection as data, a reason field, fixed it. That matters when the job state drives alerts or retries.
Validation checks structure, not behaviour. The third version had an if with no else, so the reason variable was undefined for every good record. expanso-edge validate passed it. At run time the mapping failed on the nine good records, they reached both destinations unprocessed, and the job ended failed. Validate first, then run against records whose outcome you already know.
A finished job is not proof that data arrived. The Expanso recipes make the same point from the other direction: a job can end completed with nothing written. Count the records where they landed.
From the example to real telemetry
The example reads a file and writes files so that you can see everything. In production the shape stays the same and the components change.
Sources. Swap the file input for the transport your telemetry already uses. The sensor telemetry over MQTT recipe subscribes to a wildcard topic, converts units and routes alerts and rejects the same way this example does. The component catalog also lists inputs for Kafka, HTTP, tailed files, cloud queues and OPC UA.
Destinations. Replace the file outputs with the systems you run: an HTTP endpoint, object storage, a message bus, a search cluster. Adding or swapping a destination is a change to this file, not to the applications producing the telemetry. Decide what an outage should do. By default, a fan_out broker retries a failed output until it succeeds and lets back pressure from one output block the stream, which favours completeness; wrapping an output in drop_on lets the others keep flowing instead. The webhook fan-out recipe was run with a receiver that answered 503 twice, and every receiver still got all ten events.
Placement. Deploy the same job through Expanso Cloud instead of local mode and add a selector so it runs on the nodes next to the data. What local mode leaves out is the fleet side: multi-node coordination and central management.
This run has limits. It proves how one job handled twelve synthetic records on one machine. It is not a throughput benchmark, and it did not test buffering through a network outage, which you should test against your own destinations before relying on it. The log reduction recipe shows a larger synthetic run with its limits stated the same way.
How to choose a telemetry pipeline tool
Five questions separate telemetry pipeline tools more reliably than feature lists:
- Where does it run? Only in a central cluster, or also on the hosts, gateways and sites where telemetry is produced?
- What happens to a record it cannot parse? Is it kept somewhere with a reason, or does it disappear?
- Can one stream feed several destinations, and what happens when one fails? Do the others wait, or carry on without it, and can you choose?
- Can you change destinations without re-instrumenting? Switching an analytics vendor should be a pipeline change, not an application change.
- How do you prove what it did? Can you count, per destination, what arrived?
Getting started with Expanso takes few dependencies: this example needed one agent binary, one CLI and no account. The first five nodes are free on the pricing page. For a side-by-side look at the alternatives, see Cribl alternatives, and for log-specific pipelines, the log processing use case.
FAQ
What is a telemetry pipeline?
A telemetry pipeline collects logs, metrics, traces and events from the systems that produce them, cleans them into a consistent shape, filters out what is not needed, and routes each record to the tools that use it, such as alerting, search and long-term storage.
What is the difference between a telemetry pipeline and an observability pipeline?
In practice they describe the same thing. “Observability pipeline” is more often the name of a product category; “telemetry pipeline” more often describes the data path itself.
Do I need OpenTelemetry to build a telemetry pipeline?
No. OpenTelemetry standardises how applications emit telemetry and provides its own collector. A telemetry pipeline can read whatever your systems already produce. The example in this post reads JSON lines from a file.
Where should a telemetry pipeline run?
As close to the source as the data allows. Filtering centrally reduces what you store, but you have already paid to move the data. Filtering at the source means dropped records never cross the network.
How do I know my telemetry pipeline is not dropping data I need?
Send records whose outcome you already know through the pipeline, send anything it cannot parse to a rejects destination with the reason, and count the records at each destination rather than trusting the job’s status.
Can I try Expanso without an account?
Yes. expanso-edge validate and local mode, as used in this post, run on your own machine and need no account. The first five nodes are free if you move to Expanso Cloud.
