Repository: kserve/kserve (language: python)

Issue:
LocalModelNode Agent
<!--  Thanks for sending a pull request!  Here are some tips for you:
1. If this is your first time, read our contributor guidelines https://www.kubeflow.org/docs/about/contributing/ and developer guide https://github.com/kserve/kserve/blob/master/docs/DEVELOPER_GUIDE.md
2. If you want *faster* PR reviews, read how: https://git.k8s.io/community/contributors/guide/pull-requests.md#best-practices-for-faster-reviews
3. Follow the instructions for writing a release note: https://git.k8s.io/community/contributors/guide/release-notes.md
4. If the PR is unfinished, see how to mark it: https://git.k8s.io/community/contributors/guide/pull-requests.md#marking-unfinished-pull-requests
-->

**What this PR does / why we need it**:
Implement local model node agent to reconcile the expected model cache list against the actual models in the local node volumes

**Which issue(s) this PR fixes** *(optional, in `fixes #<issue number>(, fixes #<issue_number>, ...)` format, will close the issue(s) when PR gets merged)*:
Fixes #

**Type of changes**
Please delete options that are not relevant.

- [ ] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to not work as expected)
- [ ] This change requires a documentation update

**Feature/Issue validation/testing**:

Please describe the tests that you ran to verify your changes and relevant result summary. Provide instructions so it can be reproduced.
Please also list any relevant details for your test configuration.

- [ ] Test A
- [ ] Test B

- Logs

**Special notes for your reviewer**:

1. Please confirm that if this PR changes any image versions, then that's the sole change this PR makes.

**Checklist**:

- [ ] Have you added unit/e2e tests that prove your fix is effective or that this feature works?
- [ ] Has code been commented, particularly in hard-to-understand areas?
- [ ] Have you made corresponding changes to the documentation?

**Release note**:
<!--  Write your release note:
1. Enter your extended release note in the below block. If the PR requires additional action from users switching to the new release, include the string "action required".
2. If no release note is required, just write "NONE".
-->
```release-note

```

**Re-running failed tests**

- `/rerun-all` - rerun all failed workflows.
- `/rerun-workflow <workflow name>` - rerun a specific failed workflow. Only one workflow name can be specified. Multiple /rerun-workflow commands are allowed per comment.


Discussion / hints:

