Conduit

Data Integration for Production Data Stores. :dizzy:

Overview

Conduit is a data streaming tool written in Go. It aims to provide the best user experience for building and running real-time data pipelines. Conduit comes with common connectors, processors and observability data out of the box.

Conduit pipelines are built out of simple building blocks which run in their own goroutines and are connected using Go channels. This makes Conduit pipelines incredibly performant on multi-core machines. Conduit guarantees the order of received records won't change, it also takes care of consistency by propagating acknowledgements to the start of the pipeline only when a record is successfully processed on all destinations.

Conduit connectors are plugins that communicate with Conduit via a gRPC interface. This means that plugins can be written in any language as long as they conform to the required interface.

Conduit was created and open-sourced by Meroxa.

• Quick start
• Installation guide
• Preflight checks
• Configuring Conduit
• Storage
• Connectors
• Processors
• API
• Documentation
• Contributing

Quick start

The fastest way to see Conduit working — scaffold and run a demo pipeline with a single command:

conduit quickstart

Sample records flow to your console within seconds. State is in-memory and nothing is written to your working directory; press Ctrl-C to stop. For a full walkthrough, see .

Installation guide

Download binary and run

Download a pre-built binary from the latest release and simply run it!

./conduit run

Once you see that the service is running, the configured pipeline should start processing records automatically. You can also interact with the Conduit API directly, we recommend navigating to http://localhost:8080/openapi and exploring the HTTP API through Swagger UI.

Conduit can be configured through command line parameters. To view the full list of available options, run ./conduit run --help or see configuring Conduit.

Homebrew

Make sure you have homebrew installed on your machine, then run:

brew update
brew install conduit

Debian

Download the right .deb file for your machine architecture from the latest release, then run:

dpkg -i conduit0.15.0Linuxx8664.deb

RPM

Download the right .rpm file for your machine architecture from the latest release, then run:

rpm -i conduit0.15.0Linuxx8664.rpm

Build from source

Requirements:

• Go
• Make

git clone git@github.com:ConduitIO/conduit.git
cd conduit
make
./conduit run

Docker

Our Docker images are hosted on GitHub's Container Registry. To run the latest Conduit version, you should run the following command:

docker run -p 8080:8080 conduit.docker.scarf.sh/conduitio/conduit:latest

Preflight checks

Before running Conduit for the first time (or after changing configuration), run conduit doctor to check whether your environment is ready:

conduit doctor

It runs a set of offline, non-destructive checks — config resolution and validation, database reachability, API address availability, plugin directories, the built-in plugin registry, and whether a running engine is reachable — and reports pass/warn/fail for each, grouped by category. It does not start a Runtime; this is distinct from the running server's /readyz and /healthz endpoints, which answer "is the running engine serving?" instead of "would conduit run succeed here?".

Useful flags: --json for machine-readable output, --check <name> (repeatable) to run only specific checks, --deep to additionally verify standalone connector plugin binaries in an isolated subprocess, --require-server to fail instead of warn when no Conduit server is reachable, and -q/--quiet to only print warnings, failures, and the summary. See conduit doctor --help for the full list of checks and exit codes, and the design doc for the rationale.

Configuring Conduit

Conduit accepts CLI flags, environment variables and a configuration file to configure its behavior. Each CLI flag has a corresponding environment variable and a corresponding field in the configuration file. Conduit uses the value for each configuration option based on the following priorities:

• CLI flags (highest priority) - if a CLI flag is provided it will always be

respected, regardless of the environment variable or configuration file. To see a full list of available flags run conduit run --help.

• Environment variables (lower priority) - an environment variable is only

used if no CLI flag is provided for the same option. Environment variables have the prefix CONDUIT and contain underscores instead of dots and hyphens (e.g. the flag -db.postgres.connection-string corresponds to CONDUITDBPOSTGRESCONNECTIONSTRING).

• Configuration file (lowest priority) - By default, Conduit loads a

configuration file named conduit.yaml located in the same directory as the Conduit binary. You can customize the directory path to this file using the CLI flag --config.path. The configuration file is optional, as any value specified within it can be overridden by an environment variable or a CLI flag.

The file must be a YAML document, and keys can be hierarchically structured using .. For example:

db:
    type: postgres # corresponds to flag -db.type and env variable CONDUITDBTYPE
    postgres:
      connection-string: postgres://localhost:5432/conduitdb # -db.postgres.connection-string or CONDUITDBPOSTGRESCONNECTIONSTRING

To generate a configuration file with default values, use: conduit init --path <directory where conduit.yaml will be created>.

This parsing configuration is provided thanks to our own CLI library ecdysis, which builds on top of Cobra and uses Viper under the hood.

Storage

Conduit's own data (information about pipelines, connectors, etc.) can be stored in the following databases:

• BadgerDB (default)
• PostgreSQL
• SQLite

It's also possible to store all the data in memory, which is sometimes useful for development purposes.

The database type used can be configured with the db.type parameter (through any of the configuration options in Conduit). For example, the CLI flag to use a PostgreSQL database with Conduit is as follows: -db.type=postgres.

Changing database parameters (e.g. the PostgreSQL connection string) is done through parameters of the following form: db.<db type>.<parameter name>. For example, the CLI flag to use a PostgreSQL instance listening on localhost:5432 would be: -db.postgres.connection-string=postgres://localhost:5432/conduitdb.

The full example in our case would be:

./conduit run -db.type=postgres -db.postgres.connection-string="postgresql://localhost:5432/conduitdb"

Connectors

For the full list of available connectors, see the Connector List. If there's a connector that you're looking for that isn't available in Conduit, please file an issue .

Conduit loads standalone connectors at startup. The connector binaries need to be placed in the connectors directory relative to the Conduit binary so Conduit can find them. Alternatively, the path to the standalone connectors can be adjusted using the CLI flag -connectors.path.

Conduit ships with a number of built-in connectors:

• File connector provides

a source/destination to read/write a local file (useful for quickly trying out Conduit without additional setup).

• Kafka connector

provides a source/destination for Apache Kafka.

• Postgres connector

provides a source/destination for PostgreSQL.

• S3 connector provides a

source/destination for AWS S3.

• Generator connector

provides a source which generates random data (useful for testing).

• Log connector

provides a destination which logs all records (useful for testing). Additionally, we have prepared a Kafka Connect wrapper that allows you to run any Apache Kafka Connect connector as part of a Conduit pipeline.

If you are interested in writing a connector yourself, have a look at our Go Connector SDK. Since standalone connectors communicate with Conduit through gRPC they can be written in virtually any programming language, as long as the connector follows the Conduit Connector Protocol.

Processors

A processor is a component that operates on a single record that flows through a pipeline. It can either change the record (i.e. transform it) or filter it out based on some criteria.

Conduit provides a number of built-in processors, which can be used to manipulate fields, send requests to HTTP endpoints, and more, check built-in processors for the list of built-in processors and documentations.

Conduit also provides the ability to write your own standalone processor, or you can use the built-in processor custom.javascript to write custom processors in JavaScript.

More detailed information as well as examples can be found in the Processors documentation.

API

Conduit exposes a gRPC API and an HTTP API.

The gRPC API is by default running on port 8084. You can define a custom address using the CLI flag -api.grpc.address. To learn more about the gRPC API please have a look at the protobuf file.

The HTTP API is by default running on port 8080. You can define a custom address using the CLI flag -api.http.address. It is generated using gRPC gateway and is thus providing the same functionality as the gRPC API. To learn more about the HTTP API please have a look at the API documentation, OpenAPI definition or run Conduit and navigate to http://localhost:8080/openapi to open a Swagger UI which makes it easy to try it out.

Exit codes

