migration

package
v0.8.9 Latest Latest
Warning

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

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

Documentation

Index

Constants

View Source 
const (
	StateUninitialized    = "uninitialized"
	StateInitialized      = "initialized"
	StateLagsOk           = "lags_ok"
	StateFenced           = "fenced"
	StateOffsetSyncPaused = "offset_sync_paused"
	StateFenceVerified    = "fence_verified"
	StatePromoted         = "promoted"
	StateSwitched         = "switched"
)

FSM State constants

View Source 
const (
	EventInitialize  = "initialize"
	EventWaitForLags = "wait_for_lags"
	EventFence       = "fence"
	// EventPauseOffsetSync pauses cluster-link consumer offset sync
	// (--pause-consumer-offset-sync) immediately after fencing. Without the
	// opt-in the transition still fires as a pass-through so the forward
	// walk is identical either way.
	EventPauseOffsetSync = "pause_offset_sync"
	EventVerifyFence     = "verify_fence"
	EventPromote         = "promote"
	EventSwitch          = "switch"
	// EventAbortFence rolls back to initialized when the pause_offset_sync
	// step fails (from fenced) or the verify_fence step detects unrouted
	// producers (from offset_sync_paused); the transition itself unfences
	// the gateway and restores any paused sync config (see onAbortFence in
	// orchestrator.go).
	EventAbortFence = "abort_fence"
	// EventExpireVerification demotes fence_verified to fenced at FSM
	// bootstrap: the verification is a point-in-time attestation and never
	// survives a restart, so a resume re-runs the verify_fence detection
	// window. Fired only by NewMigrationOrchestrator; it has no action.
	EventExpireVerification = "expire_verification"
	// EventExpireFence demotes fenced and offset_sync_paused to lags_ok at FSM
	// bootstrap: whether the live gateway still holds the fenced CR is equally
	// a point-in-time fact. A crash or a partially-completed abort_fence
	// rollback (initial CR applied, process gone before the rolled-back state
	// reached disk) leaves the gateway unfenced while the state file still
	// records a fenced-family state. Demoting makes the resume re-apply the
	// fenced CR — a no-op rollout when the gateway never diverged — instead of
	// verifying and promoting behind a fence that may not exist. Fired only by
	// NewMigrationOrchestrator; it has no action.
	EventExpireFence = "expire_fence"
)

FSM Event constants

Variables

View Source 
var ErrUnroutedProducers = errors.New("unrouted producers detected")

ErrUnroutedProducers is returned when the post-fence check detects producers bypassing the gateway. The orchestrator catches this to trigger an EventAbortFence transition back to initialized state.

Functions

func BuildClusterLinkConfig added in v0.8.1 

func BuildClusterLinkConfig(config *MigrationConfig, apiKey, apiSecret string) clusterlink.Config

BuildClusterLinkConfig assembles a clusterlink.Config from a migration config plus runtime API credentials. Centralized here so the bookend callers in cmd/migration/execute don't duplicate the field layout.

func RestoreOffsetSync added in v0.8.1 

func RestoreOffsetSync(
	_ context.Context,
	cl clusterlink.Service,
	clCfg clusterlink.Config,
	config *MigrationConfig,
	persist func() error,
)

RestoreOffsetSync runs the post-execute restore bookend. Soft-failure semantics: ListConfigs or AlterConfigs failure prints a remediation message to stderr but does NOT propagate an error, because the switchover itself succeeded (R13). The PauseConsumerOffsetSyncFlipped marker stays true on failure so the state file records that a restore is still owed.

Diff-based restore: when MigrationConfig.ClusterLinkConfigs holds an init snapshot, restore queries the cluster link for its current consumer.offset.* state and re-applies snapshot values for keys that the disable bookend cleared (current is missing, empty, or "false"). Keys whose current value looks like a deliberate post-disable operator change (non-empty, non-"false", differs from snapshot) are left alone. When the snapshot is empty (defensive fallback for unforeseen state-file lifecycles), restore performs a single SET consumer.offset.sync.enable=true.

