Skip to content

Event Consumers

ConfigHub records actions in an append-only event log: a Unit was applied, a Release was published, and so on. An event consumer subscribes to those events and reacts to them: trigger a deployment, post a notification, kick off a job, or synchronize an external system.

A consumer is a pure reader. Unlike a Bridge, it holds no lease and dequeues no operations, so it never contends with a Target's work. The event only tells the consumer that something happened; the reaction is entirely the consumer's own. Consumers add immediacy and integrations to ConfigHub without touching the apply path.

One example is argobot (MIT-licensed, open source): it consumes apply and release events and force-syncs the corresponding Argo CD Application so a deploy takes effect immediately instead of waiting for Argo's next reconciliation. This page uses the same SDK argobot is built on.

The Quick Start gets a consumer running against a ConfigHub instance. The sections after it cover subscriptions, handling events, delivery semantics, and a deep dive into the wire protocol.

Quick Start

A consumer authenticates as a ConfigHub Worker. It's the same worker identity used elsewhere; a consumer uses it to read the event log rather than to run operations.

1. Create a worker identity

cub worker create my-consumer --space my-space
cub worker get my-consumer --space my-space --include-secret

Note the worker's ID and secret; the consumer authenticates with them. See the worker guide for details.

2. Write the consumer

The SDK ships the event-consumer machinery in github.com/confighub/sdk/core/worker. A working consumer is a subscription, a handler, and a run loop:

package main

import (
    "context"
    "encoding/json"
    "log"
    "os"
    "os/signal"
    "syscall"

    "github.com/confighub/sdk/core/worker"
    "github.com/confighub/sdk/core/worker/api"
)

func main() {
    consumer := worker.NewEventConsumer(
        os.Getenv("CONFIGHUB_URL"), // e.g. https://hub.confighub.com
        os.Getenv("CONFIGHUB_WORKER_ID"),
        os.Getenv("CONFIGHUB_WORKER_SECRET"),
        api.EventSubscription{
            // Name keys the server-held delivery cursor — keep it stable across restarts.
            Name:       "my-consumer",
            EventTypes: []string{"apply.completed", "release.published"},
            // SpaceID and TargetID left empty: receive events from every Space and Target.
        },
        handle,
    )

    // Run blocks until the context is canceled.
    ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
    defer stop()

    if err := consumer.Run(ctx); err != nil && ctx.Err() == nil {
        log.Fatal(err)
    }
}

// handle is invoked once per delivered event.
func handle(ctx context.Context, e api.EventLogEntry) {
    log.Printf("%s in space=%s target=%s (cursor %d)", e.EventType, e.SpaceID, e.TargetID, e.CursorID)

    if e.EventType == "release.published" {
        var payload struct {
            SpaceSlug  string
            ReleaseNum int64
        }
        if err := json.Unmarshal(e.Payload, &payload); err != nil {
            log.Printf("unexpected payload: %v", err)
            return
        }
        // React here: e.g. call an external system for release ReleaseNum of SpaceSlug.
        log.Printf("release %d published for %q", payload.ReleaseNum, payload.SpaceSlug)
    }
}

3. Run it

CONFIGHUB_URL=https://hub.confighub.com \
CONFIGHUB_WORKER_ID=... CONFIGHUB_WORKER_SECRET=... \
go run .

Apply a Unit or publish a Release for a subscribed Space, and the consumer logs the event. That's the whole loop — everything else is deciding what to react to and how.

The subscription

An api.EventSubscription declares what the consumer receives:

Field Meaning
Name A stable, per-worker subscription name. The server keys the delivery cursor on it, so a restart resumes where the consumer left off. It must be identical across reconnects.
EventTypes The event types to receive, e.g. apply.completed, release.published. Empty means every type.
SpaceID Optional Space filter (UUID). Empty means any Space.
TargetID Optional Target filter (UUID). Empty means any Target.

The event vocabulary is open-ended and owned by the server; new event types may appear over time. A consumer names the ones it cares about rather than relying on a fixed enum.

Scope with TargetID to keep a consumer focused

If your consumer only cares about one deployment destination, set TargetID to that Target's UUID. It will then receive only the events recorded against that Target, instead of filtering the whole org's traffic in your handler. argobot, for instance, runs one instance per cluster, each scoped to that cluster's Target.

Handling events

