streamrun

Documentation

Index

• Constants
• Variables
• func FlavorOnce(fn func(flavor string)) func(flavor string)
• func NormalizeGTIDSet(s string) string
• func One(ctx context.Context, cfg Config) error
• func StartMetricsServer(addr string) (shutdown func(), err error)
• func StartMetricsServerWith(addr string, more map[string]http.Handler) (shutdown func(), err error)
• type Config
• type Deps
• type EarlierCleanupError
  • func (e *EarlierCleanupError) Error() string
  • func (e *EarlierCleanupError) Unwrap() error
• type GapRefusedError
  • func (e *GapRefusedError) Error() string
  • func (e *GapRefusedError) TelemetryClass() string
• type Hooks
• type SourceRenumberedError
  • func (e *SourceRenumberedError) Error() string
  • func (e *SourceRenumberedError) TelemetryClass() string

Constants

const DefaultResumeCleanupWait = time.Hour

DefaultResumeCleanupWait is how long a start waits for an earlier cleanup before giving up. The longest cleanup measured is about 30 minutes (#1690, a 48 GB index with no floor), so an hour covers one that began just before the restart with the same again to spare.

const PhaseResumeCleanup = "resume_cleanup"

PhaseResumeCleanup is the OnPhase value for the resume-time dedup: the delete (and, in GTID mode, the open-transaction straggler pass) that runs before StartSync so a replayed window cannot double-index. On a large index it is minutes of work with no events flowing, which is exactly the window #1690 found indistinguishable from a hung daemon.

const PhaseResumeCleanupWaiting = "resume_cleanup_waiting"

PhaseResumeCleanupWaiting is the OnPhase value while a start waits for a cleanup that an earlier run left executing on the index server. No cleanup of this run has started yet.

Variables

var ErrEarlierCleanupRunning = errors.New("an earlier resume cleanup is still running on the index")

ErrEarlierCleanupRunning tags the failure of a start that waited the whole ceiling for an earlier cleanup. Callers match it with errors.Is; the text is for people and may change.

var ErrStopWithoutFlush = errors.New("capture stopped without writing its pending batch or position: this process lost its capture lock")

streamLoop consumes parser events, flushes batches to MySQL, and writes checkpoints to stream_state at the given interval.

A reader of the index never sees part of a source transaction (#2391). Rows of COMPLETE transactions are batched together as before, up to the batch size, and each batch commits in one statement (or one index transaction when it needs more than one). Rows of the transaction still being read are never committed with them: once they fill a batch they go to an index transaction of their own, which commits when the source transaction's end (EventCommit or EventDDL) arrives, or in parts past maxTxnRows / maxTxnOpen. A stop in the middle of a transaction discards its rows, and the next resume reads it again.

DDL events flush the current batch and are never inserted. The DDL HANDLING (auto-snapshot + resolver swap + schema_changes record) is NOT done here on purpose: this loop runs behind the parser through a buffered channel, so by the time a DDL reaches it the parser has already decoded — and, for a new table, skipped — the rows that followed the DDL in the binlog. That work lives in the parser's synchronous DDL hook (StreamParser.SetSyncDDLHook, #396). ErrStopWithoutFlush, as the cause a caller cancels One's context with (context.WithCancelCause), stops the stream WITHOUT writing what it holds: neither the pending batch nor the position. A process that lost its capture lock must not write after it knows, because another process may be capturing into the same index by then. Nothing is lost: the events after the last durable checkpoint are read again by whoever captures next, whose resume cleanup drops any row of a batch that was already in flight. One returns this error when it stopped this way.

var ResumeCleanupWait = DefaultResumeCleanupWait

ResumeCleanupWait is the effective ceiling, set from --cleanup-wait-timeout at startup. A var for the same reason indexer.WriteTimeout is one: a flag sets it and tests shorten it. Zero turns the look off and the cleanup starts at once, as it did before #1708; negative is rejected by One.

Functions

func FlavorOnce

func FlavorOnce(fn func(flavor string)) func(flavor string)

FlavorOnce wraps fn for use as Hooks.OnFlavorResolved by a caller that runs One more than once on the same hooks (a restart loop): fn runs for the first resolved flavor only. Safe for concurrent calls.

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

One runs one complete replication stream — connect, validate, resolve identity, snapshot, gap-check, sync, index, checkpoint — until ctx is cancelled or a fatal error occurs. It is self-contained by design: no package globals, no signal handling, safe to run N instances concurrently (each against its own index database).

One restarts itself, in process, when the parser stops with parser.ErrResentCutTransaction: the connection to the source dropped in the middle of a transaction whose first rows were already captured, and the source re-sent it whole. Restarting is what indexes it once: the resume cleanup of the last checkpoint deletes the partial copy (its GTID is not in the saved set), and the transaction is read again in full. Every caller (stream, up, watch and its console sources) gets that without its own restart policy. Any other error returns unchanged. A restart that fails the same way again within a minute counts toward a limit; past it the error is returned, so a source that keeps cutting transactions stops loudly instead of spinning.

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).

func StartMetricsServerWith

func StartMetricsServerWith(addr string, more map[string]http.Handler) (shutdown func(), err error)

StartMetricsServerWith is StartMetricsServer with more routes on the same listener, by ServeMux pattern ("GET /status"): for a process that has something of its own to say where its metrics are. Give no pattern for /metrics: the same one panics as a route registered twice, and one with a method ("GET /metrics") is taken as more specific and replaces the metrics.

Types

type Config

type Config struct {
	IndexDSN  string
	SourceDSN string
	// Flavor is the DECLARED source flavor: "mysql", "mariadb", or empty.
	// One asks the server (Deps.DetectSourceFlavor) and runs as what it
	// reports: empty lets detection decide, a declared flavor that contradicts
	// it refuses, and a failed detection is accepted only when declared (with
	// a warning). The resolved value 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
	// DetectSourceFlavor asks the source what it is (metadata.DetectSourceFlavor).
	DetectSourceFlavor func(db *sql.DB) (flavor, version string, err 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 EarlierCleanupError

type EarlierCleanupError struct {
	// ConnectionID is the index server's id for the oldest cleanup still
	// running, the value KILL takes.
	ConnectionID uint64
	// Running is how long that statement had been executing at the last look.
	Running time.Duration
	// Waited is how long this start waited before giving up.
	Waited time.Duration
	// Count is how many such statements were running at the last look.
	Count int
}

EarlierCleanupError is the ceiling failure with what was seen.

func (*EarlierCleanupError) Error

func (e *EarlierCleanupError) Error() string

func (*EarlierCleanupError) Unwrap

func (e *EarlierCleanupError) Unwrap() error

type GapRefusedError

type GapRefusedError struct {
	// contains filtered or unexported fields
}

GapRefusedError is the --no-gap-fill refusal: the checkpoint names a position the source can no longer serve (binlogs purged, GTIDs inside the source's purged set, or a same-named file regenerated after RESET MASTER) and the operator asked not to auto-advance past it. msg is the gap detector's human description, which names binlog files and GTIDs — operator-facing only; the type's usage-telemetry class is binlog_not_found, the same bucket as the server's own 1236 for the same condition.

func (*GapRefusedError) Error

func (e *GapRefusedError) Error() string

func (*GapRefusedError) TelemetryClass

func (e *GapRefusedError) TelemetryClass() string

TelemetryClass implements telemetry.Classed.

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)
	// OnSourceConnected fires once per run, after the source connection is
	// open and its binlog settings pass validation, before the schema
	// snapshot: the console's first-run list shows the connection as its own
	// step (#1606).
	OnSourceConnected func()
	// OnPhase names a startup step that runs BEFORE capture and can take
	// minutes, so a supervisor shows what the stream is doing instead of an
	// unqualified "pending" (#1690). Fires with a Phase* constant when the
	// step starts and with "" when it ends — including on the error path, so
	// a failed step never leaves a phase stuck. Only steps that can outlast
	// an operator's patience get one; the fast ones stay silent.
	OnPhase func(phase string)
	// OnPhaseDetail qualifies the phase OnPhase last named, in a few words for
	// an operator ("connection 812, running for 14m0s"). It fires right after
	// OnPhase, again whenever the detail changes, and with "" when the phase
	// ends. It is text to show, never to decide on.
	OnPhaseDetail func(detail string)
	// OnFlavorResolved fires once per run with the flavor the stream will
	// capture as, right after the source was asked what it is. A caller that
	// hands the flavor to other work (extension source jobs) takes it from
	// here, not from the declared Config.Flavor, which may be empty.
	OnFlavorResolved func(flavor 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.

type SourceRenumberedError

type SourceRenumberedError struct {
	// contains filtered or unexported fields
}

SourceRenumberedError is the MariaDB refusal to resume a GTID checkpoint from a source whose GTID numbering went backwards. It carries the steps that resume capture.

func (*SourceRenumberedError) Error

func (e *SourceRenumberedError) Error() string

func (*SourceRenumberedError) TelemetryClass

func (e *SourceRenumberedError) TelemetryClass() string

TelemetryClass implements telemetry.Classed: the same bucket as the server's own 1236 for this condition.