preflight

package
v1.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 15, 2026 License: Apache-2.0 Imports: 7 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var Builders = []Builder{
	{CategoryConnectivity, "connectivity", BuildConnectivityChecks},
	{CategoryReplication, "replication", BuildReplicationChecks},
	{CategoryAccess, "access", BuildAccessChecks},
	{CategorySchema, "schema", BuildSchemaChecks},
	{CategoryResources, "resources", BuildResourcesChecks},
}

Builders is the registry of category builders. Adding a new category = one Builder entry here + one flag declaration on checkCmd.

Functions

func BuildAccessChecks 

func BuildAccessChecks(cfg *stream.Config) ([]Check, CleanupFunc)

BuildAccessChecks returns the access-preflight checks applicable to cfg, plus a cleanup function that closes the shared source connection.

func BuildChecks 

func BuildChecks(cfg *stream.Config, selected []Category) ([]Check, CleanupFunc)

BuildChecks returns the concrete checks for the selected categories, preserving the registration order in Builders, plus a single cleanup function that releases every category's resources. The returned cleanup is always non-nil; callers can defer it unconditionally. An empty selection runs every registered category.

func BuildConnectivityChecks 

func BuildConnectivityChecks(cfg *stream.Config) ([]Check, CleanupFunc)

BuildConnectivityChecks returns the connectivity checks applicable to cfg. A source check is added when a source postgres URL is configured; a target check is added when a postgres target is configured. Each check opens its own conn (to its own URL), so no shared cleanup is needed.

func BuildReplicationChecks 

func BuildReplicationChecks(cfg *stream.Config) ([]Check, CleanupFunc)

BuildReplicationChecks returns the replication-preflight checks applicable to cfg, plus a cleanup function that closes the shared source connection. Replication checks only apply when the source is configured with a replication slot.

func BuildResourcesChecks 

func BuildResourcesChecks(cfg *stream.Config) ([]Check, CleanupFunc)

