Replacing endpointSlice Reconciler with a direct Pod Reconciler (#300)

* reversion to pod reconciliation

* adding ready check and unit tests

* updating test

* ablating unnecessary func

* embedding ready status into update so non-ready pods are deleted

* scrubbing serviceName & zone as they are obsolete

* implementing pod cache flushing logic

* Renaming file so merge confilcts can find the diffs easier

* cleaning up messy merge conflict

* nil checking short circuit

* Listing fixes

* feedback cleanup

* log formatting and removing pods if not found

* removing err to provent perma-reconciliation

* removing dev image ref

* cleaning up err logic

Author: kfswain

pkg/ext-proc/backend/datastore.go

@@ -1,12 +1,16 @@
 package backend
 
 import (
+	"context"
 	"errors"
 	"math/rand"
+	"strconv"
 	"sync"
 
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/klog/v2"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/gateway-api-inference-extension/api/v1alpha1"
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/ext-proc/util/logging"
 )
@@ -111,3 +115,41 @@ func IsCritical(model *v1alpha1.InferenceModel) bool {
 	}
 	return false
 }
+
+func (ds *K8sDatastore) LabelsMatch(podLabels map[string]string) bool {
+	poolSelector := selectorFromInferencePoolSelector(ds.inferencePool.Spec.Selector)
+	podSet := labels.Set(podLabels)
+	return poolSelector.Matches(podSet)
+}
+
+func (ds *K8sDatastore) flushPodsAndRefetch(ctx context.Context, ctrlClient client.Client, newServerPool *v1alpha1.InferencePool) {
+	podList := &corev1.PodList{}
+	if err := ctrlClient.List(ctx, podList, &client.ListOptions{
+		LabelSelector: selectorFromInferencePoolSelector(newServerPool.Spec.Selector),
+		Namespace:     newServerPool.Namespace,
+	}); err != nil {
+		klog.Error(err, "error listing clients")
+	}
+	ds.pods.Clear()
+
+	for _, k8sPod := range podList.Items {
+		pod := Pod{
+			Name:    k8sPod.Name,
+			Address: k8sPod.Status.PodIP + ":" + strconv.Itoa(int(newServerPool.Spec.TargetPortNumber)),
+		}
+		ds.pods.Store(pod, true)
+	}
+
+}
+
+func selectorFromInferencePoolSelector(selector map[v1alpha1.LabelKey]v1alpha1.LabelValue) labels.Selector {
+	return labels.SelectorFromSet(stripLabelKeyAliasFromLabelMap(selector))
+}
+
+func stripLabelKeyAliasFromLabelMap(labels map[v1alpha1.LabelKey]v1alpha1.LabelValue) map[string]string {
+	outMap := make(map[string]string)
+	for k, v := range labels {
+		outMap[string(k)] = string(v)
+	}
+	return outMap
+}

pkg/ext-proc/backend/endpointslice_reconciler.go

@@ -309,3 +309,24 @@ func populateServiceMap(services ...*v1alpha1.InferenceModel) *sync.Map {
 	}
 	return returnVal
 }
+
+func mapsEqual(map1, map2 *sync.Map) bool {
+	equal := true
+
+	map1.Range(func(k, v any) bool {
+		if _, ok := map2.Load(k); !ok {
+			equal = false
+			return false
+		}
+		return true
+	})
+	map2.Range(func(k, v any) bool {
+		if _, ok := map1.Load(k); !ok {
+			equal = false
+			return false
+		}
+		return true
+	})
+
+	return equal
+}