WIP

Author: kfswain

pkg/ext-proc/backend/k8s_client.go

@@ -1,30 +1,52 @@
 package backend
 
 import (
+	"context"
+	"fmt"
+
+	v1alpha1 "inference.networking.x-k8s.io/llm-instance-gateway/api/v1alpha1"
+	clientset "inference.networking.x-k8s.io/llm-instance-gateway/client-go/clientset/versioned"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
 )
 
 type K8sClient struct {
-	serverPoolName string
-	clientSet      *kubernetes.Clientset
-	labelSelector  string // Remove this when LLMServerPool is wired up.
+	serverPoolName      string
+	namespace           string
+	kubeClientSet       *kubernetes.Clientset
+	instanceGWClientSet clientset.Interface
+
+	labelSelector string // Remove this when LLMServerPool is wired up.
 }
 
-func (k *K8sClient) GetLLMServerPool() {
+func NewK8sClient(name, namespace string, k8sClient *kubernetes.Clientset, iGWClient clientset.Interface) K8sClient {
+	return K8sClient{
+		serverPoolName:      name,
+		namespace:           namespace,
+		kubeClientSet:       k8sClient,
+		instanceGWClientSet: iGWClient,
+	}
+}
 
+func (k *K8sClient) GetLLMServerPool() (*v1alpha1.LLMServerPool, error) {
+	llmServerPool, err := k.instanceGWClientSet.ApiV1alpha1().LLMServerPools(k.namespace).Get(context.TODO(), k.serverPoolName, v1.GetOptions{})
+	fmt.Print(llmServerPool.Name)
+	if err != nil {
+		fmt.Print("oh no.")
+		return nil, err
+	}
+	return llmServerPool, nil
 }
 
-func (k *K8sClient) GetPods(namespace string) {
-	podList, err := k.clentSet.CoreV1().Pods(namespace).List(context.TODO(), metav1.ListOptions{LabelSelector: k.labelSelector})
+func (k *K8sClient) GetPods() {
+	podList, err := k.kubeClientSet.CoreV1().Pods(k.namespace).List(context.TODO(), metav1.ListOptions{LabelSelector: k.labelSelector})
 	if err != nil {
-		//Cry about it.
+		// Handle err
 	}
 
-	for p in podList.Items {
+	for _, p := range podList.Items {
+		fmt.Print(p.Name)
 		// get IP and name
 	}
 }
-
-func (k *K8sClient) GetLLMServices() {
-
-}

pkg/ext-proc/main.go

@@ -115,6 +115,52 @@ 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)