From 5e6d97c295788f529c329cd44ba0658062051466 Mon Sep 17 00:00:00 2001 From: Artur Reznikov Date: Tue, 27 Oct 2026 14:15:56 -0700 Subject: [PATCH 1/3] feat(controller): make reconciles restart-safe. resume op Signed-off-by: Artur Reznikov --- .../solo-operator/templates/deployment.yaml | 3 +- charts/solo-operator/tests/shutdown_test.yaml | 23 ++ charts/solo-operator/values.yaml | 2 + cmd/main.go | 7 + config/manager/manager.yaml | 2 +- .../controller/networkgenesis_controller.go | 4 + .../networkupgradeexecute_controller.go | 155 +++++---- internal/controller/orbit_controller.go | 29 +- internal/controller/restart_safety_test.go | 317 ++++++++++++++++++ internal/resources/helm/client.go | 39 ++- internal/resources/helm/mock_client_test.go | 15 + internal/resources/helm/reasons.go | 1 + internal/resources/helm/release.go | 28 +- internal/resources/helm/release_test.go | 47 +++ internal/resources/helmtest/mock_client.go | 30 +- test/integration/restart/restart_test.go | 268 +++++++++++++++ 16 files changed, 868 insertions(+), 102 deletions(-) create mode 100644 charts/solo-operator/tests/shutdown_test.yaml create mode 100644 internal/controller/restart_safety_test.go create mode 100644 test/integration/restart/restart_test.go diff --git a/charts/solo-operator/templates/deployment.yaml b/charts/solo-operator/templates/deployment.yaml index 32fb626de..f32c0c346 100644 --- a/charts/solo-operator/templates/deployment.yaml +++ b/charts/solo-operator/templates/deployment.yaml @@ -29,7 +29,7 @@ spec: serviceAccountName: {{ include "solo-operator.serviceAccountName" . }} securityContext: {{- toYaml .Values.podSecurityContext | nindent 8 }} - terminationGracePeriodSeconds: 10 + terminationGracePeriodSeconds: {{ add .Values.controller.gracefulShutdownTimeoutSeconds 15 }} containers: - name: manager image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}" @@ -58,6 +58,7 @@ spec: {{- with .Values.controller.nodeActiveAlertInterval }} - --node-active-alert-interval={{ . }} {{- end }} + - --graceful-shutdown-timeout={{ .Values.controller.gracefulShutdownTimeoutSeconds }}s {{- $allowedNamespaces := include "solo-operator.allowedNamespaces" . }} {{- if ne $allowedNamespaces "_all_namespaces_placeholder" }} - --allowed-namespaces={{ $allowedNamespaces | replace " " "," }} diff --git a/charts/solo-operator/tests/shutdown_test.yaml b/charts/solo-operator/tests/shutdown_test.yaml new file mode 100644 index 000000000..0d71a56e3 --- /dev/null +++ b/charts/solo-operator/tests/shutdown_test.yaml @@ -0,0 +1,23 @@ +suite: deployment shutdown window +templates: + - templates/deployment.yaml +tests: + - it: defaults to a 30s drain inside a 45s grace period + asserts: + - equal: + path: spec.template.spec.terminationGracePeriodSeconds + value: 45 + - contains: + path: spec.template.spec.containers[0].args + content: --graceful-shutdown-timeout=30s + + - it: derives the grace period from the drain timeout + set: + controller.gracefulShutdownTimeoutSeconds: 60 + asserts: + - equal: + path: spec.template.spec.terminationGracePeriodSeconds + value: 75 + - contains: + path: spec.template.spec.containers[0].args + content: --graceful-shutdown-timeout=60s diff --git a/charts/solo-operator/values.yaml b/charts/solo-operator/values.yaml index d2a05d779..8f0e3cca3 100644 --- a/charts/solo-operator/values.yaml +++ b/charts/solo-operator/values.yaml @@ -152,3 +152,5 @@ controller: # initial deadline fires. Shorter values give the node operator more frequent # visibility. Empty uses the operator default (5m). nodeActiveAlertInterval: 5m + # Best-effort SIGTERM drain in seconds; the pod grace period is this plus 15s. + gracefulShutdownTimeoutSeconds: 30 diff --git a/cmd/main.go b/cmd/main.go index a968fd161..b19cb9326 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -71,6 +71,7 @@ func main() { var allowedNamespaces string var nodeActiveTimeout time.Duration var nodeActiveAlertInterval time.Duration + var gracefulShutdownTimeout time.Duration var showVersion bool var logLevel string var consoleLogging bool @@ -106,6 +107,9 @@ func main() { flag.DurationVar(&nodeActiveAlertInterval, "node-active-alert-interval", 5*time.Minute, "Interval between repeated NodeActiveDeadlineExceeded warning events after the initial "+ "deadline fires. Shorter values give the node operator more frequent visibility.") + flag.DurationVar(&gracefulShutdownTimeout, "graceful-shutdown-timeout", 30*time.Second, + "How long in-flight reconciles may run after SIGTERM before the manager stops. "+ + "Must be shorter than the pod's terminationGracePeriodSeconds.") flag.Parse() @@ -185,6 +189,9 @@ func main() { Metrics: metricsServerOptions, HealthProbeBindAddress: probeAddr, LeaderElection: enableLeaderElection, + // Safe because main exits as soon as mgr.Start returns. + LeaderElectionReleaseOnCancel: true, + GracefulShutdownTimeout: &gracefulShutdownTimeout, // LeaderElectionID matches the API group (operator.solo.hiero.org) so the lock // name is consistent with CRD group, finalizer, and label key naming conventions. LeaderElectionID: "operator.solo.hiero.org", diff --git a/config/manager/manager.yaml b/config/manager/manager.yaml index c68e7e1a1..d869245df 100644 --- a/config/manager/manager.yaml +++ b/config/manager/manager.yaml @@ -61,4 +61,4 @@ spec: - name: tmp emptyDir: {} serviceAccountName: controller-manager - terminationGracePeriodSeconds: 10 + terminationGracePeriodSeconds: 45 diff --git a/internal/controller/networkgenesis_controller.go b/internal/controller/networkgenesis_controller.go index 8bd7f2fcc..ac6266db4 100644 --- a/internal/controller/networkgenesis_controller.go +++ b/internal/controller/networkgenesis_controller.go @@ -9,6 +9,7 @@ import ( "github.com/automa-saga/errx" corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" + apimeta "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/client-go/tools/events" @@ -210,6 +211,7 @@ func (r *NetworkGenesisReconciler) succeed(ctx context.Context, ng *operatorv1al genesis.ClusterConfigMapName, } ng.Status.GeneratedConfigMaps = generatedCMs + apimeta.RemoveStatusCondition(&ng.Status.Conditions, networkGenesisConditionFailed) setNetworkGenesisCondition(&ng.Status.Conditions, metav1.Condition{ Type: networkGenesisConditionReady, Status: metav1.ConditionTrue, @@ -226,6 +228,7 @@ func (r *NetworkGenesisReconciler) fail(ctx context.Context, ng *operatorv1alpha ng.Status.Phase = operatorv1alpha1.PhaseFailed ng.Status.ObservedGeneration = ng.Generation ng.Status.CompletionTime = &now + apimeta.RemoveStatusCondition(&ng.Status.Conditions, networkGenesisConditionReady) setNetworkGenesisCondition(&ng.Status.Conditions, metav1.Condition{ Type: networkGenesisConditionFailed, Status: metav1.ConditionTrue, @@ -241,6 +244,7 @@ func (r *NetworkGenesisReconciler) rejectMutation(ctx context.Context, ng *opera now := metav1.Now() ng.Status.Phase = operatorv1alpha1.PhaseFailed ng.Status.ObservedGeneration = ng.Generation + apimeta.RemoveStatusCondition(&ng.Status.Conditions, networkGenesisConditionReady) setNetworkGenesisCondition(&ng.Status.Conditions, metav1.Condition{ Type: networkGenesisConditionFailed, Status: metav1.ConditionTrue, diff --git a/internal/controller/networkupgradeexecute_controller.go b/internal/controller/networkupgradeexecute_controller.go index 96f412a54..5bdcf9a19 100644 --- a/internal/controller/networkupgradeexecute_controller.go +++ b/internal/controller/networkupgradeexecute_controller.go @@ -14,6 +14,7 @@ import ( appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" + apimeta "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" @@ -208,8 +209,10 @@ func (r *NetworkUpgradeExecuteReconciler) Reconcile(ctx context.Context, req ctr return ctrl.Result{}, err } - now := metav1.Now() - execute.Status.StartTime = &now + if execute.Status.StartTime == nil { + now := metav1.Now() + execute.Status.StartTime = &now + } execute.Status.ObservedGeneration = execute.Generation // Both modes transition to ReadyForProvisionerDaemon; the difference is who @@ -403,72 +406,10 @@ func (r *NetworkUpgradeExecuteReconciler) reconcilePendingNodeUpgrade(ctx contex } } - prepare, err := findPrepare(ctx, r.Client, execute.Namespace, execute.Spec.OperationID) - if err != nil { - return ctrl.Result{}, err - } - - if _, prefetchOK := prefetchCompleted(prepare); prefetchOK { - // Prefetch succeeded — skip registry re-validation. The images are - // likely cached on nodes; the kubelet uses them via IfNotPresent. - common.Info(ctx, operatorv1alpha1.ReasonPrefetchCompleted).Msg("NetworkUpgradeExecute: images prefetched, skipping registry re-validation") - ss := execute.Spec.SelectedSources - execute.Status.SelectedSources = &ss - setCondition(&execute.Status.Conditions, metav1.Condition{ - Type: operatorv1alpha1.CondRegistrySelected, - Status: metav1.ConditionTrue, - Reason: operatorv1alpha1.ReasonPrefetchCompleted.String(), - Message: "images prefetched on nodes; registry re-validation skipped", - }) - } else { - rng := rand.New(rand.NewPCG(rand.Uint64(), rand.Uint64())) - checker, err := capsuleRegistryChecker(ctx, r.Client, r.Checker, execute.Namespace, execute.Spec.ConsensusCapsule, r.AllowedRegistryHosts) - if err != nil { - return ctrl.Result{}, err - } - skipUC := map[operatorv1alpha1.ComponentName]bool{operatorv1alpha1.ComponentUC: true} - updated, registryOK := registry.RevalidateComponents( - ctx, checker, - execute.Spec.Images, execute.Spec.SelectedSources, - checkTimeout, rng, r.StrategyName, - skipUC, - func(name operatorv1alpha1.ComponentName, res registry.FailoverResult) { - r.eventf(ctx, execute, corev1.EventTypeWarning, operatorv1alpha1.EventRegistryFailover, - "%s: registry %s unreachable, failed over to %s (digest: %s)", - name, res.FailedImage, res.Selected.Image, res.Selected.Digest) - }, - func(name operatorv1alpha1.ComponentName, err error) { - r.eventf(ctx, execute, corev1.EventTypeWarning, operatorv1alpha1.EventRegistryInvalid, - "%s: %v", name, err) - }, - ) - if !registryOK { - now := metav1.Now() - execute.Status.Phase = operatorv1alpha1.ExecutePhaseFailed - execute.Status.CompletionTime = &now - execute.Status.ObservedGeneration = execute.Generation - setCondition(&execute.Status.Conditions, metav1.Condition{ - Type: operatorv1alpha1.CondRegistrySelected, - Status: metav1.ConditionFalse, - Reason: operatorv1alpha1.ReasonRegistryInvalid.String(), - Message: "no valid registry available for one or more components", - }) - if err := r.Status().Update(ctx, execute); err != nil { - return ctrl.Result{}, wrapStatusUpdate(err, "updating status to failed") - } - r.eventf(ctx, execute, corev1.EventTypeWarning, operatorv1alpha1.EventUpgradeFailed, - "Registry validation failed for operationId %q", execute.Spec.OperationID) - return ctrl.Result{}, errx.WithHints( - errx.WithReason(errReconcile.New("no valid registry for one or more components"), ReasonNoValidRegistry), - "Confirm at least one configured registry serves a valid image for each component") + if !apimeta.IsStatusConditionTrue(execute.Status.Conditions, operatorv1alpha1.CondRegistrySelected) || execute.Status.SelectedSources == nil { + if res, err := r.selectSources(ctx, execute); res != nil { + return *res, err } - execute.Status.SelectedSources = updated - setCondition(&execute.Status.Conditions, metav1.Condition{ - Type: operatorv1alpha1.CondRegistrySelected, - Status: metav1.ConditionTrue, - Reason: operatorv1alpha1.ReasonSelectionComplete.String(), - Message: "all selected registries healthy", - }) } patch := client.MergeFrom(capsule.DeepCopy()) @@ -583,6 +524,86 @@ func (r *NetworkUpgradeExecuteReconciler) reconcilePendingNodeUpgrade(ctx contex } } +// selectSources persists the registry choice before the capsule patch so a restart reuses it. +func (r *NetworkUpgradeExecuteReconciler) selectSources(ctx context.Context, execute *operatorv1alpha1.NetworkUpgradeExecute) (*ctrl.Result, error) { + prepare, err := findPrepare(ctx, r.Client, execute.Namespace, execute.Spec.OperationID) + if err != nil { + return &ctrl.Result{}, err + } + + if _, prefetchOK := prefetchCompleted(prepare); prefetchOK { + // Prefetch succeeded — skip registry re-validation. The images are + // likely cached on nodes; the kubelet uses them via IfNotPresent. + common.Info(ctx, operatorv1alpha1.ReasonPrefetchCompleted).Msg("NetworkUpgradeExecute: images prefetched, skipping registry re-validation") + ss := execute.Spec.SelectedSources + execute.Status.SelectedSources = &ss + setCondition(&execute.Status.Conditions, metav1.Condition{ + Type: operatorv1alpha1.CondRegistrySelected, + Status: metav1.ConditionTrue, + Reason: operatorv1alpha1.ReasonPrefetchCompleted.String(), + Message: "images prefetched on nodes; registry re-validation skipped", + }) + } else { + rng := rand.New(rand.NewPCG(rand.Uint64(), rand.Uint64())) + checker, err := capsuleRegistryChecker(ctx, r.Client, r.Checker, execute.Namespace, execute.Spec.ConsensusCapsule, r.AllowedRegistryHosts) + if err != nil { + return &ctrl.Result{}, err + } + skipUC := map[operatorv1alpha1.ComponentName]bool{operatorv1alpha1.ComponentUC: true} + updated, registryOK := registry.RevalidateComponents( + ctx, checker, + execute.Spec.Images, execute.Spec.SelectedSources, + checkTimeout, rng, r.StrategyName, + skipUC, + func(name operatorv1alpha1.ComponentName, res registry.FailoverResult) { + r.eventf(ctx, execute, corev1.EventTypeWarning, operatorv1alpha1.EventRegistryFailover, + "%s: registry %s unreachable, failed over to %s (digest: %s)", + name, res.FailedImage, res.Selected.Image, res.Selected.Digest) + }, + func(name operatorv1alpha1.ComponentName, err error) { + r.eventf(ctx, execute, corev1.EventTypeWarning, operatorv1alpha1.EventRegistryInvalid, + "%s: %v", name, err) + }, + ) + if !registryOK { + now := metav1.Now() + execute.Status.Phase = operatorv1alpha1.ExecutePhaseFailed + execute.Status.CompletionTime = &now + execute.Status.ObservedGeneration = execute.Generation + setCondition(&execute.Status.Conditions, metav1.Condition{ + Type: operatorv1alpha1.CondRegistrySelected, + Status: metav1.ConditionFalse, + Reason: operatorv1alpha1.ReasonRegistryInvalid.String(), + Message: "no valid registry available for one or more components", + }) + if err := r.Status().Update(ctx, execute); err != nil { + return &ctrl.Result{}, wrapStatusUpdate(err, "updating status to failed") + } + r.eventf(ctx, execute, corev1.EventTypeWarning, operatorv1alpha1.EventUpgradeFailed, + "Registry validation failed for operationId %q", execute.Spec.OperationID) + return &ctrl.Result{}, errx.WithHints( + errx.WithReason(errReconcile.New("no valid registry for one or more components"), ReasonNoValidRegistry), + "Confirm at least one configured registry serves a valid image for each component") + } + execute.Status.SelectedSources = updated + setCondition(&execute.Status.Conditions, metav1.Condition{ + Type: operatorv1alpha1.CondRegistrySelected, + Status: metav1.ConditionTrue, + Reason: operatorv1alpha1.ReasonSelectionComplete.String(), + Message: "all selected registries healthy", + }) + } + + execute.Status.ObservedGeneration = execute.Generation + if err := r.Status().Update(ctx, execute); err != nil { + if apierrors.IsConflict(err) { + return &ctrl.Result{Requeue: true}, nil + } + return &ctrl.Result{}, wrapStatusUpdate(err, "persisting selected sources") + } + return nil, nil +} + // failUpgrade drives the Execute CR to a terminal Failed phase with a // NodesUpgraded=False condition and a UpgradeFailed warning event. The reconciler // is the sole writer of Execute.status.phase. diff --git a/internal/controller/orbit_controller.go b/internal/controller/orbit_controller.go index 6ae3c6ab8..e9b3ab01a 100644 --- a/internal/controller/orbit_controller.go +++ b/internal/controller/orbit_controller.go @@ -8,6 +8,7 @@ import ( "slices" "time" + "k8s.io/apimachinery/pkg/api/equality" apierrors "k8s.io/apimachinery/pkg/api/errors" apimeta "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -350,6 +351,7 @@ func (r *OrbitReconciler) deleteOwnedCapsules(ctx context.Context, orbitObj *ope } func (r *OrbitReconciler) updateStatus(ctx context.Context, orbitObj *operatorv1alpha1.Orbit) error { + before := orbitObj.Status.DeepCopy() now := metav1.Now() if orbitObj.Status.Phase == "" { @@ -364,26 +366,17 @@ func (r *OrbitReconciler) updateStatus(ctx context.Context, orbitObj *operatorv1 } } - ready := metav1.Condition{ - Type: "Ready", - Status: metav1.ConditionTrue, - LastTransitionTime: now, - Reason: ReasonReconcileSuccess.String(), - Message: "Orbit is ready", - } - found := false - for i, c := range orbitObj.Status.Conditions { - if c.Type == "Ready" { - orbitObj.Status.Conditions[i] = ready - found = true - break - } - } - if !found { - orbitObj.Status.Conditions = append(orbitObj.Status.Conditions, ready) - } + apimeta.SetStatusCondition(&orbitObj.Status.Conditions, metav1.Condition{ + Type: "Ready", + Status: metav1.ConditionTrue, + Reason: ReasonReconcileSuccess.String(), + Message: "Orbit is ready", + }) r.setNamespaceAllowed(ctx, orbitObj) orbitObj.Status.ObservedGeneration = orbitObj.Generation + if equality.Semantic.DeepEqual(before, &orbitObj.Status) { + return nil + } return common.WarnOnConflict(ctx, r.Status().Update(ctx, orbitObj)) } diff --git a/internal/controller/restart_safety_test.go b/internal/controller/restart_safety_test.go new file mode 100644 index 000000000..58006b46b --- /dev/null +++ b/internal/controller/restart_safety_test.go @@ -0,0 +1,317 @@ +// SPDX-License-Identifier: Apache-2.0 + +package controller + +import ( + "context" + "sync/atomic" + "testing" + "time" + + batchv1 "k8s.io/api/batch/v1" + corev1 "k8s.io/api/core/v1" + apimeta "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/tools/events" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + operatorv1alpha1 "github.com/hashgraph/solo-operator/api/v1alpha1" + "github.com/hashgraph/solo-operator/internal/crashtest" + "github.com/hashgraph/solo-operator/internal/operations/genesis" + "github.com/hashgraph/solo-operator/internal/resources/applicationproperties" + "github.com/hashgraph/solo-operator/internal/resources/log4j2config" + "github.com/hashgraph/solo-operator/internal/resources/nodesettings" + "github.com/hashgraph/solo-operator/internal/subreconciler" +) + +func newRestartExecuteReconciler(c client.Client, checker fakeUpgradeChecker) *NetworkUpgradeExecuteReconciler { + r := NewNetworkUpgradeExecuteReconciler(c, newControllerTestScheme()) + r.Recorder = events.NewFakeRecorder(64) + r.Checker = checker + return r +} + +func TestRestart_Execute_CrashAfterCapsulePatchKeepsSelection(t *testing.T) { + execute := newExecuteForTest() + execute.Status.Phase = operatorv1alpha1.ExecutePhasePendingNodeUpgrade + execute.Status.NodeUpgradeStartTime = &metav1.Time{Time: time.Now()} + execute.Spec.Images.ConsensusNode.Registries = registriesFor("reg-a/cn:0.75.0", "reg-b/cn:0.75.0") + + capsule := newOfflineCapsule("node0") + capsule.Spec.PodProperties.Containers.ConsensusNode.SoftwareVersion = &operatorv1alpha1.SoftwareVersion{Repository: "old-repo", ImageName: "cn", ImageTag: "0.74.0"} + + base := withExecuteIndexes(fake.NewClientBuilder(). + WithScheme(newControllerTestScheme()). + WithObjects(execute, capsule, newTestOrbit()). + WithStatusSubresource(execute)).Build() + ctx := context.Background() + if err := base.Status().Update(ctx, execute); err != nil { + t.Fatalf("seed status: %v", err) + } + req := ctrl.Request{NamespacedName: client.ObjectKeyFromObject(execute)} + + var dead atomic.Bool + dying := crashtest.Wrap(base, &dead, func(obj client.Object) bool { + _, isCapsule := obj.(*operatorv1alpha1.ConsensusCapsule) + return isCapsule + }) + first := newRestartExecuteReconciler(dying, newExecuteChecker(map[string]bool{"reg-a/cn:0.75.0": true}, nil)) + for i := 0; i < 5 && !dead.Load(); i++ { + _, _ = first.Reconcile(ctx, req) + } + if !dead.Load() { + t.Fatal("first process never reached the capsule patch") + } + + restarted := newRestartExecuteReconciler(base, newExecuteChecker(map[string]bool{"reg-b/cn:0.75.0": true}, nil)) + for i := 0; i < 5; i++ { + if _, err := restarted.Reconcile(ctx, req); err != nil { + t.Fatalf("reconcile after restart: %v", err) + } + } + + var got operatorv1alpha1.ConsensusCapsule + if err := base.Get(ctx, client.ObjectKeyFromObject(capsule), &got); err != nil { + t.Fatalf("get capsule: %v", err) + } + if repo := got.Spec.PodProperties.Containers.ConsensusNode.SoftwareVersion.Repository; repo != "reg-b" { + t.Errorf("capsule repository = %q after restart, want the pre-crash selection reg-b", repo) + } + var ex operatorv1alpha1.NetworkUpgradeExecute + if err := base.Get(ctx, req.NamespacedName, &ex); err != nil { + t.Fatalf("get execute: %v", err) + } + if ex.Status.Phase == operatorv1alpha1.ExecutePhaseFailed { + t.Fatal("execute failed after restart; registry selection was re-run") + } + if !apimeta.IsStatusConditionTrue(ex.Status.Conditions, operatorv1alpha1.CondConsensusCapsulePatched) { + t.Errorf("expected ConsensusCapsulePatched=True after restart, got %v", ex.Status.Conditions) + } +} + +func TestRestart_Execute_InitKeepsStartTime(t *testing.T) { + execute := newExecuteForTest() + execute.Status.Phase = "" + original := metav1.NewTime(time.Now().Add(-time.Hour).Truncate(time.Second)) + execute.Status.StartTime = &original + + pre := prereqs() + objs := append([]client.Object{execute, newOfflineCapsule("node0"), newTestOrbit()}, pre...) + base := withExecuteIndexes(fake.NewClientBuilder(). + WithScheme(newControllerTestScheme()). + WithObjects(objs...). + WithStatusSubresource(execute, &operatorv1alpha1.NetworkUpgradePrepare{}, &operatorv1alpha1.NetworkUpgradeFreeze{})).Build() + ctx := context.Background() + for _, o := range append([]client.Object{execute}, pre...) { + if err := base.Status().Update(ctx, o); err != nil { + t.Fatalf("seed status: %v", err) + } + } + + r := newRestartExecuteReconciler(base, newExecuteChecker(nil, nil)) + if _, err := r.Reconcile(ctx, ctrl.Request{NamespacedName: client.ObjectKeyFromObject(execute)}); err != nil { + t.Fatalf("reconcile: %v", err) + } + var ex operatorv1alpha1.NetworkUpgradeExecute + if err := base.Get(ctx, client.ObjectKeyFromObject(execute), &ex); err != nil { + t.Fatalf("get execute: %v", err) + } + if ex.Status.Phase != operatorv1alpha1.ExecutePhaseReadyForProvisionerDaemon { + t.Fatalf("phase = %q, want ReadyForProvisionerDaemon", ex.Status.Phase) + } + if ex.Status.StartTime == nil || !ex.Status.StartTime.Equal(&original) { + t.Errorf("startTime = %v, want preserved %v", ex.Status.StartTime, original) + } +} + +func TestRestart_Genesis_CrashMidConfigMapsConverges(t *testing.T) { + scheme := ngTestScheme(t) + ng := &operatorv1alpha1.NetworkGenesis{ + ObjectMeta: metav1.ObjectMeta{Name: "genesis-1", Namespace: "test", Finalizers: []string{NetworkGenesisFinalizer}}, + Spec: operatorv1alpha1.NetworkGenesisSpec{Orbit: "test-orbit", OperationID: "genesis-2026-05-20"}, + } + objs := []client.Object{newTestOrbit(), ngCapsule("node0", 0), ng} + objs = append(objs, ngCapsuleSecrets("node0", ngSelfSignedCertPEM(t), ngSelfSignedCertPEM(t))...) + base := fake.NewClientBuilder().WithScheme(scheme). + WithStatusSubresource(&operatorv1alpha1.NetworkGenesis{}). + WithObjects(objs...).Build() + ctx := context.Background() + name := types.NamespacedName{Namespace: "test", Name: "genesis-1"} + + cmWrites := 0 + var dead atomic.Bool + dying := crashtest.Wrap(base, &dead, func(obj client.Object) bool { + if _, ok := obj.(*corev1.ConfigMap); ok { + cmWrites++ + } + return cmWrites == 2 + }) + if _, err := NewNetworkGenesisReconciler(dying, scheme).Reconcile(ctx, ctrl.Request{NamespacedName: name}); err == nil { + t.Fatal("expected the crashed reconcile to error") + } + + var got operatorv1alpha1.NetworkGenesis + if err := base.Get(ctx, name, &got); err != nil { + t.Fatalf("get: %v", err) + } + got.Status.Phase = operatorv1alpha1.PhaseFailed + setNetworkGenesisCondition(&got.Status.Conditions, metav1.Condition{ + Type: networkGenesisConditionFailed, Status: metav1.ConditionTrue, Reason: ReasonGenesisFailed.String(), LastTransitionTime: metav1.Now(), + }) + if err := base.Status().Update(ctx, &got); err != nil { + t.Fatalf("seed failed status: %v", err) + } + + reconcileToCompletion(t, NewNetworkGenesisReconciler(base, scheme), name) + + if err := base.Get(ctx, name, &got); err != nil { + t.Fatalf("get: %v", err) + } + if got.Status.Phase != operatorv1alpha1.PhaseSucceeded { + t.Fatalf("phase = %s, want Succeeded", got.Status.Phase) + } + if apimeta.FindStatusCondition(got.Status.Conditions, networkGenesisConditionFailed) != nil { + t.Error("stale Failed condition survived a successful run") + } + for _, cm := range []string{genesis.GenesisConfigMapName, genesis.AddressBookBinConfigMapName, genesis.PublicKeysConfigMapName, genesis.ClusterConfigMapName} { + if err := base.Get(ctx, types.NamespacedName{Namespace: "test", Name: cm}, &corev1.ConfigMap{}); err != nil { + t.Errorf("ConfigMap %s missing after restart: %v", cm, err) + } + } +} + +func TestRestart_Prepare_CrashMidPrefetchJobCreation(t *testing.T) { + scheme := newControllerTestScheme() + prepare := newPrepareForTest() + prepare.Finalizers = []string{FinalizerNetworkUpgradePrepare} + prepare.Status.Phase = operatorv1alpha1.PhaseRunning + prepare.Status.SelectedSources = &operatorv1alpha1.SelectedSources{} + prepare.Status.SelectedSources.Set(operatorv1alpha1.ComponentConsensusNode, + &operatorv1alpha1.SelectedSource{Image: "gcr.io/cn:v1", Digest: "sha256:abc"}) + prepare.Status.Conditions = []operatorv1alpha1.Condition{{ + Type: operatorv1alpha1.CondRegistrySelected, Status: metav1.ConditionTrue, Reason: operatorv1alpha1.ReasonSelectionComplete.String(), + }} + cap0 := capsuleForPrefetch("cap-0", "test", "test-orbit", 0, nil, "sa-cn") + cap1 := capsuleForPrefetch("cap-1", "test", "test-orbit", 1, nil, "sa-cn") + + base := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(prepare, newTestOrbit(), &cap0, &cap1, consensusPodOn(&cap0, "node-a"), consensusPodOn(&cap1, "node-b")). + WithStatusSubresource(prepare). + WithIndex(&operatorv1alpha1.NetworkUpgradePrepare{}, fieldOperationID, prepareIndexFunc). + WithIndex(&operatorv1alpha1.ConsensusCapsule{}, fieldCapsuleOrbit, capsuleOrbitIndexFunc). + Build() + ctx := context.Background() + req := ctrl.Request{NamespacedName: client.ObjectKeyFromObject(prepare)} + + var dead atomic.Bool + dying := crashtest.Wrap(base, &dead, func(obj client.Object) bool { + _, isJob := obj.(*batchv1.Job) + return isJob + }) + first := NewNetworkUpgradePrepareReconciler(dying, scheme) + first.Recorder = events.NewFakeRecorder(64) + if _, err := first.Reconcile(ctx, req); err == nil || !dead.Load() { + t.Fatalf("expected crash after the first Job create, err=%v", err) + } + + restarted := NewNetworkUpgradePrepareReconciler(base, scheme) + restarted.Recorder = events.NewFakeRecorder(64) + if _, err := restarted.Reconcile(ctx, req); err != nil { + t.Fatalf("reconcile after restart: %v", err) + } + + var jobs batchv1.JobList + if err := base.List(ctx, &jobs, client.InNamespace("test")); err != nil { + t.Fatalf("list jobs: %v", err) + } + if len(jobs.Items) != 2 { + t.Errorf("got %d prefetch Jobs after restart, want 2 (one per node, no duplicates)", len(jobs.Items)) + } + var got operatorv1alpha1.NetworkUpgradePrepare + if err := base.Get(ctx, req.NamespacedName, &got); err != nil { + t.Fatalf("get prepare: %v", err) + } + if !apimeta.IsStatusConditionTrue(got.Status.Conditions, operatorv1alpha1.CondImagePrefetchCreated) { + t.Errorf("expected ImagePrefetchCreated=True after restart, got %v", got.Status.Conditions) + } +} + +type svProbe struct { + subreconciler.BaseSubReconciler[*operatorv1alpha1.ConsensusCapsule] + seen *operatorv1alpha1.SoftwareVersion +} + +func (p *svProbe) Reconcile(_ context.Context, c *operatorv1alpha1.ConsensusCapsule) (ctrl.Result, error) { + p.seen = c.Spec.PodProperties.Containers.ConsensusNode.SoftwareVersion.DeepCopy() + return ctrl.Result{RequeueAfter: time.Minute}, nil +} + +func TestRestart_ConsensusCapsule_PinsReachChildBuild(t *testing.T) { + capsule := capsuleWithConsensusSource() + capsule.Finalizers = []string{ConsensusCapsuleFinalizer} + capsule.Spec.Orbit = "test-orbit" + capsule.Spec.Log4j2ConfigRef = "log4j2" + capsule.Spec.SettingsConfigRef = "settings" + capsule.Spec.ApplicationPropertiesRef = "app" + capsule.Spec.PodProperties.Containers.UC = &operatorv1alpha1.UcSidecar{ + Enabled: true, + SoftwareVersion: &operatorv1alpha1.SoftwareVersion{Repository: "ghcr.io/hashgraph", ImageName: "uc", ImageTag: "1.0.0"}, + } + cm := func(name string) *corev1.ConfigMap { + return &corev1.ConfigMap{ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: "test"}} + } + + scheme := newControllerTestScheme() + c := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(capsule, newTestOrbit(), + cm(log4j2config.ConfigMapNameForRef("log4j2")), + cm(nodesettings.ConfigMapNameForRef("settings")), + cm(applicationproperties.ConfigMapNameForRef("app"))). + WithStatusSubresource(capsule).Build() + + probe := &svProbe{BaseSubReconciler: *subreconciler.NewBaseSubReconciler[*operatorv1alpha1.ConsensusCapsule](c, scheme, "probe")} + r := &ConsensusCapsuleReconciler{Client: c, Scheme: scheme, Checker: &fakeImageChecker{}, + subreconciler: subreconciler.NewChain[*operatorv1alpha1.ConsensusCapsule](probe)} + + if _, err := r.Reconcile(context.Background(), ctrl.Request{NamespacedName: client.ObjectKeyFromObject(capsule)}); err != nil { + t.Fatalf("reconcile: %v", err) + } + if probe.seen == nil || probe.seen.Digest != testResolvedDigest { + t.Fatalf("child build saw SoftwareVersion %+v, want digest %s", probe.seen, testResolvedDigest) + } +} + +func TestOrbit_RepeatReconcileDoesNotRewriteStatus(t *testing.T) { + orbit := newTestOrbit() + orbit.Finalizers = []string{OrbitFinalizer} + c := fake.NewClientBuilder().WithScheme(newControllerTestScheme()). + WithObjects(orbit).WithStatusSubresource(orbit).Build() + r := NewOrbitReconciler(c, newControllerTestScheme()) + ctx := context.Background() + req := ctrl.Request{NamespacedName: client.ObjectKeyFromObject(orbit)} + + if _, err := r.Reconcile(ctx, req); err != nil { + t.Fatalf("first reconcile: %v", err) + } + var first operatorv1alpha1.Orbit + if err := c.Get(ctx, req.NamespacedName, &first); err != nil { + t.Fatalf("get: %v", err) + } + if apimeta.FindStatusCondition(first.Status.Conditions, "Ready") == nil { + t.Fatal("expected Ready condition after first reconcile") + } + + if _, err := r.Reconcile(ctx, req); err != nil { + t.Fatalf("second reconcile: %v", err) + } + var second operatorv1alpha1.Orbit + if err := c.Get(ctx, req.NamespacedName, &second); err != nil { + t.Fatalf("get: %v", err) + } + if second.ResourceVersion != first.ResourceVersion { + t.Errorf("unchanged Orbit status was rewritten: resourceVersion %s -> %s", first.ResourceVersion, second.ResourceVersion) + } +} diff --git a/internal/resources/helm/client.go b/internal/resources/helm/client.go index dfc3b1101..012facd2d 100644 --- a/internal/resources/helm/client.go +++ b/internal/resources/helm/client.go @@ -37,11 +37,16 @@ const ( ) type ReleaseInfo struct { - Name string - Namespace string - Version int - Status string - ChartString string // "chartname-version" + Name string + Namespace string + Version int + Status string + ChartString string // "chartname-version" + LastDeployed time.Time +} + +func (i *ReleaseInfo) Pending() bool { + return helmrelease.Status(i.Status).IsPending() } // ChartCredentials is read from the Secret named in HelmChartSpec.CredentialsRef. @@ -75,6 +80,7 @@ type Client interface { InstallOrUpgrade(ctx context.Context, releaseName, namespace string, chart operatorv1alpha1.HelmChartSpec, values map[string]interface{}, opts ReleaseOptions) (*ReleaseInfo, error) Uninstall(ctx context.Context, releaseName, namespace string, opts UninstallOptions) error GetRelease(releaseName, namespace string) (*ReleaseInfo, error) + Unlock(releaseName, namespace string) error } // ─── Real implementation ────────────────────────────────────────────────────── @@ -277,6 +283,28 @@ func (h *helmSDKClient) GetRelease(releaseName, namespace string) (*ReleaseInfo, return releaseInfoFrom(rel), nil } +func (h *helmSDKClient) Unlock(releaseName, namespace string) error { + cfg, err := h.actionConfig(namespace) + if err != nil { + return err + } + rel, err := cfg.Releases.Last(releaseName) + if err != nil { + if isNotFound(err) { + return nil + } + return errx.WithReason(errHelm.Wrap(err, "helm get last revision of %s", releaseName), ReasonHelm) + } + if rel.Info == nil || !rel.Info.Status.IsPending() { + return nil + } + rel.SetStatus(helmrelease.StatusFailed, "operation interrupted; unlocked by solo-operator") + if err := cfg.Releases.Update(rel); err != nil { + return errx.WithReason(errHelm.Wrap(err, "helm unlock %s", releaseName), ReasonHelm) + } + return nil +} + func releaseInfoFrom(rel *helmrelease.Release) *ReleaseInfo { if rel == nil { return nil @@ -288,6 +316,7 @@ func releaseInfoFrom(rel *helmrelease.Release) *ReleaseInfo { } if rel.Info != nil { info.Status = rel.Info.Status.String() + info.LastDeployed = rel.Info.LastDeployed.Time } if rel.Chart != nil && rel.Chart.Metadata != nil { info.ChartString = rel.Chart.Metadata.Name + "-" + rel.Chart.Metadata.Version diff --git a/internal/resources/helm/mock_client_test.go b/internal/resources/helm/mock_client_test.go index 46fd37e0c..00c33a741 100644 --- a/internal/resources/helm/mock_client_test.go +++ b/internal/resources/helm/mock_client_test.go @@ -4,6 +4,7 @@ package helm import ( "context" + "errors" operatorv1alpha1 "github.com/hashgraph/solo-operator/api/v1alpha1" ) @@ -18,6 +19,17 @@ type MockClient struct { GetReleaseFunc func(releaseName, namespace string) (*ReleaseInfo, error) } +func (m *MockClient) SetRelease(info *ReleaseInfo) { + m.releases[info.Namespace+"/"+info.Name] = info +} + +func (m *MockClient) Unlock(releaseName, namespace string) error { + if info, ok := m.releases[namespace+"/"+releaseName]; ok && info.Pending() { + info.Status = "failed" + } + return nil +} + func NewMockClient() *MockClient { return &MockClient{releases: make(map[string]*ReleaseInfo)} } @@ -28,6 +40,9 @@ func (m *MockClient) InstallOrUpgrade(ctx context.Context, releaseName, namespac } key := namespace + "/" + releaseName existing := m.releases[key] + if existing != nil && existing.Pending() { + return nil, errors.New("another operation (install/upgrade/rollback) is in progress") + } version := 1 if existing != nil { version = existing.Version + 1 diff --git a/internal/resources/helm/reasons.go b/internal/resources/helm/reasons.go index e4fc73182..9b33e03d4 100644 --- a/internal/resources/helm/reasons.go +++ b/internal/resources/helm/reasons.go @@ -15,6 +15,7 @@ const ( ReasonReleaseReconciling errx.Reason = "ReleaseReconciling" ReasonReleaseReconciled errx.Reason = "ReleaseReconciled" ReasonReleasePending errx.Reason = "ReleasePending" + ReasonReleaseUnlocked errx.Reason = "ReleaseUnlocked" ReasonAddressBookReadFailed errx.Reason = "AddressBookReadFailed" ReasonCapsuleSkipped errx.Reason = "CapsuleSkipped" diff --git a/internal/resources/helm/release.go b/internal/resources/helm/release.go index 85ad24148..dd6079a20 100644 --- a/internal/resources/helm/release.go +++ b/internal/resources/helm/release.go @@ -65,21 +65,19 @@ func (r *ReleaseSubReconciler) Reconcile(ctx context.Context, hc *operatorv1alph if err != nil { return ctrl.Result{}, err } - if hash == hc.Status.ValuesHash { - existing, err := r.helmClient.GetRelease(releaseName, hc.Namespace) - if err != nil { - return ctrl.Result{}, fmt.Errorf("checking existing release: %w", err) - } - if existing != nil && existing.Status == "deployed" { + existing, err := r.helmClient.GetRelease(releaseName, hc.Namespace) + if err != nil { + return ctrl.Result{}, fmt.Errorf("checking existing release: %w", err) + } + if hash == hc.Status.ValuesHash && existing != nil { + if existing.Status == "deployed" { if err := r.syncOutputs(ctx, hc, componentOutputs(hc, releaseName, values)); err != nil { return ctrl.Result{}, err } common.DebugL(&rl, ReasonReleaseUpToDate).Msg("Release up-to-date, skipping upgrade") return ctrl.Result{RequeueAfter: releaseCheckInterval}, nil } - if existing != nil { - common.InfoL(&rl, ReasonReleaseRetry).Str("helmStatus", existing.Status).Msg("Release not in deployed state, retrying upgrade") - } + common.InfoL(&rl, ReasonReleaseRetry).Str("helmStatus", existing.Status).Msg("Release not in deployed state, retrying upgrade") } opts := releaseOptions(hc) @@ -87,6 +85,18 @@ func (r *ReleaseSubReconciler) Reconcile(ctx context.Context, hc *operatorv1alph opts.Timeout = DefaultTimeout } + if existing != nil && existing.Pending() { + if wait := opts.Timeout - time.Since(existing.LastDeployed); wait > 0 { + common.InfoL(&rl, ReasonReleasePending).Str("helmStatus", existing.Status).Msg("Helm release has an operation in progress, waiting") + return ctrl.Result{RequeueAfter: wait}, nil + } + common.WarnL(&rl, ReasonReleaseUnlocked).Str("helmStatus", existing.Status).Int("revision", existing.Version). + Msg("Helm release stuck pending past its timeout, marking failed to unlock") + if err := r.helmClient.Unlock(releaseName, hc.Namespace); err != nil { + return ctrl.Result{}, err + } + } + if hc.Spec.Chart.CredentialsRef != nil { creds, err := r.loadChartCredentials(ctx, hc.Namespace, hc.Spec.Chart.CredentialsRef.Name) if err != nil { diff --git a/internal/resources/helm/release_test.go b/internal/resources/helm/release_test.go index 6ddeb50d8..452d9f13a 100644 --- a/internal/resources/helm/release_test.go +++ b/internal/resources/helm/release_test.go @@ -554,6 +554,53 @@ func TestReleaseSubReconciler_retriesWhenHashMatchesButNotDeployed(t *testing.T) } } +func TestReleaseSubReconciler_unlocksStalePendingRelease(t *testing.T) { + for _, status := range []string{"pending-install", "pending-upgrade", "pending-rollback"} { + t.Run(status, func(t *testing.T) { + mock := NewMockClient() + hc := newHelmCapsule("mirror", "test-ns") + r := newTestReleaseSubReconciler(newTestScheme(), mock, hc) + mock.SetRelease(&ReleaseInfo{ + Name: "mirror", Namespace: "test-ns", Version: 3, Status: status, + LastDeployed: time.Now().Add(-2 * DefaultTimeout), + }) + + if _, err := r.Reconcile(context.Background(), hc); err != nil { + t.Fatalf("reconcile: %v", err) + } + info, _ := mock.GetRelease("mirror", "test-ns") + if info.Status != "deployed" || info.Version != 4 { + t.Errorf("got status=%s revision=%d, want deployed revision 4", info.Status, info.Version) + } + if hc.Status.ReleaseStatus != "deployed" { + t.Errorf("hc.Status.ReleaseStatus = %q, want deployed", hc.Status.ReleaseStatus) + } + }) + } +} + +func TestReleaseSubReconciler_waitsOutFreshPendingRelease(t *testing.T) { + mock := NewMockClient() + hc := newHelmCapsule("mirror", "test-ns") + r := newTestReleaseSubReconciler(newTestScheme(), mock, hc) + mock.SetRelease(&ReleaseInfo{ + Name: "mirror", Namespace: "test-ns", Version: 3, Status: "pending-upgrade", + LastDeployed: time.Now().Add(-time.Minute), + }) + + result, err := r.Reconcile(context.Background(), hc) + if err != nil { + t.Fatalf("reconcile: %v", err) + } + if result.RequeueAfter <= 0 || result.RequeueAfter > DefaultTimeout-time.Minute { + t.Errorf("RequeueAfter = %v, want the remaining timeout", result.RequeueAfter) + } + info, _ := mock.GetRelease("mirror", "test-ns") + if info.Status != "pending-upgrade" || info.Version != 3 { + t.Errorf("fresh pending release was touched: status=%s revision=%d", info.Status, info.Version) + } +} + // ─── Chart credentials ──────────────────────────────────────────────────────── func TestReleaseSubReconciler_loadsCredentialsFromSecret(t *testing.T) { diff --git a/internal/resources/helmtest/mock_client.go b/internal/resources/helmtest/mock_client.go index d73202b42..67305e827 100644 --- a/internal/resources/helmtest/mock_client.go +++ b/internal/resources/helmtest/mock_client.go @@ -5,6 +5,8 @@ package helmtest import ( "context" + "errors" + "sync" operatorv1alpha1 "github.com/hashgraph/solo-operator/api/v1alpha1" "github.com/hashgraph/solo-operator/internal/resources/helm" @@ -12,6 +14,7 @@ import ( // MockClient is an in-memory helm.Client for tests. type MockClient struct { + mu sync.Mutex releases map[string]*helm.ReleaseInfo InstallOrUpgradeFunc func(ctx context.Context, releaseName, namespace string, chart operatorv1alpha1.HelmChartSpec, values map[string]interface{}, opts helm.ReleaseOptions) (*helm.ReleaseInfo, error) @@ -19,6 +22,21 @@ type MockClient struct { GetReleaseFunc func(releaseName, namespace string) (*helm.ReleaseInfo, error) } +func (m *MockClient) SetRelease(info *helm.ReleaseInfo) { + m.mu.Lock() + defer m.mu.Unlock() + m.releases[info.Namespace+"/"+info.Name] = info +} + +func (m *MockClient) Unlock(releaseName, namespace string) error { + m.mu.Lock() + defer m.mu.Unlock() + if info, ok := m.releases[namespace+"/"+releaseName]; ok && info.Pending() { + info.Status = "failed" + } + return nil +} + func NewMockClient() *MockClient { return &MockClient{releases: make(map[string]*helm.ReleaseInfo)} } @@ -27,8 +45,13 @@ func (m *MockClient) InstallOrUpgrade(ctx context.Context, releaseName, namespac if m.InstallOrUpgradeFunc != nil { return m.InstallOrUpgradeFunc(ctx, releaseName, namespace, chart, values, opts) } + m.mu.Lock() + defer m.mu.Unlock() key := namespace + "/" + releaseName existing := m.releases[key] + if existing != nil && existing.Pending() { + return nil, errors.New("another operation (install/upgrade/rollback) is in progress") + } version := 1 if existing != nil { version = existing.Version + 1 @@ -48,6 +71,8 @@ func (m *MockClient) Uninstall(ctx context.Context, releaseName, namespace strin if m.UninstallFunc != nil { return m.UninstallFunc(ctx, releaseName, namespace, opts) } + m.mu.Lock() + defer m.mu.Unlock() delete(m.releases, namespace+"/"+releaseName) return nil } @@ -56,9 +81,12 @@ func (m *MockClient) GetRelease(releaseName, namespace string) (*helm.ReleaseInf if m.GetReleaseFunc != nil { return m.GetReleaseFunc(releaseName, namespace) } + m.mu.Lock() + defer m.mu.Unlock() info, ok := m.releases[namespace+"/"+releaseName] if !ok { return nil, nil } - return info, nil + cp := *info + return &cp, nil } diff --git a/test/integration/restart/restart_test.go b/test/integration/restart/restart_test.go new file mode 100644 index 000000000..660071ea6 --- /dev/null +++ b/test/integration/restart/restart_test.go @@ -0,0 +1,268 @@ +// SPDX-License-Identifier: Apache-2.0 + +package restart + +import ( + "context" + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + + corev1 "k8s.io/api/core/v1" + apimeta "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/kubernetes/scheme" + "k8s.io/client-go/rest" + "k8s.io/utils/ptr" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/config" + "sigs.k8s.io/controller-runtime/pkg/envtest" + logf "sigs.k8s.io/controller-runtime/pkg/log" + "sigs.k8s.io/controller-runtime/pkg/log/zap" + metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server" + + operatorv1alpha1 "github.com/hashgraph/solo-operator/api/v1alpha1" + "github.com/hashgraph/solo-operator/internal/common" + "github.com/hashgraph/solo-operator/internal/controller" + "github.com/hashgraph/solo-operator/internal/crashtest" + "github.com/hashgraph/solo-operator/internal/registry" + "github.com/hashgraph/solo-operator/internal/resources/helm" + "github.com/hashgraph/solo-operator/internal/resources/helmtest" +) + +const ( + timeout = 30 * time.Second + interval = 250 * time.Millisecond + digest = operatorv1alpha1.Sha256Digest("sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc") +) + +var ( + cfg *rest.Config + k8sClient client.Client + testEnv *envtest.Environment +) + +func TestRestart(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Restart Safety Suite") +} + +var _ = BeforeSuite(func() { + logf.SetLogger(zap.New(zap.WriteTo(GinkgoWriter), zap.UseDevMode(true))) + testEnv = &envtest.Environment{ + CRDDirectoryPaths: []string{filepath.Join("..", "..", "..", "config", "crd", "bases")}, + ErrorIfCRDPathMissing: true, + } + var err error + cfg, err = testEnv.Start() + Expect(err).NotTo(HaveOccurred()) + Expect(operatorv1alpha1.AddToScheme(scheme.Scheme)).To(Succeed()) + k8sClient, err = client.New(cfg, client.Options{Scheme: scheme.Scheme}) + Expect(err).NotTo(HaveOccurred()) +}) + +var _ = AfterSuite(func() { + Expect(testEnv.Stop()).To(Succeed()) +}) + +type operator struct { + dead atomic.Bool + cancel context.CancelFunc + done chan struct{} +} + +func (o *operator) stop() { + o.cancel() + <-o.done +} + +func startOperator(setup func(ctrl.Manager), dieAfter func(client.Object) bool) *operator { + op := &operator{done: make(chan struct{})} + opts := ctrl.Options{ + Scheme: scheme.Scheme, + Metrics: metricsserver.Options{BindAddress: "0"}, + Controller: config.Controller{SkipNameValidation: ptr.To(true)}, + } + if dieAfter != nil { + opts.NewClient = func(c *rest.Config, o client.Options) (client.Client, error) { + base, err := client.NewWithWatch(c, o) + if err != nil { + return nil, err + } + return crashtest.Wrap(base, &op.dead, dieAfter), nil + } + } + mgr, err := ctrl.NewManager(cfg, opts) + Expect(err).NotTo(HaveOccurred()) + setup(mgr) + + ctx, cancel := context.WithCancel(context.Background()) + op.cancel = cancel + go func() { + defer GinkgoRecover() + defer close(op.done) + Expect(mgr.Start(ctx)).To(Succeed()) + }() + return op +} + +func createNamespace(ctx context.Context, name string) { + Expect(client.IgnoreAlreadyExists(k8sClient.Create(ctx, &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: name}}))).To(Succeed()) +} + +type checker struct{ unreachable string } + +func (c checker) Check(_ context.Context, spec operatorv1alpha1.ImageRegistrySpec, _ operatorv1alpha1.ImageDeterministicSpec) registry.RegistryResult { + if spec.Image == c.unreachable { + return registry.RegistryResult{Image: spec.Image, State: registry.StateUnreachable} + } + return registry.RegistryResult{Image: spec.Image, State: registry.StateValid, Digest: digest, LayerHashesVerified: true} +} + +func setupExecute(chk checker) func(ctrl.Manager) { + return func(mgr ctrl.Manager) { + Expect(controller.RegisterPrepareIndex(mgr)).To(Succeed()) + Expect(controller.RegisterFreezeIndex(mgr)).To(Succeed()) + Expect(controller.RegisterExecuteIndex(mgr)).To(Succeed()) + r := controller.NewNetworkUpgradeExecuteReconciler(mgr.GetClient(), mgr.GetScheme()) + r.Checker = chk + Expect(r.SetupWithManager(mgr)).To(Succeed()) + } +} + +var _ = Describe("Operator restart mid-upgrade", func() { + const ( + ns = "restart-execute" + opID = "restart-op" + ) + ctx := context.Background() + execKey := types.NamespacedName{Name: "execute", Namespace: ns} + capsKey := types.NamespacedName{Name: "node0", Namespace: ns} + + It("keeps the pre-crash registry selection when killed right after patching the capsule", func() { + createNamespace(ctx, ns) + Expect(k8sClient.Create(ctx, &operatorv1alpha1.Orbit{ + ObjectMeta: metav1.ObjectMeta{Name: ns}, + Spec: operatorv1alpha1.OrbitSpec{Consensus: operatorv1alpha1.OrbitConsensus{ + Genesis: operatorv1alpha1.OrbitGenesis{AddressBook: operatorv1alpha1.OrbitAddressBook{LedgerId: "0x01"}}, + }}, + })).To(Succeed()) + Expect(k8sClient.Create(ctx, &operatorv1alpha1.ConsensusCapsule{ + ObjectMeta: metav1.ObjectMeta{Name: capsKey.Name, Namespace: ns, Labels: map[string]string{common.LabelOrbit: ns}}, + Spec: operatorv1alpha1.ConsensusCapsuleSpec{ + Orbit: ns, NodeId: 0, AccountId: "0.0.3", Weight: 1, Offline: true, + Log4j2ConfigRef: "log4j2", SettingsConfigRef: "settings", ApplicationPropertiesRef: "app", + PodProperties: operatorv1alpha1.ConsensusPodProperties{Containers: operatorv1alpha1.ConsensusContainers{ + ConsensusNode: operatorv1alpha1.ConsensusNodeContainer{ + SoftwareVersion: &operatorv1alpha1.SoftwareVersion{Repository: "gcr.io/old", ImageName: "cn", ImageTag: "0.74.0"}, + }, + }}, + }, + })).To(Succeed()) + + images := operatorv1alpha1.UpgradeImagesSpec{ConsensusNode: operatorv1alpha1.ImageComponentSpec{ + Version: "0.75.0", + Registries: []operatorv1alpha1.ImageRegistrySpec{ + {Image: "reg-a.io/cn:0.75.0"}, {Image: "reg-b.io/cn:0.75.0"}, + }, + }} + selected := operatorv1alpha1.SelectedSources{ConsensusNode: &operatorv1alpha1.SelectedSource{ + Image: "reg-a.io/cn:0.75.0", Digest: digest, LayerHashesVerified: true, + }} + execute := &operatorv1alpha1.NetworkUpgradeExecute{ + ObjectMeta: metav1.ObjectMeta{Name: execKey.Name, Namespace: ns, Labels: map[string]string{ + operatorv1alpha1.LabelOperationID: opID, operatorv1alpha1.LabelStep: operatorv1alpha1.StepExecute, + }}, + Spec: operatorv1alpha1.NetworkUpgradeExecuteSpec{ + Orbit: ns, OperationID: opID, ConsensusCapsule: capsKey.Name, Images: images, SelectedSources: selected, + }, + } + Expect(k8sClient.Create(ctx, execute)).To(Succeed()) + now := metav1.Now() + execute.Status.Phase = operatorv1alpha1.ExecutePhasePendingNodeUpgrade + execute.Status.StartTime = &now + execute.Status.NodeUpgradeStartTime = &now + Expect(k8sClient.Status().Update(ctx, execute)).To(Succeed()) + + By("running an operator that dies right after its capsule image patch") + first := startOperator(setupExecute(checker{unreachable: "reg-a.io/cn:0.75.0"}), func(obj client.Object) bool { + _, ok := obj.(*operatorv1alpha1.ConsensusCapsule) + return ok + }) + Eventually(first.dead.Load, timeout, interval).Should(BeTrue()) + first.stop() + + By("restarting with registry health inverted, so a re-selection would pick reg-a") + second := startOperator(setupExecute(checker{unreachable: "reg-b.io/cn:0.75.0"}), nil) + defer second.stop() + + Eventually(func(g Gomega) { + got := &operatorv1alpha1.NetworkUpgradeExecute{} + g.Expect(k8sClient.Get(ctx, execKey, got)).To(Succeed()) + g.Expect(got.Status.Phase).NotTo(Equal(operatorv1alpha1.ExecutePhaseFailed)) + g.Expect(apimeta.IsStatusConditionTrue(got.Status.Conditions, operatorv1alpha1.CondConsensusCapsulePatched)).To(BeTrue()) + }, timeout, interval).Should(Succeed()) + + caps := &operatorv1alpha1.ConsensusCapsule{} + Expect(k8sClient.Get(ctx, capsKey, caps)).To(Succeed()) + Expect(caps.Spec.PodProperties.Containers.ConsensusNode.SoftwareVersion.Repository).To(Equal("reg-b.io")) + }) +}) + +var _ = Describe("Operator restart mid-Helm-install", func() { + const ns = "restart-helm" + ctx := context.Background() + key := types.NamespacedName{Name: "mirror", Namespace: ns} + + It("unlocks the release left pending by the killed operator and finishes the install", func() { + createNamespace(ctx, ns) + mock := helmtest.NewMockClient() + started := make(chan struct{}) + var once sync.Once + mock.InstallOrUpgradeFunc = func(ctx context.Context, name, namespace string, _ operatorv1alpha1.HelmChartSpec, _ map[string]interface{}, _ helm.ReleaseOptions) (*helm.ReleaseInfo, error) { + mock.SetRelease(&helm.ReleaseInfo{Name: name, Namespace: namespace, Version: 1, Status: "pending-install", LastDeployed: time.Now()}) + once.Do(func() { close(started) }) + <-ctx.Done() + return nil, ctx.Err() + } + setup := func(mgr ctrl.Manager) { + Expect(controller.NewHelmCapsuleReconciler(mgr.GetClient(), mgr.GetScheme(), mock).SetupWithManager(mgr)).To(Succeed()) + } + + Expect(k8sClient.Create(ctx, &operatorv1alpha1.HelmCapsule{ + ObjectMeta: metav1.ObjectMeta{Name: key.Name, Namespace: ns}, + Spec: operatorv1alpha1.HelmCapsuleSpec{ + OrbitRef: &corev1.LocalObjectReference{Name: ns}, + Chart: operatorv1alpha1.HelmChartSpec{Repo: "oci://ghcr.io/hashgraph", Chart: "hedera-mirror", Version: "0.129.1"}, + Timeout: &metav1.Duration{Duration: 2 * time.Second}, + }, + })).To(Succeed()) + + By("killing the operator while helm install is in flight") + first := startOperator(setup, nil) + Eventually(started, timeout).Should(BeClosed()) + first.stop() + mock.InstallOrUpgradeFunc = nil + + By("restarting; the stale pending release must be unlocked once its timeout passes") + second := startOperator(setup, nil) + defer second.stop() + + Eventually(func(g Gomega) { + got := &operatorv1alpha1.HelmCapsule{} + g.Expect(k8sClient.Get(ctx, key, got)).To(Succeed()) + g.Expect(got.Status.Phase).To(Equal(operatorv1alpha1.PhaseSucceeded)) + g.Expect(got.Status.ReleaseStatus).To(Equal("deployed")) + }, timeout, interval).Should(Succeed()) + info, err := mock.GetRelease(key.Name, ns) + Expect(err).NotTo(HaveOccurred()) + Expect(info.Version).To(Equal(2)) + }) +}) From 047853ae1938dfd11a147dacba767345a16cc88b Mon Sep 17 00:00:00 2001 From: Artur Reznikov Date: Tue, 27 Oct 2026 14:26:31 -0700 Subject: [PATCH 2/3] fix(test): add missing test Signed-off-by: Artur Reznikov --- internal/crashtest/client.go | 49 ++++++++++++++++++++++++++++++++++++ 1 file changed, 49 insertions(+) create mode 100644 internal/crashtest/client.go diff --git a/internal/crashtest/client.go b/internal/crashtest/client.go new file mode 100644 index 000000000..478a184b9 --- /dev/null +++ b/internal/crashtest/client.go @@ -0,0 +1,49 @@ +// SPDX-License-Identifier: Apache-2.0 + +// Package crashtest simulates an operator process dying mid-reconcile. +package crashtest + +import ( + "context" + "errors" + "sync/atomic" + + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" +) + +var ErrDead = errors.New("simulated operator crash") + +// Wrap lets the write that satisfies dieAfter land, then fails every later write. +func Wrap(base client.WithWatch, dead *atomic.Bool, dieAfter func(client.Object) bool) client.WithWatch { + do := func(obj client.Object, write func() error) error { + if dead.Load() { + return ErrDead + } + err := write() + if err == nil && dieAfter(obj) { + dead.Store(true) + } + return err + } + return interceptor.NewClient(base, interceptor.Funcs{ + Create: func(ctx context.Context, c client.WithWatch, obj client.Object, o ...client.CreateOption) error { + return do(obj, func() error { return c.Create(ctx, obj, o...) }) + }, + Update: func(ctx context.Context, c client.WithWatch, obj client.Object, o ...client.UpdateOption) error { + return do(obj, func() error { return c.Update(ctx, obj, o...) }) + }, + Patch: func(ctx context.Context, c client.WithWatch, obj client.Object, p client.Patch, o ...client.PatchOption) error { + return do(obj, func() error { return c.Patch(ctx, obj, p, o...) }) + }, + Delete: func(ctx context.Context, c client.WithWatch, obj client.Object, o ...client.DeleteOption) error { + return do(obj, func() error { return c.Delete(ctx, obj, o...) }) + }, + SubResourceUpdate: func(ctx context.Context, c client.Client, sub string, obj client.Object, o ...client.SubResourceUpdateOption) error { + return do(obj, func() error { return c.SubResource(sub).Update(ctx, obj, o...) }) + }, + SubResourcePatch: func(ctx context.Context, c client.Client, sub string, obj client.Object, p client.Patch, o ...client.SubResourcePatchOption) error { + return do(obj, func() error { return c.SubResource(sub).Patch(ctx, obj, p, o...) }) + }, + }) +} From 59926bd2eb42e38caf8083fdfc850866e0b30e45 Mon Sep 17 00:00:00 2001 From: Artur Reznikov Date: Wed, 28 Oct 2026 11:32:22 -0700 Subject: [PATCH 3/3] fix(api/helm): address pr review nit Signed-off-by: Artur Reznikov --- internal/resources/helm/release.go | 9 +++++++++ internal/resources/helm/release_test.go | 3 +++ 2 files changed, 12 insertions(+) diff --git a/internal/resources/helm/release.go b/internal/resources/helm/release.go index dd6079a20..5cb827f6c 100644 --- a/internal/resources/helm/release.go +++ b/internal/resources/helm/release.go @@ -87,6 +87,15 @@ func (r *ReleaseSubReconciler) Reconcile(ctx context.Context, hc *operatorv1alph if existing != nil && existing.Pending() { if wait := opts.Timeout - time.Since(existing.LastDeployed); wait > 0 { + if err := retry.RetryOnConflict(retry.DefaultRetry, func() error { + if err := r.Client.Get(ctx, types.NamespacedName{Name: hc.Name, Namespace: hc.Namespace}, hc); err != nil { + return err + } + hc.Status.ReleaseStatus = existing.Status + return common.WarnOnConflict(ctx, r.Client.Status().Update(ctx, hc)) + }); err != nil { + return ctrl.Result{}, err + } common.InfoL(&rl, ReasonReleasePending).Str("helmStatus", existing.Status).Msg("Helm release has an operation in progress, waiting") return ctrl.Result{RequeueAfter: wait}, nil } diff --git a/internal/resources/helm/release_test.go b/internal/resources/helm/release_test.go index 452d9f13a..08dde62b3 100644 --- a/internal/resources/helm/release_test.go +++ b/internal/resources/helm/release_test.go @@ -599,6 +599,9 @@ func TestReleaseSubReconciler_waitsOutFreshPendingRelease(t *testing.T) { if info.Status != "pending-upgrade" || info.Version != 3 { t.Errorf("fresh pending release was touched: status=%s revision=%d", info.Status, info.Version) } + if hc.Status.ReleaseStatus != "pending-upgrade" { + t.Errorf("hc.Status.ReleaseStatus = %q, want pending-upgrade", hc.Status.ReleaseStatus) + } } // ─── Chart credentials ────────────────────────────────────────────────────────