Build and register a connector

This guide is for developers who provide connector implementations to a tenant. The outcome is a digest-pinned source or destination that authorized users can configure in Data Hub, without a custom Portal form or a platform release.

Prerequisites

You need Docker with Buildx, jq, unzip, registry access, a tenant account with data.connector_definition.create, and source or destination test access. Coordinate registry pulls and endpoint network permissions with the platform operator before testing.

Customers build, publish and maintain connector images outside the Portal. The platform registers metadata; it does not build Dockerfiles, connect to your Git repository or install plugins on demand.

Choose the implementation

Use a MeltanoHub/Singer tap or target behind the supplied HTTP adapter, or implement the same execution contract in another language. A stock tap/target image is not sufficient unless it exposes this contract.

The worked examples below include a downloadable snapshot of the platform's PostgreSQL and S3/Ceph recipes. They are examples to build and validate, not preinstalled supported connectors. Check upstream dependency licenses, supported formats and destination-write semantics for the versions you select.

1. Implement the execution contract

Contract version 1 uses HTTPS on port 8443. The runtime supplies file paths through these environment variables:

The connector receives only its own connection's credentials. Read the mounted files at runtime; never bake credentials into the image or registration JSON. The workload runs without a Kubernetes service-account token, with a read-only root filesystem and writable temporary storage. It must support non-root execution.

Authenticate non-health operations with the supplied bearer token and serve them using the supplied TLS files. In the release sources, examples/data-connectors/singer/gateway.py and docker-images/meltano/image_runner.py define the full message shapes and bounds. In particular, a source must signal completion; a closed stream alone is not a successful transfer.

Retain useful failure codes and sanitized stage diagnostics. Do not return credentials, record contents or arbitrary credential-bearing stderr as diagnostics.

2. Build, publish and identify the image

Build the connector with the supplied gateway, push it to a registry your cluster can access, and inspect the published image digest. Register image@sha256:<digest>, with the full 64 hexadecimal digest characters. Tags alone are not accepted for new definitions. The worked examples provide complete build and definition-preparation commands.

Maintain and test the dependencies you publish. A digest gives stable artifact identity, not a security approval. Registry authentication is configured for cluster image pulls, separately from connection credentials.

3. Define the metadata and schemas

Start from the example JSON definition and adapt it to your image:

Schemas drive Portal fields and backend validation. Keep primary-key selection separate from the replication cursor. Do not advertise connection testing or incremental replication merely because the underlying ecosystem supports them; verify your particular image and adapter.

Version labels are opaque, not automatically ordered semantic versions. Existing connections remain bound to their selected version. Publishing a new version does not silently upgrade them.

4. Register and prove a transfer

1. Open Data Hub → Connectors → Register connector.
2. Enter the ID, version, name and digest-pinned image. Paste the remaining definition metadata into the JSON field and register it.
3. Create a connection and enter its configuration and credentials separately.
4. Run its supported tests and discovery. Create an ingestion with a compatible opposite-side connection.
5. Run against representative data and verify destination rows or objects, including a second run to establish replay, cursor and deduplication behavior.

Registering another version and deleting a connector family require the owner's authority as well as functional permissions. Existing connections block family deletion. Removing a registration does not delete its registry image or destination data.

Worked examples

Download the connector source bundle. It contains the shared Dockerfile and HTTPS adapter, three JSON registration templates and the optional Ceph notification helper. The executable sources and templates come from the platform's 1.4 release examples.

Unpack the download and work from its root:

unzip data-connectors-1.4.0.zip
cd data-connectors

The templates contain placeholder image tags. Replace them with your published digests using the commands below before registration. Connection secrets do not belong in these files.

Example A: PostgreSQL to PostgreSQL

This example builds both sides of a transfer, discovers a small source table and writes it into a dedicated destination schema.

Build both images

Set CONNECTOR_REGISTRY to your registry and repository namespace. Authenticate with that registry using your normal Docker login flow before pushing.

export CONNECTOR_REGISTRY=registry.example.com/my-team

docker build singer -t "$CONNECTOR_REGISTRY/postgres-source:v1" \
  --build-arg PLUGIN_PACKAGE=meltanolabs-tap-postgres==0.10.0 \
  --build-arg PLUGIN_EXECUTABLE=tap-postgres

docker build singer -t "$CONNECTOR_REGISTRY/postgres-target:v1" \
  --build-arg PLUGIN_PACKAGE=meltanolabs-target-postgres==0.8.0 \
  --build-arg PLUGIN_EXECUTABLE=target-postgres \
  --build-arg CONNECTOR_KIND=destination

