Documentation
¶
Index ¶
- func BuildClusterLinkConfig(config *types.MigrationConfig, apiKey, apiSecret string) clusterlink.Config
- func DisableOffsetSync(ctx context.Context, cl clusterlink.Service, clCfg clusterlink.Config, ...) error
- func RestoreOffsetSync(_ context.Context, cl clusterlink.Service, clCfg clusterlink.Config, ...)
- func WarnIfPausedOnExecuteFailure(config *types.MigrationConfig, execErr error)
- type ExecutionParams
- type MigrationOrchestrator
- type MigrationWorkflow
- func (s *MigrationWorkflow) CheckLags(ctx context.Context, config *types.MigrationConfig, lagThreshold int64, ...) error
- func (s *MigrationWorkflow) FenceGateway(ctx context.Context, config *types.MigrationConfig) error
- func (s *MigrationWorkflow) Initialize(ctx context.Context, config *types.MigrationConfig, ...) error
- func (s *MigrationWorkflow) PromoteTopics(ctx context.Context, config *types.MigrationConfig, ...) error
- func (s *MigrationWorkflow) SetRolloutTimeout(d time.Duration)
- func (s *MigrationWorkflow) SwitchGateway(ctx context.Context, config *types.MigrationConfig) error
- type WorkflowStep
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func BuildClusterLinkConfig ¶ added in v0.8.1
func BuildClusterLinkConfig(config *types.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 DisableOffsetSync ¶ added in v0.8.1
func DisableOffsetSync( ctx context.Context, cl clusterlink.Service, clCfg clusterlink.Config, config *types.MigrationConfig, persist func() error, ) error
DisableOffsetSync runs the pre-execute disable bookend for the --pause-consumer-offset-sync flow.
No-op cases (return nil without contacting the cluster link):
- intent flag not set
- migration is already at StateSwitched (re-run after success)
- PauseConsumerOffsetSyncFlipped is already true (resume after partial failure)
Active path: re-queries the cluster link for drift, calls AlterConfigs to set consumer.offset.sync.enable=false, sets PauseConsumerOffsetSyncFlipped, and persists the marker before returning.
A non-nil error stops execution before the FSM runs (R12).
func RestoreOffsetSync ¶ added in v0.8.1
func RestoreOffsetSync( _ context.Context, cl clusterlink.Service, clCfg clusterlink.Config, config *types.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 *types.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.
Soft-fail: never returns an error — this is best-effort messaging on top of the underlying execute error.
Types ¶
type ExecutionParams ¶
ExecutionParams holds runtime parameters needed during migration execution
type MigrationOrchestrator ¶
type MigrationOrchestrator struct {
// contains filtered or unexported fields
}
MigrationOrchestrator manages the FSM lifecycle and coordinates workflow execution
func NewMigrationOrchestrator ¶
func NewMigrationOrchestrator( config *types.MigrationConfig, workflow *MigrationWorkflow, migrationState *types.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
type MigrationWorkflow ¶
type MigrationWorkflow struct {
// contains filtered or unexported fields
}
func NewMigrationWorkflow ¶
func NewMigrationWorkflow( gatewayService gateway.Service, clusterLinkService clusterlink.Service, ) *MigrationWorkflow
func NewMigrationWorkflowWithOffsets ¶
func NewMigrationWorkflowWithOffsets( gatewayService gateway.Service, clusterLinkService clusterlink.Service, sourceOffset offset.Provider, destinationOffset offset.Provider, ) *MigrationWorkflow
func (*MigrationWorkflow) CheckLags ¶
func (s *MigrationWorkflow) CheckLags( ctx context.Context, config *types.MigrationConfig, lagThreshold int64, clusterApiKey, clusterApiSecret string, ) error
CheckLags polls source and destination offsets until lag is below threshold
func (*MigrationWorkflow) FenceGateway ¶
func (s *MigrationWorkflow) FenceGateway(ctx context.Context, config *types.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 (*MigrationWorkflow) Initialize ¶
func (s *MigrationWorkflow) Initialize( ctx context.Context, config *types.MigrationConfig, clusterApiKey, clusterApiSecret string, ) error
func (*MigrationWorkflow) PromoteTopics ¶
func (s *MigrationWorkflow) PromoteTopics(ctx context.Context, config *types.MigrationConfig, clusterApiKey, clusterApiSecret string) error
PromoteTopics polls offsets and promotes mirror topics that reach zero lag
func (*MigrationWorkflow) SetRolloutTimeout ¶ added in v0.8.1
func (s *MigrationWorkflow) SetRolloutTimeout(d time.Duration)
SetRolloutTimeout sets the deadline applied to gateway-readiness waits. A value of 0 means no deadline.
func (*MigrationWorkflow) SwitchGateway ¶
func (s *MigrationWorkflow) SwitchGateway(ctx context.Context, config *types.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.