The ctx argument is accepted for symmetry with DisableOffsetSync but the network calls use a fresh background ctx with a per-call timeout. The restore must run even when the parent ctx is already cancelled (e.g. a signal arrived between orchestrator.Execute returning and the bookend firing) — that is the case the soft-fail semantic exists for. Each PUT gets its own timeout budget so a slow endpoint on one key cannot starve the rest of the restore.

func WarnIfPausedOnExecuteFailure added in v0.8.1 

func WarnIfPausedOnExecuteFailure(config *MigrationConfig, execErr error)

WarnIfPausedOnExecuteFailure prints a stderr remediation message when orchestrator.Execute returns an error while the cluster link is still in the disabled state (PauseConsumerOffsetSyncFlipped=true). The restore bookend only runs after a successful execute, so without this warning the operator has no signal that the cluster link is paused.

The wording is shaped by the persisted state. The fenced CR applied at EventFence stays live until SwitchGateway replaces it (or the abort_fence rollback restores the initial CR), so at fenced, offset_sync_paused, fence_verified and promoted the gateway is still blocking client traffic — all four deserve urgent copy describing the observable state. The remediation differs by shape:

  • fenced / offset_sync_paused: the same signature also arises from routine resumable stops (ctx-cancel mid-detection, a verify fetch error), so the copy deliberately does not claim a rollback failed; a re-run retries the pause or completes the rollback, and manually re-applying the initial gateway CR is a safe abort.
  • fence_verified: topics are not yet promoted, so the manual abort is still safe; a re-run re-asserts the fence and resumes forward.
  • promoted: topics are already promoted, so routing clients back to the source would diverge data — completing the switchover is the only way forward that unblocks clients, and the copy explicitly warns against re-applying the initial CR.

Every other state keeps the softer restore-owed wording: at initialized and lags_ok the fence is not up (a completed rollback whose restore is still owed, or the legacy pre-FSM-pause cohort failing before the fence), and at switched the switchover CR is live.

Soft-fail: never returns an error — this is best-effort messaging on top of the underlying execute error.

Types

type ExecutionParams 

type ExecutionParams struct {
	LagThreshold     int64
	ClusterApiKey    string
	ClusterApiSecret string
}

ExecutionParams holds the per-run runtime parameters a transition may need. It is passed to fsm.Event as the sole argument and read back by callbacks via execParamsFromEvent, rather than stashed on the orchestrator, so the values flow with the event instead of living as mutable orchestrator state.

type MigrationActions added in v0.8.9 

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

func NewMigrationActions added in v0.8.9 

func NewMigrationActions(
	gatewayService gateway.Service,
	clusterLinkService clusterlink.Service,
) *MigrationActions

func NewMigrationActionsWithOffsets added in v0.8.9 

func NewMigrationActionsWithOffsets(
	gatewayService gateway.Service,
	clusterLinkService clusterlink.Service,
	sourceOffset offset.Provider,
	destinationOffset offset.Provider,
) *MigrationActions

func (*MigrationActions) CheckLags added in v0.8.9 

func (s *MigrationActions) CheckLags(
	ctx context.Context,
	config *MigrationConfig,
	lagThreshold int64,
	clusterApiKey, clusterApiSecret string,
) error

CheckLags polls source and destination offsets until lag is below threshold

func (*MigrationActions) FenceGateway added in v0.8.9 

func (s *MigrationActions) FenceGateway(ctx context.Context, config *MigrationConfig) error

FenceGateway applies the fenced gateway CR YAML to block traffic and waits for the Confluent operator to report the gateway as Ready at the new spec generation. The wait runs without a deadline by default — the operator drives convergence and the user can Ctrl-C if a rollout wedges. An optional per-workflow rolloutTimeout caps the wait when set (via SetRolloutTimeout).

func (*MigrationActions) Initialize added in v0.8.9 

func (s *MigrationActions) Initialize(
	ctx context.Context,
	config *MigrationConfig,
	clusterApiKey, clusterApiSecret string,
) error

func (*MigrationActions) PauseOffsetSync added in v0.8.9 

func (s *MigrationActions) PauseOffsetSync(
	ctx context.Context,
	config *MigrationConfig,
	clusterApiKey, clusterApiSecret string,
	persist func() error,
) error

