parser

Documentation

Overview

Package parser reads MySQL ROW-format binlog files and emits typed events. It uses go-mysql-org/go-mysql for low-level binlog decoding and the metadata.Resolver to map column ordinals to column names.

Package parser — this file provides SkipCounters, the capture-time tally of events the daemon READ from the stream and chose to DROP (#1034): per-reason monotonic counters, persisted alongside the stream checkpoint so `bintrail status` can render a Capture health verdict that survives daemon restarts.

This is the sibling of the #649 continuity verdict: continuity answers "did the stream LOSE events it never received?" (a binlog gap); these counters answer "did the stream DISCARD events it did receive?" (e.g. the column-count guard rejecting every row against a stale/corrupt snapshot). Both end the same way for the user — a restore window they believe exists and doesn't — but without these counters the second failure was invisible: the checkpoint stayed fresh and continuity honestly reported "no gaps" while 100% of rows were skipped.

Package parser — this file provides StreamParser, which reads events from a BinlogStreamer (network replication) and emits them on the same Event channel as Parser (file-based). Both use the shared handleRows function internally.

Index

• Constants
• func BuildPKValues(pkColumns []metadata.ColumnMeta, row map[string]any) string
• func ChangedColumns(before, after map[string]any) []string
• func GTIDExecutedHint(flavor string) string
• type DDLKind
• type Event
• type EventType
• type Filters
• type Parser
  • func New(binlogDir string, resolver *metadata.Resolver, filters Filters, ...) *Parser
  • func (p *Parser) ParseFile(ctx context.Context, filename string, events chan<- Event) error
  • func (p *Parser) ParseFiles(ctx context.Context, filenames []string, events chan<- Event) error
  • func (p *Parser) SetSkipCounters(s *SkipCounters)
  • func (p *Parser) SwapResolver(r *metadata.Resolver)
• type SkipAttribution
• type SkipCounters
  • func NewSkipCounters(logger *slog.Logger) *SkipCounters
  • func (c *SkipCounters) RecordCaptured()
  • func (c *SkipCounters) RecordSkip(reason string)
  • func (c *SkipCounters) RecordSkipAttributed(reason string, attr SkipAttribution)
  • func (c *SkipCounters) Seed(raw string) error
  • func (c *SkipCounters) SeedPreserving(raw string) error
  • func (c *SkipCounters) Snapshot() (string, error)
  • func (c *SkipCounters) Total() int64
• type SkipStat
• type StreamParser
  • func NewStreamParser(resolver *metadata.Resolver, filters Filters, logger *slog.Logger) *StreamParser
  • func (sp *StreamParser) Run(ctx context.Context, streamer *replication.BinlogStreamer, out chan<- Event) error
  • func (sp *StreamParser) SetFlavor(flavor string)
  • func (sp *StreamParser) SetSkipCounters(c *SkipCounters)
  • func (sp *StreamParser) SetSyncDDLHook(fn func(Event) error)
  • func (sp *StreamParser) SwapResolver(r *metadata.Resolver)

Constants

const (
	EventInsert   = event.EventInsert
	EventUpdate   = event.EventUpdate
	EventDelete   = event.EventDelete
	EventDDL      = event.EventDDL
	EventGTID     = event.EventGTID
	EventSnapshot = event.EventSnapshot
	EventCommit   = event.EventCommit
)

const (
	DDLAlterTable    = event.DDLAlterTable
	DDLCreateTable   = event.DDLCreateTable
	DDLDropTable     = event.DDLDropTable
	DDLRenameTable   = event.DDLRenameTable
	DDLTruncateTable = event.DDLTruncateTable
)

const (
	// SkipColumnCountMismatch — the binlog TABLE_MAP column count diverged
	// from the schema snapshot; indexing would map values to wrong names.
	SkipColumnCountMismatch = "column_count_mismatch"
	// SkipTableNotInSnapshot — the row's table is absent from the schema
	// snapshot (system schemas the snapshot deliberately excludes are NOT
	// counted — e.g. mysql.rds_heartbeat2 is a routine, permanent skip that
	// must never mark capture degraded).
	SkipTableNotInSnapshot = "table_not_in_snapshot"
	// SkipTableExcludedFromSnapshot — the row's table is absent from the
	// snapshot because the snapshot VALIDATOR excluded it (#1051: no explicit
	// primary key, or a non-InnoDB engine). Split from SkipTableNotInSnapshot
	// (#1296) because the two have opposite remedies and merging them hands the
	// operator a remediation that cannot converge: re-snapshotting excludes
	// this table again, forever. The fix is on the source's DDL, not here.
	SkipTableExcludedFromSnapshot = "table_excluded_from_snapshot"
	// SkipNoResolver — no schema snapshot was available at all.
	SkipNoResolver = "no_resolver"
	// SkipUnhandledRowEvent — a RowsEvent type bintrail does not decode
	// (e.g. PARTIAL_UPDATE_ROWS_EVENT under binlog_row_value_options).
	SkipUnhandledRowEvent = "unhandled_row_event"
	// SkipStatementFormatDML — a STATEMENT/MIXED-format DML whose row image
	// is not in the binlog (#999); the change cannot be captured.
	SkipStatementFormatDML = "statement_format_dml"
	// SkipUnreadablePreviousLedger — meta-reason (#1206): the previous run's
	// persisted ledger existed but could not be parsed at restart. Not an
	// event skip — it preserves the FACT that a loss tally may have been
	// destroyed, so `status` stays non-clean (and --fail-on-gap keeps failing
	// closed) instead of the next checkpoint silently laundering the evidence
	// into "{}" = OK. Cleared only by the operator acknowledgement runbook
	// (clear the ledger with the daemon stopped).
	SkipUnreadablePreviousLedger = "unreadable_previous_ledger"
)

Capture-skip reason identifiers. Stable strings: they are persisted verbatim in stream_state.capture_skips and surfaced by `bintrail status` (text and JSON), so renaming one silently splits its history into two counters.

const MaxLedgerTables = 8

MaxLedgerTables caps SkipStat.Tables — enough names to act on in one sitting without letting a persisted, monotonic document grow unbounded.

const SkipEscalationThreshold = 100

SkipEscalationThreshold is the number of CONSECUTIVE skipped events (no captured event in between) after which SkipCounters emits a single ERROR with remediation guidance — the per-event WARNs blend into noise exactly when they matter most (#1034 observed 100% skips for ~2 days). One ERROR per degraded episode: the flag re-arms only after an event is captured.

Functions

func BuildPKValues

func BuildPKValues(pkColumns []metadata.ColumnMeta, row map[string]any) string

BuildPKValues forwards to event.BuildPKValues (kept for back-compat; the canonical doc lives on event.BuildPKValues).

func ChangedColumns

func ChangedColumns(before, after map[string]any) []string

ChangedColumns forwards to event.ChangedColumns (kept for back-compat; the canonical doc lives on event.ChangedColumns).

func GTIDExecutedHint

func GTIDExecutedHint(flavor string) string

GTIDExecutedHint returns the flavor-appropriate system variable an operator should read to obtain the source's current executed GTID set, for remediation messages that direct the operator to --start-gtid. MySQL has gtid_executed; MariaDB has no such variable — its analog is gtid_binlog_pos (see detectMariaDBGTIDGap in internal/streamrun, which queries the same variable for gap detection).

Types

type DDLKind

type DDLKind = event.DDLKind

DDLKind is the source-agnostic DDL-kind, moved to internal/event (#528). Aliased here so the parser and existing callers keep using parser.DDLKind.

type Event

type Event = event.Event

Event is the source-agnostic change event, moved to internal/event (#528). Aliased here so the binlog parser and existing callers keep using parser.Event.

type EventType

type EventType = event.EventType

EventType is the source-agnostic event-type, moved to internal/event (#528). Aliased here so the capture layer and existing callers keep using parser.EventType.

type Filters

type Filters = event.Filters

Filters is the source-agnostic schema/table filter, moved to internal/event (#528). Aliased here so the parser and existing callers keep using parser.Filters.

type Parser

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

Parser reads binlog files and emits Events onto a channel.

func New

func New(binlogDir string, resolver *metadata.Resolver, filters Filters, logger *slog.Logger) *Parser

New creates a Parser that reads from binlogDir, resolves column names via resolver, and applies the given filters. logger may be nil, in which case slog.Default() is used.

func (*Parser) ParseFile

func (p *Parser) ParseFile(ctx context.Context, filename string, events chan<- Event) error

ParseFile parses a single binlog file and sends matching events to the channel. It stops early if ctx is cancelled.

func (*Parser) ParseFiles

func (p *Parser) ParseFiles(ctx context.Context, filenames []string, events chan<- Event) error

ParseFiles parses multiple binlog files in order, sending events to the channel.

func (*Parser) SetSkipCounters

func (p *Parser) SetSkipCounters(s *SkipCounters)

SetSkipCounters wires an optional skip tally into the file-index path (#1199). Unlike the stream (where the counters persist to stream_state.capture_skips), file-mode counters are run-scoped: the caller summarizes them at end of run. They exist because a VALIDATION-EXCLUDED table's skips no longer fail the file via the #778 gap tracker, and per-event WARNs alone scroll away — without a tally the run would end with no aggregate signal. Nil (the default) keeps every RecordSkip a no-op.

func (*Parser) SwapResolver

func (p *Parser) SwapResolver(r *metadata.Resolver)

SwapResolver atomically replaces the resolver used for column resolution and updates schemaVersion to the new resolver's SnapshotID. Safe to call concurrently while ParseFile is running.

type SkipAttribution

type SkipAttribution struct {
	File          string
	Pos           uint64
	StatementType string
	ConnectionID  uint32
	// Schema and Table name the table whose rows were dropped. Kept as two
	// plain strings, not a slice: RecordSkipAttributed compares this struct
	// against its zero value with ==, which a slice field would break at
	// compile time.
	Schema string
	Table  string
	// Detail is a short human explanation of THIS skip (today: the snapshot
	// validator's exclusion reason). Optional.
	Detail string
}

SkipAttribution locates one skipped event: binlog file/pos, the statement keyword (derived from the statement text, which is then discarded), and the connection id from the QUERY event post-header. Never carries the statement text itself.

type SkipCounters

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

SkipCounters aggregates capture-time skips by reason. Safe for concurrent use; every method is nil-receiver-safe so the file-index path (which has its own failure mechanism, the #778 gap tracker) can simply pass nil.

func NewSkipCounters

func NewSkipCounters(logger *slog.Logger) *SkipCounters

NewSkipCounters creates an empty counter set. logger (nil = slog.Default()) receives the single escalation ERROR when SkipEscalationThreshold consecutive events have been skipped.

func (*SkipCounters) RecordCaptured

func (c *SkipCounters) RecordCaptured()

RecordCaptured notes that an event cleared every guard and was emitted for indexing: the consecutive-skip run is broken and the escalation re-arms. The per-reason tallies are monotonic and deliberately NOT reset.

func (*SkipCounters) RecordSkip

func (c *SkipCounters) RecordSkip(reason string)

RecordSkip notes one skipped event under reason, stamping the current time. When SkipEscalationThreshold consecutive events have been skipped it emits ONE ERROR with remediation (re-armed by the next RecordCaptured).

func (*SkipCounters) RecordSkipAttributed

func (c *SkipCounters) RecordSkipAttributed(reason string, attr SkipAttribution)

RecordSkipAttributed is RecordSkip carrying the skipped event's location (#999). A zero attribution leaves any previously stamped attribution for the reason in place — an unattributed count must not erase the last useful lead (consequence: the attribution can lag last_at if a caller mixes attributed and unattributed skips of one reason).

func (*SkipCounters) Seed

func (c *SkipCounters) Seed(raw string) error

Seed replaces the counters with a previously persisted Snapshot document, so a daemon restart resumes the monotonic tallies instead of silently zeroing the DEGRADED verdict. Empty raw is a no-op (nothing persisted yet); invalid JSON is returned as an error and leaves the counters unchanged.

func (*SkipCounters) SeedPreserving

func (c *SkipCounters) SeedPreserving(raw string) error

SeedPreserving is Seed for the daemon restart path (#1206). A document that fails to parse is NOT discarded into fresh counters: the failure itself is recorded under SkipUnreadablePreviousLedger, so the next checkpoint persists a non-clean ledger instead of overwriting possible loss evidence with the affirmative "{}". The parse error is returned for the caller to log.

func (*SkipCounters) Snapshot

func (c *SkipCounters) Snapshot() (string, error)

Snapshot returns the JSON document persisted to stream_state.capture_skips: {"<reason>":{"count":N,"last_at":"RFC3339"}}. Always valid JSON — "{}" when nothing was skipped, which is the affirmative evaluated-and-clean marker that lets `status` print "Capture health: OK" (a NULL column means no skip-aware daemon ever wrote, so OK must not be asserted).

func (*SkipCounters) Total

func (c *SkipCounters) Total() int64

Total returns the sum of all per-reason counts (test/display helper).

type SkipStat

type SkipStat struct {
	Count  int64     `json:"count"`
	LastAt time.Time `json:"last_at"`
	// Last-seen attribution (#999): where the most recent skip happened, enough
	// to hunt the source without ever storing statement text (a DML statement
	// embeds row VALUES — same reason the per-event WARN omits it). Optional —
	// only the STREAMING statement-DML site stamps these today (file-mode
	// `bintrail index` detects the same drops but persists no ledger at all);
	// omitempty keeps documents persisted before this change (and reasons
	// without attribution) unchanged.
	LastFile          string `json:"last_file,omitempty"`
	LastPos           uint64 `json:"last_pos,omitempty"`
	LastStatementType string `json:"last_statement_type,omitempty"`
	LastConnectionID  uint32 `json:"last_connection_id,omitempty"`

	// Tables lists the distinct "schema.table" names this reason dropped rows
	// for, capped at MaxLedgerTables (#1296). Without it an operator is told
	// changes are missing but not WHICH table's — the first question they ask,
	// and one only the daemon log could answer before. Absent on a ledger
	// written by an older daemon: consumers must then name no table at all
	// rather than present an empty list as "none".
	Tables []string `json:"tables,omitempty"`
	// TablesTruncated records that more distinct tables were skipped than
	// Tables holds. The list is capped because this document is persisted in a
	// single stream_state column and grows monotonically — an unfiltered source
	// with thousands of unsnapshotted tables would grow it without bound. With
	// no flag, a capped list would read as the complete set.
	TablesTruncated bool `json:"tables_truncated,omitempty"`
	// LastDetail is the newest per-skip explanation the detection site could
	// state (today: the snapshot validator's exclusion reason). Display-only
	// free text, never parsed.
	LastDetail string `json:"last_detail,omitempty"`
}

SkipStat is one reason's monotonic tally. The JSON field names are the persistence format of stream_state.capture_skips — internal/status decodes the same shape independently (it deliberately does not import this package).

type StreamParser

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

StreamParser reads events from a live BinlogStreamer and sends parsed row events to an output channel, mirroring the interface of Parser but without requiring binlog files on disk.

func NewStreamParser

func NewStreamParser(resolver *metadata.Resolver, filters Filters, logger *slog.Logger) *StreamParser

NewStreamParser creates a StreamParser that resolves column names via resolver and applies the given filters. logger may be nil, in which case slog.Default() is used.

func (*StreamParser) Run

func (sp *StreamParser) Run(ctx context.Context, streamer *replication.BinlogStreamer, out chan<- Event) error

Run reads events from the streamer and sends matching row events to out. It tracks the current binlog filename (from RotateEvent) and GTID (from GTIDEvent), and uses them to populate each emitted Event.

Returns nil when the context is cancelled (graceful shutdown) or when the streamer is closed. Returns a non-nil error on network or decode failure.

func (*StreamParser) SetFlavor

func (sp *StreamParser) SetFlavor(flavor string)

SetFlavor sets the source flavor ("mysql" or "mariadb", matching gomysql.MySQLFlavor/MariaDBFlavor) used to word the position-wraparound error (see Run) flavor-appropriately. Must be called before the goroutine that runs Run is spawned — Run reads it without synchronization, relying on the happens-before edge from the spawning statement (mirrors how resolver/filters are fixed before Run begins). Empty (never called) defaults to MySQL wording.

func (*StreamParser) SetSkipCounters

func (sp *StreamParser) SetSkipCounters(c *SkipCounters)

SetSkipCounters wires the capture-skip tally (#1034) Run records discards into. Must be called before the goroutine that runs Run is spawned — Run reads it without synchronization, relying on the happens-before edge from the spawning statement (the SetFlavor contract). nil disables counting.

func (*StreamParser) SetSyncDDLHook

func (sp *StreamParser) SetSyncDDLHook(fn func(Event) error)

SetSyncDDLHook registers fn to run synchronously inside Run, for each DDL event, BEFORE that event is emitted on out and BEFORE any subsequent binlog event is decoded.

This is the ONLY correct place for the auto-snapshot-on-DDL work (#396): the binlog is sequential — `CREATE TABLE t; INSERT INTO t;` puts the row events immediately after the DDL — but the parser goroutine runs ahead of the consumer through the events channel. A consumer-side handler swaps the resolver only after the parser has already decoded (and, for an unknown table, silently skipped) the rows that followed the DDL. Blocking the parse loop until fn returns closes that window: fn typically takes a fresh schema snapshot and calls SwapResolver, so the very next row event decodes with the post-DDL schema.

fn returning a non-nil error aborts Run immediately, WITHOUT ever emitting the DDL event (#760): if the post-DDL snapshot/resolver refresh failed, the resolver is stale and every row event that follows this DDL in the binlog would otherwise be silently skipped as "column count mismatch" / "table not in snapshot" while the stream checkpoint keeps advancing past them — a permanent, unmarked loss. The checkpoint (in this package's consumers) advances off events it actually receives, including the DDL event itself; withholding that event on hook failure means the checkpoint stays exactly where it was BEFORE this DDL. A supervisor restart then resumes from there, re-reads the same DDL off the binlog, re-runs this hook (retrying the snapshot), and only then decodes the rows that follow — with a fresh resolver. Emitting the DDL event first and running fn after would let the checkpoint advance past the DDL before a failure is even known, defeating the retry.

While fn runs, no new events are produced (the replication connection buffers server-side); consumers keep draining already-emitted events. Safe to call concurrently with Run.

func (*StreamParser) SwapResolver

func (sp *StreamParser) SwapResolver(r *metadata.Resolver)

SwapResolver atomically replaces the resolver used for column resolution and updates schemaVersion to the new resolver's SnapshotID. Safe to call concurrently while Run is executing in another goroutine.