refactor: consolidate Instance phase handlers
E2E Tests / Run on Ubuntu (pull_request) Failing after 1m28s
Tests / Run on Ubuntu (pull_request) Successful in 4m35s
Lint / Run on Ubuntu (pull_request) Successful in 4m58s

This commit is contained in:
2026-09-11 15:04:15 +00:00
parent 212ca4be99
commit 89d9e2316b
3 changed files with 164 additions and 103 deletions
@@ -18,7 +18,6 @@ package controller
import ( import (
"context" "context"
"errors"
"reflect" "reflect"
"time" "time"
@@ -66,17 +65,12 @@ func (r *PostgreSQLInstanceReconciler) Reconcile(ctx context.Context, req ctrl.R
} }
before := instance.DeepCopy() before := instance.DeepCopy()
phaseResult := newInstanceStateMachine().reconcile(instance) if r.Timeout > 0 {
instance.Status.Phase = phaseResult.phase var cancel context.CancelFunc
if phaseResult.reconcilingMessage != "" { ctx, cancel = context.WithTimeout(ctx, r.Timeout)
setReconcilingCondition(&instance.Status.Conditions, instance.Generation, phaseResult.reconcilingMessage) defer cancel()
}
var reconcileErr error
if r.Initializer != nil && instance.DeletionTimestamp.IsZero() &&
before.Status.Phase != "" && before.Status.Phase != databasev1alpha1.PostgreSQLInstancePhasePending {
reconcileErr = r.reconcileDependencies(ctx, instance)
} }
result, reconcileErr := newInstanceStateMachine(r.Initializer).reconcile(ctx, instance)
if !reflect.DeepEqual(before.Status, instance.Status) { if !reflect.DeepEqual(before.Status, instance.Status) {
if err := r.Status().Patch(ctx, instance, client.MergeFrom(before)); err != nil { if err := r.Status().Patch(ctx, instance, client.MergeFrom(before)); err != nil {
@@ -87,56 +81,7 @@ func (r *PostgreSQLInstanceReconciler) Reconcile(ctx context.Context, req ctrl.R
} }
} }
return ctrl.Result{}, reconcileErr return result, reconcileErr
}
func (r *PostgreSQLInstanceReconciler) reconcileDependencies(
ctx context.Context,
instance *databasev1alpha1.PostgreSQLInstance,
) error {
if r.Timeout > 0 {
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, r.Timeout)
defer cancel()
}
if instance.Status.Phase == databasev1alpha1.PostgreSQLInstancePhaseReady &&
instance.Status.ObservedGeneration != instance.Generation {
instance.Status.Phase = databasev1alpha1.PostgreSQLInstancePhaseValidating
setReconcilingCondition(&instance.Status.Conditions, instance.Generation, "instance dependencies are being validated")
return nil
}
var version string
var err error
switch instance.Status.Phase {
case databasev1alpha1.PostgreSQLInstancePhaseValidating:
version, err = r.Initializer.Validate(ctx, instance)
if err == nil {
instance.Status.PostgreSQLVersion = version
instance.Status.Phase = databasev1alpha1.PostgreSQLInstancePhaseInitializingRegistry
setReconcilingCondition(&instance.Status.Conditions, instance.Generation, "PostgreSQL registry is being initialized")
}
case databasev1alpha1.PostgreSQLInstancePhaseInitializingRegistry:
version, err = r.Initializer.InitializeRegistry(ctx, instance)
if err == nil {
instance.Status.PostgreSQLVersion = version
instance.Status.Phase = databasev1alpha1.PostgreSQLInstancePhaseReady
instance.Status.ObservedGeneration = instance.Generation
setReadyCondition(&instance.Status.Conditions, instance.Generation, "instance dependencies are ready")
}
}
if err != nil {
instance.Status.ObservedGeneration = instance.Generation
reason := databasev1alpha1.ReasonDependencyUnavailable
var categorized interface{ ConditionReason() string }
if errors.As(err, &categorized) {
reason = categorized.ConditionReason()
}
setFailedCondition(&instance.Status.Conditions, instance.Generation, reason,
"instance dependency validation failed")
}
return err
} }
// SetupWithManager sets up the controller with the Manager. // SetupWithManager sets up the controller with the Manager.
@@ -16,58 +16,115 @@ limitations under the License.
package controller package controller
import databasev1alpha1 "git.ddupan.top/panxiao81/postgresql-tenant-operator/api/v1alpha1" import (
"context"
"errors"
"time"
type instancePhaseResult struct { databasev1alpha1 "git.ddupan.top/panxiao81/postgresql-tenant-operator/api/v1alpha1"
phase databasev1alpha1.PostgreSQLInstancePhase ctrl "sigs.k8s.io/controller-runtime"
reconcilingMessage string )
}
type instancePhaseHandler func(*databasev1alpha1.PostgreSQLInstance) instancePhaseResult type instancePhaseHandler func(context.Context, *databasev1alpha1.PostgreSQLInstance) (ctrl.Result, error)
type instanceStateMachine struct { type instanceStateMachine struct {
handlers map[databasev1alpha1.PostgreSQLInstancePhase]instancePhaseHandler initializer PostgreSQLInstanceInitializer
handlers map[databasev1alpha1.PostgreSQLInstancePhase]instancePhaseHandler
} }
func newInstanceStateMachine() instanceStateMachine { func newInstanceStateMachine(initializer PostgreSQLInstanceInitializer) *instanceStateMachine {
return instanceStateMachine{handlers: map[databasev1alpha1.PostgreSQLInstancePhase]instancePhaseHandler{ m := &instanceStateMachine{initializer: initializer}
databasev1alpha1.PostgreSQLInstancePhasePending: reconcileInstancePending, m.handlers = map[databasev1alpha1.PostgreSQLInstancePhase]instancePhaseHandler{
databasev1alpha1.PostgreSQLInstancePhaseValidating: keepInstancePhase, databasev1alpha1.PostgreSQLInstancePhasePending: m.pending,
databasev1alpha1.PostgreSQLInstancePhaseInitializingRegistry: keepInstancePhase, databasev1alpha1.PostgreSQLInstancePhaseValidating: m.validate,
databasev1alpha1.PostgreSQLInstancePhaseReady: keepInstancePhase, databasev1alpha1.PostgreSQLInstancePhaseInitializingRegistry: m.initializeRegistry,
databasev1alpha1.PostgreSQLInstancePhaseDeleting: reconcileInstanceDeleting, databasev1alpha1.PostgreSQLInstancePhaseReady: m.ready,
}} databasev1alpha1.PostgreSQLInstancePhaseDeleting: m.deleting,
}
return m
} }
func (m instanceStateMachine) reconcile(instance *databasev1alpha1.PostgreSQLInstance) instancePhaseResult { func (m *instanceStateMachine) reconcile(ctx context.Context, instance *databasev1alpha1.PostgreSQLInstance) (ctrl.Result, error) {
phase := instance.Status.Phase phase := instance.Status.Phase
if !instance.DeletionTimestamp.IsZero() { if !instance.DeletionTimestamp.IsZero() {
phase = databasev1alpha1.PostgreSQLInstancePhaseDeleting phase = databasev1alpha1.PostgreSQLInstancePhaseDeleting
} else if phase == "" { } else if phase == "" {
phase = databasev1alpha1.PostgreSQLInstancePhasePending phase = databasev1alpha1.PostgreSQLInstancePhasePending
} }
handler, found := m.handlers[phase] handler, found := m.handlers[phase]
if !found { if !found {
return instancePhaseResult{phase: phase} handler = m.pending
} }
return handler(instance) result, err := handler(ctx, instance)
if err != nil {
instance.Status.ObservedGeneration = instance.Generation
reason := databasev1alpha1.ReasonDependencyUnavailable
var categorized interface{ ConditionReason() string }
if errors.As(err, &categorized) {
reason = categorized.ConditionReason()
}
setFailedCondition(&instance.Status.Conditions, instance.Generation, reason, "instance dependency validation failed")
}
return result, err
} }
func reconcileInstancePending(*databasev1alpha1.PostgreSQLInstance) instancePhaseResult { func (m *instanceStateMachine) pending(_ context.Context, instance *databasev1alpha1.PostgreSQLInstance) (ctrl.Result, error) {
return instancePhaseResult{ return advanceInstance(instance, databasev1alpha1.PostgreSQLInstancePhaseValidating,
phase: databasev1alpha1.PostgreSQLInstancePhaseValidating, "instance dependencies are being validated"), nil
reconcilingMessage: "instance dependencies are being validated",
}
} }
func reconcileInstanceDeleting(*databasev1alpha1.PostgreSQLInstance) instancePhaseResult { func (m *instanceStateMachine) validate(ctx context.Context, instance *databasev1alpha1.PostgreSQLInstance) (ctrl.Result, error) {
return instancePhaseResult{ if m.initializer == nil {
phase: databasev1alpha1.PostgreSQLInstancePhaseDeleting, return ctrl.Result{}, nil
reconcilingMessage: "instance deletion is reconciling",
} }
version, err := m.initializer.Validate(ctx, instance)
if err != nil {
return ctrl.Result{}, err
}
instance.Status.PostgreSQLVersion = version
return advanceInstance(instance, databasev1alpha1.PostgreSQLInstancePhaseInitializingRegistry,
"PostgreSQL registry is being initialized"), nil
} }
func keepInstancePhase(instance *databasev1alpha1.PostgreSQLInstance) instancePhaseResult { func (m *instanceStateMachine) initializeRegistry(ctx context.Context, instance *databasev1alpha1.PostgreSQLInstance) (ctrl.Result, error) {
return instancePhaseResult{phase: instance.Status.Phase} if m.initializer == nil {
return ctrl.Result{}, nil
}
version, err := m.initializer.InitializeRegistry(ctx, instance)
if err != nil {
return ctrl.Result{}, err
}
instance.Status.PostgreSQLVersion = version
instance.Status.Phase = databasev1alpha1.PostgreSQLInstancePhaseReady
instance.Status.ObservedGeneration = instance.Generation
setReadyCondition(&instance.Status.Conditions, instance.Generation, "instance dependencies are ready")
return ctrl.Result{RequeueAfter: time.Minute}, nil
}
func (m *instanceStateMachine) ready(ctx context.Context, instance *databasev1alpha1.PostgreSQLInstance) (ctrl.Result, error) {
if instance.Status.ObservedGeneration != instance.Generation {
return m.pending(ctx, instance)
}
if m.initializer == nil {
return ctrl.Result{}, nil
}
version, err := m.initializer.Validate(ctx, instance)
if err != nil {
instance.Status.Phase = databasev1alpha1.PostgreSQLInstancePhaseValidating
return ctrl.Result{}, err
}
instance.Status.PostgreSQLVersion = version
return ctrl.Result{RequeueAfter: time.Minute}, nil
}
func (m *instanceStateMachine) deleting(_ context.Context, instance *databasev1alpha1.PostgreSQLInstance) (ctrl.Result, error) {
instance.Status.Phase = databasev1alpha1.PostgreSQLInstancePhaseDeleting
setReconcilingCondition(&instance.Status.Conditions, instance.Generation, "instance deletion is reconciling")
return ctrl.Result{}, nil
}
func advanceInstance(instance *databasev1alpha1.PostgreSQLInstance, phase databasev1alpha1.PostgreSQLInstancePhase, message string) ctrl.Result {
instance.Status.Phase = phase
setReconcilingCondition(&instance.Status.Conditions, instance.Generation, message)
return ctrl.Result{RequeueAfter: time.Millisecond}
} }
+71 -12
View File
@@ -34,6 +34,8 @@ type fakeInstanceInitializer struct {
err error err error
} }
const testPostgreSQLVersion = "17.6"
func (f fakeInstanceInitializer) Validate(context.Context, *databasev1alpha1.PostgreSQLInstance) (string, error) { func (f fakeInstanceInitializer) Validate(context.Context, *databasev1alpha1.PostgreSQLInstance) (string, error) {
return f.validateVersion, f.err return f.validateVersion, f.err
} }
@@ -43,19 +45,74 @@ func (f fakeInstanceInitializer) InitializeRegistry(context.Context, *databasev1
} }
var _ = Describe("phase handler state machines", func() { var _ = Describe("phase handler state machines", func() {
It("persists validation intent before touching unavailable dependencies", func() {
instance := &databasev1alpha1.PostgreSQLInstance{}
machine := newInstanceStateMachine(fakeInstanceInitializer{err: errors.New("must not be called")})
result, err := machine.reconcile(context.Background(), instance)
Expect(err).NotTo(HaveOccurred())
Expect(result.RequeueAfter).To(BeNumerically(">", 0))
Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseValidating))
Expect(instance.Status.Conditions[0].Status).To(Equal(metav1.ConditionUnknown))
})
It("persists a new validation checkpoint when a Ready Instance generation changes", func() {
instance := &databasev1alpha1.PostgreSQLInstance{
ObjectMeta: metav1.ObjectMeta{Generation: 4},
Status: databasev1alpha1.PostgreSQLInstanceStatus{
Phase: databasev1alpha1.PostgreSQLInstancePhaseReady, ObservedGeneration: 3,
},
}
machine := newInstanceStateMachine(fakeInstanceInitializer{err: errors.New("must not be called")})
_, err := machine.reconcile(context.Background(), instance)
Expect(err).NotTo(HaveOccurred())
Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseValidating))
Expect(instance.Status.ObservedGeneration).To(Equal(int64(3)))
})
It("detects a dependency outage after Ready and recovers through registry initialization", func() {
instance := &databasev1alpha1.PostgreSQLInstance{
ObjectMeta: metav1.ObjectMeta{Generation: 3},
Status: databasev1alpha1.PostgreSQLInstanceStatus{
Phase: databasev1alpha1.PostgreSQLInstancePhaseReady, ObservedGeneration: 3,
},
}
machine := newInstanceStateMachine(fakeInstanceInitializer{err: errors.New("unavailable")})
_, err := machine.reconcile(context.Background(), instance)
Expect(err).To(MatchError("unavailable"))
Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseValidating))
Expect(instance.Status.Conditions[0].Status).To(Equal(metav1.ConditionFalse))
machine.initializer = fakeInstanceInitializer{
validateVersion: testPostgreSQLVersion,
registryVersion: testPostgreSQLVersion,
}
_, err = machine.reconcile(context.Background(), instance)
Expect(err).NotTo(HaveOccurred())
Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseInitializingRegistry))
_, err = machine.reconcile(context.Background(), instance)
Expect(err).NotTo(HaveOccurred())
Expect(instance.Status.Conditions[0].Status).To(Equal(metav1.ConditionTrue))
result, err := machine.reconcile(context.Background(), instance)
Expect(err).NotTo(HaveOccurred())
Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseReady))
Expect(result.RequeueAfter).To(Equal(time.Minute))
})
It("advances an Instance through external validation and registry initialization", func() { It("advances an Instance through external validation and registry initialization", func() {
instance := &databasev1alpha1.PostgreSQLInstance{ instance := &databasev1alpha1.PostgreSQLInstance{
ObjectMeta: metav1.ObjectMeta{Generation: 3}, ObjectMeta: metav1.ObjectMeta{Generation: 3},
Status: databasev1alpha1.PostgreSQLInstanceStatus{Phase: databasev1alpha1.PostgreSQLInstancePhaseValidating}, Status: databasev1alpha1.PostgreSQLInstanceStatus{Phase: databasev1alpha1.PostgreSQLInstancePhaseValidating},
} }
reconciler := &PostgreSQLInstanceReconciler{Initializer: fakeInstanceInitializer{ machine := newInstanceStateMachine(fakeInstanceInitializer{
validateVersion: "17.6", registryVersion: "17.6", validateVersion: testPostgreSQLVersion,
}} registryVersion: testPostgreSQLVersion,
Expect(reconciler.reconcileDependencies(context.Background(), instance)).To(Succeed()) })
_, err := machine.reconcile(context.Background(), instance)
Expect(err).NotTo(HaveOccurred())
Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseInitializingRegistry)) Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseInitializingRegistry))
Expect(reconciler.reconcileDependencies(context.Background(), instance)).To(Succeed()) _, err = machine.reconcile(context.Background(), instance)
Expect(err).NotTo(HaveOccurred())
Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseReady)) Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseReady))
Expect(instance.Status.PostgreSQLVersion).To(Equal("17.6")) Expect(instance.Status.PostgreSQLVersion).To(Equal(testPostgreSQLVersion))
Expect(instance.Status.ObservedGeneration).To(Equal(int64(3))) Expect(instance.Status.ObservedGeneration).To(Equal(int64(3)))
Expect(instance.Status.Conditions).To(ConsistOf(And( Expect(instance.Status.Conditions).To(ConsistOf(And(
HaveField("Status", metav1.ConditionTrue), HaveField("Reason", databasev1alpha1.ReasonReady), HaveField("Status", metav1.ConditionTrue), HaveField("Reason", databasev1alpha1.ReasonReady),
@@ -67,8 +124,9 @@ var _ = Describe("phase handler state machines", func() {
ObjectMeta: metav1.ObjectMeta{Generation: 2}, ObjectMeta: metav1.ObjectMeta{Generation: 2},
Status: databasev1alpha1.PostgreSQLInstanceStatus{Phase: databasev1alpha1.PostgreSQLInstancePhaseValidating}, Status: databasev1alpha1.PostgreSQLInstanceStatus{Phase: databasev1alpha1.PostgreSQLInstancePhaseValidating},
} }
reconciler := &PostgreSQLInstanceReconciler{Initializer: fakeInstanceInitializer{err: errors.New("unavailable")}} machine := newInstanceStateMachine(fakeInstanceInitializer{err: errors.New("unavailable")})
Expect(reconciler.reconcileDependencies(context.Background(), instance)).To(MatchError("unavailable")) _, err := machine.reconcile(context.Background(), instance)
Expect(err).To(MatchError("unavailable"))
Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseValidating)) Expect(instance.Status.Phase).To(Equal(databasev1alpha1.PostgreSQLInstancePhaseValidating))
Expect(instance.Status.Conditions).To(ConsistOf(And( Expect(instance.Status.Conditions).To(ConsistOf(And(
HaveField("Status", metav1.ConditionFalse), HaveField("Reason", databasev1alpha1.ReasonDependencyUnavailable), HaveField("Status", metav1.ConditionFalse), HaveField("Reason", databasev1alpha1.ReasonDependencyUnavailable),
@@ -77,12 +135,13 @@ var _ = Describe("phase handler state machines", func() {
DescribeTable("dispatches Instance phases", DescribeTable("dispatches Instance phases",
func(instance *databasev1alpha1.PostgreSQLInstance, expected databasev1alpha1.PostgreSQLInstancePhase, hasMessage bool) { func(instance *databasev1alpha1.PostgreSQLInstance, expected databasev1alpha1.PostgreSQLInstancePhase, hasMessage bool) {
result := newInstanceStateMachine().reconcile(instance) _, err := newInstanceStateMachine(nil).reconcile(context.Background(), instance)
Expect(result.phase).To(Equal(expected)) Expect(err).NotTo(HaveOccurred())
Expect(instance.Status.Phase).To(Equal(expected))
if hasMessage { if hasMessage {
Expect(result.reconcilingMessage).NotTo(BeEmpty()) Expect(instance.Status.Conditions).NotTo(BeEmpty())
} else { } else {
Expect(result.reconcilingMessage).To(BeEmpty()) Expect(instance.Status.Conditions).To(BeEmpty())
} }
}, },
Entry("starts validation from an empty checkpoint", Entry("starts validation from an empty checkpoint",