feat: delete draining timeout pod and watch claim on pool

Makefile

@@ -91,7 +91,7 @@ test: api-test unit
 
 # Run unit tests
 unit: generate fmt vet manifests
-	go test ./pkg/... -coverprofile cover.out
+	CGO_ENABLED=0 go test ./pkg/... -coverprofile cover.out
 
 api-test:
 	cd api && make test

api/core/v1alpha1/cnclaim_helper.go

@@ -0,0 +1,23 @@
+// Copyright 2024 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package v1alpha1
+
+func (c *CNClaim) IsReady() bool {
+	return c.Status.Phase == CNClaimPhaseBound || c.Status.Phase == CNClaimPhaseOutdated
+}
+
+func (c *CNClaim) IsUpdated() bool {
+	return c.Status.Phase == CNClaimPhaseBound
+}

api/core/v1alpha1/cnclaim_types.go

@@ -25,6 +25,8 @@ const (
 	CNClaimPhaseBound   CNClaimPhase = "Bound"
 	CNClaimPhaseLost    CNClaimPhase = "Lost"
 
+	CNClaimPhaseOutdated CNClaimPhase = "Outdated"
+
 	ClaimOwnerNameLabel = "matrixorigin.io/claim-owner"
 )
 

pkg/controllers/cnclaim/controller.go

@@ -28,12 +28,20 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 	"slices"
 	"time"
 )
 
+const (
+	waitCacheTimeout = 10 * time.Second
+)
+
 // Actor reconciles CN Claim
 type Actor struct {
 	clientMgr *hacli.HAKeeperClientManager
@@ -146,7 +154,7 @@ func (r *Actor) doClaimCN(ctx *recon.Context[*v1alpha1.CNClaim], orphans []corev
 		return nil, errors.Wrap(err, "error list idle Pods")
 	}
 
-	slices.SortFunc(idleCNs, priorityFunc(c))
+	sortCNByPriority(c, idleCNs)
 	for i := range idleCNs {
 		pod := &idleCNs[i]
 		pod.Labels[v1alpha1.CNPodPhaseLabel] = v1alpha1.CNPodPhaseBound
@@ -191,7 +199,32 @@ func (r *Actor) bindPod(ctx *recon.Context[*v1alpha1.CNClaim], pod *corev1.Pod,
 }
 
 func (r *Actor) Sync(ctx *recon.Context[*v1alpha1.CNClaim]) error {
-	// TODO: monitor pod health
+	c := ctx.Obj
+	switch c.Status.Phase {
+	case v1alpha1.CNClaimPhasePending:
+		return errors.Errorf("CN Claim %s/%s is pending, should bind it first", c.Namespace, c.Name)
+	case v1alpha1.CNClaimPhaseLost:
+		return nil
+	case v1alpha1.CNClaimPhaseBound, v1alpha1.CNClaimPhaseOutdated:
+		// noop
+	default:
+		return errors.Errorf("CN Claim %s/%s is in unknown phase %s", c.Namespace, c.Name, c.Status.Phase)
+	}
+	pod := &corev1.Pod{}
+	err := ctx.Get(types.NamespacedName{Namespace: c.Namespace, Name: c.Spec.PodName}, pod)
+	if err != nil {
+		if apierrors.IsNotFound(err) {
+			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)
+			}
+			c.Status.Phase = v1alpha1.CNClaimPhaseLost
+			return nil
+		}
+	}
+	if pod.Status.Phase == corev1.PodSucceeded || pod.Status.Phase == corev1.PodUnknown {
+		c.Status.Phase = v1alpha1.CNClaimPhaseLost
+		return nil
+	}
 	return nil
 }
 
