etcd-cdc-rabbitmq

Etcd CDC sync service. Watches Etcd KV changes and reliably delivers them to a RabbitMQ Topic.

Fixes missed events on Watch disconnects, duplicate cross-cluster watches, and incomplete multi-key transactions. Stores offsets and elects a leader in the same source Etcd. Supports batched delivery, full config, and HA failover. One cluster runs multiple sync rules for per-business prefix isolation.

---

Features

• At-Least-Once — retry + backoff; no drop; consumer de-dup by revision+key.
• Per-prefix offset — independent batch/commit; resume on restart.
• Transaction integrity — same-revision multi-key delivered atomically.
• HA — 1 leader / N standby via Etcd Lease; auto-failover.
• Traffic shaping — batch / isolate / shard; Publisher Confirm.
• Per-business isolation — name-keyed routes; own prefix / routing key.
• Fail-fast — bad config exits at start; field-pointing errors.
• Graceful shutdown — flush buffers; advance to highest revision.
• Observable — Prometheus /metrics + stats log.

---

Architecture

Etcd Watch → rule match → group by revision → batch (batch/isolate/shard)
                                                          ↓
                                       RabbitMQ delivery (At-Least-Once)
                                                          ↓
                                       ACK → advance prefix SyncRevision

• Data source, offsets, and election share one Etcd cluster under hidden prefix /.etcd_cdc/v1/cluster/{cluster_id}/ (lock leader_lease + per-prefix sync_revision/{encoded_prefix}).
• Ordering: no global revision order across prefixes. Only hard rule: revision is monotonic within one prefix.
• Semantics: At-Least-Once. Cross-prefix same-revision txns split and emit independently; consumers de-dup.

Full design: doc/design/README.md.

---

Quick Start

Prereqs: Go 1.26 · reachable Etcd · reachable RabbitMQ (Topic exchange).

make build
make run                      # defaults to examples/config.yaml
./bin/etcd-cdc-rabbitmq -c examples/config.yaml --logging-level info

Flag	Description
-c, --config	Entry config; <path>.d/*.yaml merged sorted (last-win). Default examples/config.yaml.
--logging-level	Override logging.level (debug/info/warn/error).
--version	Print version and exit.

All business config is file-only; no runtime hot-reload.

---

Configuration

Layered load: embedded defaults (internal/config/defaults.yaml) → entry file → .d/ shards (sorted, scalars last-win, routes merged by name).

Full fields/defaults: examples/config.yaml, internal/config/defaults.yaml.

Section	Purpose
logging	Log level
server	metrics_addr (default :9090)
stats	Periodic stats log (on by default)
startup	Shared connect-retry for etcd / RabbitMQ
cluster	id, ha_enable, lease params
source_etcd	Etcd endpoints and auth
cluster_meta	Offset/election store: same_etcd or independent
batch	Ideal size, overload %, flush interval, byte cap
rabbitmq	MQ connection, exchange, confirm, retry
routes	Sync rules (name-keyed map)

Sync Rules (routes)

routes:
  app_config:
    routing_key: "etcd.config.app" # omittable under fanout
    include_prev_value: true       # carry previous value in body
    namespace: /biz/prod           # strip on delivery; "" if none
    prefixes: # watched prefixes (one or more)
      - "/config/app/"

Validation:

• Duplicate/nested prefixes across routes: allowed (one event → many destinations).
• Duplicate/nested prefixes within one route: illegal, fail-fast (duplicate delivery).

---

Operations

• Metrics: GET /metrics at server.metrics_addr (default :9090). Tracks batches, events, errors, per-prefix revision, offset, lag, leader status.
• Graceful shutdown: on SIGINT/SIGTERM, flush buffers, then exit. Offsets advance to highest delivered revision.
• Config changes: restart to apply. Each prefix resumes from its offset; no full replay.
• Stats log: stats.enabled=true prints a snapshot every stats.interval_ms (default 60s).

---

Design Tradeoffs

• Restart-to-apply, no hot-reload: avoids the "prefix change → rebuild watch state machine" complexity. Main flow stays load → run → exit.
• No global cross-prefix order: depend only on per-prefix monotonic revision. Avoids cold-prefix blocking from a global min-watermark.

---

Development

make test              # unit tests (no integration tag)
make test-integration  # e2e (needs docker; spawns etcd + rabbitmq)
make vet               # static checks
make fmt               # format
make docker-build      # build image
make dev-up            # start etcd + rabbitmq (docker compose)
make dev-down          # stop them

Integration tests: PUT/DELETE capture, multi-key txn grouping, oversized-txn sharding, offset resume, leader/follower failover.

main.go                Entry: config, signals, Server
internal/config        Structs, validation, defaults, layered load
internal/etcd         Client (election / watch / offset store)
internal/cdc          Match, group-by-revision, batch, shard
internal/pipeline     Watch → match → group → batch → deliver → offset
internal/rabbitmq     Delivery adapter
internal/ha           Lease-based election
internal/observability Logging, metrics
internal/server       Lifecycle, goroutine orchestration
internal/util         Shared utilities
doc/design/           Design docs (configuration / event processing / reliability / consumer & testing)

Docs: doc/design/README.md.