fix: add pod informer for auto-migration

pkg/controllers/automigration/controller.go

@@ -28,6 +28,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	pkgruntime "k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	dynamicclient "k8s.io/client-go/dynamic"
 	"k8s.io/client-go/informers"
 	kubeclient "k8s.io/client-go/kubernetes"
@@ -43,6 +44,7 @@ import (
 	"github.com/kubewharf/kubeadmiral/pkg/controllers/util"
 	"github.com/kubewharf/kubeadmiral/pkg/controllers/util/delayingdeliver"
 	"github.com/kubewharf/kubeadmiral/pkg/controllers/util/eventsink"
+	"github.com/kubewharf/kubeadmiral/pkg/controllers/util/managedlabel"
 	utilunstructured "github.com/kubewharf/kubeadmiral/pkg/controllers/util/unstructured"
 	"github.com/kubewharf/kubeadmiral/pkg/controllers/util/worker"
 	"github.com/kubewharf/kubeadmiral/pkg/stats"
@@ -70,7 +72,8 @@ type Controller struct {
 	federatedObjectClient   dynamicclient.NamespaceableResourceInterface
 	federatedObjectInformer informers.GenericInformer
 
-	federatedInformer util.FederatedInformer
+	federatedInformer       util.FederatedInformer
+	federatedInformerForPod util.FederatedInformer
 
 	worker worker.ReconcileWorker
 
@@ -145,6 +148,71 @@ func NewAutoMigrationController(
 		return nil, fmt.Errorf("failed to create federated informer: %w", err)
 	}
 
+	c.federatedInformerForPod, err = util.NewFederatedInformerWithEventHandler(
+		controllerConfig,
+		genericFedClient,
+		controllerConfig.KubeConfig,
+		&metav1.APIResource{
+			Name:       "pods",
+			Namespaced: true,
+			Group:      corev1.GroupName,
+			Version:    "v1",
+			Kind:       common.PodKind,
+		},
+		&util.ResourceEventHandlerFuncsForCluster{
+			UpdateFunc: func(oldObj, newObj interface{}, cluster *fedcorev1a1.FederatedCluster) {
+				keyedLogger := c.logger.WithValues("cluster", cluster.Name)
+				ctx := klog.NewContext(context.TODO(), keyedLogger)
+
+				unsNewObj, ok := newObj.(*unstructured.Unstructured)
+				if !ok {
+					keyedLogger.Error(nil, fmt.Sprintf("Internal error: newObj not of Unstructured type: %v", newObj))
+					return
+				}
+				if unsNewObj.GetDeletionTimestamp() != nil {
+					return
+				}
+				unsOldObj, ok := oldObj.(*unstructured.Unstructured)
+				if !ok {
+					keyedLogger.Error(nil, fmt.Sprintf("Internal error: oldObj not of Unstructured type: %v", oldObj))
+					return
+				}
+
+				newPod := &corev1.Pod{}
+				oldPod := &corev1.Pod{}
+				if err := pkgruntime.DefaultUnstructuredConverter.FromUnstructured(unsNewObj.Object, newPod); err != nil {
+					keyedLogger.Error(nil, fmt.Sprintf("Internal error: newObj not of Pod type: %v", unsNewObj))
+					return
+				}
+				if err := pkgruntime.DefaultUnstructuredConverter.FromUnstructured(unsOldObj.Object, oldPod); err != nil {
+					keyedLogger.Error(nil, fmt.Sprintf("Internal error: oldObj not of Pod type: %v", unsOldObj))
+					return
+				}
+
+				if !podScheduledConditionChanged(oldPod, newPod) {
+					return
+				}
+
+				qualifiedName, found, err := c.getSourceObjFromCluster(ctx, newPod, cluster.Name)
+				if err != nil || !found {
+					keyedLogger.V(3).Info(
+						"Failed to get source object form pod",
+						"pod", common.NewQualifiedName(newPod),
+						"err", err,
+					)
+					return
+				}
+				// enqueue with a delay to simulate a rudimentary rate limiter
+				c.worker.EnqueueWithDelay(*qualifiedName, 10*time.Second)
+			},
+		},
+		nil,
+		&util.ClusterLifecycleHandlerFuncs{},
+	)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create federated informer for pod: %w", err)
+	}
+
 	return c, nil
 }
 
