How OpenETL works

This page explains how OpenETL applies pipeline definitions, executes jobs and commits data. It identifies the packages responsible for each operation. CONTRIBUTING.md explains how to build and test the code.

The partsCopy Anchor Link

internal/daemon runs the roles of the openetld program. server serves the API and console. orchestrator executes workflows and assigns jobs. runner executes attempts. dev combines all three roles in one process.

The APICopy Anchor Link

internal/api serves the API defined in proto/openetl/api/v1 over Connect, gRPC, gRPC-Web and REST. Vanguard transcodes the REST routes from google.api.http rules. These rules also generate internal/api/openapi.yaml.

The CLI and console use this API. internal/webui serves the built console from the same origin.

Runners call DataPlaneService, defined in proto/openetl/dataplane/v1. A runner token limits each runner to its data plane's work. internal/auth manages API keys, runner tokens and roles.

The metadata storeCopy Anchor Link

internal/store is the metadata database, on Postgres: its goose migrations, and the queries from which sqlc generates internal/store/db. Every query outside the orgs table is scoped by org_id, which TestQueriesAreOrgScoped checks.

internal/secrets encrypts secret values before storage. Each organization has a data key that encrypts values with AES-256-GCM. The installation's local or KMS key wraps those data keys.

Secrets can also reference AWS Secrets Manager, GCP Secret Manager or Azure Key Vault. Runners resolve these references with their cloud identity through internal/secretstore.

OrchestrationCopy Anchor Link

Workflows manage schedules, continuous-pipeline supervisors, syncs and operations such as source discovery. They use the interface in internal/orchestration. The default DBOS backend stores workflow state in Postgres. Temporal is the other supported backend.

Steps perform I/O; workflow code does not. After a failure, workflow replay returns the recorded results of previous calls. Workflow code therefore cannot use its own clocks, randomness or goroutines. Changes that alter an existing workflow's execution path require Context.Version.

internal/orchestration/dispatch sends work to data planes through the API. It uses Postgres dispatch records, leases and long polling to deliver results to waiting workflows. sizing selects the resources for the next attempt. scaling adjusts continuous jobs when their lanes need more or fewer resources.

RunnersCopy Anchor Link

A runner (internal/runner/worker) registers with the API, polls for work, executes it and reports results. Each attempt goes to a runner in the pipeline's data plane and pool with sufficient capacity. A pod launcher can instead create a Kubernetes pod for one attempt. That pod runs openetld runner and obtains the attempt from the API. Install on Kubernetes explains sizes, pools and placement.

An attempt (internal/runner) hosts connectors, drives dataflow and coordinates checkpoints. Built-in connectors run inside the runner. External connectors run as separate processes and serve the connector protocol on Unix sockets. internal/connectorhost supplies host services, including HTTP retries and secrets.

The engineCopy Anchor Link

openetl-engine is the Rust data engine. The runner starts one engine process per job and communicates through two socket types. The control channel uses EngineService from proto/openetl/engine/v1. Data edges carry frames with a 12-byte header and Arrow IPC payload.

The runner sits between the connectors and the engine. Its source pump is the source connector's emitter, and sends what the source reads to the engine's inbound edges. On each lane, the engine:

• turns raw rows into Arrow, inferring each stream's schema from what it sees and only ever widening it, and coerces values to the types a source declares (crates/normalize);
• sets aside the records it can't take, with a reason, as JSON lines under the pipeline's _quarantine/ in object storage;
• applies the pipeline's schema policy, gives tables and columns their destination names, which never change once given, and adds the metadata columns, _openetl_id among them;
• holds back the columns and cells the destination can't take, and keeps their values for a later release;
• routes rows by key group to writer lanes, which write to the runner, with barriers in band, or stage Parquet files, for a destination that loads them and for the built-in parquet destination.

The runner's sink pumps hand each writer lane's batches to the destination connector's writer.

CheckpointsCopy Anchor Link

internal/checkpoint keeps each pipeline's checkpoint log under {org}/{pipeline}/_checkpoints/ in object storage. Records are immutable, and HEAD points to the latest record. Compare-and-swap updates use conditional writes on S3, GCS and Azure Blob Storage. This prevents two attempts from advancing the same HEAD concurrently.