@@ -223,7 +256,7 @@ func (r *Actor) Finalize(ctx *recon.Context[*v1alpha1.CNClaim]) (bool, error) {
 		// set the CN Pod to draining phase and let the draining process handle recycling
 		if err := ctx.Patch(&cn, func() error {
 			cn.Labels[v1alpha1.CNPodPhaseLabel] = v1alpha1.CNPodPhaseDraining
-			delete(cn.Labels, v1alpha1.ClaimOwnerNameLabel)
+			delete(cn.Labels, v1alpha1.PodClaimedByLabel)
 			if cn.Annotations == nil {
 				cn.Annotations = map[string]string{}
 			}
@@ -268,7 +301,24 @@ func (r *Actor) patchStore(ctx *recon.Context[*v1alpha1.CNClaim], pod *corev1.Po
 }
 
 func (r *Actor) Start(mgr manager.Manager) error {
-	return recon.Setup[*v1alpha1.CNClaim](&v1alpha1.CNClaim{}, "cn-claim-manager", mgr, r)
+	return recon.Setup[*v1alpha1.CNClaim](&v1alpha1.CNClaim{}, "cn-claim-manager", mgr, r, recon.WithBuildFn(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
+			}
+			claimName, ok := pod.Labels[v1alpha1.PodClaimedByLabel]
+			if !ok {
+				return nil
+			}
+			return []reconcile.Request{{
+				NamespacedName: types.NamespacedName{
+					Namespace: pod.Namespace,
+					Name:      claimName,
+				},
+			}}
+		}))
+	}))
 }
 
 func toStoreStatus(cn *metadata.CNService) v1alpha1.CNStoreStatus {
@@ -293,6 +343,10 @@ func toStoreStatus(cn *metadata.CNService) v1alpha1.CNStoreStatus {
 	}
 }
 
+func sortCNByPriority(c *v1alpha1.CNClaim, pods []corev1.Pod) {
+	slices.SortFunc(pods, priorityFunc(c))
+}
+
 func priorityFunc(c *v1alpha1.CNClaim) func(a, b corev1.Pod) int {
 	return func(a, b corev1.Pod) int {
 		// 1. claim the previously used pod first

pkg/controllers/cnclaim/controller_test.go

@@ -0,0 +1,92 @@
+// Copyright 2024 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package cnclaim
+
+import (
+	"github.com/matrixorigin/matrixone-operator/api/core/v1alpha1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/utils/pointer"
+	"math/rand"
+	"testing"
+
+	. "github.com/onsi/gomega"
+)
+
+func Test_sortCNByPriority(t *testing.T) {
+	tests := []struct {
+		name  string
+		c     *v1alpha1.CNClaim
+		pods  []corev1.Pod
+		order []string
+	}{{
+		name: "basic",
+		c: &v1alpha1.CNClaim{
+			Spec: v1alpha1.CNClaimSpec{
+				OwnerName: pointer.String("set1"),
+			},
+		},
+		pods: []corev1.Pod{
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "not-claimed-but-older",
+					Labels: map[string]string{
+						v1alpha1.CNPodPhaseLabel: v1alpha1.CNPodPhaseIdle,
+					},
+					CreationTimestamp: metav1.Unix(0, 0),
+				},
+			},
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "previously-claimed",
+					Labels: map[string]string{
+						v1alpha1.CNPodPhaseLabel:     v1alpha1.CNPodPhaseIdle,
+						v1alpha1.ClaimOwnerNameLabel: "set1",
+					},
+					CreationTimestamp: metav1.Unix(10, 0),
+				},
+			},
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "previously-claimed-by-other-set",
+					Labels: map[string]string{
+						v1alpha1.CNPodPhaseLabel:     v1alpha1.CNPodPhaseIdle,
+						v1alpha1.ClaimOwnerNameLabel: "set2",
+					},
+					CreationTimestamp: metav1.Unix(10, 0),
+				},
+			},
+		},
+		order: []string{
+			"previously-claimed",
+			"not-claimed-but-older",
+			"previously-claimed-by-other-set",
+		},
+	}}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			rand.Shuffle(len(tt.pods), func(i, j int) {
+				tt.pods[i], tt.pods[j] = tt.pods[j], tt.pods[i]
+			})
+			sortCNByPriority(tt.c, tt.pods)
+			g := NewGomegaWithT(t)
+			var res []string
+			for _, po := range tt.pods {
+				res = append(res, po.Name)
+			}
+			g.Expect(res).To(Equal(tt.order))
+		})
+	}
+}