PauseOffsetSync runs the pause_offset_sync stage: with the operator's --pause-consumer-offset-sync opt-in it pauses cluster-link consumer offset sync immediately after fencing; otherwise it passes through so the FSM still records offset_sync_paused. The already-flipped guard makes resumes (and legacy state files whose pause ran pre-FSM) idempotent.

func (*MigrationActions) PromoteTopics added in v0.8.9 

func (s *MigrationActions) PromoteTopics(ctx context.Context, config *MigrationConfig, clusterApiKey, clusterApiSecret string) error

PromoteTopics polls offsets and promotes mirror topics that reach zero lag

func (*MigrationActions) SetPromoteBatchSize added in v0.8.9 

func (s *MigrationActions) SetPromoteBatchSize(n int)

SetPromoteBatchSize caps how many mirror topics are promoted per batch during PromoteTopics. A value of 0 (the default) means unlimited — all zero-lag topics are promoted at once. When set (>0), each batch is promoted and fully confirmed STOPPED before the next batch is submitted.

func (*MigrationActions) SetRolloutTimeout added in v0.8.9 

func (s *MigrationActions) SetRolloutTimeout(d time.Duration)

SetRolloutTimeout sets the deadline applied to gateway-readiness waits. A value of 0 means no deadline.

func (*MigrationActions) SwitchGateway added in v0.8.9 

func (s *MigrationActions) SwitchGateway(ctx context.Context, config *MigrationConfig) error

SwitchGateway applies the switchover gateway CR YAML to point to Confluent Cloud and waits for the operator to report the gateway as Ready. The wait uses the same no-deadline-by-default behavior as FenceGateway.

func (*MigrationActions) VerifyFence added in v0.8.9 

func (s *MigrationActions) VerifyFence(ctx context.Context, config *MigrationConfig) error

VerifyFence verifies the fence held: source offsets must be stable, because an increasing offset after fencing indicates a producer bypassing the gateway. When detection is disabled (DetectUnroutedProducersDuration == 0) the step succeeds immediately so the FSM still records fence_verified.

detectUnroutedProducers wraps ErrUnroutedProducers only for a real detection; a network/fetch error propagates as-is. Either way we just return it — restoring traffic (unfencing the gateway) is the state machine's job on the abort_fence rollback transition, which the orchestrator triggers only for ErrUnroutedProducers.

type MigrationConfig added in v0.8.4 

type MigrationConfig struct {
	MigrationId  string `json:"migration_id"`
	CurrentState string `json:"current_state"`

	// Gateway configuration
	KubeConfigPath string `json:"kube_config_path"`

	// Source cluster configuration
	SourceBootstrap string `json:"source_bootstrap"`

	// Destination cluster configuration
	ClusterBootstrap    string   `json:"cluster_bootstrap"`
	ClusterId           string   `json:"cluster_id"`
	ClusterRestEndpoint string   `json:"cluster_rest_endpoint"`
	ClusterLinkName     string   `json:"cluster_link_name"`
	Topics              []string `json:"topics"`

	// Migration runtime data (populated during initialization)
	ClusterLinkTopics  []string          `json:"cluster_link_topics"`
	ClusterLinkConfigs map[string]string `json:"cluster_link_configs"`

	// Operator intent: pause cluster-link consumer offset sync for the duration of execute.
	// PauseConsumerOffsetSync records the operator's choice at init time.
	// PauseConsumerOffsetSyncFlipped is set when kcp has executed the disable AlterConfigs and
	// not yet restored — supports drift detection, idempotent resume, and remediation messaging.
	PauseConsumerOffsetSync        bool `json:"pause_consumer_offset_sync"`
	PauseConsumerOffsetSyncFlipped bool `json:"pause_consumer_offset_sync_flipped"`

	// DetectUnroutedProducersDuration is the monitoring window for the post-fence
	// safety check that verifies source offsets are not still increasing before
	// promoting mirror topics. A value of 0 skips the check. An increasing offset
	// after fencing indicates a producer that bypassed the gateway and is writing
	// directly to the source cluster.
	DetectUnroutedProducersDuration time.Duration `json:"detect_unrouted_producers_duration"`

	// ConsumerOffsetSyncDrainDuration is how long the pause_offset_sync stage
	// waits after fencing before disabling the cluster link's
	// consumer.offset.sync.enable. The fence freezes source consumer offsets
	// (clients can no longer commit), so holding here lets the link run one or
	// more further sync cycles and propagate those final committed offsets to
	// the destination, minimising messages reprocessed after switchover. Only
	// has effect when PauseConsumerOffsetSync is set. Best-effort: offset sync
	// is asynchronous, so this reduces but does not eliminate duplicate
	// processing. A value of 0 (the default) skips the wait — the link is
	// disabled immediately after fencing, the prior behaviour.
	ConsumerOffsetSyncDrainDuration time.Duration `json:"consumer_offset_sync_drain_duration"`

	// Gateway CR configuration
	InitialCrName    string `json:"initial_cr_name"`
	K8sNamespace     string `json:"k8s_namespace"`
	InitialCrYAML    []byte `json:"initial_cr_yaml"`
	FencedCrYAML     []byte `json:"fenced_cr_yaml"`
	SwitchoverCrYAML []byte `json:"switchover_cr_yaml"`
}