An attempt claims the pipeline by writing HEAD with the next generation. An older active attempt, called a zombie, receives ErrFenced on its next HEAD write and stops. Each record contains stream resume states and schemas, lane committables, and staged or quarantined file references.

The same storage holds, next to the log, each attempt's log as JSON lines (_logs/), staged epochs (_staging/), held values (_held/) and quarantined records (_quarantine/).

The commit protocolCopy Anchor Link

Data is committed in epochs: the records between two checkpoints. For each one:

1. The coordinator requests barrier N. The source pump inserts it after each split's next resume point. This aligns the barrier with a position the source can resume from.
2. The barrier travels through the engine to each writer lane. After all its streams cross the barrier, each lane calls PreCommit(N) and returns a committable. The lane's data is durable. Stage-and-apply and transactional destinations keep it invisible until commit; deliver-and-acknowledge destinations have already delivered it.
3. The coordinator writes record N to the checkpoint log and moves HEAD to it. This is the commit point.
4. Lane 0's writer calls Commit(N). Stage-and-apply destinations make the epoch visible with stream markers protected by the destination fence. Commit is idempotent and can be retried. If retries fail, the next attempt's Recover applies the epoch from its recorded committables.
5. The source receives the positions it can acknowledge upstream, such as event offsets or a replication position. This notification is one checkpoint behind.

Commits run one at a time, in epoch order, while barriers for later epochs go on being placed.

After claiming the pipeline, an attempt calls its writer's Recover. This raises the destination fence to the new generation, preventing zombie commits. It then checks the epoch referenced by HEAD.

The epoch can be already applied, recoverable from its committables, or lost. If it is lost, the log returns to the preceding record. The attempt reads from the resume states in the selected record.

A delayed split can use a forced barrier without a resume point only when replay preserves the required delivery guarantee. Recovery from that checkpoint reads some records again. The test/sim simulation exercises the protocol.

Delivery levelsCopy Anchor Link

internal/delivery works out each stream's guarantee from the destination's class, the source's replay, the write mode, the primary key and the level the pipeline requires:

• An exact source into a stage-and-apply or transactional destination is exactly once: the protocol alone keeps every record to one commit.
• A superset source written with merge into a stage-and-apply destination is effectively once: a repeat lands on the same key, guarded by _openetl_seq, so the table ends up the same.
• A superset source written with append or overwrite into a destination that can skip rows by _openetl_id when it applies is exactly once, when the pipeline requires more than at least once.
• A destination that delivers and acknowledges, and anything else, is at least once.

internal/plan uses it to plan a pipeline version from its source's latest catalog, and the runner checks again before each sync, against the catalog it discovers.

Around the data pathCopy Anchor Link

• Incidents (internal/heal) are opened and resolved, and their playbooks run, in the same transaction as the report that shows them.
• Platform events (internal/events) are written to an outbox in the transaction that changes what they report, and delivered to webhooks, signed as the Standard Webhooks spec says (internal/webhooks).
• Outbound connections go through internal/netguard, which checks every address after DNS resolution, against server-side request forgery.
• Transforms run on the user's own clusters: internal/spark runs Spark applications, and internal/flink a continuous pipeline's Flink job, with libs/ holding what runs on those clusters.

How it's testedCopy Anchor Link

• Unit tests (make test) need no Docker.
• The simulation, test/sim, runs the real runner with fake sources and destinations. A seed selects fault timing, failures and zombie attempts. Checks cover data loss, exactly-once duplicates and final rows under effectively-once delivery. make sim SEEDS=10000 runs more seeds. go test ./test/sim -run 'TestSimulation/kafka' -seed=42 -v replays one seed with its trace.
• End-to-end tests, test/e2e, run the whole stack against the compose services and the official connectors, one test at a time.
• The chaos suite, in test/e2e, runs randomized pipelines while runners, engines and connectors die, runners lose the API and the object store fails, and checks every row landed once.
• Recorded workflow histories, in internal/orchestration/workflows/testdata, replay against the current workflows, so a change that would break a running workflow on Temporal fails a test.