docker push "$CONNECTOR_REGISTRY/postgres-source:v1"
docker push "$CONNECTOR_REGISTRY/postgres-target:v1"

SOURCE_DIGEST=$(docker buildx imagetools inspect \
  "$CONNECTOR_REGISTRY/postgres-source:v1" --format '{{.Manifest.Digest}}')
TARGET_DIGEST=$(docker buildx imagetools inspect \
  "$CONNECTOR_REGISTRY/postgres-target:v1" --format '{{.Manifest.Digest}}')

jq --arg image "$CONNECTOR_REGISTRY/postgres-source@$SOURCE_DIGEST" \
  '.image = $image' example-postgres-source.json > postgres-source.json
jq --arg image "$CONNECTOR_REGISTRY/postgres-target@$TARGET_DIGEST" \
  '.image = $image' example-postgres-target.json > postgres-target.json

These commands build and publish two separate images. CONNECTOR_KIND=destination is essential for the target. The adapter maps its database fields and target_schema to the target's native settings inside that image; the BSQAI API does not contain PostgreSQL-specific form logic.

Register the definitions

In Data Hub → Connectors → Register connector, use each generated JSON file's id, version, name and image for the corresponding fields. Paste the remaining metadata into the JSON field. For example:

jq 'del(.id, .version, .name, .image)' postgres-source.json
jq 'del(.id, .version, .name, .image)' postgres-target.json

The source advertises test, discovery, full-table and incremental capabilities. The destination advertises full-table writes and has no standalone connection-test capability. Its write access is verified during an actual ingestion.

Prepare a small source table

Have the source database owner create this disposable example table and grant the source connection's database account read access to it:

CREATE SCHEMA connector_demo;
CREATE TABLE connector_demo.products (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    updated_at TIMESTAMPTZ NOT NULL
);
INSERT INTO connector_demo.products (id, name, updated_at) VALUES
    (1, 'Notebook', '2026-01-01T00:00:00Z'),
    (2, 'Pencil',   '2026-01-02T00:00:00Z');

Use a dedicated test database or a schema name that is not already in use. On the destination, have its owner prepare connector_demo_target and grant the destination account the rights needed to create and write its tables. Source read credentials and destination write credentials should be different.

Configure the connections

Create one connection for each registered definition. Replace the example hosts, databases and users with your values:

Both database endpoints must be reachable under the operator's endpoint policy. The example uses encrypted PostgreSQL transport; choose stricter certificate/hostname verification and the corresponding trust configuration when required by your database policy.

Test the source connection, then discover streams. Select the discovered stream for connector_demo.products; use the actual discovered identifier rather than constructing one from the table name.

Run and inspect the result

Create an ingestion with those connections and the selected stream. Start with full-table replication, select id as its primary key, and set the destination ingestion option:

{
  "target_schema": "connector_demo_target"
}

Run once. In the destination database, find the table created for the selected stream:

SELECT table_name
FROM information_schema.tables
WHERE table_schema = 'connector_demo_target'
ORDER BY table_name;

Query that table and confirm two records with IDs 1 and 2 and the expected names. The destination connector derives table names from stream identifiers, so do not assume the source-qualified table name is preserved.

Run again and verify the same two logical records rather than duplicate IDs. The example target defaults to upsert behavior, which depends on selecting stable primary keys. For incremental testing, create a new ingestion with updated_at as the cursor and id as the primary key, then change a source row and advance its timestamp. Verify the corresponding destination update. The Portal does not edit an existing ingestion's stream configuration.

Example B: S3/Ceph files to PostgreSQL

Reuse the PostgreSQL destination image and connection from Example A. This source downloads files from a configured bucket/prefix and passes them to a packaged Singer file parser.

Build and register the file source

The supplied file-source recipe uses Python 3.11 and a pinned parser revision:

docker build singer -t "$CONNECTOR_REGISTRY/s3-source:v1" \
  --build-arg PYTHON_IMAGE=python:3.11-slim \
  --build-arg 'PLUGIN_PACKAGE=tap-universal-file @ https://github.com/MeltanoLabs/tap-universal-file/archive/beb126898c26b5f64d4158d372ea94da45b40ef3.tar.gz' \
  --build-arg PLUGIN_EXECUTABLE=tap-universal-file

docker push "$CONNECTOR_REGISTRY/s3-source:v1"
S3_DIGEST=$(docker buildx imagetools inspect \
  "$CONNECTOR_REGISTRY/s3-source:v1" --format '{{.Manifest.Digest}}')
jq --arg image "$CONNECTOR_REGISTRY/s3-source@$S3_DIGEST" \
  '.image = $image' example-s3-source.json > s3-source.json
jq 'del(.id, .version, .name, .image)' s3-source.json

