feat(catalog): support for multiple catalog pods

Enables horizontal scaling through PostgreSQL-based leader election
that coordinates database writes across multiple pods.

All pods serve read requests from in-memory data and database queries.
The leader alone performs database writes: fetches models, writes
updates, and cleans up orphaned data. Leadership transfers
automatically when the leader fails.

Implementation:
- Leader election package using pglock for distributed locking
- Loader split into StartReadOnly() and StartLeader() modes
- Configuration: CATALOG_LEADER_LOCK_DURATION and
  CATALOG_LEADER_HEARTBEAT environment variables
- Integration tests for multi-pod scenarios

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

Author: pboyd

catalog/cmd/catalog.go

@@ -2,15 +2,21 @@ package cmd
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net/http"
+	"os"
+	"os/signal"
 	"reflect"
+	"sync"
+	"syscall"
 	"time"
 
 	"github.com/golang/glog"
 	"github.com/kubeflow/model-registry/catalog/internal/catalog"
 	"github.com/kubeflow/model-registry/catalog/internal/db/models"
 	"github.com/kubeflow/model-registry/catalog/internal/db/service"
+	"github.com/kubeflow/model-registry/catalog/internal/leader"
 	"github.com/kubeflow/model-registry/catalog/internal/server/openapi"
 	"github.com/kubeflow/model-registry/internal/datastore"
 	"github.com/kubeflow/model-registry/internal/datastore/embedmd"
@@ -27,6 +33,49 @@ var catalogCfg = struct {
 	PerformanceMetricsPath: []string{},
 }
 
+const (
+	leaderLockName = "catalog-leader"
+
+	defaultLeaderLockDuration = 60 * time.Second
+	defaultLeaderHeartbeat    = 15 * time.Second
+
+	envLeaderLockDuration = "CATALOG_LEADER_LOCK_DURATION"
+	envLeaderHeartbeat    = "CATALOG_LEADER_HEARTBEAT"
+)
+
+// getLeaderElectionConfig reads leader election configuration from environment
+// variables, falling back to defaults when unset or invalid.
+func getLeaderElectionConfig() (lockDuration, heartbeat time.Duration) {
+	lockDuration = defaultLeaderLockDuration
+	heartbeat = defaultLeaderHeartbeat
+
+	if envDuration := os.Getenv(envLeaderLockDuration); envDuration != "" {
+		if parsed, err := time.ParseDuration(envDuration); err == nil {
+			lockDuration = parsed
+			glog.Infof("Using leader lock duration from %s: %v", envLeaderLockDuration, lockDuration)
+		} else {
+			glog.Warningf("Invalid %s value %q: %v (using default %v)", envLeaderLockDuration, envDuration, err, defaultLeaderLockDuration)
+		}
+	}
+
+	if envHB := os.Getenv(envLeaderHeartbeat); envHB != "" {
+		if parsed, err := time.ParseDuration(envHB); err == nil {
+			heartbeat = parsed
+			glog.Infof("Using leader heartbeat from %s: %v", envLeaderHeartbeat, heartbeat)
+		} else {
+			glog.Warningf("Invalid %s value %q: %v (using default %v)", envLeaderHeartbeat, envHB, err, defaultLeaderHeartbeat)
+		}
+	}
+
+	// Validate pglock requirement: heartbeat <= lockDuration/2
+	if heartbeat > lockDuration/2 {
+		glog.Warningf("Heartbeat (%v) exceeds half of lock duration (%v), required by pglock. Using defaults.", heartbeat, lockDuration)
+		return defaultLeaderLockDuration, defaultLeaderHeartbeat
+	}
+
+	return lockDuration, heartbeat
+}
+
 var CatalogCmd = &cobra.Command{
 	Use:   "catalog",
 	Short: "Catalog API server",
@@ -81,11 +130,23 @@ func runCatalogServer(cmd *cobra.Command, args []string) error {
 		return nil
 	})
 