The handler is a func(context.Context, api.EventLogEntry), invoked once per delivered event. An api.EventLogEntry carries:

Field Meaning
EventType The event's type, e.g. apply.completed.
SpaceID / TargetID The event's Space and Target scope, if any (UUID strings).
SubjectEntityType / SubjectEntityID The entity the event is about, e.g. Release / its ID.
Payload Event-type-specific data as raw JSON (json.RawMessage), opaque to the transport.
CursorID This event's monotonic position in the log.
CreatedAt When the event was emitted.

The Payload shape depends on the event type. For release.published it includes SpaceSlug, ReleaseNum, Digest, and BundleBaseName; see the Release reference. Unmarshal only the fields you need, and tolerate unknown ones; the server may add fields.

Read the field you mean, not the one that looks close

A Release's BundleBaseName is the bundle's filename base and defaults to the Space's ID, not its slug. To act on the Space an event is about (for example, to name a resource after it), use SpaceSlug, which follows the Space's slug. Choosing the wrong field is an easy and silent mistake.

Delivery and resumption

The delivery cursor, a bookmark into the log, is held by the server, keyed by the worker and the subscription Name. The consumer is stateless: each poll sends only the filter, never a position or an acknowledgment, and the server advances the bookmark as it hands out events. Two consequences follow:

  • Restart and failover resume cleanly. A consumer that stops and reconnects with the same Name continues from the stored bookmark, so events recorded while it was down are still delivered when it returns. On the first connect for a Name (no stored bookmark), delivery starts at the log tail (only events recorded from then on) unless you set FromCursor to begin earlier.
  • Delivery is at-most-once. Because the server advances the bookmark as it delivers and the client never acknowledges, an event handed to a consumer that then crashes before finishing is not redelivered. A consumer can miss an event, but it won't receive the same one twice.

So design a consumer such that correctness belongs to the source, not to the consumer:

  • Treat the reaction as an immediacy layer. A missed event or a failed reaction should cost promptness, not correctness; the underlying system still converges. argobot relies on this: Argo CD reconciles from its source regardless, so a missed force-sync delays a deploy but never corrupts one.
  • Keep reactions idempotent anyway. You might replay from an earlier FromCursor, or run redundant consumers (below), and a repeated reaction should be harmless.
  • Keep handler work quick, or hand it to a queue/goroutine you own, so one slow reaction doesn't stall delivery of the next event.

Running more than one consumer

The Name is the cursor. Two consumers that connect with the same Name (and worker) do not each get a copy of the stream; they contend for a single server-side bookmark. Each event advances that one bookmark, so it goes to whichever poll picked it up and the other consumer never sees it; with overlapping polls the split isn't guaranteed clean. Same-Name concurrency is therefore neither a fan-out nor a supported HA model.

Choose based on what you want:

  • High availability — run a single active consumer, and rely on quick restart or an active/standby pair (only one polling at a time). The server-held bookmark makes the handoff clean, and at-most-once delivery plus source reconciliation means a brief gap costs only immediacy.
  • Fan-out — every replica sees every event — give each consumer its own Name. Each then has an independent cursor and receives the full stream; make the reactions idempotent to absorb the N× firing.

Deep dive

NewEventConsumer and Run wrap a small HTTP protocol you can implement on any stack:

  1. Authenticate. The consumer exchanges its worker ID and secret at POST /auth/worker for a short-lived JWT, and refreshes the JWT automatically on a 401.
  2. Long-poll. It polls POST /api/space/{spaceID}/bridge_worker/{workerID}/event_cursor/{name}/poll. The subscription (event types and any Space/Target filter) travels in the request body on each poll; the server persists only the cursor, under {name}. Each response delivers a batch of EventLogEntry records after the consumer's current cursor, or returns empty when the poll times out, and the consumer polls again.

Because the consumer owns no state and the server owns the cursor, a restarted or rescheduled consumer resumes where it left off. That same statelessness is why two consumers must not share a Name: they would contend for the one cursor rather than each see the stream. See Running more than one consumer.

Deploying a consumer

A consumer is an ordinary long-running process. Package it as a container and run it wherever it has network access to ConfigHub. Supply the worker ID and secret through your platform's secret mechanism, and keep the subscription Name stable so restarts resume cleanly. argobot's deployment manifests are a minimal, working reference: a Deployment plus its configuration and secrets.

See also