chore: test to prod

common/ip_resolver.go

@@ -2,7 +2,6 @@ package common
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"log"
 	"sync"
@@ -191,6 +190,100 @@ func (resolver *K8sIPResolver) StopWatching() {
 	close(resolver.stopSignal)
 }
 
+// stripPod removes all fields from a Pod that the resolver doesn't need,
+// reducing the informer cache footprint by ~90% per pod object.
+func stripPod(obj interface{}) (interface{}, error) {
+	pod, ok := obj.(*v1.Pod)
+	if !ok {
+		return obj, nil
+	}
+	pod.ManagedFields = nil
+	pod.Annotations = nil
+	pod.Finalizers = nil
+	pod.Spec.Containers = nil
+	pod.Spec.InitContainers = nil
+	pod.Spec.Volumes = nil
+	pod.Spec.EphemeralContainers = nil
+	pod.Spec.ImagePullSecrets = nil
+	pod.Spec.Tolerations = nil
+	pod.Spec.Affinity = nil
+	pod.Spec.TopologySpreadConstraints = nil
+	pod.Status.Conditions = nil
+	pod.Status.ContainerStatuses = nil
+	pod.Status.InitContainerStatuses = nil
+	pod.Status.EphemeralContainerStatuses = nil
+	return pod, nil
+}
+
+// stripNode removes unneeded fields from a Node object.
+func stripNode(obj interface{}) (interface{}, error) {
+	node, ok := obj.(*v1.Node)
+	if !ok {
+		return obj, nil
+	}
+	node.ManagedFields = nil
+	node.Annotations = nil
+	node.Spec = v1.NodeSpec{}
+	node.Status.Conditions = nil
+	node.Status.Images = nil
+	node.Status.VolumesAttached = nil
+	node.Status.VolumesInUse = nil
+	node.Status.Capacity = nil
+	node.Status.Allocatable = nil
+	return node, nil
+}
+
+// stripService removes unneeded fields from a Service object.
+func stripService(obj interface{}) (interface{}, error) {
+	svc, ok := obj.(*v1.Service)
+	if !ok {
+		return obj, nil
+	}
+	svc.ManagedFields = nil
+	svc.Annotations = nil
+	// Keep: Name, Namespace, Spec.Selector, Spec.ClusterIPs
+	svc.Spec.Ports = nil
+	svc.Status = v1.ServiceStatus{}
+	return svc, nil
+}
+
+// stripOwnerOnly removes everything except ObjectMeta (for UID and OwnerReferences)
+// from resources where we only need the ownership chain (ReplicaSet, Deployment, etc.).
+func stripOwnerOnly(obj interface{}) (interface{}, error) {
+	type objectMetaAccessor interface {
+		GetObjectMeta() metav1.Object
+	}
+	if accessor, ok := obj.(objectMetaAccessor); ok {
+		meta := accessor.GetObjectMeta()
+		meta.SetManagedFields(nil)
+		meta.SetAnnotations(nil)
+		meta.SetLabels(nil)
+	}
+	// Type-specific spec/status clearing
+	switch o := obj.(type) {
+	case *appsv1.ReplicaSet:
+		o.Spec = appsv1.ReplicaSetSpec{}
+		o.Status = appsv1.ReplicaSetStatus{}
+		// Restore OwnerReferences (cleared spec removes nothing we need, they're in ObjectMeta)
+	case *appsv1.Deployment:
+		o.Spec = appsv1.DeploymentSpec{}
+		o.Status = appsv1.DeploymentStatus{}
+	case *appsv1.DaemonSet:
+		o.Spec = appsv1.DaemonSetSpec{}
+		o.Status = appsv1.DaemonSetStatus{}
+	case *appsv1.StatefulSet:
+		o.Spec = appsv1.StatefulSetSpec{}
+		o.Status = appsv1.StatefulSetStatus{}
+	case *batchv1.Job:
+		o.Spec = batchv1.JobSpec{}
+		o.Status = batchv1.JobStatus{}
+	case *batchv1.CronJob:
+		o.Spec = batchv1.CronJobSpec{}
+		o.Status = batchv1.CronJobStatus{}
+	}
+	return obj, nil
+}
+
 func (resolver *K8sIPResolver) StartWatching() error {
 	factory := informers.NewSharedInformerFactory(resolver.clientset, time.Minute*10)
 	// Define individual informers
@@ -204,6 +297,20 @@ func (resolver *K8sIPResolver) StartWatching() error {
 	serviceInformer := factory.Core().V1().Services().Informer()
 	deploymentInformer := factory.Apps().V1().Deployments().Informer()
 
+	// Strip unneeded fields from objects before they enter the informer cache.
+	// The informer stores full K8s objects internally, but we only need minimal
+	// fields. This reduces memory by ~80% for the informer cache (the biggest
+	// contributor on clusters with hundreds of pods/replicasets).
+	podInformer.SetTransform(stripPod)
+	nodeInformer.SetTransform(stripNode)
+	replicaSetInformer.SetTransform(stripOwnerOnly)
+	daemonSetInformer.SetTransform(stripOwnerOnly)
+	statefulSetInformer.SetTransform(stripOwnerOnly)
+	jobInformer.SetTransform(stripOwnerOnly)
+	cronJobInformer.SetTransform(stripOwnerOnly)
+	serviceInformer.SetTransform(stripService)
+	deploymentInformer.SetTransform(stripOwnerOnly)
+
 	resolver.addPodHandlers(podInformer)
 	resolver.addNodeHandlers(nodeInformer)
 	resolver.addReplicaSetHandlers(replicaSetInformer)
@@ -214,12 +321,22 @@ func (resolver *K8sIPResolver) StartWatching() error {
 	resolver.addServiceHandlers(serviceInformer)
 	resolver.addDeploymentHandlers(deploymentInformer)
 
-	// get initial state
-	err := resolver.getResolvedClusterSnapshot()
-	if err != nil {
-		return fmt.Errorf("error retrieving cluster's initial state: %v", err)
+	// Start informers and wait for initial list+watch sync.
+	// The event handlers populate resolver.snapshot.* as objects arrive.
+	factory.Start(resolver.stopSignal)
+	synced := factory.WaitForCacheSync(resolver.stopSignal)
+	for informerType, ok := range synced {
+		if !ok {
+			return fmt.Errorf("informer sync failed for %v", informerType)
+		}
 	}
-	factory.Start(resolver.stopSignal) // runs in background
+
+	// Re-process IP mappings in the correct priority order
+	// (services → nodes → pods) to handle IP collisions deterministically.
+	log.Printf("Informer cache synced, building IP mappings")
+	resolver.updateIpMapping()
+	log.Printf("IP mappings built")
+
 	return nil
 }
 
@@ -530,144 +647,6 @@ func (resolver *K8sIPResolver) handleNodeEvent(node *v1.Node) bool {
 	return false
 }
 
-func (resolver *K8sIPResolver) getResolvedClusterSnapshot() error {
-	log.Printf("Generating full cluster snapshot")
-	err := resolver.getFullClusterSnapshot()
-	if err != nil {
-		return err
-	}
-	resolver.updateIpMapping()
-	log.Printf("Generated full cluster snapshot")
-	return nil
-}
-
-// iterate the API for initial coverage of the cluster's state
-func (resolver *K8sIPResolver) getFullClusterSnapshot() error {
-	pods, err := resolver.clientset.CoreV1().Pods("").List(context.Background(), metav1.ListOptions{})
-	if err != nil {
-		return errors.New("error getting pods, aborting snapshot update")
-	}
-	log.Printf("loaded pods data %d", pods.Size())
-
-	for _, pod := range pods.Items {
-		minPod := MinimalPod{
-			UID:             pod.UID,
-			Name:            pod.Name,
-			Namespace:       pod.Namespace,
-			Labels:          pod.Labels,
-			NodeName:        pod.Spec.NodeName,
-			PodIPs:          pod.Status.PodIPs,
-			OwnerReferences: pod.ObjectMeta.OwnerReferences,
-		}
-		resolver.snapshot.Pods.Store(pod.UID, minPod)
-		resolver.snapshot.PodNameIndex.Store(pod.Namespace+"/"+pod.Name, pod.UID)
-	}
-
-	nodes, err := resolver.clientset.CoreV1().Nodes().List(context.Background(), metav1.ListOptions{})
-	if err != nil {
-		return errors.New("error getting nodes, aborting snapshot update")
-	}
-	for _, node := range nodes.Items {
-		// extract region and zone from attributes
-		region := node.Labels["topology.kubernetes.io/region"]
-		zone := node.Labels["topology.kubernetes.io/zone"]
-		meta := InstanceMeta{
-			Region:   region,
-			Zone:     zone,
-			Instance: node.Name,
-		}
-		resolver.instanceMetaMap.Store(node.Name, meta)
-		resolver.snapshot.Nodes.Store(node.UID, MinimalNode{
-			Name:      node.Name,
-			Labels:    node.Labels,
-			Addresses: node.Status.Addresses,
-		})
-	}
-
-	replicasets, err := resolver.clientset.AppsV1().ReplicaSets("").List(context.Background(), metav1.ListOptions{})
-	if err != nil {
-		return errors.New("error getting replicasets, aborting snapshot update")
-	}
-	for _, rs := range replicasets.Items {
-		resolver.snapshot.ReplicaSets.Store(rs.ObjectMeta.UID, MinimalOwnerInfo{
-			OwnerReferences: rs.OwnerReferences,
-		})
-	}
-
-	daemonsets, err := resolver.clientset.AppsV1().DaemonSets("").List(context.Background(), metav1.ListOptions{})
-	if err != nil {
-		return errors.New("error getting daemonsets, aborting snapshot update")
-	}
-	for _, ds := range daemonsets.Items {
-		resolver.snapshot.DaemonSets.Store(ds.ObjectMeta.UID, MinimalOwnerInfo{
-			OwnerReferences: ds.OwnerReferences,
-		})
-	}
-
-	statefulsets, err := resolver.clientset.AppsV1().StatefulSets("").List(context.Background(), metav1.ListOptions{})
-	if err != nil {
-		return errors.New("error getting statefulsets, aborting snapshot update")
-	}
-	for _, ss := range statefulsets.Items {
-		resolver.snapshot.StatefulSets.Store(ss.ObjectMeta.UID, MinimalOwnerInfo{
-			OwnerReferences: ss.OwnerReferences,
-		})
-	}
-
-	jobs, err := resolver.clientset.BatchV1().Jobs("").List(context.Background(), metav1.ListOptions{})
-	if err != nil {
-		return errors.New("error getting jobs, aborting snapshot update")
-	}
-	for _, job := range jobs.Items {
-		resolver.snapshot.Jobs.Store(job.ObjectMeta.UID, MinimalOwnerInfo{
-			OwnerReferences: job.OwnerReferences,
-		})
-	}
-
-	services, err := resolver.clientset.CoreV1().Services("").List(context.Background(), metav1.ListOptions{})
-	if err != nil {
-		return errors.New("error getting services, aborting snapshot update")
-	}
-	for _, service := range services.Items {
-		resolver.snapshot.Services.Store(service.UID, MinimalService{
-			Name:       service.Name,
-			Namespace:  service.Namespace,
-			Selector:   service.Spec.Selector,
-			ClusterIPs: service.Spec.ClusterIPs,
-		})
-	}
-
-	deployments, err := resolver.clientset.AppsV1().Deployments("").List(context.Background(), metav1.ListOptions{})
-	if err != nil {
-		return errors.New("error getting deployments, aborting snapshot update")
-	}
-	for _, deployment := range deployments.Items {
-		resolver.snapshot.Deployments.Store(deployment.UID, MinimalOwnerInfo{
-			OwnerReferences: deployment.OwnerReferences,
-		})
-	}
-
-	cronJobs, err := resolver.clientset.BatchV1().CronJobs("").List(context.Background(), metav1.ListOptions{})
-	if err != nil {
-		cronJobs, err := resolver.clientset.BatchV1beta1().CronJobs("").List(context.Background(), metav1.ListOptions{})
-		if err != nil {
-			return errors.New("error getting cronjobs, aborting snapshot update")
-		}
-		for _, cronJob := range cronJobs.Items {
-			resolver.snapshot.CronJobs.Store(cronJob.UID, MinimalOwnerInfo{
-				OwnerReferences: cronJob.OwnerReferences,
-			})
-		}
-	}
-	for _, cronJob := range cronJobs.Items {
-		resolver.snapshot.CronJobs.Store(cronJob.UID, MinimalOwnerInfo{
-			OwnerReferences: cronJob.OwnerReferences,
-		})
-	}
-
-	return nil
-}
-
 // add mapping from ip to resolved host to an existing map,
 // based on the given cluster snapshot
 func (resolver *K8sIPResolver) updateIpMapping() {