feat: fail loudly, recover automatically

.github/workflows/e2e.yml

@@ -10,7 +10,7 @@ on:
       suite:
         description: Which suite(s) to run
         type: choice
-        options: [default, devserver, workflowrun, mtls, upgrade, opensearch, cassandra, multicluster, all]
+        options: [default, devserver, workflowrun, mtls, upgrade, upgrade-stall, opensearch, cassandra, multicluster, all]
         default: default
 
 permissions:
@@ -35,21 +35,23 @@ jobs:
           workflowrun='{"suite":"workflowrun"}'
           mtls='{"temporal":"1.31.1","persistence":"mtls","suite":"mtls"}'
           upgrade='{"temporal":"1.30.0","persistence":"upgrade","suite":"upgrade"}'
+          upgradestall='{"temporal":"1.30.4","persistence":"upgrade","suite":"upgrade-stall"}'
           opensearch='{"temporal":"1.31.1","persistence":"opensearch","suite":"opensearch"}'
           cassandra='{"temporal":"1.31.1","persistence":"cassandra","suite":"cassandra"}'
           multicluster='{"temporal":"1.31.1","persistence":"multicluster","suite":"multicluster"}'
           if [ "$EVENT" = "schedule" ]; then
-            echo "combos=[$postgres,$devserver,$workflowrun,$mtls,$upgrade,$opensearch,$cassandra,$multicluster]" >> "$GITHUB_OUTPUT"
+            echo "combos=[$postgres,$devserver,$workflowrun,$mtls,$upgrade,$upgradestall,$opensearch,$cassandra,$multicluster]" >> "$GITHUB_OUTPUT"
           elif [ "$EVENT" = "workflow_dispatch" ]; then
             case "$SUITE" in
               devserver)    echo "combos=[$devserver]" >> "$GITHUB_OUTPUT" ;;
               workflowrun)  echo "combos=[$workflowrun]" >> "$GITHUB_OUTPUT" ;;
               mtls)         echo "combos=[$mtls]" >> "$GITHUB_OUTPUT" ;;
               upgrade)      echo "combos=[$upgrade]" >> "$GITHUB_OUTPUT" ;;
+              upgrade-stall) echo "combos=[$upgradestall]" >> "$GITHUB_OUTPUT" ;;
               opensearch)   echo "combos=[$opensearch]" >> "$GITHUB_OUTPUT" ;;
               cassandra)    echo "combos=[$cassandra]" >> "$GITHUB_OUTPUT" ;;
               multicluster) echo "combos=[$multicluster]" >> "$GITHUB_OUTPUT" ;;
-              all)          echo "combos=[$postgres,$devserver,$workflowrun,$mtls,$upgrade,$opensearch,$cassandra,$multicluster]" >> "$GITHUB_OUTPUT" ;;
+              all)          echo "combos=[$postgres,$devserver,$workflowrun,$mtls,$upgrade,$upgradestall,$opensearch,$cassandra,$multicluster]" >> "$GITHUB_OUTPUT" ;;
               *)            echo "combos=[$postgres,$devserver]" >> "$GITHUB_OUTPUT" ;;
             esac
           else
@@ -95,11 +97,19 @@ jobs:
 
       - name: Install the operator via Helm
         run: |
+          extra_args=()
+          if [ "${{ matrix.combo.suite }}" = "upgrade-stall" ]; then
+            extra_args=(
+              --set-string 'manager.args[0]=--leader-elect'
+              --set-string 'manager.args[1]=--upgrade-phase-timeout=1m'
+            )
+          fi
           helm install temporal-operator dist/chart \
             --namespace temporal-system --create-namespace \
             --set manager.image.repository=temporal-operator \
             --set manager.image.tag=e2e \
-            --set manager.image.pullPolicy=Never
+            --set manager.image.pullPolicy=Never \
+            "${extra_args[@]}"
           kubectl -n temporal-system rollout status deploy/temporal-operator-controller-manager --timeout=180s
 
       # Pre-pull the Temporal images on the runner and side-load them into the

Makefile

@@ -63,7 +63,7 @@ vet: ## Run go vet against code.
 
 .PHONY: test
 test: manifests generate fmt vet setup-envtest ## Run tests.
