V1 [SINK] Native data forwarding / sink system

[austin-barrington]

Context

Users need to push HyperbyteDB data to external systems — other databases, stream processors, or visualization tools — without running a separate bridge. A native SINK system with a DSL (CREATE SINK / DROP SINK / SHOW SINKS) lets operators configure data forwarding in the same way they configure retention policies or continuous queries.

Deliverables

DSL

Add SQL-like statements:

CREATE SINK "my_kafka" TO "kafka://broker:9092/topic-metrics"
  USERNAME "user" PASSWORD "pass"
  TYPE kafka
  FORMAT line_protocol
  INCLUDE MEASUREMENTS cpu,mem
  BATCH_SIZE 1000
  FLUSH_INTERVAL 5s

CREATE SINK "my_http" TO "http://collector:8086/write?db=mydb"
  USERNAME "u" PASSWORD "p"
  TYPE http_line
  INCLUDE ALL

CREATE SINK "my_mqtt" TO "mqtt://broker:1883/hyperbyte"
  USERNAME "mqtt_user" PASSWORD "mqtt_pass"
  TYPE mqtt
  FORMAT line_protocol

DROP SINK "my_kafka"

SHOW SINKS

Supported sink types:

Type	Destination format	Description
http_line	http(s)://host/path	Forward as line protocol POST
http_msgpack	http(s)://host/path	Forward as columnar MessagePack
mqtt	mqtt(s)://host:port/topic	Publish to MQTT topic
kafka	kafka://host:port/topic	Produce to Kafka topic
stdout	—	Write to stdout (debug/testing)

Metadata & durability

• Sink definitions stored in RocksDB metadata (meta_sink:<name>).
• Sink definitions survive restarts and are included in cluster sync / backup.
• Sink state (connection status, last error, points forwarded) tracked in metrics.

Background service

• SinkForwardService runs as a Tokio background task.
• On each tick, reads flushed data from chDB (similar to CQ service) and forwards to each configured sink.
• Batch size and flush interval are configurable per sink.
• Backoff and retry on failure; errors exposed via logs + metrics.
• Graceful shutdown: complete in-flight forwards, then stop.

Metrics

Metric	Type	Description
hyperbytedb_sink_forwards_total	counter	Points forwarded per sink
hyperbytedb_sink_errors_total	counter	Forward errors per sink
hyperbytedb_sink_duration_seconds	histogram	Forward latency per sink
hyperbytedb_sink_queue_depth	gauge	Pending points per sink

Auth

• Sink credentials (password, Kafka SASL, MQTT password) stored in metadata.
• When [auth] enabled = true, SHOW SINKS and CREATE SINK require admin credentials.
• Credentials redacted from SHOW SINKS output.

Config

Add a [sink] section with global defaults:

Key	Type	Default	Description
enabled	boolean	false	Enable sink forwarding service
default_batch_size	integer	1000	Default points per batch
default_flush_interval_secs	integer	10	Default flush interval
max_retries	integer	3	Max retries per forward attempt
backoff_base_secs	integer	1	Exponential backoff base

Documentation

• docs/user-guide/sinks.md — Sink types, configuration, examples (Telegraf → Kafka, MQTT bridge, HTTP forwarding)
• Update docs/user-guide/reference.md with new DDL statements
• Add row to docs/user-guide/index.md reading order

Test plan

• Unit tests for parser: CREATE SINK, DROP SINK, SHOW SINKS
• Integration test: create sink → write data → verify data reaches mock HTTP endpoint
• Integration test: sink credentials are redacted in SHOW SINKS
• Integration test: sink survives restart (metadata persistence)

Priority

P1 — High value for data pipeline use cases