Merge pull request #1424 from kubeflow/main

[pull] main from kubeflow:main

Author: openshift-merge-bot[bot]

api/openapi/catalog.yaml

@@ -875,10 +875,12 @@ components:
       description: |-
         Operational status of a catalog source.
         - `available`: The source is functioning correctly and models can be retrieved
+        - `partially-available`: The source loaded some models successfully but encountered errors with others
         - `error`: The source is experiencing issues and cannot provide models
         - `disabled`: The source has been intentionally disabled
       enum:
         - available
+        - partially-available
         - error
         - disabled
       type: string

api/openapi/src/catalog.yaml

@@ -713,10 +713,12 @@ components:
       description: |-
         Operational status of a catalog source.
         - `available`: The source is functioning correctly and models can be retrieved
+        - `partially-available`: The source loaded some models successfully but encountered errors with others
         - `error`: The source is experiencing issues and cannot provide models
         - `disabled`: The source has been intentionally disabled
       enum:
         - available
+        - partially-available
         - error
         - disabled
       type: string

catalog/clients/python/src/catalog_openapi/models/catalog_source_status.py

@@ -145,7 +145,7 @@ def test_source_status_values(self, api_client: CatalogAPIClient, suppress_ssl_w
             assert status is not None or enabled is not None
 
             if status:
-                valid_statuses = {"available", "disabled", "error", "loading", "pending"}
+                valid_statuses = {"available", "partially-available", "disabled", "error", "loading", "pending"}
                 assert status in valid_statuses, f"Unexpected status: {status}"
 
     def test_enabled_and_status_consistency(self, api_client: CatalogAPIClient, suppress_ssl_warnings: None):

catalog/clients/python/tests/test_sources.py

@@ -9,20 +9,13 @@ import (
 )
 
 type ListModelsParams struct {
-	Query                 string
-	FilterQuery           string
-	SourceIDs             []string
-	SourceLabels          []string
-	PageSize              int32
-	OrderBy               model.OrderByField
-	SortOrder             model.SortOrder
-	NextPageToken         *string
-	Recommended           bool
-	TargetRPS             int32
-	LatencyProperty       string
-	RPSProperty           string
-	HardwareCountProperty string
-	HardwareTypeProperty  string
+	Query         string
+	FilterQuery   string
+	SourceIDs     []string
+	PageSize      int32
+	OrderBy       model.OrderByField
+	SortOrder     model.SortOrder
+	NextPageToken *string
 }
 
 type ListArtifactsParams struct {
@@ -63,7 +56,8 @@ type APIProvider interface {
 	// FindModelsWithRecommendedLatency returns models sorted by recommended latency using Pareto filtering.
 	// Models without computable latency appear at the end of results.
 	// If sourceIDs is provided, filter models by source IDs.
-	FindModelsWithRecommendedLatency(ctx context.Context, pagination mrmodels.Pagination, paretoParams dbmodels.ParetoFilteringParams, sourceIDs []string) (*model.CatalogModelList, error)
+	// If query is provided, filter models by text search.
+	FindModelsWithRecommendedLatency(ctx context.Context, pagination mrmodels.Pagination, paretoParams dbmodels.ParetoFilteringParams, sourceIDs []string, query string) (*model.CatalogModelList, error)
 
 	// GetArtifacts returns all artifacts for a particular model. If no
 	// model is found with that name, it returns nil. If the model is

catalog/internal/catalog/catalog.go

@@ -85,21 +85,6 @@ func (d *dbCatalogImpl) ListModels(ctx context.Context, params ListModelsParams)
 	}
 
 	sourceIDs := params.SourceIDs
-	if len(sourceIDs) == 0 && len(params.SourceLabels) > 0 {
-		sources := d.sources.ByLabel(params.SourceLabels)
-		if len(sources) == 0 {
-			// No matching sources, so no matching models.
-			return apimodels.CatalogModelList{
-				Items:    make([]apimodels.CatalogModel, 0),
-				PageSize: pageSize,
-			}, nil
-		}
-
-		sourceIDs = make([]string, len(sources))
-		for i, source := range sources {
-			sourceIDs[i] = source.Id
-		}
-	}
 
 	modelsList, err := d.catalogModelRepository.List(dbmodels.CatalogModelListOptions{
 		SourceIDs: &sourceIDs,
@@ -684,15 +669,22 @@ func (d *dbCatalogImpl) FindModelsWithRecommendedLatency(
 	pagination mrmodels.Pagination,
 	paretoParams dbmodels.ParetoFilteringParams,
 	sourceIDs []string,
+	query string,
 ) (*apimodels.CatalogModelList, error) {
 	// Get all models first (without pagination)
 	var sourceIDsPtr *[]string
 	if len(sourceIDs) > 0 {
 		sourceIDsPtr = &sourceIDs
 	}
 
+	var queryPtr *string
+	if query != "" {
+		queryPtr = &query
+	}
+
 	allModels, err := d.catalogModelRepository.List(dbmodels.CatalogModelListOptions{
 		SourceIDs: sourceIDsPtr,
+		Query:     queryPtr,
 		Pagination: mrmodels.Pagination{
 			FilterQuery: pagination.FilterQuery,
 			PageSize:    apiutils.Of(int32(0)), // Get all models

catalog/internal/catalog/db_catalog.go

@@ -2091,6 +2091,7 @@ func TestFindModelsWithRecommendedLatency(t *testing.T) {
 		pagination,
 		paretoParams,
 		[]string{"latency-test-source"}, // Filter by this test's source ID
+		"",                              // No query filter
 	)
 
 	require.NoError(t, err)

catalog/internal/catalog/db_catalog_test.go

@@ -549,7 +549,7 @@ func (p *hfModelProvider) getModelsFromHF(ctx context.Context) ([]ModelProviderR
 	}
 
 	if len(failedModels) > 0 {
-		return records, fmt.Errorf("Failed to fetch some models, ensure models exist and are accessible with given credentials. Failed models: %v", failedModels)
+		return records, &PartiallyAvailableError{FailedModels: failedModels}
 	}
 
 	return records, nil

catalog/internal/catalog/hf_catalog.go

@@ -150,7 +150,7 @@ func BenchmarkRecommendedLatencySorting(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		_, err := provider.FindModelsWithRecommendedLatency(ctx, mr_models.Pagination{
 			PageSize: apiutils.Of(int32(20)),
-		}, paretoParams, []string{"benchmark-source"})
+		}, paretoParams, []string{"benchmark-source"}, "")
 
 		require.NoError(b, err)
 	}

catalog/internal/catalog/integration_test.go

@@ -19,11 +19,30 @@ import (
 
 // Source status constants matching the OpenAPI enum values
 const (
-	SourceStatusAvailable = "available"
-	SourceStatusError     = "error"
-	SourceStatusDisabled  = "disabled"
+	SourceStatusAvailable          = "available"
+	SourceStatusPartiallyAvailable = "partially-available"
+	SourceStatusError              = "error"
+	SourceStatusDisabled           = "disabled"
 )
 
+// PartiallyAvailableError indicates that a source loaded some models successfully
+// but encountered errors with others.
+type PartiallyAvailableError struct {
+	FailedModels []string
+}
+
+func (e *PartiallyAvailableError) Error() string {
+	return fmt.Sprintf("Failed to fetch some models, ensure models exist and are accessible with given credentials. Failed models: %v", e.FailedModels)
+}
+
+func (e *PartiallyAvailableError) Is(target error) bool {
+	_, ok := target.(*PartiallyAvailableError)
+	return ok
+}
+
+// ErrPartiallyAvailable is used with errors.Is() to check for this error type.
+var ErrPartiallyAvailable error = &PartiallyAvailableError{}
+
 // ModelProviderRecord contains one model and its associated artifacts.
 type ModelProviderRecord struct {
 	Model     dbmodels.CatalogModel
@@ -450,9 +469,9 @@ func (l *Loader) readProviderRecords(ctx context.Context) <-chan ModelProviderRe
 					// Only save status if context is still valid (no reload in progress)
 					if ctx.Err() == nil {
 						// Check if there was a partial error (some models failed to load)
-						if r.Error != nil {
-							glog.Errorf("%s: partial error after loading models: %v", sourceID, r.Error)
-							l.saveSourceStatus(sourceID, SourceStatusError, r.Error.Error())
+						if errors.Is(r.Error, ErrPartiallyAvailable) {
+							glog.Warningf("%s: partial error after loading models: %v", sourceID, r.Error)
+							l.saveSourceStatus(sourceID, SourceStatusPartiallyAvailable, r.Error.Error())
 						} else {
 							l.saveSourceStatus(sourceID, SourceStatusAvailable, "")
 						}
@@ -620,7 +639,7 @@ func (l *Loader) removeOrphanedModelsFromSource(sourceID string, valid mapset.Se
 func (l *Loader) saveSourceStatus(sourceID, status string, errorMsg string) {
 	// Validate status is a valid enum value
 	switch status {
-	case SourceStatusAvailable, SourceStatusError, SourceStatusDisabled:
+	case SourceStatusAvailable, SourceStatusPartiallyAvailable, SourceStatusError, SourceStatusDisabled:
 		// valid
 	default:
 		glog.Errorf("invalid status %q for source %s", status, sourceID)