streamrun

Documentation

Index

• func NormalizeGTIDSet(s string) string
• func One(ctx context.Context, cfg Config) error
• func StartMetricsServer(addr string) (shutdown func(), err error)
• type Config
• type Deps
• type Hooks

Functions

func NormalizeGTIDSet

func NormalizeGTIDSet(s string) string

NormalizeGTIDSet zero-pads each UUID in a GTID set to the standard 8-4-4-4-12 format. Some MySQL-compatible services (e.g. Amazon RDS) return GTIDs with leading zeros stripped from UUID segments, producing UUIDs shorter than 36 characters (e.g. "5512139-1432-11f1-8d8d-0693b428a89b" instead of "05512139-1432-11f1-8d8d-0693b428a89b"). The go-mysql library requires standard-length UUIDs, so this function normalizes before parsing.

func One

func One(ctx context.Context, cfg Config) error

func StartMetricsServer

func StartMetricsServer(addr string) (shutdown func(), err error)

StartMetricsServer binds the Prometheus /metrics endpoint and returns a shutdown func. The bind is SYNCHRONOUS so a bad --metrics-addr fails the command fast — the operator explicitly asked for metrics; silently running without them (scrapes getting connection-refused) is worse than refusing to start. Shared by `bintrail stream` (one endpoint per stream process) and the `bintrail-console watch` daemon (one endpoint for all supervised streams; the default registry is process-global).

Types

type Config

type Config struct {
	IndexDSN  string
	SourceDSN string
	// Flavor is the source database flavor: "mysql" (default) or "mariadb".
	// Empty is normalized to "mysql" in One(). It selects the GTID parser and
	// the BinlogSyncer flavor, and is persisted to stream_state for resume.
	Flavor      string
	ServerID    uint32
	StartFile   string
	StartPos    uint32
	StartGTID   string
	BatchSize   int
	Schemas     string
	Tables      string
	Checkpoint  int // seconds
	MetricsAddr string
	// MetricsSource is the value of the Prometheus "source" label for this
	// stream — the supervisor sets it to the registry entry ID so N
	// concurrent streams stay distinguishable on one /metrics endpoint.
	// Empty = fall back to the resolved bintrail_id (or "default").
	MetricsSource string
	// MetricsScrapeInterval is how often (seconds) the bintrail_index_* gauges
	// are refreshed from a status snapshot. 0 = the 60s default.
	MetricsScrapeInterval int
	// IndexMetrics forces the bintrail_index_* scraper on even when this stream
	// sets neither MetricsAddr nor MetricsSource — the `bintrail-console watch`
	// daemon's OWN primary stream needs this, since it serves /metrics centrally
	// (MetricsAddr empty) and isn't supervisor-launched (MetricsSource empty).
	IndexMetrics bool
	SSLMode      string
	SSLCA        string
	SSLCert      string
	SSLKey       string
	Format       string
	Reset        bool
	NoGapFill    bool
	GapTimeout   int // seconds
	// Hooks, when non-nil, lets a supervisor observe this stream's liveness
	// without polling. Plain `bintrail stream` leaves it nil.
	Hooks *Hooks
	// Deps are the host-supplied helper functions One needs (see Deps). All
	// fields are required.
	Deps Deps
}

Config carries everything ONE replication stream needs — the by-value equivalent of the strm* package globals. It exists so the control plane can run N streams concurrently in one process (each with its own config), while `bintrail stream` keeps wiring its flags through a single instance unchanged. Plain data plus optional observer hooks and required Deps: the zero value is invalid; build it from the cobra layer's streamConfigFromFlags or populate it explicitly.

type Deps

type Deps struct {
	ValidateBinlogFormat   func(db *sql.DB) error
	ValidateBinlogRowImage func(db *sql.DB) error
	ValidateNoFKCascades   func(db *sql.DB, schemas []string) error
	ParseSchemaList        func(s string) []string
	ResolveServerIdentity  func(ctx context.Context, sourceDB, indexDB *sql.DB, sourceDSN string) (string, error)
	EnsureResolver         func(indexDB, sourceDB *sql.DB, schemas []string) (*metadata.Resolver, error)
	BuildIndexFilters      func(schemas, tables string) parser.Filters
	InsertSchemaChange     func(db *sql.DB, ev parser.Event, snapshotID *int) error
	ParseSourceDSN         func(dsn string) (host string, port uint16, user, password string, err error)
	OutputJSON             func(v any) error
}

Deps are the host-supplied functions One needs that are NOT part of the streaming engine itself — source preflight/validation, server-identity resolution, schema snapshot/resolver/filters, DDL schema-change recording, source-DSN parsing, and JSON summary output. They live in the binary that builds the Config so streamrun stays a pure engine and never imports the cmd layer (or anything it would drag in). Every field must be set.

type Hooks

type Hooks struct {
	// OnCheckpoint fires after every successful checkpoint save — the
	// "attached and healthy" signal (the ticker checkpoints even with zero
	// events, so an idle source still reports progress).
	OnCheckpoint func()
	// OnIndexed fires after every successful batch flush with rows written.
	OnIndexed func(n int64)
	// OnGapAutoAdvance fires when an unfillable binlog gap forced the stream
	// to advance past purged events — data in the gap is permanently lost.
	OnGapAutoAdvance func(detail string)
}

Hooks are liveness callbacks a supervisor attaches to one stream. All fields are optional. They are invoked synchronously from the stream loop — implementations must be fast and non-blocking.