Implementing pod and LLMServerPool controllers and

 adjusting provider to sync via controller datastore

Author: kfswain

pkg/ext-proc/backend/k8s_client.go

@@ -80,35 +80,3 @@ func (p *Provider) Init(refreshPodsInterval, refreshMetricsInterval time.Duratio
 
 	return nil
 }
-
-// refreshPodsOnce lists pods and updates keys in the podMetrics map.
-// Note this function doesn't update the PodMetrics value, it's done separately.
-func (p *Provider) refreshPodsOnce() error {
-	pods, err := p.pl.List()
-	if err != nil {
-		return err
-	}
-	// merge new pods with cached ones.
-	// add new pod to the map
-	for pod := range pods {
-		if _, ok := p.podMetrics.Load(pod); !ok {
-			new := &PodMetrics{
-				Pod: pod,
-				Metrics: Metrics{
-					ActiveModels: make(map[string]int),
-				},
-			}
-			p.podMetrics.Store(pod, new)
-		}
-	}
-	// remove pods that don't exist any more.
-	mergeFn := func(k, v any) bool {
-		pod := k.(Pod)
-		if _, ok := pods[pod]; !ok {
-			p.podMetrics.Delete(pod)
-		}
-		return true
-	}
-	p.podMetrics.Range(mergeFn)
-	return nil
-}

pkg/ext-proc/backend/provider.go

@@ -115,52 +115,6 @@ func main() {
 		}
 	}()
 
-	kubeInformerFactory := kubeinformers.NewSharedInformerFactory(kubeClientset, time.Second*30)
-	llmServerPoolInformerFactory := informers.NewSharedInformerFactory(iGWClientSet, time.Second*30)
-
-	llmServerPoolController := backend.NewLLMServerPoolController(
-		ctx,
-		*serverPoolName,
-		datastore,
-		kubeClientset,
-		llmServerPoolInformerFactory.Api().V1alpha1().LLMServerPools(),
-	)
-
-	podController := backend.NewPodController(
-		ctx,
-		*serverPoolName,
-		datastore,
-		kubeClientset,
-		kubeInformerFactory.Core().V1().Pods(),
-	)
-
-	kubeInformerFactory.Start(ctx.Done())
-	llmServerPoolInformerFactory.Start(ctx.Done())
-
-	errChan := make(chan error)
-	go func() {
-		if err = llmServerPoolController.Run(ctx, 1); err != nil {
-			klog.Error(err, "Error running LLMServerPoolController")
-			errChan <- err
-		}
-	}()
-
-	go func() {
-		if err = podController.Run(ctx, 2); err != nil {
-			klog.Error(err, "Error running PodController")
-			errChan <- err
-		}
-	}()
-
-	k8sClient := backend.NewK8sClient("test-pool", "default", kubeClientset, iGWClientSet)
-	serverPool, err := k8sClient.GetLLMServerPool()
-	if err != nil {
-		klog.Error(err.Error())
-	} else if serverPool != nil {
-		klog.Info(fmt.Printf("KELLEN TEST: %v", serverPool.Spec.ModelServerSelector.MatchLabels))
-	} else {
-		klog.Info("No err but serverPool is nil")
-	}
 	s := grpc.NewServer()
 
 	pp := backend.NewProvider(&backend.PodMetricsClientImpl{}, datastore)