package common import ( "context" "encoding/json" "fmt" "sort" "strconv" "strings" "time" "github.com/google/go-cmp/cmp" "github.com/google/go-cmp/cmp/cmpopts" "github.com/pingcap/errors" oplabel "github.com/pingcap/tidb-operator/pkg/apis/label" opv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1" "github.com/samber/lo" "github.com/tidbcloud/config-service/sdk/api" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" rbacv1 "k8s.io/api/rbac/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "k8s.io/apimachinery/pkg/util/sets" "k8s.io/client-go/util/retry" "k8s.io/utils/ptr" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" "sigs.k8s.io/controller-runtime/pkg/log" "github.com/tidbcloud/eks-provider/api/v1alpha1" "github.com/tidbcloud/eks-provider/controllers/cluster/task" "github.com/tidbcloud/eks-provider/controllers/cluster/task/common/audit" "github.com/tidbcloud/eks-provider/controllers/cluster/task/common/statementlog" taskutils "github.com/tidbcloud/eks-provider/controllers/cluster/task/utils" "github.com/tidbcloud/eks-provider/controllers/utils" "github.com/tidbcloud/eks-provider/controllers/utils/configmanager" "github.com/tidbcloud/eks-provider/utils/tidb-operator/tidbcluster" cloudv1 "github.com/tidbcloud/infra-api/api/v1" "github.com/tidbcloud/infra-provider/pkg/config" "github.com/tidbcloud/infra-provider/pkg/consts" "github.com/tidbcloud/infra-provider/pkg/deploy" "github.com/tidbcloud/infra-provider/pkg/gitops" sdk "github.com/tidbcloud/infra-provider/pkg/gitops-sdk" "github.com/tidbcloud/infra-provider/pkg/k8s" "github.com/tidbcloud/infra-provider/pkg/metrics" "github.com/tidbcloud/infra-provider/pkg/nodepool" "github.com/tidbcloud/infra-provider/pkg/provider" "github.com/tidbcloud/infra-provider/pkg/util" ) const ( clusterRequeueAfter = time.Minute ) const ( // annPDDeleteSlots is annotation key of pd delete slots. annPDDeleteSlots = "pd.tidb.pingcap.com/delete-slots" // annTiDBDeleteSlots is annotation key of tidb delete slots. annTiDBDeleteSlots = "tidb.tidb.pingcap.com/delete-slots" // annTiKVDeleteSlots is annotation key of tikv delete slots. annTiKVDeleteSlots = "tikv.tidb.pingcap.com/delete-slots" // annTiFlashDeleteSlots is annotation key of tiflash delete slots. annTiFlashDeleteSlots = "tiflash.tidb.pingcap.com/delete-slots" // annTiProxyDeleteSlots is annotation key of tiproxy delete slots. annTiProxyDeleteSlots = "tiproxy.tidb.pingcap.com/delete-slots" ) const ( annoKeyPDLowSpaceRatio = "pd.cloud.pingcap.com/low-space-ratio" // as 0.05 of the space is reserve for the placeholder, so this value should not be too large. // as the default value is 0.8, so we choose 0.901 now (a value unlikely specified by GitOps). pdLowSpaceRatioInScalingIn float64 = 0.901 pdConfigPrefix = "pd/api/v1/config" pdScheduleConfigPath = "pd/api/v1/config/schedule" // defaultExtraTiDBPodTerminationGracePeriodSeconds is the extra termination grace period seconds for tidb pod when the tiproxy is enabled defaultExtraTiDBPodTerminationGracePeriodSeconds int64 = 30 defaultExtraTiProxyPodTerminationGracePeriodSeconds int64 = 15 // tiproxyTerminationGracePeriodSeconds // We use the fixed value 20 minutes for tiproxy termination grace period seconds. With this, we can avoid restarts of tiproxy // when either `proxy.graceful-close-conn-timeout` or `proxy.graceful-wait-before-shutdown` is being changed. // As we set this to 20 minutes, we set the maximum value to 9 minutes for both `proxy.graceful-close-conn-timeout` and `proxy.graceful-wait-before-shutdown`. tiproxyTerminationGracePeriodSeconds int64 = 60 * 20 // 20 minutes ) var componentDeleteSlotsAnnMap = map[cloudv1.ComponentType]string{ cloudv1.ComponentPD: annPDDeleteSlots, cloudv1.ComponentTiDB: annTiDBDeleteSlots, cloudv1.ComponentTiKV: annTiKVDeleteSlots, cloudv1.ComponentTiProxy: annTiProxyDeleteSlots, cloudv1.ComponentTiFlash: annTiFlashDeleteSlots, } type TiDBCluster struct { *task.BaseTask } var _ deploy.Task = &TiDBCluster{} func NewTiDBCluster(ctx *task.Context) *TiDBCluster { return &TiDBCluster{ BaseTask: &task.BaseTask{Context: ctx}, } } func (t *TiDBCluster) Name() string { return "TiDBCluster" } func (t *TiDBCluster) buildClusterContext() (cc *ClusterContext, err error) { cc = &ClusterContext{ Target: t.Cluster, Provider: t.Provider, ConfigProvider: t.ConfigProvider, // Injected providers Cluster: t, } return enableKMS(t.BaseTask, cc) } func reconcileNormalizeLastAppliedConfigAnnotation(obj client.Object, enabled bool) { if enabled { util.SetAnnotation(obj, consts.NormalizeLastAppliedConfigAnnotation, "true") return } annotations := obj.GetAnnotations() if annotations == nil { return } delete(annotations, consts.NormalizeLastAppliedConfigAnnotation) obj.SetAnnotations(annotations) } func normalizeLastAppliedConfigAnnotationForCreateOrUpdate(obj client.Object, enabled bool) { if !enabled { return } k8s.NormalizeLastAppliedConfigurationAnnotation(obj) } func (t *TiDBCluster) Create(ctx context.Context) (res deploy.Result) { t.BeginAt = ptr.To(time.Now()) tidbMountOOMPathForNew := t.Cluster.FeatureEnabled(consts.ClusterFeatureTiDBExposeOOMPath) tidbMountOOMPathForExisting := t.Cluster.FeatureEnabled(consts.ClusterFeatureTiDBExposeOOMPath) tidbMountOOMPathForUpgrading := tidbMountOOMPathForNew && t.Cluster.Status.Phase == cloudv1.ClusterPhaseUpgrading tidbMountReplayerPathForNew := t.Cluster.FeatureEnabled(consts.ClusterFeatureTiDBExposeReplayerPath) tidbMountReplayerPathForExisting := t.Cluster.FeatureEnabled(consts.ClusterFeatureTiDBExposeReplayerPath) tidbMountReplayerPathForUpgrading := tidbMountReplayerPathForNew && t.Cluster.Status.Phase == cloudv1.ClusterPhaseUpgrading cli := t.EnvProvider.GetClient() for _, obj := range t.subResources() { if _, err := controllerutil.CreateOrUpdate(ctx, cli, obj, func() error { return nil }); err != nil { return deploy.Error(util.WrapError("create sub-resource", err)) } } c, err := t.buildClusterContext() if err != nil { return deploy.Error(err) } tc, err := t.buildTibCluster(c) if err != nil { return deploy.Error(err) } normalizeLastAppliedEnabled := config.FeatureEnabled(t.Config.FeatureGates, config.NormalizeLastAppliedConfiguration, t.Cluster.GetTenantID(), t.Cluster.GetProjectID()) reconcileNormalizeLastAppliedConfigAnnotation(tc, normalizeLastAppliedEnabled) // plumb customer's KMS key into storage class if customers enabled if t.Cluster.FeatureEnabled(consts.ClusterFeatureUseCMEK) { // set the storage class for the tidb cluster object err = t.useCMEKStorageClass(tc) if err != nil { return deploy.Error(err) } } // After `AssembleTidbClusterValues`, tc.Spec.Version is set. // TODO(chenfei): remove me after we deprecated cloud-native arch if taskutils.IsClusterCreating(t.Cluster) && t.Cluster.Spec.Cluster.TiDB.Audit == nil && audit.CheckIfSupported(ctx, tc.Spec.Version) { // Enable audit log for newly created cluster. t.Cluster.Spec.Cluster.TiDB.Audit = &cloudv1.Audit{ Enable: true, BucketURL: "", Region: "", RoleArn: "", } return deploy.Panic(nil).WithRequeuAfter(0) } obj := &opv1alpha1.TidbCluster{} obj.Annotations = map[string]string{} obj.Namespace = t.Namer.GetTiDBNamespace() obj.Name = tc.GetName() var waitForTiFlashScaleFromZero bool opResult, err := createOrUpdateTidbClusterWithRetry(ctx, cli, obj, func() error { waitForTiFlashScaleFromZero = false current := obj.DeepCopy() defer func() { diff := cmp.Diff(current, obj, cmpopts.EquateEmpty()) if diff == "" { t.Logger.Info("tc is synced") return } t.Logger.Info("tc is not synced", "diff", diff) }() obj.SetLabels(tc.GetLabels()) for k, v := range tc.GetAnnotations() { obj.Annotations[k] = v } reconcileNormalizeLastAppliedConfigAnnotation(obj, normalizeLastAppliedEnabled) // if creation timestamp is zero, these code only works for new tc if obj.CreationTimestamp.IsZero() { obj.Spec = tc.Spec initializeGKEV1MinDomains(t.Cluster, obj) setMigrateDeleteSlotsAnn(t.Cluster, obj) if tidbMountOOMPathForNew { c.MountOOMPathToHost(obj) } if tidbMountReplayerPathForNew { c.MountPlanReplayerPathToHost(obj) } if taskutils.IsBootstrapSQLEnabled(t.Context, obj.Spec.Version) { t.Logger.Info("set bootstrap sql configmap name when creating TC") obj.Spec.TiDB.BootstrapSQLConfigMapName = ptr.To(t.Namer.BootstrapSQLConfigMapName()) obj.Spec.TiDB.Config.Set(SEMKey, true) } else { obj.Spec.TiDB.Config.Set(SEMKey, false) } statementlog.Overlay(obj) // NOTE: this external id is not same as t.GetAWSAssumeRoleExternalIDForXP() externalID := t.GetAWSAuditLogRoleExternalID() if _, err := audit.Reconcile(ctx, t.Config, t.Cluster, t.Local, obj, cli, true, externalID, t.Cluster.Namespace, t.GetGCPNodePoolServiceAccountKeySecretName(), nil); err != nil { return err } } else { waitForTiFlashScaleFromZero, err = t.waitForTiFlashScaleFromZeroPVCDeletion(ctx, cli, obj, tc) if err != nil || waitForTiFlashScaleFromZero { return err } } if t.Cluster.Status.Phase != "" && t.Cluster.Status.Phase != cloudv1.ClusterPhasePending && t.Cluster.Status.Phase != cloudv1.ClusterPhaseCreating { // set MinReadySeconds after cluster created // NOTE: when the cluster becomes running, one more round of reconcile is needed to trigger this if err = c.OverlayMinReadyAnnotations(ctx, t.Cluster.GetDefaultGroupID(), obj); err != nil { return err } } if t.Cluster.FeatureEnabled(consts.ClusterFeatureSetPriorityClass) { obj.Spec.PriorityClassName = ptr.To(consts.PriorityClasName) } // Set preStop hook for tidb when the tiproxy is disabled if t.Cluster.FeatureEnabled(consts.ClusterFeatureEnableGracefulShutdown) && t.Cluster.Spec.Cluster.TiProxy.Replicas == 0 { setPreStopHook(ctx, t.Cluster, t.Cluster.GetDefaultGroupID(), obj, t.Config.ClusterController) } // Set PV retain policy to 'Delete' pvPolicy := corev1.PersistentVolumeReclaimDelete obj.Spec.PVReclaimPolicy = &pvPolicy // set config update strategy obj.Spec.ConfigUpdateStrategy = tc.Spec.ConfigUpdateStrategy // FRM-2728: https://tidb.atlassian.net/browse/FRM-2728 // Enable PPv2 For TiDB mutateTcForComponentsEnablePPv2(ctx, t.Cluster, obj) // ensure endpoint update take effect if obj.Spec.TiDB != nil && tc.Spec.TiDB != nil { obj.Spec.TiDB.Service = tc.Spec.TiDB.Service } // update replicas for non scaling up components to make sure scale up precedes scale in/out nonScaleUpComps := nonScaleUpComponents(t.Cluster, t.Provider) t.Logger.Info("non-scale-up components", "components", nonScaleUpComps) if err := t.mutateTcForReplicas(ctx, cli, obj, tc, nonScaleUpComps); err != nil { return err } // update storage-size for non instance-type scaling components comps := NonInstanceTypeScalingComponents(t.Cluster) t.mutateTcForStorageSize(obj, tc, comps) if err := t.ensureGKEVolumeAttributesClassesForModification(ctx, cli, obj, tc, comps); err != nil { return err } // Keep new volumes out of current until their VACs are prepared. if len(obj.Spec.TiKV.StorageVolumes) == 0 && len(tc.Spec.TiKV.StorageVolumes) > 0 { obj.Spec.TiKV.StorageVolumes = tc.Spec.TiKV.StorageVolumes } t.mutateTcForVolumeClasses(obj, tc, comps) // do not update PD if PD is unavailable as it will call PD HTTP API, // and rely on requeue in PostCreate if cluster is not running (because PD is unavailable). if err = t.ConfigManager.UpdateConfig(ctx, t.Cluster, t.IsRegionalManaged(), configmanager.OverrideRevisionFinder(t.Local), obj, t.PDClient, t.removeUnavailablePD(comps)); err != nil { return err } // NOTE: we should call `mutateTiDBForTokenBasedAuth` after `t.ConfigManager.UpdateConfig` to ensure // the version of components in `obj` is CORRECT(patched/updated by ConfigRevision). mutateTiDBForTokenBasedAuth(ctx, t.Cluster, obj, tc) mutateTiDBAndTiProxyForTiProxy(t.Cluster, &t.Cluster.Spec.Cluster.TiProxy, obj, tc, t.Cluster.GetDefaultGroupID()) // check if the cluster is not scaling in/out to make sure scale in/out precedes scale down if !isModifyingComponentReplicas(obj) { t.Logger.Info("mutate tc for instance type") // mutate component if a rolling-update is needed after instance-type changed (and new nodegroups are ready) if err = t.mutateTcForInstanceType(ctx, cli, obj, tc, cli); err != nil { return err } } t.mutateTcForSuspendActions(obj, tc) if err := overlayTiDBAddIndex(c, obj); err != nil { return err } if tidbMountOOMPathForExisting || tidbMountOOMPathForUpgrading { c.MountOOMPathToHost(obj) } if tidbMountReplayerPathForExisting || tidbMountReplayerPathForUpgrading { c.MountPlanReplayerPathToHost(obj) } if err := mutateTiKVForDedicatedRaftEngineStorage(obj, tc); err != nil { return err } // Only overlay node selector for GKE and AKS clusters for k8s version upgrade if t.Config.Provider == consts.ProviderTypeGKE || t.Config.Provider == consts.ProviderTypeAKS { nodepool.OverlayNodeSelectorWithK8sVersion(obj, t.GetNodeVersion(true)) } // Merge the sidecar update and the tidb component update together externalID := t.GetAWSAuditLogRoleExternalID() if err := audit.FillSidecarContainer(ctx, t.Config, t.Cluster, t.Local, obj, cli, true, externalID, t.Cluster.Namespace, t.GetGCPNodePoolServiceAccountKeySecretName(), nil); err != nil { return err } // Merge TiFlash graceful shutdown settings mutateTiFlashForGracefulShutdown(ctx, t.Cluster, tc, obj) // The v1 TiDBCluster path uses controllerutil.CreateOrUpdate instead of pkg/k8s.Apply, // so explicitly migrate existing last-applied annotations once the normalize marker is enabled. normalizeLastAppliedConfigAnnotationForCreateOrUpdate(obj, normalizeLastAppliedEnabled) return nil }) defer func() { // any error occurred, it's better to observe the current status of tidbcluster observeRes := t.observeTidbCluster(ctx, cli) if res.Error() != nil { // log the observe result but do not overwrite the original one if observeRes.Error() != nil { t.Logger.Info("observe tidb cluster when creating/updating tc failed", "observeResult", observeRes) } return } t.Context.TidbClusterLastUpdateTime = ptr.To(time.Now()) res = observeRes // return the observe result }() if err != nil { // skip operations (e.g `setPDLowSpaceRatio`) in `PostCreate` return deploy.Fatal(err) } if waitForTiFlashScaleFromZero { return deploy.Fatal(k8s.RequeueOnlyError).WithRequeuAfter(deploy.DefaultWaitRequeueAfter) } if opResult != controllerutil.OperationResultCreated { // When tc not create, audit.Reconcile will just update tc as what it is expect. // When tc already created. Since `volumeClaimTemplates` of StatefulSet do not support updated. // To enable audit log, it will need some trivial hack which will update the STS directly. // See comment of `audit.Reconcile` for detail. // NOTE: this external id is not same as t.GetAWSAssumeRoleExternalIDForXP() externalID := t.GetAWSAuditLogRoleExternalID() res, err := audit.Reconcile(ctx, t.Config, t.Cluster, t.Local, obj, cli, true, externalID, t.Cluster.Namespace, t.GetGCPNodePoolServiceAccountKeySecretName(), nil) if err != nil { return deploy.Error(err) } if !res.IsZero() { return deploy.Done().WithRequeuAfter(res.RequeueAfter) } } err = ensureSEM(ctx, t.Cluster, cli, obj) if err != nil { res := deploy.Error(err) if err == ErrWaitTidbInitialized { res = res.WithRequeuAfter(time.Second * 10) } return res } // Make sure operator already learn than we update tidb configuration to enable SEM. // Or in the PostCreate, tc may be Ready but actually has not enable SEM yes and we will update cluster phase as Running // if other tasks success too. if t.Cluster.Status.Phase == cloudv1.ClusterPhaseCreating { enable, err := anyPodEnableSEM(ctx, cli, obj) if err != nil { return deploy.Error(err) } if !enable { return deploy.Error(fmt.Errorf("operator not start to enable SEM")).WithRequeuAfter(time.Second * 30) } } return deploy.Done() } func createOrUpdateTidbClusterWithRetry(ctx context.Context, cli client.Client, obj *opv1alpha1.TidbCluster, mutate controllerutil.MutateFn) (controllerutil.OperationResult, error) { var opResult controllerutil.OperationResult err := retry.RetryOnConflict(retry.DefaultRetry, func() error { var err error opResult, err = controllerutil.CreateOrUpdate(ctx, cli, obj, mutate) return err }) return opResult, err } func (t *TiDBCluster) observeTidbCluster(ctx context.Context, remoteCli client.Client) deploy.Result { var tc opv1alpha1.TidbCluster key := types.NamespacedName{Name: t.Namer.GetTiDBClusterName(), Namespace: t.Namer.GetTiDBNamespace()} if err := remoteCli.Get(ctx, key, &tc); err != nil { return deploy.ErrorOrDone(err) } t.Tidbcluster = &tc t.TCGetter = tidbcluster.NewV1Getter(&tc) if t.CommonK8sClusterStatus != nil { t.CommonK8sClusterStatus.Kubernetes = &cloudv1.KubernetesStatus{Namespace: t.Namer.GetTiDBNamespace()} } if t.Cluster.Status.Phase != cloudv1.ClusterPhasePending && t.Cluster.Status.Phase != cloudv1.ClusterPhaseCreating { // Error from resolving the cluster status does not block the remaining tasks err := t.resolveStatus(ctx) if err != nil { return deploy.Error(err) } } return deploy.Done() } // isClusterReady checks whether the tidb cluster is ready // 1. check whether the tc is ready(If ready condition is true, it means all components are ready) // 2. check whether the tidb pods are ready(If ready condition is true, it means readiness gates are ready and tidb service is ready to accept traffic) func (t *TiDBCluster) isClusterReady(ctx context.Context) (bool, error) { var ready bool for _, c := range t.Tidbcluster.Status.Conditions { if c.Type == opv1alpha1.TidbClusterReady && c.Status == corev1.ConditionTrue { ready = true break } } if !ready { return false, fmt.Errorf("tc is not ready yet") } remoteCli := t.EnvProvider.GetClient() pods, err := k8s.SelectPods(ctx, remoteCli, t.Tidbcluster.Namespace, t.GetComponentPodLabels(cloudv1.ComponentTiDB)) if err != nil { return false, fmt.Errorf("failed to select tidb pods: %v", err) } name2Pods := make(map[string]corev1.Pod, len(pods)) for _, pod := range pods { name2Pods[pod.Name] = pod } for name := range t.Tidbcluster.Status.TiDB.Members { pod, ok := name2Pods[name] if !ok { return false, fmt.Errorf("tidb pod(%s) is not found", name) } if !k8s.IsPodReady2(pod) { return false, fmt.Errorf("tidb pod(%s) is not ready", name) } } return true, nil } func (t *TiDBCluster) PostCreate(ctx context.Context) deploy.Result { if t.Tidbcluster == nil { return deploy.Error(fmt.Errorf("tidb cluster is not ready yet")) } u, err := t.GetClusterProxyBaseURL(ctx) if err != nil { return deploy.Error(err) } t.Cluster.Status.Proxy = &cloudv1.ProxyStatus{Endpoint: u} // set cluster gateway base url gatewayEp, err := t.GetClusterGatewayBaseURL(ctx) if err != nil { return deploy.Error(err) } t.Cluster.Status.Gateway = &cloudv1.GatewayStatus{Endpoint: gatewayEp} // set/reset PD low-space-ratio if err := t.setPDLowSpaceRatio(ctx); err != nil { return deploy.Error(err) } // always requeue to sync if cluster is scaling or upgrading, // or PD is not Normal as PD's status is not considered in `Phase` phase := t.Cluster.Status.Phase if phase == cloudv1.ClusterPhaseScaling || phase == cloudv1.ClusterPhaseUpgrading || phase == cloudv1.ClusterPhaseSuspend || (phase == cloudv1.ClusterPhaseRunning && (t.Cluster.Status.DB == nil || t.Cluster.Status.DB.PD == nil || t.Cluster.Status.DB.PD.Status != cloudv1.ComponentNormal)) { t.logModifyProgress() return deploy.Error(fmt.Errorf("cluster is %s", phase)).WithRequeuAfter(clusterRequeueAfter) } ready, err := t.isClusterReady(ctx) if !ready { return deploy.Error(fmt.Errorf("tidb cluster is not ready yet: %v", err)).WithRequeuAfter(clusterRequeueAfter) } return deploy.Done() } func (t *TiDBCluster) logModifyProgress() { t.Logger.Info("up to date replicas", "pd", t.Cluster.GetUpToDateReplicas(cloudv1.ComponentPD), "tikv", t.Cluster.GetUpToDateReplicas(cloudv1.ComponentTiKV), "tidb", t.Cluster.GetUpToDateReplicas(cloudv1.ComponentTiDB), "tiproxy", t.Cluster.GetUpToDateReplicas(cloudv1.ComponentTiProxy), "tiflash", t.Cluster.GetUpToDateReplicas(cloudv1.ComponentTiFlash)) t.Logger.Info("remaining to delete replicas", "pd", t.Cluster.GetRemainingDeleteReplicas(cloudv1.ComponentPD), "tikv", t.Cluster.GetRemainingDeleteReplicas(cloudv1.ComponentTiKV), "tidb", t.Cluster.GetRemainingDeleteReplicas(cloudv1.ComponentTiDB), "tiproxy", t.Cluster.GetRemainingDeleteReplicas(cloudv1.ComponentTiProxy), "tiflash", t.Cluster.GetRemainingDeleteReplicas(cloudv1.ComponentTiFlash)) } func (t *TiDBCluster) Delete(ctx context.Context) deploy.Result { cli := t.EnvProvider.GetClient() return deploy.ErrorOrDone(k8s.DeleteObject(ctx, cli, t.TidbClusterMetaObject())) } func (t *TiDBCluster) PostDelete(ctx context.Context) deploy.Result { cli := t.EnvProvider.GetClient() key := types.NamespacedName{Name: t.Namer.GetTiDBClusterName(), Namespace: t.Namer.GetTiDBNamespace()} return deploy.ErrorOrDone(k8s.GetDeletedObject(ctx, cli, key, t.TidbClusterMetaObject())) } func (t *TiDBCluster) subResources() []client.Object { const ( backupManager = "tidb-backup-manager" ) objs := []client.Object{ // Required for PD dashboard &corev1.Secret{ ObjectMeta: metav1.ObjectMeta{ Name: "db-dashboard-session-secret", Namespace: t.Namer.GetTiDBNamespace(), }, Data: map[string][]byte{ "encryption_key": []byte(cloudv1.RandomString(32)), }, }, // Required for backups and restores &corev1.ServiceAccount{ ObjectMeta: metav1.ObjectMeta{ Name: backupManager, Namespace: t.Namer.GetTiDBNamespace(), }, }, &rbacv1.Role{ ObjectMeta: metav1.ObjectMeta{ Name: backupManager, Namespace: t.Namer.GetTiDBNamespace(), }, Rules: []rbacv1.PolicyRule{ { APIGroups: []string{""}, Resources: []string{"events"}, Verbs: []string{"*"}, }, { APIGroups: []string{"pingcap.com"}, Resources: []string{"backups", "restores", "compactbackups"}, Verbs: []string{"get", "watch", "list", "update"}, }, }, }, &rbacv1.RoleBinding{ ObjectMeta: metav1.ObjectMeta{ Name: backupManager, Namespace: t.Namer.GetTiDBNamespace(), }, Subjects: []rbacv1.Subject{ { Kind: rbacv1.ServiceAccountKind, Name: backupManager, }, }, RoleRef: rbacv1.RoleRef{ APIGroup: "rbac.authorization.k8s.io", Kind: "Role", Name: backupManager, }, }, } return objs } // Methods that implement the GitOps and Cluster interface func getClusterConfigIn3layers(cli client.Client, cluster *cloudv1.Cluster) (*api.DescribeClusterConfigOutput, error) { // If the cluster is in 3layers arch, we DON'T need to describe Config from ClusterConfig By calling DescribeClusterConfiguration. // Because changed config of existing component and newly added config of new component all are given by the upstream of cluster-controller, // for now it's `tidb-mgmt` service. conf, err := sdk.GetCurrentClusterConfiguration(cli, cluster.Spec.ConfigRevisionRef) if err != nil { return nil, err } return sdk.GenerateConfigOutput(cluster, conf) } var _ provider.Cluster = &TiDBCluster{} // GetComponentNodeLabels implements the provider.Cluster interface. func (t *TiDBCluster) GetComponentNodeLabels(componentType cloudv1.ComponentType) map[string]string { // these labels will be used to map pod to zone by node name, // so no instance-type label added here so that we can get all nodes return t.Namer.WithComponent(string(componentType)).Labels() } // GetComponentPodLabels implements the provider.Cluster interface. func (t *TiDBCluster) GetComponentPodLabels(componentType cloudv1.ComponentType) map[string]string { return t.Namer.WithComponent(string(componentType)).GetPodLabels() } func (t *TiDBCluster) buildTibCluster(c *ClusterContext) (*opv1alpha1.TidbCluster, error) { cc, err := getClusterConfigIn3layers(t.Local, t.Cluster) if err != nil { return nil, errors.Annotate(err, "error get cluster config") } tc, err := buildBasicTidbCluster(t.Cluster, cc) if err != nil { return nil, err } if err := t.buildPDConfig(tc, cc); err != nil { return nil, err } if err := t.buildTiKVConfig(tc, cc); err != nil { return nil, err } if err := c.buildTiDBConfig(tc, cc); err != nil { return nil, err } syncFeatureAnnotations(t.Cluster, tc) if err := t.Overlay(cc, tc); err != nil { return nil, err } return tc, nil } func (t *TiDBCluster) computePDConfig(cc *api.DescribeClusterConfigOutput) *opv1alpha1.PDConfigWraper { conf := opv1alpha1.NewPDConfig() if pdConfig, ok := getComponentConfig(cc, api.ComponentPD); ok { conf.GenericConfig.MP = pdConfig.Config.MP } conf.Set("dashboard.public-path-prefix", fmt.Sprintf("/proxy/cluster/%s/pd/dashboard", t.Cluster.Name)) conf.Set("security.cert-allowed-cn", []string{util.FnvHash(t.Cluster.PrefixedId())}) if t.Cluster.Spec.Cluster.TiFlash.Replicas > 0 { conf.Set("replication.enable-placement-rules", true) } return conf } func (t *TiDBCluster) buildPDConfig(tc *opv1alpha1.TidbCluster, cc *api.DescribeClusterConfigOutput) error { pdConfig, ok := getComponentConfig(cc, api.ComponentPD) if !ok { return errors.New("no PD configuration provided") } topo := t.Cluster.Spec.Cluster pdResource, err := t.ConfigProvider.GetResourceSpec(t.Provider, &topo.PD) if err != nil { return errors.Annotate(err, "get pd resources spec") } pdSpec := &opv1alpha1.PDSpec{ ComponentSpec: opv1alpha1.ComponentSpec{ Affinity: utils.Affinity(cloudv1.ComponentPD), NodeSelector: t.GetComponentNodeLabels(cloudv1.ComponentPD), Tolerations: utils.Toleration(cloudv1.ComponentPD), HostNetwork: ptr.To(true), Version: pdConfig.Image.Tag, Env: []corev1.EnvVar{ { Name: "DASHBOARD_SESSION_SECRET", ValueFrom: &corev1.EnvVarSource{ SecretKeyRef: &corev1.SecretKeySelector{ LocalObjectReference: corev1.LocalObjectReference{ Name: fmt.Sprintf("%s-dashboard-session-secret", consts.ClusterPrefix), }, Key: "encryption_key", }, }, }, }, }, ResourceRequirements: pdResource.Resources, Replicas: int32(topo.PD.Replicas), BaseImage: pdConfig.Image.BaseImage, MaxFailoverCount: ptr.To(int32(0)), Service: nil, StorageClassName: firstNonEmpty(topo.PD.StorageClassName, pdResource.StorageClass), Config: t.computePDConfig(cc), MountClusterClientSecret: ptr.To(true), } tc.Spec.PD = pdSpec return nil } func (t *TiDBCluster) computeTiKVConfig(cc *api.DescribeClusterConfigOutput) *opv1alpha1.TiKVConfigWraper { conf := opv1alpha1.NewTiKVConfig() if kvConfig, ok := getComponentConfig(cc, api.ComponentTiKV); ok { conf.GenericConfig.MP = kvConfig.Config.MP } conf.Set("security.cert-allowed-cn", []string{util.FnvHash(t.Cluster.PrefixedId())}) if enabled, keyRegion, keyID := t.GetKMSConfig(); enabled { conf.Set("security.encryption.data-encryption-method", "aes256-ctr") conf.Set("security.encryption.data-key-rotation-period", "7d") conf.Set("security.encryption.master-key.type", "kms") conf.Set("security.encryption.master-key.key-id", keyID) conf.Set("security.encryption.master-key.region", keyRegion) } return conf } func (t *TiDBCluster) buildTiKVConfig(tc *opv1alpha1.TidbCluster, cc *api.DescribeClusterConfigOutput) error { tikvConfig, ok := getComponentConfig(cc, api.ComponentTiKV) if !ok { return errors.New("no TiKV configuration provided") } topo := t.Cluster.Spec.Cluster tikvResource, err := t.ConfigProvider.GetResourceSpec(t.Provider, &topo.TiKV) if err != nil { return errors.Annotate(err, "get tikv resources spec") } tikvSpec := &opv1alpha1.TiKVSpec{ ComponentSpec: opv1alpha1.ComponentSpec{ Affinity: utils.Affinity(cloudv1.ComponentTiKV), NodeSelector: t.GetComponentNodeLabels(cloudv1.ComponentTiKV), Tolerations: utils.Toleration(cloudv1.ComponentTiKV), // cross region require tikv use hostnetwork, cdc will use pod ip to access tikv HostNetwork: ptr.To(true), PodSecurityContext: &corev1.PodSecurityContext{ Sysctls: []corev1.Sysctl{ { Name: "net.core.somaxconn", Value: "32768", }, }, }, Annotations: map[string]string{ consts.AnnSysctlInit: consts.AnnSysctlInitVal, }, Version: tikvConfig.Image.Tag, }, ResourceRequirements: tikvResource.Resources, BaseImage: tikvConfig.Image.BaseImage, Replicas: int32(topo.TiKV.Replicas), Privileged: ptr.To(false), MaxFailoverCount: ptr.To(int32(0)), StorageClassName: firstNonEmpty(topo.TiKV.StorageClassName, tikvResource.StorageClass), Config: t.computeTiKVConfig(cc), } tc.Spec.TiKV = tikvSpec return nil } // Overlay implements the provider.Cluster interface. func (t *TiDBCluster) Overlay(clusterConfig *api.DescribeClusterConfigOutput, tc *opv1alpha1.TidbCluster) error { fs := []func(t *TiDBCluster, tc *opv1alpha1.TidbCluster) error{ lo.Partial2(overlayTiFlashAndTiProxy, clusterConfig), overlaySuspendActions, overlayTiKVStorageVolumes, overlayTiKVForDedicatedRaftEngineStorage, overlayTiKVKmsCredentials, } switch t.Config.Provider { case consts.ProviderTypeGKE: // Must after overlayTiFlash fs = append(fs, overlayNodeSelectorWithInstanceType) if t.Cluster.FeatureEnabled(consts.ClusterFeatureGCP3LayerNodePool) { fs = append(fs, overlay3LayerArchToleration) fs = append(fs, overlay3LayerArchNodeSelector) } fs = append(fs, overlayArm64Toleration) case consts.ProviderTypeAKS: // Must after overlayTiFlash fs = append(fs, overlayNodeSelectorWithInstanceType) } for _, f := range fs { if err := f(t, tc); err != nil { return err } } return nil } func overlay3LayerArchNodeSelector(_ *TiDBCluster, tc *opv1alpha1.TidbCluster) error { if tc.Spec.PD != nil { tc.Spec.PD.NodeSelector[consts.LabelKeyFor3LayerArch] = consts.LabelValTrue } if tc.Spec.TiDB != nil { tc.Spec.TiDB.NodeSelector[consts.LabelKeyFor3LayerArch] = consts.LabelValTrue } if tc.Spec.TiKV != nil { tc.Spec.TiKV.NodeSelector[consts.LabelKeyFor3LayerArch] = consts.LabelValTrue } if tc.Spec.TiProxy != nil { tc.Spec.TiProxy.NodeSelector[consts.LabelKeyFor3LayerArch] = consts.LabelValTrue } if tc.Spec.TiFlash != nil { tc.Spec.TiFlash.NodeSelector[consts.LabelKeyFor3LayerArch] = consts.LabelValTrue } return nil } func overlay3LayerArchToleration(_ *TiDBCluster, tc *opv1alpha1.TidbCluster) error { if tc.Spec.PD != nil { tc.Spec.PD.Tolerations = append( tc.Spec.PD.Tolerations, *utils.TolerationWith3LayerArch(), ) } if tc.Spec.TiDB != nil { tc.Spec.TiDB.Tolerations = append( tc.Spec.TiDB.Tolerations, *utils.TolerationWith3LayerArch(), ) } if tc.Spec.TiKV != nil { tc.Spec.TiKV.Tolerations = append( tc.Spec.TiKV.Tolerations, *utils.TolerationWith3LayerArch(), ) } if tc.Spec.TiProxy != nil { tc.Spec.TiProxy.Tolerations = append( tc.Spec.TiProxy.Tolerations, *utils.TolerationWith3LayerArch(), ) } if tc.Spec.TiFlash != nil { tc.Spec.TiFlash.Tolerations = append( tc.Spec.TiFlash.Tolerations, *utils.TolerationWith3LayerArch(), ) } return nil } // overlayArm64Toleration adds kubernetes.io/arch=arm64:NoSchedule toleration // for GKE ARM instances (e.g. c4a-*). func overlayArm64Toleration(t *TiDBCluster, tc *opv1alpha1.TidbCluster) error { type compRef struct { instanceType string tolerations *[]corev1.Toleration } var comps []compRef if tc.Spec.PD != nil { comps = append(comps, compRef{t.Cluster.Spec.Cluster.PD.InstanceType, &tc.Spec.PD.Tolerations}) } if tc.Spec.TiDB != nil { comps = append(comps, compRef{t.Cluster.Spec.Cluster.TiDB.InstanceType, &tc.Spec.TiDB.Tolerations}) } if tc.Spec.TiKV != nil { comps = append(comps, compRef{t.Cluster.Spec.Cluster.TiKV.InstanceType, &tc.Spec.TiKV.Tolerations}) } if tc.Spec.TiProxy != nil { comps = append(comps, compRef{t.Cluster.Spec.Cluster.TiProxy.InstanceType, &tc.Spec.TiProxy.Tolerations}) } if tc.Spec.TiFlash != nil { comps = append(comps, compRef{t.Cluster.Spec.Cluster.TiFlash.InstanceType, &tc.Spec.TiFlash.Tolerations}) } for _, c := range comps { if c.instanceType == "" { continue } it, err := t.Provider.FindInstanceType(c.instanceType) if err != nil { return err } if it.Arch == consts.ARM64ArchName { *c.tolerations = append(*c.tolerations, *utils.TolerationWithArch(consts.ARM64ArchName)) } } return nil } func overlayNodeSelectorWithInstanceType(t *TiDBCluster, tc *opv1alpha1.TidbCluster) error { disableNodePoolDataPlaneLabel := t.Cluster.FeatureEnabled(consts.ClusterFeatureDisableNodePoolDataPlaneLabel) if tc.Spec.PD != nil { it := t.Cluster.Spec.Cluster.PD.InstanceType tc.Spec.PD.NodeSelector = map[string]string{ consts.InstanceTypeLabelKey: it, } if !disableNodePoolDataPlaneLabel { tc.Spec.PD.NodeSelector[consts.LabelKeyNodeGroupType] = consts.LabelValNodeGroupTypeDataPlane } tc.Spec.PD.Affinity.NodeAffinity = &corev1.NodeAffinity{} tc.Spec.PD.Tolerations = append( tc.Spec.PD.Tolerations, utils.TolerationWithComponentShared()..., ) } if tc.Spec.TiDB != nil { it := t.Cluster.Spec.Cluster.TiDB.InstanceType tc.Spec.TiDB.NodeSelector = map[string]string{ consts.InstanceTypeLabelKey: it, } if !disableNodePoolDataPlaneLabel { tc.Spec.TiDB.NodeSelector[consts.LabelKeyNodeGroupType] = consts.LabelValNodeGroupTypeDataPlane } tc.Spec.TiDB.Affinity.NodeAffinity = &corev1.NodeAffinity{} tc.Spec.TiDB.Tolerations = append( tc.Spec.TiDB.Tolerations, utils.TolerationWithComponentShared()..., ) } if tc.Spec.TiKV != nil { it := t.Cluster.Spec.Cluster.TiKV.InstanceType tc.Spec.TiKV.NodeSelector = map[string]string{ consts.InstanceTypeLabelKey: it, } if !disableNodePoolDataPlaneLabel { tc.Spec.TiKV.NodeSelector[consts.LabelKeyNodeGroupType] = consts.LabelValNodeGroupTypeDataPlane } tc.Spec.TiKV.Affinity.NodeAffinity = &corev1.NodeAffinity{} tc.Spec.TiKV.Tolerations = append( tc.Spec.TiKV.Tolerations, utils.TolerationWithComponentShared()..., ) } if tc.Spec.TiProxy != nil { it := t.Cluster.Spec.Cluster.TiProxy.InstanceType tc.Spec.TiProxy.NodeSelector = map[string]string{ consts.InstanceTypeLabelKey: it, } if !disableNodePoolDataPlaneLabel { tc.Spec.TiProxy.NodeSelector[consts.LabelKeyNodeGroupType] = consts.LabelValNodeGroupTypeDataPlane } tc.Spec.TiProxy.Affinity.NodeAffinity = &corev1.NodeAffinity{} tc.Spec.TiProxy.Tolerations = append( tc.Spec.TiProxy.Tolerations, utils.TolerationWithComponentShared()..., ) } if tc.Spec.TiFlash != nil { it := t.Cluster.Spec.Cluster.TiFlash.InstanceType tc.Spec.TiFlash.NodeSelector = map[string]string{ consts.InstanceTypeLabelKey: it, } if !disableNodePoolDataPlaneLabel { tc.Spec.TiFlash.NodeSelector[consts.LabelKeyNodeGroupType] = consts.LabelValNodeGroupTypeDataPlane } tc.Spec.TiFlash.Affinity.NodeAffinity = &corev1.NodeAffinity{} tc.Spec.TiFlash.Tolerations = append( tc.Spec.TiFlash.Tolerations, utils.TolerationWithComponentShared()..., ) } return nil } func OverlayTiProxy(baseTask *task.BaseTask, c *ClusterContext, tiproxySpec *cloudv1.TiProxy, tc *opv1alpha1.TidbCluster, clusterConfig *api.DescribeClusterConfigOutput) error { logger := baseTask.Logger // TODO(ideasf) wantTiproxy := tiproxySpec.Replicas > 0 if !wantTiproxy { tc.Spec.TiProxy = nil return nil } var tiproxyConfig api.ComponentOutput found := false for _, comp := range clusterConfig.Components { if comp.Component == api.ComponentTiProxy { tiproxyConfig = comp found = true break } } if !found { logger.Info("want a tiproxy but failed to get the config") return errors.New("tiproxy config not found in cluster config") } tc.Spec.TiProxy = &opv1alpha1.TiProxySpec{ ComponentSpec: opv1alpha1.ComponentSpec{ Version: tiproxyConfig.Image.Tag, HostNetwork: ptr.To(true), }, BaseImage: tiproxyConfig.Image.BaseImage, Config: opv1alpha1.NewTiProxyConfig(), Replicas: int32(tiproxySpec.Replicas), SSLEnableTiDB: true, // unsupported fields: // MaxFailoverCount: pointer.Int32Ptr(0), // Service: nil, // StorageClassName: firstNonEmpty(topo.TiProxy.StorageClassName, tiproxyResource.StorageClass), } if !c.Target.FeatureEnabled(consts.ClusterFeatureTiProxyCertLayoutLegacy) { tc.Spec.TiProxy.CertLayout = opv1alpha1.TiProxyCertLayoutV1 } if !c.Target.FeatureEnabled(consts.ClusterFeatureLegacyTiProxySettings) { tc.Spec.TiProxy.ReadinessProbe = &opv1alpha1.Probe{ // use command probe, curl with status API '/api/debug/health' Type: ptr.To(opv1alpha1.CommandProbeType), } } // copy config tc.Spec.TiProxy.Config.MP = tiproxyConfig.Config.MP // set node distribution proxySpec := tc.Spec.TiProxy setTiProxyPodAffinity(baseTask.Cluster, proxySpec) if true { // standalone deploy tiproxyResource, err := c.ConfigProvider.GetResourceSpec(c.Provider, tiproxySpec) if err != nil { return errors.Annotate(err, "get tiproxy resources spec") } proxySpec.NodeSelector = c.GetComponentNodeLabels(cloudv1.ComponentTiProxy) proxySpec.Tolerations = utils.Toleration(cloudv1.ComponentTiProxy) proxySpec.ResourceRequirements = tiproxyResource.Resources } // set tiproxy grace period k1, k2 := int64(15), int64(30) if v := tc.Spec.TiProxy.Config.Get("proxy.graceful-close-conn-timeout"); v != nil { if e, err := v.AsInt(); err == nil { k1 = e } } if v := tc.Spec.TiProxy.Config.Get("proxy.graceful-wait-before-shutdown"); v != nil { if e, err := v.AsInt(); err == nil { k2 = e } } tiproxyTerm := tiproxyTerminationGracePeriodSeconds // for compatibility with the existing clusters, we still use dynamic calculated value for terminationGracePeriodSeconds. if c.Target.FeatureEnabled(consts.ClusterFeatureLegacyTiProxySettings) { tiproxyTerm = defaultExtraTiProxyPodTerminationGracePeriodSeconds + k1 + k2 } tc.Spec.TiProxy.TerminationGracePeriodSeconds = &tiproxyTerm // mutate tidb { tidbGracefulShutdownWaitSeconds := configmanager.GetDefaultTidbGracefulShutdownWaitSeconds(baseTask.Cluster) if v := tc.Spec.TiDB.Config.Get(consts.TiDBGracefulShutdownWaitSecondsKey); v != nil { if e, err := v.AsInt(); err == nil { tidbGracefulShutdownWaitSeconds = e } } // extra `defaultExtraTiDBPodTerminationGracePeriodSeconds` is for tidb shutdown // and the sidecar `auditlog` which tries to push audit logs to remote storage. tidbTerm := defaultExtraTiDBPodTerminationGracePeriodSeconds + tidbGracefulShutdownWaitSeconds tc.Spec.TiDB.TerminationGracePeriodSeconds = &tidbTerm // Since tidb will not reload certs, but tiproxy reload certs properly. // So tiproxy may fail to verify tidbs' server cert when tiproxy reloads the new rotated tidb ca cert. // Set this to true, to let [pd, br, tiproxy] to skip verifying tidbs' old cert cert. // NOTE: this field is use by other components not the 'tidb', so that changing this field will not cause // tidb restart but cause pd, tiproxy and tiproxy etc to restart. tc.Spec.TiDB.TLSClient.SkipInternalClientCA = true } return nil } func setTiProxyPodAffinity(cluster *cloudv1.Cluster, proxySpec *opv1alpha1.TiProxySpec) { proxySpec.Affinity = utils.Affinity(cloudv1.ComponentTiProxy) // When there are only one/two tiproxy pods and one/two tidb pods, we want to run tiproxy in the same zone with tidb. // This is to avoid the case that tiproxy is scheduled to a different zone with tidb, which causes data transfer fee increase. aff := proxySpec.Affinity if aff.PodAffinity == nil { aff.PodAffinity = &corev1.PodAffinity{} } // NOTE: there is only one weighted pod affinity term in the affinity for now, so we can directly assign the value to the slice. aff.PodAffinity.PreferredDuringSchedulingIgnoredDuringExecution = []corev1.WeightedPodAffinityTerm{ { Weight: 100, PodAffinityTerm: corev1.PodAffinityTerm{ LabelSelector: &metav1.LabelSelector{ MatchLabels: map[string]string{ consts.ComponentLabelKey: string(cloudv1.ComponentTiDB), }, }, TopologyKey: corev1.LabelTopologyZone, }, }, } } func overlayTiKVKmsCredentials(t *TiDBCluster, tc *opv1alpha1.TidbCluster) error { cc, err := t.buildClusterContext() if err != nil { return err } return cc.OverlayTiKVKmsCredentials(tc) } func overlayTiFlashAndTiProxy(clusterConfig *api.DescribeClusterConfigOutput, t *TiDBCluster, tc *opv1alpha1.TidbCluster) error { cc, err := t.buildClusterContext() if err != nil { return err } err = OverlayTiProxy(t.BaseTask, cc, &cc.Target.Spec.Cluster.TiProxy, tc, clusterConfig) if err != nil { return err } err = cc.OverlayTiFlash(tc, clusterConfig) if err != nil { return err } return nil } func (t *TiDBCluster) mutateTcForReplicas(ctx context.Context, cli client.Client, currentTc, newTc *opv1alpha1.TidbCluster, comps []cloudv1.ComponentType) error { err := t.preMutateTcReplicas(ctx, cli, currentTc, newTc, comps) if err != nil { return err } var tikvTiflashScaleIn bool for _, comp := range comps { switch comp { case cloudv1.ComponentPD: currentTc.Spec.PD.Replicas = newTc.Spec.PD.Replicas case cloudv1.ComponentTiKV: if currentTc.Spec.TiKV.Replicas > newTc.Spec.TiKV.Replicas { diff := currentTc.Spec.TiKV.Replicas - newTc.Spec.TiKV.Replicas currentTc.Spec.TiKV.ScalePolicy.ScaleInParallelism = &diff tikvTiflashScaleIn = true } currentTc.Spec.TiKV.Replicas = newTc.Spec.TiKV.Replicas case cloudv1.ComponentTiDB: if !t.Cluster.FeatureEnabled(consts.ClusterFeatureNoScaleOutTiDBInParallel) { if newTc.Spec.TiDB.Replicas > currentTc.Spec.TiDB.Replicas { diff := newTc.Spec.TiDB.Replicas - currentTc.Spec.TiDB.Replicas currentTc.Spec.TiDB.ScalePolicy.ScaleOutParallelism = &diff } } else if currentTc.Spec.TiDB.ScalePolicy.ScaleOutParallelism != nil && *currentTc.Spec.TiDB.ScalePolicy.ScaleOutParallelism > 1 { // Reset ScaleOutParallelism to the default value (1) if the feature gate is not enabled. currentTc.Spec.TiDB.ScalePolicy.ScaleOutParallelism = ptr.To[int32](1) } currentTc.Spec.TiDB.Replicas = newTc.Spec.TiDB.Replicas case cloudv1.ComponentTiProxy: var currentReplicas, desiredReplicas int32 if currentTc.Spec.TiProxy != nil { currentReplicas = currentTc.Spec.TiProxy.Replicas } if newTc.Spec.TiProxy != nil { desiredReplicas = newTc.Spec.TiProxy.Replicas } if currentTc.Spec.TiProxy == nil || currentTc.Spec.TiProxy.Replicas == 0 { if newTc.Spec.TiProxy != nil { currentTc.Spec.TiProxy = newTc.Spec.TiProxy // scaling from 0 } } if currentTc.Spec.TiProxy != nil { if newTc.Spec.TiProxy == nil { currentTc.Spec.TiProxy.Replicas = 0 } else { currentTc.Spec.TiProxy.Replicas = desiredReplicas } } ensureGKEV1MinDomainsOnActivation( t.Cluster, currentTc, cloudv1.ComponentTiProxy, currentReplicas, desiredReplicas, ) case cloudv1.ComponentTiFlash: var currentReplicas, desiredReplicas int32 if currentTc.Spec.TiFlash != nil { currentReplicas = currentTc.Spec.TiFlash.Replicas } if newTc.Spec.TiFlash != nil { desiredReplicas = newTc.Spec.TiFlash.Replicas } if currentTc.Spec.TiFlash == nil || currentTc.Spec.TiFlash.Replicas == 0 { if newTc.Spec.TiFlash != nil { currentTc.Spec.TiFlash = newTc.Spec.TiFlash // scaling from 0 } } if currentTc.Spec.TiFlash != nil { if newTc.Spec.TiFlash == nil { diff := currentTc.Spec.TiFlash.Replicas if diff > 0 { currentTc.Spec.TiFlash.ScalePolicy.ScaleInParallelism = &diff } currentTc.Spec.TiFlash.Replicas = 0 } else { if currentTc.Spec.TiFlash.Replicas > newTc.Spec.TiFlash.Replicas { diff := currentTc.Spec.TiFlash.Replicas - newTc.Spec.TiFlash.Replicas currentTc.Spec.TiFlash.ScalePolicy.ScaleInParallelism = &diff // special case: not set `tikvTiflashScaleIn` to true if scaling to 0 if newTc.Spec.TiFlash.Replicas != 0 { tikvTiflashScaleIn = true } } currentTc.Spec.TiFlash.Replicas = newTc.Spec.TiFlash.Replicas } } ensureGKEV1MinDomainsOnActivation( t.Cluster, currentTc, cloudv1.ComponentTiFlash, currentReplicas, desiredReplicas, ) } } if tikvTiflashScaleIn { // record PD low space ratio for reset after scaled in if pdLowSpaceRatio := t.getPDLowSpaceRatio(ctx); pdLowSpaceRatio != nil { if currentTc.Annotations == nil { currentTc.Annotations = make(map[string]string) } currentTc.Annotations[annoKeyPDLowSpaceRatio] = fmt.Sprintf("%f", *pdLowSpaceRatio) t.Logger.Info("record PD low-space-ratio before scale in TiKV/TiFlash", "ratio", *pdLowSpaceRatio) } } return nil } // waitForTiFlashScaleFromZeroPVCDeletion prevents a retained VAC-enabled PVC // from being reused while TiDB Operator recreates the TiFlash StatefulSet. func (t *TiDBCluster) waitForTiFlashScaleFromZeroPVCDeletion( ctx context.Context, cli client.Client, currentTc, desiredTc *opv1alpha1.TidbCluster, ) (bool, error) { current, desired := currentTc.Spec.TiFlash, desiredTc.Spec.TiFlash if !t.gkeVolumeAttributesClassEnabled() || current == nil || desired == nil || current.Replicas != 0 || desired.Replicas <= 0 || len(current.StorageClaims) == 0 || ptr.Deref(current.StorageClaims[0].VolumeAttributesClassName, "") == "" { return false, nil } pvcs, err := k8s.SelectPVCs( ctx, cli, currentTc.Namespace, t.GetComponentPodLabels(cloudv1.ComponentTiFlash), ) if err != nil { return false, fmt.Errorf("list TiFlash PVCs before scaling from zero: %w", err) } return len(pvcs) > 0, nil } func (t *TiDBCluster) preMutateTcReplicas(ctx context.Context, cli client.Client, currentTc, newTc *opv1alpha1.TidbCluster, comps []cloudv1.ComponentType) error { for _, comp := range comps { var current, desired int32 = 0, 0 switch comp { case cloudv1.ComponentPD: current = currentTc.Spec.PD.Replicas desired = newTc.Spec.PD.Replicas case cloudv1.ComponentTiKV: current = currentTc.Spec.TiKV.Replicas desired = newTc.Spec.TiKV.Replicas case cloudv1.ComponentTiDB: current = currentTc.Spec.TiDB.Replicas desired = newTc.Spec.TiDB.Replicas case cloudv1.ComponentTiProxy: if currentTc.Spec.TiProxy != nil { current = currentTc.Spec.TiProxy.Replicas } if newTc.Spec.TiProxy != nil { desired = newTc.Spec.TiProxy.Replicas } case cloudv1.ComponentTiFlash: if currentTc.Spec.TiFlash != nil { current = currentTc.Spec.TiFlash.Replicas } if newTc.Spec.TiFlash != nil { desired = newTc.Spec.TiFlash.Replicas } } err := scaleInComponentBalanced(ctx, cli, current, desired, currentTc, comp, t.Zones()) if err != nil { return err } } return nil } // scaleInComponentBalanced set delete slots to make pod balanced across zones when scale in happening // https://docs.pingcap.com/zh/tidb-in-kubernetes/stable/advanced-statefulset#%E6%93%8D%E4%BD%9C-tidbcluster-%E5%AF%B9%E8%B1%A1%E6%8C%87%E5%AE%9A-pod-%E8%BF%9B%E8%A1%8C%E7%BC%A9%E5%AE%B9 func scaleInComponentBalanced(ctx context.Context, cli client.Client, current, desired int32, tc *opv1alpha1.TidbCluster, ct cloudv1.ComponentType, zones []string) error { if current <= desired { return nil } key := types.NamespacedName{Namespace: tc.GetNamespace(), Name: tc.GetName()} topo, err := k8s.GetComponentTopology(ctx, cli, key, ct) if err != nil { return err } if len(topo.UnScheduled) > 0 { // this is an exception since we only scale cluster when cluster is normal return errors.New("unscheduled pods found, stop scale-in") } // Scale out not complete if int32(topo.Replicas) < current { return errors.Errorf("scale-out did not complete, tc spec %d, pods %d, stop scale-in, component: %v", current, topo.Replicas, ct) } scaleInCount := current - desired topoScheduled := k8s.PodsByZone{} for k, v := range topo.Scheduled { topoScheduled[k] = append([]string{}, v...) } // Find victims victimOrdinals := sets.New[int]() for i := int32(0); i < scaleInCount; i++ { victim, err := findOneScaleInVictim(zones, topoScheduled) if err != nil { return errors.Annotate(err, "scaleInOnePod") } victimOrdinals.Insert(victim.Ordinal) } currentSlots := GetDeleteSlots(tc, ct) intersection := victimOrdinals.Intersection(currentSlots) if intersection.Len() > 0 { return errors.Errorf("[%s] victims has intersection with current delete-slots %+v, which means the current observation is stale", ct, sets.List(intersection)) } newSlots := currentSlots.Union(victimOrdinals) err = SetDeleteSlots(tc, ct, newSlots) if err != nil { return errors.Annotate(err, "setDeleteSlots") } return nil } // findOneScaleInVictim select the victim to be scaled-in and remove it from current topology for following selection func findOneScaleInVictim(zones []string, spreads k8s.PodsByZone) (*utils.StatefulPod, error) { maxZoneIndex := 0 maxZoneCount := 0 // find the max zone to scale-in, latter zone wins on tie for i := range zones { if len(spreads[zones[i]]) >= maxZoneCount { maxZoneCount = len(spreads[zones[i]]) maxZoneIndex = i } } maxZone := zones[maxZoneIndex] maxZonePods := spreads[maxZone] victim := &utils.StatefulPod{Ordinal: -1} victimIndex := 0 // find the max ordinal to scale-in in the max zone for i, podName := range maxZonePods { pod, err := utils.NewStatefulPodFromName(podName) if err != nil { return nil, errors.Annotatef(err, "%s is not a statefulset pod", podName) } if pod.GetOrdinal() > victim.GetOrdinal() { victim = pod victimIndex = i } } if victim.GetOrdinal() < 0 { return nil, errors.New("no pod to scale-in") } // rm victim from current topology spreads[maxZone] = append(maxZonePods[:victimIndex], maxZonePods[victimIndex+1:]...) return victim, nil } func GetDeleteSlots(tc *opv1alpha1.TidbCluster, ct cloudv1.ComponentType) sets.Set[int] { deleteSlots := sets.New[int]() annotations := tc.GetAnnotations() if annotations == nil { return deleteSlots } value, ok := annotations[componentDeleteSlotsAnnMap[ct]] if !ok { return deleteSlots } var slice []int err := json.Unmarshal([]byte(value), &slice) if err != nil { return deleteSlots } deleteSlots.Insert(slice...) return deleteSlots } func SetDeleteSlots(tc *opv1alpha1.TidbCluster, ct cloudv1.ComponentType, slots sets.Set[int]) error { annotations := tc.GetAnnotations() if annotations == nil { annotations = make(map[string]string) } deleteSlotsAnn := componentDeleteSlotsAnnMap[ct] if slots == nil || slots.Len() == 0 { // clear delete(annotations, deleteSlotsAnn) } else { // set slotsList := sets.List(slots) sort.Ints(slotsList) b, err := json.Marshal(slotsList) if err != nil { return err } annotations[deleteSlotsAnn] = string(b) } tc.SetAnnotations(annotations) return nil } func (t *TiDBCluster) mutateTcForStorageSize(currentTc, newTc *opv1alpha1.TidbCluster, comps []cloudv1.ComponentType) { for _, comp := range comps { switch comp { case cloudv1.ComponentPD: currentTc.Spec.PD.Requests[corev1.ResourceStorage] = util.MaxQuantity( // no scale down currentTc.Spec.PD.Requests[corev1.ResourceStorage], newTc.Spec.PD.Requests[corev1.ResourceStorage], ) case cloudv1.ComponentTiKV: currentTc.Spec.TiKV.Requests[corev1.ResourceStorage] = util.MaxQuantity( currentTc.Spec.TiKV.Requests[corev1.ResourceStorage], newTc.Spec.TiKV.Requests[corev1.ResourceStorage], ) if len(currentTc.Spec.TiKV.StorageVolumes) > 0 { // do not support remove volumes now newVolumeMap := convertVolumeSliceToMap(newTc.Spec.TiKV.StorageVolumes) for i, volume := range currentTc.Spec.TiKV.StorageVolumes { if newVolume, ok := newVolumeMap[volume.Name]; ok { maxSize, err := util.MaxQuantityStr(volume.StorageSize, newVolume.StorageSize) if err != nil { t.Logger.Error(err, "failed to calculate max storage size", "volume", volume.Name) continue } currentTc.Spec.TiKV.StorageVolumes[i].StorageSize = maxSize } } } case cloudv1.ComponentTiFlash: if newTc.Spec.TiFlash != nil { currentTc.Spec.TiFlash.StorageClaims[0].Resources.Requests[corev1.ResourceStorage] = util.MaxQuantity( currentTc.Spec.TiFlash.StorageClaims[0].Resources.Requests[corev1.ResourceStorage], newTc.Spec.TiFlash.StorageClaims[0].Resources.Requests[corev1.ResourceStorage], ) } } } } // mutateTcForVolumeClasses copies StorageClass changes and prepared GKE VAC assignments. func (t *TiDBCluster) mutateTcForVolumeClasses(currentTc, newTc *opv1alpha1.TidbCluster, comps []cloudv1.ComponentType) { for _, comp := range comps { switch comp { case cloudv1.ComponentPD: currentTc.Spec.PD.StorageClassName = newTc.Spec.PD.StorageClassName if t.gkeVolumeAttributesClassEnabled() && ptr.Deref(newTc.Spec.PD.VolumeAttributesClassName, "") != "" { currentTc.Spec.PD.VolumeAttributesClassName = newTc.Spec.PD.VolumeAttributesClassName } case cloudv1.ComponentTiKV: currentTc.Spec.TiKV.StorageClassName = newTc.Spec.TiKV.StorageClassName if t.gkeVolumeAttributesClassEnabled() && ptr.Deref(newTc.Spec.TiKV.VolumeAttributesClassName, "") != "" { currentTc.Spec.TiKV.VolumeAttributesClassName = newTc.Spec.TiKV.VolumeAttributesClassName } if len(currentTc.Spec.TiKV.StorageVolumes) > 0 { // do not support remove volumes now newVolumeMap := convertVolumeSliceToMap(newTc.Spec.TiKV.StorageVolumes) for i, volume := range currentTc.Spec.TiKV.StorageVolumes { if newVolume, ok := newVolumeMap[volume.Name]; ok { currentTc.Spec.TiKV.StorageVolumes[i].StorageClassName = newVolume.StorageClassName if t.gkeVolumeAttributesClassEnabled() && ptr.Deref(newVolume.VolumeAttributesClassName, "") != "" { currentTc.Spec.TiKV.StorageVolumes[i].VolumeAttributesClassName = newVolume.VolumeAttributesClassName } } } } case cloudv1.ComponentTiFlash: if newTc.Spec.TiFlash != nil { currentTc.Spec.TiFlash.StorageClaims[0].StorageClassName = newTc.Spec.TiFlash.StorageClaims[0].StorageClassName if t.gkeVolumeAttributesClassEnabled() && ptr.Deref(newTc.Spec.TiFlash.StorageClaims[0].VolumeAttributesClassName, "") != "" { currentTc.Spec.TiFlash.StorageClaims[0].VolumeAttributesClassName = newTc.Spec.TiFlash.StorageClaims[0].VolumeAttributesClassName } } } } } func (t *TiDBCluster) useClusterAutoscaler() bool { return t.Config.Provider == consts.ProviderTypeGKE || t.Config.Provider == consts.ProviderTypeAKS } func (t *TiDBCluster) mutateTcCompForInstanceType(ctx context.Context, currentTc, newTc *opv1alpha1.TidbCluster, comps []cloudv1.ComponentType, remoteCli client.Client) error { t.mutateTcForStorageSize(currentTc, newTc, comps) if err := t.ensureGKEVolumeAttributesClassesForModification(ctx, remoteCli, currentTc, newTc, comps); err != nil { return err } t.mutateTcForVolumeClasses(currentTc, newTc, comps) var ( nodeAffinityPolicy = corev1.NodeInclusionPolicyIgnore topoSpreadConstraint = opv1alpha1.TopologySpreadConstraint{ TopologyKey: "topology.kubernetes.io/zone", MaxSkew: 1, NodeAffinityPolicy: &nodeAffinityPolicy, } ) // set minDomains to zones count for azure cluster if t.Config.Provider == consts.ProviderTypeAKS { topoSpreadConstraint.MinDomains = ptr.To(int32(len(t.Zones()))) } for _, comp := range comps { switch comp { case cloudv1.ComponentPD: // mutate CPU/Memory requests/limits currentTc.Spec.PD.Requests[corev1.ResourceCPU] = newTc.Spec.PD.Requests[corev1.ResourceCPU] currentTc.Spec.PD.Requests[corev1.ResourceMemory] = newTc.Spec.PD.Requests[corev1.ResourceMemory] currentTc.Spec.PD.Limits[corev1.ResourceMemory] = newTc.Spec.PD.Limits[corev1.ResourceMemory] // no CPU limit if t.useClusterAutoscaler() { it := t.Cluster.Spec.Cluster.PD.InstanceType // `NodeSelector` should not be nil as it already initialized when creating tidbcluster currentTc.Spec.PD.NodeSelector[consts.InstanceTypeLabelKey] = it } case cloudv1.ComponentTiKV: currentTc.Spec.TiKV.Requests[corev1.ResourceCPU] = newTc.Spec.TiKV.Requests[corev1.ResourceCPU] currentTc.Spec.TiKV.Requests[corev1.ResourceMemory] = newTc.Spec.TiKV.Requests[corev1.ResourceMemory] currentTc.Spec.TiKV.Limits[corev1.ResourceMemory] = newTc.Spec.TiKV.Limits[corev1.ResourceMemory] if t.useClusterAutoscaler() { it := t.Cluster.Spec.Cluster.TiKV.InstanceType currentTc.Spec.TiKV.NodeSelector[consts.InstanceTypeLabelKey] = it // for GKE, it's better to set `NodeAffinityPolicy` to avoid scale up/down blocking caused by AZ imbalance. // ref: https://pingcap-cn.feishu.cn/wiki/H1qywaX0ei9lPqki2uMcRUMLn7d // NOTE: no need for PD as it only has 3 replicas across 3 AZs. // NOTE: should not set for CDC as it only deploys in 2 AZ. currentTc.Spec.TiKV.TopologySpreadConstraints = []opv1alpha1.TopologySpreadConstraint{topoSpreadConstraint} } case cloudv1.ComponentTiProxy: if newTc.Spec.TiProxy != nil { currentTc.Spec.TiProxy.NodeSelector = newTc.Spec.TiProxy.NodeSelector currentTc.Spec.TiProxy.Tolerations = newTc.Spec.TiProxy.Tolerations currentTc.Spec.TiProxy.StorageClassName = newTc.Spec.TiProxy.StorageClassName currentTc.Spec.TiProxy.Requests[corev1.ResourceCPU] = newTc.Spec.TiProxy.Requests[corev1.ResourceCPU] currentTc.Spec.TiProxy.Requests[corev1.ResourceMemory] = newTc.Spec.TiProxy.Requests[corev1.ResourceMemory] currentTc.Spec.TiProxy.Limits[corev1.ResourceMemory] = newTc.Spec.TiProxy.Limits[corev1.ResourceMemory] // NOTE: scale up TiProxy if needed if t.useClusterAutoscaler() { it := t.Cluster.Spec.Cluster.TiProxy.InstanceType currentTc.Spec.TiProxy.NodeSelector[consts.InstanceTypeLabelKey] = it currentTc.Spec.TiProxy.TopologySpreadConstraints = []opv1alpha1.TopologySpreadConstraint{topoSpreadConstraint} } } case cloudv1.ComponentTiDB: currentTc.Spec.TiDB.ResourceRequirements = newTc.Spec.TiDB.ResourceRequirements // no storage // reset `GOMEMLIMIT` when changing resource limit, ref `AssembleTidbClusterValues` // NOTE: we do not append if `GOMEMLIMIT` is not found, as the situation is unknown. tidbResource := newTc.Spec.TiDB.ResourceRequirements if tidbResource.Limits.Memory() != nil && !tidbResource.Limits.Memory().IsZero() { tidbGoLimit := int64(GOMEMLIMIT_RATE * float64(tidbResource.Limits.Memory().Value())) for i, env := range currentTc.Spec.TiDB.Env { if env.Name == "GOMEMLIMIT" { currentTc.Spec.TiDB.Env[i] = corev1.EnvVar{Name: "GOMEMLIMIT", Value: strconv.FormatInt(tidbGoLimit, 10)} break } } } if t.useClusterAutoscaler() { it := t.Cluster.Spec.Cluster.TiDB.InstanceType currentTc.Spec.TiDB.NodeSelector[consts.InstanceTypeLabelKey] = it currentTc.Spec.TiDB.TopologySpreadConstraints = []opv1alpha1.TopologySpreadConstraint{topoSpreadConstraint} } case cloudv1.ComponentTiFlash: if newTc.Spec.TiFlash != nil { currentTc.Spec.TiFlash.ResourceRequirements = newTc.Spec.TiFlash.ResourceRequirements // storage in `StorageClaims` if t.useClusterAutoscaler() { it := t.Cluster.Spec.Cluster.TiFlash.InstanceType currentTc.Spec.TiFlash.NodeSelector[consts.InstanceTypeLabelKey] = it } currentTc.Spec.TiFlash.TopologySpreadConstraints = []opv1alpha1.TopologySpreadConstraint{topoSpreadConstraint} } } } if err := applyInstanceTypeMinDomains( ctx, t.Cluster, remoteCli, currentTc, t.Cluster.GetDefaultGroupID(), comps..., ); err != nil { return errors.Wrap(err, "prepare GKE v1 minDomains for instance-type change") } return t.ConfigManager.UpdateConfig(ctx, t.Cluster, t.IsRegionalManaged(), configmanager.OverrideRevisionFinder(t.Local), currentTc, t.PDClient, t.removeUnavailablePD(comps)) } func (t *TiDBCluster) mutateTcForInstanceType(ctx context.Context, _ client.Client, currentTc, newTc *opv1alpha1.TidbCluster, remoteCli client.Client) error { // this annotation will be added before pods restarted and deleted after the nodegroup deletion called compAnno := t.Cluster.Annotations[consts.ClusterUpdateAnnotationKeyInstanceType] if compAnno == "" { return nil } comps := strings.Split(compAnno, consts.AnnoInstanceTypeSeparator) compsCloudV1 := make([]cloudv1.ComponentType, 0, len(comps)) t.Logger.Info("Updating TidbCluster config for instance type", "components", comps) for _, comp := range comps { switch cloudv1.ComponentType(comp) { case cloudv1.ComponentPD: compsCloudV1 = append(compsCloudV1, cloudv1.ComponentPD) if currentTc.Spec.PD.Annotations == nil { currentTc.Spec.PD.Annotations = make(map[string]string) } currentTc.Spec.PD.Annotations[consts.TcInstanceTypeAnnotationKey] = t.Cluster.Spec.Cluster.PD.InstanceType case cloudv1.ComponentTiKV: compsCloudV1 = append(compsCloudV1, cloudv1.ComponentTiKV) if currentTc.Spec.TiKV.Annotations == nil { currentTc.Spec.TiKV.Annotations = make(map[string]string) } currentTc.Spec.TiKV.Annotations[consts.TcInstanceTypeAnnotationKey] = t.Cluster.Spec.Cluster.TiKV.InstanceType case cloudv1.ComponentTiProxy: if currentTc.Spec.TiProxy != nil { compsCloudV1 = append(compsCloudV1, cloudv1.ComponentTiProxy) if currentTc.Spec.TiProxy.Annotations == nil { currentTc.Spec.TiProxy.Annotations = make(map[string]string) } currentTc.Spec.TiProxy.Annotations[consts.TcInstanceTypeAnnotationKey] = t.Cluster.Spec.Cluster.TiProxy.InstanceType } case cloudv1.ComponentTiDB: compsCloudV1 = append(compsCloudV1, cloudv1.ComponentTiDB) if currentTc.Spec.TiDB.Annotations == nil { currentTc.Spec.TiDB.Annotations = make(map[string]string) } currentTc.Spec.TiDB.Annotations[consts.TcInstanceTypeAnnotationKey] = t.Cluster.Spec.Cluster.TiDB.InstanceType case cloudv1.ComponentTiFlash: // `currentTc.Spec.TiFlash` should not be `nil` if we are here compsCloudV1 = append(compsCloudV1, cloudv1.ComponentTiFlash) if currentTc.Spec.TiFlash.Annotations == nil { currentTc.Spec.TiFlash.Annotations = make(map[string]string) } currentTc.Spec.TiFlash.Annotations[consts.TcInstanceTypeAnnotationKey] = t.Cluster.Spec.Cluster.TiFlash.InstanceType } } // update cluster's storage-size, storage-class, requests/limits and config (and image) // in one operation when restarting pods to avoid evicting-leaders, restarting-pods twice. return t.mutateTcCompForInstanceType(ctx, currentTc, newTc, compsCloudV1, remoteCli) } // useCMEKStorageClass will use CMEK storage class for TiDB cluster func (t *TiDBCluster) useCMEKStorageClass(tc *opv1alpha1.TidbCluster) error { if tc.Spec.PD != nil && tc.Spec.PD.StorageClassName != nil { tc.Spec.PD.StorageClassName = ptr.To(t.CMEKStorageClassMapper.Name(*tc.Spec.PD.StorageClassName)) } if tc.Spec.TiKV != nil { if tc.Spec.TiKV.StorageClassName != nil { tc.Spec.TiKV.StorageClassName = ptr.To(t.CMEKStorageClassMapper.Name(*tc.Spec.TiKV.StorageClassName)) } for i, vol := range tc.Spec.TiKV.StorageVolumes { if vol.StorageClassName == nil { continue } tc.Spec.TiKV.StorageVolumes[i].StorageClassName = ptr.To(t.CMEKStorageClassMapper.Name(*vol.StorageClassName)) } } if tc.Spec.TiDB != nil && tc.Spec.TiDB.StorageClassName != nil { tc.Spec.TiDB.StorageClassName = ptr.To(t.CMEKStorageClassMapper.Name(*tc.Spec.TiDB.StorageClassName)) } if tc.Spec.TiFlash != nil && tc.Spec.TiFlash.StorageClaims != nil { for i, claim := range tc.Spec.TiFlash.StorageClaims { if tc.Spec.TiFlash.StorageClaims[i].StorageClassName == nil { continue } tc.Spec.TiFlash.StorageClaims[i].StorageClassName = ptr.To(t.CMEKStorageClassMapper.Name(*claim.StorageClassName)) } } if tc.Spec.TiProxy != nil && tc.Spec.TiProxy.StorageClassName != nil { tc.Spec.TiProxy.StorageClassName = ptr.To(t.CMEKStorageClassMapper.Name(*tc.Spec.TiProxy.StorageClassName)) } return nil } func isComponentScalingUp(provider gitops.Provider, spec cloudv1.ClusterComponent, status cloudv1.ClusterComponentStatus) (bool, error) { if spec.GetInstanceType() == "" { return false, nil } desired, err := provider.FindInstanceType(spec.GetInstanceType()) if err != nil { return false, fmt.Errorf("failed to find the desired instance type %s: %v", spec.GetInstanceType(), err) } for _, it := range status.GetInstanceTypes() { if it == "" { continue } current, err := provider.FindInstanceType(it) if err != nil { return false, fmt.Errorf("failed to find the current instance type %s: %v", it, err) } if cloudv1.IsScalingUp(current, desired) { return true, nil } } return false, nil } func nonScaleUpComponents(cluster *cloudv1.Cluster, provider gitops.Provider) []cloudv1.ComponentType { db := cluster.Status.DB if db == nil { return []cloudv1.ComponentType{cloudv1.ComponentPD, cloudv1.ComponentTiKV, cloudv1.ComponentTiDB, cloudv1.ComponentTiFlash, cloudv1.ComponentTiCDC, cloudv1.ComponentTiProxy} } comps := make([]cloudv1.ComponentType, 0, 6) if db.PD == nil || func() bool { scalingUP, err := isComponentScalingUp(provider, &cluster.Spec.Cluster.PD, db.PD) return !scalingUP && err == nil }() { comps = append(comps, cloudv1.ComponentPD) } if db.TiDB == nil || func() bool { scalingUP, err := isComponentScalingUp(provider, &cluster.Spec.Cluster.TiDB, db.TiDB) return !scalingUP && err == nil }() { comps = append(comps, cloudv1.ComponentTiDB) } if db.TiKV == nil || func() bool { scalingUP, err := isComponentScalingUp(provider, &cluster.Spec.Cluster.TiKV, db.TiKV) return !scalingUP && err == nil }() { comps = append(comps, cloudv1.ComponentTiKV) } // Note that tiflash and tiproxy are optional components, // if we scale them from zero, we will consider them as non-scale-up components if db.TiFlash == nil || func() bool { scalingUP, err := isComponentScalingUp(provider, &cluster.Spec.Cluster.TiFlash, db.TiFlash) return !scalingUP && err == nil }() { comps = append(comps, cloudv1.ComponentTiFlash) } if db.TiProxy == nil || func() bool { scalingUP, err := isComponentScalingUp(provider, &cluster.Spec.Cluster.TiProxy, db.TiProxy) return !scalingUP && err == nil }() { comps = append(comps, cloudv1.ComponentTiProxy) } comps = append(comps, cloudv1.ComponentTiCDC) return comps } func replicasChanged(stsStatus *appsv1.StatefulSetStatus, desiredReplicas int32) bool { return stsStatus != nil && stsStatus.UpdatedReplicas != desiredReplicas && stsStatus.Replicas != 0 } func isModifyingComponentReplicas(tc *opv1alpha1.TidbCluster) bool { if tc.Spec.PD != nil && replicasChanged(tc.Status.PD.GetStatefulSet(), tc.Spec.PD.Replicas) { return true } if tc.Spec.TiDB != nil && replicasChanged(tc.Status.TiDB.GetStatefulSet(), tc.Spec.TiDB.Replicas) { return true } if tc.Spec.TiKV != nil && replicasChanged(tc.Status.TiKV.GetStatefulSet(), tc.Spec.TiKV.Replicas) { return true } // Note that tiflash and tiproxy are optional components, // if we scale them from/to zero, we will NOT consider them as modifying replicas if tc.Spec.TiFlash != nil && replicasChanged(tc.Status.TiFlash.GetStatefulSet(), tc.Spec.TiFlash.Replicas) && tc.Spec.TiFlash.Replicas != 0 { return true } if tc.Spec.TiProxy != nil && replicasChanged(tc.Status.TiProxy.GetStatefulSet(), tc.Spec.TiProxy.Replicas) && tc.Spec.TiProxy.Replicas != 0 { return true } return false } func NonInstanceTypeScalingComponents(cluster *cloudv1.Cluster) []cloudv1.ComponentType { db := cluster.Status.DB if db == nil { return []cloudv1.ComponentType{cloudv1.ComponentPD, cloudv1.ComponentTiKV, cloudv1.ComponentTiDB, cloudv1.ComponentTiFlash, cloudv1.ComponentTiCDC, cloudv1.ComponentTiProxy} } comps := make([]cloudv1.ComponentType, 0) pdScaling := func() bool { if db.PD != nil { if db.PD.InstanceTypeScalePhase != cloudv1.InstanceTypeScalePhaseNone { return true } for _, member := range db.PD.Members { if member.InstanceType != "" && member.InstanceType != cluster.Spec.Cluster.PD.InstanceType { return true } } } return false } tikvScaling := func() bool { if db.TiKV != nil { if db.TiKV.InstanceTypeScalePhase != cloudv1.InstanceTypeScalePhaseNone { return true } for _, member := range db.TiKV.Members { if member.InstanceType != "" && !cluster.TiKVMemberInstanceTypeAllowed(member.InstanceType) { return true } } } return false } tidbScaling := func() bool { if db.TiDB != nil { if db.TiDB.InstanceTypeScalePhase != cloudv1.InstanceTypeScalePhaseNone { return true } for _, member := range db.TiDB.Members { if member.InstanceType != "" && !cluster.TiDBMemberInstanceTypeAllowed(member.InstanceType) { return true } } } return false } tiflashScaling := func() bool { if db.TiFlash != nil { if db.TiFlash.InstanceTypeScalePhase != cloudv1.InstanceTypeScalePhaseNone { return true } for _, member := range db.TiFlash.Members { if member.InstanceType != "" && member.InstanceType != cluster.Spec.Cluster.TiFlash.InstanceType { return true } } } return false } tiproxyScaling := func() bool { if db.TiProxy != nil { if db.TiProxy.InstanceTypeScalePhase != cloudv1.InstanceTypeScalePhaseNone { return true } for _, member := range db.TiProxy.Members { if member.InstanceType != "" && member.InstanceType != cluster.Spec.Cluster.TiProxy.InstanceType { return true } } } return false } if !pdScaling() { comps = append(comps, cloudv1.ComponentPD) } if !tikvScaling() { comps = append(comps, cloudv1.ComponentTiKV) } if !tidbScaling() { comps = append(comps, cloudv1.ComponentTiDB) } if !tiflashScaling() { comps = append(comps, cloudv1.ComponentTiFlash) } if !tiproxyScaling() { comps = append(comps, cloudv1.ComponentTiProxy) } comps = append(comps, cloudv1.ComponentTiCDC) return comps } func (t *TiDBCluster) removeUnavailablePD(comps []cloudv1.ComponentType) []cloudv1.ComponentType { if t.Cluster.Status.DB == nil || t.Cluster.Status.DB.PD == nil || t.Cluster.Status.DB.PD.Status == cloudv1.ComponentUnavailable { ret := make([]cloudv1.ComponentType, 0) for _, comp := range comps { if comp != cloudv1.ComponentPD { ret = append(ret, comp) } } return ret } return comps } // mutateTcForSuspendActions mutates the TidbCluster for suspend actions. func (t *TiDBCluster) mutateTcForSuspendActions(currentTc, newTc *opv1alpha1.TidbCluster) { if currentTc.Spec.TiDB != nil && newTc.Spec.TiDB != nil { currentTc.Spec.TiDB.SuspendAction = newTc.Spec.TiDB.SuspendAction } if currentTc.Spec.PD != nil && newTc.Spec.PD != nil { currentTc.Spec.PD.SuspendAction = newTc.Spec.PD.SuspendAction } if currentTc.Spec.TiKV != nil && newTc.Spec.TiKV != nil { currentTc.Spec.TiKV.SuspendAction = newTc.Spec.TiKV.SuspendAction } if currentTc.Spec.TiProxy != nil && newTc.Spec.TiProxy != nil { currentTc.Spec.TiProxy.SuspendAction = newTc.Spec.TiProxy.SuspendAction } if currentTc.Spec.TiFlash != nil && newTc.Spec.TiFlash != nil { currentTc.Spec.TiFlash.SuspendAction = newTc.Spec.TiFlash.SuspendAction } } func mutateTiDBForTokenBasedAuth(_ context.Context, cluster *cloudv1.Cluster, currentTc, _ *opv1alpha1.TidbCluster) { // NOTE: We don't support change tokenBasedAuthEnabled from true to false, refer to: https://pingcap.feishu.cn/docx/PwgZdyBvPou8I8xp7Iqcxy9XnKh // You should clear the jwks of this cluster by call core-svc `SetClusterAuthPubKey` // if you want to disable tokenBasedAuth for this cluster. if ptr.Deref(currentTc.Spec.TiDB.TokenBasedAuthEnabled, false) { return } // refer to: infra-api/controllers/cluster/tidbcluster.go:AssembleTidbClusterValues currentTc.Spec.TiDB.TokenBasedAuthEnabled = ptr.To(utils.TokenBasedAuthEnabled(cluster, currentTc.Spec.Version)) } // mutateTiDBAndTiProxyForTiProxy mutates the TiDB when the TiProxy is enabled. func mutateTiDBAndTiProxyForTiProxy(cluster *cloudv1.Cluster, tiproxySpec *cloudv1.TiProxy, currentTc, newTc *opv1alpha1.TidbCluster, groupId string) { // Note: // 1. We don't rollback the following changes when TiProxy is being disabled to avoid tidb restart. // 2. Get the replicas from the `tiproxySpec`, not the `cluster.Spec.Cluster.TiProxy`, // because this function may be called for tidbgroups. if tiproxySpec.Replicas <= 0 { return } // To reduce impact of TiDB restart, change these fields only when the LB target is already on TiProxy. if !cluster.FeatureEnabled(consts.ClusterFeatureDisableSyncTiDBSpecChangesForEnablingTiproxy) && cluster.Annotations[consts.AnnoKeyLBTargetOnTiProxy(groupId)] == consts.AnnoValTrue { currentTc.Spec.TiDB.TerminationGracePeriodSeconds = newTc.Spec.TiDB.TerminationGracePeriodSeconds currentTc.Spec.TiDB.TLSClient.SkipInternalClientCA = newTc.Spec.TiDB.TLSClient.SkipInternalClientCA } if !cluster.FeatureEnabled(consts.ClusterFeatureDisableScheduleTiProxyInTheSameZoneWithTiDB) { currentTc.Spec.TiProxy.Affinity = newTc.Spec.TiProxy.Affinity } currentTc.Spec.TiProxy.CertLayout = newTc.Spec.TiProxy.CertLayout currentTc.Spec.TiProxy.TerminationGracePeriodSeconds = newTc.Spec.TiProxy.TerminationGracePeriodSeconds currentTc.Spec.TiProxy.ReadinessProbe = newTc.Spec.TiProxy.ReadinessProbe // TODO(ideascf): move to a better place. // These changes don't cause restarts of TiDB/TiProxy, so that we can move them to a more common place. currentTc.Spec.TiProxy.ServerLabels = map[string]string{ consts.ServerLabelKeyGroup: newTc.Name, } currentTc.Spec.TiDB.ServerLabels = map[string]string{ consts.ServerLabelKeyGroup: newTc.Name, } } // when scaling in TiKV or TiFlash, PD will choose schedule targets for regions based on `low-space-ratio` and `high-space-ratio`. // we have a pre-check for the remaining storage based on `low-space-ratio` in core-svc, // but the scaling operation may still be blocked if more new data is written, data compression ratio is changed, the data is not evenly distributed, etc. // so we set a greater value for `low-space-ratio` before scaling in and reset it after scaling in. // ref: // - https://github.com/tikv/pd/blob/c55e632b164b5c4d339f0d6247173b8a8200a92f/pkg/core/store.go#L421 // - https://github.com/tidbcloud/core-svc/pull/737 // NOTE: GitOps may also modify the `low-space-ratio` value. // // In each reconcile loop, we will: // 1. query the current `low-space-ratio` value from PD // - this value may be set by GitOps or by this cluster controller or just the default value // 2. record the current `low-space-ratio` value in the annotation of the TidbCluster if and ONLY if a scale in operation will be triggered for TiKV/TiFlash // - if apply TidbCluster to K8s failed, retry should work as no new value if updated to PD // 3. set `low-space-ratio` to a greater value if TiKV/TiFlash is scaling in and the current value (may be set by GitOps) is less than the target value // - if the current value is greater than or equal to the target, do nothing, as it may be set by GitOps // - scaling in check: len(members) > replicas-in-spec now // 4. reset `low-space-ratio` to the value recorded in the annotation of the TidbCluster if TiKV/TiFlash is normal and the current value is greater than the recorded value // - if the current value is less than or equal to the recorded value, do nothing, as it may be set by GitOps // // When scaling in, the max(target value, GitOps value) will be used. // Otherwise, the min(current value, recorded value) will be used. func (t *TiDBCluster) getPDLowSpaceRatio(ctx context.Context) *float64 { if t.PDClient == nil { return nil } // NOTE: `t.PDClient.GetConfig` may get a stale config, so we use HTTP API directly body, err := t.PDClient.HTTPGet(ctx, pdScheduleConfigPath) if err != nil { metrics.ConnectToClusterProxyFailed.WithLabelValues(t.Cluster.GetProjectID(), t.EnvProvider.GetRegion(), err.Error()).Inc() t.Logger.Error(err, "failed to get PD config") return nil } config := map[string]interface{}{} err = json.Unmarshal(body, &config) if err != nil { t.Logger.Error(err, "failed to unmarshal PD config", "config", string(body)) return nil } if schedule, ok := config["low-space-ratio"]; ok { if v, ok := schedule.(float64); ok { return &v } } return nil } func (t *TiDBCluster) setPDLowSpaceRatio(ctx context.Context) error { if t.PDClient == nil || t.Tidbcluster == nil || t.Cluster.Status.DB == nil || (t.Cluster.Status.Phase != cloudv1.ClusterPhaseScaling && t.Cluster.Status.Phase != cloudv1.ClusterPhaseRunning) || (t.Cluster.Status.DB.TiKV == nil && t.Cluster.Status.DB.TiFlash == nil) { return nil } // only set/reset PD low space ratio if the previous value is recorded in the annotation var ( previous float64 err error ) if v, ok := t.Tidbcluster.Annotations[annoKeyPDLowSpaceRatio]; !ok { return nil } else if previous, err = strconv.ParseFloat(v, 64); err != nil { t.Logger.Error(err, "failed to parse previous low space ratio", "previous", v) return nil } current := t.getPDLowSpaceRatio(ctx) if current == nil { return nil } putPDLowSpaceRatio := func(lowSpaceRatio float64) error { if t.Config.DryRun.Enabled { return fmt.Errorf("[DryRun] Cannot put low space ratio to %v, current is %v, previous is %v", lowSpaceRatio, current, previous) } body := map[string]interface{}{ "schedule.low-space-ratio": lowSpaceRatio, } data, err := json.Marshal(body) if err != nil { return err } // NOTE: `t.PDClient.SetConfig` may NOT work for `low-space-ratio` (from 0.9 to 0.8), so we use HTTP API directly return metrics.PDClientErrorHandler(t.PDClient.HTTPPost(ctx, pdConfigPrefix, data), t.Cluster.GetProjectID(), t.EnvProvider.GetRegion()) } switch t.Cluster.Status.Phase { case cloudv1.ClusterPhaseScaling: if *current >= pdLowSpaceRatioInScalingIn { return nil } tikvStatus := t.Cluster.Status.DB.TiKV tiflashStatus := t.Cluster.Status.DB.TiFlash var scalingIn bool if tikvStatus != nil && tikvStatus.Status == cloudv1.ComponentScaling && len(tikvStatus.Members) > int(t.Tidbcluster.Spec.TiKV.Replicas) { scalingIn = true } if tiflashStatus != nil && tiflashStatus.Status == cloudv1.ComponentScaling && len(tiflashStatus.Members) > int(t.Tidbcluster.Spec.TiFlash.Replicas) && t.Tidbcluster.Spec.TiFlash.Replicas > 0 { // special case: not set `low-space-ratio` if scaling to 0 scalingIn = true } if scalingIn { if err := putPDLowSpaceRatio(pdLowSpaceRatioInScalingIn); err != nil { return err } t.Logger.Info("set PD low-space-ratio when scaling in TiKV/TiFlash", "from", *current, "to", pdLowSpaceRatioInScalingIn) } case cloudv1.ClusterPhaseRunning: if *current <= previous || *current != pdLowSpaceRatioInScalingIn { // if a different value is set by GitOps, do nothing return nil } if err := putPDLowSpaceRatio(previous); err != nil { return err } t.Logger.Info("reset PD low-space-ratio after scaled in TiKV/TiFlash", "from", *current, "to", previous) } return nil } func setMigrateDeleteSlotsAnn(cluster *cloudv1.Cluster, tc *opv1alpha1.TidbCluster) { if !tc.CreationTimestamp.IsZero() { panic("should not set it after tc created") } keyMap := map[string]string{ v1alpha1.MigrateAnnTiDBDeleteSlots: oplabel.AnnTiDBDeleteSlots, v1alpha1.MigrateAnnPDDeleteSlots: oplabel.AnnPDDeleteSlots, v1alpha1.MigrateAnnTiKVDeleteSlots: oplabel.AnnTiKVDeleteSlots, v1alpha1.MigrateAnnTiFlashDeleteSlots: oplabel.AnnTiFlashDeleteSlots, } for infraKey, tcKey := range keyMap { if val, ok := cluster.Annotations[infraKey]; ok { if tc.Annotations == nil { tc.Annotations = map[string]string{} } tc.Annotations[tcKey] = val } } } func convertVolumeSliceToMap(volumes []opv1alpha1.StorageVolume) map[string]opv1alpha1.StorageVolume { volumeMap := make(map[string]opv1alpha1.StorageVolume) for _, volume := range volumes { volumeMap[volume.Name] = volume } return volumeMap } // setPreStopHook sets the pre-stop hook for TiDB to sleep for a while before stopping. // It also sets the termination grace period to the sum of the original termination grace period and the sleep time. func setPreStopHook(ctx context.Context, cluster *cloudv1.Cluster, groupId string, tc *opv1alpha1.TidbCluster, cfg config.ClusterControllerConfig) { logger := log.FromContext(ctx).WithValues("cluster", cluster.Name, "groupId", groupId) sleepSecondsInCfg := cfg.GetTiDBPreStopHookSleepSeconds() sleepSeconds := cluster.GetTiDBPreStopHookSleepSeconds(ctx, groupId, sleepSecondsInCfg) tc.Spec.TiDB.Lifecycle = &corev1.Lifecycle{ PreStop: &corev1.LifecycleHandler{ Exec: &corev1.ExecAction{ Command: []string{ "sleep", fmt.Sprintf("%d", sleepSeconds), }, }, }, } if tc.Spec.TiDB.TerminationGracePeriodSeconds == nil { tc.Spec.TiDB.TerminationGracePeriodSeconds = ptr.To(consts.DefaultPodTerminationGracePeriodSeconds + sleepSeconds) } logger.Info("Setting pre-stop hook for TiDB", "sleepSeconds", sleepSeconds, "sleepSecondsInCfg", sleepSecondsInCfg, "terminationGracePeriodSeconds", tc.Spec.TiDB.TerminationGracePeriodSeconds, ) } // Enable PPv2 For TiDB (and TiProxy) func mutateTcForComponentsEnablePPv2(_ context.Context, cluster *cloudv1.Cluster, currentTc *opv1alpha1.TidbCluster) { // So far this annotation is only set by the ops workflow. if cluster.FeatureEnabled(consts.ClusterFeatureUsePPv2TiDBComponents) { // https://docs.pingcap.com/zh/tidb/stable/tidb-configuration-file/#networks // The `fallbackable` config does not work on TiDB versions lower than v6.5.1. currentTc.Spec.TiDB.Config.Set("proxy-protocol.fallbackable", true) currentTc.Spec.TiDB.Config.Set("proxy-protocol.networks", "*") } if cluster.FeatureEnabled(consts.ClusterFeatureUsePPv2TiProxy) { if currentTc.Spec.TiProxy != nil && currentTc.Spec.TiProxy.Replicas > 0 { // https://docs.pingcap.com/tidb/stable/tiproxy-configuration/#proxy-protocol currentTc.Spec.TiProxy.Config.Set("proxy.proxy-protocol", "v2") } } } // mutateTiFlashForGracefulShutdown configures graceful shutdown settings for TiFlash component. // This function sets up the termination grace period and enables sidecar logging for TiFlash. // The sidecar configuration is based on the enhancement described in: // https://github.com/pingcap/tidb-operator/pull/6325 func mutateTiFlashForGracefulShutdown(ctx context.Context, cluster *cloudv1.Cluster, tc *opv1alpha1.TidbCluster, obj *opv1alpha1.TidbCluster) { logger := log.FromContext(ctx).WithValues("cluster", cluster.Name) // Configure termination grace period and sidecar setting from source if tc.Spec.TiFlash != nil && obj.Spec.TiFlash != nil { obj.Spec.TiFlash.TerminationGracePeriodSeconds = tc.Spec.TiFlash.TerminationGracePeriodSeconds obj.Spec.TiFlash.LogTailer = tc.Spec.TiFlash.LogTailer logger.Info("mutate TiFlash graceful shutdown setting", "cluster", cluster.Name) } }