feat(catalog): load YAML catalogs into the database (kubeflow#1697)

* feat(catalog): load YAML catalogs into the database

- Add new ModelProviderFunc interface for pluggable model providers
- Implement SourceCollection for managing catalog sources and origins
- Refactor catalog loading to support multiple source formats
- Add comprehensive test coverage for new loading functionality
- Include database migrations for foreign key constraints
- Extend catalog models to support new artifact types

This introduces a more flexible catalog loading architecture that supports
loading models from YAML sources while maintaining backward compatibility.

Signed-off-by: Paul Boyd <[email protected]>

* fix(catalog): use the correct types for logo and readme fields

Signed-off-by: Paul Boyd <[email protected]>

* fix(catalog): silence missed event warnings

Signed-off-by: Paul Boyd <[email protected]>

* fix(catalog): default language and tasks to empty slices

Signed-off-by: Paul Boyd <[email protected]>

* fix(catalog): pause after receiving fsnotify events

We see multiple change events for files that are partially written, so
pause a bit to give time for the write to finish before we process it.

Signed-off-by: Paul Boyd <[email protected]>

---------

Signed-off-by: Paul Boyd <[email protected]>

Author: pboyd

catalog/README.md

@@ -122,7 +122,7 @@ type CatalogSourceProvider interface {
 
 2. Register your provider:
 ```go
-catalog.RegisterCatalogType("my-catalog", func(source *CatalogSourceConfig) (CatalogSourceProvider, error) {
+catalog.RegisterCatalogType("my-catalog", func(source *Source) (CatalogSourceProvider, error) {
     return NewMyCatalogProvider(source)
 })
 ```
@@ -145,4 +145,4 @@ The catalog service is designed to complement the main Model Registry service by
 - Unified metadata aggregation
 - Read-only access to distributed model catalogs
 
-For complete Model Registry documentation, see the [main README](../README.md).
+For complete Model Registry documentation, see the [main README](../README.md).

catalog/cmd/catalog.go

@@ -1,6 +1,7 @@
 package cmd
 
 import (
+	"context"
 	"fmt"
 	"net/http"
 	"reflect"
@@ -53,7 +54,14 @@ func runCatalogServer(cmd *cobra.Command, args []string) error {
 		return fmt.Errorf("error initializing datastore: %v", err)
 	}
 
-	sources, err := catalog.LoadCatalogSources(catalogCfg.ConfigPath)
+	services := service.NewServices(
+		getRepo[models.CatalogModelRepository](repoSet),
+		getRepo[models.CatalogArtifactRepository](repoSet),
+		getRepo[models.CatalogModelArtifactRepository](repoSet),
+		getRepo[models.CatalogMetricsArtifactRepository](repoSet),
+	)
+
+	sources, err := catalog.LoadCatalogSources(context.Background(), services, catalogCfg.ConfigPath)
 	if err != nil {
 		return fmt.Errorf("error loading catalog sources: %v", err)
 	}

catalog/internal/catalog/catalog.go

@@ -2,13 +2,6 @@ package catalog
 
 import (
 	"context"
-	"fmt"
-	"os"
-	"path/filepath"
-	"sync"
-
-	"github.com/golang/glog"
-	"k8s.io/apimachinery/pkg/util/yaml"
 
 	model "github.com/kubeflow/model-registry/catalog/pkg/openapi"
 )
@@ -29,8 +22,8 @@ type ListArtifactsParams struct {
 	NextPageToken *string
 }
 
-// CatalogSourceProvider is implemented by catalog source types, e.g. YamlCatalog
-type CatalogSourceProvider interface {
+// APIProvider implements the API endpoints.
+type APIProvider interface {
 	// GetModel returns model metadata for a single model by its name. If
 	// nothing is found with the name provided it returns nil, without an
 	// error.
@@ -46,162 +39,3 @@ type CatalogSourceProvider interface {
 	// found, but has no artifacts, an empty list is returned.
 	GetArtifacts(ctx context.Context, modelName string, sourceID string, params ListArtifactsParams) (model.CatalogArtifactList, error)
 }
-
-// CatalogSourceConfig is a single entry from the catalog sources YAML file.
-type CatalogSourceConfig struct {
-	model.CatalogSource `json:",inline"`
-
-	// Catalog type to use, must match one of the registered types
-	Type string `json:"type"`
-
-	// Properties used for configuring the catalog connection based on catalog implementation
-	Properties map[string]any `json:"properties,omitempty"`
-}
-
-// sourceConfig is the structure for the catalog sources YAML file.
-type sourceConfig struct {
-	Catalogs []CatalogSourceConfig `json:"catalogs"`
-}
-
-type CatalogTypeRegisterFunc func(source *CatalogSourceConfig, reldir string) (CatalogSourceProvider, error)
-
-var registeredCatalogTypes = make(map[string]CatalogTypeRegisterFunc, 0)
-
-func RegisterCatalogType(catalogType string, callback CatalogTypeRegisterFunc) error {
-	if _, exists := registeredCatalogTypes[catalogType]; exists {
-		return fmt.Errorf("catalog type %s already exists", catalogType)
-	}
-	registeredCatalogTypes[catalogType] = callback
-	return nil
-}
-
-type CatalogSource struct {
-	Provider CatalogSourceProvider
-	Metadata model.CatalogSource
-}
-
-type SourceCollection struct {
-	sourcesMu sync.RWMutex
-	sources   map[string]CatalogSource
-}
-
-func NewSourceCollection(sources map[string]CatalogSource) *SourceCollection {
-	return &SourceCollection{sources: sources}
-}
-
-func (sc *SourceCollection) All() map[string]CatalogSource {
-	sc.sourcesMu.RLock()
-	defer sc.sourcesMu.RUnlock()
-
-	return sc.sources
-}
-
-func (sc *SourceCollection) Get(name string) (src CatalogSource, ok bool) {
-	sc.sourcesMu.RLock()
-	defer sc.sourcesMu.RUnlock()
-
-	src, ok = sc.sources[name]
-	return
-}
-
-func (sc *SourceCollection) load(path string) error {
-	// Get absolute path of the catalog config file
-	absConfigPath, err := filepath.Abs(path)
-	if err != nil {
-		return fmt.Errorf("failed to get absolute path for %s: %v", path, err)
-	}
-
-	// Get the directory of the config file to resolve relative paths
-	configDir := filepath.Dir(absConfigPath)
-
-	config := sourceConfig{}
-	bytes, err := os.ReadFile(absConfigPath)
-	if err != nil {
-		return err
-	}
-
-	if err = yaml.UnmarshalStrict(bytes, &config); err != nil {
-		return err
-	}
-
-	sources := make(map[string]CatalogSource, len(config.Catalogs))
-	for _, catalogConfig := range config.Catalogs {
-		// If enabled is explicitly set to false, skip
-		hasEnabled := catalogConfig.HasEnabled()
-		if hasEnabled && *catalogConfig.Enabled == false {
-			continue
-		}
-		// If not explicitly set, default to enabled
-		if !hasEnabled {
-			t := true
-			catalogConfig.CatalogSource.Enabled = &t
-		}
-
-		catalogType := catalogConfig.Type
-		glog.Infof("reading config type %s...", catalogType)
-		registerFunc, ok := registeredCatalogTypes[catalogType]
-		if !ok {
-			return fmt.Errorf("catalog type %s not registered", catalogType)
-		}
-		id := catalogConfig.GetId()
-		if len(id) == 0 {
-			return fmt.Errorf("invalid catalog id %s", id)
-		}
-		if _, exists := sources[id]; exists {
-			return fmt.Errorf("duplicate catalog id %s", id)
-		}
-
-		labels := make([]string, 0)
-		if catalogConfig.GetLabels() != nil {
-			labels = catalogConfig.GetLabels()
-		}
-		catalogConfig.CatalogSource.Labels = labels
-		provider, err := registerFunc(&catalogConfig, configDir)
-		if err != nil {
-			return fmt.Errorf("error reading catalog type %s with id %s: %v", catalogType, id, err)
-		}
-
-		sources[id] = CatalogSource{
-			Provider: provider,
-			Metadata: catalogConfig.CatalogSource,
-		}
-
-		glog.Infof("loaded config %s of type %s", id, catalogType)
-	}
-
-	sc.sourcesMu.Lock()
-	defer sc.sourcesMu.Unlock()
-	sc.sources = sources
-
-	return nil
-}
-
-func LoadCatalogSources(paths []string) (*SourceCollection, error) {
-	sc := &SourceCollection{}
-
-	for _, path := range paths {
-		err := sc.load(path)
-		if err != nil {
-			return nil, err
-		}
-
-		go func(path string) {
-			changes, err := getMonitor().Path(path)
-			if err != nil {
-				glog.Errorf("unable to watch sources file (%s): %v", path, err)
-				// Not fatal, we just won't get automatic updates.
-			}
-
-			for range changes {
-				glog.Infof("Reloading sources %s", path)
-
-				err = sc.load(path)
-				if err != nil {
-					glog.Errorf("unable to load sources: %v", err)
-				}
-			}
-		}(path)
-	}
-
-	return sc, nil
-}