BuildResourcesChecks returns the resource-capacity preflight checks applicable to cfg, plus a cleanup function that closes the shared source connection. The snapshot connection-headroom check only applies when a data snapshot is configured (it sizes snapshot_workers × table_workers against the source's max_connections).

func BuildSchemaChecks 

func BuildSchemaChecks(cfg *stream.Config) ([]Check, CleanupFunc)

BuildSchemaChecks returns the schema-preflight checks applicable to cfg, plus a cleanup function that closes the shared source (and, when the target is Postgres, target) connection. Schema checks cover every table pgstream reads (snapshot and replication), so they use the combined access table selection. The range-type and extension checks are target specific and are only added when the target is Postgres.

Types

type Builder 

type Builder struct {
	Category Category
	Flag     string
	Build    func(*stream.Config) ([]Check, CleanupFunc)
}

Builder turns a stream.Config into the concrete checks for a category, plus an optional cleanup function that releases resources the checks share (e.g. a Postgres connection). Each new category adds an entry to Builders and a matching CLI flag in cmd/root_cmd.go.

type Category 

type Category string

Category groups checks of the same concern so callers can opt in by category via CLI flags. New categories are added as new check sets land — see docs/migration_preflight_issue.md for the planned ones. 

const (
	CategoryConnectivity Category = "connectivity"
	CategoryReplication  Category = "replication"
	CategoryAccess       Category = "access"
	CategorySchema       Category = "schema"
	CategoryResources    Category = "resources"
)

type Check 

type Check interface {
	Name() string
	Run(ctx context.Context) ([]Finding, error)
}

Check is the minimal contract every preflight check must satisfy. Run returns the findings the check produced; a non-nil error means the check itself could not complete (distinct from finding a problem with the system under test).

type CheckResult 

type CheckResult struct {
	Name     string    `json:"name"`
	Findings []Finding `json:"findings"`
	Err      error     `json:"-"`
	// Details holds optional structured context from a Detailer check. Keys are
	// merged into the result's JSON object alongside name/findings/error.
	Details map[string]any `json:"-"`
}

CheckResult bundles a check's name with whatever it produced.

func (CheckResult) MarshalJSON 

func (r CheckResult) MarshalJSON() ([]byte, error)

MarshalJSON renders Err as a string so the report is consumable from a non-Go process (the default error marshaling drops the message), and merges any Details keys into the object.

type CleanupFunc 

type CleanupFunc func(context.Context) error

CleanupFunc releases any resources a builder set up (e.g. a shared Postgres connection). Builders return nil when there's nothing to clean up.

type ConnectivityCheck 

type ConnectivityCheck struct {
	Label string
	URL   string
}

ConnectivityCheck verifies a Postgres URL accepts a connection and answers a ping. A connection or ping failure is reported as a finding (not a check error), since establishing connectivity is the purpose of the check.

func (*ConnectivityCheck) Name 

func (c *ConnectivityCheck) Name() string

func (*ConnectivityCheck) Run 

func (c *ConnectivityCheck) Run(ctx context.Context) ([]Finding, error)

type Detailer 

type Detailer interface {
	Details() map[string]any
}

Detailer is an optional interface a Check may implement to attach structured, non-finding context to its result (e.g. the set of extensions it inspected). The engine calls Details after Run, so a check populates it from state it gathered while running. Each key is merged into the check's JSON object; it is only surfaced in the JSON report, not the human-readable one.

type Finding 

type Finding struct {
	Message string `json:"message"`
}

Finding describes a single issue detected by a Check. Every finding is an error — a check that finds nothing wrong returns no findings at all.

type PostgresRangeTypeCheck 

type PostgresRangeTypeCheck struct {
	Source    postgres.AcquireFunc
	Selection stream.TableSelection
}

PostgresRangeTypeCheck verifies that every in-scope range/multirange column uses a range type pgstream's Postgres target writer can actually encode.

func (*PostgresRangeTypeCheck) Name 

func (c *PostgresRangeTypeCheck) Name() string

func (*PostgresRangeTypeCheck) Run 

func (c *PostgresRangeTypeCheck) Run(ctx context.Context) ([]Finding, error)

type ProgressFunc 

type ProgressFunc func(idx, total int, name string)

ProgressFunc is invoked just before each check runs. idx is 1-based.

type ReplicaIdentityCheck 

type ReplicaIdentityCheck struct {
	Source    postgres.AcquireFunc
	Selection stream.TableSelection
}

ReplicaIdentityCheck verifies that every in-scope table has a REPLICA IDENTITY sufficient for logical replication of UPDATE/DELETE WAL events. Anything insufficient means those events would silently be skipped at run time. The check inspects only the tables that pass the user's include/exclude filter (TableSelection) so unrelated tables don't pollute the report.

func (*ReplicaIdentityCheck) Name 

func (c *ReplicaIdentityCheck) Name() string

func (*ReplicaIdentityCheck) Run 

func (c *ReplicaIdentityCheck) Run(ctx context.Context) ([]Finding, error)

type ReplicationRoleAttrCheck 

type ReplicationRoleAttrCheck struct {
	Source postgres.AcquireFunc
}

ReplicationRoleAttrCheck verifies the current source role has the REPLICATION attribute, which is required to open a logical replication slot.

func (*ReplicationRoleAttrCheck) Name 

func (c *ReplicationRoleAttrCheck) Name() string

func (*ReplicationRoleAttrCheck) Run 

func (c *ReplicationRoleAttrCheck) Run(ctx context.Context) ([]Finding, error)

type ReplicationSlotHeadroomCheck 

type ReplicationSlotHeadroomCheck struct {
	Source postgres.AcquireFunc
}

ReplicationSlotHeadroomCheck reports whether the source has at least one slot still available before max_replication_slots is reached.

func (*ReplicationSlotHeadroomCheck) Name 

func (c *ReplicationSlotHeadroomCheck) Name() string

func (*ReplicationSlotHeadroomCheck) Run 

func (c *ReplicationSlotHeadroomCheck) Run(ctx context.Context) ([]Finding, error)

type Report 

type Report struct {
	Results []CheckResult `json:"results"`
}

Report is the outcome of running a set of checks.

func Run 

func Run(ctx context.Context, checks []Check, opts ...RunOption) Report

Run executes every check in order. A check returning an error does not stop the run; subsequent checks still execute and the error is captured in the report alongside the findings.

func (Report) HasErrors 

func (r Report) HasErrors() bool

HasErrors reports whether any check produced findings or failed to complete.

type ReportPrinter 

type ReportPrinter struct {
	Report Report
}

ReportPrinter renders a Report for display. It satisfies the cmd-side printer contract (PrettyPrint string + json.Marshaler), so existing print(cmd, p) helpers can drive it without change. Flag-driven rendering options (NoColor, Verbose, …) will live on this struct.

func (ReportPrinter) MarshalJSON 

func (p ReportPrinter) MarshalJSON() ([]byte, error)

MarshalJSON delegates to the underlying Report so a printer marshals to the same shape as the data type it wraps.

func (ReportPrinter) PrettyPrint 

func (p ReportPrinter) PrettyPrint() string

PrettyPrint renders the report as a human-readable string.

type RunOption 

type RunOption func(*runOptions)

RunOption configures Run.

func WithProgress 

func WithProgress(fn ProgressFunc) RunOption

WithProgress installs a callback invoked before each check runs. Useful for updating a spinner or log line with "running X of N: <name>".

type SchemaExtensionCompatibilityCheck 

type SchemaExtensionCompatibilityCheck struct {
	Source postgres.AcquireFunc
	Target postgres.AcquireFunc
	// contains filtered or unexported fields
}

SchemaExtensionCompatibilityCheck verifies that every extension installed on the source database is also installed on the Postgres target. pgstream replicates the schema and data of objects that depend on extensions (custom types, functions, operators, index access methods), but it never installs the extensions themselves. It reports the full set of source extensions it inspected via Details (source_extensions), regardless of the outcome.

func (*SchemaExtensionCompatibilityCheck) Details 

func (c *SchemaExtensionCompatibilityCheck) Details() map[string]any

Details exposes every extension installed on the source as a string array under source_extensions, so the report records what was inspected even when nothing is missing.

func (*SchemaExtensionCompatibilityCheck) Name 

func (c *SchemaExtensionCompatibilityCheck) Name() string

func (*SchemaExtensionCompatibilityCheck) Run 

func (c *SchemaExtensionCompatibilityCheck) Run(ctx context.Context) ([]Finding, error)

type SchemaTypeCompatibilityCheck 

type SchemaTypeCompatibilityCheck struct {
	Source    postgres.AcquireFunc
	Selection stream.TableSelection
}

SchemaTypeCompatibilityCheck verifies that pgstream can decode every column of every in-scope table. A column type is considered supported when either pgx's static type map natively handles it, or pgstream adds its own handling on top of pgx (pgstreamSupportedTypes).

func (*SchemaTypeCompatibilityCheck) Name 

func (c *SchemaTypeCompatibilityCheck) Name() string

func (*SchemaTypeCompatibilityCheck) Run 

func (c *SchemaTypeCompatibilityCheck) Run(ctx context.Context) ([]Finding, error)

type SnapshotConnectionsCheck 

type SnapshotConnectionsCheck struct {
	Source postgres.AcquireFunc
	// Demand is the number of concurrent connections the snapshot will open at
	// peak (snapshot_workers × table_workers).
	Demand uint
}

SnapshotConnectionsCheck verifies the source Postgres has enough spare connection slots to serve the snapshot's peak concurrency (snapshot_workers × table_workers) on top of what is already in use, without exceeding max_connections. Non-superuser roles cannot use the slots reserved by superuser_reserved_connections, so those are excluded from the headroom.

func (*SnapshotConnectionsCheck) Name 

func (c *SnapshotConnectionsCheck) Name() string

func (*SnapshotConnectionsCheck) Run 

func (c *SnapshotConnectionsCheck) Run(ctx context.Context) ([]Finding, error)

type SourceSequenceSelectPrivilegesCheck 

type SourceSequenceSelectPrivilegesCheck struct {
	Source    postgres.AcquireFunc
	Selection stream.TableSelection
}

SourceSequenceSelectPrivilegesCheck verifies that the source Postgres role can read every in-scope sequence pgstream may need to snapshot.

func (*SourceSequenceSelectPrivilegesCheck) Name 

func (c *SourceSequenceSelectPrivilegesCheck) Name() string

func (*SourceSequenceSelectPrivilegesCheck) Run 

func (c *SourceSequenceSelectPrivilegesCheck) Run(ctx context.Context) ([]Finding, error)

type SourceTableSelectPrivilegesCheck 

type SourceTableSelectPrivilegesCheck struct {
	Source    postgres.AcquireFunc
	Selection stream.TableSelection
}

SourceTableSelectPrivilegesCheck verifies that the source Postgres role can read every table pgstream may need to snapshot or replicate.

func (*SourceTableSelectPrivilegesCheck) Name 

func (c *SourceTableSelectPrivilegesCheck) Name() string

func (*SourceTableSelectPrivilegesCheck) Run 

func (c *SourceTableSelectPrivilegesCheck) Run(ctx context.Context) ([]Finding, error)

type WAL2JSONCheck 

type WAL2JSONCheck struct {
	Source postgres.AcquireFunc
}

WAL2JSONCheck verifies that the wal2json output plugin is installed and loadable on the source. pgstream decodes WAL through wal2json.

func (*WAL2JSONCheck) Name 

func (c *WAL2JSONCheck) Name() string

func (*WAL2JSONCheck) Run 

func (c *WAL2JSONCheck) Run(ctx context.Context) ([]Finding, error)

type WALLevelCheck 

type WALLevelCheck struct {
	Source postgres.AcquireFunc
}

WALLevelCheck verifies the source Postgres has `wal_level=logical`, which pgstream's replication path requires.

func (*WALLevelCheck) Name 

func (c *WALLevelCheck) Name() string

func (*WALLevelCheck) Run 

func (c *WALLevelCheck) Run(ctx context.Context) ([]Finding, error)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.