@@ -154,6 +222,8 @@ func (c *Controller) Run(ctx context.Context) {
 
 	c.federatedInformer.Start()
 	defer c.federatedInformer.Stop()
+	c.federatedInformerForPod.Start()
+	defer c.federatedInformerForPod.Stop()
 
 	if !cache.WaitForNamedCacheSync(c.name, ctx.Done(), c.HasSynced) {
 		return
@@ -164,7 +234,9 @@ func (c *Controller) Run(ctx context.Context) {
 }
 
 func (c *Controller) HasSynced() bool {
-	if !c.federatedObjectInformer.Informer().HasSynced() || !c.federatedInformer.ClustersSynced() {
+	if !c.federatedObjectInformer.Informer().HasSynced() ||
+		!c.federatedInformer.ClustersSynced() ||
+		!c.federatedInformerForPod.ClustersSynced() {
 		return false
 	}
 
@@ -185,7 +257,7 @@ func (c *Controller) reconcile(qualifiedName common.QualifiedName) (status worke
 	keyedLogger.V(3).Info("Start reconcile")
 	defer func() {
 		c.metrics.Duration(fmt.Sprintf("%s.latency", c.name), startTime)
-		keyedLogger.V(3).Info("Finished reconcile", "duration", time.Since(startTime), "status", status.String())
+		keyedLogger.V(3).Info("Finished reconcile", "duration", time.Since(startTime), "status", status)
 	}()
 
 	fedObject, err := util.UnstructuredFromStore(c.federatedObjectInformer.Informer().GetStore(), key)
@@ -335,7 +407,6 @@ func (c *Controller) estimateCapacity(
 			"total", len(pods),
 			"desired", desiredReplicas,
 			"unschedulable", unschedulable,
-			"next-cross-in", nextCrossIn,
 		)
 
 		if nextCrossIn != nil && (retryAfter == nil || *nextCrossIn < *retryAfter) {
@@ -438,3 +509,44 @@ func (c *Controller) getPodsFromCluster(
 
 	return pods, false, nil
 }
+
+func (c *Controller) getSourceObjFromCluster(
+	ctx context.Context,
+	pod *corev1.Pod,
+	clusterName string,
+) (sourceQualified *common.QualifiedName, found bool, err error) {
+	plugin, err := plugins.ResolvePlugin(c.typeConfig)
+	if err != nil {
+		return nil, false, fmt.Errorf("failed to get plugin for FTC: %w", err)
+	}
+
+	client, err := c.federatedInformer.GetClientForCluster(clusterName)
+	if err != nil {
+		return nil, true, fmt.Errorf("failed to get client for cluster: %w", err)
+	}
+
+	object, found, err := plugin.GetTargetObjectFromPod(ctx, pod, plugins.ClusterHandle{
+		Client: client,
+	})
+	if err != nil {
+		return nil, false, fmt.Errorf("failed to get target object form pod: %w", err)
+	}
+	if !found {
+		return nil, false, nil
+	}
+
+	gv, err := schema.ParseGroupVersion(object.GetAPIVersion())
+	if err != nil {
+		return nil, false, fmt.Errorf("failed to parse APIVersion form obj: %w", err)
+	}
+	qualifiedName := common.NewQualifiedName(object)
+	targetType := c.typeConfig.GetTargetType()
+	managed := object.GetLabels()[managedlabel.ManagedByKubeAdmiralLabelKey] == managedlabel.ManagedByKubeAdmiralLabelValue
+
+	if targetType.Group != gv.Group || targetType.Kind != object.GetKind() || !managed {
+		return nil, false, fmt.Errorf("got obj %s, but it's not the target type or not managed by admiral: {Group:%s Kind:%s Managed:%t}",
+			qualifiedName.String(), gv.Group, object.GetKind(), managed)
+	}
+
+	return &qualifiedName, true, nil
+}

pkg/controllers/automigration/plugins/deployments.go

@@ -24,8 +24,11 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
+	"github.com/kubewharf/kubeadmiral/pkg/controllers/common"
+	"github.com/kubewharf/kubeadmiral/pkg/lifted/kubernetes/pkg/controller"
 	deploymentutil "github.com/kubewharf/kubeadmiral/pkg/lifted/kubernetes/pkg/controller/deployment/util"
 )
 
@@ -92,6 +95,27 @@ func (*deploymentPlugin) GetPodsForClusterObject(
 	return ret, nil
 }
 
+func (*deploymentPlugin) GetTargetObjectFromPod(
+	ctx context.Context,
+	pod *corev1.Pod,
+	handle ClusterHandle,
+) (obj *unstructured.Unstructured, found bool, err error) {
+	rs, found, err := controller.GetSpecifiedOwnerFromSourceObj(ctx, handle.Client, pod, schema.GroupVersionKind{
+		Group:   appsv1.GroupName,
+		Version: "v1",
+		Kind:    common.ReplicaSetKind,
+	})
+	if err != nil || !found {
+		return nil, false, err
+	}
+
+	return controller.GetSpecifiedOwnerFromSourceObj(ctx, handle.Client, rs, schema.GroupVersionKind{
+		Group:   appsv1.GroupName,
+		Version: "v1",
+		Kind:    common.DeploymentKind,
+	})
+}
+
 var _ Plugin = &deploymentPlugin{}
 
 func convertListOptions(ns string, opts *metav1.ListOptions) (*client.ListOptions, error) {

pkg/controllers/automigration/plugins/plugins.go

@@ -40,6 +40,12 @@ type Plugin interface {
 		obj *unstructured.Unstructured,
 		handle ClusterHandle,
 	) ([]*corev1.Pod, error)
+
+	GetTargetObjectFromPod(
+		ctx context.Context,
+		pod *corev1.Pod,
+		handle ClusterHandle,
+	) (obj *unstructured.Unstructured, found bool, err error)
 }
 
 var nativePlugins = map[schema.GroupVersionResource]Plugin{

pkg/controllers/automigration/util.go

@@ -36,17 +36,8 @@ func countUnschedulablePods(
 			continue
 		}
 
-		var scheduledCondition *corev1.PodCondition
-		for i := range pod.Status.Conditions {
-			condition := &pod.Status.Conditions[i]
-			if condition.Type == corev1.PodScheduled {
-				scheduledCondition = condition
-				break
-			}
-		}
-		if scheduledCondition == nil ||
-			scheduledCondition.Status != corev1.ConditionFalse ||
-			scheduledCondition.Reason != corev1.PodReasonUnschedulable {
+		scheduledCondition, isUnschedulable := getPodScheduledCondition(pod)
+		if !isUnschedulable {
 			continue
 		}
 
@@ -62,3 +53,34 @@ func countUnschedulablePods(
 
 	return unschedulableCount, nextCrossIn
 }
+
+func getPodScheduledCondition(pod *corev1.Pod) (scheduledCondition *corev1.PodCondition, isUnschedulable bool) {
+	for i := range pod.Status.Conditions {
+		condition := &pod.Status.Conditions[i]
+		if condition.Type == corev1.PodScheduled {
+			scheduledCondition = condition
+			break
+		}
+	}
+	if scheduledCondition == nil ||
+		scheduledCondition.Status != corev1.ConditionFalse ||
+		scheduledCondition.Reason != corev1.PodReasonUnschedulable {
+		return scheduledCondition, false
+	}
+	return scheduledCondition, true
+}
+
+func podScheduledConditionChanged(oldPod, newPod *corev1.Pod) bool {
+	condition, _ := getPodScheduledCondition(newPod)
+	oldCondition, _ := getPodScheduledCondition(oldPod)
+	if condition == nil || oldCondition == nil {
+		return condition != oldCondition
+	}
+
+	isEqual := condition.Status == oldCondition.Status &&
+		condition.Reason == oldCondition.Reason &&
+		condition.Message == oldCondition.Message &&
+		condition.LastProbeTime.Equal(&oldCondition.LastProbeTime) &&
+		condition.LastTransitionTime.Equal(&oldCondition.LastTransitionTime)
+	return !isEqual
+}

pkg/controllers/automigration/util_test.go

@@ -120,3 +120,71 @@ func newPod(terminating bool, schedulable bool, lastTransitionTimestamp time.Tim
 	}
 	return pod
 }
+
+func Test_podScheduledConditionChanged(t *testing.T) {
+	now := time.Now()
+	podWithEmptyCond := newPod(false, false, now)
+	podWithEmptyCond.Status.Conditions = nil
+
+	tests := []struct {
+		name   string
+		oldPod *corev1.Pod
+		newPod *corev1.Pod
+		want   bool
+	}{
+		{
+			name:   "both nil",
+			oldPod: podWithEmptyCond,
+			newPod: podWithEmptyCond,
+			want:   false,
+		},
+		{
+			name:   "oldPod condition is nil",
+			oldPod: podWithEmptyCond,
+			newPod: newPod(false, false, now),
+			want:   true,
+		},
+		{
+			name:   "newPod condition is nil",
+			oldPod: newPod(false, false, now),
+			newPod: podWithEmptyCond,
+			want:   true,
+		},
+		{
+			name:   "unschedulable condition equal",
+			oldPod: newPod(false, false, now),
+			newPod: newPod(false, false, now),
+			want:   false,
+		},
+		{
+			name:   "unschedulable condition not equal",
+			oldPod: newPod(false, false, now.Add(10*time.Second)),
+			newPod: newPod(false, false, now),
+			want:   true,
+		},
+		{
+			name:   "schedulable condition equal",
+			oldPod: newPod(false, true, now),
+			newPod: newPod(false, true, now),
+			want:   false,
+		},
+		{
+			name:   "schedulable condition not equal",
+			oldPod: newPod(false, true, now.Add(10*time.Second)),
+			newPod: newPod(false, true, now),
+			want:   true,
+		},
+		{
+			name:   "schedulable and unschedulable conditions",
+			oldPod: newPod(false, true, now),
+			newPod: newPod(false, false, now),
+			want:   true,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equalf(t, tt.want, podScheduledConditionChanged(tt.oldPod, tt.newPod),
+				"podScheduledConditionChanged(%v, %v)", tt.oldPod, tt.newPod)
+		})
+	}
+}

pkg/controllers/common/constants.go

@@ -46,6 +46,7 @@ const (
 	PersistentVolumeKind      = "PersistentVolume"
 	PersistentVolumeClaimKind = "PersistentVolumeClaim"
 	PodKind                   = "Pod"
+	ReplicaSetKind            = "ReplicaSet"
 )
 
 // The following consts are spec fields used to interact with unstructured resources