Fix patch:
diff --git a/charts/kserve-resources/README.md b/charts/kserve-resources/README.md
index 60762bcbf08..a83a4abc5a4 100644
--- a/charts/kserve-resources/README.md
+++ b/charts/kserve-resources/README.md
@@ -55,6 +55,10 @@ $ helm install kserve oci://ghcr.io/kserve/charts/kserve --version v0.14.0
 | kserve.controller.tag | string | `"v0.14.0"` | KServe controller contrainer image tag. |
 | kserve.controller.tolerations | list | `[]` | A list of Kubernetes Tolerations, if required. For more information, see [Toleration v1 core](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.27/#toleration-v1-core).  For example:   tolerations:   - key: foo.bar.com/role     operator: Equal     value: master     effect: NoSchedule |
 | kserve.controller.topologySpreadConstraints | list | `[]` | A list of Kubernetes TopologySpreadConstraints, if required. For more information, see [Topology spread constraint v1 core](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.27/#topologyspreadconstraint-v1-core  For example:   topologySpreadConstraints:   - maxSkew: 2     topologyKey: topology.kubernetes.io/zone     whenUnsatisfiable: ScheduleAnyway     labelSelector:       matchLabels:         app.kubernetes.io/instance: kserve-controller-manager         app.kubernetes.io/component: controller |
+| kserve.localmodel.agent.hostPath | string | `"/mnt/models"` |  |
+| kserve.localmodel.agent.image | string | `"kserve/kserve-localmodelnode-agent"` |  |
+| kserve.localmodel.agent.nodeSelector | object | `{}` |  |
+| kserve.localmodel.agent.tag | string | `"v0.14.0"` |  |
 | kserve.localmodel.controller.image | string | `"kserve/kserve-localmodel-controller"` |  |
 | kserve.localmodel.controller.tag | string | `"v0.14.0"` |  |
 | kserve.localmodel.enabled | bool | `false` |  |
diff --git a/charts/kserve-resources/templates/localmodel/role.yaml b/charts/kserve-resources/templates/localmodel/role.yaml
index bdefd53c3d3..8a7e3d2f3aa 100644
--- a/charts/kserve-resources/templates/localmodel/role.yaml
+++ b/charts/kserve-resources/templates/localmodel/role.yaml
@@ -63,7 +63,6 @@ rules:
 - apiGroups:
   - serving.kserve.io
   resources:
-  - clusterstoragecontainers
   - inferenceservices
   - localmodelnodegroups
   verbs:
diff --git a/charts/kserve-resources/templates/localmodelnode/deployment.yaml b/charts/kserve-resources/templates/localmodelnode/daemonset.yaml
similarity index 75%
rename from charts/kserve-resources/templates/localmodelnode/deployment.yaml
rename to charts/kserve-resources/templates/localmodelnode/daemonset.yaml
index bb3adb65984..e5512a15be4 100644
--- a/charts/kserve-resources/templates/localmodelnode/deployment.yaml
+++ b/charts/kserve-resources/templates/localmodelnode/daemonset.yaml
@@ -23,6 +23,10 @@ spec:
       annotations:
         kubectl.kubernetes.io/default-container: manager
     spec:
+      {{- with .Values.kserve.localmodel.agent.nodeSelector }}
+      nodeSelector:
+        {{- toYaml . | nindent 8 }}
+      {{- end }}
       serviceAccountName: kserve-localmodelnode-agent
       securityContext:
         runAsNonRoot: true
@@ -45,6 +49,14 @@ spec:
             valueFrom:
               fieldRef:
                 fieldPath: metadata.namespace
+          - name: NODE_NAME
+            valueFrom:
+              fieldRef:
+                fieldPath: spec.nodeName
+        volumeMounts:
+          - mountPath: /mnt/models
+            name: models
+            readOnly: false
         resources:
           limits:
             cpu: 100m
@@ -52,5 +64,10 @@ spec:
           requests:
             cpu: 100m
             memory: 200Mi
+      volumes:
+        - name: models
+          hostPath:
+            path: {{ .Values.kserve.localmodel.agent.hostPath }}
+            type: DirectoryOrCreate
       terminationGracePeriodSeconds: 10
 {{- end }}
diff --git a/charts/kserve-resources/templates/localmodelnode/role.yaml b/charts/kserve-resources/templates/localmodelnode/role.yaml
index 4c3924a650c..ecd67842054 100644
--- a/charts/kserve-resources/templates/localmodelnode/role.yaml
+++ b/charts/kserve-resources/templates/localmodelnode/role.yaml
@@ -26,18 +26,6 @@ rules:
   verbs:
   - get
   - watch
-- apiGroups:
-  - ""
-  resources:
-  - persistentvolumeclaims
-  verbs:
-  - create
-  - delete
-  - get
-  - list
-  - patch
-  - update
-  - watch
 - apiGroups:
   - batch
   resources:
@@ -64,7 +52,6 @@ rules:
   verbs:
   - get
   - list
-  - watch
 - apiGroups:
   - serving.kserve.io
   resources:
diff --git a/charts/kserve-resources/values.yaml b/charts/kserve-resources/values.yaml
index bff2b2a5e76..eb1a482d145 100644
--- a/charts/kserve-resources/values.yaml
+++ b/charts/kserve-resources/values.yaml
@@ -403,5 +403,10 @@ kserve:
       jobNamespace: kserve-localmodel-jobs
       securityContext:
         FSGroup: 1000
+      agent:
+        nodeSelector: {}
+        hostPath: /mnt/models
+        image: kserve/kserve-localmodelnode-agent
+        tag: *defaultVersion
   security:
     autoMountServiceAccountToken: true
diff --git a/config/localmodelnodes/manager.yaml b/config/localmodelnodes/manager.yaml
index cb0e3590b24..da69173a992 100644
--- a/config/localmodelnodes/manager.yaml
+++ b/config/localmodelnodes/manager.yaml
@@ -21,6 +21,8 @@ spec:
       annotations:
         kubectl.kubernetes.io/default-container: manager
     spec:
+      nodeSelector:
+        kserve/localmodel: worker
       serviceAccountName: kserve-localmodelnode-agent
       securityContext:
         runAsNonRoot: true
@@ -43,6 +45,14 @@ spec:
             valueFrom:
               fieldRef:
                 fieldPath: metadata.namespace
+          - name: NODE_NAME
+            valueFrom:
+              fieldRef:
+                fieldPath: spec.nodeName
+        volumeMounts:
+          - mountPath: /mnt/models
+            name: models
+            readOnly: false
         resources:
           limits:
             cpu: 100m
@@ -50,4 +60,9 @@ spec:
           requests:
             cpu: 100m
             memory: 200Mi
+      volumes:
+        - name: models
+          hostPath:
+            path: /models
+            type: DirectoryOrCreate
       terminationGracePeriodSeconds: 10
diff --git a/config/rbac/localmodel/role.yaml b/config/rbac/localmodel/role.yaml
index 27e47536db3..c9c31f5688f 100644
--- a/config/rbac/localmodel/role.yaml
+++ b/config/rbac/localmodel/role.yaml
@@ -62,7 +62,6 @@ rules:
 - apiGroups:
   - serving.kserve.io
   resources:
-  - clusterstoragecontainers
   - inferenceservices
   - localmodelnodegroups
   verbs:
diff --git a/config/rbac/localmodelnode/role.yaml b/config/rbac/localmodelnode/role.yaml
index edf0050acf4..6996b160878 100644
--- a/config/rbac/localmodelnode/role.yaml
+++ b/config/rbac/localmodelnode/role.yaml
@@ -25,18 +25,6 @@ rules:
   verbs:
   - get
   - watch
-- apiGroups:
-  - ""
-  resources:
-  - persistentvolumeclaims
-  verbs:
-  - create
-  - delete
-  - get
-  - list
-  - patch
-  - update
-  - watch
 - apiGroups:
   - batch
   resources:
@@ -63,7 +51,6 @@ rules:
   verbs:
   - get
   - list
-  - watch
 - apiGroups:
   - serving.kserve.io
   resources:
diff --git a/pkg/controller/v1alpha1/localmodel/controller.go b/pkg/controller/v1alpha1/localmodel/controller.go
index d3cec408050..eda4db3efd6 100644
--- a/pkg/controller/v1alpha1/localmodel/controller.go
+++ b/pkg/controller/v1alpha1/localmodel/controller.go
@@ -20,7 +20,6 @@ limitations under the License.
 // +kubebuilder:rbac:groups=serving.kserve.io,resources=clusterlocalmodels/status,verbs=get;update;patch
 // +kubebuilder:rbac:groups=serving.kserve.io,resources=localmodelnodes,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=serving.kserve.io,resources=localmodelnodes/status,verbs=get;watch
-// +kubebuilder:rbac:groups=serving.kserve.io,resources=clusterstoragecontainers,verbs=get;list;watch
 // +kubebuilder:rbac:groups=core,resources=configmaps,verbs=get
 // +kubebuilder:rbac:groups=core,resources=nodes,verbs=get;list;watch
 // +kubebuilder:rbac:groups=core,resources=nodes/status,verbs=get;watch
diff --git a/pkg/controller/v1alpha1/localmodelnode/controller.go b/pkg/controller/v1alpha1/localmodelnode/controller.go
index 59cf23041dc..0a4404421d4 100644
--- a/pkg/controller/v1alpha1/localmodelnode/controller.go
+++ b/pkg/controller/v1alpha1/localmodelnode/controller.go
@@ -14,8 +14,8 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-// +kubebuilder:rbac:groups=serving.kserve.io,resources=localmodelnodegroups,verbs=get;list;watch
-// +kubebuilder:rbac:groups=serving.kserve.io,resources=clusterstoragecontainers,verbs=get;list;watch
+// +kubebuilder:rbac:groups=serving.kserve.io,resources=localmodelnodegroups,verbs=get;list
+// +kubebuilder:rbac:groups=serving.kserve.io,resources=clusterstoragecontainers,verbs=get;list
 // +kubebuilder:rbac:groups=serving.kserve.io,resources=localmodelnodes,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=serving.kserve.io,resources=localmodelnodes/status,verbs=get;update;patch
 // +kubebuilder:rbac:groups=core,resources=configmaps,verbs=get
@@ -23,18 +23,26 @@ limitations under the License.
 // +kubebuilder:rbac:groups=core,resources=nodes/status,verbs=get;watch
 // +kubebuilder:rbac:groups=batch,resources=jobs,verbs=get;list;watch;create;update;patch;delete
 // +kubebuilder:rbac:groups=batch,resources=jobs/status,verbs=get
-// +kubebuilder:rbac:groups=core,resources=persistentvolumeclaims,verbs=get;list;watch;create;update;patch;delete
 package localmodelnode
 
 import (
 	"context"
+	"fmt"
+	"os"
+	"reflect"
 
 	"github.com/go-logr/logr"
 	v1alpha1api "github.com/kserve/kserve/pkg/apis/serving/v1alpha1"
+	"github.com/kserve/kserve/pkg/apis/serving/v1beta1"
+	batchv1 "k8s.io/api/batch/v1"
+	v1 "k8s.io/api/core/v1"
+	apierr "k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
 
@@ -45,13 +53,214 @@ type LocalModelNodeReconciler struct {
 	Scheme    *runtime.Scheme
 }
 
+const (
+	MountPath             = "/mnt/models" // Volume mount path for models, must be the same as the value in the DaemonSet spec
+	DownloadContainerName = "kserve-localmodel-download"
+	PvcSourceMountName    = "kserve-pvc-source"
+)
+
+var (
+	defaultJobImage = "kserve/storage-initializer:latest" // Can be overwritten by the value in the configmap
+	FSGroup         *int64
+	jobNamespace    string
+	nodeName        = os.Getenv("NODE_NAME") // Name of current node, passed as an env variable via downward API
+)
+
+// Launch a new job or return an existing job
+func (c *LocalModelNodeReconciler) launchJob(ctx context.Context, jobName string, localModelNode *v1alpha1api.LocalModelNode, modelInfo v1alpha1api.LocalModelInfo, claimName string) (*batchv1.Job, error) {
+	container, err := c.getContainerSpecForStorageUri(ctx, modelInfo.SourceModelUri)
+	if err != nil {
+		return nil, err
+	}
+	jobs := c.Clientset.BatchV1().Jobs(jobNamespace)
+
+	job, err := jobs.Get(ctx, jobName, metav1.GetOptions{})
+
+	// In tests, job is an empty struct, using this bool is easier than checking for empty struct
+	jobFound := true
+	if err != nil {
+		if apierr.IsNotFound(err) {
+			jobFound = false
+		} else {
+			c.Log.Error(err, "Failed to get job", "name", jobName)
+			return job, err
+		}
+	}
+
+	container.Args = []string{modelInfo.SourceModelUri, MountPath}
+	container.VolumeMounts = []v1.VolumeMount{
+		{
+			MountPath: MountPath,
+			Name:      PvcSourceMountName,
+			ReadOnly:  false,
+			SubPath:   "models/" + modelInfo.ModelName,
+		},
+	}
+	expectedJob := &batchv1.Job{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      jobName + "dryrun",
+			Namespace: jobNamespace,
+		},
+		Spec: batchv1.JobSpec{
+			Template: v1.PodTemplateSpec{
+				Spec: v1.PodSpec{
+					NodeName:      nodeName,
+					Containers:    []v1.Container{*container},
+					RestartPolicy: v1.RestartPolicyNever,
+					Volumes: []v1.Volume{
+						{
+							Name: PvcSourceMountName,
+							VolumeSource: v1.VolumeSource{
+								PersistentVolumeClaim: &v1.PersistentVolumeClaimVolumeSource{
+									ClaimName: claimName,
+								},
+							},
+						},
+					},
+					SecurityContext: &v1.PodSecurityContext{
+						FSGroup: FSGroup,
+					},
+				},
+			},
+		},
+	}
+	dryrunJob, err := jobs.Create(ctx, expectedJob, metav1.CreateOptions{DryRun: []string{metav1.DryRunAll}})
+	if err != nil {
+		return nil, err
+	}
+	if job != nil && reflect.DeepEqual(job.Spec.Template.Spec, dryrunJob.Spec.Template.Spec) {
+		return job, nil
+	}
+	if jobFound {
+		bg := metav1.DeletePropagationBackground
+		err = jobs.Delete(ctx, job.Name, metav1.DeleteOptions{
+			PropagationPolicy: &bg,
+		})
+		if err != nil {
+			c.Log.Error(err, "Failed to delete job.", "name", job.Name)
+			return nil, err
+		}
+	}
+
+	if err := controllerutil.SetControllerReference(localModelNode, expectedJob, c.Scheme); err != nil {
+		c.Log.Error(err, "Failed to set controller reference", "name", modelInfo.ModelName)
+		return nil, err
+	}
+	expectedJob.Name = jobName
+	job, err = jobs.Create(ctx, expectedJob, metav1.CreateOptions{})
+	c.Log.Info("Creating job", "name", job.Name, "namespace", job.Namespace)
+	if err != nil {
+		c.Log.Error(err, "Failed to create job.", "name", expectedJob.Name)
+		return nil, err
+	}
+	return job, err
+}
+
+// Fetches container spec for model download container, use the default KServe image if not found
+func (c *LocalModelNodeReconciler) getContainerSpecForStorageUri(ctx context.Context, storageUri string) (*v1.Container, error) {
+	storageContainers := &v1alpha1api.ClusterStorageContainerList{}
+	if err := c.Client.List(ctx, storageContainers); err != nil {
+		return nil, err
+	}
+
+	for _, sc := range storageContainers.Items {
+		if sc.IsDisabled() {
+			continue
+		}
+		if sc.Spec.WorkloadType != v1alpha1api.LocalModelDownloadJob {
+			continue
+		}
+		supported, err := sc.Spec.IsStorageUriSupported(storageUri)
+		if err != nil {
+			return nil, fmt.Errorf("error checking storage container %s: %w", sc.Name, err)
+		}
+		if supported {
+			return &sc.Spec.Container, nil
+		}
+	}
+
+	defaultContainer := &v1.Container{
+		Name:                     DownloadContainerName,
+		Image:                    defaultJobImage,
+		TerminationMessagePolicy: v1.TerminationMessageFallbackToLogsOnError,
+	}
+	return defaultContainer, nil
+}
+
+func (c *LocalModelNodeReconciler) downloadModels(ctx context.Context, localModelNode *v1alpha1api.LocalModelNode) error {
+	c.Log.Info("Downloading models to", "node", localModelNode.ObjectMeta.Name)
+
+	for _, modelInfo := range localModelNode.Spec.LocalModels {
+		if status, ok := localModelNode.Status.ModelStatus[modelInfo.ModelName]; ok {
+			if status == v1alpha1api.ModelDownloaded {
+				continue
+			}
+		}
+		jobName := modelInfo.ModelName + "-" + localModelNode.ObjectMeta.Name
+
+		job, err := c.launchJob(ctx, jobName, localModelNode, modelInfo, modelInfo.ModelName)
+		if err != nil {
+			c.Log.Error(err, "Job error", "name", jobName)
+			return err
+		}
+
+		localModelNode.Status.ModelStatus = map[string]v1alpha1api.ModelStatus{}
+		switch {
+		case job.Status.Succeeded > 0:
+			localModelNode.Status.ModelStatus[modelInfo.ModelName] = v1alpha1api.ModelDownloaded
+		case job.Status.Failed > 0:
+			localModelNode.Status.ModelStatus[modelInfo.ModelName] = v1alpha1api.ModelDownloadError
+		case job.Status.Ready != nil && *job.Status.Ready > 0:
+			localModelNode.Status.ModelStatus[modelInfo.ModelName] = v1alpha1api.ModelDownloading
+		default:
+			localModelNode.Status.ModelStatus[modelInfo.ModelName] = v1alpha1api.ModelDownloadPending
+		}
+	}
+
+	if err := c.Status().Update(ctx, localModelNode); err != nil {
+		c.Log.Error(err, "Update local model cache status error", "name", localModelNode.Name)
+		return err
+	}
+
+	return nil
+}
+
 func (c *LocalModelNodeReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-	c.Log.Info("Daemonset controller reconciling localmodelnode", "name", req.Name)
+	if req.Name != nodeName {
+		c.Log.Info("Skipping LocalModelNode because it is not for current node", "name", req.Name, "current node", nodeName)
+		return reconcile.Result{}, nil
+	}
+
+	c.Log.Info("Agent reconciling LocalModelNode", "name", req.Name, "node", nodeName)
+	// Create Jobs to download models if the model is not present locally.
+	// 1. Check if LocalModelNode CR is for current node
+	localModelNode := &v1alpha1api.LocalModelNode{}
+	if err := c.Get(ctx, req.NamespacedName, localModelNode); err != nil {
+		c.Log.Error(err, "Error getting LocalModelNode", "name", req.Name)
+		return reconcile.Result{}, err
+	}
+
+	// 2. Kick off download jobs for all models in spec
+	localModelConfig, err := v1beta1.NewLocalModelConfig(c.Clientset)
+	if err != nil {
+		c.Log.Error(err, "Failed to get local model config")
+		return reconcile.Result{}, err
+	}
+	defaultJobImage = localModelConfig.DefaultJobImage
+	jobNamespace = localModelConfig.JobNamespace
+	FSGroup = localModelConfig.FSGroup
+	if err := c.downloadModels(ctx, localModelNode); err != nil {
+		c.Log.Error(err, "Model download err")
+		return reconcile.Result{}, err
+	}
+
+	// Todo: Delete models that are not in the spec
 	return reconcile.Result{}, nil
 }
 
 func (c *LocalModelNodeReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&v1alpha1api.LocalModelNode{}).
+		Owns(&batchv1.Job{}).
 		Complete(c)
 }


Test patch:
diff --git a/pkg/controller/v1alpha1/localmodel/controller_test.go b/pkg/controller/v1alpha1/localmodel/controller_test.go
index 769a1742b29..6975db584af 100644
--- a/pkg/controller/v1alpha1/localmodel/controller_test.go
+++ b/pkg/controller/v1alpha1/localmodel/controller_test.go
@@ -101,7 +101,7 @@ var _ = Describe("CachedModel controller", func() {
 		}
 	)
 	Context("When creating a local model", func() {
-		It("Should create pv, pvc, and download jobs", func() {
+		It("Should create pv, pvc, localmodelnode, and update status from localmodelnode", func() {
 			var configMap = &v1.ConfigMap{
 				ObjectMeta: metav1.ObjectMeta{
 					Name:      constants.InferenceServiceConfigMapName,
@@ -440,9 +440,8 @@ var _ = Describe("CachedModel controller", func() {
 				localModelNode := &v1alpha1.LocalModelNode{}
 				Eventually(func() bool {
 					err := k8sClient.Get(ctx, types.NamespacedName{Name: node.Name}, localModelNode)
-					return err == nil
+					return err == nil && len(localModelNode.Spec.LocalModels) == 2
 				}, timeout, interval).Should(BeTrue())
-				Expect(localModelNode.Spec.LocalModels).Should(ConsistOf(v1alpha1.LocalModelInfo{ModelName: cachedModel1.Name, SourceModelUri: sourceModelUri}, v1alpha1.LocalModelInfo{ModelName: cachedModel2.Name, SourceModelUri: sourceModelUri}))
 			}
 
 			// Now let's test deletion - delete one model and expect one model exists
diff --git a/pkg/controller/v1alpha1/localmodelnode/controller_test.go b/pkg/controller/v1alpha1/localmodelnode/controller_test.go
new file mode 100644
index 00000000000..d3319e71552
--- /dev/null
+++ b/pkg/controller/v1alpha1/localmodelnode/controller_test.go
@@ -0,0 +1,173 @@
+/*
+Copyright 2024 The KServe Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package localmodelnode
+
+import (
+	"context"
+	"time"
+
+	"github.com/kserve/kserve/pkg/apis/serving/v1alpha1"
+	"github.com/kserve/kserve/pkg/constants"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	batchv1 "k8s.io/api/batch/v1"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/resource"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/utils/ptr"
+)
+
+var _ = Describe("CachedModel controller", func() {
+	const (
+		timeout             = time.Second * 10
+		duration            = time.Second * 10
+		interval            = time.Millisecond * 250
+		modelCacheNamespace = "kserve-localmodel-jobs"
+		sourceModelUri      = "s3://mybucket/mymodel"
+	)
+	var (
+		modelName          = "iris"
+		localModelNodeSpec = v1alpha1.LocalModelNodeSpec{
+			LocalModels: []v1alpha1.LocalModelInfo{
+				{
+					SourceModelUri: sourceModelUri,
+					ModelName:      modelName,
+				},
+			},
+		}
+		clusterStorageContainerSpec = v1alpha1.StorageContainerSpec{
+			SupportedUriFormats: []v1alpha1.SupportedUriFormat{{Prefix: "s3://"}},
+			Container: v1.Container{
+				Name:  "name",
+				Image: "image",
+				Args: []string{
+					"srcURI",
+					constants.DefaultModelLocalMountPath,
+				},
+				TerminationMessagePolicy: v1.TerminationMessageFallbackToLogsOnError,
+				VolumeMounts:             []v1.VolumeMount{},
+			},
+		}
+		localModelNodeGroupSpec = v1alpha1.LocalModelNodeGroupSpec{
+			PersistentVolumeSpec: v1.PersistentVolumeSpec{
+				AccessModes:                   []v1.PersistentVolumeAccessMode{v1.ReadWriteOnce},
+				VolumeMode:                    ptr.To(v1.PersistentVolumeFilesystem),
+				Capacity:                      v1.ResourceList{v1.ResourceStorage: resource.MustParse("2Gi")},
+				StorageClassName:              "standard",
+				PersistentVolumeReclaimPolicy: v1.PersistentVolumeReclaimDelete,
+				PersistentVolumeSource: v1.PersistentVolumeSource{
+					HostPath: &v1.HostPathVolumeSource{
+						Path: "/models",
+						Type: ptr.To(v1.HostPathDirectory),
+					},
+				},
+				NodeAffinity: &v1.VolumeNodeAffinity{
+					Required: &v1.NodeSelector{
+						NodeSelectorTerms: []v1.NodeSelectorTerm{
+							{
+								MatchExpressions: []v1.NodeSelectorRequirement{
+									{
+										Key:      "node.kubernetes.io/instance-type",
+										Operator: v1.NodeSelectorOpIn,
+										Values:   []string{"gpu"},
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			PersistentVolumeClaimSpec: v1.PersistentVolumeClaimSpec{
+				AccessModes: []v1.PersistentVolumeAccessMode{v1.ReadWriteOnce},
+				Resources:   v1.VolumeResourceRequirements{Requests: v1.ResourceList{v1.ResourceStorage: resource.MustParse("2Gi")}},
+			},
+		}
+		configs = map[string]string{
+			"localModel": `{
+        		"jobNamespace": "kserve-localmodel-jobs",
+                "defaultJobImage": "kserve/storage-initializer:latest"
+            }`,
+		}
+	)
+	Context("When creating a local model", func() {
+		It("Should create download jobs and update model status from jobs", func() {
+			var configMap = &v1.ConfigMap{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      constants.InferenceServiceConfigMapName,
+					Namespace: constants.KServeNamespace,
+				},
+				Data: configs,
+			}
+			Expect(k8sClient.Create(context.TODO(), configMap)).NotTo(HaveOccurred())
+			defer k8sClient.Delete(context.TODO(), configMap)
+
+			clusterStorageContainer := &v1alpha1.ClusterStorageContainer{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "test",
+				},
+				Spec: clusterStorageContainerSpec,
+			}
+			Expect(k8sClient.Create(ctx, clusterStorageContainer)).Should(Succeed())
+			defer k8sClient.Delete(ctx, clusterStorageContainer)
+
+			nodeGroup := &v1alpha1.LocalModelNodeGroup{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: "gpu",
+				},
+				Spec: localModelNodeGroupSpec,
+			}
+			Expect(k8sClient.Create(ctx, nodeGroup)).Should(Succeed())
+			defer k8sClient.Delete(ctx, nodeGroup)
+
+			nodeName = "worker"
+			localModelNode := &v1alpha1.LocalModelNode{
+				ObjectMeta: metav1.ObjectMeta{
+					Name: nodeName,
+				},
+				Spec: localModelNodeSpec,
+			}
+			Expect(k8sClient.Create(ctx, localModelNode)).Should(Succeed())
+			defer k8sClient.Delete(ctx, localModelNode)
+
+			// Wait for the download job to be created
+			job := &batchv1.Job{}
+			Eventually(func() bool {
+				err := k8sClient.Get(ctx, types.NamespacedName{Name: modelName + "-" + nodeName, Namespace: modelCacheNamespace}, job)
+				return err == nil
+			}, timeout, interval).Should(BeTrue(), "Download job should be created")
+
+			// Now let's update the job status to be successful
+			job.Status.Succeeded = 1
+			Expect(k8sClient.Status().Update(ctx, job)).Should(Succeed())
+
+			// LocalModelNode status should be updated
+			Eventually(func() bool {
+				err := k8sClient.Get(ctx, types.NamespacedName{Name: nodeName}, localModelNode)
+				if err != nil {
+					return false
+				}
+				modelStatus, ok := localModelNode.Status.ModelStatus[modelName]
+				if !ok {
+					return false
+				}
+				return modelStatus == v1alpha1.ModelDownloaded
+			}, timeout, interval).Should(BeTrue(), "LocaModelNode status should be downloaded")
+			Expect(localModelNode.Spec).Should(Equal(localModelNodeSpec), "spec should not be changed")
+		})
+	})
+})
diff --git a/pkg/controller/v1alpha1/localmodelnode/suite_test.go b/pkg/controller/v1alpha1/localmodelnode/suite_test.go
new file mode 100644
index 00000000000..8cc7b9fe126
--- /dev/null
+++ b/pkg/controller/v1alpha1/localmodelnode/suite_test.go
@@ -0,0 +1,133 @@
+/*
+Copyright 2024 The KServe Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package localmodelnode
+
+import (
+	"context"
+	"path/filepath"
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	"k8s.io/client-go/rest"
+	knservingv1 "knative.dev/serving/pkg/apis/serving/v1"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
+	logf "sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/log/zap"
+	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
+
+	"github.com/kserve/kserve/pkg/apis/serving/v1alpha1"
+	kservev1beta1 "github.com/kserve/kserve/pkg/apis/serving/v1beta1"
+	"github.com/kserve/kserve/pkg/constants"
+	pkgtest "github.com/kserve/kserve/pkg/testing"
+	// +kubebuilder:scaffold:imports
+)
+
+// These tests use Ginkgo (BDD-style Go testing framework). Refer to
+// http://onsi.github.io/ginkgo/ to learn more about Ginkgo.
+
+var (
+	cfg       *rest.Config
+	k8sClient client.Client
+	testEnv   *envtest.Environment
+	cancel    context.CancelFunc
+	ctx       context.Context
+)
+
+func TestAPIs(t *testing.T) {
+	RegisterFailHandler(Fail)
+
+	RunSpecs(t, "v1alpha1 Controller Suite")
+}
+
+var _ = BeforeSuite(func() {
+	logf.SetLogger(zap.New(zap.WriteTo(GinkgoWriter), zap.UseDevMode(true)))
+	ctx, cancel = context.WithCancel(context.TODO())
+	By("bootstrapping test environment")
+	crdDirectoryPaths := []string{
+		filepath.Join("..", "..", "..", "..", "test", "crds"),
+	}
+	testEnv = pkgtest.SetupEnvTest(crdDirectoryPaths)
+	var err error
+	cfg, err = testEnv.Start()
+	Expect(err).ToNot(HaveOccurred())
+	Expect(cfg).ToNot(BeNil())
+
+	err = v1alpha1.AddToScheme(scheme.Scheme)
+	Expect(err).NotTo(HaveOccurred())
+	err = kservev1beta1.AddToScheme(scheme.Scheme)
+	Expect(err).NotTo(HaveOccurred())
+	err = knservingv1.AddToScheme(scheme.Scheme)
+	Expect(err).NotTo(HaveOccurred())
+	// +kubebuilder:scaffold:scheme
+
+	k8sClient, err = client.New(cfg, client.Options{Scheme: scheme.Scheme})
+	Expect(err).ToNot(HaveOccurred())
+	Expect(k8sClient).ToNot(BeNil())
+
+	clientset, err := kubernetes.NewForConfig(cfg)
+	Expect(err).ToNot(HaveOccurred())
+	Expect(clientset).ToNot(BeNil())
+
+	// Creates namespace
+	kserveNamespaceObj := &v1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: constants.KServeNamespace,
+		},
+	}
+	jobsNamespaceObj := &v1.Namespace{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "kserve-localmodel-jobs",
+		},
+	}
+	Expect(k8sClient.Create(context.Background(), kserveNamespaceObj)).Should(Succeed())
+	Expect(k8sClient.Create(context.Background(), jobsNamespaceObj)).Should(Succeed())
+
+	k8sManager, err := ctrl.NewManager(cfg, ctrl.Options{
+		Scheme: scheme.Scheme,
+		Metrics: metricsserver.Options{
+			BindAddress: "0",
+		},
+	})
+
+	Expect(err).ToNot(HaveOccurred())
+	err = (&LocalModelNodeReconciler{
+		Client:    k8sManager.GetClient(),
+		Clientset: clientset,
+		Scheme:    scheme.Scheme,
+		Log:       ctrl.Log.WithName("v1alpha1LocalModelAgent"),
+	}).SetupWithManager(k8sManager)
+	Expect(err).ToNot(HaveOccurred())
+	defer GinkgoRecover()
+	go func() {
+		err = k8sManager.Start(ctx)
+		Expect(err).ToNot(HaveOccurred())
+	}()
+})
+
+var _ = AfterSuite(func() {
+	cancel()
+	By("tearing down the test environment")
+	err := testEnv.Stop()
+	Expect(err).ToNot(HaveOccurred())
+})