Register the generated definition through the same Portal flow as Example A. Its adapter downloads into temporary local storage, then supplies that location to the file parser. The example limits the total staged input to 512 MiB per execution; this is not a platform-wide file-size limit.

Configure and run a file transfer

Save this as products.jsonl and upload it to incoming/products.jsonl in a dedicated connector-demo bucket:

{"id": 1, "name": "Notebook"}
{"id": 2, "name": "Pencil"}

Use this non-secret source configuration, replacing the endpoint with your S3-compatible service:

{
  "endpoint_url": "https://s3.example.com",
  "bucket": "connector-demo",
  "prefix": "incoming/",
  "region": "us-east-1",
  "file_type": "jsonl",
  "stream_name": "products"
}

1. Keep this example prefix limited to files of the selected format.
2. Create the source connection with these settings. The packaged parser uses jsonl for newline-delimited JSON and delimited for CSV/TSV; a .csv extension does not make csv a valid file_type. Other formats require their corresponding parser settings.
3. Put access_key_id and secret_access_key in credential fields. The account needs list/get access to the chosen bucket/prefix and version-read permission if using versioned object events.
4. Test and discover the source, choose its stream and stable record keys, then create an ingestion into the PostgreSQL destination with a dedicated target_schema.
5. Run manually and verify the destination records match the file. Validate your actual file format and schema before enabling automation.

A manual run lists the configured prefix. An event-triggered run instead reads the notified object/version and validates its bucket and prefix. For an unversioned event, the ETag must still match; the adapter refuses to substitute an object that has changed since the notification.

Optional: run when a file is uploaded

First enable the ingestion's upload trigger in Data Hub. Save the issued token privately. The bucket owner can then use the bundled ceph-notifications.py for Ceph RGW, with this protected configuration shape:

{
  "endpoint_url": "https://s3.example.com",
  "region": "us-east-1",
  "bucket": "connector-demo",
  "prefix": "incoming/",
  "access_key_id": "<bucket-owner-access-key>",
  "secret_access_key": "<bucket-owner-secret-key>",
  "callback_url": "https://api.example.com/v1/data-connectors/events/<trigger-id>",
  "trigger_id": "<trigger-id-from-data-hub>",
  "trigger_token": "<one-time-token-from-data-hub>"
}

Create notification.private.json in a protected location with mode 0600 before entering real values; keep it out of source control. This setup identity needs bucket-notification/topic management rights, not just the ingestion source's read permissions. With Python and boto3 installed:

python ceph-notifications.py --config notification.private.json

This command changes the selected bucket's notification configuration. It preserves unrelated rules and updates the rule for this trigger. Configure the trusted object-store CA through AWS_CA_BUNDLE when needed. Upload a new file, inspect the resulting Data Hub run and verify its destination records. Updating or rotating the trigger token requires rerunning producer setup with the new token.

The helper targets Ceph RGW. For AWS S3, use its supported event-delivery integration to call the authenticated event endpoint instead. Do not run this Ceph helper against an AWS bucket.

API integration

The BSQAI API exposes managed ingestion under /v1/data-connectors. Use the deployment's live API reference for complete request and response schemas.

Management routes require an authenticated user session and active tenant, with X-Tenant-Id selecting the tenant for users with multiple memberships. An ordinary tenant API key is not a delegated user identity for these routes. Scheduled execution uses a selected authorized user or operator-registered tenant workload principal, not a connector image's Kubernetes identity.

Test, discovery, run, reset and restore requests return an asynchronous run. Poll its status instead of treating HTTP 202 as completion. These operations accept an Idempotency-Key UUID: reuse it when retrying the same submission after an uncertain response, and use a new one for a new execution. This protects submission identity, not exactly-once destination writes.

Event-aware sources

Enable an ingestion trigger through the Portal or its trigger API. The resulting callback is POST /v1/data-connectors/events/{trigger_id}. The event producer authenticates with HTTP Basic using the trigger ID as username and its token as password. Protect these credentials, and update the producer when rotating the token.

Event interpretation belongs to the source image. The example S3 source validates the configured bucket/prefix and the notified object version or ETag; another source can implement a different payload contract. The backend does not turn an arbitrary event into source-specific ingestion logic.

The release example's README.md and ceph-notifications.py cover Ceph notifications. AWS S3 needs its supported event delivery integration or an authorized producer calling the same endpoint; it does not use Ceph's direct notification setup.

Operational handoff

Give the resource owner the registered definition/version, supported capabilities and tested source/destination behavior. Give the operator the image-pull requirements, endpoint access needs, native-service permissions and resource expectations. Use managed-ingestion operations for the runtime boundary and Data Hub for the user workflow.