-	KUBEBUILDER_ASSETS="$(shell "$(ENVTEST)" use $(ENVTEST_K8S_VERSION) --bin-dir "$(LOCALBIN)" -p path)" go test $$(go list ./... | grep -v /e2e) -coverprofile cover.out
+	KUBEBUILDER_ASSETS="$(shell "$(ENVTEST)" use $(ENVTEST_K8S_VERSION) --bin-dir "$(LOCALBIN)" -p path)" go test -race $$(go list ./... | grep -v /e2e) -coverprofile cover.out
 
 .PHONY: test-golden-update
 test-golden-update: ## Regenerate config-template golden files.

api/v1alpha1/accessors.go

@@ -0,0 +1,51 @@
+/*
+Copyright 2026 Brian Morton.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1alpha1
+
+import metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+// This file provides the uniform status accessors that internal/status uses to
+// operate on any Temporal CRD without type switches. Keep one pair of methods
+// per API type; there is intentionally no reflection here.
+
+func (t *TemporalCluster) GetConditions() *[]metav1.Condition { return &t.Status.Conditions }
+func (t *TemporalCluster) SetObservedGeneration(g int64)      { t.Status.ObservedGeneration = g }
+
+func (t *TemporalClusterClient) GetConditions() *[]metav1.Condition { return &t.Status.Conditions }
+func (t *TemporalClusterClient) SetObservedGeneration(g int64)      { t.Status.ObservedGeneration = g }
+
+func (t *TemporalClusterConnection) GetConditions() *[]metav1.Condition { return &t.Status.Conditions }
+func (t *TemporalClusterConnection) SetObservedGeneration(g int64) {
+	t.Status.ObservedGeneration = g
+}
+
+func (t *TemporalDevServer) GetConditions() *[]metav1.Condition { return &t.Status.Conditions }
+func (t *TemporalDevServer) SetObservedGeneration(g int64)      { t.Status.ObservedGeneration = g }
+
+func (t *TemporalNamespace) GetConditions() *[]metav1.Condition { return &t.Status.Conditions }
+func (t *TemporalNamespace) SetObservedGeneration(g int64)      { t.Status.ObservedGeneration = g }
+
+func (t *TemporalSchedule) GetConditions() *[]metav1.Condition { return &t.Status.Conditions }
+func (t *TemporalSchedule) SetObservedGeneration(g int64)      { t.Status.ObservedGeneration = g }
+
+func (t *TemporalSearchAttribute) GetConditions() *[]metav1.Condition { return &t.Status.Conditions }
+func (t *TemporalSearchAttribute) SetObservedGeneration(g int64) {
+	t.Status.ObservedGeneration = g
+}
+
+func (t *TemporalWorkflowRun) GetConditions() *[]metav1.Condition { return &t.Status.Conditions }
+func (t *TemporalWorkflowRun) SetObservedGeneration(g int64)      { t.Status.ObservedGeneration = g }

api/v1alpha1/accessors_test.go

@@ -0,0 +1,56 @@
+/*
+Copyright 2026 Brian Morton.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package v1alpha1
+
+import (
+	"testing"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// conditionAccessor is the contract internal/status depends on.
+type conditionAccessor interface {
+	GetConditions() *[]metav1.Condition
+	SetObservedGeneration(int64)
+}
+
+func TestAllTypesImplementConditionAccessor(t *testing.T) {
+	objs := map[string]conditionAccessor{
+		"TemporalCluster":           &TemporalCluster{},
+		"TemporalClusterClient":     &TemporalClusterClient{},
+		"TemporalClusterConnection": &TemporalClusterConnection{},
+		"TemporalDevServer":         &TemporalDevServer{},
+		"TemporalNamespace":         &TemporalNamespace{},
+		"TemporalSchedule":          &TemporalSchedule{},
+		"TemporalSearchAttribute":   &TemporalSearchAttribute{},
+		"TemporalWorkflowRun":       &TemporalWorkflowRun{},
+	}
+
+	for name, obj := range objs {
+		t.Run(name, func(t *testing.T) {
+			conds := obj.GetConditions()
+			if conds == nil {
+				t.Fatal("GetConditions returned nil pointer")
+			}
+			*conds = append(*conds, metav1.Condition{Type: "Ready"})
+			if len(*obj.GetConditions()) != 1 {
+				t.Error("GetConditions does not return a live pointer into status")
+			}
+			obj.SetObservedGeneration(7)
+		})
+	}
+}

api/v1alpha1/conditions.go

@@ -87,4 +87,21 @@ const (
 	// ReasonFrontendUnavailable indicates the Temporal frontend is reachable
 	// but not yet accepting RPCs (transient startup window).
 	ReasonFrontendUnavailable = "FrontendUnavailable"
+	// ReasonUpgradeStalled indicates an upgrade phase exceeded its timeout.
+	ReasonUpgradeStalled = "UpgradeStalled"
+	// ReasonRolloutStalled indicates a service Deployment did not roll out in time.
+	ReasonRolloutStalled = "RolloutStalled"
+	// ReasonUpgradeProgressing indicates an upgrade is advancing normally.
+	ReasonUpgradeProgressing = "UpgradeProgressing"
+	// ReasonSchemaMigrationFailed indicates schema migration failed and the
+	// retry budget is exhausted.
+	ReasonSchemaMigrationFailed = "SchemaMigrationFailed"
+	// ReasonSchemaMigrationRetrying indicates a failed schema migration will be retried.
+	ReasonSchemaMigrationRetrying = "SchemaMigrationRetrying"
+	// ReasonCleanupAbandoned indicates remote cleanup was abandoned after the
+	// cleanup deadline elapsed with the target unreachable.
+	ReasonCleanupAbandoned = "CleanupAbandoned"
+	// ReasonCleanupPending indicates remote cleanup is being retried while
+	// waiting for the target to become reachable.
+	ReasonCleanupPending = "CleanupPending"
 )

api/v1alpha1/persistence_types.go

@@ -237,6 +237,12 @@ type PersistenceStatus struct {
 	// Reachable indicates whether the datastores were reachable at last reconcile.
 	// +optional
 	Reachable bool `json:"reachable,omitempty"`
+
+	// SchemaAttempts records failed schema-migration attempts per store, so
+	// recreation is bounded across operator restarts and leader-election
+	// failover. Entries are removed when a store's migration succeeds.
+	// +optional
+	SchemaAttempts map[string]SchemaAttemptStatus `json:"schemaAttempts,omitempty"`
 }
 
 // SchemaUpgradeRecord records a single schema migration.
@@ -246,3 +252,17 @@ type SchemaUpgradeRecord struct {
 	ToVersion   string      `json:"toVersion"`
 	Time        metav1.Time `json:"time"`
 }
+
+// SchemaAttemptStatus counts consecutive failed schema-migration attempts for a
+// single store.
+type SchemaAttemptStatus struct {
+	// Count is the number of attempts that have failed.
+	// +optional
+	Count int32 `json:"count,omitempty"`
+	// FirstFailedAt is when the first of the current run of failures occurred.
+	// +optional
+	FirstFailedAt *metav1.Time `json:"firstFailedAt,omitempty"`
+	// LastError is the failure reason reported by the most recent Job.
+	// +optional
+	LastError string `json:"lastError,omitempty"`
+}

api/v1alpha1/temporalcluster_types.go

@@ -246,6 +246,17 @@ type UpgradeStatus struct {
 	Rollbackable bool `json:"rollbackable,omitempty"`
 	// +optional
 	StartedAt *metav1.Time `json:"startedAt,omitempty"`
+	// PhaseStartedAt is when the current phase was entered. It is the basis for
+	// stall detection.
+	// +optional
+	PhaseStartedAt *metav1.Time `json:"phaseStartedAt,omitempty"`
+	// StalledService names the service whose rollout has exceeded the phase
+	// timeout, or is empty when the upgrade is progressing normally.
+	// +optional
+	StalledService string `json:"stalledService,omitempty"`
+	// Message explains why the upgrade is not progressing, when it is not.
+	// +optional
+	Message string `json:"message,omitempty"`
 }
 
 // +kubebuilder:object:root=true

cmd/main.go

@@ -38,10 +38,13 @@ import (
 
 	temporalv1alpha1 "github.com/bmorton/temporal-operator/api/v1alpha1"
 	"github.com/bmorton/temporal-operator/internal/controller"
+	opevents "github.com/bmorton/temporal-operator/internal/events"
+	"github.com/bmorton/temporal-operator/internal/metrics"
 	"github.com/bmorton/temporal-operator/internal/ui"
 	webhookv1alpha1 "github.com/bmorton/temporal-operator/internal/webhook/v1alpha1"
 
 	certmanagerv1 "github.com/cert-manager/cert-manager/pkg/apis/certmanager/v1"
+	ctrlmetrics "sigs.k8s.io/controller-runtime/pkg/metrics"
 	// +kubebuilder:scaffold:imports
 )
 
@@ -99,12 +102,16 @@ func main() {
 	flag.StringVar(&uiBasePath, "ui-base-path", "/", "URL base path the UI is served under.")
 	flag.BoolVar(&uiRequireAuth, "ui-require-auth", false,
 		"Require a trusted forward-auth user header; return 401 when absent.")
+	var upgradePhaseTimeout time.Duration
+	flag.DurationVar(&upgradePhaseTimeout, "upgrade-phase-timeout", 15*time.Minute,
+		"How long a single upgrade phase may run before it is reported as stalled.")
 	opts := zap.Options{
 		Development: true,
 	}
 	opts.BindFlags(flag.CommandLine)
 	flag.Parse()
 
+	controller.SetUpgradePhaseTimeout(upgradePhaseTimeout)
 	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&opts)))
 
 	// if the enable-http2 flag is false (the default), http/2 should be disabled
@@ -198,10 +205,14 @@ func main() {
 		os.Exit(1)
 	}
 
+	metrics.Register()
+	ctrlmetrics.Registry.MustRegister(metrics.NewConditionCollector(mgr.GetClient()))
+
 	if err := (&controller.TemporalClusterReconciler{
 		Client:        mgr.GetClient(),
 		Scheme:        mgr.GetScheme(),
 		Recorder:      mgr.GetEventRecorder("temporalcluster-controller"),
+		Events:        opevents.New(mgr.GetEventRecorder("temporalcluster-controller")),
 		OperatorImage: os.Getenv("OPERATOR_IMAGE"),
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "TemporalCluster")
@@ -210,48 +221,55 @@ func main() {
 	if err := (&controller.TemporalClusterClientReconciler{
 		Client: mgr.GetClient(),
 		Scheme: mgr.GetScheme(),
+		Events: opevents.New(mgr.GetEventRecorder("temporalclusterclient-controller")),
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "TemporalClusterClient")
 		os.Exit(1)
 	}
 	if err := (&controller.TemporalNamespaceReconciler{
 		Client: mgr.GetClient(),
 		Scheme: mgr.GetScheme(),
+		Events: opevents.New(mgr.GetEventRecorder("temporalnamespace-controller")),
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "TemporalNamespace")
 		os.Exit(1)
 	}
 	if err := (&controller.TemporalSearchAttributeReconciler{
 		Client: mgr.GetClient(),
 		Scheme: mgr.GetScheme(),
+		Events: opevents.New(mgr.GetEventRecorder("temporalsearchattribute-controller")),
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "TemporalSearchAttribute")
 		os.Exit(1)
 	}
 	if err := (&controller.TemporalScheduleReconciler{
 		Client: mgr.GetClient(),
 		Scheme: mgr.GetScheme(),
+		Events: opevents.New(mgr.GetEventRecorder("temporalschedule-controller")),
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "TemporalSchedule")
 		os.Exit(1)
 	}
 	if err := (&controller.TemporalDevServerReconciler{
 		Client: mgr.GetClient(),
 		Scheme: mgr.GetScheme(),
+		Events: opevents.New(mgr.GetEventRecorder("temporaldevserver-controller")),
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "TemporalDevServer")
 		os.Exit(1)
 	}
 	if err := (&controller.TemporalClusterConnectionReconciler{
 		Client: mgr.GetClient(),
 		Scheme: mgr.GetScheme(),
+		Events: opevents.New(mgr.GetEventRecorder("temporalclusterconnection-controller")),
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "TemporalClusterConnection")
 		os.Exit(1)
 	}
 	if err := (&controller.TemporalWorkflowRunReconciler{
 		Client: mgr.GetClient(),
 		Scheme: mgr.GetScheme(),
+		Events: opevents.New(mgr.GetEventRecorder("temporalworkflowrun-controller")),
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "TemporalWorkflowRun")
 		os.Exit(1)