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

pkg/controllers/cnclaim/controller.go

@@ -146,7 +146,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
@@ -293,6 +293,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))
+		})
+	}
+}

pkg/controllers/cnpool/controller.go

@@ -125,7 +125,7 @@ func (r *Actor) Sync(ctx *recon.Context[*v1alpha1.CNPool]) error {
 				return nil
 			}
 			scaleInCount := desired.Spec.Replicas - desiredReplicas
-			slices.SortFunc(idlePods, deletionOrder)
+			sortPodByDeletionOrder(idlePods)
 			if int32(len(idlePods)) > scaleInCount {
 				// pick first N to scale-in
 				idlePods = idlePods[0:scaleInCount]
@@ -294,6 +294,11 @@ func podInUse(pod *corev1.Pod) bool {
 		pod.Labels[v1alpha1.CNPodPhaseLabel] == v1alpha1.CNPodPhaseDraining
 }
 
+// sortPodByDeletionOrder sort the pool pods to be deleted
+func sortPodByDeletionOrder(pods []*corev1.Pod) {
+	slices.SortFunc(pods, deletionOrder)
+}
+
 func deletionOrder(a, b *corev1.Pod) int {
 	c := deletionCost(a) - deletionCost(b)
 	if c == 0 {
@@ -303,14 +308,6 @@ func deletionOrder(a, b *corev1.Pod) int {
 	return c
 }
 
-func podNames(pods []*corev1.Pod) []string {
-	var ss []string
-	for _, pod := range pods {
-		ss = append(ss, pod.Name)
-	}
-	return ss
-}
-
 func deletionCost(pod *corev1.Pod) int {
 	score := 0
 	switch pod.Labels[v1alpha1.CNPodPhaseLabel] {
@@ -332,7 +329,6 @@ func deletionCost(pod *corev1.Pod) int {
 
 func (r *Actor) Start(mgr manager.Manager) error {
 	return recon.Setup[*v1alpha1.CNPool](&v1alpha1.CNPool{}, "cn-pool-manager", mgr, r, recon.WithBuildFn(func(b *builder.Builder) {
-		// TODO: watch Pods in Pool
 		b.Watches(&corev1.Pod{}, handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, object client.Object) []reconcile.Request {
 			pod, ok := object.(*corev1.Pod)
 			if !ok {
@@ -349,6 +345,28 @@ func (r *Actor) Start(mgr manager.Manager) error {
 				},
 			}}
 		}))
+		b.Watches(&v1alpha1.CNClaim{}, handler.EnqueueRequestsFromMapFunc(func(ctx context.Context, object client.Object) []reconcile.Request {
+			claim, ok := object.(*v1alpha1.CNClaim)
+			if !ok {
+				return nil
+			}
+			if claim.Spec.PoolName == "" {
+				return nil
+			}
+			return []reconcile.Request{{
+				NamespacedName: types.NamespacedName{
+					Namespace: claim.Namespace,
+					Name:      claim.Spec.PoolName,
+				},
+			}}
+		}))
 		b.Owns(&v1alpha1.CNSet{})
 	}))
 }
+func podNames(pods []*corev1.Pod) []string {
+	var ss []string
+	for _, pod := range pods {
+		ss = append(ss, pod.Name)
+	}
+	return ss
+}

pkg/controllers/cnpool/controller_test.go

@@ -0,0 +1,93 @@
+// 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 cnpool
+
+import (
+	"github.com/matrixorigin/matrixone-operator/api/core/v1alpha1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"math/rand"
+	"testing"
+
+	. "github.com/onsi/gomega"
+)
+
+func Test_sortPodByDeletionOrder(t *testing.T) {
+	tests := []struct {
+		name  string
+		pods  []*corev1.Pod
+		order []string
+	}{{
+		name: "basic",
+		pods: []*corev1.Pod{
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "idle-new",
+					Labels: map[string]string{
+						v1alpha1.CNPodPhaseLabel: v1alpha1.CNPodPhaseIdle,
+					},
+					CreationTimestamp: metav1.Unix(10, 0),
+				},
+			},
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "unknown",
+					Labels: map[string]string{
+						v1alpha1.CNPodPhaseLabel: v1alpha1.CNPodPhaseUnknown,
+					},
+					CreationTimestamp: metav1.Unix(10, 0),
+				},
+			},
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "idle-old",
+					Labels: map[string]string{
+						v1alpha1.CNPodPhaseLabel: v1alpha1.CNPodPhaseIdle,
+					},
+					CreationTimestamp: metav1.Unix(0, 0),
+				},
+			},
+			{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "terminating",
+					Labels: map[string]string{
+						v1alpha1.CNPodPhaseLabel: v1alpha1.CNPodPhaseTerminating,
+					},
+					CreationTimestamp: metav1.Unix(10, 0),
+				},
+			},
+		},
+		order: []string{
+			"terminating",
+			"unknown",
+			"idle-new",
+			"idle-old",
+		},
+	}}
+	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]
+			})
+			sortPodByDeletionOrder(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))
+		})
+	}
+}

pkg/controllers/cnstore/pooling.go

@@ -17,11 +17,13 @@ package cnstore
 import (
 	"context"
 	recon "github.com/matrixorigin/controller-runtime/pkg/reconciler"
+	"github.com/matrixorigin/controller-runtime/pkg/util"
 	"github.com/matrixorigin/matrixone-operator/api/core/v1alpha1"
 	"github.com/matrixorigin/matrixone-operator/pkg/controllers/common"
 	"github.com/matrixorigin/matrixone-operator/pkg/hacli"
 	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"time"
 )
 
@@ -47,15 +49,16 @@ func (c *withCNSet) poolingCNReconcile(ctx *recon.Context[*corev1.Pod]) error {
 		// recycle the pod
 		timeStr, ok := pod.Annotations[common.ReclaimedAt]
 		if !ok {
-			// legacy pod, simply timeout
+			// legacy pod, simply return it to the pool
 			return c.patchPhase(ctx, v1alpha1.CNPodPhaseIdle)
 		}
 		parsed, err := time.Parse(time.RFC3339, timeStr)
 		if err != nil {
 			return errors.Wrapf(err, "error parsing start time %s", timeStr)
 		}
 		if time.Since(parsed) > recycleTimeout {
-			return c.patchPhase(ctx, v1alpha1.CNPodPhaseIdle)
+			ctx.Log.Info("drain pool pod timeout, delete it to avoid workload intervention")
+			return util.Ignore(apierrors.IsNotFound, ctx.Delete(pod))
 		}
 		count, err := common.GetStoreConnection(pod)
 		if err != nil {