-	err = loader.Start(context.Background())
-	if err != nil {
-		return fmt.Errorf("error loading catalog sources: %v", err)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	sigCh := make(chan os.Signal, 1)
+	signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT)
+	go func() {
+		sig := <-sigCh
+		glog.Infof("Received signal %v, initiating graceful shutdown", sig)
+		cancel()
+	}()
+
+	glog.Info("Starting loader in read-only mode (standby)")
+	if err := loader.StartReadOnly(ctx); err != nil {
+		return fmt.Errorf("error starting loader in read-only mode: %v", err)
 	}
 
+	// Set up HTTP server (runs continuously regardless of leadership)
 	svc := openapi.NewModelCatalogServiceAPIService(
 		catalog.NewDBCatalog(services, loader.Sources),
 		loader.Sources,
@@ -94,8 +155,97 @@ func runCatalogServer(cmd *cobra.Command, args []string) error {
 	)
 	ctrl := openapi.NewModelCatalogServiceAPIController(svc)
 
-	glog.Infof("Catalog API server listening on %s", catalogCfg.ListenAddress)
-	return http.ListenAndServe(catalogCfg.ListenAddress, openapi.NewRouter(ctrl))
+	server := &http.Server{
+		Addr:    catalogCfg.ListenAddress,
+		Handler: openapi.NewRouter(ctrl),
+	}
+
+	var wg sync.WaitGroup
+
+	errCh := make(chan error, 3)
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		glog.Infof("Catalog API server listening on %s", catalogCfg.ListenAddress)
+		if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+			errCh <- fmt.Errorf("HTTP server failed: %w", err)
+		}
+	}()
+
+	go func() {
+		<-ctx.Done()
+		glog.Info("Shutting down HTTP server...")
+		shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second)
+		defer shutdownCancel()
+		if err := server.Shutdown(shutdownCtx); err != nil {
+			glog.Errorf("HTTP server shutdown error: %v", err)
+		}
+	}()
+
+	gormDB, err := ds.DB()
+	if err != nil {
+		return fmt.Errorf("error getting database connection: %w", err)
+	}
+
+	lockDuration, heartbeat := getLeaderElectionConfig()
+	glog.Infof("Leader election configured: lock duration=%v, heartbeat=%v", lockDuration, heartbeat)
+
+	elector, err := leader.NewLeaderElector(
+		gormDB,
+		ctx,
+		leaderLockName,
+		lockDuration,
+		heartbeat,
+		func(leaderCtx context.Context) {
+			glog.Info("Became leader - starting leader-only operations")
+
+			// Monitor leaderCtx in separate goroutine and call StopLeader when lost
+			go func() {
+				<-leaderCtx.Done()
+				glog.Info("Lost leadership, stopping leader operations...")
+				if err := loader.StopLeader(); err != nil {
+					glog.Errorf("Error stopping leader: %v", err)
+				}
+			}()
+
+			// StartLeader blocks until StopLeader is called or ctx cancels
+			// Pass program context (ctx), NOT leaderCtx
+			if err := loader.StartLeader(ctx); err != nil && !errors.Is(err, context.Canceled) {
+				glog.Errorf("StartLeader exited with error: %v", err)
+			}
+
+			glog.Info("Leader callback complete")
+		},
+	)
+	if err != nil {
+		return fmt.Errorf("error creating leader elector: %w", err)
+	}
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if err := elector.Run(ctx); err != nil {
+			errCh <- fmt.Errorf("leader elector failed: %w", err)
+		}
+	}()
+
+	go func() {
+		defer close(errCh)
+		wg.Wait()
+	}()
+
+	errs := []error{}
+	for err := range errCh {
+		if !errors.Is(err, context.Canceled) {
+			errs = append(errs, err)
+		}
+	}
+
+	if err := loader.Shutdown(); err != nil {
+		errs = append(errs, fmt.Errorf("loader shutdown error: %w", err))
+	}
+
+	return errors.Join(errs...)
 }
 
 func getRepo[T any](repoSet datastore.RepoSet) T {

catalog/cmd/catalog_config_test.go

@@ -0,0 +1,108 @@
+package cmd
+
+import (
+	"os"
+	"testing"
+	"time"
+)
+
+func TestGetLeaderElectionConfig(t *testing.T) {
+	tests := []struct {
+		name                 string
+		lockDurationEnv      string
+		heartbeatEnv         string
+		expectedLockDuration time.Duration
+		expectedHeartbeat    time.Duration
+	}{
+		{
+			name:                 "defaults when no env vars set",
+			lockDurationEnv:      "",
+			heartbeatEnv:         "",
+			expectedLockDuration: defaultLeaderLockDuration,
+			expectedHeartbeat:    defaultLeaderHeartbeat,
+		},
+		{
+			name:                 "custom valid values",
+			lockDurationEnv:      "30s",
+			heartbeatEnv:         "10s",
+			expectedLockDuration: 30 * time.Second,
+			expectedHeartbeat:    10 * time.Second,
+		},
+		{
+			name:                 "fast failover for local dev",
+			lockDurationEnv:      "10s",
+			heartbeatEnv:         "3s",
+			expectedLockDuration: 10 * time.Second,
+			expectedHeartbeat:    3 * time.Second,
+		},
+		{
+			name:                 "long lease for production",
+			lockDurationEnv:      "120s",
+			heartbeatEnv:         "30s",
+			expectedLockDuration: 120 * time.Second,
+			expectedHeartbeat:    30 * time.Second,
+		},
+		{
+			name:                 "invalid lock duration uses default lock, keeps valid heartbeat",
+			lockDurationEnv:      "invalid",
+			heartbeatEnv:         "10s",
+			expectedLockDuration: defaultLeaderLockDuration, // falls back to default
+			expectedHeartbeat:    10 * time.Second,          // uses parsed value
+		},
+		{
+			name:                 "invalid heartbeat uses default heartbeat, keeps valid lock",
+			lockDurationEnv:      "30s",
+			heartbeatEnv:         "invalid",
+			expectedLockDuration: 30 * time.Second,       // uses parsed value
+			expectedHeartbeat:    defaultLeaderHeartbeat, // falls back to default
+		},
+		{
+			name:                 "heartbeat exceeds lock duration/2 uses defaults",
+			lockDurationEnv:      "30s",
+			heartbeatEnv:         "20s", // 20s > 15s (half of 30s)
+			expectedLockDuration: defaultLeaderLockDuration,
+			expectedHeartbeat:    defaultLeaderHeartbeat,
+		},
+		{
+			name:                 "heartbeat exactly at lock duration/2 is valid",
+			lockDurationEnv:      "30s",
+			heartbeatEnv:         "15s",
+			expectedLockDuration: 30 * time.Second,
+			expectedHeartbeat:    15 * time.Second,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			// Clear environment
+			os.Unsetenv(envLeaderLockDuration)
+			os.Unsetenv(envLeaderHeartbeat)
+
+			// Set test environment
+			if tt.lockDurationEnv != "" {
+				os.Setenv(envLeaderLockDuration, tt.lockDurationEnv)
+				defer os.Unsetenv(envLeaderLockDuration)
+			}
+			if tt.heartbeatEnv != "" {
+				os.Setenv(envLeaderHeartbeat, tt.heartbeatEnv)
+				defer os.Unsetenv(envLeaderHeartbeat)
+			}
+
+			// Get configuration
+			lockDuration, heartbeat := getLeaderElectionConfig()
+
+			// Verify
+			if lockDuration != tt.expectedLockDuration {
+				t.Errorf("lock duration = %v, want %v", lockDuration, tt.expectedLockDuration)
+			}
+			if heartbeat != tt.expectedHeartbeat {
+				t.Errorf("heartbeat = %v, want %v", heartbeat, tt.expectedHeartbeat)
+			}
+
+			// Verify pglock requirement: heartbeat <= lockDuration/2
+			if heartbeat > lockDuration/2 {
+				t.Errorf("heartbeat (%v) exceeds half of lock duration (%v), violates pglock requirement", heartbeat, lockDuration)
+			}
+		})
+	}
+}