Merge pull request llm-d-incubation#190 from waltforme/use-cr

Begin to use custom resources

Author: MikeSpreitzer

charts/dpctlr/templates/role.yaml

@@ -6,6 +6,9 @@ metadata:
     app.kubernetes.io/component: controller
   name: dual-pods-controller
 rules:
+- apiGroups: [ "fma.llm-d.ai" ]
+  resources: [ inferenceserverconfigs, launcherconfigs ]
+  verbs: [ get, list, watch ]
 - apiGroups: [ "" ]
   resources: [ configmaps ]
   verbs: [ get, list, watch ]

cmd/dual-pods-controller/main.go

@@ -32,6 +32,8 @@ import (
 	"k8s.io/klog/v2"
 
 	dpctlr "github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/controller/dual-pods"
+	fmaclient "github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/generated/clientset/versioned"
+	fmainformers "github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/generated/informers/externalversions"
 )
 
 func main() {
@@ -83,17 +85,20 @@ func main() {
 
 	kubeClient := kubernetes.NewForConfigOrDie(restConfig)
 	kubePreInformers := kubeinformers.NewSharedInformerFactoryWithOptions(kubeClient, 0, kubeinformers.WithNamespace(overrides.Context.Namespace))
-
+	fmaClient := fmaclient.NewForConfigOrDie(restConfig)
+	fmaPreInformers := fmainformers.NewSharedInformerFactoryWithOptions(fmaClient, 0, fmainformers.WithNamespace(overrides.Context.Namespace))
 	ctlr, err := config.NewController(
 		logger,
 		kubeClient.CoreV1(),
 		overrides.Context.Namespace,
 		kubePreInformers.Core().V1(),
+		fmaPreInformers,
 	)
 	if err != nil {
 		klog.Fatal(err)
 	}
 	kubePreInformers.Start(ctx.Done())
+	fmaPreInformers.Start(ctx.Done())
 	err = ctlr.Start(ctx)
 	if err != nil {
 		klog.Fatal(err)

pkg/api/interface.go

@@ -41,9 +41,17 @@ package api
 // to define the server-providing Pod.
 // 1. Remove all annotations;
 // 2. Apply the patch
-
+// This annotation is the single source of truth to signal that the server-requesting Pod
+// is using the server patch to define its server-providing Pod.
+// This annotation is mutually exclusive with the 'InferenceServerConfigAnnotationName' annotation.
 const ServerPatchAnnotationName = "dual-pods.llm-d.ai/server-patch"
 
+// InferenceServerConfigAnnotationName is the name of an annotation on the
+// server-requesting Pod. The value of the annotation is the name of the
+// InferenceServerConfig object that the server-providing Pod uses.
+// This annotation is mutually exclusive with the 'ServerPatchAnnotationName' annotation.
+const InferenceServerConfigAnnotationName = "dual-pods.llm-d.ai/inference-server-config"
+
 // StatusAnnotationName is the name of an annotation that the dual-pods controller
 // maintains reporting the ServerRequestingPodStatus. The value of this annotation is the
 // JSON rendering of the status.

pkg/controller/dual-pods/controller.go

@@ -40,8 +40,11 @@ import (
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog/v2"
 
+	fmav1alpha1 "github.com/llm-d-incubation/llm-d-fast-model-actuation/api/fma/v1alpha1"
 	"github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/api"
 	genctlr "github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/controller/generic"
+	fmainformers "github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/generated/informers/externalversions"
+	fmalisters "github.com/llm-d-incubation/llm-d-fast-model-actuation/pkg/generated/listers/fma/v1alpha1"
 )
 
 // This package implements the dual-pods controller.
@@ -147,6 +150,7 @@ func (config ControllerConfig) NewController(
 	coreClient coreclient.CoreV1Interface,
 	namespace string,
 	corev1PreInformers corev1preinformers.Interface,
+	fmaInformerFactory fmainformers.SharedInformerFactory,
 ) (*controller, error) {
 	ctl := &controller{
 		enqueueLogger:       logger.WithName(ControllerName),
@@ -158,16 +162,21 @@ func (config ControllerConfig) NewController(
 		cmLister:            corev1PreInformers.ConfigMaps().Lister(),
 		nodeInformer:        corev1PreInformers.Nodes().Informer(),
 		nodeLister:          corev1PreInformers.Nodes().Lister(),
+		iscInformer:         fmaInformerFactory.Fma().V1alpha1().InferenceServerConfigs().Informer(),
+		iscLister:           fmaInformerFactory.Fma().V1alpha1().InferenceServerConfigs().Lister(),
+		lcInformer:          fmaInformerFactory.Fma().V1alpha1().LauncherConfigs().Informer(),
+		lcLister:            fmaInformerFactory.Fma().V1alpha1().LauncherConfigs().Lister(),
 		sleeperLimit:        config.SleeperLimit,
 		debugAccelMemory:    config.AcceleratorSleepingMemoryLimitMiB < math.MaxInt32,
 		accelMemoryLimitMiB: config.AcceleratorSleepingMemoryLimitMiB,
 		nodeNameToData:      map[string]*nodeData{},
 	}
 	ctl.gpuMap.Store(&map[string]GpuLocation{})
 	err := ctl.podInformer.AddIndexers(cache.Indexers{
-		requesterIndexName:   requesterIndexFunc,
-		nominalHashIndexName: nominalHashIndexFunc,
-		GPUIndexName:         GPUIndexFunc})
+		inferenceServerConfigIndexName: inferenceServerConfigIndexFunc,
+		requesterIndexName:             requesterIndexFunc,
+		nominalHashIndexName:           nominalHashIndexFunc,
+		GPUIndexName:                   GPUIndexFunc})
 	if err != nil { //impossible
 		return nil, err
 	}
@@ -194,6 +203,14 @@ func (config ControllerConfig) NewController(
 	if err != nil {
 		panic(err)
 	}
+	_, err = ctl.iscInformer.AddEventHandler(ctl)
+	if err != nil {
+		panic(err)
+	}
+	_, err = ctl.lcInformer.AddEventHandler(ctl)
+	if err != nil {
+		panic(err)
+	}
 	return ctl, nil
 }
 
@@ -207,6 +224,10 @@ type controller struct {
 	cmLister      corev1listers.ConfigMapLister
 	nodeInformer  cache.SharedIndexInformer
 	nodeLister    corev1listers.NodeLister
+	iscInformer   cache.SharedIndexInformer
+	iscLister     fmalisters.InferenceServerConfigLister
+	lcInformer    cache.SharedIndexInformer
+	lcLister      fmalisters.LauncherConfigLister
 	genctlr.KnowsProcessedSync[queueItem]
 
 	sleeperLimit        int
@@ -327,7 +348,7 @@ const (
 // The controller cares about server-requesting Pods, bound direct server-providing Pods, and launcher-based server-providing Pods.
 // The controller doesn't care about unbound direct providers and other Pods.
 func careAbout(pod *corev1.Pod) (item infSvrItem, it infSvrItemType) {
-	if len(pod.Annotations[api.ServerPatchAnnotationName]) > 0 {
+	if len(pod.Annotations[api.ServerPatchAnnotationName]) > 0 || len(pod.Annotations[api.InferenceServerConfigAnnotationName]) > 0 {
 		return infSvrItem{pod.UID, pod.Name}, infSvrItemRequester
 	}
 	requesterStr := pod.Annotations[requesterAnnotationKey]
@@ -338,6 +359,17 @@ func careAbout(pod *corev1.Pod) (item infSvrItem, it infSvrItemType) {
 	return infSvrItem{apitypes.UID(requesterParts[0]), requesterParts[1]}, infSvrItemBoundDirectProvider
 }
 
+const inferenceServerConfigIndexName = "inferenceserverconfig"
+
+func inferenceServerConfigIndexFunc(obj any) ([]string, error) {
+	pod := obj.(*corev1.Pod)
+	inferenceServerConfigName := pod.Annotations[api.InferenceServerConfigAnnotationName]
+	if len(inferenceServerConfigName) == 0 {
+		return []string{}, nil
+	}
+	return []string{inferenceServerConfigName}, nil
+}
+
 const requesterIndexName = "requester"
 
 func requesterIndexFunc(obj any) ([]string, error) {
@@ -373,6 +405,8 @@ func (ctl *controller) OnAdd(obj any, isInInitialList bool) {
 			nd.add(item)
 			ctl.Queue.Add(nodeItem{nodeName})
 		}
+	case *fmav1alpha1.InferenceServerConfig:
+		ctl.enqueueRequestersByInferenceServerConfig(typed, isInInitialList)
 	case *corev1.ConfigMap:
 		if typed.Name != GPUMapName {
 			ctl.enqueueLogger.V(5).Info("Ignoring ConfigMap that is not the GPU map", "ref", cache.MetaObjectToName(typed))
@@ -412,6 +446,8 @@ func (ctl *controller) OnUpdate(prev, obj any) {
 			nd.add(item)
 			ctl.Queue.Add(nodeItem{nodeName})
 		}
+	case *fmav1alpha1.InferenceServerConfig:
+		ctl.enqueueRequestersByInferenceServerConfig(typed, false)
 	case *corev1.ConfigMap:
 		if typed.Name != GPUMapName {
 			ctl.enqueueLogger.V(5).Info("Ignoring ConfigMap that is not the GPU map", "ref", cache.MetaObjectToName(typed))
@@ -454,6 +490,8 @@ func (ctl *controller) OnDelete(obj any) {
 			nd.add(item)
 			ctl.Queue.Add(nodeItem{nodeName})
 		}
+	case *fmav1alpha1.InferenceServerConfig:
+		ctl.enqueueRequestersByInferenceServerConfig(typed, false)
 	case *corev1.ConfigMap:
 		if typed.Name != GPUMapName {
 			ctl.enqueueLogger.V(5).Info("Ignoring ConfigMap that is not the GPU map", "ref", cache.MetaObjectToName(typed))
@@ -478,7 +516,7 @@ func getProviderNodeName(pod *corev1.Pod) (string, error) {
 }
 
 func (ctl *controller) Start(ctx context.Context) error {
-	if !cache.WaitForNamedCacheSync(ControllerName, ctx.Done(), ctl.cmInformer.HasSynced, ctl.podInformer.HasSynced, ctl.nodeInformer.HasSynced) {
+	if !cache.WaitForNamedCacheSync(ControllerName, ctx.Done(), ctl.cmInformer.HasSynced, ctl.podInformer.HasSynced, ctl.nodeInformer.HasSynced, ctl.iscInformer.HasSynced, ctl.lcInformer.HasSynced) {
 		return fmt.Errorf("caches not synced before end of Start context")
 	}
 	err := ctl.StartWorkers(ctx)
@@ -550,6 +588,37 @@ func (ctl *controller) enqueueRequesters(ctx context.Context) {
 	}
 }
 
+func (ctl *controller) enqueueRequestersByInferenceServerConfig(isc *fmav1alpha1.InferenceServerConfig, isInInitialList bool) {
+	inferenceServerConfigName := isc.Name
+	requesters, err := ctl.podInformer.GetIndexer().ByIndex(inferenceServerConfigIndexName, inferenceServerConfigName)
+	if err != nil {
+		ctl.enqueueLogger.Error(err, "Failed to get server requesting pods that use InferenceServerConfig", "ref", cache.MetaObjectToName(isc))
+		return
+	}
+	nodeNames := sets.New[string]()
+	for _, podObj := range requesters {
+		pod := podObj.(*corev1.Pod)
+		item, it := careAbout(pod)
+		if it != infSvrItemRequester {
+			// should not happen because of the nature of inferenceServerConfigIndexFunc
+			ctl.enqueueLogger.V(5).Info("Ignoring Pod that is not a server-requesting Pod", "pod", pod.Name)
+			continue
+		}
+		nodeName := pod.Spec.NodeName
+		if nodeName == "" {
+			ctl.enqueueLogger.V(5).Info("Ignoring non-scheduled server-requesting Pod that uses InferenceServerConfig", "pod", pod.Name, "inferenceServerConfigName", inferenceServerConfigName)
+			continue
+		}
+		nd := ctl.getNodeData(nodeName)
+		ctl.enqueueLogger.V(5).Info("Enqueuing inference server reference due to notification of InferenceServerConfig", "nodeName", nodeName, "item", item, "infSvrItemType", it, "inferenceServerConfigName", inferenceServerConfigName, "isInInitialList", isInInitialList, "resourceVersion", isc.ResourceVersion)
+		nd.add(item)
+		nodeNames.Insert(nodeName)
+	}
+	for nodeName := range nodeNames {
+		ctl.Queue.Add(nodeItem{nodeName})
+	}
+}
+
 func (ctl *controller) getNodeData(nodeName string) *nodeData {
 	ctl.mutex.Lock()
 	defer ctl.mutex.Unlock()

test/e2e/run.sh

@@ -2,7 +2,7 @@
 
 # Usage: $0
 # Current working directory must be the root of the Git repository.
-# The only reason for this is the `make` commands.
+# The reasons for this are the `make` commands and the location of CRDs.
 
 set -euo pipefail
 
@@ -67,12 +67,23 @@ kubectl get pods -A -o wide
 
 kubectl create clusterrole node-viewer --verb=get,list,watch --resource=nodes
 
+kubectl create -f ./config/crd/
+
 kubectl apply -f - <<EOF
 apiVersion: rbac.authorization.k8s.io/v1
 kind: Role
 metadata:
   name: testreq
 rules:
+- apiGroups:
+  - "fma.llm-d.ai"
+  resources:
+  - inferenceserverconfigs
+  - launcherconfigs
+  verbs:
+  - get
+  - list
+  - watch
 - apiGroups:
   - ""
   resourceNames: