Add StorageVersionMigrator controller to clean up storedVersions

cmd/thv-operator/api/v1beta1/embeddingserver_types.go

@@ -211,6 +211,7 @@ const (
 //+kubebuilder:object:root=true
 //+kubebuilder:storageversion
 //+kubebuilder:subresource:status
+//+kubebuilder:metadata:labels=toolhive.stacklok.dev/auto-migrate-storage-version=true
 //+kubebuilder:resource:shortName=emb;embedding,categories=toolhive
 //+kubebuilder:printcolumn:name="Status",type="string",JSONPath=".status.phase"
 //+kubebuilder:printcolumn:name="Model",type="string",JSONPath=".spec.model"

cmd/thv-operator/api/v1beta1/mcpexternalauthconfig_types.go

@@ -1058,6 +1058,7 @@ type MCPExternalAuthConfigStatus struct {
 // +kubebuilder:object:root=true
 // +kubebuilder:storageversion
 // +kubebuilder:subresource:status
+// +kubebuilder:metadata:labels=toolhive.stacklok.dev/auto-migrate-storage-version=true
 // +kubebuilder:resource:shortName=extauth;mcpextauth,categories=toolhive
 // +kubebuilder:printcolumn:name="Type",type=string,JSONPath=`.spec.type`
 // +kubebuilder:printcolumn:name="Valid",type=string,JSONPath=`.status.conditions[?(@.type=='Valid')].status`

cmd/thv-operator/api/v1beta1/mcpgroup_types.go

@@ -88,6 +88,7 @@ const (
 //+kubebuilder:object:root=true
 //+kubebuilder:storageversion
 //+kubebuilder:subresource:status
+//+kubebuilder:metadata:labels=toolhive.stacklok.dev/auto-migrate-storage-version=true
 //+kubebuilder:resource:shortName=mcpg;mcpgroup,categories=toolhive
 //+kubebuilder:printcolumn:name="Servers",type="integer",JSONPath=".status.serverCount"
 //+kubebuilder:printcolumn:name="Phase",type="string",JSONPath=".status.phase"

cmd/thv-operator/api/v1beta1/mcpoidcconfig_types.go

@@ -200,6 +200,7 @@ type MCPOIDCConfigStatus struct {
 // +kubebuilder:object:root=true
 // +kubebuilder:storageversion
 // +kubebuilder:subresource:status
+// +kubebuilder:metadata:labels=toolhive.stacklok.dev/auto-migrate-storage-version=true
 // +kubebuilder:resource:shortName=mcpoidc,categories=toolhive
 // +kubebuilder:printcolumn:name="Source",type=string,JSONPath=`.spec.type`
 // +kubebuilder:printcolumn:name="Valid",type=string,JSONPath=`.status.conditions[?(@.type=='Valid')].status`

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

@@ -205,6 +205,7 @@ const (
 //+kubebuilder:object:root=true
 //+kubebuilder:storageversion
 //+kubebuilder:subresource:status
+//+kubebuilder:metadata:labels=toolhive.stacklok.dev/auto-migrate-storage-version=true
 //+kubebuilder:printcolumn:name="Status",type="string",JSONPath=".status.phase"
 //+kubebuilder:printcolumn:name="Ready",type="string",JSONPath=".status.conditions[?(@.type=='Ready')].status"
 //+kubebuilder:printcolumn:name="Replicas",type="integer",JSONPath=".status.readyReplicas"

cmd/thv-operator/api/v1beta1/mcpremoteproxy_types.go

@@ -348,6 +348,7 @@ const (
 //+kubebuilder:object:root=true
 //+kubebuilder:storageversion
 //+kubebuilder:subresource:status
+//+kubebuilder:metadata:labels=toolhive.stacklok.dev/auto-migrate-storage-version=true
 //+kubebuilder:resource:shortName=rp;mcprp,categories=toolhive
 //+kubebuilder:printcolumn:name="Phase",type="string",JSONPath=".status.phase"
 //+kubebuilder:printcolumn:name="Remote URL",type="string",JSONPath=".spec.remoteUrl"

cmd/thv-operator/api/v1beta1/mcpserver_types.go

@@ -884,6 +884,7 @@ const (
 //+kubebuilder:object:root=true
 //+kubebuilder:storageversion
 //+kubebuilder:subresource:status
+//+kubebuilder:metadata:labels=toolhive.stacklok.dev/auto-migrate-storage-version=true
 //+kubebuilder:resource:shortName=mcpserver;mcpservers,categories=toolhive
 //+kubebuilder:printcolumn:name="Status",type="string",JSONPath=".status.phase"
 //+kubebuilder:printcolumn:name="Ready",type="string",JSONPath=".status.conditions[?(@.type=='Ready')].status"

cmd/thv-operator/api/v1beta1/mcpserverentry_types.go

@@ -164,6 +164,7 @@ const (
 //+kubebuilder:object:root=true
 //+kubebuilder:storageversion
 //+kubebuilder:subresource:status
+//+kubebuilder:metadata:labels=toolhive.stacklok.dev/auto-migrate-storage-version=true
 //+kubebuilder:resource:shortName=mcpentry,categories=toolhive
 //+kubebuilder:printcolumn:name="Phase",type="string",JSONPath=".status.phase"
 //+kubebuilder:printcolumn:name="Transport",type="string",JSONPath=".spec.transport"

cmd/thv-operator/api/v1beta1/mcptelemetryconfig_types.go

@@ -139,6 +139,7 @@ type MCPTelemetryConfigStatus struct {
 // +kubebuilder:object:root=true
 // +kubebuilder:storageversion
 // +kubebuilder:subresource:status
+// +kubebuilder:metadata:labels=toolhive.stacklok.dev/auto-migrate-storage-version=true
 // +kubebuilder:resource:shortName=mcpotel,categories=toolhive
 // +kubebuilder:printcolumn:name="Endpoint",type=string,JSONPath=`.spec.openTelemetry.endpoint`
 // +kubebuilder:printcolumn:name="Valid",type=string,JSONPath=`.status.conditions[?(@.type=='Valid')].status`

cmd/thv-operator/api/v1beta1/toolconfig_types.go

@@ -112,6 +112,7 @@ type MCPToolConfigStatus struct {
 // +kubebuilder:object:root=true
 // +kubebuilder:storageversion
 // +kubebuilder:subresource:status
+// +kubebuilder:metadata:labels=toolhive.stacklok.dev/auto-migrate-storage-version=true
 // +kubebuilder:resource:shortName=tc;toolconfig,categories=toolhive
 // +kubebuilder:printcolumn:name="Valid",type=string,JSONPath=`.status.conditions[?(@.type=='Valid')].status`
 // +kubebuilder:printcolumn:name="References",type=integer,JSONPath=`.status.referenceCount`

cmd/thv-operator/api/v1beta1/virtualmcpcompositetooldefinition_types.go

@@ -100,6 +100,7 @@ const (
 //+kubebuilder:object:root=true
 //+kubebuilder:storageversion
 //+kubebuilder:subresource:status
+//+kubebuilder:metadata:labels=toolhive.stacklok.dev/auto-migrate-storage-version=true
 //+kubebuilder:resource:shortName=vmcpctd;compositetool,categories=toolhive
 //+kubebuilder:printcolumn:name="Workflow",type="string",JSONPath=".spec.name",description="Workflow name"
 //+kubebuilder:printcolumn:name="Steps",type="integer",JSONPath=".spec.steps[*]",description="Number of steps"

cmd/thv-operator/api/v1beta1/virtualmcpserver_types.go

@@ -464,6 +464,7 @@ const (
 //+kubebuilder:object:root=true
 //+kubebuilder:storageversion
 //+kubebuilder:subresource:status
+//+kubebuilder:metadata:labels=toolhive.stacklok.dev/auto-migrate-storage-version=true
 //+kubebuilder:resource:shortName=vmcp;virtualmcp,categories=toolhive
 //+kubebuilder:printcolumn:name="Phase",type="string",JSONPath=".status.phase",description="The phase of the VirtualMCPServer"
 //+kubebuilder:printcolumn:name="URL",type="string",JSONPath=".status.url",description="Virtual MCP server URL"

cmd/thv-operator/app/app.go

@@ -12,9 +12,11 @@
 	"fmt"
 	"log/slog"
 	"os"
+	"strconv"
 	"strings"
 
 	"github.com/go-logr/logr"
+	apiextensionsv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
@@ -42,10 +44,20 @@
 	setupLog = log.Log.WithName("setup")
 )
 
+// envEnableStorageVersionMigrator is the operator-wide opt-out for the
+// StorageVersionMigrator controller. Unlike the other controllers — which are
+// always-on after the operator-wide feature flag system was removed — the
+// migrator retains a feature flag because admins running an external
+// kube-storage-version-migrator deployment need a way to disable the in-operator
+// controller. Setting this env var to a falsey value ("false", "0", "f")
+// skips registering the migrator with the manager.
+const envEnableStorageVersionMigrator = "ENABLE_STORAGE_VERSION_MIGRATOR"
+
 func init() {
 	utilruntime.Must(clientgoscheme.AddToScheme(scheme))
 	utilruntime.Must(mcpv1alpha1.AddToScheme(scheme))
 	utilruntime.Must(mcpv1beta1.AddToScheme(scheme))
+	utilruntime.Must(apiextensionsv1.AddToScheme(scheme))
 	//+kubebuilder:scaffold:scheme
 }
 
@@ -150,10 +162,53 @@
 	if err := setupAggregationControllers(mgr, imagePullSecretsDefaults); err != nil {
 		return err
 	}
+	if isStorageVersionMigratorEnabled() {
+		if err := setupStorageVersionMigrator(mgr); err != nil {
+			return err
+		}
+	} else {
+		setupLog.Info(envEnableStorageVersionMigrator + " is disabled, skipping StorageVersionMigrator controller")
+	}
 	//+kubebuilder:scaffold:builder
 	return nil
 }
 
+// setupStorageVersionMigrator wires the StorageVersionMigrator controller into
+// the manager. The controller reconciles status.storedVersions on opted-in
+// toolhive.stacklok.dev CRDs so a future operator release can drop deprecated
+// versions from spec.versions without orphaning etcd objects.
+func setupStorageVersionMigrator(mgr ctrl.Manager) error {
+	if err := (&controllers.StorageVersionMigratorReconciler{
+		Client:    mgr.GetClient(),
+		APIReader: mgr.GetAPIReader(),
+		Scheme:    mgr.GetScheme(),
+		Recorder:  mgr.GetEventRecorderFor("storageversionmigrator-controller"),
+	}).SetupWithManager(mgr); err != nil {
+		return fmt.Errorf("unable to create controller StorageVersionMigrator: %w", err)
+	}
+	return nil
+}
+
+// isStorageVersionMigratorEnabled reports whether the StorageVersionMigrator
+// controller should be registered. Defaults to true; can be disabled by setting
+// ENABLE_STORAGE_VERSION_MIGRATOR to a falsey value. An unparseable value logs
+// a warning and uses the default.
+func isStorageVersionMigratorEnabled() bool {
+	value, found := os.LookupEnv(envEnableStorageVersionMigrator)
+	if !found {
+		return true
+	}
+	enabled, err := strconv.ParseBool(value)
+	if err != nil {
+		setupLog.Info(
+			"invalid boolean value for "+envEnableStorageVersionMigrator+", using default (enabled)",
+			"value", value,
+		)
+		return true
+	}
+	return enabled
+}
+
 // setupGroupRefFieldIndexes sets up field indexing for spec.groupRef on all resource types
 // that can reference an MCPGroup. This enables efficient lookups by groupRef in controllers.
 func setupGroupRefFieldIndexes(mgr ctrl.Manager) error {

cmd/thv-operator/controllers/marker_coverage_test.go

@@ -0,0 +1,151 @@
+// SPDX-FileCopyrightText: Copyright 2025 Stacklok, Inc.
+// SPDX-License-Identifier: Apache-2.0
+
+package controllers
+
+import (
+	"bufio"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestV1beta1TypesMarkerCoverage guards against a subtle failure mode in the
+// opt-in-label design for storage-version migration: if a new CRD root type is
+// added to cmd/thv-operator/api/v1beta1/ without the
+//
+//	+kubebuilder:metadata:labels=toolhive.stacklok.dev/auto-migrate-storage-version=true
+//
+// marker, the StorageVersionMigrator controller silently excludes it from
+// reconciliation. The problem surfaces only when a future release tries to
+// drop a deprecated version — at which point it is far too late.
+//
+// The test scans every root type (marker block contains
+// +kubebuilder:object:root=true AND +kubebuilder:storageversion) and fails if
+// any lacks either the migrate marker or an explicit
+// +thv:storage-version-migrator:exclude sibling marker. List types
+// (+kubebuilder:object:root=true WITHOUT +kubebuilder:storageversion) are
+// intentionally skipped — CRD-level labels are keyed on the root type, not
+// the list type.
+func TestV1beta1TypesMarkerCoverage(t *testing.T) {
+	t.Parallel()
+
+	typesDir := filepath.Join("..", "api", "v1beta1")
+	entries, err := os.ReadDir(typesDir)
+	require.NoError(t, err, "reading %s", typesDir)
+
+	const (
+		rootMarker         = "+kubebuilder:object:root=true"
+		storageMarker      = "+kubebuilder:storageversion"
+		migrateMarker      = "+kubebuilder:metadata:labels=toolhive.stacklok.dev/auto-migrate-storage-version=true"
+		excludeMarker      = "+thv:storage-version-migrator:exclude"
+		groupversionInfoGo = "groupversion_info.go"
+		zzGeneratedPrefix  = "zz_generated"
+		suffixTypesGo      = "_types.go"
+	)
+
+	type rootType struct {
+		file        string
+		typeName    string
+		markerBlock []string
+	}
+
+	var roots []rootType
+
+	for _, e := range entries {
+		if e.IsDir() {
+			continue
+		}
+		name := e.Name()
+		if !strings.HasSuffix(name, suffixTypesGo) {
+			continue
+		}
+		if name == groupversionInfoGo || strings.HasPrefix(name, zzGeneratedPrefix) {
+			continue
+		}
+
+		path := filepath.Join(typesDir, name)
+		f, err := os.Open(path)
+		require.NoError(t, err, "open %s", path)
+
+		scanner := bufio.NewScanner(f)
+		// Raise the max token size — some of the *_types.go files have very
+		// long comment lines (printcolumn JSONPath expressions).
+		scanner.Buffer(make([]byte, 64*1024), 1024*1024)
+
+		var block []string
+		for scanner.Scan() {
+			line := strings.TrimSpace(scanner.Text())
+
+			switch {
+			case strings.HasPrefix(line, "//"):
+				// Accumulate every comment/marker line. kubebuilder convention
+				// often separates markers from the godoc comment with a blank
+				// line, so we keep the block alive across blanks and only
+				// reset on a non-comment, non-blank, non-type line.
+				block = append(block, strings.TrimSpace(strings.TrimPrefix(line, "//")))
+			case strings.HasPrefix(line, "type "):
+				// Only root-level types matter (must contain object:root=true
+				// AND storageversion markers). List types have only object:root=true.
+				if containsMarker(block, rootMarker) && containsMarker(block, storageMarker) {
+					typeName := extractTypeName(line)
+					if typeName != "" {
+						copied := append([]string(nil), block...)
+						roots = append(roots, rootType{file: name, typeName: typeName, markerBlock: copied})
+					}
+				}
+				block = nil
+			case line == "":
+				// Blank line — keep block alive (comment-then-blank-then-type
+				// is idiomatic Go + kubebuilder).
+			default:
+				// Anything else (e.g. struct body, package clause, import) —
+				// drop any in-flight comment block.
+				block = nil
+			}
+		}
+		require.NoError(t, scanner.Err(), "scan %s", path)
+		require.NoError(t, f.Close())
+	}
+
+	require.NotEmpty(t, roots,
+		"no v1beta1 root types found — scanner likely broken; this test is meaningless without coverage")
+
+	for _, r := range roots {
+		hasMigrate := containsMarker(r.markerBlock, migrateMarker)
+		hasExclude := containsMarker(r.markerBlock, excludeMarker)
+		assert.Truef(t, hasMigrate || hasExclude,
+			"v1beta1 root type %s.%s is missing either\n"+
+				"  %s\n"+
+				"(opt in to storage-version migration) or\n"+
+				"  %s\n"+
+				"(explicit opt-out). Every root type must declare one. See\n"+
+				"cmd/thv-operator/controllers/storageversionmigrator_controller.go for context.",
+			r.file, r.typeName, migrateMarker, excludeMarker)
+	}
+}
+
+// containsMarker returns true if any line in block contains the given
+// marker substring.
+func containsMarker(block []string, marker string) bool {
+	for _, l := range block {
+		if strings.Contains(l, marker) {
+			return true
+		}
+	}
+	return false
+}
+
+// extractTypeName returns the identifier in a line of the form `type Foo struct {`.
+// Returns empty string if the line is not a type declaration we care about.
+func extractTypeName(line string) string {
+	fields := strings.Fields(line)
+	if len(fields) < 2 || fields[0] != "type" {
+		return ""
+	}
+	return fields[1]
+}