From a3cb6c102b842322a3231101f1dc9c20ece63981 Mon Sep 17 00:00:00 2001 From: Chengjie Yao Date: Fri, 10 Jul 2026 14:43:58 -0700 Subject: [PATCH 1/3] [Core] Prioritize real downloads over validation replay --- pkg/modelagent/gopher.go | 107 +++++++++++++++++++++-- pkg/modelagent/gopher_task_queue.go | 35 +++++--- pkg/modelagent/gopher_task_queue_test.go | 49 +++++++++++ pkg/modelagent/gopher_test.go | 82 +++++++++++++++++ 4 files changed, 251 insertions(+), 22 deletions(-) diff --git a/pkg/modelagent/gopher.go b/pkg/modelagent/gopher.go index cde64a52e..a70948e9c 100644 --- a/pkg/modelagent/gopher.go +++ b/pkg/modelagent/gopher.go @@ -46,6 +46,7 @@ type GopherTask struct { TensorRTLLMShapeFilter *TensorRTLLMShapeFilter SamePathWaitStartedAt time.Time NormalPriorityOnly bool + NormalValidationOnly bool } type activeDownload struct { @@ -77,6 +78,8 @@ type Gopher struct { taskQueue *gopherTaskQueue samePathWaitDelay time.Duration samePathWaitTimeout time.Duration + + startupReadyModelKeys map[string]struct{} } const ( @@ -133,6 +136,8 @@ func NewGopher( } func (s *Gopher) Run(stopCh <-chan struct{}, numWorker int, numHighPriorityWorker int) { + s.captureStartupReadyModels(context.Background()) + // Start the ConfigMap reconciliation service s.configMapReconciler.StartReconciliation() s.logger.Info("Started ConfigMap reconciliation service") @@ -192,6 +197,8 @@ func (s *Gopher) enqueueTask(task *GopherTask) { } if task.TaskType == Delete { s.cancelActiveDownload(task) + } else { + s.markValidationOnlyIfStartupReady(task) } s.taskQueue.enqueue(task) } @@ -209,6 +216,9 @@ func (s *Gopher) runWorker() { if task.TaskType == Delete { s.cancelActiveDownload(task) } + if s.deferStartupReadyValidationIfLocalPathExists(task) { + continue + } err := s.processTask(task) if err != nil { s.logger.Errorf("Gopher task failed with error: %s", err.Error()) @@ -679,10 +689,82 @@ func (s *Gopher) demoteToNormalPriority(task *GopherTask) { return } task.NormalPriorityOnly = true + s.markValidationOnlyIfStartupReady(task) s.logger.Infof("Demoting %s to normal priority for fallback download/validation", getModelInfoForLogging(task)) s.enqueueTask(task) } +func (s *Gopher) deferStartupReadyValidationIfLocalPathExists(task *GopherTask) bool { + if s.markValidationOnlyIfStartupReady(task) { + s.logger.Infof("Deferring MD5 validation for %s behind normal download work", getModelInfoForLogging(task)) + s.enqueueTask(task) + return true + } + return false +} + +func (s *Gopher) markValidationOnlyIfStartupReady(task *GopherTask) bool { + if task == nil || task.NormalValidationOnly || task.TaskType != Download { + return false + } + if !s.wasReadyAtStartupWithLocalPath(task) { + return false + } + task.NormalPriorityOnly = true + task.NormalValidationOnly = true + return true +} + +func (s *Gopher) captureStartupReadyModels(ctx context.Context) { + if s.configMapReconciler == nil { + return + } + configMap, err := s.configMapReconciler.getConfigMap(ctx) + if err != nil { + s.logger.Warnf("Cannot capture startup Ready model snapshot: %v", err) + s.startupReadyModelKeys = map[string]struct{}{} + return + } + readyModelKeys := make(map[string]struct{}) + for key, data := range configMap.Data { + if hasModelEntryStatus(data, ModelStatusReady) { + readyModelKeys[key] = struct{}{} + } + } + s.startupReadyModelKeys = readyModelKeys + s.logger.Infof("Captured %d Ready models from startup ConfigMap snapshot", len(readyModelKeys)) +} + +func (s *Gopher) wasReadyAtStartupWithLocalPath(task *GopherTask) bool { + if len(s.startupReadyModelKeys) == 0 { + return false + } + modelKey := getModelID(task.BaseModel, task.ClusterBaseModel) + if _, wasReady := s.startupReadyModelKeys[modelKey]; !wasReady { + return false + } + + var baseModelSpec v1beta1.BaseModelSpec + if task.BaseModel != nil { + baseModelSpec = task.BaseModel.Spec + } else if task.ClusterBaseModel != nil { + baseModelSpec = task.ClusterBaseModel.Spec + } else { + return false + } + if baseModelSpec.Storage == nil || baseModelSpec.Storage.StorageUri == nil || baseModelSpec.Storage.Path == nil { + return false + } + storageType, err := storage.GetStorageType(*baseModelSpec.Storage.StorageUri) + if err != nil || storageType != storage.StorageTypeOCI { + return false + } + + destPath := getDestPath(&baseModelSpec, s.modelRootDir) + fileInfo, err := os.Stat(destPath) + return err == nil && fileInfo.IsDir() +} + func shouldUseSamePathObjectStorageReuse(task *GopherTask) bool { return task != nil && task.TaskType == Download } @@ -1101,6 +1183,21 @@ func sameModelStoragePath(currentStorage *v1beta1.StorageSpec, candidateStorage return getDestPath(&candidateSpec, modelRootDir) == destPath } +func filterObjectStorageObjectsForTask(objects []objectstorage.ObjectSummary, task *GopherTask) []objectstorage.ObjectSummary { + if task == nil || task.TensorRTLLMShapeFilter == nil || + !task.TensorRTLLMShapeFilter.IsTensorrtLLMModel || + task.TensorRTLLMShapeFilter.ModelType != string(constants.ServingBaseModel) { + return objects + } + shapeFilteredObjects := make([]objectstorage.ObjectSummary, 0) + for _, object := range objects { + if object.Name != nil && strings.Contains(*object.Name, fmt.Sprintf("/%s/", task.TensorRTLLMShapeFilter.ShapeAlias)) { + shapeFilteredObjects = append(shapeFilteredObjects, object) + } + } + return shapeFilteredObjects +} + func (s *Gopher) downloadModel(ctx context.Context, uri *ociobjectstore.ObjectURI, destPath string, task *GopherTask) error { startTime := time.Now() defer func() { @@ -1146,15 +1243,7 @@ func (s *Gopher) downloadModel(ctx context.Context, uri *ociobjectstore.ObjectUR // Shape filtering for TensorRTLLM if task.TensorRTLLMShapeFilter != nil && task.TensorRTLLMShapeFilter.IsTensorrtLLMModel && task.TensorRTLLMShapeFilter.ModelType == string(constants.ServingBaseModel) { s.logger.Infof("TensorRTLLM Serving model detected. Start filtering model files that doesn't belong to the node shape %s in model bucket folder", task.TensorRTLLMShapeFilter.ShapeAlias) - shapeFilteredObjects := make([]objectstorage.ObjectSummary, 0) - for _, object := range objects { - if object.Name != nil { - if strings.Contains(*object.Name, fmt.Sprintf("/%s/", task.TensorRTLLMShapeFilter.ShapeAlias)) { - shapeFilteredObjects = append(shapeFilteredObjects, object) - } - } - } - objects = shapeFilteredObjects + objects = filterObjectStorageObjectsForTask(objects, task) if len(objects) == 0 { return fmt.Errorf("no suitable objects found for shape %s", task.TensorRTLLMShapeFilter.ShapeAlias) diff --git a/pkg/modelagent/gopher_task_queue.go b/pkg/modelagent/gopher_task_queue.go index 288223b9f..d990073ec 100644 --- a/pkg/modelagent/gopher_task_queue.go +++ b/pkg/modelagent/gopher_task_queue.go @@ -8,11 +8,12 @@ import ( ) type gopherTaskQueue struct { - mutex sync.Mutex - cond *sync.Cond - high []*GopherTask - normal []*GopherTask - closed bool + mutex sync.Mutex + cond *sync.Cond + high []*GopherTask + normalDownload []*GopherTask + normalValidation []*GopherTask + closed bool } func newGopherTaskQueue() *gopherTaskQueue { @@ -34,12 +35,15 @@ func (q *gopherTaskQueue) enqueue(task *GopherTask) { // Delete preempts pending work for the same model and should run before // reuse-wait tasks, so it is the only non-FIFO insertion. q.high = removeSupersededTasks(q.high, task) - q.normal = removeSupersededTasks(q.normal, task) + q.normalDownload = removeSupersededTasks(q.normalDownload, task) + q.normalValidation = removeSupersededTasks(q.normalValidation, task) q.high = append([]*GopherTask{task}, q.high...) } else if shouldUseHighPriorityQueue(task) { q.high = append(q.high, task) + } else if task.NormalValidationOnly { + q.normalValidation = append(q.normalValidation, task) } else { - q.normal = append(q.normal, task) + q.normalDownload = append(q.normalDownload, task) } q.cond.Broadcast() } @@ -47,12 +51,17 @@ func (q *gopherTaskQueue) enqueue(task *GopherTask) { func (q *gopherTaskQueue) popNormal() (*GopherTask, bool) { q.mutex.Lock() defer q.mutex.Unlock() - for len(q.normal) == 0 && !q.closed { + for len(q.normalDownload) == 0 && len(q.normalValidation) == 0 && !q.closed { q.cond.Wait() } - if len(q.normal) > 0 { - task := q.normal[0] - q.normal = q.normal[1:] + if len(q.normalDownload) > 0 { + task := q.normalDownload[0] + q.normalDownload = q.normalDownload[1:] + return task, true + } + if len(q.normalValidation) > 0 { + task := q.normalValidation[0] + q.normalValidation = q.normalValidation[1:] return task, true } return nil, false @@ -82,7 +91,7 @@ func (q *gopherTaskQueue) close() { func (q *gopherTaskQueue) len() int { q.mutex.Lock() defer q.mutex.Unlock() - return len(q.high) + len(q.normal) + return len(q.high) + len(q.normalDownload) + len(q.normalValidation) } func shouldUseHighPriorityQueue(task *GopherTask) bool { @@ -90,7 +99,7 @@ func shouldUseHighPriorityQueue(task *GopherTask) bool { } func isObjectStorageDownloadTask(task *GopherTask) bool { - if task == nil || task.TaskType != Download || task.NormalPriorityOnly { + if task == nil || task.TaskType != Download || task.NormalPriorityOnly || task.NormalValidationOnly { return false } var storageSpec *v1beta1.StorageSpec diff --git a/pkg/modelagent/gopher_task_queue_test.go b/pkg/modelagent/gopher_task_queue_test.go index c582ed834..389a85839 100644 --- a/pkg/modelagent/gopher_task_queue_test.go +++ b/pkg/modelagent/gopher_task_queue_test.go @@ -103,6 +103,35 @@ func TestGopherTaskQueueKeepsDownloadOverrideNormal(t *testing.T) { assert.Equal(t, "oci-model", queued.BaseModel.Name) } +func TestGopherTaskQueuePrioritizesNormalDownloadBeforeValidation(t *testing.T) { + queue := newGopherTaskQueue() + validation := &GopherTask{ + TaskType: Download, + NormalPriorityOnly: true, + NormalValidationOnly: true, + BaseModel: &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{Name: "validation", Namespace: "service-ns", UID: "validation-uid"}, + }, + } + download := &GopherTask{ + TaskType: Download, + NormalPriorityOnly: true, + BaseModel: &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{Name: "download", Namespace: "service-ns", UID: "download-uid"}, + }, + } + + queue.enqueue(validation) + queue.enqueue(download) + + task, ok := queue.popNormal() + require.True(t, ok) + assert.Equal(t, "download", task.BaseModel.Name) + task, ok = queue.popNormal() + require.True(t, ok) + assert.Equal(t, "validation", task.BaseModel.Name) +} + func TestGopherTaskQueueDeleteSupersedesPendingDownloadsForSameModel(t *testing.T) { queue := newGopherTaskQueue() model := &v1beta1.BaseModel{ @@ -203,6 +232,26 @@ func TestGopherTaskQueueDemotedSamePathWaitUsesNormalQueue(t *testing.T) { assert.Equal(t, "demoted", task.BaseModel.Name) } +func TestGopherTaskQueueDeleteSupersedesPendingValidationForSameModel(t *testing.T) { + queue := newGopherTaskQueue() + model := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{Name: "model", Namespace: "service-ns", UID: "model-uid"}, + } + + queue.enqueue(&GopherTask{ + TaskType: Download, + BaseModel: model, + NormalPriorityOnly: true, + NormalValidationOnly: true, + }) + queue.enqueue(&GopherTask{TaskType: Delete, BaseModel: model}) + + task, ok := queue.popHighPriority() + require.True(t, ok) + assert.Equal(t, Delete, task.TaskType) + assert.Equal(t, 0, queue.len()) +} + func TestGopherTaskQueueEnqueueWakesMatchingBlockedWorker(t *testing.T) { queue := newGopherTaskQueue() normalDone := make(chan struct{}) diff --git a/pkg/modelagent/gopher_test.go b/pkg/modelagent/gopher_test.go index a77b6003b..9106ca9d7 100644 --- a/pkg/modelagent/gopher_test.go +++ b/pkg/modelagent/gopher_test.go @@ -1060,6 +1060,88 @@ func TestProcessTaskWithOptions_HighPriorityDemotesFallbackDownload(t *testing.T assert.Equal(t, 0, g.taskQueue.len()) } +func TestDemoteToNormalPriorityClassifiesValidationBeforeEnqueue(t *testing.T) { + storageURI := "oci://n/object-ns/b/model-bucket/o/models/already-downloaded" + modelPath := filepath.Join(t.TempDir(), "already-downloaded") + require.NoError(t, os.MkdirAll(modelPath, 0755)) + alreadyDownloaded := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{Name: "already-downloaded", Namespace: "service-ns", UID: "already-downloaded-uid"}, + Spec: v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{StorageUri: &storageURI, Path: &modelPath}, + }, + } + missingArtifact := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{Name: "missing-artifact", Namespace: "service-ns", UID: "missing-artifact-uid"}, + Spec: v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{StorageUri: &storageURI, Path: &modelPath}, + }, + } + g := &Gopher{ + taskQueue: newGopherTaskQueue(), + logger: zap.NewNop().Sugar(), + startupReadyModelKeys: map[string]struct{}{getModelID(alreadyDownloaded, nil): {}}, + modelRootDir: t.TempDir(), + } + + g.demoteToNormalPriority(&GopherTask{TaskType: Download, BaseModel: alreadyDownloaded}) + g.demoteToNormalPriority(&GopherTask{TaskType: Download, BaseModel: missingArtifact}) + + task, ok := g.taskQueue.popNormal() + require.True(t, ok) + assert.Equal(t, missingArtifact.Name, task.BaseModel.Name) + assert.True(t, task.NormalPriorityOnly) + assert.False(t, task.NormalValidationOnly) + + task, ok = g.taskQueue.popNormal() + require.True(t, ok) + assert.Equal(t, alreadyDownloaded.Name, task.BaseModel.Name) + assert.True(t, task.NormalPriorityOnly) + assert.True(t, task.NormalValidationOnly) +} + +func TestEnqueueTaskClassifiesStartupReadyLocalPathAsValidation(t *testing.T) { + storageURI := "oci://n/object-ns/b/model-bucket/o/models/ready-model" + modelPath := filepath.Join(t.TempDir(), "ready-model") + require.NoError(t, os.MkdirAll(modelPath, 0755)) + model := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{Name: "ready-model", Namespace: "service-ns", UID: "ready-model-uid"}, + Spec: v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{StorageUri: &storageURI, Path: &modelPath}, + }, + } + g := &Gopher{ + taskQueue: newGopherTaskQueue(), + logger: zap.NewNop().Sugar(), + startupReadyModelKeys: map[string]struct{}{getModelID(model, nil): {}}, + } + + g.enqueueTask(&GopherTask{TaskType: Download, BaseModel: model}) + + task, ok := g.taskQueue.popNormal() + require.True(t, ok) + assert.Equal(t, model.Name, task.BaseModel.Name) + assert.True(t, task.NormalPriorityOnly) + assert.True(t, task.NormalValidationOnly) + assert.Equal(t, 0, g.taskQueue.len()) +} + +func TestCaptureStartupReadyModelsCapturesOnlyReadyEntries(t *testing.T) { + readyKey := constants.GetModelConfigMapKey("service-ns", "ready-model", false) + updatingKey := constants.GetModelConfigMapKey("service-ns", "updating-model", false) + cm := makeConfigMap("node-1", map[string]string{ + readyKey: modelEntryJSON(ModelStatusReady), + updatingKey: modelEntryJSON(ModelStatusUpdating), + "invalid": "not-json", + }) + g := newGopherForProcessTask(cm) + + g.captureStartupReadyModels(context.Background()) + + assert.Contains(t, g.startupReadyModelKeys, readyKey) + assert.NotContains(t, g.startupReadyModelKeys, updatingKey) + assert.NotContains(t, g.startupReadyModelKeys, "invalid") +} + func TestGopherEnqueueDeleteCancelsActiveDownload(t *testing.T) { model := &v1beta1.BaseModel{ ObjectMeta: metav1.ObjectMeta{Name: "model", Namespace: "service-ns", UID: "model-uid"}, From 6d934319cf40e19cf38d94c34c46615adb8356b8 Mon Sep 17 00:00:00 2001 From: Chengjie Yao Date: Tue, 14 Jul 2026 11:01:24 -0700 Subject: [PATCH 2/3] Add tests to address comments --- pkg/modelagent/gopher_test.go | 38 ++++++++++++++++++++++++++++++++++- 1 file changed, 37 insertions(+), 1 deletion(-) diff --git a/pkg/modelagent/gopher_test.go b/pkg/modelagent/gopher_test.go index 9106ca9d7..5c33eda91 100644 --- a/pkg/modelagent/gopher_test.go +++ b/pkg/modelagent/gopher_test.go @@ -1064,6 +1064,8 @@ func TestDemoteToNormalPriorityClassifiesValidationBeforeEnqueue(t *testing.T) { storageURI := "oci://n/object-ns/b/model-bucket/o/models/already-downloaded" modelPath := filepath.Join(t.TempDir(), "already-downloaded") require.NoError(t, os.MkdirAll(modelPath, 0755)) + missingStorageURI := "oci://n/object-ns/b/model-bucket/o/models/missing-artifact" + missingModelPath := filepath.Join(t.TempDir(), "missing-artifact") alreadyDownloaded := &v1beta1.BaseModel{ ObjectMeta: metav1.ObjectMeta{Name: "already-downloaded", Namespace: "service-ns", UID: "already-downloaded-uid"}, Spec: v1beta1.BaseModelSpec{ @@ -1073,7 +1075,7 @@ func TestDemoteToNormalPriorityClassifiesValidationBeforeEnqueue(t *testing.T) { missingArtifact := &v1beta1.BaseModel{ ObjectMeta: metav1.ObjectMeta{Name: "missing-artifact", Namespace: "service-ns", UID: "missing-artifact-uid"}, Spec: v1beta1.BaseModelSpec{ - Storage: &v1beta1.StorageSpec{StorageUri: &storageURI, Path: &modelPath}, + Storage: &v1beta1.StorageSpec{StorageUri: &missingStorageURI, Path: &missingModelPath}, }, } g := &Gopher{ @@ -1082,6 +1084,8 @@ func TestDemoteToNormalPriorityClassifiesValidationBeforeEnqueue(t *testing.T) { startupReadyModelKeys: map[string]struct{}{getModelID(alreadyDownloaded, nil): {}}, modelRootDir: t.TempDir(), } + _, err := os.Stat(missingModelPath) + require.True(t, os.IsNotExist(err)) g.demoteToNormalPriority(&GopherTask{TaskType: Download, BaseModel: alreadyDownloaded}) g.demoteToNormalPriority(&GopherTask{TaskType: Download, BaseModel: missingArtifact}) @@ -1142,6 +1146,38 @@ func TestCaptureStartupReadyModelsCapturesOnlyReadyEntries(t *testing.T) { assert.NotContains(t, g.startupReadyModelKeys, "invalid") } +func TestCaptureStartupReadyModelsFeedsValidationClassification(t *testing.T) { + storageURI := "oci://n/object-ns/b/model-bucket/o/models/ready-model" + modelPath := filepath.Join(t.TempDir(), "ready-model") + require.NoError(t, os.MkdirAll(modelPath, 0755)) + readyModel := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{Name: "ready-model", Namespace: "service-ns", UID: "ready-model-uid"}, + Spec: v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{StorageUri: &storageURI, Path: &modelPath}, + }, + } + updatingModel := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{Name: "updating-model", Namespace: "service-ns", UID: "updating-model-uid"}, + } + g := newGopherForProcessTask(makeConfigMap("node-1", map[string]string{ + constants.GetModelConfigMapKey(readyModel.Namespace, readyModel.Name, false): modelEntryJSON(ModelStatusReady), + constants.GetModelConfigMapKey(updatingModel.Namespace, updatingModel.Name, false): modelEntryJSON(ModelStatusUpdating), + })) + g.taskQueue = newGopherTaskQueue() + + g.captureStartupReadyModels(context.Background()) + g.enqueueTask(&GopherTask{TaskType: Download, BaseModel: readyModel}) + + task, ok := g.taskQueue.popNormal() + require.True(t, ok) + assert.Equal(t, readyModel.Name, task.BaseModel.Name) + assert.True(t, task.NormalPriorityOnly) + assert.True(t, task.NormalValidationOnly) + assert.Contains(t, g.startupReadyModelKeys, getModelID(readyModel, nil)) + assert.NotContains(t, g.startupReadyModelKeys, getModelID(updatingModel, nil)) + assert.Equal(t, 0, g.taskQueue.len()) +} + func TestGopherEnqueueDeleteCancelsActiveDownload(t *testing.T) { model := &v1beta1.BaseModel{ ObjectMeta: metav1.ObjectMeta{Name: "model", Namespace: "service-ns", UID: "model-uid"}, From 8838eb666653b8fd7cc4ec4827fc550e3d89e1a9 Mon Sep 17 00:00:00 2001 From: Chengjie Yao Date: Tue, 14 Jul 2026 16:47:36 -0700 Subject: [PATCH 3/3] Address comments --- pkg/modelagent/configmap_reconciler.go | 4 ++ pkg/modelagent/gopher.go | 44 ++++++++--------- pkg/modelagent/gopher_task_queue.go | 30 ++++++------ pkg/modelagent/gopher_task_queue_test.go | 18 +++---- pkg/modelagent/gopher_test.go | 60 +++++++++++++++++++++--- 5 files changed, 101 insertions(+), 55 deletions(-) diff --git a/pkg/modelagent/configmap_reconciler.go b/pkg/modelagent/configmap_reconciler.go index 5697fd007..8d0d6e083 100644 --- a/pkg/modelagent/configmap_reconciler.go +++ b/pkg/modelagent/configmap_reconciler.go @@ -712,6 +712,10 @@ Returns: func (c *ConfigMapReconciler) getConfigMap(ctx context.Context) (*corev1.ConfigMap, error) { existingConfigMap, err := c.kubeClient.CoreV1().ConfigMaps(c.namespace).Get(ctx, c.nodeName, metav1.GetOptions{}) if err != nil { + if errors.IsNotFound(err) { + c.logger.Infof("Node %s configmap does not exist yet", c.nodeName) + return nil, err + } c.logger.Errorf("Failed retrieve node %s configmap: %v", c.nodeName, err) return nil, err } diff --git a/pkg/modelagent/gopher.go b/pkg/modelagent/gopher.go index a70948e9c..f66f95a05 100644 --- a/pkg/modelagent/gopher.go +++ b/pkg/modelagent/gopher.go @@ -46,7 +46,7 @@ type GopherTask struct { TensorRTLLMShapeFilter *TensorRTLLMShapeFilter SamePathWaitStartedAt time.Time NormalPriorityOnly bool - NormalValidationOnly bool + RevalidationReplay bool } type activeDownload struct { @@ -85,8 +85,9 @@ type Gopher struct { const ( BigFileSizeInMB = 200 - defaultSamePathWaitDelay = 30 * time.Second - defaultSamePathWaitTimeout = 30 * time.Minute + defaultSamePathWaitDelay = 30 * time.Second + defaultSamePathWaitTimeout = 30 * time.Minute + defaultStartupReadySnapshotTimeout = 5 * time.Second ) func NewGopher( @@ -136,7 +137,9 @@ func NewGopher( } func (s *Gopher) Run(stopCh <-chan struct{}, numWorker int, numHighPriorityWorker int) { - s.captureStartupReadyModels(context.Background()) + startupSnapshotCtx, cancelStartupSnapshot := context.WithTimeout(context.Background(), defaultStartupReadySnapshotTimeout) + defer cancelStartupSnapshot() + s.captureStartupReadyModels(startupSnapshotCtx) // Start the ConfigMap reconciliation service s.configMapReconciler.StartReconciliation() @@ -198,7 +201,7 @@ func (s *Gopher) enqueueTask(task *GopherTask) { if task.TaskType == Delete { s.cancelActiveDownload(task) } else { - s.markValidationOnlyIfStartupReady(task) + s.classifyStartupRevalidation(task) } s.taskQueue.enqueue(task) } @@ -216,9 +219,6 @@ func (s *Gopher) runWorker() { if task.TaskType == Delete { s.cancelActiveDownload(task) } - if s.deferStartupReadyValidationIfLocalPathExists(task) { - continue - } err := s.processTask(task) if err != nil { s.logger.Errorf("Gopher task failed with error: %s", err.Error()) @@ -689,29 +689,20 @@ func (s *Gopher) demoteToNormalPriority(task *GopherTask) { return } task.NormalPriorityOnly = true - s.markValidationOnlyIfStartupReady(task) + s.classifyStartupRevalidation(task) s.logger.Infof("Demoting %s to normal priority for fallback download/validation", getModelInfoForLogging(task)) s.enqueueTask(task) } -func (s *Gopher) deferStartupReadyValidationIfLocalPathExists(task *GopherTask) bool { - if s.markValidationOnlyIfStartupReady(task) { - s.logger.Infof("Deferring MD5 validation for %s behind normal download work", getModelInfoForLogging(task)) - s.enqueueTask(task) - return true - } - return false -} - -func (s *Gopher) markValidationOnlyIfStartupReady(task *GopherTask) bool { - if task == nil || task.NormalValidationOnly || task.TaskType != Download { +func (s *Gopher) classifyStartupRevalidation(task *GopherTask) bool { + if task == nil || task.RevalidationReplay || task.TaskType != Download { return false } - if !s.wasReadyAtStartupWithLocalPath(task) { + if !s.isStartupRevalidation(task) { return false } task.NormalPriorityOnly = true - task.NormalValidationOnly = true + task.RevalidationReplay = true return true } @@ -721,6 +712,11 @@ func (s *Gopher) captureStartupReadyModels(ctx context.Context) { } configMap, err := s.configMapReconciler.getConfigMap(ctx) if err != nil { + if apierrors.IsNotFound(err) { + s.logger.Infof("No startup Ready model snapshot because node ConfigMap does not exist yet") + s.startupReadyModelKeys = map[string]struct{}{} + return + } s.logger.Warnf("Cannot capture startup Ready model snapshot: %v", err) s.startupReadyModelKeys = map[string]struct{}{} return @@ -735,7 +731,7 @@ func (s *Gopher) captureStartupReadyModels(ctx context.Context) { s.logger.Infof("Captured %d Ready models from startup ConfigMap snapshot", len(readyModelKeys)) } -func (s *Gopher) wasReadyAtStartupWithLocalPath(task *GopherTask) bool { +func (s *Gopher) isStartupRevalidation(task *GopherTask) bool { if len(s.startupReadyModelKeys) == 0 { return false } @@ -752,7 +748,7 @@ func (s *Gopher) wasReadyAtStartupWithLocalPath(task *GopherTask) bool { } else { return false } - if baseModelSpec.Storage == nil || baseModelSpec.Storage.StorageUri == nil || baseModelSpec.Storage.Path == nil { + if baseModelSpec.Storage == nil || baseModelSpec.Storage.StorageUri == nil || baseModelSpec.Storage.Path == nil || *baseModelSpec.Storage.Path == "" { return false } storageType, err := storage.GetStorageType(*baseModelSpec.Storage.StorageUri) diff --git a/pkg/modelagent/gopher_task_queue.go b/pkg/modelagent/gopher_task_queue.go index d990073ec..1cd55b30c 100644 --- a/pkg/modelagent/gopher_task_queue.go +++ b/pkg/modelagent/gopher_task_queue.go @@ -8,12 +8,12 @@ import ( ) type gopherTaskQueue struct { - mutex sync.Mutex - cond *sync.Cond - high []*GopherTask - normalDownload []*GopherTask - normalValidation []*GopherTask - closed bool + mutex sync.Mutex + cond *sync.Cond + high []*GopherTask + normalDownload []*GopherTask + normalRevalidation []*GopherTask + closed bool } func newGopherTaskQueue() *gopherTaskQueue { @@ -36,12 +36,12 @@ func (q *gopherTaskQueue) enqueue(task *GopherTask) { // reuse-wait tasks, so it is the only non-FIFO insertion. q.high = removeSupersededTasks(q.high, task) q.normalDownload = removeSupersededTasks(q.normalDownload, task) - q.normalValidation = removeSupersededTasks(q.normalValidation, task) + q.normalRevalidation = removeSupersededTasks(q.normalRevalidation, task) q.high = append([]*GopherTask{task}, q.high...) } else if shouldUseHighPriorityQueue(task) { q.high = append(q.high, task) - } else if task.NormalValidationOnly { - q.normalValidation = append(q.normalValidation, task) + } else if task.RevalidationReplay { + q.normalRevalidation = append(q.normalRevalidation, task) } else { q.normalDownload = append(q.normalDownload, task) } @@ -51,7 +51,7 @@ func (q *gopherTaskQueue) enqueue(task *GopherTask) { func (q *gopherTaskQueue) popNormal() (*GopherTask, bool) { q.mutex.Lock() defer q.mutex.Unlock() - for len(q.normalDownload) == 0 && len(q.normalValidation) == 0 && !q.closed { + for len(q.normalDownload) == 0 && len(q.normalRevalidation) == 0 && !q.closed { q.cond.Wait() } if len(q.normalDownload) > 0 { @@ -59,9 +59,9 @@ func (q *gopherTaskQueue) popNormal() (*GopherTask, bool) { q.normalDownload = q.normalDownload[1:] return task, true } - if len(q.normalValidation) > 0 { - task := q.normalValidation[0] - q.normalValidation = q.normalValidation[1:] + if len(q.normalRevalidation) > 0 { + task := q.normalRevalidation[0] + q.normalRevalidation = q.normalRevalidation[1:] return task, true } return nil, false @@ -91,7 +91,7 @@ func (q *gopherTaskQueue) close() { func (q *gopherTaskQueue) len() int { q.mutex.Lock() defer q.mutex.Unlock() - return len(q.high) + len(q.normalDownload) + len(q.normalValidation) + return len(q.high) + len(q.normalDownload) + len(q.normalRevalidation) } func shouldUseHighPriorityQueue(task *GopherTask) bool { @@ -99,7 +99,7 @@ func shouldUseHighPriorityQueue(task *GopherTask) bool { } func isObjectStorageDownloadTask(task *GopherTask) bool { - if task == nil || task.TaskType != Download || task.NormalPriorityOnly || task.NormalValidationOnly { + if task == nil || task.TaskType != Download || task.NormalPriorityOnly || task.RevalidationReplay { return false } var storageSpec *v1beta1.StorageSpec diff --git a/pkg/modelagent/gopher_task_queue_test.go b/pkg/modelagent/gopher_task_queue_test.go index 389a85839..8e0c0036e 100644 --- a/pkg/modelagent/gopher_task_queue_test.go +++ b/pkg/modelagent/gopher_task_queue_test.go @@ -103,12 +103,12 @@ func TestGopherTaskQueueKeepsDownloadOverrideNormal(t *testing.T) { assert.Equal(t, "oci-model", queued.BaseModel.Name) } -func TestGopherTaskQueuePrioritizesNormalDownloadBeforeValidation(t *testing.T) { +func TestGopherTaskQueuePrioritizesNormalDownloadBeforeRevalidationReplay(t *testing.T) { queue := newGopherTaskQueue() validation := &GopherTask{ - TaskType: Download, - NormalPriorityOnly: true, - NormalValidationOnly: true, + TaskType: Download, + NormalPriorityOnly: true, + RevalidationReplay: true, BaseModel: &v1beta1.BaseModel{ ObjectMeta: metav1.ObjectMeta{Name: "validation", Namespace: "service-ns", UID: "validation-uid"}, }, @@ -232,17 +232,17 @@ func TestGopherTaskQueueDemotedSamePathWaitUsesNormalQueue(t *testing.T) { assert.Equal(t, "demoted", task.BaseModel.Name) } -func TestGopherTaskQueueDeleteSupersedesPendingValidationForSameModel(t *testing.T) { +func TestGopherTaskQueueDeleteSupersedesPendingRevalidationReplayForSameModel(t *testing.T) { queue := newGopherTaskQueue() model := &v1beta1.BaseModel{ ObjectMeta: metav1.ObjectMeta{Name: "model", Namespace: "service-ns", UID: "model-uid"}, } queue.enqueue(&GopherTask{ - TaskType: Download, - BaseModel: model, - NormalPriorityOnly: true, - NormalValidationOnly: true, + TaskType: Download, + BaseModel: model, + NormalPriorityOnly: true, + RevalidationReplay: true, }) queue.enqueue(&GopherTask{TaskType: Delete, BaseModel: model}) diff --git a/pkg/modelagent/gopher_test.go b/pkg/modelagent/gopher_test.go index 5c33eda91..4fca0cc4e 100644 --- a/pkg/modelagent/gopher_test.go +++ b/pkg/modelagent/gopher_test.go @@ -10,7 +10,9 @@ import ( "testing" "time" + "go.uber.org/zap/zapcore" "go.uber.org/zap/zaptest" + "go.uber.org/zap/zaptest/observer" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -1060,7 +1062,7 @@ func TestProcessTaskWithOptions_HighPriorityDemotesFallbackDownload(t *testing.T assert.Equal(t, 0, g.taskQueue.len()) } -func TestDemoteToNormalPriorityClassifiesValidationBeforeEnqueue(t *testing.T) { +func TestDemoteToNormalPriorityClassifiesRevalidationBeforeEnqueue(t *testing.T) { storageURI := "oci://n/object-ns/b/model-bucket/o/models/already-downloaded" modelPath := filepath.Join(t.TempDir(), "already-downloaded") require.NoError(t, os.MkdirAll(modelPath, 0755)) @@ -1094,16 +1096,16 @@ func TestDemoteToNormalPriorityClassifiesValidationBeforeEnqueue(t *testing.T) { require.True(t, ok) assert.Equal(t, missingArtifact.Name, task.BaseModel.Name) assert.True(t, task.NormalPriorityOnly) - assert.False(t, task.NormalValidationOnly) + assert.False(t, task.RevalidationReplay) task, ok = g.taskQueue.popNormal() require.True(t, ok) assert.Equal(t, alreadyDownloaded.Name, task.BaseModel.Name) assert.True(t, task.NormalPriorityOnly) - assert.True(t, task.NormalValidationOnly) + assert.True(t, task.RevalidationReplay) } -func TestEnqueueTaskClassifiesStartupReadyLocalPathAsValidation(t *testing.T) { +func TestEnqueueTaskClassifiesStartupReadyLocalPathAsRevalidation(t *testing.T) { storageURI := "oci://n/object-ns/b/model-bucket/o/models/ready-model" modelPath := filepath.Join(t.TempDir(), "ready-model") require.NoError(t, os.MkdirAll(modelPath, 0755)) @@ -1125,7 +1127,7 @@ func TestEnqueueTaskClassifiesStartupReadyLocalPathAsValidation(t *testing.T) { require.True(t, ok) assert.Equal(t, model.Name, task.BaseModel.Name) assert.True(t, task.NormalPriorityOnly) - assert.True(t, task.NormalValidationOnly) + assert.True(t, task.RevalidationReplay) assert.Equal(t, 0, g.taskQueue.len()) } @@ -1146,7 +1148,7 @@ func TestCaptureStartupReadyModelsCapturesOnlyReadyEntries(t *testing.T) { assert.NotContains(t, g.startupReadyModelKeys, "invalid") } -func TestCaptureStartupReadyModelsFeedsValidationClassification(t *testing.T) { +func TestCaptureStartupReadyModelsFeedsRevalidationClassification(t *testing.T) { storageURI := "oci://n/object-ns/b/model-bucket/o/models/ready-model" modelPath := filepath.Join(t.TempDir(), "ready-model") require.NoError(t, os.MkdirAll(modelPath, 0755)) @@ -1172,12 +1174,56 @@ func TestCaptureStartupReadyModelsFeedsValidationClassification(t *testing.T) { require.True(t, ok) assert.Equal(t, readyModel.Name, task.BaseModel.Name) assert.True(t, task.NormalPriorityOnly) - assert.True(t, task.NormalValidationOnly) + assert.True(t, task.RevalidationReplay) assert.Contains(t, g.startupReadyModelKeys, getModelID(readyModel, nil)) assert.NotContains(t, g.startupReadyModelKeys, getModelID(updatingModel, nil)) assert.Equal(t, 0, g.taskQueue.len()) } +func TestCaptureStartupReadyModelsTreatsMissingConfigMapAsColdStart(t *testing.T) { + core, recorded := observer.New(zap.DebugLevel) + logger := zap.New(core).Sugar() + cm := makeConfigMap("node-1", map[string]string{}) + reconciler := NewConfigMapReconciler(cm.Name, cm.Namespace, k8sfake.NewSimpleClientset(), logger) + g := &Gopher{ + configMapReconciler: reconciler, + logger: logger, + startupReadyModelKeys: map[string]struct{}{"stale-model": {}}, + } + + g.captureStartupReadyModels(context.Background()) + + assert.Empty(t, g.startupReadyModelKeys) + assert.Empty(t, recorded.FilterLevelExact(zapcore.ErrorLevel).All()) + assert.Empty(t, recorded.FilterLevelExact(zapcore.WarnLevel).All()) +} + +func TestClassifyStartupRevalidationIgnoresEmptyStoragePath(t *testing.T) { + storageURI := "oci://n/object-ns/b/model-bucket/o/models/ready-model" + emptyPath := "" + modelRootDir := t.TempDir() + require.NoError(t, os.MkdirAll(modelRootDir+"/"+storageURI, 0755)) + model := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{Name: "ready-model", Namespace: "service-ns", UID: "ready-model-uid"}, + Spec: v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{ + StorageUri: &storageURI, + Path: &emptyPath, + }, + }, + } + g := &Gopher{ + logger: zap.NewNop().Sugar(), + startupReadyModelKeys: map[string]struct{}{getModelID(model, nil): {}}, + modelRootDir: modelRootDir, + } + task := &GopherTask{TaskType: Download, BaseModel: model} + + assert.False(t, g.classifyStartupRevalidation(task)) + assert.False(t, task.NormalPriorityOnly) + assert.False(t, task.RevalidationReplay) +} + func TestGopherEnqueueDeleteCancelsActiveDownload(t *testing.T) { model := &v1beta1.BaseModel{ ObjectMeta: metav1.ObjectMeta{Name: "model", Namespace: "service-ns", UID: "model-uid"},