feat(cluster): add feature gates (#6076)

Author: liubog2008

api/core/v1alpha1/cluster_types.go

@@ -17,6 +17,8 @@ package v1alpha1
 import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	meta "github.com/pingcap/tidb-operator/api/v2/meta/v1alpha1"
 )
 
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
@@ -74,6 +76,8 @@ type ClusterSpec struct {
 	// The default value is 10.
 	// +kubebuilder:validation:Minimum=0
 	RevisionHistoryLimit *int32 `json:"revisionHistoryLimit,omitempty"`
+
+	FeatureGates []meta.FeatureGate `json:"featureGates,omitempty"`
 }
 
 type SuspendAction struct {
@@ -142,6 +146,9 @@ type ClusterStatus struct {
 	// PD means url of the pd service, it's prepared for internal use
 	// e.g. https://pd:2379
 	PD string `json:"pd,omitempty"`
+
+	// FeatureGates of this cluster
+	FeatureGates []meta.FeatureGateStatus `json:"featureGates,omitempty"`
 }
 
 type ComponentKind string

api/core/v1alpha1/zz_generated.deepcopy.go

@@ -0,0 +1,41 @@
+// Copyright 2024 PingCAP, Inc.
+//
+// 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
+
+type Feature string
+
+type FeatureStage string
+
+const (
+	FeatureStageAlpha      FeatureStage = "ALPHA"
+	FeatureStageBeta       FeatureStage = "BETA"
+	FeatureStageStable     FeatureStage = "STABLE"
+	FeatureStageDeprecated FeatureStage = "DEPRECATED"
+)
+
+type FeatureGate struct {
+	Name Feature `json:"name"`
+}
+
+type FeatureGateStatus struct {
+	FeatureGate `json:",inline"`
+	Stage       FeatureStage `json:"stage"`
+}
+
+const (
+	// Support modify volume by VolumeAttributeClass
+	VolumeAttributeClass      Feature      = "VolumeAttributeClass"
+	VolumeAttributeClassStage FeatureStage = FeatureStageAlpha
+)

api/meta/v1alpha1/feature.go

@@ -147,10 +147,7 @@ func setup(ctx context.Context, mgr ctrl.Manager) error {
 	pdcm := pdm.NewPDClientManager(mgr.GetLogger(), c)
 
 	logger.Info("setup volume modifier")
-	vm, err := volumes.NewModifier(ctx, mgr.GetLogger().WithName("VolumeModifier"), c)
-	if err != nil {
-		return fmt.Errorf("failed to create volume modifier: %w", err)
-	}
+	vm := volumes.NewModifierFactory(mgr.GetLogger().WithName("VolumeModifier"), c)
 
 	setupLog.Info("setup controllers")
 	if err := setupControllers(mgr, c, pdcm, vm); err != nil {
@@ -208,7 +205,7 @@ func addIndexer(ctx context.Context, mgr ctrl.Manager) error {
 	return nil
 }
 
-func setupControllers(mgr ctrl.Manager, c client.Client, pdcm pdm.PDClientManager, vm volumes.Modifier) error {
+func setupControllers(mgr ctrl.Manager, c client.Client, pdcm pdm.PDClientManager, vm volumes.ModifierFactory) error {
 	if err := cluster.Setup(mgr, c, pdcm); err != nil {
 		return fmt.Errorf("unable to create controller Cluster: %w", err)
 	}

cmd/operator/main.go

@@ -7,7 +7,6 @@ require (
 	github.com/Azure/azure-sdk-for-go/sdk/azidentity v1.8.1
 	github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/compute/armcompute v1.0.0
 	github.com/Masterminds/semver/v3 v3.3.0
-	github.com/aws/aws-sdk-go-v2 v1.30.5
 	github.com/aws/aws-sdk-go-v2/config v1.27.35
 	github.com/aws/aws-sdk-go-v2/service/ec2 v1.177.3
 	github.com/aws/smithy-go v1.20.4
@@ -60,6 +59,7 @@ require (
 	github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect
 	github.com/AzureAD/microsoft-authentication-library-for-go v1.3.2 // indirect
 	github.com/MakeNowJust/heredoc v1.0.0 // indirect
+	github.com/aws/aws-sdk-go-v2 v1.30.5 // indirect
 	github.com/aws/aws-sdk-go-v2/credentials v1.17.33 // indirect
 	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.13 // indirect
 	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.17 // indirect

go.mod

@@ -76,6 +76,15 @@ spec:
                     type: string
                 type: object
                 x-kubernetes-map-type: atomic
+              featureGates:
+                items:
+                  properties:
+                    name:
+                      type: string
+                  required:
+                  - name
+                  type: object
+                type: array
               paused:
                 description: Paused specifies whether to pause the reconciliation
                   loop for all components of the cluster.
@@ -215,6 +224,19 @@ spec:
                 x-kubernetes-list-map-keys:
                 - type
                 x-kubernetes-list-type: map
+              featureGates:
+                description: FeatureGates of this cluster
+                items:
+                  properties:
+                    name:
+                      type: string
+                    stage:
+                      type: string
+                  required:
+                  - name
+                  - stage
+                  type: object
+                type: array
               id:
                 description: ID is the cluster id.
                 type: string

manifests/crd/core.pingcap.com_clusters.yaml

@@ -24,6 +24,7 @@ import (
 	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
 	coreutil "github.com/pingcap/tidb-operator/pkg/apiutil/core/v1alpha1"
 	"github.com/pingcap/tidb-operator/pkg/client"
+	"github.com/pingcap/tidb-operator/pkg/features"
 	"github.com/pingcap/tidb-operator/pkg/utils/task"
 )
 
@@ -67,10 +68,12 @@ func (t *TaskContext) Sync(ctx task.Context[ReconcileContext]) task.Result {
 		if !errors.IsNotFound(err) {
 			return task.Fail().With("can't get tidb cluster: %w", err)
 		}
+		features.Deregister(rtx.Key.Namespace, rtx.Key.Name)
 
 		return task.Complete().Break().With("tidb cluster has been deleted")
 	}
 	rtx.Cluster = &cluster
+	features.Register(rtx.Cluster)
 
 	if coreutil.ShouldPauseReconcile(rtx.Cluster) {
 		return task.Complete().Break().With("cluster reconciliation is paused")

pkg/controllers/cluster/tasks/ctx.go

@@ -158,8 +158,8 @@ func FakeGroupAndInstanceSliceState[
 	RI runtime.Instance,
 ](g RG, s ...RI) GroupAndInstanceSliceState[RG, RI] {
 	return &fakeGroupAndInstanceSliceState[RG, RI]{
-		GroupState:         FakeGroupState[RG](g),
-		InstanceSliceState: FakeInstanceSliceState[RI](s),
+		GroupState:         FakeGroupState(g),
+		InstanceSliceState: FakeInstanceSliceState(s),
 	}
 }
 
@@ -207,8 +207,8 @@ func FakeGroupAndInstanceSliceAndRevisionState[
 	s ...RI,
 ) GroupAndInstanceSliceAndRevisionState[RG, RI] {
 	return &fakeGroupAndInstanceSliceAndRevisionState[RG, RI]{
-		GroupState:         FakeGroupState[RG](g),
-		InstanceSliceState: FakeInstanceSliceState[RI](s),
+		GroupState:         FakeGroupState(g),
+		InstanceSliceState: FakeInstanceSliceState(s),
 		RevisionState:      FakeRevisionState(update, current, collisionCount),
 	}
 }

pkg/controllers/common/interfaces_test.go

@@ -28,6 +28,7 @@ import (
 
 	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
 	"github.com/pingcap/tidb-operator/pkg/client"
+	"github.com/pingcap/tidb-operator/pkg/features"
 	"github.com/pingcap/tidb-operator/pkg/utils/task/v3"
 )
 
@@ -182,3 +183,13 @@ func TaskSuspendPod(state PodState, c client.Client) task.Task {
 		return task.Retry(task.DefaultRequeueAfter).With("pod is deleting")
 	})
 }
+
+func TaskFeatureGates(state ClusterState) task.Task {
+	return task.NameTaskFunc("FeatureGates", func(context.Context) task.Result {
+		if err := features.Verify(state.Cluster()); err != nil {
+			return task.Fail().With("feature gates are not up to date: %v", err)
+		}
+
+		return task.Complete().With("feature gates are initialized")
+	})
+}

pkg/controllers/common/task.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
 	"github.com/pingcap/tidb-operator/pkg/client"
+	"github.com/pingcap/tidb-operator/pkg/features"
 	"github.com/pingcap/tidb-operator/pkg/utils/fake"
 	"github.com/pingcap/tidb-operator/pkg/utils/task/v3"
 )
@@ -395,3 +396,67 @@ func TestTaskSuspendPod(t *testing.T) {
 		})
 	}
 }
+
+func TestFeatureGates(t *testing.T) {
+	cases := []struct {
+		desc          string
+		state         *fakeState[v1alpha1.Cluster]
+		objs          []client.Object
+		unexpectedErr bool
+
+		expectedResult task.Status
+		expectedObj    *corev1.Pod
+	}{
+		{
+			desc: "up to date",
+			state: &fakeState[v1alpha1.Cluster]{
+				name: "xxx",
+				obj: fake.FakeObj("xxx", func(obj *v1alpha1.Cluster) *v1alpha1.Cluster {
+					obj.Generation = 1
+					return obj
+				}),
+			},
+			expectedResult: task.SComplete,
+		},
+		{
+			desc: "generation is changed",
+			state: &fakeState[v1alpha1.Cluster]{
+				name: "xxx",
+				obj: fake.FakeObj("xxx", func(obj *v1alpha1.Cluster) *v1alpha1.Cluster {
+					obj.Generation = 2
+					return obj
+				}),
+			},
+			expectedResult: task.SFail,
+		},
+		{
+			desc: "uid is changed",
+			state: &fakeState[v1alpha1.Cluster]{
+				name: "xxx",
+				obj: fake.FakeObj("xxx", func(obj *v1alpha1.Cluster) *v1alpha1.Cluster {
+					obj.Generation = 1
+					obj.UID = "newuid"
+					return obj
+				}),
+			},
+			expectedResult: task.SFail,
+		},
+	}
+
+	for i := range cases {
+		c := &cases[i]
+		t.Run(c.desc, func(tt *testing.T) {
+			tt.Parallel()
+
+			features.Register(fake.FakeObj("xxx", func(obj *v1alpha1.Cluster) *v1alpha1.Cluster {
+				obj.Generation = 1
+				return obj
+			}))
+
+			s := &fakeClusterState{s: c.state}
+			res, done := task.RunTask(context.Background(), TaskFeatureGates(s))
+			assert.Equal(tt, c.expectedResult, res.Status(), c.desc)
+			assert.False(tt, done, c.desc)
+		})
+	}
+}

pkg/controllers/common/task_test.go

@@ -30,6 +30,7 @@ func (r *Reconciler) NewRunner(state *tasks.ReconcileContext, reporter task.Task
 
 		// get cluster
 		common.TaskContextCluster(state, r.Client),
+		common.TaskFeatureGates(state),
 		// if it's paused just return
 		task.IfBreak(common.CondClusterIsPaused(state)),
 
@@ -51,7 +52,7 @@ func (r *Reconciler) NewRunner(state *tasks.ReconcileContext, reporter task.Task
 
 		common.TaskContextPDSlice(state, r.Client),
 		tasks.TaskConfigMap(state, r.Client),
-		tasks.TaskPVC(state, r.Logger, r.Client, r.VolumeModifier),
+		tasks.TaskPVC(state, r.Logger, r.Client, r.VolumeModifierFactory),
 		tasks.TaskPod(state, r.Client),
 		// If pd client has not been registered yet, do not update status of the pd
 		task.IfBreak(tasks.CondPDClientIsNotRegisterred(state),

pkg/controllers/pd/builder.go

@@ -37,18 +37,18 @@ import (
 )
 
 type Reconciler struct {
-	Logger          logr.Logger
-	Client          client.Client
-	PDClientManager pdm.PDClientManager
-	VolumeModifier  volumes.Modifier
+	Logger                logr.Logger
+	Client                client.Client
+	PDClientManager       pdm.PDClientManager
+	VolumeModifierFactory volumes.ModifierFactory
 }
 
-func Setup(mgr manager.Manager, c client.Client, pdcm pdm.PDClientManager, vm volumes.Modifier) error {
+func Setup(mgr manager.Manager, c client.Client, pdcm pdm.PDClientManager, vm volumes.ModifierFactory) error {
 	r := &Reconciler{
-		Logger:          mgr.GetLogger().WithName("PD"),
-		Client:          c,
-		PDClientManager: pdcm,
-		VolumeModifier:  vm,
+		Logger:                mgr.GetLogger().WithName("PD"),
+		Client:                c,
+		PDClientManager:       pdcm,
+		VolumeModifierFactory: vm,
 	}
 	return ctrl.NewControllerManagedBy(mgr).For(&v1alpha1.PD{}).
 		Owns(&corev1.Pod{}).

pkg/controllers/pd/controller.go

@@ -31,10 +31,11 @@ import (
 	"github.com/pingcap/tidb-operator/pkg/volumes"
 )
 
-func TaskPVC(state *ReconcileContext, logger logr.Logger, c client.Client, vm volumes.Modifier) task.Task {
+func TaskPVC(state *ReconcileContext, logger logr.Logger, c client.Client, vm volumes.ModifierFactory) task.Task {
 	return task.NameTaskFunc("PVC", func(ctx context.Context) task.Result {
+		ck := state.Cluster()
 		pvcs := newPVCs(state)
-		if wait, err := volumes.SyncPVCs(ctx, c, pvcs, vm, logger); err != nil {
+		if wait, err := volumes.SyncPVCs(ctx, c, pvcs, vm.New(ck.Namespace, ck.Name), logger); err != nil {
 			return task.Fail().With("failed to sync pvcs: %v", err)
 		} else if wait {
 			return task.Wait().With("waiting for pvcs to be synced")

pkg/controllers/pd/tasks/pvc.go

@@ -143,6 +143,8 @@ func TestTaskPVC(t *testing.T) {
 
 			ctrl := gomock.NewController(tt)
 			vm := volumes.NewMockModifier(ctrl)
+			vf := volumes.NewMockModifierFactory(ctrl)
+			vf.EXPECT().New("", "aaa").Return(vm)
 			expectedPVCs := newPVCs(c.state)
 			for _, expected := range expectedPVCs {
 				for _, current := range c.pvcs {
@@ -164,7 +166,7 @@ func TestTaskPVC(t *testing.T) {
 				fc.WithError("patch", "*", errors.NewInternalError(fmt.Errorf("fake internal err")))
 			}
 
-			res, done := task.RunTask(ctx, TaskPVC(c.state, logr.Discard(), fc, vm))
+			res, done := task.RunTask(ctx, TaskPVC(c.state, logr.Discard(), fc, vf))
 			assert.Equal(tt, c.expectedStatus.String(), res.Status().String(), res.Message())
 			assert.False(tt, done, c.desc)