Documentation
¶
Overview ¶
Package indexer consumes parsed binlog events and batch-inserts them into the binlog_events table in the index MySQL database.
Index ¶
- Constants
- Variables
- func CreateIndexTables(ctx context.Context, db *sql.DB, partitions int, encrypt bool, ...) error
- func EnsureArchiveStateSchema(db *sql.DB) error
- func EnsureDatabase(cfg *mysql.Config, dbName string, log func(string)) error
- func EnsureSchema(db *sql.DB) error
- func InsertSchemaChange(db *sql.DB, ev event.Event, snapshotID *int) error
- func PartitionDate(name string) (time.Time, bool)
- func PartitionName(d time.Time) string
- func WrapSchemaMigrationErr(err error) error
- type Indexer
Constants ¶
const DefaultWriteTimeout = 3 * time.Minute
WriteTimeout bounds a single index write (e.g. a batch INSERT, the checkpoint UPSERT, or a statement-digest lookup) so a mid-statement network stall surfaces as an error in minutes instead of the kernel's ~13-16 min TCP give-up. config.Connect sets only a connect timeout, never a read/write deadline, so without this a frozen VM or an idle-dropped NLB leaves the daemon blocked — healthy to `watch`, capturing nothing and advancing no checkpoint (#959). The window sits well above any healthy batch INSERT and below the kernel give-up, and is tunable per deployment via --write-timeout for the cases where a healthy write legitimately runs long: a very large batch over a slow link, or index-side MDL contention from a concurrent partition rotation (#959).
const ( // MaxBatchSize is the largest batch whose INSERT stays under the // placeholder cap. Derived from insertColumnCount so adding a column to // the INSERT lowers the limit instead of silently re-breaking large // batches (#956). MaxBatchSize = maxPreparedStmtParams / insertColumnCount )
Variables ¶
var ErrWriteDeadline = errors.New("index write deadline exceeded")
ErrWriteDeadline tags an index write that ran out of WriteTimeout, as opposed to one the server refused. It exists because nothing else distinguished the two: every flush failure reached the caller as an opaque error, so an index that was merely SLOW (a heavy analytical read starving it) ended capture just as surely as an un-indexable event did — the daemon exited and nothing brought it back (#1482).
A deadline is the only failure class where "the same write, later" is a reasonable thing to attempt: the write never got a verdict, so the server neither accepted nor rejected it. Match with errors.Is; never on message text.
Deliberately NOT a licence to re-execute the INSERT in place. The deadline fires client-side, and the driver reacts by closing the socket without a KILL QUERY, so a slow batch usually goes on to COMMIT on the server after the client has given up. Re-running that statement would duplicate every row it wrote, permanently. The only duplicate-safe way to act on this tag is to restart the stream, which replays from the checkpoint through streamrun's dedup-on-resume — the step that exists to drop exactly those rows.
var WriteTimeout = DefaultWriteTimeout
WriteTimeout is the effective deadline, set from --write-timeout at startup (default DefaultWriteTimeout). A var (not const) so the flag can set it and tests can shrink it — a test that mutates it must NOT call t.Parallel() (it is a shared global).
Functions ¶
func CreateIndexTables ¶
func CreateIndexTables(ctx context.Context, db *sql.DB, partitions int, encrypt bool, logTable func(string)) error
CreateIndexTables creates every index table (idempotent — all DDL is CREATE TABLE IF NOT EXISTS). Shared by `bintrail init` and the control-plane supervisor, which provisions a per-source index database with the same set. logTable is invoked per table for progress output; nil is allowed (no output).
func EnsureArchiveStateSchema ¶ added in v0.77.0
EnsureArchiveStateSchema adds the archive_state columns introduced after the initial schema. Idempotent, like EnsureSchema, which calls it.
Split out so a caller that needs ONLY this table does not migrate binlog_events as a side effect (#1535). `archive reconcile` is that caller: its dry run is documented as a read-only cron drift monitor, and adding a column to the largest table in the deployment is not something such a run should ever start.
func EnsureDatabase ¶
EnsureDatabase creates the index database if it does not already exist. cfg is a parsed DSN config pointing at the MySQL server (its DBName is ignored — the connection is opened without a database so CREATE DATABASE can run); dbName is created with utf8mb4/utf8mb4_unicode_ci.
The optional log callback receives a human-readable "ready" line. Pass nil for silent provisioning (e.g. the monitor supervisor); the CLI passes a closure that prints to stdout.
func EnsureSchema ¶
EnsureSchema adds any columns introduced after the initial schema to binlog_events, schema_snapshots, and stream_state. It is idempotent — safe to call on every startup.
func InsertSchemaChange ¶
InsertSchemaChange records a DDL detection in the schema_changes table. snapshotID may be nil when no auto-snapshot was taken (file mode).
func PartitionDate ¶
PartitionDate parses the hour from a partition name like "p_2026021914". Returns the time and true on success; zero and false for p_future or other names.
func PartitionName ¶
PartitionName returns the partition name for a given hour ("p_YYYYMMDDHH").
func WrapSchemaMigrationErr ¶ added in v0.30.0
WrapSchemaMigrationErr rewrites an EnsureSchema failure for the READ plane (query, recover, verify, shim, reconstruct, recover-cascade, the MCP tools) into an actionable error.
EnsureSchema issues ALTER TABLE / CREATE TABLE statements, which a SELECT-only DSN — the natural pairing with the RBAC --profile feature — lacks the privilege for. MySQL then returns errno 1142 (ER_TABLEACCESS_DENIED_ERROR) or 1044 (ER_DBACCESS_DENIED_ERROR), whose raw message names neither the privilege cause nor the fix, so the read plane refuses to read data it could otherwise serve with an unhelpful error. Detect those two errors specifically and rewrite them to name both cause and fix; every other EnsureSchema failure (connection drop, context deadline, etc.) is not a privilege issue and passes through with a plain "schema migration" wrap so it isn't misdiagnosed.
The CAPTURE plane (index, stream, agent, rotate) legitimately requires a privileged DSN and must keep failing hard on any EnsureSchema error — those call sites must NOT use this helper.
Types ¶
type Indexer ¶
type Indexer struct {
// contains filtered or unexported fields
}
Indexer consumes event.Events from a channel and batch-inserts them into the binlog_events table.
func (*Indexer) InsertBatch ¶
InsertBatch writes a batch of events and returns the count of rows inserted. This exported method allows callers (e.g. the stream command) that need manual checkpoint control between batches.
Source Files