MigrationConfig holds all domain configuration for a migration This is pure data with no behavior - just fields that get serialized

type MigrationOrchestrator 

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

MigrationOrchestrator manages the FSM lifecycle and coordinates workflow execution

func NewMigrationOrchestrator 

func NewMigrationOrchestrator(
	config *MigrationConfig,
	actions *MigrationActions,
	migrationState *MigrationState,
	stateFilePath string,
) *MigrationOrchestrator

NewMigrationOrchestrator creates a new migration orchestrator with injected dependencies

func (*MigrationOrchestrator) Execute 

func (o *MigrationOrchestrator) Execute(ctx context.Context, lagThreshold int64, clusterApiKey, clusterApiSecret string) error

Execute runs the full migration workflow from the current state

func (*MigrationOrchestrator) Initialize 

func (o *MigrationOrchestrator) Initialize(ctx context.Context, clusterApiKey, clusterApiSecret string) error

Initialize triggers the initialization event

func (*MigrationOrchestrator) PersistState added in v0.8.9 

func (o *MigrationOrchestrator) PersistState() error

PersistState saves the current migration config to the state file. It is the single writer for migration state: the orchestrator calls it after each successful FSM transition, and the offset-sync bookends (which run outside the FSM) are handed this method so they persist through the same path rather than duplicating the write.

type MigrationState added in v0.8.4 

type MigrationState struct {
	Migrations   []MigrationConfig  `json:"migrations"`
	KcpBuildInfo types.KcpBuildInfo `json:"kcp_build_info"`
	Timestamp    time.Time          `json:"timestamp"`
}

MigrationState represents the migration state file structure This is a dedicated state file for migration commands (init, execute, list)

func NewMigrationState added in v0.8.4 

func NewMigrationState() *MigrationState

NewMigrationState creates a new empty MigrationState with metadata

func NewMigrationStateFromFile added in v0.8.4 

func NewMigrationStateFromFile(filePath string) (*MigrationState, error)

NewMigrationStateFromFile loads a MigrationState from a JSON file

func (*MigrationState) GetMigrationById added in v0.8.4 

func (ms *MigrationState) GetMigrationById(migrationId string) (*MigrationConfig, error)

GetMigrationById retrieves a migration by its ID

func (*MigrationState) UpsertMigration added in v0.8.4 

func (ms *MigrationState) UpsertMigration(config MigrationConfig)

UpsertMigration adds a new migration or updates an existing one by ID

func (*MigrationState) WriteToFile added in v0.8.4 

func (ms *MigrationState) WriteToFile(filePath string) error

WriteToFile saves the MigrationState to a JSON file using atomic write

type WorkflowStep 

type WorkflowStep struct {
	Event       string
	Description string
	FromState   string
	ToState     string
}

WorkflowStep defines a single step in the migration workflow. It is pure FSM topology plus an ops-facing Description; user-facing presentation lives in stepHeaders, keyed by event, so the edge definitions stay presentation-free.

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.