K8SPS-852 errantTransactionsPolicy: report quarantined members NotReady

api/v1/perconaservermysql_types.go

@@ -227,6 +227,22 @@ const (
 	ClusterTypeAsync ClusterType = "async"
 )
 
+// ErrantTransactionsPolicy defines how the operator handles an async replica
+// holding transactions that were never replicated to the current primary.
+type ErrantTransactionsPolicy string
+
+const (
+	// ErrantTransactionsManual leaves resolution to the user (default).
+	ErrantTransactionsManual ErrantTransactionsPolicy = "manual"
+	// ErrantTransactionsRebuild discards the unreplicated data by deleting
+	// the member's pod and PVC; it re-provisions from the current primary.
+	ErrantTransactionsRebuild ErrantTransactionsPolicy = "rebuild"
+	// ErrantTransactionsInjectEmpty reconciles GTID sets via Orchestrator's
+	// gtid-errant-inject-empty and rejoins the member, keeping its extra
+	// rows locally.
+	ErrantTransactionsInjectEmpty ErrantTransactionsPolicy = "inject-empty"
+)
+
 const (
 	MinSafeProxySize = 2
 	MinSafeGRSize    = 3
@@ -255,6 +271,19 @@ type MySQLSpec struct {
 	Expose        ServiceExposeTogglable `json:"expose,omitempty"`
 	AutoRecovery  bool                   `json:"autoRecovery,omitempty"`
 
+	// ErrantTransactionsPolicy controls what the operator does with an async
+	// replica that holds errant GTIDs (transactions never replicated to the
+	// current primary, typically a former primary returning after failover):
+	// `manual` (default) - emit an event and do nothing; the user resolves it.
+	// `rebuild` - discard the unreplicated data: delete the member's pod and
+	// PVC so it re-provisions by cloning the current primary.
+	// `inject-empty` - keep the member's extra rows locally: reconcile the
+	// GTID sets via Orchestrator's gtid-errant-inject-empty and rejoin. The
+	// diverged rows remain on that member only.
+	// +kubebuilder:validation:Enum=manual;rebuild;inject-empty
+	// +kubebuilder:default=manual
+	ErrantTransactionsPolicy ErrantTransactionsPolicy `json:"errantTransactionsPolicy,omitempty"`
+
 	Sidecars       []corev1.Container `json:"sidecars,omitempty"`
 	SidecarVolumes []corev1.Volume    `json:"sidecarVolumes,omitempty"`
 	SidecarPVCs    []SidecarPVC       `json:"sidecarPVCs,omitempty"`

api/v1/perconaservermysql_types_test.go

@@ -133,6 +133,8 @@ func TestCheckNSetDefaults(t *testing.T) {
 		assert.EqualError(t, err, "ASYNC_SOURCE_CONNECT_RETRY should be a positive value")
 	})
 	t.Run("backups disabled without image should succeed", func(t *testing.T) {
+		// the xtrabackup sidecar is not deployed when backups are disabled,
+		// so backup.image is not required in that case.
 		cr := new(PerconaServerMySQL)
 		cr.Spec.Backup = &BackupSpec{
 			Enabled: false,
@@ -150,6 +152,25 @@ func TestCheckNSetDefaults(t *testing.T) {
 		err := cr.CheckNSetDefaults(t.Context(), nil)
 		assert.NoError(t, err)
 	})
+	t.Run("backups disabled with image should succeed", func(t *testing.T) {
+		cr := new(PerconaServerMySQL)
+		cr.Spec.Backup = &BackupSpec{
+			Enabled: false,
+			Image:   "backup-image",
+		}
+		cr.Spec.MySQL.VolumeSpec = &VolumeSpec{
+			PersistentVolumeClaim: &corev1.PersistentVolumeClaimSpec{
+				Resources: corev1.VolumeResourceRequirements{
+					Requests: corev1.ResourceList{
+						corev1.ResourceStorage: resource.MustParse("1G"),
+					},
+				},
+			},
+		}
+
+		err := cr.CheckNSetDefaults(t.Context(), nil)
+		assert.NoError(t, err)
+	})
 	t.Run("without backup image, with volume spec", func(t *testing.T) {
 		cr := new(PerconaServerMySQL)
 		cr.Spec.MySQL.VolumeSpec = &VolumeSpec{

cmd/bootstrap/async/async_replication.go

@@ -118,7 +118,19 @@ func Bootstrap(ctx context.Context) error {
 		return err
 	}
 
+	readOnly, err := db.IsReadonly(ctx)
+	if err != nil {
+		return errors.Wrap(err, "check read only status")
+	}
+
 	switch {
+	case !readOnly:
+		if err := db.ResetReplication(ctx); err != nil {
+			return err
+		}
+
+		log.Printf("I'm writable and therefore the primary.")
+		return nil
 	case donor == "":
 		if err := db.ResetReplication(ctx); err != nil {
 			return err
@@ -148,6 +160,24 @@ func Bootstrap(ctx context.Context) error {
 		return errors.Wrap(err, "check if clone is required")
 	}
 
+	if requireClone {
+		// Never clone over a datadir that already executed transactions:
+		// a former primary returning after a failover has no clone.lock but
+		// may hold writes that were never replicated; cloning destroys them.
+		gtidExecuted, err := db.GetGTIDExecuted(ctx)
+		if err != nil {
+			return errors.Wrap(err, "get gtid_executed")
+		}
+		if gtidExecuted != "" {
+			log.Printf("Datadir has executed GTIDs (%s), skipping clone to preserve local data", gtidExecuted)
+			requireClone = false
+
+			if err := createCloneLock(cloneLock); err != nil {
+				return errors.Wrap(err, "create clone lock")
+			}
+		}
+	}
+
 	log.Printf("Clone required: %t", requireClone)
 	if requireClone {
 		log.Println("Checking if a clone in progress")
@@ -196,6 +226,40 @@ func Bootstrap(ctx context.Context) error {
 	}
 
 	if rStatus == mysqldb.ReplicationStatusNotInitiated || rStatus == mysqldb.ReplicationStatusStopped {
+		// Joining is only safe when this member has no transactions the
+		// primary doesn't know about.
+		errant, err := errantGTIDs(ctx, db, primaryIp)
+		if err != nil {
+			return errors.Wrap(err, "check errant GTIDs")
+		}
+		if errant != "" {
+			// Quarantine only against a confirmed (writable) primary. After
+			// a full cluster restart nobody is writable and the topology
+			// guess is unreliable; leave the member unjoined and let the
+			// operator resolve it.
+			primaryWritable, err := isPrimaryWritable(ctx, primaryIp)
+			if err != nil {
+				return errors.Wrapf(err, "check if primary %s is writable", primary)
+			}
+			if !primaryWritable {
+				log.Printf("Local transactions not present on presumed primary %s (errant GTIDs: %s), "+
+					"but the presumed primary is not writable; leaving the member unjoined for the operator to resolve", primary, errant)
+				if err := db.EnableSuperReadonly(ctx); err != nil {
+					return errors.Wrap(err, "enable super read only")
+				}
+				return nil
+			}
+
+			log.Printf("QUARANTINE: local transactions not present on primary %s (errant GTIDs: %s); refusing to join the cluster", primary, errant)
+			if err := os.WriteFile(mysql.QuarantineFile, []byte(errant+"\n"), 0o640); err != nil {
+				return errors.Wrap(err, "create quarantine file")
+			}
+			if err := db.EnableSuperReadonly(ctx); err != nil {
+				return errors.Wrap(err, "enable super read only")
+			}
+			return nil
+		}
+
 		log.Println("configuring replication")
 
 		replicaPass, err := utils.GetSecret(apiv1.UserReplication)
@@ -212,16 +276,87 @@ func Bootstrap(ctx context.Context) error {
 		}
 	}
 
+	// The member is joined: drop a stale quarantine marker if any.
+	if err := os.Remove(mysql.QuarantineFile); err != nil && !os.IsNotExist(err) {
+		return errors.Wrap(err, "remove quarantine file")
+	}
+
 	if err := db.EnableSuperReadonly(ctx); err != nil {
 		return errors.Wrap(err, "enable super read only")
 	}
 
 	return nil
 }
 
+// isPrimaryWritable reports whether the node at primaryIp accepts writes.
+func isPrimaryWritable(ctx context.Context, primaryIp string) (bool, error) {
+	operatorPass, err := utils.GetSecret(apiv1.UserOperator)
+	if err != nil {
+		return false, errors.Wrapf(err, "get %s password", apiv1.UserOperator)
+	}
+	readTimeout, err := utils.GetReadTimeout()
+	if err != nil {
+		return false, errors.Wrap(err, "get read timeout")
+	}
+	primaryDB, err := database.NewDatabase(ctx, database.DBParams{
+		User:               apiv1.UserOperator,
+		Pass:               operatorPass,
+		Host:               primaryIp,
+		ReadTimeoutSeconds: readTimeout,
+	})
+	if err != nil {
+		return false, errors.Wrapf(err, "connect to primary %s", primaryIp)
+	}
+	defer func() { _ = primaryDB.Close() }()
+
+	readOnly, err := primaryDB.IsReadonly(ctx)
+	if err != nil {
+		return false, errors.Wrap(err, "check primary read only status")
+	}
+	return !readOnly, nil
+}
+
+// errantGTIDs returns the GTIDs executed locally but absent on the primary.
+func errantGTIDs(ctx context.Context, localDB *database.DB, primaryIp string) (string, error) {
+	localGTIDs, err := localDB.GetGTIDExecuted(ctx)
+	if err != nil {
+		return "", errors.Wrap(err, "get local gtid_executed")
+	}
+	if localGTIDs == "" {
+		return "", nil
+	}
+
+	operatorPass, err := utils.GetSecret(apiv1.UserOperator)
+	if err != nil {
+		return "", errors.Wrapf(err, "get %s password", apiv1.UserOperator)
+	}
+	readTimeout, err := utils.GetReadTimeout()
+	if err != nil {
+		return "", errors.Wrap(err, "get read timeout")
+	}
+	primaryDB, err := database.NewDatabase(ctx, database.DBParams{
+		User:               apiv1.UserOperator,
+		Pass:               operatorPass,
+		Host:               primaryIp,
+		ReadTimeoutSeconds: readTimeout,
+	})
+	if err != nil {
+		return "", errors.Wrapf(err, "connect to primary %s", primaryIp)
+	}
+	defer func() { _ = primaryDB.Close() }()
+
+	primaryGTIDs, err := primaryDB.GetGTIDExecuted(ctx)
+	if err != nil {
+		return "", errors.Wrap(err, "get primary gtid_executed")
+	}
+
+	return localDB.GTIDSubtract(ctx, localGTIDs, primaryGTIDs)
+}
+
 func getTopology(ctx context.Context, fqdn string, peers sets.Set[string]) (string, []string, error) {
 	replicas := sets.New[string]()
 	primary := ""
+	stoppedSource := ""
 
 	operatorPass, err := utils.GetSecret(apiv1.UserOperator)
 	if err != nil {
@@ -262,6 +397,21 @@ func getTopology(ctx context.Context, fqdn string, peers sets.Set[string]) (stri
 
 		if status == mysqldb.ReplicationStatusActive {
 			primary = source
+		} else if status == mysqldb.ReplicationStatusStopped && source != "" && source != replicaHost {
+			stoppedSource = source
+		}
+	}
+
+	if primary == "" && stoppedSource != "" {
+		// A stopped channel remembers its source — the primary. Honor the
+		// hint only if the source resolves: after a pause/resume the lowest
+		// ordinal boots alone and its channel points at a peer that does not
+		// exist yet.
+		if _, err := utils.GetPodIP(stoppedSource); err == nil {
+			log.Printf("No active replication, using stopped channel source as primary: %s", stoppedSource)
+			primary = stoppedSource
+		} else {
+			log.Printf("Stopped channel source %s does not resolve, ignoring the hint", stoppedSource)
 		}
 	}
 

cmd/example-gen/pkg/defaults/manual.go

@@ -35,6 +35,7 @@ func mysqlDefaults(spec *apiv1.MySQLSpec) {
 	podSpecDefaults(&spec.PodSpec, ImageMySQL, resources("2Gi", "", "4Gi", ""), configurationMySQL, 600, envList("BOOTSTRAP_READ_TIMEOUT", "600", "ASYNC_SOURCE_RETRY_COUNT", "3", "ASYNC_SOURCE_CONNECT_RETRY", "60"), envFromList("mysql-env-secret"))
 
 	spec.AutoRecovery = true
+	spec.ErrantTransactionsPolicy = apiv1.ErrantTransactionsManual
 	spec.VolumeSpec = nil
 	spec.ExposePrimary.Enabled = true
 

cmd/example-gen/scripts/lib/ps.sh

@@ -116,6 +116,7 @@ del_fields_to_comment() {
 		| yq "del(.spec.mysql.imagePullSecrets)" \
 		| yq "del(.spec.mysql.initContainer)" \
 		| yq "del(.spec.mysql.vaultSecretName)" \
+		| yq "del(.spec.mysql.errantTransactionsPolicy)" \
 		| yq "del(.spec.orchestrator.configuration)" \
 		| yq "del(.spec.mysql.env)" \
 		| yq "del(.spec.mysql.envFrom)" \

cmd/healthcheck/main.go

@@ -17,6 +17,7 @@ import (
 	state "github.com/percona/percona-server-mysql-operator/cmd/internal/naming"
 	mysqldb "github.com/percona/percona-server-mysql-operator/pkg/db"
 	"github.com/percona/percona-server-mysql-operator/pkg/k8s"
+	"github.com/percona/percona-server-mysql-operator/pkg/mysql"
 	"github.com/percona/percona-server-mysql-operator/pkg/naming"
 	"github.com/percona/percona-server-mysql-operator/pkg/xtrabackup"
 )
@@ -96,6 +97,14 @@ func main() {
 }
 
 func checkReadinessAsync(ctx context.Context) error {
+	// A quarantined member (errant transactions, not joined to the cluster) is
+	// reported NotReady so the divergence surfaces in the cluster status and
+	// halts rollouts. rebuild/inject-empty clear the marker quickly; only the
+	// manual policy leaves it in place.
+	if _, err := os.Stat(mysql.QuarantineFile); err == nil {
+		return errors.New("member is quarantined due to errant transactions; see the ErrantGTIDsDetected event")
+	}
+
 	podIP, err := getPodIP()
 	if err != nil {
 		return errors.Wrap(err, "get pod IP")

cmd/internal/db/db.go

@@ -118,6 +118,7 @@ func (d *DB) StartReplication(ctx context.Context, host, replicaPass string, por
                 SOURCE_HOST=?,
                 SOURCE_PORT=?,
                 SOURCE_SSL=1,
+                GET_SOURCE_PUBLIC_KEY=1,
                 SOURCE_CONNECTION_AUTO_FAILOVER=1,
                 SOURCE_AUTO_POSITION=1,
                 SOURCE_RETRY_COUNT=?,
@@ -167,7 +168,8 @@ func (d *DB) ReplicationStatus(ctx context.Context) (db.ReplicationStatus, strin
 		return db.ReplicationStatusActive, host, nil
 	}
 
-	return db.ReplicationStatusStopped, "", nil
+	// A stopped channel still knows its source.
+	return db.ReplicationStatusStopped, host, nil
 }
 
 func (d *DB) IsReplica(ctx context.Context) (bool, error) {
@@ -427,3 +429,10 @@ func (d *DB) GetGTIDExecuted(ctx context.Context) (string, error) {
 	err := d.db.QueryRowContext(ctx, "SELECT @@GTID_EXECUTED").Scan(&gtid)
 	return gtid, errors.Wrap(err, "get GTID_EXECUTED")
 }
+
+// GTIDSubtract returns the GTIDs in set a that are not in set b.
+func (d *DB) GTIDSubtract(ctx context.Context, a, b string) (string, error) {
+	var diff string
+	err := d.db.QueryRowContext(ctx, "SELECT GTID_SUBTRACT(?, ?)", a, b).Scan(&diff)
+	return diff, errors.Wrap(err, "gtid_subtract")
+}

config/crd/bases/ps.percona.com_perconaservermysqls.yaml

@@ -3422,6 +3422,13 @@ spec:
                           x-kubernetes-map-type: atomic
                       type: object
                     type: array
+                  errantTransactionsPolicy:
+                    default: manual
+                    enum:
+                    - manual
+                    - rebuild
+                    - inject-empty
+                    type: string
                   expose:
                     properties:
                       allocateLoadBalancerNodePorts:

deploy/bundle.yaml

@@ -7427,6 +7427,13 @@ spec:
                           x-kubernetes-map-type: atomic
                       type: object
                     type: array
+                  errantTransactionsPolicy:
+                    default: manual
+                    enum:
+                    - manual
+                    - rebuild
+                    - inject-empty
+                    type: string
                   expose:
                     properties:
                       allocateLoadBalancerNodePorts:

deploy/cr.yaml

@@ -257,6 +257,7 @@ spec:
 #              storage: 1Gi
 #    bootstrap:
 #      mode: auto
+#    errantTransactionsPolicy: manual
   proxy:
     haproxy:
       enabled: true

deploy/crd.yaml

@@ -7427,6 +7427,13 @@ spec:
                           x-kubernetes-map-type: atomic
                       type: object
                     type: array
+                  errantTransactionsPolicy:
+                    default: manual
+                    enum:
+                    - manual
+                    - rebuild
+                    - inject-empty
+                    type: string
                   expose:
                     properties:
                       allocateLoadBalancerNodePorts: