Data synchronization sources

Reading data out of an existing store — a database, a search engine, a message queue — and forwarding it into Sondar. The reader names and configuration below are the shipped agent’s consumer registry, verified against the source.

How the consumers work

A consumer is a reader that polls an external store on an interval and hands the rows it finds to the rest of the pipeline (transformers, then senders). Every consumer shares three generic settings injected by the runner:

key	default	notes
`interval`	`30s`	how often the store is polled
`cron_spec`	—	optional cron expression instead of a plain interval
`cron_type`	`skip`	`none` (don’t block), `skip` (skip if the previous run is still going), `delay` (queue until the previous run finishes)

Two modes, and the distinction matters:

_append — incremental. Each run reads only what is newer than the last checkpoint, keyed on an incrementing column you name. Polling cost is constant; this is the mode to leave running.
_snapshot — full. Each run reads the whole table / index / collection. Used for one-off backfills, for data with updates and deletes, and for tables with no incrementing column.

A checkpoint (offset.meta or a per-source meta file) is written next to the agent’s meta directory and consulted on restart, so an interrupted incremental run resumes where it stopped rather than re-reading.

Clearing the target before a snapshot. A full read that must replace the target index instead of appending uses the shared sondar_host + sondar_truncate_repo keys — the reader empties the named repos before each cycle. The repos named here must be the same repos the sender writes, or data is deleted by mistake.

Use the new names. Older examples use the old keta_* keys and pre-rename consumer spellings. The shipped keys are sondar_host / sondar_truncate_repo and the readers are sondardb_consumer_*. The old spellings are gone; do not copy them into a config.

SQL databases

All seven dialects share one append/snapshot schema — only the reader name and the DSN shape differ. The sql_consumer family:

engine	reader names	DSN
MySQL	`mysql_consumer_append` / `mysql_consumer_snapshot`	`<user>:<password>@tcp(host:port)/?parseTime=true`
PostgreSQL	`postgres_consumer_append` / `postgres_consumer_snapshot`	`postgres://<user>:<password>@host:port/db?sslmode=disable`
Oracle	`oracle_consumer_append` / `oracle_consumer_snapshot`	`user/password@host:port/service`
SQL Server	`mssql_consumer_append` / `mssql_consumer_snapshot`	`sqlserver://<user>:<password>@host:port?database=...`
Dameng	`dm_consumer_append` / `dm_consumer_snapshot`	Dameng DSN
Trino	`trino_consumer_append` / `trino_consumer_snapshot`	`http://user@host:port` (Trino’s client DSN)
Druid	`druid_consumer_append` / `druid_consumer_snapshot`	`http://host:port/druid/v2/sql` (Avatica JDBC)

The append schema

readers:
  - mysql_consumer_append:
      interval: 30s
      dsn: "<user>:<password>@tcp(localhost:3306)/?parseTime=true"
      database: test
      tables_include:
        table_a: time      # table -> incrementing key
        table_b: id
      tables_exclude:
        - table_c
      data_base_options:
        max_open_conns: 10

Key semantics (verified in sql_consumer/append/types.go):

dsn — required, single source; dsns — a list of sources, each collected in the same task.
database / schema — collect every table in the database; the dialect decides which word it calls it (MySQL database, Oracle/SQL Server schema).
tables_include — a map of table: incrementing_key. A table listed without a key is read in full each cycle even in append mode.
tables_exclude — tables skipped when collecting a whole database.
queries — a map of SQL statement: incrementing_key, taking priority over database when both are set. Incrementing key only works here if the query returns it as a column, and the query must not return two columns with the same name.
meta — extra fields stamped onto every row, chosen from dsn, database, schema, table, query (plus db_host/db_port/db_instance where the dialect can derive them).
data_base_options — max_open_conns (default 10), max_idle_conns, conn_max_idle_time, conn_max_life_time.

Incrementing keys must be strictly increasing — a time column with backfilled rows, or an auto-increment id, both work; a volatile last_updated does not.

The snapshot schema

The same keys, with three differences:

tables_include / queries are lists, not maps — no incrementing key in full mode.
is_full_query (default false) — must be true for statements that are not a plain SELECT ... FROM ... (a SHOW ... or a stored-procedure call), so the reader does not try to derive a table name.
The sondar_host / sondar_truncate_repo clearing keys, as above.

Dialect notes

SQL Server builds offsets with OFFSET ... ROWS FETCH NEXT ... ROWS ONLY and paginates by CURRENT_TIMESTAMP ordering — the offset column must be orderable the same way (types.go: SelectWithOffsetTemplateForTable).
Oracle offset queries add a synthetic sondarRowNum column so the reader can page without a stable key; the row-number column is stripped before data is handed on, so it never reaches your index.
MySQL append also supports change-data-capture — see MySQL binlog below.

