diff --git a/Dockerfile b/Dockerfile index f8384bac..6457e1f5 100644 --- a/Dockerfile +++ b/Dockerfile @@ -4,8 +4,6 @@ ARG GOPROXY="https://proxy.golang.org,direct" WORKDIR /workspace -RUN go env -w GOMODCACHE=/root/.cache/go-build - RUN go env -w GOPROXY=${GOPROXY} COPY go.mod go.mod @@ -15,12 +13,14 @@ COPY api/go.sum api/go.sum # cache deps before building and copying source so that we don't need to re-download as much # and so that source changes don't invalidate our downloaded layer -RUN --mount=type=cache,target=/root/.cache/go-build go mod download +RUN --mount=type=cache,id=gomodcache,target=/go/pkg/mod go mod download COPY . . # Build -RUN --mount=type=cache,target=/root/.cache/go-build CGO_ENABLED=0 go build -a -o manager cmd/operator/main.go +RUN --mount=type=cache,id=gomodcache,target=/go/pkg/mod \ + --mount=type=cache,id=gobuildcache,target=/root/.cache/go-build \ + CGO_ENABLED=0 go build -a -o manager cmd/operator/main.go # Use distroless as minimal base image to package the manager binary # Refer to https://github.com/GoogleContainerTools/distroless for more details diff --git a/api/go.mod b/api/go.mod index 41333099..bdcf24aa 100644 --- a/api/go.mod +++ b/api/go.mod @@ -5,6 +5,7 @@ go 1.19 require ( github.com/blang/semver/v4 v4.0.0 github.com/go-errors/errors v1.5.1 + github.com/google/go-cmp v0.6.0 github.com/matrixorigin/controller-runtime v0.0.0-20240909085031-5f706d779ec6 github.com/mohae/deepcopy v0.0.0-20170929034955-c48cc78d4826 github.com/onsi/gomega v1.27.7 @@ -31,7 +32,6 @@ require ( github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect github.com/golang/protobuf v1.5.4 // indirect github.com/google/gnostic v0.5.7-v3refs // indirect - github.com/google/go-cmp v0.6.0 // indirect github.com/google/gofuzz v1.2.0 // indirect github.com/google/pprof v0.0.0-20230510103437-eeec1cb781c3 // indirect github.com/google/uuid v1.6.0 // indirect diff --git a/hack/lib.sh b/hack/lib.sh index b6793f17..ce51c8f2 100644 --- a/hack/lib.sh +++ b/hack/lib.sh @@ -83,14 +83,21 @@ function kind::ensure-kind() { function kind::load-image() { local kruise_image + local kruise_hook_image kruise_image=$(helm template kruise "${ROOT}/charts/kruise" | awk '/^[[:space:]]+image:.*kruise-manager/ {print $2; exit}') if [[ -z "${kruise_image}" ]]; then echo "error: failed to resolve Kruise manager image from local chart" return 1 fi + kruise_hook_image=$(helm template kruise "${ROOT}/charts/kruise" | awk '/^[[:space:]]+image:.*kruise-helm-hook/ {print $2; exit}') + if [[ -z "${kruise_hook_image}" ]]; then + echo "error: failed to resolve Kruise Helm hook image from local chart" + return 1 + fi kind::prepare_image ${CLUSTER} ${MO_IMAGE_REPO}:${MO_VERSION} kind::prepare_image ${CLUSTER} "${kruise_image}" + kind::prepare_image ${CLUSTER} "${kruise_hook_image}" kind::prepare_image ${CLUSTER} minio/minio:RELEASE.2023-11-01T01-57-10Z } @@ -111,9 +118,10 @@ function e2e::check() { } function e2e::run() { + local nodes="${E2E_NODES:-4}" echo "> Run e2e test" make ginkgo - ./bin/ginkgo -nodes=4 -stream=true -slowSpecThreshold=3000 ./test/e2e/... -- \ + ./bin/ginkgo -nodes="${nodes}" -stream=true -slowSpecThreshold=3000 ./test/e2e/... -- \ -mo-version="${MO_VERSION}" \ -mo-image-repo="${MO_IMAGE_REPO}" @@ -140,16 +148,62 @@ function e2e::install() { fi rm -rf -- "${chart_root}" - echo "> Wait webhook certificate injected" - sleep 30 + e2e::wait-webhook-ready +} + +function e2e::wait-webhook-ready() { + local selector="app.kubernetes.io/name=matrixone-operator,app.kubernetes.io/instance=mo" + local mutating="matrixone-operator-mutating-webhook-mo" + local validating="matrixone-operator-validating-webhook-mo" + local timeout_seconds=300 + local deadline + + echo "> Wait for operator deployment" + if ! kubectl -n "${OPNAMESPACE}" wait deployment \ + -l "${selector}" \ + --for=condition=Available \ + --timeout="${timeout_seconds}s"; then + kubectl -n "${OPNAMESPACE}" get pods -o wide || true + return 1 + fi + + echo "> Wait for webhook CA injection" + deadline=$((SECONDS + timeout_seconds)) + while ((SECONDS < deadline)); do + local mutating_ca + local validating_ca + mutating_ca=$(kubectl get mutatingwebhookconfiguration "${mutating}" \ + -o jsonpath='{.webhooks[0].clientConfig.caBundle}' 2>/dev/null || true) + validating_ca=$(kubectl get validatingwebhookconfiguration "${validating}" \ + -o jsonpath='{.webhooks[0].clientConfig.caBundle}' 2>/dev/null || true) + if [[ -n "${mutating_ca}" && "${mutating_ca}" != "Cg==" && \ + -n "${validating_ca}" && "${validating_ca}" != "Cg==" ]]; then + echo "> Webhook CA injection completed" + return 0 + fi + sleep 2 + done + + echo "error: webhook CA injection did not complete within ${timeout_seconds}s" + kubectl get mutatingwebhookconfiguration "${mutating}" -o yaml || true + kubectl get validatingwebhookconfiguration "${validating}" -o yaml || true + kubectl -n "${OPNAMESPACE}" logs deployment/mo-matrixone-operator --tail=100 || true + return 1 } function e2e::cleanup() { echo "Delete e2e test namespace" - kubectl get ns --all-namespaces --no-headers=true | awk '/^e2e/{print $1}' | xargs kubectl delete ns + if ! kubectl delete namespace -l managed-by=e2e-suite \ + --ignore-not-found --wait=true --timeout=600s; then + kubectl get namespace -l managed-by=e2e-suite -o yaml || true + return 1 + fi # Uninstall helm charts echo "Uninstall helm charts..." - helm uninstall mo -n "${OPNAMESPACE}" + if ! helm uninstall mo -n "${OPNAMESPACE}"; then + kubectl -n kruise-system logs job/mo-finalizer --all-containers=true || true + return 1 + fi echo "Wait for charts uninstall" sleep 10 echo "Delete operator namespace" @@ -160,5 +214,9 @@ function e2e::workflow() { e2e::check trap "e2e::cleanup" EXIT e2e::install || return 1 - e2e::run + local run_status=0 + e2e::run || run_status=$? + trap - EXIT + e2e::cleanup || return 1 + return "${run_status}" } diff --git a/pkg/controllers/cnclaim/controller.go b/pkg/controllers/cnclaim/controller.go index 79de36c6..402fbd06 100644 --- a/pkg/controllers/cnclaim/controller.go +++ b/pkg/controllers/cnclaim/controller.go @@ -37,12 +37,15 @@ import ( "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/handler" + ctrllog "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/manager" "sigs.k8s.io/controller-runtime/pkg/predicate" "sigs.k8s.io/controller-runtime/pkg/reconcile" ) const ( + claimPodNameField = "spec.podName" + waitCacheTimeout = 10 * time.Second retryBindInterval = 5 * time.Second @@ -60,6 +63,12 @@ func NewActor(mgr *mocli.MORPCClientManager) *Actor { } func (r *Actor) Observe(ctx *recon.Context[*v1alpha1.CNClaim]) (recon.Action[*v1alpha1.CNClaim], error) { + // Lost is terminal. In particular, do not race a CNClaimSet deleting a lost + // claim by binding the claim to another Pod after its stale reference is + // cleared. Automatic recovery, if desired, needs an explicit state transition. + if ctx.Obj.Status.Phase == v1alpha1.CNClaimPhaseLost { + return nil, nil + } if ctx.Obj.Spec.PodName == "" { return r.Bind, nil } @@ -159,7 +168,7 @@ func (r *Actor) selectCN(ctx *recon.Context[*v1alpha1.CNClaim], orphans []corev1 } sortCNByPriority(c, idleCNs) - // build index once: podName -> claimName for all other CNClaims + // Build an index once for every Pod reference held by other CNClaims. claimIndex, err := buildPodClaimIndex(ctx, c.Namespace, c.Name) if err != nil { return nil, errors.WrapPrefix(err, "error building pod claim index", 0) @@ -167,8 +176,8 @@ func (r *Actor) selectCN(ctx *recon.Context[*v1alpha1.CNClaim], orphans []corev1 for i := range idleCNs { pod := &idleCNs[i] // skip pod already referenced by another CNClaim's spec.podName - if holder, ok := claimIndex[pod.Name]; ok { - ctx.Log.Info("skip pod claimed by other CNClaim", "podName", pod.Name, "holder", holder) + if holders := claimIndex[pod.Name]; len(holders) > 0 { + ctx.Log.Info("skip pod claimed by other CNClaim", "podName", pod.Name, "holders", claimNames(holders)) continue } if err := r.ensureOwnership(ctx, pod); err != nil { @@ -265,9 +274,19 @@ func (r *Actor) Sync(ctx *recon.Context[*v1alpha1.CNClaim]) error { if c.Status.BoundTime != nil && time.Since(c.Status.BoundTime.Time) < waitCacheTimeout { return recon.ErrReSync("pod status may be not update to date, wait", waitCacheTimeout) } + if err := ctx.Patch(c, func() error { + c.Spec.PodName = "" + c.Spec.NodeName = "" + return nil + }); err != nil { + return errors.WrapPrefix(err, "error clearing lost claim spec", 0) + } + // Patch refreshes c with the API response, including the previously + // persisted status, so set Lost only after the spec patch completes. c.Status.Phase = v1alpha1.CNClaimPhaseLost return nil } + return errors.WrapPrefix(err, "error get claimed Pod", 0) } if pod.Status.Phase == corev1.PodSucceeded || pod.Status.Phase == corev1.PodUnknown { c.Status.Phase = v1alpha1.CNClaimPhaseLost @@ -350,16 +369,31 @@ func (r *Actor) Finalize(ctx *recon.Context[*v1alpha1.CNClaim]) (bool, error) { if len(ownedCNs) == 0 { return true, nil } - // build index once: podName -> claimName for all other CNClaims + // Build an index once for every Pod reference held by other CNClaims. claimIndex, err := buildPodClaimIndex(ctx, c.Namespace, c.Name) if err != nil { return false, errors.WrapPrefix(err, "error building pod claim index", 0) } for i := range ownedCNs { cn := ownedCNs[i] - // skip reclaim if another CNClaim still references this pod via spec.podName - if holder, ok := claimIndex[cn.Name]; ok { - ctx.Log.Info("skip reclaim, pod still claimed by other CNClaim", "pod", cn.Name, "holder", holder) + holders := claimIndex[cn.Name] + if len(holders) > 1 { + return false, errors.Errorf("cannot transfer pod %s ownership: multiple CNClaims reference it: %v", cn.Name, claimNames(holders)) + } + // If another CNClaim references this pod via spec.podName, transfer all + // claim-managed Pod labels directly. Removing only claimed-by would leave + // stale ownership metadata, and label-only changes do not trigger this + // controller to let the surviving claim repair them later. + if len(holders) == 1 { + holder := &holders[0] + if !holder.IsReady() { + ctx.Log.Info("wait for target CNClaim before transferring pod ownership", "pod", cn.Name, "holder", holder.Name, "phase", holder.Status.Phase) + return false, nil + } + ctx.Log.Info("transfer pod ownership to other CNClaim", "pod", cn.Name, "holder", holder.Name) + if err := transferPodOwnership(ctx, &cn, c, holder); err != nil { + return false, errors.WrapPrefix(err, "error transferring pod ownership", 0) + } continue } ctx.Log.Info("finalize CNClaim, reclaim bound CN", "cn", cn.Name) @@ -367,9 +401,44 @@ func (r *Actor) Finalize(ctx *recon.Context[*v1alpha1.CNClaim]) (bool, error) { return false, err } } + // Keep the finalizer for one more reconciliation so the cached Pod list can + // confirm that no Bound Pod is still labelled as owned by this claim. return false, nil } +func transferPodOwnership( + cli recon.KubeClient, + pod *corev1.Pod, + from, to *v1alpha1.CNClaim, +) error { + return cli.Patch(pod, func() error { + if pod.Labels == nil { + pod.Labels = map[string]string{} + } + + for key := range from.Spec.AdditionalPodLabels { + delete(pod.Labels, key) + } + for key, value := range to.Spec.AdditionalPodLabels { + pod.Labels[key] = value + } + + pod.Labels[v1alpha1.CNPodPhaseLabel] = v1alpha1.CNPodPhaseBound + pod.Labels[v1alpha1.PodClaimedByLabel] = to.Name + if claimSetName := to.Labels[v1alpha1.ClaimSetNameLabel]; claimSetName != "" { + pod.Labels[v1alpha1.ClaimSetNameLabel] = claimSetName + } else { + delete(pod.Labels, v1alpha1.ClaimSetNameLabel) + } + if to.Spec.OwnerName != nil { + pod.Labels[v1alpha1.PodOwnerNameLabel] = *to.Spec.OwnerName + } else { + delete(pod.Labels, v1alpha1.PodOwnerNameLabel) + } + return nil + }) +} + // podClaimedByOthers checks if the given pod is referenced by any CNClaim's // spec.podName other than excludeClaim in the same namespace. // It skips CNClaims that are being deleted (DeletionTimestamp != nil). @@ -391,24 +460,34 @@ func podClaimedByOthers(cli recon.KubeClient, namespace, podName, excludeClaim s } // buildPodClaimIndex lists all CNClaims in the namespace and returns a map -// from podName to the claiming CNClaim name, excluding the given claim and -// CNClaims that are being deleted. -func buildPodClaimIndex(cli recon.KubeClient, namespace, excludeClaim string) (map[string]string, error) { +// from podName to the CNClaims that reference it, excluding the given claim +// and CNClaims that are being deleted. Keeping every claimant lets callers +// detect an ambiguous ownership transfer instead of choosing one arbitrarily. +func buildPodClaimIndex(cli recon.KubeClient, namespace, excludeClaim string) (map[string][]v1alpha1.CNClaim, error) { claimList := &v1alpha1.CNClaimList{} if err := cli.List(claimList, client.InNamespace(namespace)); err != nil { return nil, err } - index := make(map[string]string, len(claimList.Items)) + index := make(map[string][]v1alpha1.CNClaim, len(claimList.Items)) for i := range claimList.Items { claim := &claimList.Items[i] if claim.Name == excludeClaim || claim.DeletionTimestamp != nil || claim.Spec.PodName == "" { continue } - index[claim.Spec.PodName] = claim.Name + index[claim.Spec.PodName] = append(index[claim.Spec.PodName], *claim) } return index, nil } +func claimNames(claims []v1alpha1.CNClaim) []string { + names := make([]string, 0, len(claims)) + for i := range claims { + names = append(names, claims[i].Name) + } + slices.Sort(names) + return names +} + func (r *Actor) patchStore(ctx *recon.Context[*v1alpha1.CNClaim], pod *corev1.Pod, req logpb.CNStateLabel) (*metadata.CNService, error) { cs, err := common.ResolveCNSet(ctx, pod) if err != nil { @@ -451,29 +530,74 @@ func (r *Actor) patchStore(ctx *recon.Context[*v1alpha1.CNClaim], pod *corev1.Po } func (r *Actor) Start(mgr manager.Manager) error { + if err := mgr.GetFieldIndexer().IndexField(context.Background(), &v1alpha1.CNClaim{}, claimPodNameField, indexClaimByPodName); err != nil { + return errors.WrapPrefix(err, "error indexing CNClaims by pod name", 0) + } return recon.Setup(&v1alpha1.CNClaim{}, "cn-claim-manager", mgr, r, recon.WithPredicate(predicate.ResourceVersionChangedPredicate{}), - recon.WithBuildFn(watchPodChange), + recon.WithBuildFn(watchPodChangeFn(mgr.GetClient())), ) } -func watchPodChange(b *builder.Builder) { - b.Watches(&corev1.Pod{}, handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, object client.Object) []reconcile.Request { - pod, ok := object.(*corev1.Pod) - if !ok { - return nil - } - claimName, ok := pod.Labels[v1alpha1.PodClaimedByLabel] - if !ok { - return nil - } - return []reconcile.Request{{ +func watchPodChangeFn(cli client.Reader) func(*builder.Builder) { + return func(b *builder.Builder) { + b.Watches(&corev1.Pod{}, handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, object client.Object) []reconcile.Request { + pod, ok := object.(*corev1.Pod) + if !ok { + return nil + } + return requestsForPod(ctx, cli, pod) + }), builder.WithPredicates(common.PodStatusChangedPredicate{})) + } +} + +func indexClaimByPodName(object client.Object) []string { + claim, ok := object.(*v1alpha1.CNClaim) + if !ok || claim.Spec.PodName == "" { + return nil + } + return []string{claim.Spec.PodName} +} + +func requestsForPod(ctx context.Context, cli client.Reader, pod *corev1.Pod) []reconcile.Request { + var requests []reconcile.Request + if claimName, ok := pod.Labels[v1alpha1.PodClaimedByLabel]; ok && claimName != "" { + requests = append(requests, reconcile.Request{ NamespacedName: types.NamespacedName{ Namespace: pod.Namespace, Name: claimName, }, - }} - }), builder.WithPredicates(common.PodStatusChangedPredicate{})) + }) + } + + claimList := &v1alpha1.CNClaimList{} + if err := cli.List(ctx, claimList, + client.InNamespace(pod.Namespace), + client.MatchingFields{claimPodNameField: pod.Name}); err != nil { + ctrllog.FromContext(ctx).Error(err, "error listing CNClaims for Pod event", "pod", pod.Name) + return requests + } + for i := range claimList.Items { + req := reconcile.Request{ + NamespacedName: types.NamespacedName{ + Namespace: pod.Namespace, + Name: claimList.Items[i].Name, + }, + } + if !containsRequest(requests, req) { + requests = append(requests, req) + } + } + return requests +} + +func containsRequest(reqs []reconcile.Request, req reconcile.Request) bool { + for _, r := range reqs { + if r.NamespacedName == req.NamespacedName { + return true + } + } + return false } func toStoreStatus(cn *metadata.CNService, pod *corev1.Pod) v1alpha1.CNStoreStatus { diff --git a/pkg/controllers/cnclaim/controller_test.go b/pkg/controllers/cnclaim/controller_test.go index 2a784a52..b87c07e5 100644 --- a/pkg/controllers/cnclaim/controller_test.go +++ b/pkg/controllers/cnclaim/controller_test.go @@ -16,25 +16,37 @@ package cnclaim import ( "context" + stderrors "errors" "math/rand" "testing" + reconfake "github.com/matrixorigin/controller-runtime/pkg/fake" "github.com/matrixorigin/matrixone-operator/api/core/v1alpha1" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" "k8s.io/utils/pointer" "sigs.k8s.io/controller-runtime/pkg/client" - "sigs.k8s.io/controller-runtime/pkg/client/fake" + clientfake "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/client/interceptor" + "sigs.k8s.io/controller-runtime/pkg/reconcile" . "github.com/onsi/gomega" ) -func newFakeClient(objs ...client.Object) client.Client { +func newFakeClientBuilder(objs ...client.Object) *clientfake.ClientBuilder { scheme := runtime.NewScheme() _ = v1alpha1.SchemeBuilder.AddToScheme(scheme) _ = corev1.AddToScheme(scheme) - return fake.NewClientBuilder().WithScheme(scheme).WithObjects(objs...).Build() + return clientfake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(objs...). + WithIndex(&v1alpha1.CNClaim{}, claimPodNameField, indexClaimByPodName) +} + +func newFakeClient(objs ...client.Object) client.Client { + return newFakeClientBuilder(objs...).Build() } // fakeKubeClient adapts client.Client to recon.KubeClient for testing. @@ -195,7 +207,330 @@ func Test_buildPodClaimIndex(t *testing.T) { index, err := buildPodClaimIndex(&fakeKubeClient{cli}, "ns", "self") g.Expect(err).NotTo(HaveOccurred()) // "self" excluded, "deleting" filtered, "pending" has no podName - g.Expect(index).To(Equal(map[string]string{"pod-2": "other"})) + g.Expect(index).To(HaveLen(1)) + g.Expect(claimNames(index["pod-2"])).To(Equal([]string{"other"})) +} + +func Test_Finalize_transfersLabelWhenPodClaimedByOther(t *testing.T) { + g := NewGomegaWithT(t) + now := metav1.Now() + + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "pod-1", + Namespace: "ns", + Labels: map[string]string{ + v1alpha1.CNPodPhaseLabel: v1alpha1.CNPodPhaseBound, + v1alpha1.PodClaimedByLabel: "claim-a", + v1alpha1.ClaimSetNameLabel: "claimset-a", + v1alpha1.PodOwnerNameLabel: "owner-a", + "source-only": "old", + "shared": "old", + }, + }, + } + claimA := &v1alpha1.CNClaim{ + ObjectMeta: metav1.ObjectMeta{ + Name: "claim-a", + Namespace: "ns", + DeletionTimestamp: &now, + Finalizers: []string{"test"}, + Labels: map[string]string{ + v1alpha1.ClaimSetNameLabel: "claimset-a", + }, + }, + Spec: v1alpha1.CNClaimSpec{ + ClaimPodRef: v1alpha1.ClaimPodRef{PodName: "pod-1"}, + OwnerName: pointer.String("owner-a"), + AdditionalPodLabels: map[string]string{"source-only": "old", "shared": "old"}, + }, + } + claimB := &v1alpha1.CNClaim{ + ObjectMeta: metav1.ObjectMeta{ + Name: "claim-b", + Namespace: "ns", + Labels: map[string]string{ + v1alpha1.ClaimSetNameLabel: "claimset-b", + }, + }, + Spec: v1alpha1.CNClaimSpec{ + ClaimPodRef: v1alpha1.ClaimPodRef{PodName: "pod-1"}, + OwnerName: pointer.String("owner-b"), + AdditionalPodLabels: map[string]string{"target-only": "new", "shared": "new"}, + }, + Status: v1alpha1.CNClaimStatus{Phase: v1alpha1.CNClaimPhaseBound}, + } + + cli := newFakeClient(pod, claimA, claimB) + ctx := reconfake.NewContext(claimA, cli, nil) + done, err := (&Actor{}).Finalize(ctx) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(done).To(BeFalse(), "ownership transfer must be confirmed by another reconciliation") + + updatedPod := &corev1.Pod{} + g.Expect(cli.Get(context.Background(), client.ObjectKeyFromObject(pod), updatedPod)).To(Succeed()) + g.Expect(updatedPod.Labels).To(HaveKeyWithValue(v1alpha1.PodClaimedByLabel, "claim-b")) + g.Expect(updatedPod.Labels).To(HaveKeyWithValue(v1alpha1.CNPodPhaseLabel, v1alpha1.CNPodPhaseBound)) + g.Expect(updatedPod.Labels).To(HaveKeyWithValue(v1alpha1.ClaimSetNameLabel, "claimset-b")) + g.Expect(updatedPod.Labels).To(HaveKeyWithValue(v1alpha1.PodOwnerNameLabel, "owner-b")) + g.Expect(updatedPod.Labels).NotTo(HaveKey("source-only")) + g.Expect(updatedPod.Labels).To(HaveKeyWithValue("target-only", "new")) + g.Expect(updatedPod.Labels).To(HaveKeyWithValue("shared", "new")) + + done, err = (&Actor{}).Finalize(ctx) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(done).To(BeTrue()) +} + +func Test_transferPodOwnership_removesLabelsNotManagedByTarget(t *testing.T) { + g := NewGomegaWithT(t) + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{ + Name: "pod-1", + Namespace: "ns", + Labels: map[string]string{ + v1alpha1.CNPodPhaseLabel: v1alpha1.CNPodPhaseBound, + v1alpha1.PodClaimedByLabel: "claim-a", + v1alpha1.ClaimSetNameLabel: "claimset-a", + v1alpha1.PodOwnerNameLabel: "owner-a", + "source-only": "old", + }, + }} + from := &v1alpha1.CNClaim{ + ObjectMeta: metav1.ObjectMeta{Name: "claim-a", Namespace: "ns"}, + Spec: v1alpha1.CNClaimSpec{ + AdditionalPodLabels: map[string]string{"source-only": "old"}, + }, + } + to := &v1alpha1.CNClaim{ + ObjectMeta: metav1.ObjectMeta{Name: "claim-b", Namespace: "ns"}, + } + + cli := newFakeClient(pod) + storedPod := &corev1.Pod{} + g.Expect(cli.Get(context.Background(), client.ObjectKeyFromObject(pod), storedPod)).To(Succeed()) + ctx := reconfake.NewContext(from, cli, nil) + g.Expect(transferPodOwnership(ctx, storedPod, from, to)).To(Succeed()) + + updatedPod := &corev1.Pod{} + g.Expect(cli.Get(context.Background(), client.ObjectKeyFromObject(pod), updatedPod)).To(Succeed()) + g.Expect(updatedPod.Labels).To(HaveKeyWithValue(v1alpha1.PodClaimedByLabel, to.Name)) + g.Expect(updatedPod.Labels).NotTo(HaveKey(v1alpha1.ClaimSetNameLabel)) + g.Expect(updatedPod.Labels).NotTo(HaveKey(v1alpha1.PodOwnerNameLabel)) + g.Expect(updatedPod.Labels).NotTo(HaveKey("source-only")) +} + +func Test_Finalize_rejectsAmbiguousOwnershipTransfer(t *testing.T) { + g := NewGomegaWithT(t) + now := metav1.Now() + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{ + Name: "pod-1", + Namespace: "ns", + Labels: map[string]string{ + v1alpha1.CNPodPhaseLabel: v1alpha1.CNPodPhaseBound, + v1alpha1.PodClaimedByLabel: "claim-a", + }, + }} + claimA := &v1alpha1.CNClaim{ + ObjectMeta: metav1.ObjectMeta{ + Name: "claim-a", + Namespace: "ns", + DeletionTimestamp: &now, + Finalizers: []string{"test"}, + }, + Spec: v1alpha1.CNClaimSpec{ClaimPodRef: v1alpha1.ClaimPodRef{PodName: pod.Name}}, + } + claimB := &v1alpha1.CNClaim{ + ObjectMeta: metav1.ObjectMeta{Name: "claim-b", Namespace: "ns"}, + Spec: v1alpha1.CNClaimSpec{ClaimPodRef: v1alpha1.ClaimPodRef{PodName: pod.Name}}, + } + claimC := &v1alpha1.CNClaim{ + ObjectMeta: metav1.ObjectMeta{Name: "claim-c", Namespace: "ns"}, + Spec: v1alpha1.CNClaimSpec{ClaimPodRef: v1alpha1.ClaimPodRef{PodName: pod.Name}}, + } + + cli := newFakeClient(pod, claimA, claimB, claimC) + ctx := reconfake.NewContext(claimA, cli, nil) + done, err := (&Actor{}).Finalize(ctx) + g.Expect(done).To(BeFalse()) + g.Expect(err).To(MatchError(ContainSubstring("multiple CNClaims reference it: [claim-b claim-c]"))) + + updatedPod := &corev1.Pod{} + g.Expect(cli.Get(context.Background(), client.ObjectKeyFromObject(pod), updatedPod)).To(Succeed()) + g.Expect(updatedPod.Labels).To(HaveKeyWithValue(v1alpha1.PodClaimedByLabel, claimA.Name)) +} + +func Test_Finalize_waitsForTargetClaimToBecomeReady(t *testing.T) { + g := NewGomegaWithT(t) + now := metav1.Now() + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{ + Name: "pod-1", + Namespace: "ns", + Labels: map[string]string{ + v1alpha1.CNPodPhaseLabel: v1alpha1.CNPodPhaseBound, + v1alpha1.PodClaimedByLabel: "claim-a", + }, + }} + claimA := &v1alpha1.CNClaim{ + ObjectMeta: metav1.ObjectMeta{ + Name: "claim-a", + Namespace: "ns", + DeletionTimestamp: &now, + Finalizers: []string{"test"}, + }, + Spec: v1alpha1.CNClaimSpec{ClaimPodRef: v1alpha1.ClaimPodRef{PodName: pod.Name}}, + } + claimB := &v1alpha1.CNClaim{ + ObjectMeta: metav1.ObjectMeta{Name: "claim-b", Namespace: "ns"}, + Spec: v1alpha1.CNClaimSpec{ClaimPodRef: v1alpha1.ClaimPodRef{PodName: pod.Name}}, + Status: v1alpha1.CNClaimStatus{Phase: v1alpha1.CNClaimPhasePending}, + } + + cli := newFakeClient(pod, claimA, claimB) + ctx := reconfake.NewContext(claimA, cli, nil) + done, err := (&Actor{}).Finalize(ctx) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(done).To(BeFalse()) + + updatedPod := &corev1.Pod{} + g.Expect(cli.Get(context.Background(), client.ObjectKeyFromObject(pod), updatedPod)).To(Succeed()) + g.Expect(updatedPod.Labels).To(HaveKeyWithValue(v1alpha1.PodClaimedByLabel, claimA.Name)) +} + +func Test_Sync_clearsSpecOnPodNotFound(t *testing.T) { + g := NewGomegaWithT(t) + + // Setup: claim references a pod that doesn't exist + claim := &v1alpha1.CNClaim{ + ObjectMeta: metav1.ObjectMeta{ + Name: "claim-lost", + Namespace: "ns", + }, + Spec: v1alpha1.CNClaimSpec{ + ClaimPodRef: v1alpha1.ClaimPodRef{ + PodName: "pod-deleted", + NodeName: "node-1", + }, + }, + Status: v1alpha1.CNClaimStatus{ + Phase: v1alpha1.CNClaimPhaseBound, + }, + } + + cli := newFakeClient(claim) + storedClaim := &v1alpha1.CNClaim{} + g.Expect(cli.Get(context.Background(), client.ObjectKeyFromObject(claim), storedClaim)).To(Succeed()) + ctx := reconfake.NewContext(storedClaim, cli, nil) + g.Expect((&Actor{}).Sync(ctx)).To(Succeed()) + + updatedClaim := &v1alpha1.CNClaim{} + g.Expect(cli.Get(context.Background(), client.ObjectKeyFromObject(claim), updatedClaim)).To(Succeed()) + g.Expect(updatedClaim.Spec.PodName).To(BeEmpty()) + g.Expect(updatedClaim.Spec.NodeName).To(BeEmpty()) + g.Expect(ctx.Obj.Status.Phase).To(Equal(v1alpha1.CNClaimPhaseLost)) +} + +func Test_Observe_doesNotRebindLostClaim(t *testing.T) { + g := NewGomegaWithT(t) + claim := &v1alpha1.CNClaim{ + ObjectMeta: metav1.ObjectMeta{Name: "claim-lost", Namespace: "ns"}, + Status: v1alpha1.CNClaimStatus{Phase: v1alpha1.CNClaimPhaseLost}, + } + ctx := reconfake.NewContext(claim, newFakeClient(claim), nil) + + action, err := (&Actor{}).Observe(ctx) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(action).To(BeNil()) +} + +func Test_Sync_returnsPodGetError(t *testing.T) { + g := NewGomegaWithT(t) + getErr := stderrors.New("transient Pod read failure") + claim := &v1alpha1.CNClaim{ + ObjectMeta: metav1.ObjectMeta{Name: "claim", Namespace: "ns"}, + Spec: v1alpha1.CNClaimSpec{ClaimPodRef: v1alpha1.ClaimPodRef{PodName: "pod"}}, + } + cli := newFakeClientBuilder(claim).WithInterceptorFuncs(interceptor.Funcs{ + Get: func(context.Context, client.WithWatch, client.ObjectKey, client.Object, ...client.GetOption) error { + return getErr + }, + }).Build() + ctx := reconfake.NewContext(claim, cli, nil) + + err := (&Actor{}).Sync(ctx) + g.Expect(err).To(MatchError(And(ContainSubstring("error get claimed Pod"), ContainSubstring(getErr.Error())))) + g.Expect(claim.Spec.PodName).To(Equal("pod")) +} + +func Test_migrate_returnsSourcePodGetError(t *testing.T) { + g := NewGomegaWithT(t) + getErr := stderrors.New("transient source Pod read failure") + claim := &v1alpha1.CNClaim{ + ObjectMeta: metav1.ObjectMeta{Name: "claim", Namespace: "ns"}, + Spec: v1alpha1.CNClaimSpec{ + ClaimPodRef: v1alpha1.ClaimPodRef{PodName: "target"}, + SourcePod: &v1alpha1.ClaimPodRef{PodName: "source"}, + }, + } + cli := newFakeClientBuilder(claim).WithInterceptorFuncs(interceptor.Funcs{ + Get: func(context.Context, client.WithWatch, client.ObjectKey, client.Object, ...client.GetOption) error { + return getErr + }, + }).Build() + ctx := reconfake.NewContext(claim, cli, nil) + + err := (&Actor{}).migrate(ctx) + g.Expect(err).To(MatchError(And(ContainSubstring("error get source Pod"), ContainSubstring(getErr.Error())))) +} + +func Test_watchPodChangeFn_enqueuesClaimBySpecPodName(t *testing.T) { + g := NewGomegaWithT(t) + + // Setup: pod with NO claimed-by label, but a CNClaim references it via spec.podName + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: "pod-1", + Namespace: "ns", + Labels: map[string]string{}, + }, + } + claim := &v1alpha1.CNClaim{ + ObjectMeta: metav1.ObjectMeta{ + Name: "claim-refs-pod", + Namespace: "ns", + }, + Spec: v1alpha1.CNClaimSpec{ + ClaimPodRef: v1alpha1.ClaimPodRef{PodName: "pod-1"}, + }, + } + + cli := newFakeClient(pod, claim) + + requests := requestsForPod(context.Background(), cli, pod) + g.Expect(requests).To(HaveLen(1)) + g.Expect(requests[0].Name).To(Equal("claim-refs-pod")) + + // The label and field-index paths may identify the same claim; only enqueue it once. + pod.Labels[v1alpha1.PodClaimedByLabel] = claim.Name + requests = requestsForPod(context.Background(), cli, pod) + g.Expect(requests).To(HaveLen(1)) + g.Expect(requests[0].Name).To(Equal("claim-refs-pod")) +} + +func Test_containsRequest(t *testing.T) { + g := NewGomegaWithT(t) + reqs := []reconcile.Request{ + {NamespacedName: types.NamespacedName{Namespace: "ns", Name: "claim-a"}}, + {NamespacedName: types.NamespacedName{Namespace: "ns", Name: "claim-b"}}, + } + g.Expect(containsRequest(reqs, reconcile.Request{ + NamespacedName: types.NamespacedName{Namespace: "ns", Name: "claim-a"}, + })).To(BeTrue()) + g.Expect(containsRequest(reqs, reconcile.Request{ + NamespacedName: types.NamespacedName{Namespace: "ns", Name: "claim-c"}, + })).To(BeFalse()) + g.Expect(containsRequest(nil, reconcile.Request{ + NamespacedName: types.NamespacedName{Namespace: "ns", Name: "claim-a"}, + })).To(BeFalse()) } func Test_sortCNByPriority(t *testing.T) { diff --git a/pkg/controllers/cnclaim/migrate.go b/pkg/controllers/cnclaim/migrate.go index 21360205..439040f0 100644 --- a/pkg/controllers/cnclaim/migrate.go +++ b/pkg/controllers/cnclaim/migrate.go @@ -44,6 +44,7 @@ func (r *Actor) migrate(ctx *recon.Context[*v1alpha1.CNClaim]) error { if apierrors.IsNotFound(err) { return r.completeMigration(ctx) } + return errors.WrapPrefix(err, "error get source Pod", 0) } if c.Status.Store.BoundTime == nil { return errors.New(fmt.Sprintf("claim store %s/%s bound time is nil", c.Namespace, c.Name)) diff --git a/pkg/controllers/cnclaimset/controller.go b/pkg/controllers/cnclaimset/controller.go index 8046179c..2d7af065 100644 --- a/pkg/controllers/cnclaimset/controller.go +++ b/pkg/controllers/cnclaimset/controller.go @@ -1,4 +1,4 @@ -// Copyright 2025 Matrix Origin +// Copyright 2025-2026 Matrix Origin // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -215,8 +215,13 @@ func (c *ClaimAndPod) scoreHasPod() int { func (r *Actor) scaleIn(ctx *recon.Context[*v1alpha1.CNClaimSet], oc *ownedClaims, count int) error { var cps []ClaimAndPod + var migrating []v1alpha1.CNClaim for i := range oc.owned { c := oc.owned[i] + if c.Spec.SourcePod != nil { + migrating = append(migrating, c) + continue + } pod, err := getClaimedPod(ctx, &c) if err != nil { return errors.WrapPrefix(err, "error get claimed Pod", 0) @@ -226,8 +231,10 @@ func (r *Actor) scaleIn(ctx *recon.Context[*v1alpha1.CNClaimSet], oc *ownedClaim Pod: pod, }) } + if len(migrating) > 0 { + ctx.Log.Info("skip migrating claims from scale-in", "count", len(migrating)) + } if count >= len(cps) { - // simply delete all claims count = len(cps) } else { sortClaimsToDelete(cps) @@ -246,6 +253,7 @@ func (r *Actor) scaleIn(ctx *recon.Context[*v1alpha1.CNClaimSet], oc *ownedClaim for ; i < len(cps); i++ { left = append(left, *cps[i].Claim) } + left = append(left, migrating...) oc.owned = left return nil } diff --git a/pkg/controllers/cnclaimset/controller_test.go b/pkg/controllers/cnclaimset/controller_test.go index 389d1c39..7b040e51 100644 --- a/pkg/controllers/cnclaimset/controller_test.go +++ b/pkg/controllers/cnclaimset/controller_test.go @@ -1,4 +1,4 @@ -// Copyright 2025 Matrix Origin +// Copyright 2025-2026 Matrix Origin // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -15,17 +15,77 @@ package cnclaimset import ( + "context" "testing" "time" + "github.com/matrixorigin/controller-runtime/pkg/fake" "github.com/matrixorigin/matrixone-operator/api/core/v1alpha1" "github.com/matrixorigin/matrixone-operator/pkg/controllers/common" corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" . "github.com/onsi/gomega" ) +func Test_scaleIn_skipsMigratingClaims(t *testing.T) { + g := NewGomegaWithT(t) + now := time.Now() + + oc := &ownedClaims{ + owned: []v1alpha1.CNClaim{ + { + ObjectMeta: metav1.ObjectMeta{ + Name: "claim-migrating", + Namespace: "ns", + }, + Spec: v1alpha1.CNClaimSpec{ + ClaimPodRef: v1alpha1.ClaimPodRef{PodName: "pod-target"}, + SourcePod: &v1alpha1.ClaimPodRef{PodName: "pod-source"}, + }, + Status: v1alpha1.CNClaimStatus{ + Phase: v1alpha1.CNClaimPhaseBound, + }, + }, + { + ObjectMeta: metav1.ObjectMeta{ + Name: "claim-normal", + Namespace: "ns", + CreationTimestamp: metav1.NewTime(now), + }, + Spec: v1alpha1.CNClaimSpec{ + ClaimPodRef: v1alpha1.ClaimPodRef{PodName: ""}, + }, + Status: v1alpha1.CNClaimStatus{ + Phase: v1alpha1.CNClaimPhasePending, + }, + }, + }, + } + + scheme := runtime.NewScheme() + g.Expect(v1alpha1.AddToScheme(scheme)).To(Succeed()) + cli := fake.KubeClientBuilder(). + WithScheme(scheme). + WithObjects(&oc.owned[0], &oc.owned[1]). + Build() + ctx := fake.NewContext(&v1alpha1.CNClaimSet{ + ObjectMeta: metav1.ObjectMeta{Name: "claimset", Namespace: "ns"}, + }, cli, nil) + + g.Expect((&Actor{}).scaleIn(ctx, oc, 1)).To(Succeed()) + g.Expect(oc.owned).To(HaveLen(1)) + g.Expect(oc.owned[0].Name).To(Equal("claim-migrating")) + + stored := &v1alpha1.CNClaim{} + g.Expect(cli.Get(context.Background(), client.ObjectKey{Namespace: "ns", Name: "claim-migrating"}, stored)).To(Succeed()) + err := cli.Get(context.Background(), client.ObjectKey{Namespace: "ns", Name: "claim-normal"}, &v1alpha1.CNClaim{}) + g.Expect(apierrors.IsNotFound(err)).To(BeTrue()) +} + func Test_sortClaimsToDelete(t *testing.T) { type args struct { cps []ClaimAndPod diff --git a/test/e2e/claim_test.go b/test/e2e/claim_test.go index 4d4cb966..cd518add 100644 --- a/test/e2e/claim_test.go +++ b/test/e2e/claim_test.go @@ -1,4 +1,4 @@ -// Copyright 2025 Matrix Origin +// Copyright 2025-2026 Matrix Origin // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -24,6 +24,7 @@ import ( . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/utils/pointer" @@ -166,6 +167,13 @@ var _ = Describe("CNClaim and CNPool test", func() { }, }, } + DeferCleanup(func() { + By("Teardown CNPool dependencies") + deleteE2EObject(pool) + deleteE2EObject(proxy) + deleteE2EObject(d) + deleteE2EObject(l) + }) Expect(kubeCli.Create(ctx, l)).To(Succeed()) Expect(kubeCli.Create(ctx, d)).To(Succeed()) Expect(kubeCli.Create(ctx, pool)).To(Succeed()) @@ -209,6 +217,10 @@ var _ = Describe("CNClaim and CNPool test", func() { }, } Expect(kubeCli.Create(ctx, claim)).To(Succeed()) + DeferCleanup(func() { + By("Teardown CNClaim") + deleteE2EObject(claim) + }) Eventually(func() error { if err := kubeCli.Get(ctx, client.ObjectKeyFromObject(claim), claim); err != nil { @@ -246,11 +258,35 @@ var _ = Describe("CNClaim and CNPool test", func() { if err := kubeCli.Get(ctx, client.ObjectKeyFromObject(claim), claim); err != nil { return err } - if claim.Status.Store.PodName == target.Name { + if claim.Status.Store.PodName == target.Name && + claim.Spec.SourcePod == nil && + claim.Status.Migrate == nil { return nil } - logger.Infow("wait migrate complete", "claim", claim.Name) + logger.Infow("wait migrate complete", + "claim", claim.Name, + "targetPod", target.Name, + "storePod", claim.Status.Store.PodName, + "sourcePod", claim.Spec.SourcePod, + "progress", claim.Status.Migrate) return errWait }, migrateTimeout, pollInterval).Should(Succeed()) }) }) + +func deleteE2EObject(obj client.Object) { + key := client.ObjectKeyFromObject(obj) + Expect(util.Ignore(apierrors.IsNotFound, kubeCli.Delete(ctx, obj))).To(Succeed()) + Eventually(func() error { + err := kubeCli.Get(ctx, key, obj) + if apierrors.IsNotFound(err) { + return nil + } + if err != nil { + logger.Errorw("error getting resource during teardown", "resource", fmt.Sprintf("%T", obj), "key", key, "error", err) + return err + } + logger.Infow("wait resource teardown", "resource", fmt.Sprintf("%T", obj), "key", key) + return errWait + }, teardownClusterTimeout, pollInterval).Should(Succeed(), "%T %s should be deleted", obj, key) +} diff --git a/test/e2e/suite_test.go b/test/e2e/suite_test.go index 3c46802a..a6252614 100644 --- a/test/e2e/suite_test.go +++ b/test/e2e/suite_test.go @@ -1,4 +1,4 @@ -// Copyright 2025 Matrix Origin +// Copyright 2025-2026 Matrix Origin // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -103,11 +103,15 @@ var _ = SynchronizedBeforeSuite(func() []byte { ns := &corev1.Namespace{ ObjectMeta: metav1.ObjectMeta{ - Name: env.Namespace, + Name: env.Namespace, + Labels: e2eResourceLabels(), }, } Expect(util.Ignore(apierrors.IsAlreadyExists, kubeCli.Create(ctx, ns))).To(Succeed()) - poolNS := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: env.PoolNamespace}} + poolNS := &corev1.Namespace{ObjectMeta: metav1.ObjectMeta{ + Name: env.PoolNamespace, + Labels: e2eResourceLabels(), + }} Expect(util.Ignore(apierrors.IsAlreadyExists, kubeCli.Create(ctx, poolNS))).To(Succeed()) buf, err := json.Marshal(env)