MCPRegistry e2e tests

.github/workflows/operator-ci.yml

@@ -187,3 +187,12 @@ jobs:
         chainsaw test --test-dir test/e2e/chainsaw/operator/single-tenancy/setup --config .chainsaw.yaml
         chainsaw test --test-dir test/e2e/chainsaw/operator/single-tenancy/test-scenarios --config .chainsaw.yaml
         chainsaw test --test-dir test/e2e/chainsaw/operator/single-tenancy/cleanup --config .chainsaw.yaml
+
+    - name: Install Ginkgo CLI
+      run: |
+        go install github.com/onsi/ginkgo/v2/ginkgo@latest
+
+    - name: Run Ginkgo tests
+      run: |
+        go install sigs.k8s.io/controller-runtime/tools/setup-envtest@latest
+        KUBEBUILDER_ASSETS="$(setup-envtest use 1.28.0 -p path)" ginkgo --github-output test/e2e/operator

cmd/thv-operator/Taskfile.yml

@@ -198,7 +198,16 @@ tasks:
       - go install sigs.k8s.io/controller-runtime/tools/setup-envtest@latest
       - KUBEBUILDER_ASSETS="$(shell setup-envtest use 1.28.0 -p path)" go test ./cmd/thv-operator/... -coverprofile cover.out
 
+  operator-e2e-test-ginkgo:
+    desc: Run E2E tests for the operator with Ginkgo
+    cmds:
+      - KUBEBUILDER_ASSETS="$(setup-envtest use 1.28.0 -p path)" ginkgo -v test/e2e/operator
+
+  # Backwards compatibility
   operator-e2e-test:
+    deps: [operator-e2e-test-chainsaw]
+
+  operator-e2e-test-chainsaw:
     desc: Run E2E tests for the operator
     cmds:
       - |

cmd/thv-operator/api/v1alpha1/mcpregistry_types.go