MongoDB

Reader names mongodb_consumer_append / mongodb_consumer_snapshot (in the sql_consumer tree but a separate dialect). Requires MongoDB 2.6 or higher.

readers:
  - mongodb_consumer_append:
      interval: 600s
      host: "mongodb://<user>:<password>@<host>:<port>/?retryWrites=true"
      database: app
      collection: orders
      offset_key: _id
      query:
        name: example
        age:
          $gte: 12

host — a standard MongoDB connection string (mongodb:// or mongodb+srv://). Options carried in the URL override client_option.
database, collection — required.
query — a nested document, written YAML-style, matching db.collection.find(...), e.g. a field filter like {"name":"example","age":{"$gte":12}}.
offset_key — defaults to _id. When set, the reader replaces sort with {"<offset_key>": 1} and the query’s offset condition with {"<offset_key>": {"$gt": <value>}}.
find_option — hint, projection, sort, min, max, let (MongoDB FindOptions).
connect_timeout / collect_timeout — both default 30s.
client_option — raw Go driver client options (the supported URI params).

Full mode reads the whole collection each cycle.

Elasticsearch

Reader names es_go_es_consumer_append / es_go_es_consumer_snapshot (go-elasticsearch client, the default) and es_olivere_consumer_append / es_olivere_consumer_snapshot (Olivere client). Both speak to ES 6, 7 and 8 (version, default 7).

readers:
  - es_go_es_consumer_append:
      interval: 30s
      host: http://localhost:9200
      username: elastic
      password: changeme
      index:
        - name: my_index
          key: clock          # incrementing field for append mode
          query: '{"range": {"count": {"gt": 5}}}'
        - name: my_second_index

host single, or addresses a list of cluster nodes.
username/password, service_token, or cert_file / cert_fingerprint for TLS — the config takes whichever auth fits.
index — a list of index specs, each with name (wildcards allowed), key (the incrementing field used as the offset in append mode), and query (an Elasticsearch DSL filter applied to the read, e.g. {"range":{"count":{"gt":5}}}).
conn_timeout / read_timeout (default 30s), scroll (default 5m), es_batch_size (default 1000).

Append mode tracks the max key per index; snapshot reads the whole index each cycle and, with sondar_host/sondar_truncate_repo, replaces the target.

Why the two clients. Nothing in the product says which to prefer; the choice is not obviously reversible once a checkpoint exists. This was already flagged as a documentation gap in the sources page — pick one per task and stick with it.

HBase

Reader names hbase_consumer_append / hbase_consumer_snapshot, driven by gohbase (uses ZooKeeper, not a direct HBase address).

readers:
  - hbase_consumer_append:
      zookeeper_host: "zk1:2181,zk2:2181"
      table: metrics
      columns: ["cf:col1", "cf:col2"]

zookeeper_host — required, the ZooKeeper ensemble.
table — required, case-sensitive.
user (default root), compression_codec (snappy), and the usual timeouts (flush_interval 20ms, region_lookup_timeout / region_read_timeout / zookeeper_timeout 30s, collect_timeout 10m).
columns — which columns to read (empty = all).
rows — zero, or a start/end row pair.
HBase’s own row timestamp is carried through as the hbase_time field.

Redis

Reader names redis_goredis_consumer (Go-Redis, the recommended default) and redis_redigo_consumer (Redigo). Same schema.

readers:
  - redis_goredis_consumer:
      interval: 30s
      host: localhost:6379
      password: secret
      db: 0
      key:
        string: ["key1", "key2"]
        list: ["list1"]
        hash: ["hash1"]
        set: ["set1"]
        zset: ["zset1"]
        channel: ["chan1"]
        pattern_channel: ["p_chan1"]
        command: ["GET key1"]

host — required, host:port.
password, db, conn_timeout / read_timeout (defaults 5s).
key — a map of type → key list, one entry per Redis data type (string, list, hash, set, zset, channel, pattern_channel, command). Commands let you read something the built-in types do not cover — the value returned is the raw command result.
version lets the two clients negotiate protocol.

The two-client choice exists here too (same reason as Elasticsearch and Kafka) — and nothing documents which to pick.

Kafka

Reader names kafka-segmentio (segmentio/kafka-go, recommended) and kafka-sarama (Shopify sarama). Both consume via consumer groups, so parallelism is horizontal: run several tasks with the same group_id and Kafka splits the partitions between them.

readers:
  - kafka-segmentio:
      topics: ["test"]
      brokers: ["127.0.0.1:9092"]
      group_id: my-group
      sasl_mechanism: PLAIN      # PLAIN | SCRAM-SHA-256 | SCRAM-SHA-512
      sasl_username: kafka
      sasl_password: secret
      startoffset: -2            # -2 = earliest (default), -1 = latest

topics, brokers, group_id — required.
sasl_*, tls_* — auth. TLS keys: tls_ca, tls_cert, tls_key, tls_key_pwd, insecure_skip_verify, tls_server_name.
startoffset — -2 (earliest, default) or -1 (latest). Applies on first consumption; a group with a checkpoint resumes from it.
The long tail of kafkago.ReaderConfig options (segmentio): minbytes (1), maxbytes (10 MB), maxwait (60s), heartbeatinterval (3s), commitinterval (1s), sessiontimeout (30s), rebalancetimeout (30s), joingroupbackoff (5s), retentiontime (24h), readbackoffmin (100ms), readbackoffmax (1s), maxattempts (3), queuecapacity (100).
add_header — optionally keep the Kafka headers on the record.
sarama exposes its own KafkaConfig inline instead of the segmentio option names — the top-level keys (topics/brokers/group_id/sasl/tls) are the same.

Sondar itself (index-to-index)

Reader names sondardb_consumer_append / sondardb_consumer_snapshot — read one index to write another. This is the rollup mechanism: build a coarse-grained index from a raw one, or migrate between indexes.

readers:
  - sondardb_consumer_append:
      hosts: ["127.0.0.1:9500"]     # gRPC, not the HTTP port
      token: ""
      query: "repo=app_logs | ..."   # any SonQL
      offset_key: _time

hosts — a list (or host, deprecated, for a single one). The gRPC service address, not the HTTP access port.
token — required when auth is on.
query — any SonQL query over the source repo. Start/end time bounds (start_time/end_time) and result size (default 500 for search-type queries) are available as inline fields.
Append mode: offset_key (default _time), start_value (default 0), and offset_template (default {OFFSET}). If the offset column is _time, the reader advances start_time instead of substituting the template; for any other column it rewrites the placeholder in the query.
Snapshot mode: run the whole query every cycle.
Connection tuning: connect_timeout (30s), drop_null, metadata (header pairs), plus the gRPC/TLS inline config (keepalive_params, TLS CA/cert files, etc.). The config struct still lives in a source file with the pre-rename file name; the keys are the sondardb_* reader’s.

MySQL binlog — change data capture

Reader name mysql_binlog_consumer — Linux builds only (all_linux.go), and it is the one consumer that does not poll: it streams MySQL’s binlog for row-level changes (via go-mysql-org/go-mysql).

readers:
  - mysql_binlog_consumer:
      addr: 127.0.0.1:3306
      user: root
      password: secret
      sync_from_now: false
      include_table_regex: [".*\\.canal"]
      exclude_table_regex: ["mysql\\..*"]
      dump:
        mysqldump: mysqldump
        tables: [table1]
        table_db: db1

addr/user/password — required connection.
dump — optional backfill stage: run mysqldump for the named tables/databases before switching to binlog streaming. During the dump the tables are locked against writes, so keep it off unless history is needed. discard_err, skip_master_data, max_allowed_packet_mb, protocol, where, ignore_tables, extra_options tune the dump.
sync_from_now — start from the newest binlog position (default) or from the oldest.
include_table_regex / exclude_table_regex — table filters (matched against db.table). Both empty = all tables.
The rest are canal’s: charset (utf8), flavor (mysql/mariadb), heartbeat_period, read_timeout, parse_time, use_decimal, semi_sync_enabled, disable_retry_sync, max_reconnect_attempts, meta (database/table/addr).

This is the mode for tables with updates and deletes where append mode’s “newer rows only” is not enough — every change is an event.

Zabbix metrics

Reader name zabbix_mysql_binlog_consumer — Linux only, and it exists to pull Zabbix monitoring data (only Zabbix backed by MySQL). It is built on the MySQL binlog consumer: item metadata (items, hosts, host_groups, interfaces, macros) is polled on a schedule, while item data (history, history_uint, history_str) streams through dump + binlog.

readers:
  - zabbix_mysql_binlog_consumer:
      zabbix_dsn: "root:secret@tcp(127.0.0.1:3306)/zabbix"
      template_dir: ".zabbix_templates"
      query_cond:
        items:
          - key: "agent.ping"
        hosts:
          - host: "%Zabbix server%"
      host_parse_regex:
        node:
          - '(?P<ip>\d{1,3}(\.\d{1,3}){3})(_(?P<port>\d{1,5}))?'

zabbix_dsn — required.
template_dir — the metric templates that decide which items are collected and how they are named.
query_cond — filters over Zabbix’s own tables (items, hosts, interface, hstgrp), supporting % wildcards; default is all items.
host_parse_regex — regexes per template that pull IP/port out of Zabbix host names.
query_period (24h default), query_batch, server_id, bus_group_dsn/bus_group_query, host_ip_list_path, sync_from_now.