sync-controller

Sync Controller

Sync Controller is a component that is responsible for running sync tasks in kubernetes cluster. Sync Controller supports source connectors that use Airbyte protocol. All methods are supported: spec, check, discover, read.

Sync Controller run all tasks in separate Pods. It runs pod and monitor its status, updating task status information in database tables. When task is finished, Sync Controller deletes Pod and frees resources. K8s Secret used to pass source credentials to Source connector in Pod.

Each Pod contains of two main containers: Source connector and Sync Sidecar.

Sync Sidecar captures data rows, logs, state and results of spec, discover and check command from Source connector:

• Data rows are sent to the target destination in Bulker instance,
• Logs are sent to preconfigured Bulkers destinations.
• spec, discover and check results goes to Postgres database tables.

Authorization

All request should have Authorization header with value Bearer ${token}.

Token should be one of the tokens defined in SYNCCTL_AUTH_TOKENS env variable, see configuration section.

If SYNCCTL_AUTH_TOKENS is not configured, then no authorization is required.

Endpoints

GET /spec

The endpoint is async. Run task to obtain Source connector specification. Resulting specification is stored in database table source_spec.

Query parameters:

• package* - source package image (for airbyte-based sources it's docker image)
• version* - source package version
• connectorType (optional) - airbyte by default

Result table: source_spec. It writes a last JSON schema with update time by [connectorType, package, version].

It returns either {"status": "ok"} or {"error": ...}

POST /check

The endpoint is async. Run task to check Source connector credentials. Check result is stored in database table source_check.

Query parameters:

• storageKey - key to store results of check in a database
• package - source package image (for airbyte-based sources it's docker image)
• version - source package version

Request body:

{
  "config": {
    // credentials json object
  }
}

Result table: source_check

POST /discover

Run task to obtain Source connector streams catalog. Catalog is stored in database table source_catalog.

Query parameters:

• storageKey - key to store results of discover in a database
• package - source package image (for airbyte-based sources it's docker image)
• version - source package version

Request body:

{
  "config": {
    // credentials json object
  }
}

Result table: source_catalog

POST /read

Run task to pull data from Source connector.

• Pulled data is sent to Bulker instance to syncId connection.
• Task logs are sent to Bulker instance to SYNCCTL_BULKER_LOGS_CONNECTION_ID connection.
• Task status is kept updated in source_task database table.
• Saved state of task if any is stored in source_state database table.

Query parameters:

• package - source package image (for airbyte-based sources it's docker image)
• version - source package version
• taskId - id of this task. Should be unique for each task
• syncId - id of sync entity (bulker destination id) where pulled events should be sent

Request body:

{
  "config": {
    // credentials json object
  },
  "catalog": {
    // catalog json object with selected streams and their configurations
  },
  "state": {
    // json object with state from previous task run
  }
}

Configuration

Database schema