@@ -97,7 +97,7 @@ type GitSource struct {
 	// Repository is the Git repository URL (HTTP/HTTPS/SSH)
 	// +kubebuilder:validation:Required
 	// +kubebuilder:validation:MinLength=1
-	// +kubebuilder:validation:Pattern="^(https?://|git@|ssh://|git://).*"
+	// +kubebuilder:validation:Pattern="^(file:///|https?://|git@|ssh://|git://).*"
 	Repository string `json:"repository"`
 
 	// Branch is the Git branch to use (mutually exclusive with Tag and Commit)
@@ -186,6 +186,10 @@ type MCPRegistryStatus struct {
 	// +optional
 	APIStatus *APIStatus `json:"apiStatus,omitempty"`
 
+	// LastAppliedFilterHash is the hash of the last applied filter
+	// +optional
+	LastAppliedFilterHash string `json:"lastAppliedFilterHash,omitempty"`
+
 	// StorageRef is a reference to the internal storage location
 	// +optional
 	StorageRef *StorageReference `json:"storageRef,omitempty"`

cmd/thv-operator/controllers/mcpregistry_controller.go

@@ -2,6 +2,9 @@ package controllers
 
 import (
 	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
 	"fmt"
 	"time"
 
@@ -208,7 +211,7 @@ func (r *MCPRegistryReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 	r.deriveOverallStatus(ctx, mcpRegistry, statusManager, statusDeriver)
 
 	// 8. Apply all status changes in a single batch update
-	if statusUpdateErr := statusManager.Apply(ctx, r.Client); statusUpdateErr != nil {
+	if statusUpdateErr := r.applyStatusUpdates(ctx, r.Client, mcpRegistry, statusManager); statusUpdateErr != nil {
 		ctxLogger.Error(statusUpdateErr, "Failed to apply batched status update")
 		// Return the status update error only if there was no main reconciliation error
 		if syncErr == nil {
@@ -386,20 +389,23 @@ func (*MCPRegistryReconciler) deriveOverallStatus(
 	statusManager mcpregistrystatus.StatusManager, statusDeriver mcpregistrystatus.StatusDeriver) {
 	ctxLogger := log.FromContext(ctx)
 
+	syncStatus := statusManager.Sync().Status()
+	if syncStatus == nil {
+		syncStatus = mcpRegistry.Status.SyncStatus
+	}
+	apiStatus := statusManager.API().Status()
+	if apiStatus == nil {
+		apiStatus = mcpRegistry.Status.APIStatus
+	}
 	// Use the StatusDeriver to determine the overall phase and message
 	// based on current sync and API statuses
-	derivedPhase, derivedMessage := statusDeriver.DeriveOverallStatus(
-		mcpRegistry.Status.SyncStatus,
-		mcpRegistry.Status.APIStatus,
-	)
+	derivedPhase, derivedMessage := statusDeriver.DeriveOverallStatus(syncStatus, apiStatus)
 
 	// Only update phase and message if they've changed
 	statusManager.SetOverallStatus(derivedPhase, derivedMessage)
-	ctxLogger.Info("Updated overall status",
-		"oldPhase", mcpRegistry.Status.Phase,
-		"newPhase", derivedPhase,
-		"oldMessage", mcpRegistry.Status.Message,
-		"newMessage", derivedMessage)
+	ctxLogger.Info("Updated overall status", "syncStatus", syncStatus, "apiStatus", apiStatus,
+		"oldPhase", mcpRegistry.Status.Phase, "newPhase", derivedPhase,
+		"oldMessage", mcpRegistry.Status.Message, "newMessage", derivedMessage)
 }
 
 // SetupWithManager sets up the controller with the Manager.
@@ -411,3 +417,86 @@ func (r *MCPRegistryReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		Owns(&corev1.ConfigMap{}).
 		Complete(r)
 }
+
+// Apply applies all collected status changes in a single batch update.
+// Only actual changes are applied to the status to avoid unnecessary reconciliations
+func (r *MCPRegistryReconciler) applyStatusUpdates(
+	ctx context.Context, k8sClient client.Client,
+	mcpRegistry *mcpv1alpha1.MCPRegistry, statusManager mcpregistrystatus.StatusManager) error {
+
+	ctxLogger := log.FromContext(ctx)
+
+	// Refetch the latest version of the resource to avoid conflicts
+	latestRegistry := &mcpv1alpha1.MCPRegistry{}
+	if err := k8sClient.Get(ctx, client.ObjectKeyFromObject(mcpRegistry), latestRegistry); err != nil {
+		ctxLogger.Error(err, "Failed to fetch latest MCPRegistry version for status update")
+		return fmt.Errorf("failed to fetch latest MCPRegistry version: %w", err)
+	}
+	latestRegistryStatus := latestRegistry.Status
+	hasUpdates := false
+
+	// Apply manual sync trigger change if necessary
+	if mcpRegistry.Annotations != nil {
+		if triggerValue := mcpRegistry.Annotations[mcpregistrystatus.SyncTriggerAnnotation]; triggerValue != "" {
+			if latestRegistryStatus.LastManualSyncTrigger != triggerValue {
+				latestRegistryStatus.LastManualSyncTrigger = triggerValue
+				hasUpdates = true
+				ctxLogger.Info("Updated LastManualSyncTrigger", "trigger", triggerValue)
+			}
+		}
+	}
+
+	// Apply filter change if necessary
+	currentFilterJSON, err := json.Marshal(mcpRegistry.Spec.Filter)
+	if err != nil {
+		ctxLogger.Error(err, "Failed to marshal current filter")
+		return fmt.Errorf("failed to marshal current filter: %w", err)
+	}
+	currentFilterHash := sha256.Sum256(currentFilterJSON)
+	currentFilterHashStr := hex.EncodeToString(currentFilterHash[:])
+	if latestRegistryStatus.LastAppliedFilterHash != currentFilterHashStr {
+		latestRegistryStatus.LastAppliedFilterHash = currentFilterHashStr
+		hasUpdates = true
+		ctxLogger.Info("Updated LastAppliedFilterHash", "hash", currentFilterHashStr)
+	}
+
+	// Update storage reference if necessary
+	storageRef := r.storageManager.GetStorageReference(latestRegistry)
+	if storageRef != nil {
+		if latestRegistryStatus.StorageRef == nil || latestRegistryStatus.StorageRef.ConfigMapRef.Name != storageRef.ConfigMapRef.Name {
+			latestRegistryStatus.StorageRef = storageRef
+			hasUpdates = true
+			ctxLogger.Info("Updated StorageRef", "storageRef", storageRef)
+		}
+	}
+
+	// Apply status changes from status manager
+	hasUpdates = statusManager.UpdateStatus(ctx, &latestRegistryStatus) || hasUpdates
+
+	// Single status update using the latest version
+	if hasUpdates {
+		latestRegistry.Status = latestRegistryStatus
+		if err := k8sClient.Status().Update(ctx, latestRegistry); err != nil {
+			ctxLogger.Error(err, "Failed to apply batched status update")
+			return fmt.Errorf("failed to apply batched status update: %w", err)
+		}
+		var syncPhase mcpv1alpha1.SyncPhase
+		if latestRegistryStatus.SyncStatus != nil {
+			syncPhase = latestRegistryStatus.SyncStatus.Phase
+		}
+		var apiPhase string
+		if latestRegistryStatus.APIStatus != nil {
+			apiPhase = string(latestRegistryStatus.APIStatus.Phase)
+		}
+		ctxLogger.V(1).Info("Applied batched status updates",
+			"phase", latestRegistryStatus.Phase,
+			"syncPhase", syncPhase,
+			"apiPhase", apiPhase,
+			"message", latestRegistryStatus.Message,
+			"conditionsCount", len(latestRegistryStatus.Conditions))
+	} else {
+		ctxLogger.V(1).Info("No batched status updates applied")
+	}
+
+	return nil
+}

cmd/thv-operator/pkg/mcpregistrystatus/collector.go

@@ -3,11 +3,9 @@ package mcpregistrystatus
 
 import (
 	"context"
-	"fmt"
 
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
 	mcpv1alpha1 "github.com/stacklok/toolhive/cmd/thv-operator/api/v1alpha1"
@@ -122,58 +120,46 @@ func (s *StatusCollector) SetAPIStatus(phase mcpv1alpha1.APIPhase, message strin
 	s.hasChanges = true
 }
 
-// Apply applies all collected status changes in a single batch update.
-func (s *StatusCollector) Apply(ctx context.Context, k8sClient client.Client) error {
-	if !s.hasChanges {
-		return nil
-	}
+// UpdateStatus applies all collected status changes in a single batch update.
+// Requires the MCPRegistryStatus being the updated version from the cluster
+func (s *StatusCollector) UpdateStatus(ctx context.Context, mcpRegistryStatus *mcpv1alpha1.MCPRegistryStatus) bool {
 
 	ctxLogger := log.FromContext(ctx)
 
-	// Refetch the latest version of the resource to avoid conflicts
-	latestRegistry := &mcpv1alpha1.MCPRegistry{}
-	if err := k8sClient.Get(ctx, client.ObjectKeyFromObject(s.mcpRegistry), latestRegistry); err != nil {
-		ctxLogger.Error(err, "Failed to fetch latest MCPRegistry version for status update")
-		return fmt.Errorf("failed to fetch latest MCPRegistry version: %w", err)
-	}
-
-	// Apply phase change
-	if s.phase != nil {
-		latestRegistry.Status.Phase = *s.phase
-	}
-
-	// Apply message change
-	if s.message != nil {
-		latestRegistry.Status.Message = *s.message
+	if s.hasChanges {
+		// Apply phase change
+		if s.phase != nil {
+			mcpRegistryStatus.Phase = *s.phase
+		}
+
+		// Apply message change
+		if s.message != nil {
+			mcpRegistryStatus.Message = *s.message
+		}
+
+		// Apply sync status change
+		if s.syncStatus != nil {
+			mcpRegistryStatus.SyncStatus = s.syncStatus
+		}
+
+		// Apply API status change
+		if s.apiStatus != nil {
+			mcpRegistryStatus.APIStatus = s.apiStatus
+		}
+
+		// Apply condition changes
+		for _, condition := range s.conditions {
+			meta.SetStatusCondition(&mcpRegistryStatus.Conditions, condition)
+		}
+
+		ctxLogger.V(1).Info("Batched status update applied",
+			"phase", s.phase,
+			"message", s.message,
+			"conditionsCount", len(s.conditions))
+		return true
 	}
-
-	// Apply sync status change
-	if s.syncStatus != nil {
-		latestRegistry.Status.SyncStatus = s.syncStatus
-	}
-
-	// Apply API status change
-	if s.apiStatus != nil {
-		latestRegistry.Status.APIStatus = s.apiStatus
-	}
-
-	// Apply condition changes
-	for _, condition := range s.conditions {
-		meta.SetStatusCondition(&latestRegistry.Status.Conditions, condition)
-	}
-
-	// Single status update using the latest version
-	if err := k8sClient.Status().Update(ctx, latestRegistry); err != nil {
-		ctxLogger.Error(err, "Failed to apply batched status update")
-		return fmt.Errorf("failed to apply batched status update: %w", err)
-	}
-
-	ctxLogger.V(1).Info("Applied batched status update",
-		"phase", s.phase,
-		"message", s.message,
-		"conditionsCount", len(s.conditions))
-
-	return nil
+	ctxLogger.V(1).Info("No batched status update needed")
+	return false
 }
 
 // StatusManager interface methods
@@ -196,6 +182,11 @@ func (s *StatusCollector) SetOverallStatus(phase mcpv1alpha1.MCPRegistryPhase, m
 
 // SyncStatusCollector implementation
 
+// Status returns the sync status
+func (sc *syncStatusCollector) Status() *mcpv1alpha1.SyncStatus {
+	return sc.parent.syncStatus
+}
+
 // SetSyncCondition sets a sync-related condition
 func (sc *syncStatusCollector) SetSyncCondition(condition metav1.Condition) {
 	sc.parent.conditions[condition.Type] = condition
@@ -210,6 +201,11 @@ func (sc *syncStatusCollector) SetSyncStatus(phase mcpv1alpha1.SyncPhase, messag
 
 // APIStatusCollector implementation
 
+// Status returns the API status
+func (ac *apiStatusCollector) Status() *mcpv1alpha1.APIStatus {
+	return ac.parent.apiStatus
+}
+
 // SetAPIStatus delegates to the parent's SetAPIStatus method
 func (ac *apiStatusCollector) SetAPIStatus(phase mcpv1alpha1.APIPhase, message string, endpoint string) {
 	ac.parent.SetAPIStatus(phase, message, endpoint)

cmd/thv-operator/pkg/mcpregistrystatus/collector_test.go

@@ -322,9 +322,8 @@ func TestStatusCollector_Apply(t *testing.T) {
 		t.Parallel()
 		collector := NewStatusManager(registry).(*StatusCollector)
 
-		err := collector.Apply(ctx, k8sClient)
-
-		assert.NoError(t, err)
+		hasUpdates := collector.UpdateStatus(ctx, &registry.Status)
+		assert.False(t, hasUpdates)
 	})
 
 	t.Run("verifies hasChanges behavior", func(t *testing.T) {
@@ -394,7 +393,7 @@ func TestStatusCollector_MultipleConditions(t *testing.T) {
 	assert.Contains(t, collector.conditions, mcpv1alpha1.ConditionAPIReady)
 }
 
-func TestStatusCollector_ApplyErrors(t *testing.T) {
+func TestStatusCollector_NoUpdates(t *testing.T) {
 	t.Parallel()
 
 	ctx := context.Background()
@@ -406,9 +405,6 @@ func TestStatusCollector_ApplyErrors(t *testing.T) {
 	t.Run("error fetching latest registry", func(t *testing.T) {
 		t.Parallel()
 
-		// Create client that will fail on Get
-		k8sClient := fake.NewClientBuilder().WithScheme(scheme).Build()
-
 		// Create collector with registry that doesn't exist in client
 		registry := &mcpv1alpha1.MCPRegistry{
 			ObjectMeta: metav1.ObjectMeta{
@@ -417,12 +413,10 @@ func TestStatusCollector_ApplyErrors(t *testing.T) {
 			},
 		}
 
-		collector := newStatusCollector(registry)
-		collector.SetPhase(mcpv1alpha1.MCPRegistryPhaseReady) // Make some changes
+		collector := newStatusCollector(registry) // No changes
+		hasUpdates := collector.UpdateStatus(ctx, &registry.Status)
+		assert.False(t, hasUpdates)
 
-		err := collector.Apply(ctx, k8sClient)
-		assert.Error(t, err)
-		assert.Contains(t, err.Error(), "failed to fetch latest MCPRegistry version")
 	})
 
 }

cmd/thv-operator/pkg/sync/constants.go → cmd/thv-operator/pkg/mcpregistrystatus/constants.go

@@ -1,4 +1,4 @@
-package sync
+package mcpregistrystatus
 
 const (
 	// SyncTriggerAnnotation is the annotation key used to trigger registry synchronization