diff --git a/pkg/constants/constants.go b/pkg/constants/constants.go index 24f6a06dc..5221e5a35 100644 --- a/pkg/constants/constants.go +++ b/pkg/constants/constants.go @@ -424,8 +424,12 @@ const ( LLamaVllmFTServingServedModelNamePrefix = "/data" ) -// DefaultModelLocalMountPath is where models will be mounted by the storage-initializer -const DefaultModelLocalMountPath = "/mnt/models" +const ( + // DefaultModelLocalMountPath is where models will be mounted by the storage-initializer. + DefaultModelLocalMountPath = "/mnt/models" + // ModelArtifactsDirectory is the node-local shared artifact directory under a model store. + ModelArtifactsDirectory = "_artifacts" +) var ( ServiceAnnotationDisallowedList = []string{ diff --git a/pkg/modelagent/artifact_object_filter.go b/pkg/modelagent/artifact_object_filter.go new file mode 100644 index 000000000..81c6057b6 --- /dev/null +++ b/pkg/modelagent/artifact_object_filter.go @@ -0,0 +1,30 @@ +package modelagent + +import ( + "strings" + + "github.com/oracle/oci-go-sdk/v65/objectstorage" +) + +func filterInternalArtifactObjectSummaries(objects []objectstorage.ObjectSummary) []objectstorage.ObjectSummary { + filtered := make([]objectstorage.ObjectSummary, 0, len(objects)) + for _, object := range objects { + if object.Name != nil && isInternalArtifactObjectName(*object.Name) { + continue + } + filtered = append(filtered, object) + } + return filtered +} + +func isInternalArtifactObjectName(objectName string) bool { + return isArtifactCompleteMarkerObjectName(objectName) || isArtifactUploadLockObjectName(objectName) +} + +func isArtifactCompleteMarkerObjectName(objectName string) bool { + return objectName == artifactCompleteMarkerFileName || strings.HasSuffix(objectName, "/"+artifactCompleteMarkerFileName) +} + +func isArtifactUploadLockObjectName(objectName string) bool { + return objectName == artifactUploadLockFileName || strings.HasSuffix(objectName, "/"+artifactUploadLockFileName) +} diff --git a/pkg/modelagent/configmap_reconciler.go b/pkg/modelagent/configmap_reconciler.go index 8d0d6e083..809a71fdf 100644 --- a/pkg/modelagent/configmap_reconciler.go +++ b/pkg/modelagent/configmap_reconciler.go @@ -26,6 +26,10 @@ const ( ConfigAttr = "config" ArtifactAttr = "artifact" ShaAttr = "sha" + OriginAttr = "origin" + OriginTypeAttr = "type" + HFModelIDAttr = "hfModelId" + HFCommitSHAAttr = "hfCommitSha" ParentPathAttr = "parentPath" ChildrenPathsAttr = "childrenPaths" ) @@ -1043,6 +1047,100 @@ func (c *ConfigMapReconciler) FindMatchedModelFromConfigMap(configMap *corev1.Co return matchedParentName, matchedParentPath, searchingError } +/* +FindMatchedModelFromConfigMapByArtifactIdentity scans Ready entries in the node +ConfigMap for an artifact that has the same provenance identity. This is used +for OCI artifacts that were originally imported from Hugging Face, where +config.artifact.origin carries the HF model ID and resolved commit SHA. + +Returns the parent model key and parent artifact path for the best match, +preferring the parent that already has the most children so new children keep +pointing at the established parent artifact. +*/ +func (c *ConfigMapReconciler) FindMatchedModelFromConfigMapByArtifactIdentity(configMap *corev1.ConfigMap, identity ArtifactIdentity, modelType string, currentModelTypeAndNodeName string) (string, string, error) { + var searchingError error + + var matchedParentName, matchedParentPath string + var matchedParentChildrenNum = -1 + for modelKey, jsonStr := range configMap.Data { + if !strings.HasPrefix(strings.ToLower(modelKey), strings.ToLower(modelType)) { + continue + } + if strings.EqualFold(modelKey, currentModelTypeAndNodeName) { + continue + } + + var obj map[string]interface{} + if err := json.Unmarshal([]byte(jsonStr), &obj); err != nil { + searchingError = fmt.Errorf("fail to Unmarshal %s during FindMatchedModelFromConfigMapByArtifactIdentity: %s", jsonStr, err) + c.logger.Errorf(searchingError.Error()) + continue + } + status, ok := obj["status"].(string) + if !ok || !strings.EqualFold(status, string(ModelStatusReady)) { + continue + } + config, ok := obj[ConfigAttr].(map[string]interface{}) + if !ok { + continue + } + artifact, ok := config[ArtifactAttr].(map[string]interface{}) + if !ok || !artifactMatchesIdentity(artifact, identity) { + continue + } + + rawParent, ok := artifact[ParentPathAttr].(map[string]interface{}) + if !ok { + searchingError = fmt.Errorf("parentPath is not an object") + continue + } + if len(rawParent) != 1 { + searchingError = fmt.Errorf("expected exactly one parentPath entry, got %d", len(rawParent)) + continue + } + + for parentName, value := range rawParent { + parentPath, ok := value.(string) + if !ok { + searchingError = fmt.Errorf("parentPath value for %q is not a string", parentName) + continue + } + if strings.EqualFold(parentName, currentModelTypeAndNodeName) { + continue + } + + childrenNum := len(extractChildrenPaths(artifact)) + if childrenNum > matchedParentChildrenNum { + matchedParentChildrenNum = childrenNum + matchedParentName = parentName + matchedParentPath = parentPath + } + } + } + c.logger.Infof("matched artifact identity parentName: %s, matchedParentPath: %s", matchedParentName, matchedParentPath) + return matchedParentName, matchedParentPath, searchingError +} + +func artifactMatchesIdentity(artifact map[string]interface{}, identity ArtifactIdentity) bool { + origin, ok := artifact[OriginAttr].(map[string]interface{}) + if !ok { + return false + } + originType, ok := origin[OriginTypeAttr].(string) + if !ok || !strings.EqualFold(originType, identity.OriginType) { + return false + } + hfModelID, ok := origin[HFModelIDAttr].(string) + if !ok || hfModelID != identity.HFModelID { + return false + } + hfCommitSHA, ok := origin[HFCommitSHAAttr].(string) + if !ok { + return false + } + return strings.EqualFold(hfCommitSHA, identity.HFCommitSHA) +} + func extractChildrenPaths(artifact map[string]interface{}) []string { // Read childrenPaths leniently and convert to []string children := make([]string, 0) @@ -1081,6 +1179,21 @@ func (c *ConfigMapReconciler) getModelDataByArtifactSha(ctx context.Context, tar return c.FindMatchedModelFromConfigMap(cm, targetSha, modelType, currentModelTypeAndNodeName) } +// getModelDataByArtifactIdentity fetches the node ConfigMap and searches for a +// Ready entry with the same normalized artifact identity. +func (c *ConfigMapReconciler) getModelDataByArtifactIdentity(ctx context.Context, identity ArtifactIdentity, modelType string, currentModelTypeAndNodeName string) (string, string, error) { + cm, err := c.kubeClient.CoreV1().ConfigMaps("ome").Get(ctx, c.nodeName, metav1.GetOptions{}) + if err != nil { + if errors.IsNotFound(err) { + c.logger.Warn("cannot find configmap %s", c.nodeName) + return "", "", fmt.Errorf("cannot find configmap %s", c.nodeName) + } + c.logger.Errorf("Failed to get ConfigMap %s: %v", c.nodeName, err) + return "", "", fmt.Errorf("failed to get ConfigMap %s: %v", c.nodeName, err) + } + return c.FindMatchedModelFromConfigMapByArtifactIdentity(cm, identity, modelType, currentModelTypeAndNodeName) +} + // addPathToChildrenPaths appends newPath to the config.artifact.childrenPaths array if the newPath is not contained in the children paths func (c *ConfigMapReconciler) addPathToChildrenPaths(modelTypeAndModelName string, newPath string, dataEntry string) (string, error) { // Parse the JSON into a generic map @@ -1314,6 +1427,221 @@ func (c *ConfigMapReconciler) updateConfigMapWithRemovedChildPath(ctx context.Co return err } +func (c *ConfigMapReconciler) upsertHuggingFaceArtifactParentEntry(ctx context.Context, parentName string, identity ArtifactIdentity, parentPath string, childPath string) error { + updateConfigMap := func(currentConfigMap *corev1.ConfigMap) (bool, *corev1.ConfigMap, error) { + if currentConfigMap.Data == nil { + currentConfigMap.Data = make(map[string]string) + } + + modelEntry := ModelEntry{ + Name: parentName, + Status: ModelStatusReady, + Config: &ModelConfig{ + Artifact: Artifact{ + Sha: strings.ToLower(identity.HFCommitSHA), + Origin: identity.toOrigin(), + ParentPath: map[string]string{parentName: parentPath}, + ChildrenPaths: []string{}, + }, + }, + } + if existingDataEntry, exists := currentConfigMap.Data[parentName]; exists { + var existing ModelEntry + if err := json.Unmarshal([]byte(existingDataEntry), &existing); err != nil { + c.logger.Warnf("failed to parse existing Hugging Face artifact parent entry %s, rebuilding it: %v", parentName, err) + } else { + modelEntry = existing + modelEntry.Name = parentName + modelEntry.Status = ModelStatusReady + if modelEntry.Config == nil { + modelEntry.Config = &ModelConfig{} + } + modelEntry.Config.Artifact.Sha = strings.ToLower(identity.HFCommitSHA) + modelEntry.Config.Artifact.Origin = identity.toOrigin() + modelEntry.Config.Artifact.ParentPath = map[string]string{parentName: parentPath} + if modelEntry.Config.Artifact.ChildrenPaths == nil { + modelEntry.Config.Artifact.ChildrenPaths = []string{} + } + } + } + + if strings.TrimSpace(childPath) != "" && !containsString(modelEntry.Config.Artifact.ChildrenPaths, childPath) { + modelEntry.Config.Artifact.ChildrenPaths = append(modelEntry.Config.Artifact.ChildrenPaths, childPath) + } + + entryJSON, err := json.Marshal(modelEntry) + if err != nil { + return false, currentConfigMap, err + } + currentConfigMap.Data[parentName] = string(entryJSON) + return true, currentConfigMap, nil + } + return c.updateConfigMapWithRetry(ctx, updateConfigMap) +} + +// setHuggingFaceArtifactParentStatus updates the synthetic Hugging Face parent +// status while preserving its origin, parent path, and childrenPaths metadata. +func (c *ConfigMapReconciler) setHuggingFaceArtifactParentStatus(ctx context.Context, parentName string, identity ArtifactIdentity, parentPath string, status ModelStatus, skipIfUpdating bool) (string, bool, error) { + observedParentPath := parentPath + updated := false + updateConfigMap := func(currentConfigMap *corev1.ConfigMap) (bool, *corev1.ConfigMap, error) { + if currentConfigMap.Data == nil { + currentConfigMap.Data = make(map[string]string) + } + + modelEntry := ModelEntry{ + Name: parentName, + Status: status, + Config: &ModelConfig{ + Artifact: Artifact{ + Sha: strings.ToLower(identity.HFCommitSHA), + Origin: identity.toOrigin(), + ParentPath: map[string]string{parentName: parentPath}, + ChildrenPaths: []string{}, + }, + }, + } + if existingDataEntry, exists := currentConfigMap.Data[parentName]; exists { + var existing ModelEntry + if err := json.Unmarshal([]byte(existingDataEntry), &existing); err != nil { + return false, currentConfigMap, fmt.Errorf("failed to parse existing Hugging Face artifact parent entry %s: %w", parentName, err) + } + if existing.Config == nil || existing.Config.Artifact.Origin == nil { + return false, currentConfigMap, fmt.Errorf("existing Hugging Face artifact parent entry %s is missing origin metadata", parentName) + } + origin := existing.Config.Artifact.Origin + if !strings.EqualFold(origin.Type, identity.OriginType) || origin.HFModelID != identity.HFModelID || !strings.EqualFold(origin.HFCommitSHA, identity.HFCommitSHA) { + return false, currentConfigMap, fmt.Errorf("existing Hugging Face artifact parent entry %s has mismatched origin metadata", parentName) + } + if existingParentPath := existing.Config.Artifact.ParentPath[parentName]; strings.TrimSpace(existingParentPath) != "" { + observedParentPath = existingParentPath + } + if skipIfUpdating && existing.Status == ModelStatusUpdating { + updated = false + return false, currentConfigMap, nil + } + modelEntry = existing + modelEntry.Name = parentName + modelEntry.Status = status + modelEntry.Config.Artifact.Sha = strings.ToLower(identity.HFCommitSHA) + modelEntry.Config.Artifact.Origin = identity.toOrigin() + modelEntry.Config.Artifact.ParentPath = map[string]string{parentName: observedParentPath} + if modelEntry.Config.Artifact.ChildrenPaths == nil { + modelEntry.Config.Artifact.ChildrenPaths = []string{} + } + } + + entryJSON, err := json.Marshal(modelEntry) + if err != nil { + return false, currentConfigMap, err + } + currentConfigMap.Data[parentName] = string(entryJSON) + updated = true + return true, currentConfigMap, nil + } + + err := c.updateConfigMapWithRetry(ctx, updateConfigMap) + return observedParentPath, updated, err +} + +// markHuggingFaceArtifactParentUpdating marks the synthetic Hugging Face parent +// as Updating before a DownloadOverride repair writes to the shared parent path. +// Existing childrenPaths are preserved so ready child entries keep pointing to +// the same parent metadata while the repair runs. If another worker already +// owns the Updating parent, acquired is false and the caller should wait. +func (c *ConfigMapReconciler) markHuggingFaceArtifactParentUpdating(ctx context.Context, parentName string, identity ArtifactIdentity, parentPath string) (string, bool, error) { + return c.setHuggingFaceArtifactParentStatus(ctx, parentName, identity, parentPath, ModelStatusUpdating, true) +} + +func (c *ConfigMapReconciler) markHuggingFaceArtifactParentFailed(ctx context.Context, parentName string, identity ArtifactIdentity, parentPath string) error { + _, _, err := c.setHuggingFaceArtifactParentStatus(ctx, parentName, identity, parentPath, ModelStatusFailed, false) + return err +} + +// reserveHuggingFaceArtifactParentEntry creates the synthetic parent in Updating +// state before the first worker downloads to the canonical path. Other workers +// see the reservation and wait instead of writing to the same directory. +func (c *ConfigMapReconciler) reserveHuggingFaceArtifactParentEntry(ctx context.Context, parentName string, identity ArtifactIdentity, parentPath string) (string, ModelStatus, bool, error) { + var observedParentPath string + var observedStatus ModelStatus + var reserved bool + + updateConfigMap := func(currentConfigMap *corev1.ConfigMap) (bool, *corev1.ConfigMap, error) { + if currentConfigMap.Data == nil { + currentConfigMap.Data = make(map[string]string) + } + + if existingDataEntry, exists := currentConfigMap.Data[parentName]; exists { + var existing ModelEntry + if err := json.Unmarshal([]byte(existingDataEntry), &existing); err != nil { + return false, currentConfigMap, fmt.Errorf("failed to parse existing Hugging Face artifact parent entry %s: %w", parentName, err) + } + if existing.Config == nil || existing.Config.Artifact.Origin == nil { + return false, currentConfigMap, fmt.Errorf("existing Hugging Face artifact parent entry %s is missing origin metadata", parentName) + } + origin := existing.Config.Artifact.Origin + if !strings.EqualFold(origin.Type, identity.OriginType) || origin.HFModelID != identity.HFModelID || !strings.EqualFold(origin.HFCommitSHA, identity.HFCommitSHA) { + return false, currentConfigMap, fmt.Errorf("existing Hugging Face artifact parent entry %s has mismatched origin metadata", parentName) + } + observedParentPath = existing.Config.Artifact.ParentPath[parentName] + if strings.TrimSpace(observedParentPath) == "" { + return false, currentConfigMap, fmt.Errorf("existing Hugging Face artifact parent entry %s is missing parent path", parentName) + } + observedStatus = existing.Status + reserved = false + return false, currentConfigMap, nil + } + + modelEntry := ModelEntry{ + Name: parentName, + Status: ModelStatusUpdating, + Config: &ModelConfig{ + Artifact: Artifact{ + Sha: strings.ToLower(identity.HFCommitSHA), + Origin: identity.toOrigin(), + ParentPath: map[string]string{parentName: parentPath}, + ChildrenPaths: []string{}, + }, + }, + } + entryJSON, err := json.Marshal(modelEntry) + if err != nil { + return false, currentConfigMap, err + } + currentConfigMap.Data[parentName] = string(entryJSON) + observedParentPath = parentPath + observedStatus = ModelStatusUpdating + reserved = true + return true, currentConfigMap, nil + } + + err := c.updateConfigMapWithRetry(ctx, updateConfigMap) + return observedParentPath, observedStatus, reserved, err +} + +func (c *ConfigMapReconciler) deleteConfigMapDataEntry(ctx context.Context, key string) error { + updateConfigMap := func(currentConfigMap *corev1.ConfigMap) (bool, *corev1.ConfigMap, error) { + if currentConfigMap.Data == nil { + return false, currentConfigMap, nil + } + if _, exists := currentConfigMap.Data[key]; !exists { + return false, currentConfigMap, nil + } + delete(currentConfigMap.Data, key) + return true, currentConfigMap, nil + } + return c.updateConfigMapWithRetry(ctx, updateConfigMap) +} + +func containsString(values []string, target string) bool { + for _, value := range values { + if strings.EqualFold(value, target) { + return true + } + } + return false +} + func (c *ConfigMapReconciler) getDataEntryBasedOnModelKey(ctx context.Context, modelKey string) (bool, string, error) { // get cm existingConfigMap, err := c.getConfigMap(ctx) diff --git a/pkg/modelagent/gopher.go b/pkg/modelagent/gopher.go index f66f95a05..8dd081dca 100644 --- a/pkg/modelagent/gopher.go +++ b/pkg/modelagent/gopher.go @@ -37,6 +37,11 @@ const ( Download GopherTaskType = "Download" DownloadOverride GopherTaskType = "DownloadOverride" Delete GopherTaskType = "Delete" + + huggingFaceArtifactConfigMapKeyPrefix = "artifact.huggingface." + huggingFaceArtifactReadyMarkerFile = ".ome-hf-artifact-ready" + artifactCompleteMarkerFileName = ".ome-artifact-complete" + artifactUploadLockFileName = ".ome-artifact-upload.lock" ) type GopherTask struct { @@ -80,14 +85,21 @@ type Gopher struct { samePathWaitTimeout time.Duration startupReadyModelKeys map[string]struct{} + + startupHuggingFaceParentRecoveryMutex sync.Mutex + startupHuggingFaceParentRecoveryPending bool + startupHuggingFaceParentValidationMutex sync.Mutex + startupHuggingFaceParentValidationKeys map[string]struct{} + startupHuggingFaceParentValidationTried map[string]struct{} + startupHuggingFaceParentSnapshotMissing bool } const ( BigFileSizeInMB = 200 - defaultSamePathWaitDelay = 30 * time.Second - defaultSamePathWaitTimeout = 30 * time.Minute - defaultStartupReadySnapshotTimeout = 5 * time.Second + defaultSamePathWaitDelay = 30 * time.Second + defaultSamePathWaitTimeout = 30 * time.Minute + defaultStartupConfigMapSnapshotTimeout = 5 * time.Second ) func NewGopher( @@ -137,9 +149,13 @@ func NewGopher( } func (s *Gopher) Run(stopCh <-chan struct{}, numWorker int, numHighPriorityWorker int) { - startupSnapshotCtx, cancelStartupSnapshot := context.WithTimeout(context.Background(), defaultStartupReadySnapshotTimeout) - defer cancelStartupSnapshot() + startupRecoveryCtx, cancelStartupRecovery := startupConfigMapContext() + s.recoverStartupHuggingFaceArtifactParents(startupRecoveryCtx) + cancelStartupRecovery() + + startupSnapshotCtx, cancelStartupSnapshot := startupConfigMapContext() s.captureStartupReadyModels(startupSnapshotCtx) + cancelStartupSnapshot() // Start the ConfigMap reconciliation service s.configMapReconciler.StartReconciliation() @@ -219,6 +235,9 @@ func (s *Gopher) runWorker() { if task.TaskType == Delete { s.cancelActiveDownload(task) } + if s.deferStartupReadyRevalidationIfLocalPathExists(task) { + continue + } err := s.processTask(task) if err != nil { s.logger.Errorf("Gopher task failed with error: %s", err.Error()) @@ -441,13 +460,38 @@ func (s *Gopher) processTaskWithOptions(task *GopherTask, allowFallbackDownload case storage.StorageTypeOCI: osUri, err := getTargetDirPath(&baseModelSpec) destPath := getDestPath(&baseModelSpec, s.modelRootDir) + var artifact *Artifact + hfOriginIdentity, hasHFOriginIdentity := huggingFaceArtifactIdentityFromTask(task) + useHuggingFaceOriginReuse := hasHFOriginIdentity && shouldUseHuggingFaceOriginObjectStorageReuse(task, baseModelSpec) + useHuggingFaceOriginRepair := hasHFOriginIdentity && shouldRepairHuggingFaceOriginObjectStorageParent(task, baseModelSpec) + hfArtifactParentKey := "" + hfArtifactParentPath := "" + if hasHFOriginIdentity { + s.logger.Infof("OCI model %s has Hugging Face origin metadata %s@%s for artifact reuse", + modelInfo, hfOriginIdentity.HFModelID, hfOriginIdentity.HFCommitSHA) + } + if useHuggingFaceOriginReuse || useHuggingFaceOriginRepair { + hfArtifactParentKey = huggingFaceArtifactConfigMapKey(hfOriginIdentity) + hfArtifactParentPath = canonicalHuggingFaceArtifactPathForTask(task, s.modelRootDir, destPath, hfOriginIdentity) + s.logger.Infof("OCI model %s will use canonical Hugging Face artifact parent %s at %s for %s", + modelInfo, hfArtifactParentKey, hfArtifactParentPath, task.TaskType) + } + if useHuggingFaceOriginRepair { + s.logger.Infof("OCI model %s will repair canonical Hugging Face artifact parent %s at %s", + modelInfo, hfArtifactParentKey, hfArtifactParentPath) + } if err != nil { s.logger.Errorf("Failed to get target directory path for model %s: %v", modelInfo, err) return err } - downloadObjectStorageModel := func() error { + if useHuggingFaceOriginReuse || useHuggingFaceOriginRepair { + if stop, recoveryErr := s.ensureStartupHuggingFaceArtifactParentsRecovered(ctx, task, hfArtifactParentKey); stop || recoveryErr != nil { + return recoveryErr + } + } + downloadObjectStorageModel := func(downloadPath string) error { err = utils.Retry(s.downloadRetry, 100*time.Millisecond, func() error { - downloadErr := s.downloadModel(ctx, osUri, destPath, task) + downloadErr := s.downloadModel(ctx, osUri, downloadPath, task) if downloadErr != nil { // Check if context was cancelled if ctx.Err() != nil { @@ -475,8 +519,97 @@ func (s *Gopher) processTaskWithOptions(task *GopherTask, allowFallbackDownload return nil } - if shouldUseSamePathObjectStorageReuse(task) { - if matchedKey, reused := s.findReadyObjectStorageModelWithSamePath(ctx, task, baseModelSpec, destPath); reused { + if useHuggingFaceOriginRepair { + var repaired bool + artifact, repaired, err = s.repairHuggingFaceOriginArtifactParent(ctx, task, name, destPath, hfArtifactParentKey, hfArtifactParentPath, hfOriginIdentity, downloadObjectStorageModel) + if err != nil { + return err + } + if !repaired { + return nil + } + } else if shouldUseSamePathObjectStorageReuse(task) { + var reused bool + if useHuggingFaceOriginReuse { + if stop, validationErr := s.validateStartupHuggingFaceArtifactParentIfNeeded(ctx, task, hfArtifactParentKey, hfOriginIdentity, downloadObjectStorageModel); stop || validationErr != nil { + return validationErr + } + artifact, reused, err = s.reuseHuggingFaceOriginArtifactIfPossible(ctx, task, baseModelSpec, modelType, namespace, name, destPath, hfOriginIdentity) + if err != nil { + return err + } + } + if reused { + s.logger.Infof("Reusing Ready Hugging Face origin model artifact for %s/%s at %s", namespace, name, destPath) + } else if useHuggingFaceOriginReuse { + if wait, waitErr := s.requeueIfHuggingFaceArtifactParentUpdating(ctx, task, hfOriginIdentity); wait || waitErr != nil { + return waitErr + } + if !allowFallbackDownload { + s.demoteToNormalPriority(task) + return nil + } + parentPath, parentStatus, reserved, err := s.configMapReconciler.reserveHuggingFaceArtifactParentEntry(ctx, hfArtifactParentKey, hfOriginIdentity, hfArtifactParentPath) + if err != nil { + return err + } + rebuildParentPath := hfArtifactParentPath + if strings.TrimSpace(parentPath) != "" { + rebuildParentPath = parentPath + } + if !reserved { + switch parentStatus { + case ModelStatusUpdating: + if s.requeueSamePathInFlightReuseWait(task, hfArtifactParentKey) { + return nil + } + return fmt.Errorf("timed out waiting for Hugging Face artifact parent %s to become Ready", hfArtifactParentKey) + case ModelStatusReady: + artifact, reused, err = s.reuseHuggingFaceOriginArtifactIfPossible(ctx, task, baseModelSpec, modelType, namespace, name, destPath, hfOriginIdentity) + if err != nil { + return err + } + if reused { + s.logger.Infof("Reusing Ready Hugging Face origin model artifact for %s/%s at %s after reservation check", namespace, name, destPath) + } else { + s.logger.Warnf("Hugging Face artifact parent %s is Ready but cannot be reused from %s; rebuilding canonical parent", hfArtifactParentKey, parentPath) + } + } + } + if !reused { + if !reserved { + var acquired bool + rebuildParentPath, acquired, err = s.acquireHuggingFaceArtifactParentRebuild(ctx, task, hfArtifactParentKey, rebuildParentPath, hfOriginIdentity) + if err != nil { + return err + } + if !acquired { + return nil + } + } + if err := downloadObjectStorageModel(rebuildParentPath); err != nil { + if reserved { + if cleanupErr := s.configMapReconciler.deleteConfigMapDataEntry(ctx, hfArtifactParentKey); cleanupErr != nil { + s.logger.Warnf("failed to remove Hugging Face artifact parent reservation %s after download failure: %v", hfArtifactParentKey, cleanupErr) + } + } else if failErr := s.configMapReconciler.markHuggingFaceArtifactParentFailed(ctx, hfArtifactParentKey, hfOriginIdentity, rebuildParentPath); failErr != nil { + s.logger.Warnf("failed to mark Hugging Face artifact parent %s at %s as Failed after rebuild error: %v", hfArtifactParentKey, rebuildParentPath, failErr) + } + return err + } + if markerErr := writeHuggingFaceArtifactReadyMarker(rebuildParentPath); markerErr != nil { + s.logger.Warnf("failed to write Hugging Face artifact ready marker for parent %s at %s: %v", hfArtifactParentKey, rebuildParentPath, markerErr) + } + if err := s.markHuggingFaceArtifactParentReady(ctx, hfArtifactParentKey, rebuildParentPath, hfOriginIdentity); err != nil { + s.logger.Errorf("downloaded Hugging Face artifact parent %s at %s but failed to mark it Ready: %v", hfArtifactParentKey, rebuildParentPath, err) + return err + } + artifact, err = s.linkHuggingFaceOriginArtifact(ctx, task, name, destPath, hfArtifactParentKey, rebuildParentPath, hfOriginIdentity) + if err != nil { + return err + } + } + } else if matchedKey, reused := s.findReadyObjectStorageModelWithSamePath(ctx, task, baseModelSpec, destPath); reused { s.logger.Infof("Reusing Ready same-path model artifact for %s/%s from %s at %s", namespace, name, matchedKey, destPath) } else if matchedKey, wait := s.findUpdatingObjectStorageModelWithSamePath(ctx, task, baseModelSpec, destPath); wait && s.requeueSamePathInFlightReuseWait(task, matchedKey) { @@ -484,12 +617,15 @@ func (s *Gopher) processTaskWithOptions(task *GopherTask, allowFallbackDownload } else if !allowFallbackDownload { s.demoteToNormalPriority(task) return nil - } else if err := downloadObjectStorageModel(); err != nil { + } else if err := downloadObjectStorageModel(destPath); err != nil { return err } - } else if err := downloadObjectStorageModel(); err != nil { + } else if err := downloadObjectStorageModel(destPath); err != nil { return err } + if artifact == nil && hasHFOriginIdentity { + artifact = s.buildSelfParentArtifactFromIdentity(ctx, task, hfOriginIdentity, destPath) + } // Parse model config and update ConfigMap // We can pass either BaseModel or ClusterBaseModel based on the task's model type var baseModel *v1beta1.BaseModel @@ -506,7 +642,7 @@ func (s *Gopher) processTaskWithOptions(task *GopherTask, allowFallbackDownload s.logger.Warnf("No model object found in task, skipping config parsing") } - if err := s.safeParseAndUpdateModelConfig(destPath, baseModel, clusterBaseModel, nil); err != nil { + if err := s.safeParseAndUpdateModelConfig(destPath, baseModel, clusterBaseModel, artifact); err != nil { s.logger.Errorf("Failed to parse and update model config: %v", err) } case storage.StorageTypeVendor: @@ -591,7 +727,7 @@ func (s *Gopher) processTaskWithOptions(task *GopherTask, allowFallbackDownload s.logger.Infof("Starting deletion for model %s", modelInfo) destPath := getDestPath(&baseModelSpec, s.modelRootDir) // check if it needs to skip artifact deletion - isSkippingDeletion, _, _, _ := s.isSkippingArtifactDeletion(ctx, task, destPath, false) + isSkippingDeletion, isRemoveParent, parentName, parentDir := s.isSkippingArtifactDeletion(ctx, task, destPath, s.hasSharedArtifactMetadata(ctx, task)) if !isSkippingDeletion { err = s.deleteModel(destPath, task) if err != nil { @@ -603,7 +739,10 @@ func (s *Gopher) processTaskWithOptions(task *GopherTask, allowFallbackDownload } else { s.logger.Infof("Successfully deleted the ClusterBaseModel %s", task.ClusterBaseModel.Name) } + } else { + s.logger.Infof("model %s artifact deletion will be skipped", modelInfo) } + s.deleteParentArtifactIfUnreferenced(ctx, isRemoveParent, parentName, parentDir) case storage.StorageTypeVendor: s.logger.Infof("Skipping deletion for model %s", modelInfo) case storage.StorageTypeHuggingFace: @@ -624,24 +763,7 @@ func (s *Gopher) processTaskWithOptions(task *GopherTask, allowFallbackDownload } else { s.logger.Infof("model %s artifact deletion will be skipped", modelInfo) } - if isRemoveParent && parentName != "" && parentDir != "" { - // check whether the parent directory still has other directory points to it using symbolic link - isParentHasSymbolicLinkPointedTo, symbolicLinkSearchErr := utils.HasSymlinkPointingToDir(s.modelRootDir, parentDir) - if symbolicLinkSearchErr != nil { - s.logger.Infof("fails to search for the SymbolicLink pointing to parent Dir %s: %v. will regard the parent is still being pointed conservatively", parentDir, symbolicLinkSearchErr) - isParentHasSymbolicLinkPointedTo = true - } - s.logger.Infof("parent %s:%s has other directory points to: %v", parentName, parentDir, isParentHasSymbolicLinkPointedTo) - if !isParentHasSymbolicLinkPointedTo { - err = s.deleteModel(parentDir, nil) - if err != nil { - s.logger.Errorf("fail to delete parent model artifact directory %s: %s", parentName, parentDir) - } - s.logger.Infof("Successfully delete parent model artifact directory %s: %s", parentName, parentDir) - } - } else { - s.logger.Infof("no need to delete parent model artifact directory %s: %s", parentName, parentDir) - } + s.deleteParentArtifactIfUnreferenced(ctx, isRemoveParent, parentName, parentDir) case storage.StorageTypeLocal: s.logger.Infof("Skipping deletion for local storage model %s (local files should not be deleted)", modelInfo) // For local storage, we should NOT delete the actual files @@ -714,21 +836,34 @@ func (s *Gopher) captureStartupReadyModels(ctx context.Context) { 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 + } else { + s.logger.Warnf("Cannot capture startup Ready model snapshot: %v", err) } - s.logger.Warnf("Cannot capture startup Ready model snapshot: %v", err) s.startupReadyModelKeys = map[string]struct{}{} + s.startupHuggingFaceParentValidationMutex.Lock() + s.startupHuggingFaceParentValidationKeys = map[string]struct{}{} + s.startupHuggingFaceParentValidationTried = map[string]struct{}{} + s.startupHuggingFaceParentSnapshotMissing = true + s.startupHuggingFaceParentValidationMutex.Unlock() return } readyModelKeys := make(map[string]struct{}) + readyHuggingFaceParentKeys := make(map[string]struct{}) for key, data := range configMap.Data { if hasModelEntryStatus(data, ModelStatusReady) { readyModelKeys[key] = struct{}{} + if isHuggingFaceArtifactConfigMapKey(key) { + readyHuggingFaceParentKeys[key] = struct{}{} + } } } s.startupReadyModelKeys = readyModelKeys - s.logger.Infof("Captured %d Ready models from startup ConfigMap snapshot", len(readyModelKeys)) + s.startupHuggingFaceParentValidationMutex.Lock() + s.startupHuggingFaceParentValidationKeys = readyHuggingFaceParentKeys + s.startupHuggingFaceParentValidationTried = map[string]struct{}{} + s.startupHuggingFaceParentSnapshotMissing = false + s.startupHuggingFaceParentValidationMutex.Unlock() + s.logger.Infof("Captured %d Ready models and %d Ready Hugging Face artifact parents from startup ConfigMap snapshot", len(readyModelKeys), len(readyHuggingFaceParentKeys)) } func (s *Gopher) isStartupRevalidation(task *GopherTask) bool { @@ -1023,7 +1158,8 @@ func (s *Gopher) createOCIOSDataStore(baseModelSpec v1beta1.BaseModelSpec) (*oci // model entry on this node that resolves to the same local destination path. // This is intentionally independent of downloadPolicy for normal Download tasks: // copied model CRs with the same source and destination can reuse Ready local -// files. DownloadOverride keeps the existing download/validation path. +// files. DownloadOverride is handled separately so it can validate and repair +// the relevant artifact instead of skipping work. func (s *Gopher) findReadyObjectStorageModelWithSamePath(ctx context.Context, task *GopherTask, baseModelSpec v1beta1.BaseModelSpec, destPath string) (string, bool) { return s.findObjectStorageModelWithSamePathAndStatus(ctx, task, baseModelSpec, destPath, ModelStatusReady, true) } @@ -1229,6 +1365,7 @@ func (s *Gopher) downloadModel(ctx context.Context, uri *ociobjectstore.ObjectUR if err != nil { return fmt.Errorf("failed to list objects: %w", err) } + objects = filterInternalArtifactObjectSummaries(objects) if len(objects) == 0 { return fmt.Errorf("no objects found under namespace %s, bucket %s, object prefix %s", uri.Namespace, uri.BucketName, uri.Prefix) @@ -1370,6 +1507,70 @@ func (s *Gopher) deleteModel(destPath string, task *GopherTask) error { return err } +func (s *Gopher) hasSharedArtifactMetadata(ctx context.Context, task *GopherTask) bool { + if task == nil || s.configMapReconciler == nil { + return false + } + modelTypeAndModelName := s.configMapReconciler.getModelConfigMapKey(task.BaseModel, task.ClusterBaseModel) + exists, dataEntry, err := s.configMapReconciler.getDataEntryBasedOnModelKey(ctx, modelTypeAndModelName) + if err != nil { + s.logger.Warnf("cannot inspect artifact metadata for %s during deletion: %v", modelTypeAndModelName, err) + return false + } + if !exists { + return false + } + + var entry ModelEntry + if err := json.Unmarshal([]byte(dataEntry), &entry); err != nil { + s.logger.Warnf("cannot parse artifact metadata for %s during deletion: %v", modelTypeAndModelName, err) + return false + } + if entry.Config == nil { + return false + } + hasParentPath := false + for _, parentPath := range entry.Config.Artifact.ParentPath { + if strings.TrimSpace(parentPath) != "" { + hasParentPath = true + break + } + } + if hasParentPath || len(entry.Config.Artifact.ChildrenPaths) > 0 { + return true + } + return false +} + +func (s *Gopher) deleteParentArtifactIfUnreferenced(ctx context.Context, isRemoveParent bool, parentName string, parentDir string) { + if !isRemoveParent || parentName == "" || parentDir == "" { + s.logger.Infof("no need to delete parent model artifact directory %s: %s", parentName, parentDir) + return + } + // Check whether any child path still points to the parent directory before + // removing the shared artifact. + isParentHasSymbolicLinkPointedTo, symbolicLinkSearchErr := utils.HasSymlinkPointingToDir(s.modelRootDir, parentDir) + if symbolicLinkSearchErr != nil { + s.logger.Infof("fails to search for the SymbolicLink pointing to parent Dir %s: %v. will regard the parent is still being pointed conservatively", parentDir, symbolicLinkSearchErr) + isParentHasSymbolicLinkPointedTo = true + } + s.logger.Infof("parent %s:%s has other directory points to: %v", parentName, parentDir, isParentHasSymbolicLinkPointedTo) + if isParentHasSymbolicLinkPointedTo { + return + } + + if err := s.deleteModel(parentDir, nil); err != nil { + s.logger.Errorf("fail to delete parent model artifact directory %s: %s", parentName, parentDir) + return + } + s.logger.Infof("Successfully delete parent model artifact directory %s: %s", parentName, parentDir) + if isHuggingFaceArtifactConfigMapKey(parentName) { + if err := s.configMapReconciler.deleteConfigMapDataEntry(ctx, parentName); err != nil { + s.logger.Errorf("failed to delete Hugging Face artifact parent entry %s from ConfigMap: %v", parentName, err) + } + } +} + // isReservingModelArtifact determines whether to preserve the model artifact directory during deletion. // Behavior: // - Returns true if either ClusterBaseModel or BaseModel has the label models.ome/reserve-model-artifact @@ -1379,6 +1580,7 @@ func (s *Gopher) deleteModel(destPath string, task *GopherTask) error { // Precedence: // - If both BaseModel and ClusterBaseModel exist and at least one has the reserve label set to "true", // the function returns true (i.e., preserve the artifact). + func (s *Gopher) isReservingModelArtifact(task *GopherTask) bool { // Guard against nil task or BaseModel; reserve logic applies only to BaseModel labels if task == nil { @@ -1423,6 +1625,11 @@ func (s *Gopher) processHuggingFaceModel(ctx context.Context, task *GopherTask, // fetch sha value based on model ID from Huggingface model API shaStr, isShaAvailable := s.fetchSha(ctx, hfComponents.ModelID, name) + hfOriginIdentity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: hfComponents.ModelID, + HFCommitSHA: strings.ToLower(shaStr), + } isReuseEligible, matchedModelTypeAndModeName, parentPath := s.isEligibleForOptimization(ctx, task, baseModelSpec, modelType, namespace, isShaAvailable, shaStr, name) var artifact *Artifact @@ -1444,7 +1651,7 @@ func (s *Gopher) processHuggingFaceModel(ctx context.Context, task *GopherTask, childrenPaths := make([]string, 0) childrenPaths, _, _, _ = s.parseModelConfigDataEntry(ctx, s.configMapReconciler.getModelConfigMapKey(task.BaseModel, task.ClusterBaseModel)) - artifact = s.modelConfigParser.BuildArtifactAttribute(shaStr, matchedModelTypeAndModeName, parentPath, childrenPaths) + artifact = s.buildArtifactAttributeFromIdentity(hfOriginIdentity, matchedModelTypeAndModeName, parentPath, childrenPaths) } else { childrenPaths := make([]string, 0) // handle the case when download Policy is updated from ReuseIfExists to AlwaysDownload @@ -1596,7 +1803,7 @@ func (s *Gopher) processHuggingFaceModel(ctx context.Context, task *GopherTask, s.logger.Infof("Successfully downloaded HuggingFace model %s to %s", modelInfo, downloadPath) - artifact = s.modelConfigParser.BuildArtifactAttribute(shaStr, s.configMapReconciler.getModelConfigMapKey(task.BaseModel, task.ClusterBaseModel), destPath, childrenPaths) + artifact = s.buildArtifactAttributeFromIdentity(hfOriginIdentity, s.configMapReconciler.getModelConfigMapKey(task.BaseModel, task.ClusterBaseModel), destPath, childrenPaths) } // Parse model config and update ConfigMap @@ -1882,11 +2089,20 @@ func (s *Gopher) isRemoveParentArtifactDirectory(ctx context.Context, hasChildre } // If the parent entry still exists in the node ConfigMap, don't remove. - exists, _, err := s.configMapReconciler.getDataEntryBasedOnModelKey(ctx, parentName) + exists, dataEntry, err := s.configMapReconciler.getDataEntryBasedOnModelKey(ctx, parentName) if err != nil && strings.Contains(err.Error(), "cannot retrieve node configmap") { s.logger.Infof("cannot retrieve node configmap and cannot determine parent entry existence, will not remove artifact") return false } + if exists && isHuggingFaceArtifactConfigMapKey(parentName) { + _, parentChildrenPaths, parseErr := s.configMapReconciler.getParentPathAndChildrenPaths(parentName, dataEntry) + if parseErr != nil { + s.logger.Infof("cannot parse Hugging Face artifact parent entry %s, will not remove artifact: %v", parentName, parseErr) + return false + } + s.logger.Infof("Hugging Face artifact parent entry %s:%s has children paths: %v", parentName, parentDir, parentChildrenPaths) + return len(parentChildrenPaths) == 0 + } s.logger.Infof("parent entry %s:%s exists on node configmap: %v", parentName, parentDir, exists) return !exists } diff --git a/pkg/modelagent/gopher_hf_artifact.go b/pkg/modelagent/gopher_hf_artifact.go new file mode 100644 index 000000000..f55dcda10 --- /dev/null +++ b/pkg/modelagent/gopher_hf_artifact.go @@ -0,0 +1,244 @@ +package modelagent + +import ( + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + + "sigs.k8s.io/ome/pkg/apis/ome/v1beta1" + "sigs.k8s.io/ome/pkg/utils" +) + +func (s *Gopher) buildArtifactAttributeFromIdentity(identity ArtifactIdentity, matchedParentName string, parentPath string, childrenPaths []string) *Artifact { + return &Artifact{ + Sha: identity.HFCommitSHA, + Origin: identity.toOrigin(), + ParentPath: map[string]string{matchedParentName: parentPath}, + ChildrenPaths: childrenPaths, + } +} + +func (s *Gopher) buildSelfParentArtifactFromIdentity(ctx context.Context, task *GopherTask, identity ArtifactIdentity, destPath string) *Artifact { + currentModelKey := s.configMapReconciler.getModelConfigMapKey(task.BaseModel, task.ClusterBaseModel) + childrenPaths, _, _, _ := s.parseModelConfigDataEntry(ctx, currentModelKey) + return s.buildArtifactAttributeFromIdentity(identity, currentModelKey, destPath, childrenPaths) +} + +func (s *Gopher) reuseHuggingFaceOriginArtifactIfPossible(ctx context.Context, task *GopherTask, baseModelSpec v1beta1.BaseModelSpec, + modelType string, namespace string, name string, destPath string, identity ArtifactIdentity) (*Artifact, bool, error) { + if !shouldUseHuggingFaceOriginObjectStorageReuse(task, baseModelSpec) { + return nil, false, nil + } + if !identity.isValid() { + return nil, false, nil + } + + currentModelKey := s.configMapReconciler.getModelConfigMapKey(task.BaseModel, task.ClusterBaseModel) + matchedModelKey, parentPath := s.handleHuggingFaceOriginReuseIfNecessary(ctx, modelType, name, namespace, identity, currentModelKey) + if matchedModelKey == "" || parentPath == "" { + return nil, false, nil + } + if _, err := os.Stat(parentPath); err != nil { + s.logger.Warnf("Cannot reuse Hugging Face origin artifact for %s from %s because parent path %s is not available locally: %v", name, matchedModelKey, parentPath, err) + return nil, false, nil + } + + artifact, err := s.linkHuggingFaceOriginArtifact(ctx, task, name, destPath, matchedModelKey, parentPath, identity) + if err != nil { + return nil, false, err + } + return artifact, true, nil +} + +func (s *Gopher) repairHuggingFaceOriginArtifactParent(ctx context.Context, task *GopherTask, name string, destPath string, + parentKey string, parentPath string, identity ArtifactIdentity, downloadObjectStorageModel func(string) error) (*Artifact, bool, error) { + if !identity.isValid() || strings.TrimSpace(parentKey) == "" || strings.TrimSpace(parentPath) == "" { + return nil, false, nil + } + if wait, waitErr := s.requeueIfHuggingFaceArtifactParentUpdating(ctx, task, identity); wait || waitErr != nil { + return nil, false, waitErr + } + + repairPath := parentPath + if _, observedParentPath, _, ok := s.getHuggingFaceArtifactParent(ctx, identity); ok { + repairPath = observedParentPath + } + repairPath, acquired, err := s.configMapReconciler.markHuggingFaceArtifactParentUpdating(ctx, parentKey, identity, repairPath) + if err != nil { + return nil, false, err + } + if !acquired { + if s.requeueSamePathInFlightReuseWait(task, parentKey) { + return nil, false, nil + } + return nil, false, fmt.Errorf("timed out waiting for Hugging Face artifact parent %s to become available for repair", parentKey) + } + if markerErr := removeHuggingFaceArtifactReadyMarker(repairPath); markerErr != nil { + if failErr := s.configMapReconciler.markHuggingFaceArtifactParentFailed(ctx, parentKey, identity, repairPath); failErr != nil { + s.logger.Warnf("failed to mark Hugging Face artifact parent %s at %s as Failed after ready marker removal error: %v", parentKey, repairPath, failErr) + } + return nil, false, fmt.Errorf("failed to remove Hugging Face artifact ready marker for parent %s at %s before repair: %w", parentKey, repairPath, markerErr) + } + s.logger.Infof("Repairing Hugging Face artifact parent %s at %s for OCI model %s using origin %s@%s", + parentKey, repairPath, name, identity.HFModelID, identity.HFCommitSHA) + + if err := downloadObjectStorageModel(repairPath); err != nil { + s.logger.Errorf("failed to repair Hugging Face artifact parent %s at %s: %v", parentKey, repairPath, err) + if failErr := s.configMapReconciler.markHuggingFaceArtifactParentFailed(ctx, parentKey, identity, repairPath); failErr != nil { + s.logger.Warnf("failed to mark Hugging Face artifact parent %s at %s as Failed after repair error: %v", parentKey, repairPath, failErr) + } + return nil, false, err + } + if markerErr := writeHuggingFaceArtifactReadyMarker(repairPath); markerErr != nil { + s.logger.Warnf("failed to write Hugging Face artifact ready marker for repaired parent %s at %s: %v", parentKey, repairPath, markerErr) + } + if err := s.markHuggingFaceArtifactParentReady(ctx, parentKey, repairPath, identity); err != nil { + s.logger.Errorf("repaired Hugging Face artifact parent %s at %s but failed to mark it Ready: %v", parentKey, repairPath, err) + return nil, false, err + } + artifact, err := s.linkHuggingFaceOriginArtifact(ctx, task, name, destPath, parentKey, repairPath, identity) + if err != nil { + return nil, false, err + } + return artifact, true, nil +} + +func (s *Gopher) acquireHuggingFaceArtifactParentRebuild(ctx context.Context, task *GopherTask, parentKey string, parentPath string, identity ArtifactIdentity) (string, bool, error) { + rebuildPath, acquired, err := s.configMapReconciler.markHuggingFaceArtifactParentUpdating(ctx, parentKey, identity, parentPath) + if err != nil { + return "", false, err + } + if acquired { + if markerErr := removeHuggingFaceArtifactReadyMarker(rebuildPath); markerErr != nil { + if failErr := s.configMapReconciler.markHuggingFaceArtifactParentFailed(ctx, parentKey, identity, rebuildPath); failErr != nil { + s.logger.Warnf("failed to mark Hugging Face artifact parent %s at %s as Failed after ready marker removal error: %v", parentKey, rebuildPath, failErr) + } + return "", false, fmt.Errorf("failed to remove Hugging Face artifact ready marker for parent %s at %s before rebuild: %w", parentKey, rebuildPath, markerErr) + } + return rebuildPath, true, nil + } + if s.requeueSamePathInFlightReuseWait(task, parentKey) { + return "", false, nil + } + return "", false, fmt.Errorf("timed out waiting for Hugging Face artifact parent %s to become available for rebuild", parentKey) +} + +func (s *Gopher) linkHuggingFaceOriginArtifact(ctx context.Context, task *GopherTask, name string, destPath string, parentKey string, parentPath string, identity ArtifactIdentity) (*Artifact, error) { + if filepath.Clean(destPath) != filepath.Clean(parentPath) { + if err := utils.CreateSymbolicLink(destPath, parentPath); err != nil { + s.logger.Errorf("failed to create symbolic link from %s to %s for OCI model %s with Hugging Face origin %s@%s: %s", + destPath, parentPath, name, identity.HFModelID, identity.HFCommitSHA, err) + return nil, err + } + s.logger.Infof("Successfully created symbolic link from %s to %s for OCI model %s using Hugging Face origin %s@%s", + destPath, parentPath, name, identity.HFModelID, identity.HFCommitSHA) + } else { + s.logger.Infof("OCI model %s already uses canonical Hugging Face artifact path %s", name, parentPath) + } + + if err := s.recordHuggingFaceOriginChildPath(ctx, parentKey, parentPath, destPath, identity); err != nil { + s.logger.Errorf("fail to update configmap to add OCI model path %s to parent %s childrenPaths: %s", destPath, parentKey, err) + return nil, err + } + s.logger.Infof("Successfully added OCI model path %s to parent %s childrenPaths", destPath, parentKey) + + // A linked OCI model is a child of the canonical Hugging Face artifact entry. + // The child entry may only have Updating status at this point, so do not parse + // it for artifact children; the parent entry owns the childrenPaths list. + return s.buildArtifactAttributeFromIdentity(identity, parentKey, parentPath, []string{}), nil +} + +func (s *Gopher) recordHuggingFaceOriginChildPath(ctx context.Context, parentKey string, parentPath string, childPath string, identity ArtifactIdentity) error { + if isHuggingFaceArtifactConfigMapKey(parentKey) { + return s.configMapReconciler.upsertHuggingFaceArtifactParentEntry(ctx, parentKey, identity, parentPath, childPath) + } + return s.configMapReconciler.updateConfigMapWithUpdatedChildrenPaths(ctx, parentKey, childPath) +} + +func (s *Gopher) markHuggingFaceArtifactParentReady(ctx context.Context, parentKey string, parentPath string, identity ArtifactIdentity) error { + return s.configMapReconciler.upsertHuggingFaceArtifactParentEntry(ctx, parentKey, identity, parentPath, "") +} + +func (s *Gopher) getHuggingFaceArtifactParent(ctx context.Context, identity ArtifactIdentity) (string, string, ModelStatus, bool) { + parentKey, parentPath, status, ok, _ := s.getHuggingFaceArtifactParentWithError(ctx, identity) + return parentKey, parentPath, status, ok +} + +func (s *Gopher) getHuggingFaceArtifactParentWithError(ctx context.Context, identity ArtifactIdentity) (string, string, ModelStatus, bool, error) { + parentKey := huggingFaceArtifactConfigMapKey(identity) + exists, dataEntry, err := s.configMapReconciler.getDataEntryBasedOnModelKey(ctx, parentKey) + if err != nil { + s.logger.Warnf("cannot inspect Hugging Face artifact parent %s: %v", parentKey, err) + return "", "", "", false, err + } + if !exists { + return "", "", "", false, nil + } + + var entry ModelEntry + if err := json.Unmarshal([]byte(dataEntry), &entry); err != nil { + s.logger.Warnf("cannot parse Hugging Face artifact parent %s: %v", parentKey, err) + return "", "", "", false, nil + } + if entry.Config == nil { + return "", "", "", false, nil + } + origin := entry.Config.Artifact.Origin + if origin == nil || !strings.EqualFold(origin.Type, identity.OriginType) || origin.HFModelID != identity.HFModelID || !strings.EqualFold(origin.HFCommitSHA, identity.HFCommitSHA) { + return "", "", "", false, nil + } + parentPath := entry.Config.Artifact.ParentPath[parentKey] + if strings.TrimSpace(parentPath) == "" { + return "", "", "", false, nil + } + return parentKey, parentPath, entry.Status, true, nil +} + +func (s *Gopher) getReadyHuggingFaceArtifactParent(ctx context.Context, identity ArtifactIdentity) (string, string, bool) { + parentKey, parentPath, status, ok := s.getHuggingFaceArtifactParent(ctx, identity) + return parentKey, parentPath, ok && status == ModelStatusReady +} + +// requeueIfHuggingFaceArtifactParentUpdating prevents concurrent workers from +// downloading into the same canonical parent path. +func (s *Gopher) requeueIfHuggingFaceArtifactParentUpdating(ctx context.Context, task *GopherTask, identity ArtifactIdentity) (bool, error) { + parentKey, parentPath, status, ok := s.getHuggingFaceArtifactParent(ctx, identity) + if !ok || status != ModelStatusUpdating { + return false, nil + } + // The ready marker is written only after object storage download and checksum + // verification complete. It lets a later task recover if ConfigMap or symlink + // finalization failed after the canonical parent was fully downloaded. + if hasHuggingFaceArtifactReadyMarker(parentPath) { + if err := s.markHuggingFaceArtifactParentReady(ctx, parentKey, parentPath, identity); err != nil { + s.logger.Warnf("Hugging Face artifact parent %s has a ready marker at %s but cannot be marked Ready yet: %v", parentKey, parentPath, err) + } else { + s.logger.Infof("Recovered Hugging Face artifact parent %s from ready marker at %s", parentKey, parentPath) + return false, nil + } + } + if s.requeueSamePathInFlightReuseWait(task, parentKey) { + return true, nil + } + return true, fmt.Errorf("timed out waiting for Hugging Face artifact parent %s to become Ready", parentKey) +} + +// handleHuggingFaceOriginReuseIfNecessary determines whether an OCI model can +// reuse an existing Ready artifact by comparing provenance metadata. The source +// storage remains OCI; the Hugging Face identity is only used to prove that the +// OCI object prefix was imported from the same HF model revision. +func (s *Gopher) handleHuggingFaceOriginReuseIfNecessary(ctx context.Context, modelType string, modelName string, namespace string, identity ArtifactIdentity, currentModelTypeAndNodeName string) (string, string) { + if !identity.isValid() { + return "", "" + } + if parentKey, parentPath, ok := s.getReadyHuggingFaceArtifactParent(ctx, identity); ok { + s.logger.Infof("found canonical Hugging Face artifact parent %s for model %s, parentPath is %s", parentKey, modelName, parentPath) + return parentKey, parentPath + } + + s.logger.Infof("no canonical Hugging Face artifact parent found for model %s with identity %s@%s", modelName, identity.HFModelID, identity.HFCommitSHA) + return "", "" +} diff --git a/pkg/modelagent/gopher_hf_artifact_startup.go b/pkg/modelagent/gopher_hf_artifact_startup.go new file mode 100644 index 000000000..0ce119cdb --- /dev/null +++ b/pkg/modelagent/gopher_hf_artifact_startup.go @@ -0,0 +1,223 @@ +package modelagent + +import ( + "context" + "encoding/json" + "fmt" + "strings" + "time" + + apierrors "k8s.io/apimachinery/pkg/api/errors" +) + +var startupConfigMapContext = func() (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), defaultStartupConfigMapSnapshotTimeout) +} + +func (s *Gopher) deferStartupReadyRevalidationIfLocalPathExists(task *GopherTask) bool { + if s.classifyStartupRevalidation(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) recoverStartupHuggingFaceArtifactParents(ctx context.Context) bool { + s.startupHuggingFaceParentRecoveryMutex.Lock() + defer s.startupHuggingFaceParentRecoveryMutex.Unlock() + + recovered := s.recoverStartupHuggingFaceArtifactParentsLocked(ctx) + s.startupHuggingFaceParentRecoveryPending = !recovered + return recovered +} + +func (s *Gopher) recoverStartupHuggingFaceArtifactParentsLocked(ctx context.Context) bool { + if s.configMapReconciler == nil { + return true + } + configMap, err := s.configMapReconciler.getConfigMap(ctx) + if err != nil { + if apierrors.IsNotFound(err) { + s.logger.Info("No startup ConfigMap found for Hugging Face artifact parent recovery") + return true + } + s.logger.Warnf("Cannot recover startup Hugging Face artifact parents: %v", err) + return false + } + recovered := true + for parentKey, dataEntry := range configMap.Data { + if !isHuggingFaceArtifactConfigMapKey(parentKey) { + continue + } + var entry ModelEntry + if err := json.Unmarshal([]byte(dataEntry), &entry); err != nil { + s.logger.Warnf("Cannot parse Hugging Face artifact parent %s during startup recovery: %v", parentKey, err) + continue + } + if entry.Status != ModelStatusUpdating { + continue + } + identity, parentPath, ok := huggingFaceArtifactParentIdentityAndPath(parentKey, entry) + if !ok { + s.logger.Warnf("Cannot recover Updating Hugging Face artifact parent %s because it is missing valid identity or parent path", parentKey) + continue + } + if hasHuggingFaceArtifactReadyMarker(parentPath) { + if err := s.markHuggingFaceArtifactParentReady(ctx, parentKey, parentPath, identity); err != nil { + s.logger.Warnf("Cannot mark Hugging Face artifact parent %s Ready during startup recovery: %v", parentKey, err) + recovered = false + } else { + s.logger.Infof("Recovered Hugging Face artifact parent %s as Ready from startup ready marker at %s", parentKey, parentPath) + } + continue + } + if err := s.configMapReconciler.markHuggingFaceArtifactParentFailed(ctx, parentKey, identity, parentPath); err != nil { + s.logger.Warnf("Cannot mark stale Hugging Face artifact parent %s Failed during startup recovery: %v", parentKey, err) + recovered = false + } else { + s.logger.Infof("Marked stale Hugging Face artifact parent %s Failed during startup recovery because ready marker is missing at %s", parentKey, parentPath) + } + } + return recovered +} + +func (s *Gopher) isStartupHuggingFaceArtifactParentRecoveryPending() bool { + s.startupHuggingFaceParentRecoveryMutex.Lock() + defer s.startupHuggingFaceParentRecoveryMutex.Unlock() + return s.startupHuggingFaceParentRecoveryPending +} + +func (s *Gopher) ensureStartupHuggingFaceArtifactParentsRecovered(ctx context.Context, task *GopherTask, parentKey string) (bool, error) { + if !s.isStartupHuggingFaceArtifactParentRecoveryPending() { + return false, nil + } + if s.recoverStartupHuggingFaceArtifactParents(ctx) { + return false, nil + } + if s.requeueStartupHuggingFaceArtifactParentRecoveryWait(task, parentKey) { + return true, nil + } + return true, fmt.Errorf("timed out waiting for startup Hugging Face artifact parent recovery before processing %s", parentKey) +} + +func (s *Gopher) requeueStartupHuggingFaceArtifactParentRecoveryWait(task *GopherTask, parentKey string) bool { + if task == nil || s.gopherChan == nil { + return false + } + now := time.Now() + timeout := s.samePathWaitTimeout + if timeout <= 0 { + timeout = defaultSamePathWaitTimeout + } + if task.SamePathWaitStartedAt.IsZero() { + task.SamePathWaitStartedAt = now + } else if now.Sub(task.SamePathWaitStartedAt) >= timeout { + s.logger.Warnf("Timed out waiting for startup Hugging Face artifact parent recovery for %s before processing %s", parentKey, getModelInfoForLogging(task)) + return false + } + + delay := s.samePathWaitDelay + if delay <= 0 { + delay = defaultSamePathWaitDelay + } + s.logger.Infof("Startup Hugging Face artifact parent recovery is pending for %s; requeueing %s after %s", parentKey, getModelInfoForLogging(task), delay) + time.AfterFunc(delay, func() { + defer func() { + if r := recover(); r != nil { + s.logger.Warnf("Cannot requeue startup Hugging Face artifact parent recovery task for %s because gopher channel is closed: %v", getModelInfoForLogging(task), r) + } + }() + s.gopherChan <- task + }) + return true +} + +func (s *Gopher) claimStartupHuggingFaceArtifactParentValidation(parentKey string) bool { + if strings.TrimSpace(parentKey) == "" { + return false + } + s.startupHuggingFaceParentValidationMutex.Lock() + defer s.startupHuggingFaceParentValidationMutex.Unlock() + return s.claimStartupHuggingFaceArtifactParentValidationLocked(parentKey) +} + +func (s *Gopher) claimStartupHuggingFaceArtifactParentValidationLocked(parentKey string) bool { + if len(s.startupHuggingFaceParentValidationKeys) > 0 { + if _, ok := s.startupHuggingFaceParentValidationKeys[parentKey]; ok { + delete(s.startupHuggingFaceParentValidationKeys, parentKey) + return true + } + } + if !s.startupHuggingFaceParentSnapshotMissing { + return false + } + if s.startupHuggingFaceParentValidationTried == nil { + s.startupHuggingFaceParentValidationTried = map[string]struct{}{} + } + if _, ok := s.startupHuggingFaceParentValidationTried[parentKey]; ok { + return false + } + s.startupHuggingFaceParentValidationTried[parentKey] = struct{}{} + return true +} + +func (s *Gopher) restoreStartupHuggingFaceArtifactParentValidationLocked(parentKey string) { + if strings.TrimSpace(parentKey) == "" { + return + } + if s.startupHuggingFaceParentValidationKeys == nil { + s.startupHuggingFaceParentValidationKeys = map[string]struct{}{} + } + s.startupHuggingFaceParentValidationKeys[parentKey] = struct{}{} +} + +func (s *Gopher) validateStartupHuggingFaceArtifactParentIfNeeded(ctx context.Context, task *GopherTask, parentKey string, identity ArtifactIdentity, downloadObjectStorageModel func(string) error) (bool, error) { + if strings.TrimSpace(parentKey) == "" { + return false, nil + } + var validationPath string + var acquired bool + s.startupHuggingFaceParentValidationMutex.Lock() + if s.claimStartupHuggingFaceArtifactParentValidationLocked(parentKey) { + observedParentKey, parentPath, parentStatus, ok, lookupErr := s.getHuggingFaceArtifactParentWithError(ctx, identity) + if lookupErr != nil { + s.restoreStartupHuggingFaceArtifactParentValidationLocked(parentKey) + s.startupHuggingFaceParentValidationMutex.Unlock() + return false, nil + } + if ok && observedParentKey == parentKey && parentStatus == ModelStatusReady { + var err error + validationPath, acquired, err = s.acquireHuggingFaceArtifactParentRebuild(ctx, task, parentKey, parentPath, identity) + if err != nil { + s.restoreStartupHuggingFaceArtifactParentValidationLocked(parentKey) + s.startupHuggingFaceParentValidationMutex.Unlock() + return false, err + } + } + } + s.startupHuggingFaceParentValidationMutex.Unlock() + if strings.TrimSpace(validationPath) == "" { + return false, nil + } + if !acquired { + return true, nil + } + + s.logger.Infof("Validating startup Hugging Face artifact parent %s at %s", parentKey, validationPath) + if err := downloadObjectStorageModel(validationPath); err != nil { + if failErr := s.configMapReconciler.markHuggingFaceArtifactParentFailed(ctx, parentKey, identity, validationPath); failErr != nil { + s.logger.Warnf("failed to mark Hugging Face artifact parent %s at %s as Failed after startup validation error: %v", parentKey, validationPath, failErr) + } + return false, err + } + if markerErr := writeHuggingFaceArtifactReadyMarker(validationPath); markerErr != nil { + s.logger.Warnf("failed to write Hugging Face artifact ready marker for startup-validated parent %s at %s: %v", parentKey, validationPath, markerErr) + } + if err := s.markHuggingFaceArtifactParentReady(ctx, parentKey, validationPath, identity); err != nil { + s.logger.Errorf("validated startup Hugging Face artifact parent %s at %s but failed to mark it Ready: %v", parentKey, validationPath, err) + return false, err + } + s.logger.Infof("Validated startup Hugging Face artifact parent %s at %s", parentKey, validationPath) + return false, nil +} diff --git a/pkg/modelagent/gopher_test.go b/pkg/modelagent/gopher_test.go index 4fca0cc4e..e13eb0e48 100644 --- a/pkg/modelagent/gopher_test.go +++ b/pkg/modelagent/gopher_test.go @@ -7,13 +7,16 @@ import ( "fmt" "os" "path/filepath" + "sync" "testing" "time" + "github.com/prometheus/client_golang/prometheus" "go.uber.org/zap/zapcore" "go.uber.org/zap/zaptest" "go.uber.org/zap/zaptest/observer" + "github.com/oracle/oci-go-sdk/v65/objectstorage" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "go.uber.org/zap" @@ -21,12 +24,16 @@ import ( apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/runtime/schema" k8sfake "k8s.io/client-go/kubernetes/fake" + ktesting "k8s.io/client-go/testing" "sigs.k8s.io/ome/pkg/apis/ome/v1beta1" + omefake "sigs.k8s.io/ome/pkg/client/clientset/versioned/fake" omev1beta1lister "sigs.k8s.io/ome/pkg/client/listers/ome/v1beta1" "sigs.k8s.io/ome/pkg/constants" + "sigs.k8s.io/ome/pkg/modelparser" "sigs.k8s.io/ome/pkg/utils/storage" ) @@ -667,77 +674,1731 @@ func TestIsReservingModelArtifact_NilTaskReturnsFalse(t *testing.T) { assert.False(t, s.isReservingModelArtifact(nil), "nil task should not reserve artifact") } +func TestFilterInternalArtifactObjectSummariesSkipsReplicationControlObjects(t *testing.T) { + configName := "models/config.json" + weightName := "models/model.safetensors" + markerName := "models/" + artifactCompleteMarkerFileName + rootMarkerName := artifactCompleteMarkerFileName + lockName := "models/" + artifactUploadLockFileName + rootLockName := artifactUploadLockFileName + size := int64(1) + + objects := []objectstorage.ObjectSummary{ + {Name: &configName, Size: &size}, + {Name: &markerName, Size: &size}, + {Name: &lockName, Size: &size}, + {Name: &weightName, Size: &size}, + {Name: &rootMarkerName, Size: &size}, + {Name: &rootLockName, Size: &size}, + } + + filtered := filterInternalArtifactObjectSummaries(objects) + + require.Len(t, filtered, 2) + assert.Equal(t, configName, *filtered[0].Name) + assert.Equal(t, weightName, *filtered[1].Name) +} + func makeConfigMap(nodeName string, data map[string]string) *corev1.ConfigMap { return &corev1.ConfigMap{ ObjectMeta: metav1.ObjectMeta{ - Name: nodeName, - Namespace: "ome", + Name: nodeName, + Namespace: "ome", + }, + Data: data, + } +} + +func newGopherWithConfigMap(cm *corev1.ConfigMap) *Gopher { + client := k8sfake.NewSimpleClientset(cm) + logger := zap.NewNop().Sugar() + cmr := NewConfigMapReconciler(cm.Name, cm.Namespace, client, logger) + return &Gopher{ + configMapReconciler: cmr, + logger: logger, + } +} + +func newGopherForProcessTask(cm *corev1.ConfigMap, nodeLabels ...map[string]string) *Gopher { + labels := map[string]string{} + if len(nodeLabels) > 0 { + labels = nodeLabels[0] + } + node := &corev1.Node{ + ObjectMeta: metav1.ObjectMeta{ + Name: cm.Name, + Labels: labels, + }, + } + client := k8sfake.NewSimpleClientset(cm, node) + logger := zap.NewNop().Sugar() + cmr := NewConfigMapReconciler(cm.Name, cm.Namespace, client, logger) + return &Gopher{ + configMapReconciler: cmr, + nodeLabelReconciler: NewNodeLabelReconciler(cm.Name, client, 1, logger), + logger: logger, + activeDownloads: map[string]activeDownload{}, + } +} + +func newGopherForArtifactReuseProcessTask(t *testing.T, cm *corev1.ConfigMap, modelRootDir string, nodeLabels ...map[string]string) *Gopher { + t.Helper() + g := newGopherForProcessTask(cm, nodeLabels...) + g.modelRootDir = modelRootDir + g.downloadRetry = 1 + g.metrics = NewMetrics(prometheus.NewRegistry()) + g.baseModelLister = &mockBaseModelLister{} + g.clusterBaseModelLister = &mockClusterBaseModelLister{} + g.modelConfigParser = modelparser.NewModelConfigParser(omefake.NewSimpleClientset(), g.logger) + return g +} + +func writeMinimalModelConfig(t *testing.T, modelPath string) { + t.Helper() + require.NoError(t, os.MkdirAll(modelPath, 0755)) + require.NoError(t, os.WriteFile(filepath.Join(modelPath, "config.json"), []byte(`{ + "architectures": ["LlamaForCausalLM"], + "hidden_size": 16, + "intermediate_size": 64, + "max_position_embeddings": 128, + "model_type": "llama", + "num_attention_heads": 2, + "num_hidden_layers": 2, + "torch_dtype": "float16", + "transformers_version": "4.44.0", + "vocab_size": 128 +}`), 0644)) +} + +func entryJSON(sha, parentName string, parentPath string) string { + entry := struct { + Config struct { + Artifact Artifact `json:"artifact"` + } `json:"config"` + }{} + entry.Config.Artifact = Artifact{ + Sha: sha, + ParentPath: map[string]string{parentName: parentPath}, + ChildrenPaths: []string{}, + } + b, err := json.Marshal(entry) + if err != nil { + return "" + } + return string(b) +} + +func entryJSONWithOrigin(status ModelStatus, modelID string, sha string, parentName string, parentPath string, childrenPaths []string) string { + entry := ModelEntry{ + Status: status, + Config: &ModelConfig{ + Artifact: Artifact{ + Sha: sha, + Origin: &ArtifactOrigin{ + Type: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + }, + ParentPath: map[string]string{parentName: parentPath}, + ChildrenPaths: childrenPaths, + }, + }, + } + b, err := json.Marshal(entry) + if err != nil { + return "" + } + return string(b) +} + +func modelEntryJSON(status ModelStatus) string { + entry := ModelEntry{Status: status} + b, err := json.Marshal(entry) + if err != nil { + return "" + } + return string(b) +} + +func dp(v v1beta1.DownloadPolicy) *v1beta1.DownloadPolicy { + return &v +} + +func TestHuggingFaceArtifactIdentityFromAnnotations(t *testing.T) { + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + + identity, ok := huggingFaceArtifactIdentityFromAnnotations(map[string]string{ + HuggingFaceModelIDAnnotationKey: "Qwen/Qwen3-4B-Instruct-2507", + HuggingFaceSHAAnnotationKey: sha, + }) + + require.True(t, ok) + assert.Equal(t, ArtifactOriginTypeHuggingFace, identity.OriginType) + assert.Equal(t, "Qwen/Qwen3-4B-Instruct-2507", identity.HFModelID) + assert.Equal(t, sha, identity.HFCommitSHA) + + _, ok = huggingFaceArtifactIdentityFromAnnotations(map[string]string{ + HuggingFaceModelIDAnnotationKey: "Qwen/Qwen3-4B-Instruct-2507", + HuggingFaceSHAAnnotationKey: "main", + }) + assert.False(t, ok) + + _, ok = huggingFaceArtifactIdentityFromAnnotations(nil) + assert.False(t, ok) +} + +func TestHuggingFaceArtifactIdentityRejectsPathUnsafeModelIDs(t *testing.T) { + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + unsafeModelIDs := []string{ + "../outside", + "meta-llama/../../outside", + "/absolute", + "namespace//repo", + "namespace/repo\\evil", + "other..repo", + } + + for _, modelID := range unsafeModelIDs { + t.Run(modelID, func(t *testing.T) { + _, ok := huggingFaceArtifactIdentityFromAnnotations(map[string]string{ + HuggingFaceModelIDAnnotationKey: modelID, + HuggingFaceSHAAnnotationKey: sha, + }) + assert.False(t, ok) + }) + } +} + +func TestHuggingFaceArtifactKeyAndCanonicalPath(t *testing.T) { + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: "deepseek-ai/DeepSeek-V4-Pro", + HFCommitSHA: sha, + } + destPath := filepath.Join(t.TempDir(), "customer-model-store", "model-ocid") + + key := huggingFaceArtifactConfigMapKey(identity) + path := canonicalHuggingFaceArtifactPath(destPath, identity) + + assert.Contains(t, key, "artifact.huggingface.deepseek-ai.DeepSeek-V4-Pro.") + assert.Contains(t, key, "."+sha) + assert.NotContains(t, key, "/") + assert.Equal(t, filepath.Join(filepath.Dir(destPath), constants.ModelArtifactsDirectory, "deepseek-ai", "DeepSeek-V4-Pro", sha), path) +} + +func TestHuggingFaceArtifactParentPathForTaskPreservesClusterBaseModelLayout(t *testing.T) { + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: "deepseek-ai/DeepSeek-V4-Pro", + HFCommitSHA: sha, + } + root := t.TempDir() + destPath := filepath.Join(root, "customer-model-store", "model-ocid") + + baseModelPath := canonicalHuggingFaceArtifactPathForTask(&GopherTask{BaseModel: &v1beta1.BaseModel{}}, root, destPath, identity) + clusterBaseModelPath := canonicalHuggingFaceArtifactPathForTask(&GopherTask{ClusterBaseModel: &v1beta1.ClusterBaseModel{}}, root, destPath, identity) + + assert.Equal(t, filepath.Join(filepath.Dir(destPath), constants.ModelArtifactsDirectory, "deepseek-ai", "DeepSeek-V4-Pro", sha), baseModelPath) + assert.Equal(t, filepath.Join(root, "deepseek-ai", "DeepSeek-V4-Pro", sha), clusterBaseModelPath) +} + +func TestHuggingFaceArtifactConfigMapKeyAvoidsModelIDSanitizationCollision(t *testing.T) { + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + first := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: "org/foo.bar", + HFCommitSHA: sha, + } + second := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: "org.foo/bar", + HFCommitSHA: sha, + } + + assert.NotEqual(t, huggingFaceArtifactConfigMapKey(first), huggingFaceArtifactConfigMapKey(second)) +} + +func TestShouldUseHuggingFaceOriginObjectStorageReuse(t *testing.T) { + policy := v1beta1.ReuseIfExists + alwaysDownload := v1beta1.AlwaysDownload + + assert.True(t, shouldUseHuggingFaceOriginObjectStorageReuse(&GopherTask{TaskType: Download}, v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{DownloadPolicy: &policy}, + })) + assert.False(t, shouldUseHuggingFaceOriginObjectStorageReuse(&GopherTask{TaskType: DownloadOverride}, v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{DownloadPolicy: &policy}, + })) + assert.False(t, shouldUseHuggingFaceOriginObjectStorageReuse(&GopherTask{TaskType: Download}, v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{DownloadPolicy: &alwaysDownload}, + })) + assert.False(t, shouldUseHuggingFaceOriginObjectStorageReuse(&GopherTask{TaskType: Download}, v1beta1.BaseModelSpec{})) +} + +func TestShouldRepairHuggingFaceOriginObjectStorageParent(t *testing.T) { + policy := v1beta1.ReuseIfExists + alwaysDownload := v1beta1.AlwaysDownload + + assert.True(t, shouldRepairHuggingFaceOriginObjectStorageParent(&GopherTask{TaskType: DownloadOverride}, v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{DownloadPolicy: &policy}, + })) + assert.False(t, shouldRepairHuggingFaceOriginObjectStorageParent(&GopherTask{TaskType: Download}, v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{DownloadPolicy: &policy}, + })) + assert.False(t, shouldRepairHuggingFaceOriginObjectStorageParent(&GopherTask{TaskType: DownloadOverride}, v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{DownloadPolicy: &alwaysDownload}, + })) + assert.False(t, shouldRepairHuggingFaceOriginObjectStorageParent(&GopherTask{TaskType: DownloadOverride}, v1beta1.BaseModelSpec{})) +} + +func TestFindReadyHuggingFaceOriginArtifactMatchesModelIDAndSHA(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + parentPath := filepath.Join(t.TempDir(), "parent") + currentKey := constants.GetModelConfigMapKey("default", "current", false) + oneChildKey := constants.GetModelConfigMapKey("default", "one-child", false) + twoChildrenKey := constants.GetModelConfigMapKey("default", "two-children", false) + updatingKey := constants.GetModelConfigMapKey("default", "updating", false) + wrongModelKey := constants.GetModelConfigMapKey("default", "wrong-model", false) + wrongSHAKey := constants.GetModelConfigMapKey("default", "wrong-sha", false) + cm := makeConfigMap("node-1", map[string]string{ + oneChildKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, oneChildKey, parentPath, []string{"/child-a"}), + twoChildrenKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, twoChildrenKey, parentPath, []string{"/child-a", "/child-b"}), + updatingKey: entryJSONWithOrigin(ModelStatusUpdating, modelID, sha, updatingKey, parentPath, []string{"/child-a", "/child-b", "/child-c"}), + wrongModelKey: entryJSONWithOrigin(ModelStatusReady, "Qwen/Other", sha, wrongModelKey, parentPath, []string{"/child-a", "/child-b", "/child-c", "/child-d"}), + wrongSHAKey: entryJSONWithOrigin(ModelStatusReady, modelID, "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", wrongSHAKey, parentPath, []string{"/child-a", "/child-b", "/child-c", "/child-d", "/child-e", "/child-f"}), + currentKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, currentKey, parentPath, []string{"/child-a", "/child-b", "/child-c", "/child-d", "/child-e"}), + }) + g := newGopherWithConfigMap(cm) + + matchedKey, matchedPath, err := g.configMapReconciler.FindMatchedModelFromConfigMapByArtifactIdentity(cm, ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + }, "default."+constants.LowerCaseBaseModel, currentKey) + + require.NoError(t, err) + assert.Equal(t, twoChildrenKey, matchedKey) + assert.Equal(t, parentPath, matchedPath) +} + +func TestReuseHuggingFaceOriginArtifactUsesArtifactParentEntry(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + tmpDir := t.TempDir() + childPath := filepath.Join(tmpDir, "models", "child") + parentPath := canonicalHuggingFaceArtifactPath(childPath, identity) + writeMinimalModelConfig(t, parentPath) + + artifactKey := huggingFaceArtifactConfigMapKey(identity) + model := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{ + Name: "child", + Namespace: "default", + Annotations: map[string]string{ + HuggingFaceModelIDAnnotationKey: modelID, + HuggingFaceSHAAnnotationKey: sha, + }, + }, + Spec: v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{ + StorageUri: stringPtr("oci://n/ns/b/bucket/o/customer-imported-basemodels/qwen/qwen3/" + sha), + Path: &childPath, + DownloadPolicy: dp(v1beta1.ReuseIfExists), + }, + }, + } + g := newGopherWithConfigMap(makeConfigMap("node-1", map[string]string{ + artifactKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, artifactKey, parentPath, []string{}), + })) + + artifact, reused, err := g.reuseHuggingFaceOriginArtifactIfPossible(context.Background(), &GopherTask{ + TaskType: Download, + BaseModel: model, + }, model.Spec, constants.BaseModel, model.Namespace, model.Name, childPath, identity) + + require.NoError(t, err) + require.True(t, reused) + require.NotNil(t, artifact) + assert.Equal(t, parentPath, artifact.ParentPath[artifactKey]) + + resolvedChild, err := filepath.EvalSymlinks(childPath) + require.NoError(t, err) + resolvedParent, err := filepath.EvalSymlinks(parentPath) + require.NoError(t, err) + assert.Equal(t, resolvedParent, resolvedChild) + + exists, dataEntry, err := g.configMapReconciler.getDataEntryBasedOnModelKey(context.Background(), artifactKey) + require.NoError(t, err) + require.True(t, exists) + _, childrenPaths, err := g.configMapReconciler.getParentPathAndChildrenPaths(artifactKey, dataEntry) + require.NoError(t, err) + assert.Contains(t, childrenPaths, childPath) +} + +func TestReuseHuggingFaceOriginArtifactIgnoresModelEntryWithoutArtifactParent(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + tmpDir := t.TempDir() + oldModelPath := filepath.Join(tmpDir, "old-model") + childPath := filepath.Join(tmpDir, "models", "child") + writeMinimalModelConfig(t, oldModelPath) + + oldModelKey := constants.GetModelConfigMapKey("default", "old-model", false) + model := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{ + Name: "child", + Namespace: "default", + Annotations: map[string]string{ + HuggingFaceModelIDAnnotationKey: modelID, + HuggingFaceSHAAnnotationKey: sha, + }, + }, + Spec: v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{ + StorageUri: stringPtr("oci://n/ns/b/bucket/o/customer-imported-basemodels/qwen/qwen3/" + sha), + Path: &childPath, + DownloadPolicy: dp(v1beta1.ReuseIfExists), + }, + }, + } + g := newGopherWithConfigMap(makeConfigMap("node-1", map[string]string{ + oldModelKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, oldModelKey, oldModelPath, []string{}), + })) + + artifact, reused, err := g.reuseHuggingFaceOriginArtifactIfPossible(context.Background(), &GopherTask{ + TaskType: Download, + BaseModel: model, + }, model.Spec, constants.BaseModel, model.Namespace, model.Name, childPath, identity) + + require.NoError(t, err) + assert.False(t, reused) + assert.Nil(t, artifact) + _, err = os.Lstat(childPath) + assert.True(t, os.IsNotExist(err), "new model path should not symlink to an old model-local parent") +} + +func TestGetHuggingFaceArtifactParentReturnsUpdatingParent(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + tmpDir := t.TempDir() + parentPath := canonicalHuggingFaceArtifactPath(filepath.Join(tmpDir, "child"), identity) + parentKey := huggingFaceArtifactConfigMapKey(identity) + g := newGopherWithConfigMap(makeConfigMap("node-1", map[string]string{ + parentKey: entryJSONWithOrigin(ModelStatusUpdating, modelID, sha, parentKey, parentPath, []string{}), + })) + + gotKey, gotPath, gotStatus, ok := g.getHuggingFaceArtifactParent(context.Background(), identity) + + require.True(t, ok) + assert.Equal(t, parentKey, gotKey) + assert.Equal(t, parentPath, gotPath) + assert.Equal(t, ModelStatusUpdating, gotStatus) +} + +func TestReserveHuggingFaceArtifactParentEntryCreatesUpdatingParent(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + tmpDir := t.TempDir() + parentPath := canonicalHuggingFaceArtifactPath(filepath.Join(tmpDir, "child"), identity) + parentKey := huggingFaceArtifactConfigMapKey(identity) + g := newGopherWithConfigMap(makeConfigMap("node-1", map[string]string{})) + + gotPath, gotStatus, reserved, err := g.configMapReconciler.reserveHuggingFaceArtifactParentEntry(context.Background(), parentKey, identity, parentPath) + + require.NoError(t, err) + assert.True(t, reserved) + assert.Equal(t, parentPath, gotPath) + assert.Equal(t, ModelStatusUpdating, gotStatus) + + gotKey, gotParentPath, parentStatus, ok := g.getHuggingFaceArtifactParent(context.Background(), identity) + require.True(t, ok) + assert.Equal(t, parentKey, gotKey) + assert.Equal(t, parentPath, gotParentPath) + assert.Equal(t, ModelStatusUpdating, parentStatus) +} + +func TestMarkHuggingFaceArtifactParentUpdatingPreservesChildren(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + tmpDir := t.TempDir() + parentPath := canonicalHuggingFaceArtifactPath(filepath.Join(tmpDir, "child"), identity) + parentKey := huggingFaceArtifactConfigMapKey(identity) + childPath := filepath.Join(tmpDir, "models", "child") + g := newGopherWithConfigMap(makeConfigMap("node-1", map[string]string{ + parentKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, parentKey, parentPath, []string{childPath}), + })) + + gotPath, acquired, err := g.configMapReconciler.markHuggingFaceArtifactParentUpdating(context.Background(), parentKey, identity, parentPath) + + require.NoError(t, err) + assert.True(t, acquired) + assert.Equal(t, parentPath, gotPath) + + gotKey, gotParentPath, parentStatus, ok := g.getHuggingFaceArtifactParent(context.Background(), identity) + require.True(t, ok) + assert.Equal(t, parentKey, gotKey) + assert.Equal(t, parentPath, gotParentPath) + assert.Equal(t, ModelStatusUpdating, parentStatus) + + exists, dataEntry, err := g.configMapReconciler.getDataEntryBasedOnModelKey(context.Background(), parentKey) + require.NoError(t, err) + require.True(t, exists) + _, childrenPaths, err := g.configMapReconciler.getParentPathAndChildrenPaths(parentKey, dataEntry) + require.NoError(t, err) + assert.Contains(t, childrenPaths, childPath) +} + +func TestMarkHuggingFaceArtifactParentUpdatingDoesNotAcquireUpdatingParent(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + tmpDir := t.TempDir() + parentPath := canonicalHuggingFaceArtifactPath(filepath.Join(tmpDir, "child"), identity) + parentKey := huggingFaceArtifactConfigMapKey(identity) + childPath := filepath.Join(tmpDir, "models", "child") + g := newGopherWithConfigMap(makeConfigMap("node-1", map[string]string{ + parentKey: entryJSONWithOrigin(ModelStatusUpdating, modelID, sha, parentKey, parentPath, []string{childPath}), + })) + + gotPath, acquired, err := g.configMapReconciler.markHuggingFaceArtifactParentUpdating(context.Background(), parentKey, identity, parentPath) + + require.NoError(t, err) + assert.False(t, acquired) + assert.Equal(t, parentPath, gotPath) + + _, gotParentPath, parentStatus, ok := g.getHuggingFaceArtifactParent(context.Background(), identity) + require.True(t, ok) + assert.Equal(t, parentPath, gotParentPath) + assert.Equal(t, ModelStatusUpdating, parentStatus) +} + +func TestAcquireHuggingFaceArtifactParentRebuildMarksFailedParentUpdating(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + tmpDir := t.TempDir() + childPath := filepath.Join(tmpDir, "models", "child") + parentPath := canonicalHuggingFaceArtifactPath(childPath, identity) + parentKey := huggingFaceArtifactConfigMapKey(identity) + require.NoError(t, writeHuggingFaceArtifactReadyMarker(parentPath)) + model := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{ + Name: "child", + Namespace: "default", + Annotations: map[string]string{ + HuggingFaceModelIDAnnotationKey: modelID, + HuggingFaceSHAAnnotationKey: sha, + }, + }, + } + g := newGopherWithConfigMap(makeConfigMap("node-1", map[string]string{ + parentKey: entryJSONWithOrigin(ModelStatusFailed, modelID, sha, parentKey, parentPath, []string{}), + })) + + rebuildPath, acquired, err := g.acquireHuggingFaceArtifactParentRebuild(context.Background(), &GopherTask{ + TaskType: Download, + BaseModel: model, + }, parentKey, parentPath, identity) + + require.NoError(t, err) + require.True(t, acquired) + assert.Equal(t, parentPath, rebuildPath) + assert.False(t, hasHuggingFaceArtifactReadyMarker(parentPath), "rebuild owner should clear stale ready marker before writing parent") + + _, gotParentPath, parentStatus, ok := g.getHuggingFaceArtifactParent(context.Background(), identity) + require.True(t, ok) + assert.Equal(t, parentPath, gotParentPath) + assert.Equal(t, ModelStatusUpdating, parentStatus) +} + +func TestRecoverStartupHuggingFaceArtifactParentsMarksUpdatingWithoutReadyMarkerFailed(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + tmpDir := t.TempDir() + parentPath := canonicalHuggingFaceArtifactPath(filepath.Join(tmpDir, "child"), identity) + parentKey := huggingFaceArtifactConfigMapKey(identity) + g := newGopherWithConfigMap(makeConfigMap("node-1", map[string]string{ + parentKey: entryJSONWithOrigin(ModelStatusUpdating, modelID, sha, parentKey, parentPath, []string{}), + })) + + g.recoverStartupHuggingFaceArtifactParents(context.Background()) + + _, gotParentPath, parentStatus, ok := g.getHuggingFaceArtifactParent(context.Background(), identity) + require.True(t, ok) + assert.Equal(t, parentPath, gotParentPath) + assert.Equal(t, ModelStatusFailed, parentStatus) +} + +func TestRecoverStartupHuggingFaceArtifactParentsMarksUpdatingWithReadyMarkerReady(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + tmpDir := t.TempDir() + parentPath := canonicalHuggingFaceArtifactPath(filepath.Join(tmpDir, "child"), identity) + parentKey := huggingFaceArtifactConfigMapKey(identity) + require.NoError(t, writeHuggingFaceArtifactReadyMarker(parentPath)) + g := newGopherWithConfigMap(makeConfigMap("node-1", map[string]string{ + parentKey: entryJSONWithOrigin(ModelStatusUpdating, modelID, sha, parentKey, parentPath, []string{}), + })) + + g.recoverStartupHuggingFaceArtifactParents(context.Background()) + + _, gotParentPath, parentStatus, ok := g.getHuggingFaceArtifactParent(context.Background(), identity) + require.True(t, ok) + assert.Equal(t, parentPath, gotParentPath) + assert.Equal(t, ModelStatusReady, parentStatus) +} + +func TestEnsureStartupHuggingFaceArtifactParentsRecoveredDoesNotWaitWhenConfigMapMissing(t *testing.T) { + client := k8sfake.NewSimpleClientset() + logger := zap.NewNop().Sugar() + g := &Gopher{ + configMapReconciler: NewConfigMapReconciler("node-1", "ome", client, logger), + logger: logger, + gopherChan: make(chan *GopherTask, 1), + samePathWaitDelay: time.Millisecond, + samePathWaitTimeout: time.Hour, + } + + recovered := g.recoverStartupHuggingFaceArtifactParents(context.Background()) + require.True(t, recovered) + require.False(t, g.isStartupHuggingFaceArtifactParentRecoveryPending()) + + stop, err := g.ensureStartupHuggingFaceArtifactParentsRecovered(context.Background(), &GopherTask{TaskType: Download}, "artifact.huggingface.test") + + require.NoError(t, err) + assert.False(t, stop) + assert.False(t, g.isStartupHuggingFaceArtifactParentRecoveryPending()) + select { + case <-g.gopherChan: + t.Fatal("missing startup ConfigMap should not requeue or block first download") + default: + } +} + +func TestStartupConfigMapContextHasDeadline(t *testing.T) { + before := time.Now() + ctx, cancel := startupConfigMapContext() + defer cancel() + + deadline, ok := ctx.Deadline() + + require.True(t, ok) + assert.True(t, deadline.After(before)) + assert.LessOrEqual(t, time.Until(deadline), defaultStartupConfigMapSnapshotTimeout) +} + +func TestRunUsesSeparateStartupContextsForRecoveryAndSnapshot(t *testing.T) { + originalStartupConfigMapContext := startupConfigMapContext + defer func() { + startupConfigMapContext = originalStartupConfigMapContext + }() + startupContextCalls := 0 + startupConfigMapContext = func() (context.Context, context.CancelFunc) { + startupContextCalls++ + return context.WithCancel(context.Background()) + } + g := newGopherForProcessTask(makeConfigMap("node-1", map[string]string{})) + g.gopherChan = make(chan *GopherTask) + + stopCh := make(chan struct{}) + done := make(chan struct{}) + go func() { + g.Run(stopCh, 1, 1) + close(done) + }() + close(stopCh) + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("gopher Run did not stop") + } + assert.Equal(t, 2, startupContextCalls) +} + +func TestValidateStartupHuggingFaceArtifactParentRunsOncePerRestart(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + tmpDir := t.TempDir() + childPath := filepath.Join(tmpDir, "models", "child") + parentPath := canonicalHuggingFaceArtifactPath(childPath, identity) + parentKey := huggingFaceArtifactConfigMapKey(identity) + require.NoError(t, writeHuggingFaceArtifactReadyMarker(parentPath)) + g := newGopherWithConfigMap(makeConfigMap("node-1", map[string]string{ + parentKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, parentKey, parentPath, []string{}), + })) + g.startupHuggingFaceParentValidationKeys = map[string]struct{}{parentKey: {}} + downloadCalls := 0 + + stop, err := g.validateStartupHuggingFaceArtifactParentIfNeeded(context.Background(), &GopherTask{ + TaskType: Download, + BaseModel: &v1beta1.BaseModel{}, + }, parentKey, identity, func(downloadPath string) error { + downloadCalls++ + assert.Equal(t, parentPath, downloadPath) + assert.False(t, hasHuggingFaceArtifactReadyMarker(parentPath), "startup validation owner should clear ready marker before validating parent files") + return nil + }) + + require.NoError(t, err) + assert.False(t, stop) + assert.Equal(t, 1, downloadCalls) + assert.True(t, hasHuggingFaceArtifactReadyMarker(parentPath)) + _, gotParentPath, parentStatus, ok := g.getHuggingFaceArtifactParent(context.Background(), identity) + require.True(t, ok) + assert.Equal(t, parentPath, gotParentPath) + assert.Equal(t, ModelStatusReady, parentStatus) + + stop, err = g.validateStartupHuggingFaceArtifactParentIfNeeded(context.Background(), &GopherTask{ + TaskType: Download, + BaseModel: &v1beta1.BaseModel{}, + }, parentKey, identity, func(downloadPath string) error { + downloadCalls++ + return nil + }) + + require.NoError(t, err) + assert.False(t, stop) + assert.Equal(t, 1, downloadCalls) +} + +func TestValidateStartupHuggingFaceArtifactParentRetainsValidationAfterAcquireError(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + tmpDir := t.TempDir() + childPath := filepath.Join(tmpDir, "models", "child") + parentPath := canonicalHuggingFaceArtifactPath(childPath, identity) + parentKey := huggingFaceArtifactConfigMapKey(identity) + require.NoError(t, writeHuggingFaceArtifactReadyMarker(parentPath)) + g := newGopherWithConfigMap(makeConfigMap("node-1", map[string]string{ + parentKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, parentKey, parentPath, []string{}), + })) + g.startupHuggingFaceParentValidationKeys = map[string]struct{}{parentKey: {}} + fakeClient, ok := g.configMapReconciler.kubeClient.(*k8sfake.Clientset) + require.True(t, ok) + updateErr := fmt.Errorf("transient configmap update failure") + failNextUpdate := true + fakeClient.PrependReactor("update", "configmaps", func(action ktesting.Action) (bool, runtime.Object, error) { + if failNextUpdate { + failNextUpdate = false + return true, nil, updateErr + } + return false, nil, nil + }) + downloadCalls := 0 + + stop, err := g.validateStartupHuggingFaceArtifactParentIfNeeded(context.Background(), &GopherTask{ + TaskType: Download, + BaseModel: &v1beta1.BaseModel{}, + }, parentKey, identity, func(downloadPath string) error { + downloadCalls++ + return nil + }) + + require.ErrorIs(t, err, updateErr) + assert.False(t, stop) + assert.Equal(t, 0, downloadCalls) + + stop, err = g.validateStartupHuggingFaceArtifactParentIfNeeded(context.Background(), &GopherTask{ + TaskType: Download, + BaseModel: &v1beta1.BaseModel{}, + }, parentKey, identity, func(downloadPath string) error { + downloadCalls++ + assert.Equal(t, parentPath, downloadPath) + assert.False(t, hasHuggingFaceArtifactReadyMarker(parentPath), "startup validation owner should clear ready marker before validating parent files") + return nil + }) + + require.NoError(t, err) + assert.False(t, stop) + assert.Equal(t, 1, downloadCalls) + assert.True(t, hasHuggingFaceArtifactReadyMarker(parentPath)) + _, gotParentPath, parentStatus, ok := g.getHuggingFaceArtifactParent(context.Background(), identity) + require.True(t, ok) + assert.Equal(t, parentPath, gotParentPath) + assert.Equal(t, ModelStatusReady, parentStatus) +} + +func TestValidateStartupHuggingFaceArtifactParentValidatesWhenStartupSnapshotUnavailable(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + tmpDir := t.TempDir() + childPath := filepath.Join(tmpDir, "models", "child") + parentPath := canonicalHuggingFaceArtifactPath(childPath, identity) + parentKey := huggingFaceArtifactConfigMapKey(identity) + require.NoError(t, writeHuggingFaceArtifactReadyMarker(parentPath)) + g, client := newGopherWithEmptyClient("node-1", "ome", t) + + g.captureStartupReadyModels(context.Background()) + _, err := client.CoreV1().ConfigMaps("ome").Create(context.Background(), makeConfigMap("node-1", map[string]string{ + parentKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, parentKey, parentPath, []string{}), + }), metav1.CreateOptions{}) + require.NoError(t, err) + downloadCalls := 0 + + stop, err := g.validateStartupHuggingFaceArtifactParentIfNeeded(context.Background(), &GopherTask{ + TaskType: Download, + BaseModel: &v1beta1.BaseModel{}, + }, parentKey, identity, func(downloadPath string) error { + downloadCalls++ + assert.Equal(t, parentPath, downloadPath) + assert.False(t, hasHuggingFaceArtifactReadyMarker(parentPath), "startup validation owner should clear ready marker before validating parent files") + return nil + }) + + require.NoError(t, err) + assert.False(t, stop) + assert.Equal(t, 1, downloadCalls) + assert.True(t, hasHuggingFaceArtifactReadyMarker(parentPath)) + + stop, err = g.validateStartupHuggingFaceArtifactParentIfNeeded(context.Background(), &GopherTask{ + TaskType: Download, + BaseModel: &v1beta1.BaseModel{}, + }, parentKey, identity, func(downloadPath string) error { + downloadCalls++ + return nil + }) + + require.NoError(t, err) + assert.False(t, stop) + assert.Equal(t, 1, downloadCalls) +} + +func TestValidateStartupHuggingFaceArtifactParentRetainsValidationAfterParentLookupUnavailable(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + tmpDir := t.TempDir() + childPath := filepath.Join(tmpDir, "models", "child") + parentPath := canonicalHuggingFaceArtifactPath(childPath, identity) + parentKey := huggingFaceArtifactConfigMapKey(identity) + require.NoError(t, writeHuggingFaceArtifactReadyMarker(parentPath)) + g, client := newGopherWithEmptyClient("node-1", "ome", t) + g.startupHuggingFaceParentValidationKeys = map[string]struct{}{parentKey: {}} + downloadCalls := 0 + + stop, err := g.validateStartupHuggingFaceArtifactParentIfNeeded(context.Background(), &GopherTask{ + TaskType: Download, + BaseModel: &v1beta1.BaseModel{}, + }, parentKey, identity, func(downloadPath string) error { + downloadCalls++ + return nil + }) + + require.NoError(t, err) + assert.False(t, stop) + assert.Equal(t, 0, downloadCalls) + + _, err = client.CoreV1().ConfigMaps("ome").Create(context.Background(), makeConfigMap("node-1", map[string]string{ + parentKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, parentKey, parentPath, []string{}), + }), metav1.CreateOptions{}) + require.NoError(t, err) + + stop, err = g.validateStartupHuggingFaceArtifactParentIfNeeded(context.Background(), &GopherTask{ + TaskType: Download, + BaseModel: &v1beta1.BaseModel{}, + }, parentKey, identity, func(downloadPath string) error { + downloadCalls++ + assert.Equal(t, parentPath, downloadPath) + assert.False(t, hasHuggingFaceArtifactReadyMarker(parentPath), "startup validation owner should clear ready marker before validating parent files") + return nil + }) + + require.NoError(t, err) + assert.False(t, stop) + assert.Equal(t, 1, downloadCalls) + assert.True(t, hasHuggingFaceArtifactReadyMarker(parentPath)) +} + +func TestValidateStartupHuggingFaceArtifactParentBlocksSiblingUntilParentUpdating(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + tmpDir := t.TempDir() + childPath := filepath.Join(tmpDir, "models", "child") + parentPath := canonicalHuggingFaceArtifactPath(childPath, identity) + parentKey := huggingFaceArtifactConfigMapKey(identity) + require.NoError(t, writeHuggingFaceArtifactReadyMarker(parentPath)) + g := newGopherWithConfigMap(makeConfigMap("node-1", map[string]string{ + parentKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, parentKey, parentPath, []string{}), + })) + g.startupHuggingFaceParentValidationKeys = map[string]struct{}{parentKey: {}} + fakeClient, ok := g.configMapReconciler.kubeClient.(*k8sfake.Clientset) + require.True(t, ok) + + updateStarted := make(chan struct{}) + releaseUpdate := make(chan struct{}) + var blockOnce sync.Once + fakeClient.PrependReactor("update", "configmaps", func(action ktesting.Action) (bool, runtime.Object, error) { + blockOnce.Do(func() { + close(updateStarted) + <-releaseUpdate + }) + return false, nil, nil + }) + + firstDone := make(chan error, 1) + go func() { + _, err := g.validateStartupHuggingFaceArtifactParentIfNeeded(context.Background(), &GopherTask{ + TaskType: Download, + BaseModel: &v1beta1.BaseModel{}, + }, parentKey, identity, func(downloadPath string) error { + return nil + }) + firstDone <- err + }() + + select { + case <-updateStarted: + case <-time.After(time.Second): + t.Fatal("expected first validation to start marking the parent Updating") + } + + secondDone := make(chan error, 1) + go func() { + _, err := g.validateStartupHuggingFaceArtifactParentIfNeeded(context.Background(), &GopherTask{ + TaskType: Download, + BaseModel: &v1beta1.BaseModel{}, + }, parentKey, identity, func(downloadPath string) error { + return nil + }) + secondDone <- err + }() + + select { + case <-secondDone: + t.Fatal("sibling validation should wait until the startup parent is marked Updating") + case <-time.After(50 * time.Millisecond): + } + + close(releaseUpdate) + require.NoError(t, <-firstDone) + select { + case err := <-secondDone: + require.NoError(t, err) + case <-time.After(time.Second): + t.Fatal("expected sibling validation to finish after parent Updating transition completes") + } +} + +func TestRepairHuggingFaceOriginArtifactParentDownloadsParentAndLinksChild(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + tmpDir := t.TempDir() + childPath := filepath.Join(tmpDir, "models", "child") + parentPath := canonicalHuggingFaceArtifactPath(childPath, identity) + parentKey := huggingFaceArtifactConfigMapKey(identity) + siblingPath := filepath.Join(tmpDir, "models", "sibling") + writeMinimalModelConfig(t, parentPath) + require.NoError(t, writeHuggingFaceArtifactReadyMarker(parentPath)) + + model := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{ + Name: "child", + Namespace: "default", + Annotations: map[string]string{ + HuggingFaceModelIDAnnotationKey: modelID, + HuggingFaceSHAAnnotationKey: sha, + }, + }, + Spec: v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{ + StorageUri: stringPtr("oci://n/ns/b/bucket/o/customer-imported-basemodels/qwen/qwen3/" + sha), + Path: &childPath, + DownloadPolicy: dp(v1beta1.ReuseIfExists), + }, + }, + } + g := newGopherWithConfigMap(makeConfigMap("node-1", map[string]string{ + parentKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, parentKey, parentPath, []string{siblingPath}), + })) + downloadCalled := false + downloadParent := func(downloadPath string) error { + downloadCalled = true + assert.Equal(t, parentPath, downloadPath) + assert.False(t, hasHuggingFaceArtifactReadyMarker(parentPath), "repair should clear stale ready marker before writing parent") + writeMinimalModelConfig(t, downloadPath) + return nil + } + + artifact, repaired, err := g.repairHuggingFaceOriginArtifactParent(context.Background(), &GopherTask{ + TaskType: DownloadOverride, + BaseModel: model, + }, model.Name, childPath, parentKey, parentPath, identity, downloadParent) + + require.NoError(t, err) + require.True(t, repaired) + require.True(t, downloadCalled) + require.NotNil(t, artifact) + assert.Equal(t, parentPath, artifact.ParentPath[parentKey]) + + resolvedChild, err := filepath.EvalSymlinks(childPath) + require.NoError(t, err) + resolvedParent, err := filepath.EvalSymlinks(parentPath) + require.NoError(t, err) + assert.Equal(t, resolvedParent, resolvedChild) + assert.True(t, hasHuggingFaceArtifactReadyMarker(parentPath)) + + _, gotParentPath, parentStatus, ok := g.getHuggingFaceArtifactParent(context.Background(), identity) + require.True(t, ok) + assert.Equal(t, parentPath, gotParentPath) + assert.Equal(t, ModelStatusReady, parentStatus) + + exists, dataEntry, err := g.configMapReconciler.getDataEntryBasedOnModelKey(context.Background(), parentKey) + require.NoError(t, err) + require.True(t, exists) + _, childrenPaths, err := g.configMapReconciler.getParentPathAndChildrenPaths(parentKey, dataEntry) + require.NoError(t, err) + assert.Contains(t, childrenPaths, siblingPath) + assert.Contains(t, childrenPaths, childPath) +} + +func TestRepairHuggingFaceOriginArtifactParentMarksParentFailedOnDownloadError(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + tmpDir := t.TempDir() + childPath := filepath.Join(tmpDir, "models", "child") + parentPath := canonicalHuggingFaceArtifactPath(childPath, identity) + parentKey := huggingFaceArtifactConfigMapKey(identity) + writeMinimalModelConfig(t, parentPath) + + model := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{ + Name: "child", + Namespace: "default", + Annotations: map[string]string{ + HuggingFaceModelIDAnnotationKey: modelID, + HuggingFaceSHAAnnotationKey: sha, + }, + }, + Spec: v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{ + StorageUri: stringPtr("oci://n/ns/b/bucket/o/customer-imported-basemodels/qwen/qwen3/" + sha), + Path: &childPath, + DownloadPolicy: dp(v1beta1.ReuseIfExists), + }, }, - Data: data, } + g := newGopherWithConfigMap(makeConfigMap("node-1", map[string]string{ + parentKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, parentKey, parentPath, []string{}), + })) + downloadErr := fmt.Errorf("download failed") + + artifact, repaired, err := g.repairHuggingFaceOriginArtifactParent(context.Background(), &GopherTask{ + TaskType: DownloadOverride, + BaseModel: model, + }, model.Name, childPath, parentKey, parentPath, identity, func(downloadPath string) error { + return downloadErr + }) + + assert.Nil(t, artifact) + assert.False(t, repaired) + assert.ErrorIs(t, err, downloadErr) + + _, gotParentPath, parentStatus, ok := g.getHuggingFaceArtifactParent(context.Background(), identity) + require.True(t, ok) + assert.Equal(t, parentPath, gotParentPath) + assert.Equal(t, ModelStatusFailed, parentStatus) } -func newGopherWithConfigMap(cm *corev1.ConfigMap) *Gopher { - client := k8sfake.NewSimpleClientset(cm) - logger := zap.NewNop().Sugar() - cmr := NewConfigMapReconciler(cm.Name, cm.Namespace, client, logger) - return &Gopher{ - configMapReconciler: cmr, - logger: logger, +func TestRepairHuggingFaceOriginArtifactParentKeepsReadyMarkerWhenRepairNotAcquired(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + tmpDir := t.TempDir() + childPath := filepath.Join(tmpDir, "models", "child") + parentPath := canonicalHuggingFaceArtifactPath(childPath, identity) + parentKey := huggingFaceArtifactConfigMapKey(identity) + writeMinimalModelConfig(t, parentPath) + require.NoError(t, writeHuggingFaceArtifactReadyMarker(parentPath)) + + model := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{ + Name: "child", + Namespace: "default", + Annotations: map[string]string{ + HuggingFaceModelIDAnnotationKey: modelID, + HuggingFaceSHAAnnotationKey: sha, + }, + }, + Spec: v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{ + StorageUri: stringPtr("oci://n/ns/b/bucket/o/customer-imported-basemodels/qwen/qwen3/" + sha), + Path: &childPath, + DownloadPolicy: dp(v1beta1.ReuseIfExists), + }, + }, + } + readyParent := makeConfigMap("node-1", map[string]string{ + parentKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, parentKey, parentPath, []string{}), + }) + updatingParent := readyParent.DeepCopy() + updatingParent.Data[parentKey] = entryJSONWithOrigin(ModelStatusUpdating, modelID, sha, parentKey, parentPath, []string{}) + g := newGopherWithConfigMap(readyParent) + fakeClient, ok := g.configMapReconciler.kubeClient.(*k8sfake.Clientset) + require.True(t, ok) + getCount := 0 + fakeClient.PrependReactor("get", "configmaps", func(action ktesting.Action) (bool, runtime.Object, error) { + getCount++ + if getCount >= 3 { + return true, updatingParent.DeepCopy(), nil + } + return false, nil, nil + }) + downloadCalled := false + + artifact, repaired, err := g.repairHuggingFaceOriginArtifactParent(context.Background(), &GopherTask{ + TaskType: DownloadOverride, + BaseModel: model, + SamePathWaitStartedAt: time.Now().Add(-defaultSamePathWaitTimeout), + }, model.Name, childPath, parentKey, parentPath, identity, func(downloadPath string) error { + downloadCalled = true + return nil + }) + + assert.Nil(t, artifact) + assert.False(t, repaired) + assert.ErrorContains(t, err, "timed out waiting for Hugging Face artifact parent") + assert.False(t, downloadCalled) + assert.True(t, hasHuggingFaceArtifactReadyMarker(parentPath), "non-owner repair must not remove the ready marker") +} + +func TestProcessTaskRequeuesWhenHuggingFaceArtifactParentIsUpdating(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + tmpDir := t.TempDir() + childPath := filepath.Join(tmpDir, "child") + parentPath := canonicalHuggingFaceArtifactPath(childPath, identity) + parentKey := huggingFaceArtifactConfigMapKey(identity) + model := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{ + Name: "child", + Namespace: "default", + UID: "child-uid", + Annotations: map[string]string{ + HuggingFaceModelIDAnnotationKey: modelID, + HuggingFaceSHAAnnotationKey: sha, + }, + }, + Spec: v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{ + StorageUri: stringPtr("oci://n/ns/b/bucket/o/customer-imported-basemodels/qwen/qwen3/" + sha), + Path: &childPath, + DownloadPolicy: dp(v1beta1.ReuseIfExists), + }, + }, + } + g := newGopherForArtifactReuseProcessTask(t, makeConfigMap("node-1", map[string]string{ + parentKey: entryJSONWithOrigin(ModelStatusUpdating, modelID, sha, parentKey, parentPath, []string{}), + }), tmpDir) + g.baseModelLister = &mockBaseModelLister{models: []*v1beta1.BaseModel{model}} + g.gopherChan = make(chan *GopherTask, 1) + g.samePathWaitDelay = time.Millisecond + g.samePathWaitTimeout = time.Hour + + err := g.processTaskWithOptions(&GopherTask{ + TaskType: Download, + BaseModel: model, + }, true) + + require.NoError(t, err) + select { + case task := <-g.gopherChan: + assert.Equal(t, model.Name, task.BaseModel.Name) + assert.False(t, task.SamePathWaitStartedAt.IsZero()) + case <-time.After(100 * time.Millisecond): + t.Fatal("expected Updating Hugging Face artifact parent task to be requeued") } + _, err = os.Lstat(childPath) + assert.True(t, os.IsNotExist(err), "child path should not be created while parent artifact is Updating") } -func newGopherForProcessTask(cm *corev1.ConfigMap, nodeLabels ...map[string]string) *Gopher { - labels := map[string]string{} - if len(nodeLabels) > 0 { - labels = nodeLabels[0] +func TestProcessTaskWaitsOnUpdatingParentCreatedAfterMissingStartupConfigMap(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, } - node := &corev1.Node{ + tmpDir := t.TempDir() + childPath := filepath.Join(tmpDir, "child") + parentPath := canonicalHuggingFaceArtifactPath(childPath, identity) + parentKey := huggingFaceArtifactConfigMapKey(identity) + model := &v1beta1.BaseModel{ ObjectMeta: metav1.ObjectMeta{ - Name: cm.Name, - Labels: labels, + Name: "child", + Namespace: "default", + UID: "child-uid", + Annotations: map[string]string{ + HuggingFaceModelIDAnnotationKey: modelID, + HuggingFaceSHAAnnotationKey: sha, + }, + }, + Spec: v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{ + StorageUri: stringPtr("oci://n/ns/b/bucket/o/customer-imported-basemodels/qwen/qwen3/" + sha), + Path: &childPath, + DownloadPolicy: dp(v1beta1.ReuseIfExists), + }, }, } - client := k8sfake.NewSimpleClientset(cm, node) + + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node-1"}} + client := k8sfake.NewSimpleClientset(node) logger := zap.NewNop().Sugar() - cmr := NewConfigMapReconciler(cm.Name, cm.Namespace, client, logger) - return &Gopher{ - configMapReconciler: cmr, - nodeLabelReconciler: NewNodeLabelReconciler(cm.Name, client, 1, logger), - logger: logger, - activeDownloads: map[string]activeDownload{}, + g := newGopherForArtifactReuseProcessTask(t, makeConfigMap("node-1", map[string]string{}), tmpDir) + g.configMapReconciler = NewConfigMapReconciler("node-1", "ome", client, logger) + g.nodeLabelReconciler = NewNodeLabelReconciler("node-1", client, 1, logger) + g.logger = logger + g.baseModelLister = &mockBaseModelLister{models: []*v1beta1.BaseModel{model}} + g.gopherChan = make(chan *GopherTask, 1) + g.samePathWaitDelay = time.Millisecond + g.samePathWaitTimeout = time.Hour + + g.recoverStartupHuggingFaceArtifactParents(context.Background()) + _, err := client.CoreV1().ConfigMaps("ome").Create(context.Background(), makeConfigMap("node-1", map[string]string{ + parentKey: entryJSONWithOrigin(ModelStatusUpdating, modelID, sha, parentKey, parentPath, []string{}), + }), metav1.CreateOptions{}) + require.NoError(t, err) + + err = g.processTaskWithOptions(&GopherTask{ + TaskType: Download, + BaseModel: model, + }, false) + + require.NoError(t, err) + _, _, parentStatus, ok := g.getHuggingFaceArtifactParent(context.Background(), identity) + require.True(t, ok) + assert.Equal(t, ModelStatusUpdating, parentStatus) + select { + case task := <-g.gopherChan: + assert.Equal(t, model.Name, task.BaseModel.Name) + case <-time.After(100 * time.Millisecond): + t.Fatal("Updating parent created after missing startup ConfigMap should be waited on, not marked Failed") } } -func entryJSON(sha, parentName string, parentPath string) string { - entry := struct { - Config struct { - Artifact Artifact `json:"artifact"` - } `json:"config"` - }{} - entry.Config.Artifact = Artifact{ - Sha: sha, - ParentPath: map[string]string{parentName: parentPath}, - ChildrenPaths: []string{}, +func TestProcessTaskRecoversUpdatingHuggingFaceArtifactParentWithReadyMarker(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, } - b, err := json.Marshal(entry) - if err != nil { - return "" + tmpDir := t.TempDir() + childPath := filepath.Join(tmpDir, "child") + parentPath := canonicalHuggingFaceArtifactPath(childPath, identity) + writeMinimalModelConfig(t, parentPath) + require.NoError(t, os.WriteFile(huggingFaceArtifactReadyMarkerPath(parentPath), []byte("ready\n"), 0644)) + + parentKey := huggingFaceArtifactConfigMapKey(identity) + model := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{ + Name: "child", + Namespace: "default", + UID: "child-uid", + Annotations: map[string]string{ + HuggingFaceModelIDAnnotationKey: modelID, + HuggingFaceSHAAnnotationKey: sha, + }, + }, + Spec: v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{ + StorageUri: stringPtr("oci://n/ns/b/bucket/o/customer-imported-basemodels/qwen/qwen3/" + sha), + Path: &childPath, + DownloadPolicy: dp(v1beta1.ReuseIfExists), + }, + }, } - return string(b) + labelKey, err := getModelLabelKey(&NodeLabelOp{ModelStateOnNode: Ready, BaseModel: model}) + require.NoError(t, err) + g := newGopherForArtifactReuseProcessTask(t, makeConfigMap("node-1", map[string]string{ + parentKey: entryJSONWithOrigin(ModelStatusUpdating, modelID, sha, parentKey, parentPath, []string{}), + }), tmpDir, map[string]string{labelKey: string(Ready)}) + g.baseModelLister = &mockBaseModelLister{models: []*v1beta1.BaseModel{model}} + + err = g.processTaskWithOptions(&GopherTask{ + TaskType: Download, + BaseModel: model, + }, true) + + require.NoError(t, err) + resolvedChild, err := filepath.EvalSymlinks(childPath) + require.NoError(t, err) + resolvedParent, err := filepath.EvalSymlinks(parentPath) + require.NoError(t, err) + assert.Equal(t, resolvedParent, resolvedChild) + + latest, err := g.configMapReconciler.kubeClient.CoreV1().ConfigMaps("ome").Get(context.Background(), "node-1", metav1.GetOptions{}) + require.NoError(t, err) + require.Contains(t, latest.Data, parentKey) + var parentEntry ModelEntry + require.NoError(t, json.Unmarshal([]byte(latest.Data[parentKey]), &parentEntry)) + assert.Equal(t, ModelStatusReady, parentEntry.Status) + assert.Contains(t, parentEntry.Config.Artifact.ChildrenPaths, childPath) +} + +func TestProcessTaskWithOptionsReusesOCIArtifactByHuggingFaceOrigin(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + tmpDir := t.TempDir() + childPath := filepath.Join(tmpDir, "child") + parentPath := canonicalHuggingFaceArtifactPath(childPath, identity) + writeMinimalModelConfig(t, parentPath) + + parentKey := huggingFaceArtifactConfigMapKey(identity) + model := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{ + Name: "child", + Namespace: "default", + UID: "child-uid", + Annotations: map[string]string{ + HuggingFaceModelIDAnnotationKey: modelID, + HuggingFaceSHAAnnotationKey: sha, + }, + }, + Spec: v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{ + StorageUri: stringPtr("oci://n/ns/b/bucket/o/customer-imported-basemodels/qwen/qwen3"), + Path: &childPath, + DownloadPolicy: dp(v1beta1.ReuseIfExists), + }, + }, + } + labelKey, err := getModelLabelKey(&NodeLabelOp{ModelStateOnNode: Ready, BaseModel: model}) + require.NoError(t, err) + g := newGopherForArtifactReuseProcessTask(t, makeConfigMap("node-1", map[string]string{ + parentKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, parentKey, parentPath, []string{}), + }), tmpDir, map[string]string{labelKey: string(Ready)}) + g.baseModelLister = &mockBaseModelLister{models: []*v1beta1.BaseModel{model}} + + err = g.processTaskWithOptions(&GopherTask{ + TaskType: Download, + BaseModel: model, + }, true) + + require.NoError(t, err) + resolvedChild, err := filepath.EvalSymlinks(childPath) + require.NoError(t, err) + resolvedParent, err := filepath.EvalSymlinks(parentPath) + require.NoError(t, err) + assert.Equal(t, resolvedParent, resolvedChild) + + latest, err := g.configMapReconciler.kubeClient.CoreV1().ConfigMaps("ome").Get(context.Background(), "node-1", metav1.GetOptions{}) + require.NoError(t, err) + assert.Contains(t, getChildrenPaths(latest.Data[parentKey]), childPath) + + currentKey := constants.GetModelConfigMapKey(model.Namespace, model.Name, false) + var currentEntry ModelEntry + require.NoError(t, json.Unmarshal([]byte(latest.Data[currentKey]), ¤tEntry)) + assert.Equal(t, ModelStatusReady, currentEntry.Status) + require.NotNil(t, currentEntry.Config) + require.NotNil(t, currentEntry.Config.Artifact.Origin) + assert.Equal(t, ArtifactOriginTypeHuggingFace, currentEntry.Config.Artifact.Origin.Type) + assert.Equal(t, modelID, currentEntry.Config.Artifact.Origin.HFModelID) + assert.Equal(t, sha, currentEntry.Config.Artifact.Origin.HFCommitSHA) + assert.Equal(t, parentPath, currentEntry.Config.Artifact.ParentPath[parentKey]) +} + +func TestReuseHuggingFaceOriginArtifactCreatesSymlinkAndUpdatesChildren(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + tmpDir := t.TempDir() + childPath := filepath.Join(tmpDir, "child") + parentPath := canonicalHuggingFaceArtifactPath(childPath, identity) + require.NoError(t, os.MkdirAll(parentPath, 0755)) + require.NoError(t, os.WriteFile(filepath.Join(parentPath, "config.json"), []byte("{}"), 0644)) + + parentKey := huggingFaceArtifactConfigMapKey(identity) + model := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{ + Name: "child", + Namespace: "default", + Annotations: map[string]string{ + HuggingFaceModelIDAnnotationKey: modelID, + HuggingFaceSHAAnnotationKey: sha, + }, + }, + Spec: v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{ + StorageUri: stringPtr("oci://n/ns/b/bucket/o/customer-imported-basemodels/qwen/qwen3"), + Path: &childPath, + DownloadPolicy: dp(v1beta1.ReuseIfExists), + }, + }, + } + g := newGopherWithConfigMap(makeConfigMap("node-1", map[string]string{ + parentKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, parentKey, parentPath, []string{}), + })) + + artifact, reused, err := g.reuseHuggingFaceOriginArtifactIfPossible(context.Background(), &GopherTask{ + TaskType: Download, + BaseModel: model, + }, model.Spec, constants.BaseModel, model.Namespace, model.Name, childPath, identity) + + require.NoError(t, err) + require.True(t, reused) + require.NotNil(t, artifact) + assert.Equal(t, sha, artifact.Sha) + require.NotNil(t, artifact.Origin) + assert.Equal(t, modelID, artifact.Origin.HFModelID) + assert.Equal(t, parentPath, artifact.ParentPath[parentKey]) + + resolvedChild, err := filepath.EvalSymlinks(childPath) + require.NoError(t, err) + resolvedParent, err := filepath.EvalSymlinks(parentPath) + require.NoError(t, err) + assert.Equal(t, resolvedParent, resolvedChild) + + exists, dataEntry, err := g.configMapReconciler.getDataEntryBasedOnModelKey(context.Background(), parentKey) + require.NoError(t, err) + require.True(t, exists) + _, childrenPaths, err := g.configMapReconciler.getParentPathAndChildrenPaths(parentKey, dataEntry) + require.NoError(t, err) + assert.Contains(t, childrenPaths, childPath) } -func modelEntryJSON(status ModelStatus) string { - entry := ModelEntry{Status: status} - b, err := json.Marshal(entry) - if err != nil { - return "" +func TestProcessTaskDeletesOCIOriginChildAndPreservesExistingParent(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + tmpDir := t.TempDir() + parentPath := filepath.Join(tmpDir, "parent") + childPath := filepath.Join(tmpDir, "child") + writeMinimalModelConfig(t, parentPath) + require.NoError(t, os.Symlink(parentPath, childPath)) + + parentKey := constants.GetModelConfigMapKey("default", "parent", false) + child := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{Name: "child", Namespace: "default", UID: "child-uid"}, + Spec: v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{ + StorageUri: stringPtr("oci://n/ns/b/bucket/o/customer-imported-basemodels/qwen/qwen3"), + Path: &childPath, + }, + }, } - return string(b) + childKey := constants.GetModelConfigMapKey(child.Namespace, child.Name, false) + g := newGopherForArtifactReuseProcessTask(t, makeConfigMap("node-1", map[string]string{ + parentKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, parentKey, parentPath, []string{childPath}), + childKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, parentKey, parentPath, []string{}), + }), tmpDir) + + err := g.processTask(&GopherTask{TaskType: Delete, BaseModel: child}) + + require.NoError(t, err) + _, err = os.Lstat(childPath) + assert.True(t, os.IsNotExist(err), "child symlink should be removed") + _, err = os.Stat(parentPath) + assert.NoError(t, err, "parent artifact should remain while the parent ConfigMap entry exists") + + latest, err := g.configMapReconciler.kubeClient.CoreV1().ConfigMaps("ome").Get(context.Background(), "node-1", metav1.GetOptions{}) + require.NoError(t, err) + assert.NotContains(t, latest.Data, childKey) + assert.NotContains(t, getChildrenPaths(latest.Data[parentKey]), childPath) +} + +func TestProcessTaskDeletesOCIOriginChildAndRemovesOrphanParent(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + tmpDir := t.TempDir() + parentPath := filepath.Join(tmpDir, "parent") + childPath := filepath.Join(tmpDir, "child") + writeMinimalModelConfig(t, parentPath) + require.NoError(t, os.Symlink(parentPath, childPath)) + + parentKey := constants.GetModelConfigMapKey("default", "deleted-parent", false) + child := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{Name: "child", Namespace: "default", UID: "child-uid"}, + Spec: v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{ + StorageUri: stringPtr("oci://n/ns/b/bucket/o/customer-imported-basemodels/qwen/qwen3"), + Path: &childPath, + }, + }, + } + childKey := constants.GetModelConfigMapKey(child.Namespace, child.Name, false) + g := newGopherForArtifactReuseProcessTask(t, makeConfigMap("node-1", map[string]string{ + childKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, parentKey, parentPath, []string{}), + }), tmpDir) + + err := g.processTask(&GopherTask{TaskType: Delete, BaseModel: child}) + + require.NoError(t, err) + _, err = os.Lstat(childPath) + assert.True(t, os.IsNotExist(err), "child symlink should be removed") + _, err = os.Stat(parentPath) + assert.True(t, os.IsNotExist(err), "orphan parent artifact should be removed after the last child is deleted") + + latest, err := g.configMapReconciler.kubeClient.CoreV1().ConfigMaps("ome").Get(context.Background(), "node-1", metav1.GetOptions{}) + require.NoError(t, err) + assert.NotContains(t, latest.Data, childKey) +} + +func TestProcessTaskDeletesLastOCIOriginChildAndRemovesArtifactParentEntry(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + tmpDir := t.TempDir() + childPath := filepath.Join(tmpDir, "models", "child") + parentPath := canonicalHuggingFaceArtifactPath(childPath, identity) + writeMinimalModelConfig(t, parentPath) + require.NoError(t, os.MkdirAll(filepath.Dir(childPath), 0755)) + require.NoError(t, os.Symlink(parentPath, childPath)) + + artifactKey := huggingFaceArtifactConfigMapKey(identity) + child := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{Name: "child", Namespace: "default", UID: "child-uid"}, + Spec: v1beta1.BaseModelSpec{ + Storage: &v1beta1.StorageSpec{ + StorageUri: stringPtr("oci://n/ns/b/bucket/o/customer-imported-basemodels/qwen/qwen3/" + sha), + Path: &childPath, + }, + }, + } + childKey := constants.GetModelConfigMapKey(child.Namespace, child.Name, false) + g := newGopherForArtifactReuseProcessTask(t, makeConfigMap("node-1", map[string]string{ + artifactKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, artifactKey, parentPath, []string{childPath}), + childKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, artifactKey, parentPath, []string{}), + }), tmpDir) + + err := g.processTask(&GopherTask{TaskType: Delete, BaseModel: child}) + + require.NoError(t, err) + _, err = os.Lstat(childPath) + assert.True(t, os.IsNotExist(err), "child symlink should be removed") + _, err = os.Stat(parentPath) + assert.True(t, os.IsNotExist(err), "canonical artifact should be removed after the last child is deleted") + + latest, err := g.configMapReconciler.kubeClient.CoreV1().ConfigMaps("ome").Get(context.Background(), "node-1", metav1.GetOptions{}) + require.NoError(t, err) + assert.NotContains(t, latest.Data, childKey) + assert.NotContains(t, latest.Data, artifactKey) } -func dp(v v1beta1.DownloadPolicy) *v1beta1.DownloadPolicy { - return &v +func TestConvertMetadataToModelConfigPreservesArtifactOrigin(t *testing.T) { + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + + config := ConvertMetadataToModelConfig(ModelMetadata{ + Artifact: Artifact{ + Sha: sha, + Origin: &ArtifactOrigin{ + Type: ArtifactOriginTypeHuggingFace, + HFModelID: "Qwen/Qwen3-4B-Instruct-2507", + HFCommitSHA: sha, + }, + ParentPath: map[string]string{"default.basemodel.parent": "/models/parent"}, + ChildrenPaths: []string{"/models/child"}, + }, + }) + + require.NotNil(t, config.Artifact.Origin) + assert.Equal(t, ArtifactOriginTypeHuggingFace, config.Artifact.Origin.Type) + assert.Equal(t, "Qwen/Qwen3-4B-Instruct-2507", config.Artifact.Origin.HFModelID) + assert.Equal(t, sha, config.Artifact.Origin.HFCommitSHA) +} + +func TestBuildSelfParentArtifactFromIdentityPreservesExistingChildren(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + modelPath := filepath.Join(t.TempDir(), "parent") + existingChildPath := filepath.Join(t.TempDir(), "child") + model := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{Name: "model", Namespace: "default"}, + } + currentKey := constants.GetModelConfigMapKey(model.Namespace, model.Name, false) + g := newGopherWithConfigMap(makeConfigMap("node-1", map[string]string{ + currentKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, currentKey, modelPath, []string{existingChildPath}), + })) + + artifact := g.buildSelfParentArtifactFromIdentity(context.Background(), &GopherTask{BaseModel: model}, ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + }, modelPath) + + require.NotNil(t, artifact) + assert.Equal(t, sha, artifact.Sha) + assert.Equal(t, modelPath, artifact.ParentPath[currentKey]) + assert.Equal(t, []string{existingChildPath}, artifact.ChildrenPaths) +} + +func TestLinkHuggingFaceOriginArtifactDoesNotParseIncompleteChildEntry(t *testing.T) { + modelID := "Qwen/Qwen3-8B" + sha := "b968826d9c46dd6066d109eabc6255188de91218" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + parentKey := huggingFaceArtifactConfigMapKey(identity) + parentPath := filepath.Join(t.TempDir(), "_artifacts", "Qwen", "Qwen3-8B", sha) + childPath := filepath.Join(t.TempDir(), "model-ocid") + require.NoError(t, os.MkdirAll(parentPath, 0755)) + writeMinimalModelConfig(t, parentPath) + + model := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{Name: "model-ocid", Namespace: "default"}, + } + currentKey := constants.GetModelConfigMapKey(model.Namespace, model.Name, false) + + core, observedLogs := observer.New(zapcore.InfoLevel) + logger := zap.New(core).Sugar() + cm := makeConfigMap("node-1", map[string]string{ + parentKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, parentKey, parentPath, []string{}), + currentKey: modelEntryJSON(ModelStatusUpdating), + }) + client := k8sfake.NewSimpleClientset(cm) + g := &Gopher{ + configMapReconciler: NewConfigMapReconciler(cm.Name, cm.Namespace, client, logger), + logger: logger, + } + + artifact, err := g.linkHuggingFaceOriginArtifact(context.Background(), &GopherTask{BaseModel: model}, model.Name, childPath, parentKey, parentPath, identity) + + require.NoError(t, err) + require.NotNil(t, artifact) + assert.Equal(t, parentPath, artifact.ParentPath[parentKey]) + assert.Empty(t, artifact.ChildrenPaths) + assert.Zero(t, observedLogs.FilterLevelExact(zapcore.ErrorLevel).Len()) +} + +func TestHasSharedArtifactMetadataUsesParentChildRelationships(t *testing.T) { + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + model := &v1beta1.BaseModel{ + ObjectMeta: metav1.ObjectMeta{Name: "model", Namespace: "default"}, + } + currentKey := constants.GetModelConfigMapKey(model.Namespace, model.Name, false) + + plainGopher := newGopherWithConfigMap(makeConfigMap("node-1", map[string]string{ + currentKey: modelEntryJSON(ModelStatusReady), + })) + assert.False(t, plainGopher.hasSharedArtifactMetadata(context.Background(), &GopherTask{BaseModel: model})) + + originGopher := newGopherWithConfigMap(makeConfigMap("node-1", map[string]string{ + currentKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, currentKey, "/models/parent", []string{}), + })) + assert.True(t, originGopher.hasSharedArtifactMetadata(context.Background(), &GopherTask{BaseModel: model})) + + legacyGopher := newGopherWithConfigMap(makeConfigMap("node-1", map[string]string{ + currentKey: entryJSON(sha, currentKey, "/models/parent"), + })) + assert.True(t, legacyGopher.hasSharedArtifactMetadata(context.Background(), &GopherTask{BaseModel: model})) + + noParentGopher := newGopherWithConfigMap(makeConfigMap("node-1", map[string]string{ + currentKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, currentKey, "", []string{}), + })) + assert.False(t, noParentGopher.hasSharedArtifactMetadata(context.Background(), &GopherTask{BaseModel: model})) + + annotatedModel := model.DeepCopy() + annotatedModel.Annotations = map[string]string{ + HuggingFaceModelIDAnnotationKey: modelID, + HuggingFaceSHAAnnotationKey: sha, + } + assert.False(t, plainGopher.hasSharedArtifactMetadata(context.Background(), &GopherTask{BaseModel: annotatedModel})) } func TestFindReadyObjectStorageModelWithSamePath(t *testing.T) { @@ -1134,10 +2795,25 @@ func TestEnqueueTaskClassifiesStartupReadyLocalPathAsRevalidation(t *testing.T) func TestCaptureStartupReadyModelsCapturesOnlyReadyEntries(t *testing.T) { readyKey := constants.GetModelConfigMapKey("service-ns", "ready-model", false) updatingKey := constants.GetModelConfigMapKey("service-ns", "updating-model", false) + modelID := "Qwen/Qwen3-4B-Instruct-2507" + sha := "cdbee75f17c01a7cc42f958dc650907174af0554" + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: sha, + } + readyParentKey := huggingFaceArtifactConfigMapKey(identity) + updatingParentKey := huggingFaceArtifactConfigMapKey(ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: "Qwen/Qwen3-8B", + HFCommitSHA: sha, + }) cm := makeConfigMap("node-1", map[string]string{ - readyKey: modelEntryJSON(ModelStatusReady), - updatingKey: modelEntryJSON(ModelStatusUpdating), - "invalid": "not-json", + readyKey: modelEntryJSON(ModelStatusReady), + updatingKey: modelEntryJSON(ModelStatusUpdating), + readyParentKey: entryJSONWithOrigin(ModelStatusReady, modelID, sha, readyParentKey, "/tmp/ready-parent", []string{}), + updatingParentKey: entryJSONWithOrigin(ModelStatusUpdating, "Qwen/Qwen3-8B", sha, updatingParentKey, "/tmp/updating-parent", []string{}), + "invalid": "not-json", }) g := newGopherForProcessTask(cm) @@ -1146,6 +2822,9 @@ func TestCaptureStartupReadyModelsCapturesOnlyReadyEntries(t *testing.T) { assert.Contains(t, g.startupReadyModelKeys, readyKey) assert.NotContains(t, g.startupReadyModelKeys, updatingKey) assert.NotContains(t, g.startupReadyModelKeys, "invalid") + assert.True(t, g.claimStartupHuggingFaceArtifactParentValidation(readyParentKey)) + assert.False(t, g.claimStartupHuggingFaceArtifactParentValidation(readyParentKey)) + assert.False(t, g.claimStartupHuggingFaceArtifactParentValidation(updatingParentKey)) } func TestCaptureStartupReadyModelsFeedsRevalidationClassification(t *testing.T) { diff --git a/pkg/modelagent/hf_artifact_identity.go b/pkg/modelagent/hf_artifact_identity.go new file mode 100644 index 000000000..96602f795 --- /dev/null +++ b/pkg/modelagent/hf_artifact_identity.go @@ -0,0 +1,222 @@ +package modelagent + +import ( + "crypto/sha256" + "encoding/hex" + "os" + "path/filepath" + "strings" + + "sigs.k8s.io/ome/pkg/apis/ome/v1beta1" + "sigs.k8s.io/ome/pkg/constants" +) + +func shouldUseHuggingFaceOriginObjectStorageReuse(task *GopherTask, baseModelSpec v1beta1.BaseModelSpec) bool { + return shouldUseSamePathObjectStorageReuse(task) && + baseModelSpec.Storage != nil && + baseModelSpec.Storage.DownloadPolicy != nil && + *baseModelSpec.Storage.DownloadPolicy == v1beta1.ReuseIfExists +} + +func shouldRepairHuggingFaceOriginObjectStorageParent(task *GopherTask, baseModelSpec v1beta1.BaseModelSpec) bool { + return task != nil && + task.TaskType == DownloadOverride && + baseModelSpec.Storage != nil && + baseModelSpec.Storage.DownloadPolicy != nil && + *baseModelSpec.Storage.DownloadPolicy == v1beta1.ReuseIfExists +} + +func (i ArtifactIdentity) isValid() bool { + return strings.EqualFold(i.OriginType, ArtifactOriginTypeHuggingFace) && + isValidHuggingFaceModelID(i.HFModelID) && + isValidHuggingFaceCommitSHA(i.HFCommitSHA) +} + +func (i ArtifactIdentity) toOrigin() *ArtifactOrigin { + if !i.isValid() { + return nil + } + return &ArtifactOrigin{ + Type: ArtifactOriginTypeHuggingFace, + HFModelID: i.HFModelID, + HFCommitSHA: strings.ToLower(i.HFCommitSHA), + } +} + +func huggingFaceArtifactParentIdentityAndPath(parentKey string, entry ModelEntry) (ArtifactIdentity, string, bool) { + if entry.Config == nil { + return ArtifactIdentity{}, "", false + } + origin := entry.Config.Artifact.Origin + if origin == nil { + return ArtifactIdentity{}, "", false + } + identity := ArtifactIdentity{ + OriginType: origin.Type, + HFModelID: origin.HFModelID, + HFCommitSHA: strings.ToLower(origin.HFCommitSHA), + } + parentPath := entry.Config.Artifact.ParentPath[parentKey] + if !identity.isValid() || strings.TrimSpace(parentPath) == "" { + return ArtifactIdentity{}, "", false + } + return identity, parentPath, true +} + +func isValidHuggingFaceCommitSHA(sha string) bool { + if len(sha) != 40 { + return false + } + for _, c := range sha { + if !((c >= '0' && c <= '9') || (c >= 'a' && c <= 'f') || (c >= 'A' && c <= 'F')) { + return false + } + } + return true +} + +func isValidHuggingFaceModelID(modelID string) bool { + modelID = strings.TrimSpace(modelID) + if modelID == "" || + strings.HasPrefix(modelID, "/") || + strings.HasSuffix(modelID, "/") || + strings.Contains(modelID, "\\") || + strings.Contains(modelID, "//") || + strings.Contains(modelID, "..") || + strings.Contains(modelID, "--") || + strings.HasSuffix(modelID, ".git") { + return false + } + parts := strings.Split(modelID, "/") + if len(parts) > 2 { + return false + } + for _, part := range parts { + if !isValidHuggingFaceModelIDSegment(part) { + return false + } + } + return true +} + +func isValidHuggingFaceModelIDSegment(segment string) bool { + if segment == "" || + segment == "." || + segment == ".." || + strings.HasPrefix(segment, ".") || + strings.HasPrefix(segment, "-") || + strings.HasSuffix(segment, ".") || + strings.HasSuffix(segment, "-") || + len(segment) > 96 { + return false + } + for _, c := range segment { + if (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '-' || c == '_' || c == '.' { + continue + } + return false + } + return true +} + +func huggingFaceArtifactIdentityFromTask(task *GopherTask) (ArtifactIdentity, bool) { + if task == nil { + return ArtifactIdentity{}, false + } + if task.BaseModel != nil { + return huggingFaceArtifactIdentityFromAnnotations(task.BaseModel.Annotations) + } + if task.ClusterBaseModel != nil { + return huggingFaceArtifactIdentityFromAnnotations(task.ClusterBaseModel.Annotations) + } + return ArtifactIdentity{}, false +} + +func huggingFaceArtifactIdentityFromAnnotations(annotations map[string]string) (ArtifactIdentity, bool) { + if annotations == nil { + return ArtifactIdentity{}, false + } + modelID := strings.TrimSpace(annotations[HuggingFaceModelIDAnnotationKey]) + sha := strings.TrimSpace(annotations[HuggingFaceSHAAnnotationKey]) + identity := ArtifactIdentity{ + OriginType: ArtifactOriginTypeHuggingFace, + HFModelID: modelID, + HFCommitSHA: strings.ToLower(sha), + } + if !identity.isValid() { + return ArtifactIdentity{}, false + } + return identity, true +} + +func huggingFaceArtifactConfigMapKey(identity ArtifactIdentity) string { + return huggingFaceArtifactConfigMapKeyPrefix + sanitizeConfigMapKeyComponent(identity.HFModelID) + "." + shortConfigMapKeyHash(identity.HFModelID) + "." + strings.ToLower(identity.HFCommitSHA) +} + +func isHuggingFaceArtifactConfigMapKey(key string) bool { + return strings.HasPrefix(key, huggingFaceArtifactConfigMapKeyPrefix) +} + +// shortConfigMapKeyHash keeps the synthetic parent key readable while avoiding +// collisions from lossy path sanitization of Hugging Face model IDs. +func shortConfigMapKeyHash(value string) string { + sum := sha256.Sum256([]byte(strings.TrimSpace(value))) + return hex.EncodeToString(sum[:])[:12] +} + +func sanitizeConfigMapKeyComponent(value string) string { + var b strings.Builder + for _, c := range strings.TrimSpace(value) { + if (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') || c == '-' || c == '_' || c == '.' { + b.WriteRune(c) + continue + } + b.WriteByte('.') + } + sanitized := strings.Trim(b.String(), ".") + if sanitized == "" { + return "unknown" + } + return sanitized +} + +func canonicalHuggingFaceArtifactPath(destPath string, identity ArtifactIdentity) string { + return filepath.Join(filepath.Dir(destPath), constants.ModelArtifactsDirectory, filepath.FromSlash(strings.Trim(strings.TrimSpace(identity.HFModelID), "/")), strings.ToLower(identity.HFCommitSHA)) +} + +// BaseModel downloads can have model-local child paths under a local store, so +// place shared HF parents in that same store. ClusterBaseModel keeps the +// existing model-root layout for backward compatibility. +func canonicalHuggingFaceArtifactPathForTask(task *GopherTask, modelRootDir string, destPath string, identity ArtifactIdentity) string { + if task != nil && task.BaseModel != nil { + return canonicalHuggingFaceArtifactPath(destPath, identity) + } + return filepath.Join(modelRootDir, filepath.FromSlash(strings.Trim(strings.TrimSpace(identity.HFModelID), "/")), strings.ToLower(identity.HFCommitSHA)) +} + +func huggingFaceArtifactReadyMarkerPath(parentPath string) string { + return filepath.Join(parentPath, huggingFaceArtifactReadyMarkerFile) +} + +func writeHuggingFaceArtifactReadyMarker(parentPath string) error { + if err := os.MkdirAll(parentPath, 0755); err != nil { + return err + } + return os.WriteFile(huggingFaceArtifactReadyMarkerPath(parentPath), []byte("ready\n"), 0644) +} + +func removeHuggingFaceArtifactReadyMarker(parentPath string) error { + err := os.Remove(huggingFaceArtifactReadyMarkerPath(parentPath)) + if err == nil || os.IsNotExist(err) { + return nil + } + return err +} + +func hasHuggingFaceArtifactReadyMarker(parentPath string) bool { + if strings.TrimSpace(parentPath) == "" { + return false + } + info, err := os.Stat(huggingFaceArtifactReadyMarkerPath(parentPath)) + return err == nil && !info.IsDir() +} diff --git a/pkg/modelagent/model_data.go b/pkg/modelagent/model_data.go index 47716ff7c..6a6cbe1a2 100644 --- a/pkg/modelagent/model_data.go +++ b/pkg/modelagent/model_data.go @@ -25,6 +25,17 @@ const ( // ConfigParsingAnnotation is the annotation key to skip config parsing const ConfigParsingAnnotation = "ome.oracle.com/skip-config-parsing" +// Hugging Face origin annotations are written by the control plane for OCI +// imports that were originally resolved from Hugging Face. The model agent uses +// them only as provenance metadata for local artifact reuse; the model source +// remains OCI Object Storage. +const ( + ArtifactOriginTypeHuggingFace = "huggingface" + + HuggingFaceModelIDAnnotationKey = "hf-model-id" + HuggingFaceSHAAnnotationKey = "hf-model-sha" +) + // ModelMetadata is the metadata produced by the shared modelparser bridge. // It is aliased so existing modelagent code (cache, configmap reconciler, // gopher) keeps using the unqualified name while the parser implementation @@ -59,6 +70,19 @@ type ModelConfig struct { // Aliased to the shared modelparser type. type Artifact = modelparser.Artifact +// ArtifactOrigin records the source identity used to prove two local artifacts +// are equivalent even when they were downloaded through different storage +// backends. +type ArtifactOrigin = modelparser.ArtifactOrigin + +// ArtifactIdentity is the normalized identity used while searching the node +// ConfigMap for a reusable artifact. +type ArtifactIdentity struct { + OriginType string + HFModelID string + HFCommitSHA string +} + // DownloadProgress tracks the progress of a model download type DownloadProgress struct { Phase string `json:"phase"` // Scanning, Downloading, Finalizing @@ -128,8 +152,16 @@ func ConvertMetadataToModelConfig(metadata ModelMetadata) *ModelConfig { // convert artifact var artifact Artifact - if metadata.Artifact.Sha != "" || metadata.Artifact.ParentPath != nil || metadata.Artifact.ChildrenPaths != nil { + if metadata.Artifact.Sha != "" || metadata.Artifact.Origin != nil || metadata.Artifact.ParentPath != nil || metadata.Artifact.ChildrenPaths != nil { currentArtifact := metadata.Artifact + var origin *ArtifactOrigin + if currentArtifact.Origin != nil { + origin = &ArtifactOrigin{ + Type: currentArtifact.Origin.Type, + HFModelID: currentArtifact.Origin.HFModelID, + HFCommitSHA: currentArtifact.Origin.HFCommitSHA, + } + } // Deep copy ParentPath to avoid aliasing var parent map[string]string if currentArtifact.ParentPath != nil { @@ -146,6 +178,7 @@ func ConvertMetadataToModelConfig(metadata ModelMetadata) *ModelConfig { } artifact = Artifact{ Sha: currentArtifact.Sha, + Origin: origin, ParentPath: parent, ChildrenPaths: children, } diff --git a/pkg/modelparser/model_metadata.go b/pkg/modelparser/model_metadata.go index d29a0a76b..5c7a82bf0 100644 --- a/pkg/modelparser/model_metadata.go +++ b/pkg/modelparser/model_metadata.go @@ -32,6 +32,9 @@ type ModelMetadata struct { // Artifact records the information of model artifact, including version (Sha) and storage paths type Artifact struct { Sha string `json:"sha"` // sha string fetched from HuggingFace + // Origin captures optional provenance for artifact reuse across storage + // backends, for example OCI objects imported from a Hugging Face revision. + Origin *ArtifactOrigin `json:"origin,omitempty"` // parent model name -> parent model artifact storage path // parent name convention is // For ClusterBaseModel: clusterbasemodel.{model_name} @@ -39,3 +42,12 @@ type Artifact struct { ParentPath map[string]string `json:"parentPath"` ChildrenPaths []string `json:"childrenPaths"` // an array of children paths } + +// ArtifactOrigin records the source identity used to prove two local artifacts +// are equivalent even when they were downloaded through different storage +// backends. +type ArtifactOrigin struct { + Type string `json:"type,omitempty"` + HFModelID string `json:"hfModelId,omitempty"` + HFCommitSHA string `json:"hfCommitSha,omitempty"` +}