diff --git a/cmd/manager/main.go b/cmd/manager/main.go index 64d7a435068..d6badb5d10b 100644 --- a/cmd/manager/main.go +++ b/cmd/manager/main.go @@ -559,9 +559,10 @@ func main() { } if err = (&opscontrollers.OpsRequestReconciler{ - Client: client, - Scheme: mgr.GetScheme(), - Recorder: mgr.GetEventRecorderFor("ops-request-controller"), + Client: client, + APIReader: mgr.GetAPIReader(), + Scheme: mgr.GetScheme(), + Recorder: mgr.GetEventRecorderFor("ops-request-controller"), }).SetupWithManager(mgr, multiClusterMgr); err != nil { setupLog.Error(err, "unable to create controller", "controller", "OpsRequest") os.Exit(1) diff --git a/controllers/operations/opsrequest_controller.go b/controllers/operations/opsrequest_controller.go index d183737acde..7ef0ee2fb91 100644 --- a/controllers/operations/opsrequest_controller.go +++ b/controllers/operations/opsrequest_controller.go @@ -58,8 +58,9 @@ import ( // OpsRequestReconciler reconciles a OpsRequest object type OpsRequestReconciler struct { client.Client - Scheme *runtime.Scheme - Recorder record.EventRecorder + APIReader client.Reader + Scheme *runtime.Scheme + Recorder record.EventRecorder } // +kubebuilder:rbac:groups=operations.kubeblocks.io,resources=opsrequests,verbs=get;list;watch;create;update;patch;delete @@ -80,7 +81,7 @@ func (r *OpsRequestReconciler) Reconcile(ctx context.Context, req ctrl.Request) } reqCtx.Log.Info("reconcile", "opsRequest", req.NamespacedName) opsCtrlHandler := &opsControllerHandler{} - return opsCtrlHandler.Handle(reqCtx, &operations.OpsResource{Recorder: r.Recorder}, + return opsCtrlHandler.Handle(reqCtx, &operations.OpsResource{Recorder: r.Recorder, APIReader: r.APIReader}, r.fetchOpsRequest, r.fetchCluster, r.handleDeletion, diff --git a/controllers/operations/suite_test.go b/controllers/operations/suite_test.go index 39a57ba57ca..42e89c0e468 100644 --- a/controllers/operations/suite_test.go +++ b/controllers/operations/suite_test.go @@ -215,9 +215,10 @@ var _ = BeforeSuite(func() { Expect(err).ToNot(HaveOccurred()) err = (&OpsRequestReconciler{ - Client: k8sManager.GetClient(), - Scheme: k8sManager.GetScheme(), - Recorder: k8sManager.GetEventRecorderFor("ops-request-controller"), + Client: k8sManager.GetClient(), + APIReader: k8sManager.GetAPIReader(), + Scheme: k8sManager.GetScheme(), + Recorder: k8sManager.GetEventRecorderFor("ops-request-controller"), }).SetupWithManager(k8sManager, nil) Expect(err).ToNot(HaveOccurred()) diff --git a/pkg/operations/switchover.go b/pkg/operations/switchover.go index 1584b4cae36..2b2ffebd9e7 100644 --- a/pkg/operations/switchover.go +++ b/pkg/operations/switchover.go @@ -23,6 +23,7 @@ import ( "context" "fmt" "reflect" + "strings" "time" "github.com/pkg/errors" @@ -31,7 +32,9 @@ import ( "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/util/uuid" "k8s.io/client-go/tools/record" + "k8s.io/client-go/util/retry" "sigs.k8s.io/controller-runtime/pkg/client" appsv1 "github.com/apecloud/kubeblocks/apis/apps/v1" @@ -43,9 +46,98 @@ import ( // switchover constants const ( - KBSwitchoverKey = "Switchover" + KBSwitchoverKey = "Switchover" + switchoverDispatchClaimMessagePrefix = "switchover dispatch claimed before lifecycle call: SwitchoverDispatch/" + switchoverDispatchClaimMessageFmt = switchoverDispatchClaimMessagePrefix + "%s/%s/%s/%s/%s" + switchoverDispatchOutcomeMessagePrefix = "switchover dispatch outcome persisted: SwitchoverDispatch/" + switchoverDispatchOutcomeMessageFmt = switchoverDispatchOutcomeMessagePrefix + "%s/%s/%s/%s/%s; %s" ) +var errSwitchoverDispatchClaimLost = errors.New("switchover dispatch claim was lost") + +type switchoverDispatchClaim struct { + opsRequestUID string + componentName string + instanceName string + candidateName string + token string +} + +func newSwitchoverDispatchClaim(opsRequest *opsv1alpha1.OpsRequest, compName string, + switchover opsv1alpha1.Switchover) switchoverDispatchClaim { + return switchoverDispatchClaim{ + opsRequestUID: string(opsRequest.UID), + componentName: compName, + instanceName: switchover.InstanceName, + candidateName: switchover.CandidateName, + token: string(uuid.NewUUID()), + } +} + +func parseSwitchoverDispatchClaim(message string) (switchoverDispatchClaim, bool) { + if !strings.HasPrefix(message, switchoverDispatchClaimMessagePrefix) { + return switchoverDispatchClaim{}, false + } + fields := strings.Split(strings.TrimPrefix(message, switchoverDispatchClaimMessagePrefix), "/") + if len(fields) != 5 || fields[0] == "" || fields[1] == "" || fields[2] == "" || fields[4] == "" { + return switchoverDispatchClaim{}, false + } + return switchoverDispatchClaim{ + opsRequestUID: fields[0], + componentName: fields[1], + instanceName: fields[2], + candidateName: fields[3], + token: fields[4], + }, true +} + +func (claim switchoverDispatchClaim) message() string { + return fmt.Sprintf(switchoverDispatchClaimMessageFmt, claim.opsRequestUID, claim.componentName, + claim.instanceName, claim.candidateName, claim.token) +} + +func (claim switchoverDispatchClaim) outcomeMessage(message string) string { + return fmt.Sprintf(switchoverDispatchOutcomeMessageFmt, claim.opsRequestUID, claim.componentName, + claim.instanceName, claim.candidateName, claim.token, message) +} + +func parseSwitchoverDispatchOutcomeMessage(message string) (switchoverDispatchClaim, string, bool) { + if !strings.HasPrefix(message, switchoverDispatchOutcomeMessagePrefix) { + return switchoverDispatchClaim{}, "", false + } + parts := strings.SplitN(strings.TrimPrefix(message, switchoverDispatchOutcomeMessagePrefix), "; ", 2) + if len(parts) != 2 || parts[1] == "" { + return switchoverDispatchClaim{}, "", false + } + fields := strings.Split(parts[0], "/") + if len(fields) != 5 || fields[0] == "" || fields[1] == "" || fields[2] == "" || fields[4] == "" { + return switchoverDispatchClaim{}, "", false + } + return switchoverDispatchClaim{ + opsRequestUID: fields[0], + componentName: fields[1], + instanceName: fields[2], + candidateName: fields[3], + token: fields[4], + }, parts[1], true +} + +func (claim switchoverDispatchClaim) matchesIdentity(opsRequest *opsv1alpha1.OpsRequest, compName string, + switchover opsv1alpha1.Switchover) bool { + return claim.opsRequestUID == string(opsRequest.UID) && + claim.componentName == compName && + claim.instanceName == switchover.InstanceName && + claim.candidateName == switchover.CandidateName +} + +func (claim switchoverDispatchClaim) equal(other switchoverDispatchClaim) bool { + return claim.opsRequestUID == other.opsRequestUID && + claim.componentName == other.componentName && + claim.instanceName == other.instanceName && + claim.candidateName == other.candidateName && + claim.token == other.token +} + type switchoverOpsHandler struct{} var _ OpsHandler = switchoverOpsHandler{} @@ -174,17 +266,16 @@ func handleSwitchovers(reqCtx intctrlutil.RequestCtx, cli client.Client, opsRes var completedCount, failedCount int32 opsRequest := opsRes.OpsRequest - oldOpsRequestStatus := opsRequest.Status.DeepCopy() - patch := client.MergeFrom(opsRequest.DeepCopy()) - for _, switchover := range opsRequest.Spec.SwitchoverList { if err := handleSwitchover(reqCtx, cli, opsRes, &switchover, opsRequest, &completedCount, &failedCount); err != nil { return expectCount, completedCount, failedCount, err } } - opsRequest.Status.Progress = fmt.Sprintf("%d/%d", completedCount, expectCount) - if !reflect.DeepEqual(*oldOpsRequestStatus, opsRequest.Status) { + progress := fmt.Sprintf("%d/%d", completedCount, expectCount) + if opsRequest.Status.Progress != progress { + patch := client.MergeFromWithOptions(opsRequest.DeepCopy(), client.MergeFromWithOptimisticLock{}) + opsRequest.Status.Progress = progress if err := cli.Status().Patch(reqCtx.Ctx, opsRequest, patch); err != nil { return expectCount, completedCount, failedCount, err } @@ -229,40 +320,251 @@ func handleSwitchover(reqCtx intctrlutil.RequestCtx, cli client.Client, opsRes * switch progressDetail.Status { case opsv1alpha1.PendingProgressStatus: - if err = runtime.Switchover(reqCtx.Ctx, synthesizedComp.Namespace, synthesizedComp.ClusterName, synthesizedComp.Name, switchover.InstanceName, switchover.CandidateName); err != nil { - progressDetail.Status = opsv1alpha1.FailedProgressStatus - progressDetail.Message = fmt.Sprintf("component %s %s", compName, err.Error()) - } else { - progressDetail.Message = "doing switchover" - progressDetail.Status = opsv1alpha1.ProcessingProgressStatus + if opsRes.APIReader == nil { + return errors.New("APIReader is required to confirm a switchover dispatch claim") } - progressDetail.StartTime = metav1.Now() + if opsRequest.UID == "" { + return errors.New("OpsRequest UID is required to create a switchover dispatch claim") + } + claim := newSwitchoverDispatchClaim(opsRequest, compName, *switchover) + claimMessage := claim.message() + patchErr := patchSwitchoverProgressStatus(reqCtx.Ctx, cli, opsRequest, compName, objectKey, + func(detail *opsv1alpha1.ProgressStatusDetail) { + detail.Status = opsv1alpha1.ProcessingProgressStatus + detail.Message = claimMessage + detail.StartTime = metav1.Now() + }) + ownsClaim, confirmErr := confirmSwitchoverDispatchClaim(reqCtx.Ctx, opsRes, opsRequest, compName, objectKey, claim) + if confirmErr != nil { + if patchErr != nil { + return errors.Wrapf(patchErr, "failed to confirm switchover dispatch claim: %v", confirmErr) + } + return confirmErr + } + if !ownsClaim { + if patchErr != nil { + return patchErr + } + return fmt.Errorf("switchover dispatch claim belongs to another writer for component %s", compName) + } + + actionErr := runtime.Switchover(reqCtx.Ctx, synthesizedComp.Namespace, synthesizedComp.ClusterName, synthesizedComp.Name, switchover.InstanceName, switchover.CandidateName) + outcomeStatus := opsv1alpha1.ProcessingProgressStatus + outcomeMessage := "doing switchover" + if actionErr != nil { + outcomeStatus = opsv1alpha1.FailedProgressStatus + outcomeMessage = fmt.Sprintf("component %s %s", compName, actionErr.Error()) + } else if !progressDetail.StartTime.IsZero() && time.Now().After(progressDetail.StartTime.Add(5*time.Minute)) { + // StartTime is committed before the lifecycle call. A successful response that arrives after + // this deadline remains a fail-closed timeout instead of being treated as a fresh success. + outcomeStatus = opsv1alpha1.FailedProgressStatus + outcomeMessage = "switchover timeout after 5 minutes" + } + progressDetail, err = persistKnownSwitchoverDispatchOutcome(reqCtx, cli, opsRes.APIReader, opsRequest, + compName, objectKey, claim, outcomeStatus, outcomeMessage) + if err != nil { + return err + } + if isCompletedProgressStatus(progressDetail.Status) { + *completedCount++ + if progressDetail.Status == opsv1alpha1.FailedProgressStatus { + *failedCount++ + } + } + return nil case opsv1alpha1.ProcessingProgressStatus: - targetRole := progressDetail.Group - if switchover.CandidateName != "" { - candidateInstance, err := getSwitchoverPodBackedInstance(runtime, synthesizedComp.Namespace, synthesizedComp.ClusterName, synthesizedComp.Name, switchover.CandidateName) - switch { - case err != nil && !apierrors.IsNotFound(err): - return err - case err != nil: - progressDetail.Message = fmt.Sprintf(`component %s candidate instance "%s" not found`, compName, switchover.CandidateName) - progressDetail.Status = opsv1alpha1.FailedProgressStatus - case targetRole == candidateInstance.GetRole(): + oldOpsRequestStatus := opsRequest.Status.DeepCopy() + patch := client.MergeFromWithOptions(opsRequest.DeepCopy(), client.MergeFromWithOptimisticLock{}) + claim, hasClaim := parseSwitchoverDispatchClaim(progressDetail.Message) + outcomeClaim, _, hasOutcome := parseSwitchoverDispatchOutcomeMessage(progressDetail.Message) + claimProtocolMismatch := strings.HasPrefix(progressDetail.Message, switchoverDispatchClaimMessagePrefix) && + (!hasClaim || !claim.matchesIdentity(opsRequest, compName, *switchover)) + outcomeProtocolMismatch := strings.HasPrefix(progressDetail.Message, switchoverDispatchOutcomeMessagePrefix) && + (!hasOutcome || !outcomeClaim.matchesIdentity(opsRequest, compName, *switchover)) + protocolIdentityMismatch := claimProtocolMismatch || outcomeProtocolMismatch + switch { + case protocolIdentityMismatch: + progressDetail.Message = fmt.Sprintf("component %s switchover dispatch protocol identity changed or is malformed; outcome is unknown and lifecycle action will not be retried", compName) + progressDetail.Status = opsv1alpha1.FailedProgressStatus + case hasClaim && switchover.CandidateName == "": + progressDetail.Message = fmt.Sprintf("component %s switchover dispatch outcome is unknown; lifecycle action will not be retried", compName) + progressDetail.Status = opsv1alpha1.FailedProgressStatus + default: + targetRole := progressDetail.Group + if switchover.CandidateName != "" { + candidateInstance, err := getSwitchoverPodBackedInstance(runtime, synthesizedComp.Namespace, synthesizedComp.ClusterName, synthesizedComp.Name, switchover.CandidateName) + switch { + case err != nil && !apierrors.IsNotFound(err): + return err + case err != nil: + progressDetail.Message = fmt.Sprintf(`component %s candidate instance "%s" not found`, compName, switchover.CandidateName) + progressDetail.Status = opsv1alpha1.FailedProgressStatus + case targetRole == candidateInstance.GetRole(): + progressDetail.Message = "do switchover succeed" + progressDetail.Status = opsv1alpha1.SucceedProgressStatus + default: + progressDetail.Message = fmt.Sprintf("component %s is waiting for candidate pod %s role change, current role %q, expected role %q", + compName, switchover.CandidateName, candidateInstance.GetRole(), targetRole) + } + } else { progressDetail.Message = "do switchover succeed" progressDetail.Status = opsv1alpha1.SucceedProgressStatus - default: - progressDetail.Message = fmt.Sprintf("component %s is waiting for candidate pod %s role change, current role %q, expected role %q", - compName, switchover.CandidateName, candidateInstance.GetRole(), targetRole) } - } else { - progressDetail.Message = "do switchover succeed" - progressDetail.Status = opsv1alpha1.SucceedProgressStatus } + handleProgressDetail(reqCtx, opsRequest, progressDetail, compName, completedCount, failedCount) + if !reflect.DeepEqual(*oldOpsRequestStatus, opsRequest.Status) { + if err := cli.Status().Patch(reqCtx.Ctx, opsRequest, patch); err != nil { + return err + } + } + return nil } handleProgressDetail(reqCtx, opsRequest, progressDetail, compName, completedCount, failedCount) return nil } +func patchSwitchoverProgressStatus(ctx context.Context, cli client.Client, opsRequest *opsv1alpha1.OpsRequest, + compName, objectKey string, mutate func(*opsv1alpha1.ProgressStatusDetail)) error { + patch := client.MergeFromWithOptions(opsRequest.DeepCopy(), client.MergeFromWithOptimisticLock{}) + progressDetail := findStatusProgressDetail(opsRequest.Status.Components[compName].ProgressDetails, objectKey) + if progressDetail == nil { + return fmt.Errorf("progress detail not found for component %s", compName) + } + mutate(progressDetail) + return cli.Status().Patch(ctx, opsRequest, patch) +} + +func confirmSwitchoverDispatchClaim(ctx context.Context, opsRes *OpsResource, + opsRequest *opsv1alpha1.OpsRequest, compName, objectKey string, expected switchoverDispatchClaim) (bool, error) { + reader := opsRes.APIReader + if reader == nil { + return false, errors.New("APIReader is required to confirm a switchover dispatch claim") + } + var confirmed *opsv1alpha1.OpsRequest + var live switchoverDispatchClaim + err := retry.OnError(retry.DefaultBackoff, shouldRetrySwitchoverStatusError, func() error { + fresh := &opsv1alpha1.OpsRequest{} + if err := reader.Get(ctx, client.ObjectKeyFromObject(opsRequest), fresh); err != nil { + return err + } + progressDetail := findStatusProgressDetail(fresh.Status.Components[compName].ProgressDetails, objectKey) + if progressDetail == nil || progressDetail.Status != opsv1alpha1.ProcessingProgressStatus { + return errors.Wrapf(errSwitchoverDispatchClaimLost, "claim was not committed for component %s", compName) + } + parsed, ok := parseSwitchoverDispatchClaim(progressDetail.Message) + if !ok || parsed.opsRequestUID != expected.opsRequestUID || parsed.componentName != expected.componentName || + parsed.instanceName != expected.instanceName || parsed.candidateName != expected.candidateName { + return errors.Wrapf(errSwitchoverDispatchClaimLost, "claim identity does not match component %s", compName) + } + confirmed = fresh + live = parsed + return nil + }) + if err != nil { + return false, err + } + confirmed.DeepCopyInto(opsRequest) + return live.token == expected.token, nil +} + +func shouldRetrySwitchoverStatusError(err error) bool { + if errors.Is(err, errSwitchoverDispatchClaimLost) || + errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) { + return false + } + return !apierrors.IsNotFound(err) && !apierrors.IsForbidden(err) && + !apierrors.IsUnauthorized(err) && !apierrors.IsInvalid(err) && + !apierrors.IsRequestEntityTooLargeError(err) && !apierrors.IsBadRequest(err) && + !apierrors.IsMethodNotSupported(err) && !apierrors.IsUnsupportedMediaType(err) && + !apierrors.IsNotAcceptable(err) +} + +func retryKnownSwitchoverOutcomeUntilContextDone(ctx context.Context, fn func() error) error { + delay := 10 * time.Millisecond + const maxDelay = time.Second + for { + err := fn() + if err == nil || !shouldRetrySwitchoverStatusError(err) { + return err + } + select { + case <-ctx.Done(): + return errors.Wrapf(ctx.Err(), "stopped persisting the known switchover outcome after retryable error: %v", err) + case <-time.After(delay): + } + if delay < maxDelay { + delay *= 2 + if delay > maxDelay { + delay = maxDelay + } + } + } +} + +func persistKnownSwitchoverDispatchOutcome( + reqCtx intctrlutil.RequestCtx, + cli client.Client, + reader client.Reader, + opsRequest *opsv1alpha1.OpsRequest, + compName, + objectKey string, + expected switchoverDispatchClaim, + status opsv1alpha1.ProgressStatus, + message string) (*opsv1alpha1.ProgressStatusDetail, error) { + var persisted *opsv1alpha1.OpsRequest + persistedMessage := expected.outcomeMessage(message) + err := retryKnownSwitchoverOutcomeUntilContextDone(reqCtx.Ctx, func() error { + fresh := &opsv1alpha1.OpsRequest{} + if err := reader.Get(reqCtx.Ctx, client.ObjectKeyFromObject(opsRequest), fresh); err != nil { + return err + } + componentStatus, ok := fresh.Status.Components[compName] + if !ok { + return errors.Wrapf(errSwitchoverDispatchClaimLost, "component %s status is missing", compName) + } + progressDetail := findStatusProgressDetail(componentStatus.ProgressDetails, objectKey) + if progressDetail == nil { + return errors.Wrapf(errSwitchoverDispatchClaimLost, "component %s progress detail is missing", compName) + } + persistedClaim, persistedOutcome, hasPersistedOutcome := parseSwitchoverDispatchOutcomeMessage(progressDetail.Message) + if progressDetail.Status == status && hasPersistedOutcome && persistedClaim.equal(expected) && persistedOutcome == message { + persisted = fresh + return nil + } + live, ok := parseSwitchoverDispatchClaim(progressDetail.Message) + if !ok || progressDetail.Status != opsv1alpha1.ProcessingProgressStatus || !live.equal(expected) { + return errors.Wrapf(errSwitchoverDispatchClaimLost, "component %s", compName) + } + + old := fresh.DeepCopy() + progressDetail.Status = status + progressDetail.Message = persistedMessage + updateProgressDetailTime(progressDetail) + componentStatus.Phase = appsv1.UpdatingComponentPhase + componentStatus.ProgressDetails = append([]opsv1alpha1.ProgressStatusDetail(nil), componentStatus.ProgressDetails...) + fresh.Status.Components[compName] = componentStatus + patch := client.MergeFromWithOptions(old, client.MergeFromWithOptimisticLock{}) + if err := cli.Status().Patch(reqCtx.Ctx, fresh, patch); err != nil { + return err + } + persisted = fresh + return nil + }) + if err != nil { + return nil, err + } + if persisted == nil { + return nil, errors.New("switchover dispatch outcome was not persisted") + } + persisted.DeepCopyInto(opsRequest) + progressDetail := findStatusProgressDetail(opsRequest.Status.Components[compName].ProgressDetails, objectKey) + if progressDetail == nil { + return nil, fmt.Errorf("progress detail not found for component %s after persisting switchover outcome", compName) + } + sendProgressDetailEvent(reqCtx.Recorder, opsRequest, *progressDetail) + return progressDetail, nil +} + func getSwitchoverPodBackedInstance(runtime OpsRuntime, namespace, clusterName, compName, instanceName string) (Instance, error) { instance, err := runtime.GetInstance(namespace, clusterName, compName, instanceName) if err != nil { diff --git a/pkg/operations/switchover_test.go b/pkg/operations/switchover_test.go index 6accd930265..39575f99a61 100644 --- a/pkg/operations/switchover_test.go +++ b/pkg/operations/switchover_test.go @@ -22,6 +22,7 @@ package operations import ( "context" "fmt" + "net/http" "github.com/golang/mock/gomock" . "github.com/onsi/ginkgo/v2" @@ -29,6 +30,9 @@ import ( corev1 "k8s.io/api/core/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/client-go/util/retry" "sigs.k8s.io/controller-runtime/pkg/client" appsv1 "github.com/apecloud/kubeblocks/apis/apps/v1" @@ -42,6 +46,53 @@ import ( testops "github.com/apecloud/kubeblocks/pkg/testutil/operations" ) +type interceptStatusClient struct { + client.Client + patch func(context.Context, client.Object, client.Patch, ...client.SubResourcePatchOption) error +} + +type interceptReader struct { + client.Reader + get func(context.Context, client.ObjectKey, client.Object, ...client.GetOption) error +} + +func (r *interceptReader) Get(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if r.get != nil { + return r.get(ctx, key, obj, opts...) + } + return r.Reader.Get(ctx, key, obj, opts...) +} + +func (c *interceptStatusClient) Status() client.SubResourceWriter { + return &interceptStatusWriter{ + SubResourceWriter: c.Client.Status(), + patch: c.patch, + } +} + +type interceptStatusWriter struct { + client.SubResourceWriter + patch func(context.Context, client.Object, client.Patch, ...client.SubResourcePatchOption) error +} + +func (w *interceptStatusWriter) Patch(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error { + if w.patch != nil { + return w.patch(ctx, obj, patch, opts...) + } + return w.SubResourceWriter.Patch(ctx, obj, patch, opts...) +} + +func switchoverDispatchClaimMessageForTest(opsRequest *opsv1alpha1.OpsRequest, componentName string, + switchover opsv1alpha1.Switchover, token string) string { + return switchoverDispatchClaim{ + opsRequestUID: string(opsRequest.UID), + componentName: componentName, + instanceName: switchover.InstanceName, + candidateName: switchover.CandidateName, + token: token, + }.message() +} + var _ = Describe("", func() { var ( compDefName = "test-compdef-" @@ -152,9 +203,671 @@ var _ = Describe("", func() { } opsRes = &OpsResource{ - Cluster: clusterObj, - Recorder: k8sManager.GetEventRecorderFor("opsrequest-controller"), + Cluster: clusterObj, + Recorder: k8sManager.GetEventRecorderFor("opsrequest-controller"), + APIReader: k8sClient, + } + }) + + preparePendingSwitchover := func(candidateName string) (client.ObjectKey, string) { + ops := testops.NewOpsRequestObj("ops-switchover-"+testCtx.GetRandomStr(), testCtx.DefaultNamespace, + clusterObj.Name, opsv1alpha1.SwitchoverType) + instanceName := fmt.Sprintf("%s-%s-%d", clusterObj.Name, defaultCompName, 1) + ops.Spec.SwitchoverList = []opsv1alpha1.Switchover{{ + ComponentName: defaultCompName, + InstanceName: instanceName, + CandidateName: candidateName, + }} + opsRes.OpsRequest = testops.CreateOpsRequest(ctx, testCtx, ops) + opsRes.OpsRequest.Status.Phase = opsv1alpha1.OpsPendingPhase + + _, err := GetOpsManager().Do(reqCtx, k8sClient, opsRes) + Expect(err).ShouldNot(HaveOccurred()) + _, err = GetOpsManager().Do(reqCtx, k8sClient, opsRes) + Expect(err).ShouldNot(HaveOccurred()) + + key := client.ObjectKeyFromObject(opsRes.OpsRequest) + stored := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, key, stored)).Should(Succeed()) + patch := client.MergeFrom(stored.DeepCopy()) + stored.Status = opsRes.OpsRequest.DeepCopy().Status + stored.Status.Phase = opsv1alpha1.OpsRunningPhase + Expect(k8sClient.Status().Patch(ctx, stored, patch)).Should(Succeed()) + opsRes.OpsRequest = stored + return key, instanceName + } + + It("persists the Processing dispatch claim before the lifecycle call", func() { + key, instanceName := preparePendingSwitchover("") + actionCalls := 0 + testapps.MockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) { + recorder.Action(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(func(ctx context.Context, req kbagentproto.ActionRequest) (kbagentproto.ActionResponse, error) { + actionCalls++ + Expect(req.Parameters["KB_SWITCHOVER_CURRENT_NAME"]).Should(Equal(instanceName)) + claimed := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, key, claimed)).Should(Succeed()) + progressDetail := findStatusProgressDetail(claimed.Status.Components[defaultCompName].ProgressDetails, + getProgressObjectKey(KBSwitchoverKey, defaultCompName)) + Expect(progressDetail).ShouldNot(BeNil()) + Expect(progressDetail.Status).Should(Equal(opsv1alpha1.ProcessingProgressStatus)) + claim, ok := parseSwitchoverDispatchClaim(progressDetail.Message) + Expect(ok).Should(BeTrue()) + Expect(claim.matchesIdentity(claimed, defaultCompName, claimed.Spec.SwitchoverList[0])).Should(BeTrue()) + Expect(claim.token).ShouldNot(BeEmpty()) + return kbagentproto.ActionResponse{Message: "mock success"}, nil + }) + }) + + _, _, _, err := handleSwitchovers(reqCtx, k8sClient, opsRes) + Expect(err).ShouldNot(HaveOccurred()) + Expect(actionCalls).Should(Equal(1)) + }) + + It("retries a transient direct-read failure after the dispatch claim commits", func() { + preparePendingSwitchover("") + actionCalls := 0 + testapps.MockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) { + recorder.Action(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(func(context.Context, kbagentproto.ActionRequest) (kbagentproto.ActionResponse, error) { + actionCalls++ + return kbagentproto.ActionResponse{Message: "mock success"}, nil + }) + }) + + injectedErr := fmt.Errorf("injected transient direct-read failure") + failedReads := 0 + opsRes.APIReader = &interceptReader{ + Reader: k8sClient, + get: func(ctx context.Context, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error { + if _, ok := obj.(*opsv1alpha1.OpsRequest); ok && failedReads == 0 { + failedReads++ + return injectedErr + } + return k8sClient.Get(ctx, key, obj, opts...) + }, + } + + _, _, _, err := handleSwitchovers(reqCtx, k8sClient, opsRes) + Expect(err).ShouldNot(HaveOccurred()) + Expect(failedReads).Should(Equal(1)) + Expect(actionCalls).Should(Equal(1)) + }) + + It("does not call the lifecycle action when the Processing claim loses a resource-version race", func() { + key, _ := preparePendingSwitchover("") + actionCalls := 0 + testapps.MockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) { + recorder.Action(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(func(context.Context, kbagentproto.ActionRequest) (kbagentproto.ActionResponse, error) { + actionCalls++ + return kbagentproto.ActionResponse{Message: "mock success"}, nil + }) + }) + + injected := false + conflictingClient := &interceptStatusClient{ + Client: k8sClient, + patch: func(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error { + if _, ok := obj.(*opsv1alpha1.OpsRequest); ok && !injected { + injected = true + concurrent := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, key, concurrent)).Should(Succeed()) + concurrentPatch := client.MergeFrom(concurrent.DeepCopy()) + concurrent.Status.Progress = "99/99" + Expect(k8sClient.Status().Patch(ctx, concurrent, concurrentPatch)).Should(Succeed()) + } + return k8sClient.Status().Patch(ctx, obj, patch, opts...) + }, + } + + _, _, _, err := handleSwitchovers(reqCtx, conflictingClient, opsRes) + Expect(apierrors.IsConflict(err)).Should(BeTrue()) + Expect(actionCalls).Should(Equal(0)) + }) + + It("lets only the writer whose unique token was committed dispatch after ambiguous patch responses", func() { + preparePendingSwitchover("") + writerOneOps := *opsRes + writerOneOps.OpsRequest = opsRes.OpsRequest.DeepCopy() + writerTwoOps := *opsRes + writerTwoOps.OpsRequest = opsRes.OpsRequest.DeepCopy() + + writerOneResponseLoss := fmt.Errorf("writer one claim response lost") + writerTwoResponseLoss := fmt.Errorf("writer two conflict response lost") + var writerOneClaim, writerTwoClaim string + actionCalls := 0 + var writerTwoErr error + + writerTwoClient := &interceptStatusClient{ + Client: k8sClient, + patch: func(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error { + if writerTwoClaim == "" { + opsRequest := obj.(*opsv1alpha1.OpsRequest) + progressDetail := findStatusProgressDetail(opsRequest.Status.Components[defaultCompName].ProgressDetails, + getProgressObjectKey(KBSwitchoverKey, defaultCompName)) + Expect(progressDetail).ShouldNot(BeNil()) + writerTwoClaim = progressDetail.Message + conflictErr := k8sClient.Status().Patch(ctx, obj, patch, opts...) + Expect(apierrors.IsConflict(conflictErr)).Should(BeTrue()) + return writerTwoResponseLoss + } + return k8sClient.Status().Patch(ctx, obj, patch, opts...) + }, + } + writerOneClient := &interceptStatusClient{ + Client: k8sClient, + patch: func(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error { + if writerOneClaim == "" { + opsRequest := obj.(*opsv1alpha1.OpsRequest) + progressDetail := findStatusProgressDetail(opsRequest.Status.Components[defaultCompName].ProgressDetails, + getProgressObjectKey(KBSwitchoverKey, defaultCompName)) + Expect(progressDetail).ShouldNot(BeNil()) + writerOneClaim = progressDetail.Message + Expect(k8sClient.Status().Patch(ctx, obj, patch, opts...)).Should(Succeed()) + return writerOneResponseLoss + } + return k8sClient.Status().Patch(ctx, obj, patch, opts...) + }, + } + + testapps.MockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) { + recorder.Action(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(func(context.Context, kbagentproto.ActionRequest) (kbagentproto.ActionResponse, error) { + actionCalls++ + if actionCalls == 1 { + _, _, _, writerTwoErr = handleSwitchovers(reqCtx, writerTwoClient, &writerTwoOps) + live := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(writerOneOps.OpsRequest), live)).Should(Succeed()) + progressDetail := findStatusProgressDetail(live.Status.Components[defaultCompName].ProgressDetails, + getProgressObjectKey(KBSwitchoverKey, defaultCompName)) + Expect(progressDetail).ShouldNot(BeNil()) + Expect(progressDetail.Message).Should(Equal(writerOneClaim)) + Expect(progressDetail.Message).ShouldNot(Equal(writerTwoClaim)) + } + return kbagentproto.ActionResponse{Message: "mock success"}, nil + }) + }) + + _, _, _, writerOneErr := handleSwitchovers(reqCtx, writerOneClient, &writerOneOps) + Expect(writerOneErr).ShouldNot(HaveOccurred()) + Expect(writerTwoErr).Should(MatchError(writerTwoResponseLoss)) + Expect(actionCalls).Should(Equal(1)) + Expect(writerOneClaim).Should(HavePrefix(switchoverDispatchClaimMessagePrefix)) + Expect(writerTwoClaim).Should(HavePrefix(switchoverDispatchClaimMessagePrefix)) + Expect(writerOneClaim).ShouldNot(Equal(writerTwoClaim)) + }) + + It("fails closed after restart when a no-candidate dispatch outcome is unknown", func() { + key, _ := preparePendingSwitchover("") + fresh := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, key, fresh)).Should(Succeed()) + patch := client.MergeFrom(fresh.DeepCopy()) + progressDetail := findStatusProgressDetail(fresh.Status.Components[defaultCompName].ProgressDetails, + getProgressObjectKey(KBSwitchoverKey, defaultCompName)) + Expect(progressDetail).ShouldNot(BeNil()) + progressDetail.Status = opsv1alpha1.ProcessingProgressStatus + progressDetail.Message = switchoverDispatchClaimMessageForTest(fresh, defaultCompName, + fresh.Spec.SwitchoverList[0], "restart-token") + progressDetail.StartTime = metav1.Now() + Expect(k8sClient.Status().Patch(ctx, fresh, patch)).Should(Succeed()) + opsRes.OpsRequest = fresh + + actionCalls := 0 + testapps.MockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) { + recorder.Action(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(func(context.Context, kbagentproto.ActionRequest) (kbagentproto.ActionResponse, error) { + actionCalls++ + return kbagentproto.ActionResponse{}, fmt.Errorf("unexpected redispatch") + }) + }) + + _, completedCount, failedCount, err := handleSwitchovers(reqCtx, k8sClient, opsRes) + Expect(err).ShouldNot(HaveOccurred()) + Expect(actionCalls).Should(Equal(0)) + Expect(completedCount).Should(Equal(int32(1))) + Expect(failedCount).Should(Equal(int32(1))) + + Expect(k8sClient.Get(ctx, key, fresh)).Should(Succeed()) + progressDetail = findStatusProgressDetail(fresh.Status.Components[defaultCompName].ProgressDetails, + getProgressObjectKey(KBSwitchoverKey, defaultCompName)) + Expect(progressDetail.Status).Should(Equal(opsv1alpha1.FailedProgressStatus)) + Expect(progressDetail.Message).Should(ContainSubstring("outcome is unknown")) + }) + + It("continues candidate role observation without redispatch after restart with a retained dispatch claim", func() { + candidateName := fmt.Sprintf("%s-%s-%d", clusterObj.Name, defaultCompName, 0) + key, _ := preparePendingSwitchover(candidateName) + fresh := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, key, fresh)).Should(Succeed()) + patch := client.MergeFrom(fresh.DeepCopy()) + progressDetail := findStatusProgressDetail(fresh.Status.Components[defaultCompName].ProgressDetails, + getProgressObjectKey(KBSwitchoverKey, defaultCompName)) + Expect(progressDetail).ShouldNot(BeNil()) + progressDetail.Status = opsv1alpha1.ProcessingProgressStatus + progressDetail.Message = switchoverDispatchClaimMessageForTest(fresh, defaultCompName, + fresh.Spec.SwitchoverList[0], "restart-with-candidate-token") + progressDetail.StartTime = metav1.Now() + Expect(k8sClient.Status().Patch(ctx, fresh, patch)).Should(Succeed()) + opsRes.OpsRequest = fresh + + candidatePod := &corev1.Pod{} + Expect(k8sClient.Get(ctx, client.ObjectKey{Namespace: testCtx.DefaultNamespace, Name: candidateName}, candidatePod)).Should(Succeed()) + candidatePod.Labels[constant.RoleLabelKey] = "unexpected-role" + Expect(k8sClient.Update(ctx, candidatePod)).Should(Succeed()) + + actionCalls := 0 + testapps.MockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) { + recorder.Action(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(func(context.Context, kbagentproto.ActionRequest) (kbagentproto.ActionResponse, error) { + actionCalls++ + return kbagentproto.ActionResponse{}, fmt.Errorf("unexpected redispatch") + }) + }) + + _, completedCount, failedCount, err := handleSwitchovers(reqCtx, k8sClient, opsRes) + Expect(err).ShouldNot(HaveOccurred()) + Expect(actionCalls).Should(Equal(0)) + Expect(completedCount).Should(Equal(int32(0))) + Expect(failedCount).Should(Equal(int32(0))) + + Expect(k8sClient.Get(ctx, key, fresh)).Should(Succeed()) + progressDetail = findStatusProgressDetail(fresh.Status.Components[defaultCompName].ProgressDetails, + getProgressObjectKey(KBSwitchoverKey, defaultCompName)) + Expect(progressDetail.Status).Should(Equal(opsv1alpha1.ProcessingProgressStatus)) + Expect(progressDetail.Message).Should(ContainSubstring("waiting for candidate pod")) + Expect(progressDetail.Message).Should(ContainSubstring("unexpected-role")) + }) + + It("fails closed when a persisted dispatch claim no longer matches the request identity", func() { + key, _ := preparePendingSwitchover("") + fresh := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, key, fresh)).Should(Succeed()) + patch := client.MergeFrom(fresh.DeepCopy()) + progressDetail := findStatusProgressDetail(fresh.Status.Components[defaultCompName].ProgressDetails, + getProgressObjectKey(KBSwitchoverKey, defaultCompName)) + Expect(progressDetail).ShouldNot(BeNil()) + progressDetail.Status = opsv1alpha1.ProcessingProgressStatus + progressDetail.Message = switchoverDispatchClaimMessagePrefix + "different-request/default/instance/candidate/foreign-token" + progressDetail.StartTime = metav1.Now() + Expect(k8sClient.Status().Patch(ctx, fresh, patch)).Should(Succeed()) + opsRes.OpsRequest = fresh + + actionCalls := 0 + testapps.MockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) { + recorder.Action(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(func(context.Context, kbagentproto.ActionRequest) (kbagentproto.ActionResponse, error) { + actionCalls++ + return kbagentproto.ActionResponse{}, fmt.Errorf("unexpected redispatch") + }) + }) + + _, completedCount, failedCount, err := handleSwitchovers(reqCtx, k8sClient, opsRes) + Expect(err).ShouldNot(HaveOccurred()) + Expect(actionCalls).Should(Equal(0)) + Expect(completedCount).Should(Equal(int32(1))) + Expect(failedCount).Should(Equal(int32(1))) + Expect(k8sClient.Get(ctx, key, fresh)).Should(Succeed()) + progressDetail = findStatusProgressDetail(fresh.Status.Components[defaultCompName].ProgressDetails, + getProgressObjectKey(KBSwitchoverKey, defaultCompName)) + Expect(progressDetail.Message).Should(ContainSubstring("protocol identity changed or is malformed")) + }) + + DescribeTable("fails closed after restart on invalid persisted outcome markers", func(buildMessage func(*opsv1alpha1.OpsRequest) string) { + key, _ := preparePendingSwitchover("") + fresh := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, key, fresh)).Should(Succeed()) + patch := client.MergeFrom(fresh.DeepCopy()) + progressDetail := findStatusProgressDetail(fresh.Status.Components[defaultCompName].ProgressDetails, + getProgressObjectKey(KBSwitchoverKey, defaultCompName)) + Expect(progressDetail).ShouldNot(BeNil()) + progressDetail.Status = opsv1alpha1.ProcessingProgressStatus + progressDetail.Message = buildMessage(fresh) + progressDetail.StartTime = metav1.Now() + Expect(k8sClient.Status().Patch(ctx, fresh, patch)).Should(Succeed()) + opsRes.OpsRequest = fresh + + actionCalls := 0 + testapps.MockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) { + recorder.Action(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(func(context.Context, kbagentproto.ActionRequest) (kbagentproto.ActionResponse, error) { + actionCalls++ + return kbagentproto.ActionResponse{}, fmt.Errorf("unexpected redispatch") + }) + }) + + _, completedCount, failedCount, err := handleSwitchovers(reqCtx, k8sClient, opsRes) + Expect(err).ShouldNot(HaveOccurred()) + Expect(completedCount).Should(Equal(int32(1))) + Expect(failedCount).Should(Equal(int32(1))) + Expect(actionCalls).Should(Equal(0)) + + Expect(k8sClient.Get(ctx, key, fresh)).Should(Succeed()) + progressDetail = findStatusProgressDetail(fresh.Status.Components[defaultCompName].ProgressDetails, + getProgressObjectKey(KBSwitchoverKey, defaultCompName)) + Expect(progressDetail.Status).Should(Equal(opsv1alpha1.FailedProgressStatus)) + Expect(progressDetail.Message).Should(ContainSubstring("protocol identity changed or is malformed")) + }, + Entry("malformed outcome", func(*opsv1alpha1.OpsRequest) string { + return switchoverDispatchOutcomeMessagePrefix + "malformed" + }), + Entry("foreign outcome identity and token", func(fresh *opsv1alpha1.OpsRequest) string { + claim := switchoverDispatchClaim{ + opsRequestUID: "different-request", + componentName: defaultCompName, + instanceName: fresh.Spec.SwitchoverList[0].InstanceName, + token: "foreign-token", + } + return claim.outcomeMessage("doing switchover") + }), + ) + + It("does not call the lifecycle action without a live API reader", func() { + key, _ := preparePendingSwitchover("") + before := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, key, before)).Should(Succeed()) + opsRes.APIReader = nil + actionCalls := 0 + testapps.MockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) { + recorder.Action(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(func(context.Context, kbagentproto.ActionRequest) (kbagentproto.ActionResponse, error) { + actionCalls++ + return kbagentproto.ActionResponse{}, fmt.Errorf("unexpected dispatch") + }) + }) + + _, _, _, err := handleSwitchovers(reqCtx, k8sClient, opsRes) + Expect(err).Should(MatchError("APIReader is required to confirm a switchover dispatch claim")) + Expect(actionCalls).Should(Equal(0)) + + after := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, key, after)).Should(Succeed()) + Expect(after.ResourceVersion).Should(Equal(before.ResourceVersion)) + Expect(after.Status).Should(Equal(before.Status)) + }) + + It("does not write a dispatch claim without a complete OpsRequest UID", func() { + key, _ := preparePendingSwitchover("") + before := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, key, before)).Should(Succeed()) + opsRes.OpsRequest.UID = "" + actionCalls := 0 + testapps.MockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) { + recorder.Action(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(func(context.Context, kbagentproto.ActionRequest) (kbagentproto.ActionResponse, error) { + actionCalls++ + return kbagentproto.ActionResponse{}, fmt.Errorf("unexpected dispatch") + }) + }) + + _, _, _, err := handleSwitchovers(reqCtx, k8sClient, opsRes) + Expect(err).Should(MatchError("OpsRequest UID is required to create a switchover dispatch claim")) + Expect(actionCalls).Should(Equal(0)) + + after := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, key, after)).Should(Succeed()) + Expect(after.ResourceVersion).Should(Equal(before.ResourceVersion)) + Expect(after.Status).Should(Equal(before.Status)) + }) + + It("refetches and retries a known successful outcome after a status conflict", func() { + key, _ := preparePendingSwitchover("") + actionCalls := 0 + testapps.MockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) { + recorder.Action(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(func(context.Context, kbagentproto.ActionRequest) (kbagentproto.ActionResponse, error) { + actionCalls++ + return kbagentproto.ActionResponse{Message: "mock success"}, nil + }) + }) + + conflicted := false + conflictingClient := &interceptStatusClient{ + Client: k8sClient, + patch: func(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error { + if _, ok := obj.(*opsv1alpha1.OpsRequest); ok && actionCalls > 0 && !conflicted { + conflicted = true + concurrent := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, key, concurrent)).Should(Succeed()) + concurrentPatch := client.MergeFromWithOptions(concurrent.DeepCopy(), client.MergeFromWithOptimisticLock{}) + concurrent.Status.Progress = "99/99" + Expect(k8sClient.Status().Patch(ctx, concurrent, concurrentPatch)).Should(Succeed()) + conflictErr := k8sClient.Status().Patch(ctx, obj, patch, opts...) + Expect(apierrors.IsConflict(conflictErr)).Should(BeTrue()) + return conflictErr + } + return k8sClient.Status().Patch(ctx, obj, patch, opts...) + }, + } + + _, _, _, err := handleSwitchovers(reqCtx, conflictingClient, opsRes) + Expect(err).ShouldNot(HaveOccurred()) + Expect(actionCalls).Should(Equal(1)) + Expect(conflicted).Should(BeTrue()) + + fresh := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, key, fresh)).Should(Succeed()) + progressDetail := findStatusProgressDetail(fresh.Status.Components[defaultCompName].ProgressDetails, + getProgressObjectKey(KBSwitchoverKey, defaultCompName)) + Expect(progressDetail).ShouldNot(BeNil()) + Expect(progressDetail.Status).Should(Equal(opsv1alpha1.ProcessingProgressStatus)) + outcomeClaim, outcome, ok := parseSwitchoverDispatchOutcomeMessage(progressDetail.Message) + Expect(ok).Should(BeTrue()) + Expect(outcomeClaim.matchesIdentity(fresh, defaultCompName, fresh.Spec.SwitchoverList[0])).Should(BeTrue()) + Expect(outcomeClaim.token).ShouldNot(BeEmpty()) + Expect(outcome).Should(Equal("doing switchover")) + }) + + It("keeps retrying a known failed outcome beyond the default bounded backoff", func() { + key, _ := preparePendingSwitchover("") + actionCalls := 0 + testapps.MockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) { + recorder.Action(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(func(context.Context, kbagentproto.ActionRequest) (kbagentproto.ActionResponse, error) { + actionCalls++ + return kbagentproto.ActionResponse{}, fmt.Errorf("injected lifecycle failure") + }) + }) + + injectedErr := fmt.Errorf("injected failed-outcome status failure") + failuresRemaining := retry.DefaultBackoff.Steps + 2 + failedAttempts := 0 + failingClient := &interceptStatusClient{ + Client: k8sClient, + patch: func(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error { + if _, ok := obj.(*opsv1alpha1.OpsRequest); ok && actionCalls > 0 && failuresRemaining > 0 { + failuresRemaining-- + failedAttempts++ + return injectedErr + } + return k8sClient.Status().Patch(ctx, obj, patch, opts...) + }, + } + + _, completedCount, failedCount, err := handleSwitchovers(reqCtx, failingClient, opsRes) + Expect(err).ShouldNot(HaveOccurred()) + Expect(actionCalls).Should(Equal(1)) + Expect(failedAttempts).Should(Equal(retry.DefaultBackoff.Steps + 2)) + Expect(completedCount).Should(Equal(int32(1))) + Expect(failedCount).Should(Equal(int32(1))) + + fresh := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, key, fresh)).Should(Succeed()) + progressDetail := findStatusProgressDetail(fresh.Status.Components[defaultCompName].ProgressDetails, + getProgressObjectKey(KBSwitchoverKey, defaultCompName)) + Expect(progressDetail).ShouldNot(BeNil()) + Expect(progressDetail.Status).Should(Equal(opsv1alpha1.FailedProgressStatus)) + outcomeClaim, outcome, ok := parseSwitchoverDispatchOutcomeMessage(progressDetail.Message) + Expect(ok).Should(BeTrue()) + Expect(outcomeClaim.matchesIdentity(fresh, defaultCompName, fresh.Spec.SwitchoverList[0])).Should(BeTrue()) + Expect(outcome).Should(ContainSubstring("injected lifecycle failure")) + }) + + It("stops retrying a known outcome when the caller context ends", func() { + retryCtx, cancel := context.WithCancel(context.Background()) + calls := 0 + err := retryKnownSwitchoverOutcomeUntilContextDone(retryCtx, func() error { + calls++ + if calls == 2 { + cancel() + } + return fmt.Errorf("injected retryable status failure") + }) + + Expect(err).Should(MatchError(ContainSubstring("context canceled"))) + Expect(calls).Should(Equal(2)) + }) + + DescribeTable("does not retry a known outcome after a permanent API response", func(statusCode int) { + calls := 0 + permanentErr := apierrors.NewGenericServerResponse(statusCode, "PATCH", schema.GroupResource{ + Group: "apps.kubeblocks.io", Resource: "opsrequests", + }, "ops", "injected permanent status failure", 0, false) + err := retryKnownSwitchoverOutcomeUntilContextDone(context.Background(), func() error { + calls++ + return permanentErr + }) + + Expect(err).Should(MatchError(permanentErr)) + Expect(calls).Should(Equal(1)) + }, + Entry("413 request entity too large", http.StatusRequestEntityTooLarge), + Entry("400 bad request", http.StatusBadRequest), + Entry("405 method not supported", http.StatusMethodNotAllowed), + Entry("415 unsupported media type", http.StatusUnsupportedMediaType), + Entry("406 not acceptable", http.StatusNotAcceptable), + ) + + It("accepts a known outcome after its committed status response is lost", func() { + key, _ := preparePendingSwitchover("") + actionCalls := 0 + testapps.MockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) { + recorder.Action(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(func(context.Context, kbagentproto.ActionRequest) (kbagentproto.ActionResponse, error) { + actionCalls++ + return kbagentproto.ActionResponse{Message: "mock success"}, nil + }) + }) + + injectedErr := fmt.Errorf("injected committed status response loss") + lost := false + responseLossClient := &interceptStatusClient{ + Client: k8sClient, + patch: func(ctx context.Context, obj client.Object, patch client.Patch, opts ...client.SubResourcePatchOption) error { + if _, ok := obj.(*opsv1alpha1.OpsRequest); ok && actionCalls > 0 && !lost { + lost = true + Expect(k8sClient.Status().Patch(ctx, obj, patch, opts...)).Should(Succeed()) + return injectedErr + } + return k8sClient.Status().Patch(ctx, obj, patch, opts...) + }, } + + _, _, _, err := handleSwitchovers(reqCtx, responseLossClient, opsRes) + Expect(err).ShouldNot(HaveOccurred()) + Expect(actionCalls).Should(Equal(1)) + Expect(lost).Should(BeTrue()) + + fresh := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, key, fresh)).Should(Succeed()) + progressDetail := findStatusProgressDetail(fresh.Status.Components[defaultCompName].ProgressDetails, + getProgressObjectKey(KBSwitchoverKey, defaultCompName)) + Expect(progressDetail).ShouldNot(BeNil()) + Expect(progressDetail.Status).Should(Equal(opsv1alpha1.ProcessingProgressStatus)) + outcomeClaim, outcome, ok := parseSwitchoverDispatchOutcomeMessage(progressDetail.Message) + Expect(ok).Should(BeTrue()) + Expect(outcomeClaim.matchesIdentity(fresh, defaultCompName, fresh.Spec.SwitchoverList[0])).Should(BeTrue()) + Expect(outcome).Should(Equal("doing switchover")) + }) + + It("does not accept matching status and outcome text without the exact token", func() { + key, _ := preparePendingSwitchover("") + actionCalls := 0 + testapps.MockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) { + recorder.Action(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(func(ctx context.Context, req kbagentproto.ActionRequest) (kbagentproto.ActionResponse, error) { + actionCalls++ + fresh := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, key, fresh)).Should(Succeed()) + patch := client.MergeFromWithOptions(fresh.DeepCopy(), client.MergeFromWithOptimisticLock{}) + progressDetail := findStatusProgressDetail(fresh.Status.Components[defaultCompName].ProgressDetails, + getProgressObjectKey(KBSwitchoverKey, defaultCompName)) + Expect(progressDetail).ShouldNot(BeNil()) + progressDetail.Status = opsv1alpha1.ProcessingProgressStatus + progressDetail.Message = "doing switchover" + Expect(k8sClient.Status().Patch(ctx, fresh, patch)).Should(Succeed()) + return kbagentproto.ActionResponse{Message: "mock success"}, nil + }) + }) + + _, _, _, err := handleSwitchovers(reqCtx, k8sClient, opsRes) + Expect(err).Should(MatchError(ContainSubstring("switchover dispatch claim was lost"))) + Expect(actionCalls).Should(Equal(1)) + + fresh := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, key, fresh)).Should(Succeed()) + progressDetail := findStatusProgressDetail(fresh.Status.Components[defaultCompName].ProgressDetails, + getProgressObjectKey(KBSwitchoverKey, defaultCompName)) + Expect(progressDetail).ShouldNot(BeNil()) + Expect(progressDetail.Status).Should(Equal(opsv1alpha1.ProcessingProgressStatus)) + Expect(progressDetail.Message).Should(Equal("doing switchover")) + }) + + It("does not accept a matching failed outcome persisted by a foreign token", func() { + key, _ := preparePendingSwitchover("") + actionCalls := 0 + foreignMessage := "" + testapps.MockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) { + recorder.Action(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(func(ctx context.Context, req kbagentproto.ActionRequest) (kbagentproto.ActionResponse, error) { + actionCalls++ + fresh := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, key, fresh)).Should(Succeed()) + patch := client.MergeFromWithOptions(fresh.DeepCopy(), client.MergeFromWithOptimisticLock{}) + progressDetail := findStatusProgressDetail(fresh.Status.Components[defaultCompName].ProgressDetails, + getProgressObjectKey(KBSwitchoverKey, defaultCompName)) + Expect(progressDetail).ShouldNot(BeNil()) + foreignClaim, ok := parseSwitchoverDispatchClaim(progressDetail.Message) + Expect(ok).Should(BeTrue()) + foreignClaim.token = "foreign-token" + foreignMessage = foreignClaim.outcomeMessage("component " + defaultCompName + " injected lifecycle failure") + progressDetail.Status = opsv1alpha1.FailedProgressStatus + progressDetail.Message = foreignMessage + Expect(k8sClient.Status().Patch(ctx, fresh, patch)).Should(Succeed()) + return kbagentproto.ActionResponse{}, fmt.Errorf("injected lifecycle failure") + }) + }) + + _, _, _, err := handleSwitchovers(reqCtx, k8sClient, opsRes) + Expect(err).Should(MatchError(ContainSubstring("switchover dispatch claim was lost"))) + Expect(actionCalls).Should(Equal(1)) + + fresh := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, key, fresh)).Should(Succeed()) + progressDetail := findStatusProgressDetail(fresh.Status.Components[defaultCompName].ProgressDetails, + getProgressObjectKey(KBSwitchoverKey, defaultCompName)) + Expect(progressDetail).ShouldNot(BeNil()) + Expect(progressDetail.Status).Should(Equal(opsv1alpha1.FailedProgressStatus)) + Expect(progressDetail.Message).Should(Equal(foreignMessage)) + }) + + It("does not overwrite a foreign token while persisting a known outcome", func() { + key, _ := preparePendingSwitchover("") + actionCalls := 0 + foreignMessage := "" + testapps.MockKBAgentClient(func(recorder *kbacli.MockClientMockRecorder) { + recorder.Action(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(func(ctx context.Context, req kbagentproto.ActionRequest) (kbagentproto.ActionResponse, error) { + actionCalls++ + fresh := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, key, fresh)).Should(Succeed()) + patch := client.MergeFromWithOptions(fresh.DeepCopy(), client.MergeFromWithOptimisticLock{}) + progressDetail := findStatusProgressDetail(fresh.Status.Components[defaultCompName].ProgressDetails, + getProgressObjectKey(KBSwitchoverKey, defaultCompName)) + Expect(progressDetail).ShouldNot(BeNil()) + foreignMessage = switchoverDispatchClaimMessageForTest(fresh, defaultCompName, + fresh.Spec.SwitchoverList[0], "foreign-token") + progressDetail.Message = foreignMessage + Expect(k8sClient.Status().Patch(ctx, fresh, patch)).Should(Succeed()) + return kbagentproto.ActionResponse{Message: "mock success"}, nil + }) + }) + + _, _, _, err := handleSwitchovers(reqCtx, k8sClient, opsRes) + Expect(err).Should(MatchError(ContainSubstring("switchover dispatch claim was lost"))) + Expect(actionCalls).Should(Equal(1)) + + fresh := &opsv1alpha1.OpsRequest{} + Expect(k8sClient.Get(ctx, key, fresh)).Should(Succeed()) + progressDetail := findStatusProgressDetail(fresh.Status.Components[defaultCompName].ProgressDetails, + getProgressObjectKey(KBSwitchoverKey, defaultCompName)) + Expect(progressDetail).ShouldNot(BeNil()) + Expect(progressDetail.Status).Should(Equal(opsv1alpha1.ProcessingProgressStatus)) + Expect(progressDetail.Message).Should(Equal(foreignMessage)) }) It("Test switchover OpsRequest", func() { diff --git a/pkg/operations/type.go b/pkg/operations/type.go index 6ad37987028..38aa9dfc258 100644 --- a/pkg/operations/type.go +++ b/pkg/operations/type.go @@ -82,6 +82,7 @@ type OpsResource struct { OpsRequest *opsv1alpha1.OpsRequest Cluster *appsv1.Cluster Recorder record.EventRecorder + APIReader client.Reader ToClusterPhase appsv1.ClusterPhase Runtimes map[string]OpsRuntime }