Every conduit CLI invocation — one-shot commands (conduit pipelines list, ...) and the long-running conduit run — exits with one of a small set of deterministic codes, so scripts and agents can branch on failure kind without parsing error text:

| Code | Meaning | Examples | | ----- | ------------- | ----------------------------------------------------------------------------- | | 0 | Success | The command succeeded, or conduit run shut down gracefully on a single SIGINT/SIGTERM. | | 1 | Runtime error | An internal or unclassified error (a bug, an unexpected failure). | | 2 | Validation | The request or config was rejected: invalid argument, not found, already exists, failed precondition. | | 3 | Environment | A required external dependency is unreachable: the server, the database, a rate limit, or an already-bound listen address. |

conduit run treats a second SIGINT/SIGTERM (received after the first one has already started a graceful shutdown) as a forced kill and exits with the POSIX 128+signum convention instead (SIGINT → 130, SIGTERM → 143) — the same value a shell reports for a process killed by that signal, so a forced kill is distinguishable from an ordinary classified exit code.

A multi-result command like conduit doctor (many checks, one process) reduces its checks to a single exit code by taking the worst bucket across every failing check — a database that can't be reached (environment, 3) always outranks an invalid config field (validation, 2) in the final exit code, regardless of which check ran first. Warnings never affect the exit code.

See ADR: deterministic CLI exit codes for the full mapping and rationale, pkg/conduit/exitcode for the implementation, and pkg/conduit/check for the multi-result aggregation doctor (and future multi-check commands) share.

Validating pipeline configs

conduit pipelines validate <path> (alias: conduit pipeline validate) statically checks one pipeline config file, or every .yml/.yaml file in a directory (not recursed), without starting Conduit or contacting a running instance — it's fully offline, so it's safe to run in CI or before conduit run with no server anywhere nearby.

$ conduit pipelines validate pipelines/orders.yaml
✗ pipelines/orders.yaml
✗ config.field_required   /connectors/0/plugin
    connector "orders:postgres": "plugin" is mandatory
    → set connectors[0].plugin (e.g. "builtin:postgres")
Summary: 1 files · 0 passed · 1 failed · 1 problems Fix the ✗ items above, then re-run.

(the connector's ID is shown enriched — <pipelineID>:<connectorID> — since validation runs after the same default-enrichment step conduit run applies; --json's configPath still points at your original connectors[0].)

Every problem in every file is reported — a bad file never stops the rest from being checked. Exits 0 if every pipeline is valid, 2 otherwise (see Exit codes above). Add --json for the structured {command, ok, summary, result, error} envelope, or -q/--quiet to suppress passing (✓) lines and show only failures and the summary.

lint (advisory warnings, e.g. deprecated fields) and dry-run (shows the enriched pipeline graph and optionally resolves builtin plugin references) share the same engine and are planned as follow-ups — see the design doc.

Documentation

To learn more about how to use Conduit visit Conduit.io/docs.

If you are interested in internals of Conduit we have prepared some technical documentation:

• Pipeline Semantics explains the internals of how

a Conduit pipeline works.

• Pipeline Configuration Files

explains how you can define pipelines using YAML files.

• Processors contains examples and more information about

Conduit processors.

• Conduit Architecture

will give you a high-level overview of Conduit.

• Conduit Metrics

provides more information about how Conduit exposes metrics.

• Conduit Package structure

provides more information about the different packages in Conduit.

Contributing

For a complete guide to contributing to Conduit, see the contribution guide.

We welcome you to join the community and contribute to Conduit to make it better! When something does not work as intended please check if there is already an issue that describes your problem, otherwise please open an issue and let us know. When you are not sure how to do something please open a discussion or hit us up on Discord.

We also value contributions in the form of pull requests. When opening a PR please ensure:

• You have followed

the Code Guidelines.

• There is no other pull request

for the same update/change.

• You have written unit tests.
• You have made sure that the PR is of reasonable size and can be easily

reviewed.