// Copyright 2018 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, // See the License for the specific language governing permissions and // limitations under the License. package member import ( "context" "encoding/hex" "encoding/json" "fmt" "path" "regexp" "strconv" "strings" "sync" "time" mngerutils "github.com/pingcap/tidb-operator/pkg/manager/utils" "github.com/pingcap/advanced-statefulset/client/apis/apps/v1/helper" "github.com/pingcap/tidb-operator/pkg/apis/label" "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1" "github.com/pingcap/tidb-operator/pkg/apis/util/toml" "github.com/pingcap/tidb-operator/pkg/controller" "github.com/pingcap/tidb-operator/pkg/manager/coordinator" "github.com/pingcap/tidb-operator/pkg/manager/member/startscript" "github.com/pingcap/tidb-operator/pkg/pdapi" "github.com/pingcap/tidb-operator/pkg/pdapi/placement" "github.com/pingcap/tidb-operator/pkg/util" "github.com/Masterminds/semver" apps "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" v1 "k8s.io/api/core/v1" apiequality "k8s.io/apimachinery/pkg/api/equality" "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/util/sets" "k8s.io/apimachinery/pkg/util/strategicpatch" corelisters "k8s.io/client-go/listers/core/v1" "k8s.io/klog/v2" podutil "k8s.io/kubernetes/pkg/api/v1/pod" "k8s.io/utils/pointer" ) const ( // LastAppliedConfigAnnotation is annotation key of last applied configuration LastAppliedConfigAnnotation = "pingcap.com/last-applied-configuration" // ImagePullBackOff is the pod state of image pull failed ImagePullBackOff = "ImagePullBackOff" // ErrImagePull is the pod state of image pull failed ErrImagePull = "ErrImagePull" // MetricsUpdateInterval represents the interval time of detect store' status. // This must be long enough to cover at least 2 scraping intervals. // Or you will get nothing when querying CPU usage. MetricsUpdateInterval = time.Minute // CAScaleDownDisabledAnnotation is annotation key of ca scale down disabled CAScaleDownDisabledAnnotation = "cluster-autoscaler.kubernetes.io/scale-down-disabled" // MemberReadyTime is the key of member ready time MemberReadyTime = "memberReadyTime" ) var ( // The first version that moves the rocksdb info and raft info log to store and rotate as the TiKV log is v5.0.0 // https://github.com/tikv/tikv/pull/7358 tikvLessThanV500, _ = semver.NewConstraint(" 0 { return fmt.Sprintf("%s.%s-pd-peer.%s.svc.%s", PdPodName(tcName, ordinal), tcName, namespace, clusterDomain) } // clusterDomain is not set if acrossK8s { return fmt.Sprintf("%s.%s-pd-peer.%s.svc", PdPodName(tcName, ordinal), tcName, namespace) } return PdPodName(tcName, ordinal) } // NeedForceUpgrade check if force upgrade is necessary func NeedForceUpgrade(ann map[string]string) bool { // Check if annotation 'pingcap.com/force-upgrade: "true"' is set if ann != nil { forceVal, ok := ann[label.AnnForceUpgradeKey] if ok && (forceVal == label.AnnForceUpgradeVal) { return true } } return false } // MarshalTOML is a template function that try to marshal a go value to toml func MarshalTOML(v interface{}) ([]byte, error) { return toml.Marshal(v) } func UnmarshalTOML(b []byte, obj interface{}) error { return toml.Unmarshal(b, obj) } // getStsAnnotations gets annotations for statefulset of given component. func getStsAnnotations(tcAnns map[string]string, component string) map[string]string { anns := map[string]string{} if tcAnns == nil { return anns } // ensure the delete-slots annotation var key string switch component { case label.PDLabelVal: key = label.AnnPDDeleteSlots case label.TiDBLabelVal: key = label.AnnTiDBDeleteSlots case label.TiKVLabelVal: key = label.AnnTiKVDeleteSlots case label.TiFlashLabelVal: key = label.AnnTiFlashDeleteSlots case label.DMMasterLabelVal: key = label.AnnDMMasterDeleteSlots case label.DMWorkerLabelVal: key = label.AnnDMWorkerDeleteSlots default: return anns } if val, ok := tcAnns[key]; ok { anns[helper.DeleteSlotsAnn] = val } return anns } // MapContainers index containers of Pod by container name in favor of looking up func MapContainers(podSpec *corev1.PodSpec) map[string]corev1.Container { m := map[string]corev1.Container{} for _, c := range podSpec.Containers { m[c.Name] = c } return m } // MapInitContainers index init containers of Pod by container name in favor of looking up func MapInitContainers(podSpec *corev1.PodSpec) map[string]corev1.Container { m := map[string]corev1.Container{} for _, c := range podSpec.InitContainers { m[c.Name] = c } return m } // findContainerByName finds targetContainer by containerName, If not find, then return nil func findContainerByName(sts *apps.StatefulSet, containerName string) *corev1.Container { for _, c := range sts.Spec.Template.Spec.Containers { if c.Name == containerName { return &c } } return nil } func getTikVConfigMapForTiKVSpec(tikvSpec *v1alpha1.TiKVSpec, tc *v1alpha1.TidbCluster) (*corev1.ConfigMap, error) { config := tikvSpec.Config.DeepCopy() if tc.IsTLSClusterEnabled() { config.Set("security.ca-path", path.Join(tikvClusterCertPath, tlsSecretRootCAKey)) config.Set("security.cert-path", path.Join(tikvClusterCertPath, corev1.TLSCertKey)) config.Set("security.key-path", path.Join(tikvClusterCertPath, corev1.TLSPrivateKeyKey)) } config.Set("black-list-path", "/etc/tikv/blacklist.json") confText, err := config.MarshalTOML() if err != nil { return nil, err } startScript, err := startscript.RenderTiKVStartScript(tc) if err != nil { return nil, fmt.Errorf("render start-script for tc %s/%s failed: %v", tc.Namespace, tc.Name, err) } cm := &corev1.ConfigMap{ Data: map[string]string{ "config-file": transformTiKVConfigMap(string(confText), tc), "startup-script": startScript, "blacklist-file": getBlacklistText(tc), }, } return cm, nil } // shouldRecover checks whether we should perform recovery operation. func shouldRecover(tc *v1alpha1.TidbCluster, component string, podLister corelisters.PodLister) bool { var stores map[string]v1alpha1.TiKVStore var failureStores map[string]v1alpha1.TiKVFailureStore var ordinals sets.Int32 var podPrefix string switch component { case label.TiKVLabelVal: stores = tc.Status.TiKV.Stores failureStores = tc.Status.TiKV.FailureStores ordinals = tc.TiKVStsDesiredOrdinals(true) podPrefix = controller.TiKVMemberName(tc.Name) case label.TiFlashLabelVal: stores = tc.Status.TiFlash.Stores failureStores = tc.Status.TiFlash.FailureStores ordinals = tc.TiFlashStsDesiredOrdinals(true) podPrefix = controller.TiFlashMemberName(tc.Name) default: klog.Warningf("Unexpected component %s for %s/%s in shouldRecover", component, tc.Namespace, tc.Name) return false } if failureStores == nil { return false } // If all desired replicas (excluding failover pods) of tidb cluster are // healthy, we can perform our failover recovery operation. // Note that failover pods may fail (e.g. lack of resources) and we don't care // about them because we're going to delete them. for ordinal := range ordinals { name := fmt.Sprintf("%s-%d", podPrefix, ordinal) pod, err := podLister.Pods(tc.Namespace).Get(name) if err != nil { klog.Errorf("pod %s/%s does not exist: %v", tc.Namespace, name, err) return false } if !podutil.IsPodReady(pod) { return false } var exist bool for _, v := range stores { if v.PodName == pod.Name { exist = true if v.State != v1alpha1.TiKVStateUp { return false } } } if !exist { return false } } return true } // shouldRecover checks whether we should perform recovery operation. func shouldRecoverDM(dc *v1alpha1.DMCluster, component string, podLister corelisters.PodLister) bool { var members map[string]v1alpha1.WorkerMember var failureMembers map[string]v1alpha1.WorkerFailureMember var ordinals sets.Int32 var podPrefix string switch component { case label.DMWorkerLabelVal: members = dc.Status.Worker.Members failureMembers = dc.Status.Worker.FailureMembers ordinals = dc.WorkerStsDesiredOrdinals(true) podPrefix = controller.DMWorkerMemberName(dc.Name) default: klog.Warningf("Unexpected component %s for %s/%s in shouldRecover", component, dc.Namespace, dc.Name) return false } if failureMembers == nil { return false } // If all desired replicas (excluding failover pods) of dm cluster are // healthy, we can perform our failover recovery operation. // Note that failover pods may fail (e.g. lack of resources) and we don't care // about them because we're going to delete them. for ordinal := range ordinals { name := fmt.Sprintf("%s-%d", podPrefix, ordinal) pod, err := podLister.Pods(dc.Namespace).Get(name) if err != nil { klog.Errorf("pod %s/%s does not exist: %v", dc.Namespace, name, err) return false } if !podutil.IsPodReady(pod) { return false } var exist bool for _, v := range members { if v.Name == pod.Name { exist = true if v.Stage == v1alpha1.DMWorkerStateOffline { return false } } } if !exist { return false } } return true } func CreateOrUpdateService(serviceLister corelisters.ServiceLister, serviceControl controller.ServiceControlInterface, newSvc *corev1.Service, obj runtime.Object) error { oldSvcTmp, err := serviceLister.Services(newSvc.Namespace).Get(newSvc.Name) if errors.IsNotFound(err) { err = controller.SetServiceLastAppliedConfigAnnotation(newSvc) if err != nil { return err } return serviceControl.CreateService(obj, newSvc) } if err != nil { return fmt.Errorf("createOrUpdateService: fail to get svc %s for obj %v, error: %s", newSvc.Name, obj, err) } oldSvc := oldSvcTmp.DeepCopy() util.RetainManagedFields(newSvc, oldSvc) equal, err := controller.ServiceEqual(newSvc, oldSvc) if err != nil { return err } annoEqual := util.IsSubMapOf(newSvc.Annotations, oldSvc.Annotations) isOrphan := metav1.GetControllerOf(oldSvc) == nil if !equal || !annoEqual || isOrphan { svc := *oldSvc svc.Spec = newSvc.Spec err = controller.SetServiceLastAppliedConfigAnnotation(&svc) if err != nil { return err } svc.Spec.ClusterIP = oldSvc.Spec.ClusterIP // apply change of annotations if any for k, v := range newSvc.Annotations { svc.Annotations[k] = v } // also override labels when adopt orphan if isOrphan { svc.OwnerReferences = newSvc.OwnerReferences svc.Labels = newSvc.Labels } _, err = serviceControl.UpdateService(obj, &svc) return err } return nil } // addDeferDeletingAnnoToPVC set the label func addDeferDeletingAnnoToPVC(tc *v1alpha1.TidbCluster, pvc *corev1.PersistentVolumeClaim, pvcControl controller.PVCControlInterface, scaleInTime ...string) error { if pvc.Annotations == nil { pvc.Annotations = map[string]string{} } now := time.Now().Format(time.RFC3339) pvc.Annotations[label.AnnPVCDeferDeleting] = now // scaleInTime indicates the time when call scale in, for test only since pvc defer deleting time may be different in same scale in call. if len(scaleInTime) > 0 { pvc.Annotations[label.AnnPVCScaleInTime] = scaleInTime[0] } if _, err := pvcControl.UpdatePVC(tc, pvc); err != nil { klog.Errorf("failed to set PVC %s/%s annotation %q to %q", tc.Namespace, pvc.Name, label.AnnPVCDeferDeleting, now) return err } klog.Infof("set PVC %s/%s annotation %q to %q successfully", tc.Namespace, pvc.Name, label.AnnPVCDeferDeleting, now) return nil } // GetPVCSelectorForPod compose a PVC selector from a tc/dm-cluster member pod at ordinal position func GetPVCSelectorForPod(controller runtime.Object, memberType v1alpha1.MemberType, ordinal int32) (labels.Selector, error) { meta := controller.(metav1.Object) var podName string var l label.Label switch controller.(type) { case *v1alpha1.TidbCluster: podName = ordinalPodName(memberType, meta.GetName(), ordinal) l = label.New().Instance(meta.GetName()) l[label.AnnPodNameKey] = podName case *v1alpha1.DMCluster: // podName = ordinalPodName(memberType, meta.GetName(), ordinal) l = label.NewDM().Instance(meta.GetName()) // just delete all defer Deleting pvc for convenience. Or dm have to support sync meta info labels for pod/pvc which seems unnecessary // l[label.AnnPodNameKey] = podName default: kind := controller.GetObjectKind().GroupVersionKind().Kind return nil, fmt.Errorf("object %s/%s of kind %s has unknown controller", meta.GetNamespace(), meta.GetName(), kind) } return l.Selector() } // TiKVLessThanV50 checks whether the `image` is less than v5.0.0 func TiKVLessThanV50(image string) bool { _, version := parseImage(image) v, err := semver.NewVersion(version) if err != nil { klog.Errorf("Parse version %s failure, error: %v", version, err) return false } if tikvLessThanV500.Check(v) { return true } return false } // parseImage returns the image name and the tag from the input image string func parseImage(image string) (string, string) { var name, tag string colonIdx := strings.LastIndexByte(image, ':') if colonIdx >= 0 { name = image[:colonIdx] tag = image[colonIdx+1:] } else { name = image } return name, tag } var ErrNotFoundStoreID = fmt.Errorf("not found") func TiKVStoreIDFromStatus(tc *v1alpha1.TidbCluster, podName string) (uint64, error) { for _, store := range tc.Status.TiKV.Stores { if store.PodName == podName { storeID, err := strconv.ParseUint(store.ID, 10, 64) if err != nil { return 0, err } return storeID, nil } } return 0, ErrNotFoundStoreID } // MergePatchContainers adds patches to base using a strategic merge patch and // iterating by container name, failing on the first error func MergePatchContainers(base, patches []corev1.Container) ([]corev1.Container, error) { var out []corev1.Container containersByName := make(map[string]corev1.Container) for _, c := range patches { containersByName[c.Name] = c } // Patch the containers that exist in base. for _, container := range base { patchContainer, ok := containersByName[container.Name] if !ok { // This container didn't need to be patched. out = append(out, container) continue } containerBytes, err := json.Marshal(container) if err != nil { return nil, fmt.Errorf("failed to marshal JSON for container %s, error: %v", container.Name, err) } patchBytes, err := json.Marshal(patchContainer) if err != nil { return nil, fmt.Errorf("failed to marshal JSON for patch container %s, error: %v", container.Name, err) } // Calculate the patch result. jsonResult, err := strategicpatch.StrategicMergePatch(containerBytes, patchBytes, corev1.Container{}) if err != nil { return nil, fmt.Errorf("failed to generate merge patch for container %s, error: %v", container.Name, err) } var patchResult corev1.Container if err := json.Unmarshal(jsonResult, &patchResult); err != nil { return nil, fmt.Errorf("failed to unmarshal merged container %s, error: %v", container.Name, err) } // Add the patch result and remove the corresponding key from the to do list. out = append(out, patchResult) delete(containersByName, container.Name) } // Append containers that are left in containersByName. // Iterate over patches to preserve the order. for _, c := range patches { if container, found := containersByName[c.Name]; found { out = append(out, container) } } return out, nil } func BuildProbeCommand(tc *v1alpha1.TidbCluster, componentType string) (command []string) { host := "127.0.0.1" var readinessURL string if componentType == label.PDLabelVal { readinessURL = fmt.Sprintf("%s://%s:2379/status", tc.Scheme(), host) } if componentType == label.TiDBLabelVal { readinessURL = fmt.Sprintf("%s://%s:10080/status", tc.Scheme(), host) } command = append(command, "curl") command = append(command, readinessURL) // Fail silently (no output at all) on server errors // without this if the server return 500, the exist code will be 0 // and probe is success. command = append(command, "--fail") // follow 301 or 302 redirect command = append(command, "--location") if tc.IsTLSClusterEnabled() { cacert := path.Join(clusterCertPath, tlsSecretRootCAKey) cert := path.Join(clusterCertPath, corev1.TLSCertKey) key := path.Join(clusterCertPath, corev1.TLSPrivateKeyKey) command = append(command, "--cacert", cacert) command = append(command, "--cert", cert) command = append(command, "--key", key) } return } // unmarshalSpec unmarshals the component's spec from bytes. func unmarshalSpec[S any](specBytes []byte, spec *S) error { return json.Unmarshal(specBytes, spec) } // specEquals compares the new component's spec with old component's spec func specEquals[S any](tc *v1alpha1.TidbCluster, newSpec *S, oldAppliedSpec string, component string) bool { oldSpec := new(S) if len(oldAppliedSpec) > 0 { err := unmarshalSpec([]byte(oldAppliedSpec), oldSpec) if err != nil { klog.Errorf("unmarshal [%s/%s]'s %s old applied spec failed, error: %v", tc.GetNamespace(), tc.GetName(), component, err) return false } return apiequality.Semantic.DeepEqual(oldSpec, newSpec) } return false } // syncAppliedConfigStatus syncs the component spec to status. func syncAppliedConfigStatus[S any](tc *v1alpha1.TidbCluster, newSpec *S, appliedConfig *v1alpha1.AppliedConfig, component string, saveLastNormal bool, deps *controller.Dependencies) (err error) { if appliedConfig == nil { return nil } if !specEquals(tc, newSpec, appliedConfig.Current, component) { if err := setAppliedConfigStatus(tc, newSpec, appliedConfig, component, saveLastNormal, deps); err != nil { return err } } return nil } // setAppliedConfigStatus set current applied spec and last normal applied spec info to component's status func setAppliedConfigStatus[S any](tc *v1alpha1.TidbCluster, spec *S, appliedConfig *v1alpha1.AppliedConfig, component string, saveLastNormal bool, deps *controller.Dependencies) error { b, err := json.Marshal(spec) if err != nil { klog.Errorf("marshal [%s/%s]'s %s spec failed, error: %v", tc.GetNamespace(), tc.GetName(), component, err) return nil } if currentAppliedSpec := appliedConfig.Current; len(currentAppliedSpec) > 0 && saveLastNormal { klog.Infof("set last normal applied config to [%s/%s]'s %s status", tc.GetNamespace(), tc.GetName(), component) appliedConfig.LastNormal = currentAppliedSpec } klog.Infof("set current applied config to [%s/%s]'s %s status", tc.GetNamespace(), tc.GetName(), component) appliedConfig.Current = string(b) if deps != nil { if _, err = deps.TiDBClusterControl.Update(tc.DeepCopy()); err != nil { klog.Errorf("set applied config: failed to update TidbCluster: [%s/%s], error: %v", tc.GetNamespace(), tc.GetName(), err) return nil } return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting for the %s applied spec to be saved", tc.GetNamespace(), tc.GetName(), component) } return nil } func rollbackConfig[S any](tc *v1alpha1.TidbCluster, spec *S, appliedConfig *v1alpha1.AppliedConfig, enableConfigRollback *bool, component string, set *apps.StatefulSet, deps *controller.Dependencies) error { if enableConfigRollback == nil || !*enableConfigRollback || appliedConfig == nil { return nil } if lastNormalAppliedSpec := appliedConfig.LastNormal; len(lastNormalAppliedSpec) > 0 { oldSpec := new(S) err := json.Unmarshal([]byte(lastNormalAppliedSpec), oldSpec) if err != nil { klog.Errorf("rollback: unmarshal Spec [%s/%s]'s %s last normal applied spec failed, error: %v", tc.GetNamespace(), tc.GetName(), component, err) return nil } if set != nil { klog.Infof("rollback: %s cluster %s/%s deletes failed statefulset", component, tc.GetNamespace(), tc.GetName()) err := deps.StatefulSetControl.DeleteStatefulSet(tc, set) if err != nil { klog.Errorf("rollback: failed to delete %s statefulset: [%s/%s], error: %v", tc.GetNamespace(), tc.GetName(), component, err) return nil } return controller.RequeueErrorf("tidbcluster: deletes %s statefulset for tc [%s/%s]", component, tc.GetNamespace(), tc.GetName()) } *spec = *oldSpec appliedConfig.Current = lastNormalAppliedSpec appliedConfig.LastNormal = "" klog.Infof("rollback: %s cluster %s/%s reverts spec to last normal applied configuration", component, tc.GetNamespace(), tc.GetName()) if _, err = deps.TiDBClusterControl.Update(tc.DeepCopy()); err != nil { klog.Errorf("rollback: failed to update TidbCluster: [%s/%s], error: %v", tc.GetNamespace(), tc.GetName(), err) return nil } return controller.RequeueErrorf("tidbcluster: revert %s spec to last normal applied configuration for tc [%s/%s]", component, tc.GetNamespace(), tc.GetName()) } *enableConfigRollback = false klog.Infof("rollback: %s cluster %s/%s is unable to roll back and resets enableConfigRollback to false", component, tc.GetNamespace(), tc.GetName()) if _, err := deps.TiDBClusterControl.Update(tc.DeepCopy()); err != nil { klog.Errorf("rollback: failed to update TidbCluster: [%s/%s], error: %v", tc.GetNamespace(), tc.GetName(), err) return nil } return controller.RequeueErrorf("tidbcluster: reset %s enableConfigRollback to false for tc [%s/%s]", component, tc.GetNamespace(), tc.GetName()) } type BlacklistConfig struct { KeyspaceIDs []uint32 `json:"keyspace_ids,omitempty"` RegionIDs []uint64 `json:"region_ids,omitempty"` } // syncBlacklistStatus syncs the blacklist spec and config to status. func syncBlacklistStatus(tc *v1alpha1.TidbCluster, deps *controller.Dependencies) (err error) { if !tc.PDIsAvailable() { return nil } if tc.Status.Blacklist == nil { tc.Status.Blacklist = &v1alpha1.BlacklistStatus{} } if !specEquals(tc, tc.Spec.Blacklist, tc.Status.Blacklist.Spec, "blacklist") { if err := setBlacklistStatus(tc, deps); err != nil { return err } } return nil } // setBlacklistStatus set the blacklist spec and config info to blacklist's status func setBlacklistStatus(tc *v1alpha1.TidbCluster, deps *controller.Dependencies) error { config, err := getBlacklistConfig(tc, deps) if err != nil { return err } configBytes, err := json.Marshal(config) if err != nil { klog.Errorf("marshal [%s/%s]'s blacklist config failed, error: %v", tc.GetNamespace(), tc.GetName(), err) return nil } tc.Status.Blacklist.Config = string(configBytes) specBytes, err := json.Marshal(tc.Spec.Blacklist) if err != nil { klog.Errorf("marshal [%s/%s]'s blacklist spec failed, error: %v", tc.GetNamespace(), tc.GetName(), err) return nil } tc.Status.Blacklist.Spec = string(specBytes) if deps != nil { if _, err = deps.TiDBClusterControl.Update(tc.DeepCopy()); err != nil { klog.Errorf("sync blacklist: failed to update TidbCluster: [%s/%s], error: %v", tc.GetNamespace(), tc.GetName(), err) return nil } return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting for the blacklist spec and config to be saved", tc.GetNamespace(), tc.GetName()) } return nil } func getBlacklistConfig(tc *v1alpha1.TidbCluster, deps *controller.Dependencies) (*BlacklistConfig, error) { config := &BlacklistConfig{} if tc.Spec.Blacklist != nil { blacklist := tc.Spec.Blacklist.DeepCopy() // use copy to not update tc spec config.RegionIDs = blacklist.RegionIDs if tc.PDIsAvailable() && len(blacklist.KeyspaceNames) > 0 { pdCli := controller.GetPDClient(deps.PDControl, tc) for _, name := range blacklist.KeyspaceNames { keyspaceInfo, err := pdCli.GetKeyspace(name) if err != nil { return nil, err } if keyspaceInfo.Name == name { config.KeyspaceIDs = append(config.KeyspaceIDs, keyspaceInfo.Id) } } } } return config, nil } func getBlacklistText(tc *v1alpha1.TidbCluster) string { blacklistText := "{}" if tc.Status.Blacklist != nil { if blacklistConfig := tc.Status.Blacklist.Config; len(blacklistConfig) > 0 { blacklistText = blacklistConfig } } return blacklistText } func getCanaryReplicasList(tc *v1alpha1.TidbCluster, component string, numPods int32, canaryUpgrade *v1alpha1.CanaryUpgrade) []int32 { canaryReplicasList := make([]int32, 0, len(canaryUpgrade.StagePercents)) if canaryUpgrade != nil && *canaryUpgrade.Enabled && len(canaryUpgrade.StagePercents) > 0 { for stageIndex := 0; stageIndex < len(canaryUpgrade.StagePercents); stageIndex++ { replicas := canaryUpgrade.StagePercents[stageIndex] * numPods / 100 if replicas < 1 { replicas = 1 } if len(canaryReplicasList) == 0 || canaryReplicasList[len(canaryReplicasList)-1] < replicas { canaryReplicasList = append(canaryReplicasList, replicas) } } } klog.Infof("tidbcluster: [%s/%s] %s canary upgrade replicas list %v", tc.GetNamespace(), tc.GetName(), component, canaryReplicasList) return canaryReplicasList } func checkCanaryUpgrade(tc *v1alpha1.TidbCluster, component string, pod *corev1.Pod, canaryUpgrade *v1alpha1.CanaryUpgrade, numPods, idx int32, canaryReplicasList []int32) error { if len(canaryReplicasList) > 0 && idx > 0 && isLastPodInStage(numPods, idx, canaryReplicasList) { lastReplicas := canaryReplicasList[len(canaryReplicasList)-1] if numPods-lastReplicas == idx { lastPercent := canaryUpgrade.StagePercents[len(canaryUpgrade.StagePercents)-1] return controller.IgnoreErrorf("tidbcluster: [%s/%s] finishes %s canary upgrade replicas %d percent %d%%", tc.GetNamespace(), tc.GetName(), component, lastReplicas, lastPercent) } else { if err := checkPodAvailable(tc, component, pod, *canaryUpgrade.StageReadySeconds); err != nil { return err } } } return nil } func isLastPodInStage(numPods, idx int32, canaryReplicasList []int32) bool { for _, replicas := range canaryReplicasList { if numPods-replicas == idx { return true } } return false } func checkPodAvailable(tc *v1alpha1.TidbCluster, component string, pod *corev1.Pod, minReadySeconds int32) error { readyCond := podutil.GetPodReadyCondition(pod.Status) if readyCond == nil || readyCond.Status != corev1.ConditionTrue { return controller.RequeueErrorf("tidbcluster: [%s/%s]'s upgraded %s pod: [%s] is not ready", tc.GetNamespace(), tc.GetName(), component, pod.GetName()) } minReadySecondsDuration := time.Duration(minReadySeconds) * time.Second if minReadySeconds == 0 || !readyCond.LastTransitionTime.IsZero() && readyCond.LastTransitionTime.Add(minReadySecondsDuration).Before(metav1.Now().Time) { return nil } return controller.RequeueErrorf("tidbcluster: [%s/%s]'s upgraded %s pod: [%s] is not available, last transition time is %v", tc.GetNamespace(), tc.GetName(), component, pod.GetName(), readyCond.LastTransitionTime) } type ResourceUsageClient interface { GetResourceUsage() (*v1alpha1.ResourceUsage, error) } type ResourceUsageClientWithPodCPUMetrics struct { namespace string podName string client ResourceUsageClient deps *controller.Dependencies } func (r *ResourceUsageClientWithPodCPUMetrics) GetResourceUsage() (resourceUsage *v1alpha1.ResourceUsage, err error) { resourceUsage, err = r.client.GetResourceUsage() if err != nil { return nil, err } cpuMetrics, err := getPodCPUMetrics(r.namespace, r.podName, r.deps) if err != nil { return nil, err } resourceUsage.CpuMilliSeconds = 0 resourceUsage.CpuMilliCores = cpuMetrics.MilliCores resourceUsage.CpuUsedMilliCores = cpuMetrics.UsedMilliCores resourceUsage.UpdateTime = metav1.NewTime(cpuMetrics.Timestamp) return resourceUsage, nil } func withPodCPUMetrics(namespace string, podName string, client ResourceUsageClient, deps *controller.Dependencies) ResourceUsageClient { return &ResourceUsageClientWithPodCPUMetrics{ namespace: namespace, podName: podName, client: client, deps: deps, } } func syncStoreResourceUsage(newStore *v1alpha1.TiKVStore, oldStore v1alpha1.TiKVStore, exist bool, resourceUsageClient ResourceUsageClient, podCPUMetrics bool, namespace string, deps *controller.Dependencies) { if exist { newStore.ResourceUsages = *oldStore.ResourceUsages.DeepCopy() } newStore.ResourceUsages.PodCPU = podCPUMetrics if len(newStore.ID) == 0 || newStore.State != v1alpha1.TiKVStateUp { return } if newStore.ResourceUsages.PodCPU { resourceUsageClient = withPodCPUMetrics(namespace, newStore.PodName, resourceUsageClient, deps) newResourceUsage, err := resourceUsageClient.GetResourceUsage() if err != nil { klog.Warningf("failed to get resource usage with pod CPU metrics for %s, error: %v", newStore.PodName, err) } else { newStore.ResourceUsages.Current = newResourceUsage } newStore.ResourceUsages.Last = nil return } old := newStore.ResourceUsages.Current if old == nil || old.UpdateTime.Add(MetricsUpdateInterval).Before(time.Now()) { newResourceUsage, err := resourceUsageClient.GetResourceUsage() if err != nil { klog.Warningf("failed to get resource usage for %s, error: %v", newStore.PodName, err) } else { newStore.ResourceUsages.Current = newResourceUsage newStore.ResourceUsages.Last = nil if old != nil && newResourceUsage.CpuMilliSeconds >= old.CpuMilliSeconds && newResourceUsage.CpuMilliCores == old.CpuMilliCores && newResourceUsage.StorageCapacity == old.StorageCapacity { newStore.ResourceUsages.Last = old } } } } type storeResource struct { cpuUsed float64 cpu float64 storageUsed float64 storageCapacity float64 } func (s *storeResource) add(a storeResource) { s.cpuUsed += a.cpuUsed s.cpu += a.cpu s.storageUsed += a.storageUsed s.storageCapacity += a.storageCapacity } func (s *storeResource) addUsed(a storeResource) { s.cpuUsed += a.cpuUsed s.storageUsed += a.storageUsed } func (s *storeResource) subtractCapacity(a storeResource) { s.cpu -= a.cpu s.storageCapacity -= a.storageCapacity } func (s *storeResource) cpuUsage() float64 { if s.cpu > 0 { return s.cpuUsed / s.cpu } return 0 } func (s *storeResource) storageUsage() float64 { if s.storageCapacity > 0 { return s.storageUsed / s.storageCapacity } return 0 } func getStoreResource(store v1alpha1.TiKVStore) *storeResource { if len(store.ID) > 0 && (store.State == v1alpha1.TiKVStateDown || store.State == v1alpha1.TiKVStateDisconnected) { return nil } current := store.ResourceUsages.Current if current == nil { return nil } var s = &storeResource{} s.storageUsed = float64(current.StorageCapacity - current.StorageAvailable) s.storageCapacity = float64(current.StorageCapacity) if store.ResourceUsages.PodCPU { s.cpuUsed = float64(current.CpuUsedMilliCores) s.cpu = float64(current.CpuMilliCores) } else { last := store.ResourceUsages.Last if last != nil { s.cpuUsed = float64(current.CpuMilliSeconds-last.CpuMilliSeconds) / 1000 s.cpu = float64(current.CpuMilliCores) / 1000 * current.UpdateTime.Sub(last.UpdateTime.Time).Seconds() } } return s } func syncResourceUsageStatus(stores map[string]v1alpha1.TiKVStore, maxPodName string, desiredReadyReplicas int32, resourceUsageStatus *v1alpha1.ResourceUsageStatus) { if resourceUsageStatus == nil { return } if desiredReadyReplicas <= 0 { *resourceUsageStatus = v1alpha1.ResourceUsageStatus{} return } var total storeResource var cpuReadyReplicas int32 var storageReadyReplicas int32 var lastStore storeResource var maxStoreCPU float64 for _, store := range stores { s := getStoreResource(store) if s == nil { continue } storageReadyReplicas++ total.add(*s) if store.PodName == maxPodName { lastStore = *s } storeCPUUsage := s.cpuUsage() if s.cpu > 0 { cpuReadyReplicas++ if storeCPUUsage > maxStoreCPU { maxStoreCPU = storeCPUUsage } } } resourceUsageStatus.MaxStoreCPU = formatUsage(maxStoreCPU) resourceUsageStatus.CPUReadyReplicas = cpuReadyReplicas resourceUsageStatus.CPUReady = cpuReadyReplicas >= desiredReadyReplicas resourceUsageStatus.CPU = formatUsage(total.cpuUsage()) resourceUsageStatus.StorageReadyReplicas = storageReadyReplicas resourceUsageStatus.StorageReady = storageReadyReplicas >= desiredReadyReplicas resourceUsageStatus.Storage = formatUsage(total.storageUsage()) var predicted = total predicted.subtractCapacity(lastStore) resourceUsageStatus.PredictedScaleInCPU = formatUsage(predicted.cpuUsage()) resourceUsageStatus.PredictedScaleInStorage = formatUsage(predicted.storageUsage()) } func formatUsage(usage float64) string { if usage == 0 { return "" } return fmt.Sprintf("%.5f", usage) } func parseUsage(usage string) float64 { if len(usage) == 0 { return 0 } value, err := strconv.ParseFloat(usage, 64) if err != nil { klog.Errorf("failed to parse float %v, err: %v", usage, err) return 0 } return value } func parseResourceUsageStatus(resourceUsageStatus v1alpha1.ResourceUsageStatus) (cpuUsage, storageUsage, predictedScaleInCPU, predictedScaleInStorage, maxStoreCPU float64, cpuReady, storageReady bool) { return parseUsage(resourceUsageStatus.CPU), parseUsage(resourceUsageStatus.Storage), parseUsage(resourceUsageStatus.PredictedScaleInCPU), parseUsage(resourceUsageStatus.PredictedScaleInStorage), parseUsage(resourceUsageStatus.MaxStoreCPU), resourceUsageStatus.CPUReady, resourceUsageStatus.StorageReady } func isAutoScalingEnabled(autoScaling *v1alpha1.AutoScaling) bool { return autoScaling != nil && autoScaling.Enabled != nil && *autoScaling.Enabled } func isAutoScalingPaused(autoScaling *v1alpha1.AutoScaling) bool { return autoScaling != nil && autoScaling.Paused != nil && *autoScaling.Paused } func isAllowScaleInCancel(autoScaling *v1alpha1.AutoScaling) bool { if autoScaling == nil || autoScaling.AllowScaleInCancel == nil { return true } return *autoScaling.AllowScaleInCancel } func isAutoScalingPodCPUMetrics(autoScaling *v1alpha1.AutoScaling) bool { return autoScaling != nil && autoScaling.PodCPUMetrics != nil && *autoScaling.PodCPUMetrics } func isAutoScalingNotAvailable(autoScaling *v1alpha1.AutoScaling, cpuUsageReady, storageUsageReady bool) (notAvailable bool) { var storageThreshold = autoScaling.Storage var cpuThreshold = autoScaling.CPU if storageThreshold == nil && cpuThreshold == nil { // Neither storage threshold nor cpu threshold is configured notAvailable = true return } // At least one of auto-scaling thresholds is configured. // The configured resource usage must be ready. if storageThreshold != nil && !storageUsageReady { notAvailable = true return } if cpuThreshold != nil && !cpuUsageReady { notAvailable = true return } return } func setAutoScalingStatus(autoScalingStatus *v1alpha1.AutoScalingStatus, targetReplicas int32) { autoScalingStatus.Replicas = targetReplicas resetAutoScalingStatusTime(autoScalingStatus) } func resetAutoScalingStatusTime(autoScalingStatus *v1alpha1.AutoScalingStatus) { autoScalingStatus.ScaleInStartTime = metav1.Time{} autoScalingStatus.ScaleOutStartTime = metav1.Time{} autoScalingStatus.LastIdleTime = metav1.Time{} } func isEmptyHotCluster(hotCluster *v1alpha1.HotCluster) bool { if hotCluster == nil || hotCluster.Cluster == nil || len(hotCluster.Cluster.Name) == 0 { return true } return false } func isHotCluster(tc *v1alpha1.TidbCluster, hotCluster *v1alpha1.HotCluster) bool { if isEmptyHotCluster(hotCluster) { return false } return hotCluster.Cluster.Name == tc.GetName() && hotCluster.Cluster.Namespace == tc.GetNamespace() } func hasHotCluster(tc *v1alpha1.TidbCluster, hotCluster *v1alpha1.HotCluster) bool { if isEmptyHotCluster(hotCluster) { return false } return hotCluster.Cluster.Name != tc.GetName() || hotCluster.Cluster.Namespace != tc.GetNamespace() } func syncAutoScalingStatus(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType, isScaling bool, hotCluster *v1alpha1.HotCluster, autoScaling *v1alpha1.AutoScaling, resourceUsageStatus v1alpha1.ResourceUsageStatus, autoScalingStatus *v1alpha1.AutoScalingStatus, pdCli pdapi.PDClient, coord coordinator.CoordinatorInterface, deps *controller.Dependencies) error { if autoScalingStatus == nil { return nil } var targetReplicas int32 if !isAutoScalingEnabled(autoScaling) { if autoScalingStatus.Replicas == 0 { resetAutoScalingStatusTime(autoScalingStatus) return nil } klog.Infof("autoscaling: [%s/%s] %s cluster updates auto scaling replicas from %v to %v when auto scaling is disabled", tc.GetNamespace(), tc.GetName(), memberType, autoScalingStatus.Replicas, targetReplicas) setAutoScalingStatus(autoScalingStatus, targetReplicas) return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to sync %s auto scaling status", tc.GetNamespace(), tc.GetName(), memberType) } if isAutoScalingPaused(autoScaling) { return cancelScalingIn(tc, memberType, deps) } var cpuUsage, storageUsage, predictedScaleInCPU, predictedScaleInStorage, maxStoreCPU, cpuUsageReady, storageUsageReady = parseResourceUsageStatus(resourceUsageStatus) isScalingIn := coordinator.GetTidbClusterStatus(tc).GetMemberStatus(memberType).IsScalingIn() if isScaling && !isScalingIn || isAutoScalingNotAvailable(autoScaling, cpuUsageReady, storageUsageReady) { return nil } var storageReplicas, cpuReplicas int32 if autoScaling.Storage != nil { storageReplicas = scaleReplicas(autoScalingStatus.Replicas, storageUsage, predictedScaleInStorage, autoScaling.Storage) } if autoScaling.CPU != nil { cpuReplicas = scaleReplicas(autoScalingStatus.Replicas, cpuUsage, predictedScaleInCPU, autoScaling.CPU) if hasHotCluster(tc, hotCluster) && isHotClusterEnabled(hotCluster) && coord != nil && maxStoreCPU > *hotCluster.CPUThreshold { if autoScalingStatus.HotStartTime.IsZero() { autoScalingStatus.HotStartTime = metav1.Now() } if autoScalingStatus.HotStartTime.Add(time.Duration(*hotCluster.HotDurationSeconds) * time.Second).Before(time.Now()) { var ( err error try = true do = func() { err = activateHotCluster(hotCluster, memberType, pdCli, deps) } ) executed := coord.SynchronouslyDoByGroup(tc, memberType, try, do) if err != nil { klog.Errorf("activateHotCluster: [%s/%s] %s, error: %v", tc.GetNamespace(), tc.GetName(), memberType, err) } else if executed { autoScalingStatus.HotStartTime = metav1.Time{} } } } else { autoScalingStatus.HotStartTime = metav1.Time{} } } if cpuReplicas > storageReplicas { targetReplicas = cpuReplicas } else { targetReplicas = storageReplicas } if isScalingIn && isAllowScaleInCancel(autoScaling) { // Generally, scaling-in takes a long time, and new workloads may cause insufficient resources during scaling-in period. if targetReplicas > autoScalingStatus.Replicas { klog.Infof("autoscaling: [%s/%s] %s cluster cancel delete all stores", tc.GetNamespace(), tc.GetName(), memberType) if err := cancelDeleteAllStores(tc, memberType, deps); err != nil { return err } setAutoScalingStatus(autoScalingStatus, targetReplicas) return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to sync %s auto scaling status", tc.GetNamespace(), tc.GetName(), memberType) } return nil } if targetReplicas < 0 { targetReplicas = 0 } else if autoScaling.MaxReplicas != nil && targetReplicas > *autoScaling.MaxReplicas { targetReplicas = *autoScaling.MaxReplicas } if autoScalingStatus.Replicas < targetReplicas { autoScalingStatus.ScaleInStartTime = metav1.Time{} if autoScalingStatus.ScaleOutStartTime.IsZero() { autoScalingStatus.ScaleOutStartTime = metav1.Now() } var scaleOutIntervalSeconds int32 if cpuReplicas > storageReplicas { scaleOutIntervalSeconds = *autoScaling.CPU.ScaleOutIntervalSeconds } else if cpuReplicas < storageReplicas { scaleOutIntervalSeconds = *autoScaling.Storage.ScaleOutIntervalSeconds } else { if *autoScaling.CPU.ScaleOutIntervalSeconds < *autoScaling.Storage.ScaleOutIntervalSeconds { scaleOutIntervalSeconds = *autoScaling.CPU.ScaleOutIntervalSeconds } else { scaleOutIntervalSeconds = *autoScaling.Storage.ScaleOutIntervalSeconds } } if autoScalingStatus.ScaleOutStartTime.Add(time.Duration(scaleOutIntervalSeconds) * time.Second).Before(time.Now()) { klog.Infof("autoscaling: [%s/%s] %s cluster automatically scales out from %v to %v based on cpu usage %.5f and storage usage %.5f for %v seconds", tc.GetNamespace(), tc.GetName(), memberType, autoScalingStatus.Replicas, targetReplicas, cpuUsage, storageUsage, scaleOutIntervalSeconds) setAutoScalingStatus(autoScalingStatus, targetReplicas) return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to sync %s auto scaling status", tc.GetNamespace(), tc.GetName(), memberType) } } else if autoScalingStatus.Replicas > targetReplicas { autoScalingStatus.ScaleOutStartTime = metav1.Time{} if autoScalingStatus.ScaleInStartTime.IsZero() { autoScalingStatus.ScaleInStartTime = metav1.Now() } var scaleInIntervalSeconds int32 if autoScaling.CPU != nil && autoScaling.Storage != nil { if *autoScaling.CPU.ScaleInIntervalSeconds > *autoScaling.Storage.ScaleInIntervalSeconds { scaleInIntervalSeconds = *autoScaling.CPU.ScaleInIntervalSeconds } else { scaleInIntervalSeconds = *autoScaling.Storage.ScaleInIntervalSeconds } } else if autoScaling.CPU != nil { scaleInIntervalSeconds = *autoScaling.CPU.ScaleInIntervalSeconds } else { scaleInIntervalSeconds = *autoScaling.Storage.ScaleInIntervalSeconds } if autoScalingStatus.ScaleInStartTime.Add(time.Duration(scaleInIntervalSeconds) * time.Second).Before(time.Now()) { klog.Infof("autoscaling: [%s/%s] %s cluster automatically scales in from %v to %v based on cpu usage %.5f and storage usage %.5f for %v seconds", tc.GetNamespace(), tc.GetName(), memberType, autoScalingStatus.Replicas, targetReplicas, cpuUsage, storageUsage, scaleInIntervalSeconds) setAutoScalingStatus(autoScalingStatus, targetReplicas) return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to sync %s auto scaling status", tc.GetNamespace(), tc.GetName(), memberType) } } else { var replicas int32 var isUpgrading bool var hotClusterActive *bool switch memberType { case v1alpha1.TiKVMemberType: replicas = tc.Spec.TiKV.Replicas isUpgrading = tc.TiKVUpgrading() hotClusterActive = &tc.Status.TiKV.HotCluster.Active default: } if isHotCluster(tc, hotCluster) && replicas > 0 && autoScalingStatus.Replicas == 0 && cpuUsage < autoScaling.CPU.MinThreshold && *hotCluster.ScaleToZero && !isUpgrading { autoScalingStatus.ScaleInStartTime = metav1.Time{} autoScalingStatus.ScaleOutStartTime = metav1.Time{} if autoScalingStatus.LastIdleTime.IsZero() { autoScalingStatus.LastIdleTime = metav1.Now() } var scaleInIntervalSeconds = *autoScaling.CPU.ScaleInIntervalSeconds if autoScalingStatus.LastIdleTime.Add(time.Duration(scaleInIntervalSeconds) * time.Second).Before(time.Now()) { klog.Infof("autoscaling: [%s/%s] %s deactivate hot cluster based on cpu usage %.5f for %v seconds", tc.GetNamespace(), tc.GetName(), memberType, cpuUsage, scaleInIntervalSeconds) *hotClusterActive = false return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to deactivate %s hot cluster", tc.GetNamespace(), tc.GetName(), memberType) } } else { resetAutoScalingStatusTime(autoScalingStatus) } } return nil } func activateHotCluster(hotCluster *v1alpha1.HotCluster, memberType v1alpha1.MemberType, pdCli pdapi.PDClient, deps *controller.Dependencies) error { hotTC, err := deps.TiDBClusterLister.TidbClusters(hotCluster.Cluster.Namespace).Get(hotCluster.Cluster.Name) if err != nil { if errors.IsNotFound(err) { return controller.IgnoreErrorf("failed to get hot TidbCluster: [%s/%s] %s, error: %v", hotCluster.Cluster.Namespace, hotCluster.Cluster.Name, memberType, err) } return err } var hot *v1alpha1.HotCluster var hotClusterActive *bool var isScaling bool switch memberType { case v1alpha1.TiKVMemberType: hot = hotTC.Spec.TiKV.HotCluster hotClusterActive = &hotTC.Status.TiKV.HotCluster.Active isScaling = hotTC.TiKVScaling() default: return nil } if !isHotClusterEnabled(hot) { return controller.IgnoreErrorf("Hot TidbCluster: [%s/%s] %s is disabled", hotTC.GetNamespace(), hotTC.GetName(), memberType) } if !*hot.ScaleToZero || *hot.InitialReplicas == 0 { return controller.IgnoreErrorf("Hot TidbCluster: [%s/%s] %s does not need to be activated", hotTC.GetNamespace(), hotTC.GetName(), memberType) } if isScaling { klog.Warningf("Hot TidbCluster: [%s/%s] %s is scaling", hotTC.GetNamespace(), hotTC.GetName(), memberType) return nil } history, err := pdCli.GetHotRegions(*hot.RegionHeartbeatSeconds, *hot.StoreHeartbeatSeconds, *hot.HotRegionSeconds) if err != nil { return err } if !*hotClusterActive { *hotClusterActive = true klog.Infof("activateHotCluster: activate hot cluster [%s/%s]'s %s based on hot regions %v, hot cnt %d", hotTC.GetNamespace(), hotTC.GetName(), memberType.String(), len(history.HistoryHotRegion), history.HotRegionsCount) if _, err := deps.TiDBClusterControl.Update(hotTC.DeepCopy()); err != nil { klog.Errorf("failed to update TidbCluster: [%s/%s], error: %v", hotTC.GetNamespace(), hotTC.GetName(), err) } } return nil } func syncHotCluster(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType, hotCluster *v1alpha1.HotCluster, pdCli pdapi.PDClient, coord coordinator.CoordinatorInterface) error { if !isHotCluster(tc, hotCluster) { return nil } notEmpty, err := isNotEmptyTierCluster(tc, memberType, pdCli, placement.HotRegionRuleGroup) if err != nil { return err } if !notEmpty { return nil } var hot *v1alpha1.HotCluster upCount := 0 var isUpgrading bool switch memberType { case v1alpha1.TiKVMemberType: hot = tc.Spec.TiKV.HotCluster for _, store := range tc.Status.TiKV.Stores { if store.State == v1alpha1.TiKVStateUp { upCount++ } } isUpgrading = tc.TiKVUpgrading() default: return nil } if *hot.InitialReplicas == 0 { return nil } singleAZTemplateRule, err := getSingleAZTemplateRule(tc, pdCli, placement.HotRegionRuleGroup, placement.HotTierLabelConstraint) if err != nil { return fmt.Errorf("failed to find hot single az placement rule for TidbCluster [%s/%s]", tc.GetNamespace(), tc.GetName()) } historyHotRegions, err := pdCli.GetHotRegions(*hot.RegionHeartbeatSeconds, *hot.StoreHeartbeatSeconds, *hot.HotRegionSeconds) if err != nil { return err } klog.Infof("syncHotCluster: get hot regions %v, hot cnt %d, %s", len(historyHotRegions.HistoryHotRegion), historyHotRegions.HotRegionsCount, historyHotRegions.String()) storeTiers, err := coordinator.GetStoreTiers(pdCli) if err != nil { return err } if tc.Status.TiKV.HotCluster.CoolingRegions == nil { tc.Status.TiKV.HotCluster.CoolingRegions = make(map[string]metav1.Time) } var coolingRegions = tc.Status.TiKV.HotCluster.CoolingRegions var history = &pdapi.HistoryHotRegions{} for idx, region := range historyHotRegions.HistoryHotRegion { if idx >= historyHotRegions.HotRegionsCount { continue } delete(coolingRegions, fmt.Sprintf("%d", region.RegionID)) standardPeerCnt := coordinator.GetTierPeerCountWithRegion(region.Region, storeTiers, placement.StandardTier) existHotPeer := coordinator.CheckIfTierPeerExistsWithRegion(region.Region, storeTiers, placement.HotTier) if standardPeerCnt > 2 || (standardPeerCnt == 2 && existHotPeer) { history.HistoryHotRegion = append(history.HistoryHotRegion, region) } } if len(history.HistoryHotRegion) > int(*hot.MaxHotRegionsCount) { history.HistoryHotRegion = history.HistoryHotRegion[:int(*hot.MaxHotRegionsCount)] } var leaderRuleRole = placement.Leader if isUpgrading { leaderRuleRole = placement.Voter } hotRules, err := getCurrentRules(pdCli, placement.HotRegionRuleGroup, "") if err != nil { return err } hotRulesCount := len(hotRules) if hotRulesCount == 0 && upCount == 0 { return nil } storeZones, err := coordinator.GetStoreZones(pdCli) if err != nil { return err } var currentHotRegions = map[uint64]struct{}{} for _, rule := range hotRules { var regionID uint64 if placement.IsSingleAZRegionRuleID(rule.ID) { regionID, _, err = placement.ParseSingleAZRegionRuleID(rule.ID) } else if placement.IsMultiAZRegionRuleID(rule.ID) { regionID, _, err = placement.ParseMultiAZRegionRuleID(rule.ID) } else { continue } currentHotRegions[regionID] = struct{}{} } for regionKey := range coolingRegions { regionID, err := strconv.ParseUint(regionKey, 10, 64) if err != nil { continue } if _, ok := currentHotRegions[regionID]; !ok { delete(coolingRegions, regionKey) } } var primaryTemplateRule, standbyTemplateRule, witnessTemplateRule *placement.Rule var multiAZEnabled = tc.IsMultiAZEnabled() if multiAZEnabled { primaryTemplateRule, standbyTemplateRule, witnessTemplateRule, err = getMultiAZTemplateRule(pdCli, placement.HotRegionRuleGroup, placement.HotTierLabelConstraint) if err != nil { return err } if primaryTemplateRule == nil || standbyTemplateRule == nil || witnessTemplateRule == nil { return nil } } var updateRulesList = [][]*placement.Rule{} var updateCount int var multiAZCount int for _, region := range history.HistoryHotRegion { var newRules []*placement.Rule var zoneRedundant = false if multiAZEnabled { zoneRedundant, err = isZoneRedundantRegion(tc, coord, region) if err != nil { return err } ready, err := checkHotRegionReady(region, hotRules, storeZones, zoneRedundant) if err != nil { return err } if !ready { continue } } if zoneRedundant { ruleBatch := placement.GetMultiAZRegionRuleBatch(region.RegionID, region.StartKey, region.EndKey) newRules = getMultiAZRules(ruleBatch, hotRules, leaderRuleRole, true, primaryTemplateRule, standbyTemplateRule, witnessTemplateRule) if len(newRules) > 0 { multiAZCount++ } } else { ruleBatch := placement.GetSingleAZRegionRuleBatch(region.RegionID, region.StartKey, region.EndKey) newRules = getSingleAZRulesWithOnlyLeader(ruleBatch, hotRules, leaderRuleRole, singleAZTemplateRule) } if len(newRules) > 0 { updateRulesList = append(updateRulesList, newRules) updateCount += len(newRules) klog.Infof("syncHotCluster: add hot region %d to hot tier. hot type %s, hot degree %d, zone redundant %v", region.RegionID, region.HotRegionType, region.HotDegree, zoneRedundant) } } if len(updateRulesList) > 0 { var singleAZCount = len(updateRulesList) - multiAZCount klog.Infof("syncHotCluster: [%s/%s] %s hot cluster already has hot rules %d, hot regions %d, update regions %d (single-az %d, multi-az %d), update rules %d", tc.GetNamespace(), tc.GetName(), memberType, hotRulesCount, len(history.HistoryHotRegion), len(updateRulesList), singleAZCount, multiAZCount, updateCount) err = coordinator.UpdateRulesList(pdCli, updateRulesList, "") if err != nil { return err } } var now = metav1.Now() var coldDuration = time.Second * time.Duration(*tc.Spec.TiKV.HotCluster.ColdRegionSeconds) var notHotRegionRules = make([]*placement.Rule, 0, len(hotRules)) for _, rule := range hotRules { var regionID uint64 if placement.IsSingleAZRegionRuleID(rule.ID) { regionID, _, err = placement.ParseSingleAZRegionRuleID(rule.ID) } else if placement.IsMultiAZRegionRuleID(rule.ID) { regionID, _, err = placement.ParseMultiAZRegionRuleID(rule.ID) } else { continue } key := fmt.Sprintf("%d", regionID) lastTime, ok := coolingRegions[key] if !ok { coolingRegions[key] = now continue } if now.Time.Before(lastTime.Add(coldDuration)) { klog.Infof("The region %d is currently in a transitional state between hot and cold. last cold time %v, rule %s", regionID, lastTime, rule.ID) continue } klog.Infof("Remove hot region %d from hot tier. rule %s", regionID, rule.ID) notHotRegionRules = append(notHotRegionRules, rule) } checkReady := func(rule *placement.Rule) (bool, error) { return checkHotRuleDeletionReady(tc, rule, coord) } return coordinator.DisableHotTierRules(pdCli, notHotRegionRules, checkReady) } func checkHotRuleDeletionReady(tc *v1alpha1.TidbCluster, rule *placement.Rule, coordinator coordinator.CoordinatorInterface) (ready bool, err error) { if tc.IsMultiAZEnabled() { // Check if the region is switching zone redundancy. keyspaceID, err := placement.ParseKeyspaceID(rule.StartKey) if err == nil { zoneRedundant, err := coordinator.IsZoneRedundant(tc, keyspaceID) if err != nil { return false, err } if strings.HasPrefix(rule.ID, "single-az-") && zoneRedundant { return false, nil } if strings.HasPrefix(rule.ID, "multi-az-") && !zoneRedundant { return false, nil } } } return true, nil } func checkHotRegionReady(region *pdapi.HistoryHotRegion, hotRules map[string]*placement.Rule, storeZones map[uint64]string, zoneRedundant bool) (ready bool, err error) { // Check if the hot region is switching zone redundancy. if zoneRedundant { leaderRuleID := placement.FormatSingleAZRegionRuleID(region.RegionID, "leader") followerRuleID := placement.FormatSingleAZRegionRuleID(region.RegionID, "follower") _, leaderOk := hotRules[leaderRuleID] _, followerOk := hotRules[followerRuleID] if leaderOk || followerOk { delete(hotRules, leaderRuleID) delete(hotRules, followerRuleID) return false, nil } } else { primaryRuleID := placement.FormatMultiAZRegionRuleID(region.RegionID, placement.PrimaryZone) standbyRuleID := placement.FormatMultiAZRegionRuleID(region.RegionID, placement.StandbyZone) witnessRuleID := placement.FormatMultiAZRegionRuleID(region.RegionID, placement.WitnessZone) _, primaryOk := hotRules[primaryRuleID] _, standbyOk := hotRules[standbyRuleID] _, witnessOk := hotRules[witnessRuleID] if primaryOk || standbyOk || witnessOk { delete(hotRules, primaryRuleID) delete(hotRules, standbyRuleID) delete(hotRules, witnessRuleID) return false, nil } } // Check if the region peers are ready. var zones = map[string]string{} for _, peer := range region.Region.Peers { if strings.ToLower(peer.RoleName) != string(placement.Voter) { continue } zone, ok := storeZones[peer.StoreId] if !ok { return false, fmt.Errorf("store zone for store %d not exist", peer.StoreId) } zones[zone] = zone } if zoneRedundant { return len(zones) > 2, nil } return len(zones) == 1, nil } func getCurrentRules(pdCli pdapi.PDClient, groupID, trimPrefix string) (rules map[string]*placement.Rule, err error) { regionRules, err := pdCli.GetRulesByGroup(groupID) if err != nil { return nil, err } rules = map[string]*placement.Rule{} for _, rule := range regionRules { if len(trimPrefix) > 0 && strings.HasPrefix(rule.ID, trimPrefix) { rule.ID = strings.TrimPrefix(rule.ID, trimPrefix) } if strings.HasPrefix(rule.ID, "single-az-") || strings.HasPrefix(rule.ID, "multi-az-") { rules[rule.ID] = rule } } return rules, nil } func getSingleAZTemplateRule(tc *v1alpha1.TidbCluster, pdCli pdapi.PDClient, groupID string, labelConstraint placement.LabelConstraint) (templateRule *placement.Rule, err error) { var defaultRule *placement.Rule if tc.IsMultiAZEnabled() { defaultRule, err = pdCli.GetRuleByGroupAndID(placement.SingleAZKeyspaceRuleGroup, "default") if err != nil { return nil, fmt.Errorf("failed to find default single az pd placement rule for TidbCluster [%s/%s]", tc.GetNamespace(), tc.GetName()) } } else { defaultRule, err = pdCli.GetRuleByGroupAndID("pd", "default") if err != nil { return nil, fmt.Errorf("failed to find default pd placement rule for TidbCluster [%s/%s]", tc.GetNamespace(), tc.GetName()) } } updateTier := placement.UpdateRuleLabelConstraint(defaultRule, placement.StandardTierLabelConstraint) if updateTier { err := pdCli.SetRules([]*placement.Rule{defaultRule}) if err != nil { klog.Errorf("syncDefaultPD: failed to set tier label constraint for default pd placement rule, err %s", err) return nil, err } } var defaultRuleClone = defaultRule.Clone() templateRule = &placement.Rule{} templateRule.GroupID = groupID templateRule.LocationLabels = defaultRuleClone.LocationLabels templateRule.IsolationLevel = defaultRuleClone.IsolationLevel templateRule.LabelConstraints = defaultRuleClone.LabelConstraints placement.UpdateRuleLabelConstraint(templateRule, labelConstraint) return templateRule, nil } func getMultiAZTemplateRule(pdCli pdapi.PDClient, groupID string, labelConstraint placement.LabelConstraint) (primaryRule, standbyRule, witnessRule *placement.Rule, err error) { multiAZRules, err := pdCli.GetRulesByGroup(placement.DefaultMultiAZRuleGroup) if err != nil || len(multiAZRules) < 3 { return nil, nil, nil, err } for _, rule := range multiAZRules { rule.GroupID = groupID rule.StartKey = "" rule.EndKey = "" rule.Version = 0 placement.UpdateRuleLabelConstraint(rule, labelConstraint) if rule.ID == placement.PrimaryZone { primaryRule = rule.Clone() } else if rule.ID == placement.StandbyZone { standbyRule = rule.Clone() } else if rule.ID == placement.WitnessZone { witnessRule = rule.Clone() } } return primaryRule, standbyRule, witnessRule, nil } func getMultiAZRules( ruleBatch *placement.MultiAZRuleBatch, rules map[string]*placement.Rule, primaryRuleRole placement.PeerRoleType, onlyLeader bool, primaryTemplateRule, standbyTemplateRule, witnessTemplateRule *placement.Rule, ) []*placement.Rule { primaryRuleID := ruleBatch.PrimaryRuleID standbyRuleID := ruleBatch.StandbyRuleID witnessRuleID := ruleBatch.WitnessRuleID oldPrimaryRule, primaryOk := rules[primaryRuleID] oldStandbyRule, standbyOk := rules[standbyRuleID] oldWitnessRule, witnessOk := rules[witnessRuleID] if primaryOk { delete(rules, primaryRuleID) } if standbyOk { delete(rules, standbyRuleID) } if witnessOk { delete(rules, witnessRuleID) } if primaryOk && standbyOk && witnessOk && oldPrimaryRule.StartKey == ruleBatch.StartKey && oldPrimaryRule.EndKey == ruleBatch.EndKey && oldStandbyRule.StartKey == ruleBatch.StartKey && oldStandbyRule.EndKey == ruleBatch.EndKey && oldWitnessRule.StartKey == ruleBatch.StartKey && oldWitnessRule.EndKey == ruleBatch.EndKey && oldPrimaryRule.Count == 1 && oldStandbyRule.Count == 1 && oldWitnessRule.Count == 1 && oldPrimaryRule.Role == primaryRuleRole { return nil } primaryRule, standbyRule, witnessRule := primaryTemplateRule.Clone(), standbyTemplateRule.Clone(), witnessTemplateRule.Clone() primaryRule.ID, standbyRule.ID, witnessRule.ID = primaryRuleID, standbyRuleID, witnessRuleID primaryRule.StartKey, primaryRule.EndKey = ruleBatch.StartKey, ruleBatch.EndKey standbyRule.StartKey, standbyRule.EndKey = ruleBatch.StartKey, ruleBatch.EndKey witnessRule.StartKey, witnessRule.EndKey = ruleBatch.StartKey, ruleBatch.EndKey primaryRule.Role = primaryRuleRole if onlyLeader { placement.UpdateRuleLabelConstraint(standbyRule, placement.StandardTierLabelConstraint) placement.UpdateRuleLabelConstraint(witnessRule, placement.StandardTierLabelConstraint) } return []*placement.Rule{primaryRule, standbyRule, witnessRule} } func getSingleAZRules(ruleBatch *placement.SingleAZRuleBatch, rules map[string]*placement.Rule, singleAZTemplateRule *placement.Rule, checkReady func(rule *placement.Rule) bool) []*placement.Rule { ruleID := ruleBatch.VoterRuleID oldRule, ok := rules[ruleID] if ok { delete(rules, ruleID) } if ok && oldRule.StartKey == ruleBatch.StartKey && oldRule.EndKey == ruleBatch.EndKey && oldRule.Count == 3 && oldRule.Role == placement.Voter { if oldRule.IsolationLevel == "" { ready := checkReady(oldRule) klog.Infof("getSingleAZRules: check rule %s ready %v", oldRule.ID, ready) if ready { oldRule.IsolationLevel = singleAZTemplateRule.IsolationLevel return []*placement.Rule{oldRule} } } return nil } rule := singleAZTemplateRule.Clone() rule.ID = ruleID rule.StartKey, rule.EndKey = ruleBatch.StartKey, ruleBatch.EndKey rule.Count = 3 rule.Role = placement.Voter rule.IsolationLevel = "" return []*placement.Rule{rule} } func getSingleAZRulesWithOnlyLeader(ruleBatch *placement.SingleAZRuleBatch, rules map[string]*placement.Rule, leaderRuleRole placement.PeerRoleType, singleAZTemplateRule *placement.Rule) []*placement.Rule { leaderRuleID := ruleBatch.LeaderRuleID followerRuleID := ruleBatch.FollowerRuleID oldLeaderRule, leaderOk := rules[leaderRuleID] oldFollowerRule, followerOk := rules[followerRuleID] if leaderOk { delete(rules, leaderRuleID) } if followerOk { delete(rules, followerRuleID) } if leaderOk && followerOk && oldLeaderRule.StartKey == ruleBatch.StartKey && oldLeaderRule.EndKey == ruleBatch.EndKey && oldFollowerRule.StartKey == ruleBatch.StartKey && oldFollowerRule.EndKey == ruleBatch.EndKey && oldLeaderRule.Count == 1 && oldFollowerRule.Count == 2 && oldLeaderRule.Role == leaderRuleRole { return nil } leaderRule, followerRule := singleAZTemplateRule.Clone(), singleAZTemplateRule.Clone() leaderRule.ID, followerRule.ID = leaderRuleID, followerRuleID leaderRule.StartKey, leaderRule.EndKey = ruleBatch.StartKey, ruleBatch.EndKey followerRule.StartKey, followerRule.EndKey = ruleBatch.StartKey, ruleBatch.EndKey leaderRule.Count, followerRule.Count = 1, 2 leaderRule.Role = leaderRuleRole leaderRule.IsolationLevel = "" followerRule.Role = placement.Voter placement.UpdateRuleLabelConstraint(followerRule, placement.StandardTierLabelConstraint) return []*placement.Rule{leaderRule, followerRule} } func isZoneRedundantRegion(tc *v1alpha1.TidbCluster, coordinator coordinator.CoordinatorInterface, region *pdapi.HistoryHotRegion) (bool, error) { startZoneRedundant, err := isZoneRedundant(tc, coordinator, region.StartKey) if err != nil { klog.Errorf("isZoneRedundantRegion: [%s/%s] region id %d, start key [%s], err %v", tc.GetNamespace(), tc.GetName(), region.RegionID, region.StartKey, err) return false, err } endZoneRedundant, err := isZoneRedundant(tc, coordinator, region.EndKey) if err != nil { klog.Errorf("isZoneRedundantRegion: [%s/%s] region id %d, end key [%s], err %v", tc.GetNamespace(), tc.GetName(), region.RegionID, region.EndKey, err) return false, err } return startZoneRedundant || endZoneRedundant, nil } func isZoneRedundant(tc *v1alpha1.TidbCluster, coordinator coordinator.CoordinatorInterface, key string) (bool, error) { if len(key) == 0 { // The empty key of full range is zone redundant. return true, nil } if len(key) < 2 { return false, fmt.Errorf("invalid key [%s]", key) } if key[:2] != "78" { return true, nil } if len(key) < 8 { return false, fmt.Errorf("invalid key [%s]", key) } id := key[2:8] d, err := hex.DecodeString(id) if err != nil { return false, err } keyspaceID := uint32(d[0])<<16 + uint32(d[1])<<8 + uint32(d[2]) return coordinator.IsZoneRedundant(tc, keyspaceID) } func deactivateHotCluster(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType, pdCli pdapi.PDClient, coordinator coordinator.CoordinatorInterface) error { err := disableHotRegionRules(tc, pdCli, coordinator) if err != nil { return err } var stores map[string]v1alpha1.TiKVStore switch memberType { case v1alpha1.TiKVMemberType: stores = tc.Status.TiKV.Stores tc.Status.TiKV.HotCluster.CoolingRegions = nil default: return nil } for _, store := range stores { id, err := strconv.ParseUint(store.ID, 10, 64) if err != nil { return err } storeInfo, err := pdCli.GetStore(id) if storeInfo.Status.RegionCount > 0 { klog.Errorf("deactivateHotCluster: There should be no hot peers for store %d", storeInfo.Store.GetId()) return controller.RequeueErrorf("TidbCluster: [%s/%s]'s store %s regions %d, waiting for the region of the hot cluster to be empty", tc.GetNamespace(), tc.GetName(), store.ID, storeInfo.Status.RegionCount) } } klog.Infof("deactivateHotCluster: [%s/%s] %s hot cluster have migrated all the hot regions", tc.GetNamespace(), tc.GetName(), memberType) return nil } func disableHotRegionRules(tc *v1alpha1.TidbCluster, pdCli pdapi.PDClient, coord coordinator.CoordinatorInterface) error { hotRegionRules, err := pdCli.GetRulesByGroup(placement.HotRegionRuleGroup) if err != nil { return err } var hotRegions = map[uint64]struct{}{} checkReady := func(rule *placement.Rule) (bool, error) { if strings.HasPrefix(rule.ID, "single-az-") { regionID, _, err := placement.ParseSingleAZRegionRuleID(rule.ID) if err != nil { return false, nil } hotRegions[regionID] = struct{}{} } else if strings.HasPrefix(rule.ID, "multi-az-") { regionID, _, err := placement.ParseMultiAZRegionRuleID(rule.ID) if err != nil { return false, nil } hotRegions[regionID] = struct{}{} } ready, err := checkHotRuleDeletionReady(tc, rule, coord) if err != nil { return false, err } if !ready { return false, controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to switch zone redundancy, rule %s", tc.GetNamespace(), tc.GetName(), rule.ID) } return true, nil } err = coordinator.DisableHotTierRules(pdCli, hotRegionRules, checkReady) if err != nil { return err } klog.Infof("disableHotRegionRules: [%s/%s] hot cluster disable hot rules %d, hot regions %d", tc.GetNamespace(), tc.GetName(), len(hotRegionRules), len(hotRegions)) if len(hotRegions) > 0 { return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to migrate hot peers to standard tier, hot regions %d", tc.GetNamespace(), tc.GetName(), len(hotRegions)) } return err } func isHotClusterEnabled(hotCluster *v1alpha1.HotCluster) bool { return hotCluster != nil && hotCluster.Enabled != nil && *hotCluster.Enabled } func syncHotClusterStatus(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType, pdCli pdapi.PDClient, coord coordinator.CoordinatorInterface, deps *controller.Dependencies) error { var hot *v1alpha1.HotCluster var replicas *int32 var numStores int var protection *v1alpha1.ScaleInProtection var autoScaling *v1alpha1.AutoScaling var hotClusterActive *bool var autoScalingStatus *v1alpha1.AutoScalingStatus switch memberType { case v1alpha1.TiKVMemberType: hot = tc.Spec.TiKV.HotCluster replicas = &tc.Spec.TiKV.Replicas numStores = len(tc.Status.TiKV.Stores) protection = tc.Spec.TiKV.ScaleInProtection autoScaling = tc.Spec.TiKV.AutoScaling hotClusterActive = &tc.Status.TiKV.HotCluster.Active autoScalingStatus = &tc.Status.TiKV.AutoScaling default: return nil } if !isHotCluster(tc, hot) { return nil } if isHotClusterEnabled(hot) { if !*hot.ScaleToZero && !*hotClusterActive { *hotClusterActive = true return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to activate %s hot cluster", tc.GetNamespace(), tc.GetName(), memberType) } } else if *hotClusterActive { *hotClusterActive = false return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to deactivate %s hot cluster", tc.GetNamespace(), tc.GetName(), memberType) } if *hotClusterActive && *hot.InitialReplicas > 0 { var targetReplicas = *hot.InitialReplicas if *hot.ScaleToZero { targetReplicas = 1 } if targetReplicas != *replicas { if err := cancelScaleInBeforeHotClusterReplicaReset(tc, memberType, deps); err != nil { return err } if autoScalingStatus.Replicas < *hot.InitialReplicas-targetReplicas { autoScalingStatus.Replicas = *hot.InitialReplicas - targetReplicas } *replicas = targetReplicas return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to reset %s hot cluster replicas to %d", tc.GetNamespace(), tc.GetName(), memberType, targetReplicas) } if *hot.ScaleToZero { if protection != nil && *protection.Enabled { *protection.Enabled = false return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to disable scale-in protection allows %s hot cluster to scale to zero", tc.GetNamespace(), tc.GetName(), memberType) } } else if protection != nil && (!*protection.Enabled || *protection.MinReplicas != *replicas) { *protection.Enabled = true *protection.MinReplicas = *replicas return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to enable scale-in protection for %s hot cluster", tc.GetNamespace(), tc.GetName(), memberType) } if autoScaling != nil && !*autoScaling.Enabled { *autoScaling.Enabled = true return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to enable auto scaling for %s hot cluster", tc.GetNamespace(), tc.GetName(), memberType) } if numStores > 0 { err := syncHotCluster(tc, memberType, hot, pdCli, coord) if err != nil { if controller.IsRequeueError(err) { return err } klog.Errorf("syncHotCluster: [%s/%s] %s, error: %v", tc.GetNamespace(), tc.GetName(), memberType, err) return nil } } } else if *replicas > 0 { if *hot.InitialReplicas > 0 { err := deactivateHotCluster(tc, memberType, pdCli, coord) if err != nil { if controller.IsRequeueError(err) { return err } klog.Errorf("deactivateHotCluster: [%s/%s] %s, error: %v", tc.GetNamespace(), tc.GetName(), memberType, err) return nil } klog.Infof("TidbCluster: [%s/%s] %s hot cluster automatically scales to zero", tc.GetNamespace(), tc.GetName(), memberType) } if protection != nil && *protection.Enabled { *protection.Enabled = false return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to disable scale-in protection allows %s hot cluster to scale to zero", tc.GetNamespace(), tc.GetName(), memberType) } *replicas = 0 autoScalingStatus.Replicas = 0 autoScalingStatus.LastIdleTime = metav1.Time{} return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to reset %s hot cluster replicas to %d", tc.GetNamespace(), tc.GetName(), memberType, *replicas) } return nil } func scaleReplicas(replicas int32, usage float64, predictedScaleInUsage float64, threshold *v1alpha1.AutoScalingThreshold) int32 { if usage > threshold.MaxThreshold { replicas += *threshold.ScaleOutReplicas } else if usage < threshold.MinThreshold { desiredThreshold := (threshold.MinThreshold + threshold.MaxThreshold) / 2 if replicas > 0 && predictedScaleInUsage < desiredThreshold { replicas -= *threshold.ScaleInReplicas if replicas < 0 { replicas = 0 } } } return replicas } func syncSpecReplicasStatus(specReplicas int32, specReplicasStatus *v1alpha1.SpecReplicasStatus) { if specReplicasStatus == nil { return } if specReplicas != specReplicasStatus.Current { specReplicasStatus.Last = specReplicasStatus.Current specReplicasStatus.Current = specReplicas } } func scaleInProtection(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType, scaleInProtection *v1alpha1.ScaleInProtection, oldSet *apps.StatefulSet, newSet *apps.StatefulSet, replicas *int32, specReplicasStatus *v1alpha1.SpecReplicasStatus, deps *controller.Dependencies) error { currentReplicas := *oldSet.Spec.Replicas targetReplicas := *newSet.Spec.Replicas if currentReplicas-targetReplicas > 0 { var stores map[string]v1alpha1.TiKVStore switch memberType { case v1alpha1.TiKVMemberType: stores = tc.Status.TiKV.Stores case v1alpha1.TiFlashMemberType: stores = tc.Status.TiFlash.Stores default: return nil } currentPodOrdinals := helper.GetPodOrdinals(currentReplicas, oldSet) targetPodOrdinals := helper.GetPodOrdinals(targetReplicas, newSet) deletions := currentPodOrdinals.Difference(targetPodOrdinals) var scaleInPodNames = make(map[string]struct{}, deletions.Len()) for _, ordinal := range deletions.List() { var podName string switch memberType { case v1alpha1.TiKVMemberType: podName = TikvPodName(tc.GetName(), ordinal) case v1alpha1.TiFlashMemberType: podName = TiFlashPodName(tc.GetName(), ordinal) default: return nil } scaleInPodNames[podName] = struct{}{} } var predicted storeResource for _, store := range stores { if store.State == v1alpha1.TiKVStateOffline { continue } s := getStoreResource(store) if s == nil { resetReplicas(newSet, oldSet) return controller.RequeueErrorf("scaleInProtection: [%s/%s]'s %s %s storage usage is not ready %s", tc.GetNamespace(), tc.GetName(), memberType, store.PodName, store.State) } if _, ok := scaleInPodNames[store.PodName]; ok { predicted.addUsed(*s) } else { predicted.add(*s) } } var predictedStorageUsage = float64(1) if targetReplicas > 0 { predictedStorageUsage = predicted.storageUsage() } if predictedStorageUsage > *scaleInProtection.StorageThreshold || targetReplicas < *scaleInProtection.MinReplicas { resetReplicas(newSet, oldSet) klog.Errorf("scaleInProtection: [%s/%s]'s %s failed to scale in cluster because predicted storage usage %v is higher than storage threshold %v or target replicas %v is less than min replicas %v", tc.GetNamespace(), tc.GetName(), memberType, predictedStorageUsage, *scaleInProtection.StorageThreshold, targetReplicas, *scaleInProtection.MinReplicas) return resetSpecReplicas(tc, memberType, replicas, specReplicasStatus, deps) } } return nil } func cancelScaleInBeforeHotClusterReplicaReset(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType, deps *controller.Dependencies) error { if !coordinator.GetTidbClusterStatus(tc).GetMemberStatus(memberType).IsScalingIn() { return nil } // If activation interrupts scale-to-zero, finish canceling pending store deletion first. // Tombstone stores must not keep their old PVCs because TiKV would reuse the tombstoned store ID. klog.Infof("syncHotClusterStatus: [%s/%s] %s hot cluster cancel scale-in before resetting replicas", tc.GetNamespace(), tc.GetName(), memberType) return cancelDeleteAllStores(tc, memberType, deps) } func resetSpecReplicas(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType, replicas *int32, specReplicasStatus *v1alpha1.SpecReplicasStatus, deps *controller.Dependencies) error { lastReplicas := specReplicasStatus.Last currentReplicas := *replicas if lastReplicas <= currentReplicas { err := cancelScalingIn(tc, memberType, deps) if err != nil { return err } return controller.IgnoreErrorf("reset spec replicas: cannot reset %s current replicas %v to last replicas %v for tc [%s/%s]", memberType, currentReplicas, lastReplicas, tc.GetNamespace(), tc.GetName()) } *replicas = lastReplicas specReplicasStatus.Current = lastReplicas specReplicasStatus.Last = 0 if _, err := deps.TiDBClusterControl.Update(tc.DeepCopy()); err != nil { return controller.IgnoreErrorf("reset spec replicas: failed to update TidbCluster: [%s/%s], error: %v", tc.GetNamespace(), tc.GetName(), err) } return controller.RequeueErrorf("tidbcluster: reset %s replicas from %v to %v for tc [%s/%s]", memberType, currentReplicas, lastReplicas, tc.GetNamespace(), tc.GetName()) } func cancelScalingIn(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType, deps *controller.Dependencies) error { var autoScalingStatus *v1alpha1.AutoScalingStatus switch memberType { case v1alpha1.TiKVMemberType: autoScalingStatus = &tc.Status.TiKV.AutoScaling case v1alpha1.TiFlashMemberType: autoScalingStatus = &tc.Status.TiFlash.AutoScaling default: return nil } memberStatus := coordinator.GetTidbClusterStatus(tc).GetMemberStatus(memberType) if memberStatus.IsScalingIn() { if err := cancelDeleteAllStores(tc, memberType, deps); err != nil { return err } targetReplicas := autoScalingStatus.Replicas + memberStatus.ScaleInReplicas() setAutoScalingStatus(autoScalingStatus, targetReplicas) if _, err := deps.TiDBClusterControl.Update(tc.DeepCopy()); err != nil { return controller.IgnoreErrorf("cancelScalingIn: failed to update TidbCluster: [%s/%s], error: %v", tc.GetNamespace(), tc.GetName(), err) } return controller.RequeueErrorf("TidbCluster: [%s/%s], reset %s auto scaling status replicas to %d", tc.GetNamespace(), tc.GetName(), memberType, targetReplicas) } return nil } func isNotEmptyTierCluster(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType, pdCli pdapi.PDClient, groupID string) (notEmpty bool, err error) { var replicas *int32 var storeLabels []string numStores := 0 switch memberType { case v1alpha1.TiKVMemberType: replicas = &tc.Spec.TiKV.Replicas storeLabels = tc.Spec.TiKV.StoreLabels numStores = len(tc.Status.TiKV.Stores) default: return false, nil } var hasTierLabel bool for _, storeLabel := range storeLabels { if storeLabel == placement.TierLabel { hasTierLabel = true break } } if !hasTierLabel { return false, fmt.Errorf("failed to find tier store label %s in TidbCluster [%s/%s] %s", placement.TierLabel, tc.GetNamespace(), tc.GetName(), memberType) } ruleGroup, _ := pdCli.GetRuleGroup(groupID) if ruleGroup == nil { return false, fmt.Errorf("failed to find rule group %s in TidbCluster [%s/%s] %s", groupID, tc.GetNamespace(), tc.GetName(), memberType) } if *replicas == 0 || numStores == 0 { return false, nil } return true, nil } func isIsolatedCluster(isolatedCluster *v1alpha1.IsolatedCluster) bool { if isolatedCluster == nil || len(isolatedCluster.Tier) == 0 { return false } return true } func getIsolatedTier(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType) (string, error) { tier, err := getClusterTier(tc, memberType) if err != nil { return "", err } if tier == placement.StandardTier || tier == placement.HotTier { return "", nil } return tier, nil } func syncIsolatedCluster(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType, pdCli pdapi.PDClient, coord coordinator.CoordinatorInterface) error { tier, err := getIsolatedTier(tc, memberType) if err != nil || len(tier) == 0 { return err } switch memberType { case v1alpha1.TiKVMemberType: isolatedClusterReplicas := tc.Spec.TiKV.IsolatedClusterReplicas if isolatedClusterReplicas != nil { tc.Spec.TiKV.Replicas = *isolatedClusterReplicas } default: return nil } notEmpty, err := isNotEmptyTierCluster(tc, memberType, pdCli, placement.IsolatedRuleGroup) if err != nil { return err } if !notEmpty { return nil } var isolated *v1alpha1.IsolatedCluster var replicas int32 var stores map[string]v1alpha1.TiKVStore var protection *v1alpha1.ScaleInProtection upCount := 0 var isUpgrading bool switch memberType { case v1alpha1.TiKVMemberType: isolated = tc.Spec.TiKV.IsolatedCluster replicas = tc.Spec.TiKV.Replicas stores = tc.Status.TiKV.Stores protection = tc.Spec.TiKV.ScaleInProtection for _, store := range tc.Status.TiKV.Stores { if store.State == v1alpha1.TiKVStateUp { upCount++ } } isUpgrading = tc.TiKVUpgrading() default: return nil } if protection != nil { tierRegionCount := 0 for _, store := range stores { id, err := strconv.ParseUint(store.ID, 10, 64) if err != nil { return err } storeInfo, err := pdCli.GetStore(id) if err != nil { return err } tierRegionCount += storeInfo.Status.RegionCount } if tierRegionCount == 0 { if *protection.Enabled { *protection.Enabled = false klog.Infof("TidbCluster: [%s/%s], disable scale-in protection for %s isolated cluster", tc.GetNamespace(), tc.GetName(), memberType) } } else if !*protection.Enabled || *protection.MinReplicas != replicas { *protection.Enabled = true *protection.MinReplicas = replicas return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to enable scale-in protection for %s isolated cluster, replicas %d", tc.GetNamespace(), tc.GetName(), memberType, replicas) } } if !isIsolatedCluster(isolated) { return nil } if tier != isolated.Tier { klog.Errorf("TidbCluster: [%s/%s], Invalid tier %s for %s %s isolated cluster", tc.GetNamespace(), tc.GetName(), isolated.Tier, tier, memberType) return nil } var onlyLeader bool if isolated.OnlyLeader != nil { onlyLeader = *isolated.OnlyLeader } if !onlyLeader { if !coord.IsReadyForPhase(tc, memberType, v1alpha1.IsolatedPhase) { return nil } } var zoneRedundant bool if isolated.ZoneRedundant != nil { zoneRedundant = *isolated.ZoneRedundant } tierLabelConstraint := placement.TierLabelConstraint(isolated.Tier) singleAZTemplateRule, err := getSingleAZTemplateRule(tc, pdCli, placement.IsolatedRuleGroup, tierLabelConstraint) if err != nil { return fmt.Errorf("failed to find single az placement rule for TidbCluster [%s/%s]", tc.GetNamespace(), tc.GetName()) } var leaderRuleRole = placement.Leader if isUpgrading { leaderRuleRole = placement.Voter } tierPrefix := fmt.Sprintf("tier-%s-", isolated.Tier) rules, err := getCurrentRules(pdCli, placement.IsolatedRuleGroup, tierPrefix) if err != nil { return err } rulesCount := len(rules) if rulesCount == 0 && upCount == 0 { return nil } var primaryTemplateRule, standbyTemplateRule, witnessTemplateRule *placement.Rule if tc.IsMultiAZEnabled() { primaryTemplateRule, standbyTemplateRule, witnessTemplateRule, err = getMultiAZTemplateRule(pdCli, placement.IsolatedRuleGroup, tierLabelConstraint) if err != nil { return err } if primaryTemplateRule == nil || standbyTemplateRule == nil || witnessTemplateRule == nil { return nil } } storeTiers, err := coordinator.GetStoreTiers(pdCli) if err != nil { return err } var updateRulesList = [][]*placement.Rule{} var updateCount int var multiAZCount int getBatchRules := func(multiAZRuleBatch func() *placement.MultiAZRuleBatch, singleAZRuleBatch func() *placement.SingleAZRuleBatch) { var newRules []*placement.Rule if zoneRedundant { ruleBatch := multiAZRuleBatch() newRules = getMultiAZRules(ruleBatch, rules, primaryTemplateRule.Role, onlyLeader, primaryTemplateRule, standbyTemplateRule, witnessTemplateRule) if len(newRules) > 0 { multiAZCount++ } } else { ruleBatch := singleAZRuleBatch() if onlyLeader { newRules = getSingleAZRulesWithOnlyLeader(ruleBatch, rules, leaderRuleRole, singleAZTemplateRule) } else { checkReady := func(rule *placement.Rule) bool { ready, err := coordinator.CheckIfTierMigrationReady(pdCli, storeTiers, placement.StandardTier, isolated.Tier, rule) if err != nil { return false } return ready } newRules = getSingleAZRules(ruleBatch, rules, singleAZTemplateRule, checkReady) } } if len(newRules) > 0 { updateRulesList = append(updateRulesList, newRules) updateCount += len(newRules) } } for _, regionID := range isolated.RegionIDs { region, err := pdCli.GetRegionByID(regionID) if err != nil { return err } getBatchRules(func() *placement.MultiAZRuleBatch { return placement.GetMultiAZRegionRuleBatch(region.ID, region.StartKey, region.EndKey) }, func() *placement.SingleAZRuleBatch { return placement.GetSingleAZRegionRuleBatch(region.ID, region.StartKey, region.EndKey) }) } for _, isolatedKeyspace := range isolated.Keyspaces { keyspaceInfo, err := pdCli.GetKeyspace(isolatedKeyspace.KeyspaceName) if err != nil { return err } keyspaceID := keyspaceInfo.Id if len(isolatedKeyspace.TableIDs) == 0 { getBatchRules(func() *placement.MultiAZRuleBatch { return placement.GetMultiAZKeyspaceRuleBatch(keyspaceID) }, func() *placement.SingleAZRuleBatch { return placement.GetSingleAZKeyspaceRuleBatch(keyspaceID) }) } else { for _, tableID := range isolatedKeyspace.TableIDs { getBatchRules(func() *placement.MultiAZRuleBatch { return placement.GetMultiAZKeyspaceTableRuleBatch(keyspaceID, tableID) }, func() *placement.SingleAZRuleBatch { return placement.GetSingleAZKeyspaceTableRuleBatch(keyspaceID, tableID) }) } } } if len(updateRulesList) > 0 { var singleAZCount = len(updateRulesList) - multiAZCount klog.Infof("syncIsolatedCluster: [%s/%s] %s isolated cluster already has rules %d, update regions %d (single-az %d, multi-az %d), update rules %d", tc.GetNamespace(), tc.GetName(), memberType, rulesCount, len(updateRulesList), singleAZCount, multiAZCount, updateCount) err = coordinator.UpdateRulesList(pdCli, updateRulesList, tierPrefix) if err != nil { return err } } var disableRules = make([]*placement.Rule, 0, len(rules)) for _, rule := range rules { disableRules = append(disableRules, rule) } return coordinator.DisableTierRules(pdCli, isolated.Tier, tierPrefix, disableRules, nil) } func checkPDAvailable(tc *v1alpha1.TidbCluster, deps *controller.Dependencies) error { pdClient := controller.GetPDClient(deps.PDControl, tc) healthInfo, err := pdClient.GetHealth() healths := 0 if err == nil && healthInfo != nil && len(healthInfo.Healths) > 0 { for _, m := range healthInfo.Healths { if m.Health { healths++ } } } if healths == 0 { return controller.RequeueErrorf("tidbcluster: external cluster [%s/%s]'s pd is not available", tc.Spec.Cluster.Namespace, tc.Spec.Cluster.Name) } if len(tc.GetClusterID()) == 0 { cluster, err := pdClient.GetCluster() if err != nil { return err } tc.Status.ClusterID = strconv.FormatUint(cluster.Id, 10) } return nil } func isSuperseding(superseding *v1alpha1.TidbClusterRef) bool { return superseding != nil && len(superseding.Name) > 0 } func hasIsolatedClusterRanges(isolatedCluster *v1alpha1.IsolatedCluster) bool { return isolatedCluster != nil && (len(isolatedCluster.Keyspaces) > 0 || len(isolatedCluster.RegionIDs) > 0) } func clearIsolatedClusterRanges(isolatedCluster *v1alpha1.IsolatedCluster) { if isolatedCluster == nil { return } isolatedCluster.Keyspaces = nil isolatedCluster.RegionIDs = nil } func supersedeTiDBCluster(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType, deps *controller.Dependencies) error { var superseding *v1alpha1.TidbClusterRef switch memberType { case v1alpha1.TiKVMemberType: superseding = tc.Spec.TiKV.SupersedingCluster case v1alpha1.TiFlashMemberType: superseding = tc.Spec.TiFlash.SupersedingCluster default: return nil } if !isSuperseding(superseding) { return nil } cachedSupersedingTC, err := deps.TiDBClusterLister.TidbClusters(superseding.Namespace).Get(superseding.Name) if err != nil { if errors.IsNotFound(err) { switch memberType { case v1alpha1.TiKVMemberType: tc.Spec.TiKV.SupersedingCluster = nil case v1alpha1.TiFlashMemberType: tc.Spec.TiFlash.SupersedingCluster = nil default: return nil } klog.Infof("set [%s/%s]'s %s superseding cluster to empty", tc.GetNamespace(), tc.GetName(), memberType.String()) if _, err := deps.TiDBClusterControl.Update(tc.DeepCopy()); err != nil { klog.Errorf("failed to update TidbCluster: [%s/%s], error: %v", tc.GetNamespace(), tc.GetName(), err) return nil } return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to set superseding cluster to empty", tc.GetNamespace(), tc.GetName()) } } else { supersedingTC := cachedSupersedingTC.DeepCopy() var replicas *int32 var protection *v1alpha1.ScaleInProtection var autoScaling *v1alpha1.AutoScaling var hot *v1alpha1.HotCluster var isolatedCluster *v1alpha1.IsolatedCluster var isolatedClusterReplicas *int32 switch memberType { case v1alpha1.TiKVMemberType: replicas = &supersedingTC.Spec.TiKV.Replicas protection = supersedingTC.Spec.TiKV.ScaleInProtection autoScaling = supersedingTC.Spec.TiKV.AutoScaling isolatedCluster = supersedingTC.Spec.TiKV.IsolatedCluster isolatedClusterReplicas = supersedingTC.Spec.TiKV.IsolatedClusterReplicas if isHotCluster(tc, tc.Spec.TiKV.HotCluster) && isHotCluster(supersedingTC, supersedingTC.Spec.TiKV.HotCluster) { if supersedingTC.Status.TiKV.HotCluster.Active && len(supersedingTC.Status.TiKV.Stores) > 0 { tc.Status.TiKV.HotCluster.Active = true if len(tc.Status.TiKV.Stores) > 0 { hot = supersedingTC.Spec.TiKV.HotCluster } } } isolatedTier, _ := getIsolatedTier(tc, memberType) supersedingIsolatedTier, _ := getIsolatedTier(supersedingTC, memberType) if len(isolatedTier) > 0 && isolatedTier == supersedingIsolatedTier { if isIsolatedCluster(supersedingTC.Spec.TiKV.IsolatedCluster) { if tc.Spec.TiKV.IsolatedCluster == nil { tc.Spec.TiKV.IsolatedCluster = supersedingTC.Spec.TiKV.IsolatedCluster.DeepCopy() } } if isolatedClusterReplicas != nil && *isolatedClusterReplicas > 0 { tc.Spec.TiKV.IsolatedClusterReplicas = pointer.Int32Ptr(*isolatedClusterReplicas) } } case v1alpha1.TiFlashMemberType: replicas = &supersedingTC.Spec.TiFlash.Replicas protection = supersedingTC.Spec.TiFlash.ScaleInProtection autoScaling = supersedingTC.Spec.TiFlash.AutoScaling default: return nil } needUpdate := *replicas > 0 || (protection != nil && *protection.Enabled) || (autoScaling != nil && *autoScaling.Enabled) || (hot != nil && *hot.InitialReplicas > 0) || (isolatedClusterReplicas != nil && *isolatedClusterReplicas > 0) || hasIsolatedClusterRanges(isolatedCluster) if needUpdate { *replicas = 0 if protection != nil { *protection.Enabled = false } if autoScaling != nil { *autoScaling.Enabled = false } if hot != nil { *hot.InitialReplicas = 0 } if isolatedClusterReplicas != nil { *isolatedClusterReplicas = 0 } // Ranges are desired placement inputs. They were copied to the new TC // above when tiers match, so clearing them only lets the old TC scale to 0. clearIsolatedClusterRanges(isolatedCluster) klog.Infof("set [%s/%s]'s %s replicas to 0, disable scale-in protection, auto scaling, hot cluster and isolated cluster", supersedingTC.GetNamespace(), supersedingTC.GetName(), memberType.String()) if _, err := deps.TiDBClusterControl.Update(supersedingTC); err != nil { klog.Errorf("failed to update TidbCluster: [%s/%s], error: %v", supersedingTC.GetNamespace(), supersedingTC.GetName(), err) } } } return nil } func isEmptyFallbackCluster(fallback *v1alpha1.FallbackCluster) bool { if fallback == nil || fallback.Cluster == nil || len(fallback.Cluster.Name) == 0 { return true } return false } func isFallbackCluster(tc *v1alpha1.TidbCluster, fallback *v1alpha1.FallbackCluster) bool { if isEmptyFallbackCluster(fallback) { return false } return fallback.Cluster.Name == tc.GetName() && fallback.Cluster.Namespace == tc.GetNamespace() } func hasFallbackCluster(tc *v1alpha1.TidbCluster, fallback *v1alpha1.FallbackCluster) bool { if isEmptyFallbackCluster(fallback) { return false } return fallback.Cluster.Name != tc.GetName() || fallback.Cluster.Namespace != tc.GetNamespace() } func syncFallbackCluster(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType, set *apps.StatefulSet, upgrading bool, deps *controller.Dependencies) error { var fallback *v1alpha1.FallbackCluster switch memberType { case v1alpha1.TiKVMemberType: fallback = tc.Spec.TiKV.FallbackCluster default: return nil } if hasFallbackCluster(tc, fallback) { return checkPodsForFallbackCluster(tc, memberType, set, upgrading, deps) } return syncFallbackClusterStatus(tc, memberType) } func syncFallbackClusterStatus(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType) error { var fallback *v1alpha1.FallbackCluster var replicas *int32 var protection *v1alpha1.ScaleInProtection var autoScaling *v1alpha1.AutoScaling var fallbackClusterStatus *v1alpha1.FallbackClusterStatus var autoScalingStatus *v1alpha1.AutoScalingStatus switch memberType { case v1alpha1.TiKVMemberType: fallback = tc.Spec.TiKV.FallbackCluster replicas = &tc.Spec.TiKV.Replicas protection = tc.Spec.TiKV.ScaleInProtection autoScaling = tc.Spec.TiKV.AutoScaling fallbackClusterStatus = &tc.Status.TiKV.FallbackCluster autoScalingStatus = &tc.Status.TiKV.AutoScaling default: return nil } if !isFallbackCluster(tc, fallback) { return nil } if !tc.IsTiKVFallbackClusterEnabled() && fallbackClusterStatus.Active { fallbackClusterStatus.Active = false return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to deactivate %s fallback cluster", tc.GetNamespace(), tc.GetName(), memberType) } if fallbackClusterStatus.Active && fallbackClusterStatus.InitialReplicas > 0 { if *replicas != 1 { *replicas = 1 autoScalingStatus.Replicas = fallbackClusterStatus.InitialReplicas - 1 return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to reset %s fallback cluster replicas to 1", tc.GetNamespace(), tc.GetName(), memberType) } if protection != nil && *protection.Enabled { *protection.Enabled = false return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to disable scale-in protection allows %s fallback cluster to scale to zero", tc.GetNamespace(), tc.GetName(), memberType) } if autoScaling != nil && !*autoScaling.Enabled { *autoScaling.Enabled = true return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to enable auto scaling for %s fallback cluster", tc.GetNamespace(), tc.GetName(), memberType) } } else if *replicas > 0 { if fallbackClusterStatus.InitialReplicas > 0 { fallbackClusterStatus.InitialReplicas = 0 return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to reset %s fallback cluster initial replicas to 0", tc.GetNamespace(), tc.GetName(), memberType) } if protection != nil && *protection.Enabled { *protection.Enabled = false return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to disable scale-in protection allows %s fallback cluster to scale to zero", tc.GetNamespace(), tc.GetName(), memberType) } *replicas = 0 autoScalingStatus.Replicas = 0 autoScalingStatus.LastIdleTime = metav1.Time{} return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to reset %s fallback cluster replicas to %d", tc.GetNamespace(), tc.GetName(), memberType, *replicas) } return nil } func checkPodsForFallbackCluster(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType, set *apps.StatefulSet, upgrading bool, deps *controller.Dependencies) error { ns := tc.GetNamespace() tcName := tc.GetName() var fallback *v1alpha1.FallbackCluster var replicas int32 var fallbackClusterStatus *v1alpha1.FallbackClusterStatus var status *v1alpha1.TiKVStatus switch memberType { case v1alpha1.TiKVMemberType: fallback = tc.Spec.TiKV.FallbackCluster replicas = tc.Spec.TiKV.Replicas fallbackClusterStatus = &tc.Status.TiKV.FallbackCluster status = &tc.Status.TiKV default: return nil } if replicas == 0 { *fallbackClusterStatus = v1alpha1.FallbackClusterStatus{} return updateFallbackClusterActive(tc, memberType, fallback, false, 0, deps) } if set.Status.Replicas == 0 { return nil } upCnt := 0 pendingCnt := int32(0) podOrdinals := helper.GetPodOrdinals(*set.Spec.Replicas, set).List() for _i := len(podOrdinals) - 1; _i >= 0; _i-- { i := podOrdinals[_i] store := getStoreByOrdinal(tc.GetName(), *status, i) if store != nil && store.State == v1alpha1.TiKVStateUp { upCnt++ continue } podName := TikvPodName(tcName, i) pod, err := deps.PodLister.Pods(ns).Get(podName) if err != nil { return fmt.Errorf("checkPodsForFallbackCluster: failed to get pod %s for cluster %s/%s, error: %s", podName, ns, tcName, err) } if store != nil { klog.Infof("checkPodsForFallbackCluster: pod %s/%s store %s state %s", podName, ns, store.ID, store.State) continue } if pod.Status.Phase == corev1.PodPending { pendingCnt++ } } fallbackTC, err := deps.TiDBClusterLister.TidbClusters(fallback.Cluster.Namespace).Get(fallback.Cluster.Name) if err != nil { if errors.IsNotFound(err) { return controller.IgnoreErrorf("failed to get fallback TidbCluster: [%s/%s] %s, error: %v", fallback.Cluster.Namespace, fallback.Cluster.Name, memberType, err) } return err } if pendingCnt != fallbackClusterStatus.PendingReplicas { klog.Infof("checkPodsForFallbackCluster: upgrading %t, pending replicas new %d, old %d", upgrading, pendingCnt, fallbackClusterStatus.PendingReplicas) } fallbackClusterStatus.Upgrading = upgrading if upgrading && fallbackClusterStatus.PendingReplicas > 0 { klog.Infof("checkPodsForFallbackCluster: TidbCluster [%s/%s] %s is upgrading; temporarily scaling in pending pods", ns, tcName, memberType) return nil } fallbackClusterStatus.PendingReplicas = pendingCnt if pendingCnt > 0 { fallbackClusterStatus.LastReadyTime = metav1.Time{} if fallbackClusterStatus.PendingStartTime.IsZero() { fallbackClusterStatus.PendingStartTime = metav1.Now() } var timeout = time.Minute * 10 if fallback.ScaleOutTimeout != nil { d, err := time.ParseDuration(*fallback.ScaleOutTimeout) if err == nil { timeout = d } } now := time.Now() pendingDeadline := fallbackClusterStatus.PendingStartTime.Add(timeout) if pendingDeadline.After(now) { dur := pendingDeadline.Sub(now) klog.Infof("checkPodsForFallbackCluster: [%s/%s] is not pending enough time, wait %s, pending start time is %v", ns, tcName, dur, fallbackClusterStatus.PendingStartTime) return nil } err := updateFallbackClusterActive(tc, memberType, fallback, true, pendingCnt, deps) if err != nil { return err } } else if int(tc.TiKVStsDesiredReplicas()) == len(podOrdinals) { fallbackClusterStatus.PendingStartTime = metav1.Time{} if len(podOrdinals) == upCnt && fallbackTC.Status.TiKV.FallbackCluster.Active { if fallbackClusterStatus.LastReadyTime.IsZero() { fallbackClusterStatus.LastReadyTime = metav1.Now() } var readyDur = time.Minute * 10 if fallback.ReadyDuration != nil { d, err := time.ParseDuration(*fallback.ReadyDuration) if err == nil { readyDur = d } } now := time.Now() readyTime := fallbackClusterStatus.LastReadyTime.Add(readyDur) if readyTime.After(now) { dur := readyTime.Sub(now) klog.Infof("checkPodsForFallbackCluster: [%s/%s] is not ready enough time, wait %s, last ready time is %v", ns, tcName, dur, fallbackClusterStatus.LastReadyTime) return nil } err := updateFallbackClusterActive(tc, memberType, fallback, false, 0, deps) if err != nil { return err } } fallbackClusterStatus.LastReadyTime = metav1.Time{} } return nil } func updateFallbackClusterActive(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType, fallbackCluster *v1alpha1.FallbackCluster, active bool, replicas int32, deps *controller.Dependencies) error { if !tc.IsTiKVFallbackClusterEnabled() { active = false } fallbackTC, err := deps.TiDBClusterLister.TidbClusters(fallbackCluster.Cluster.Namespace).Get(fallbackCluster.Cluster.Name) if err != nil { if errors.IsNotFound(err) { return controller.IgnoreErrorf("failed to get fallback TidbCluster: [%s/%s] %s, error: %v", fallbackCluster.Cluster.Namespace, fallbackCluster.Cluster.Name, memberType, err) } return err } tier, err := getClusterTier(tc, memberType) if err != nil { return err } fallbackTier, err := getClusterTier(fallbackTC, memberType) if err != nil { return err } if tier != fallbackTier { return controller.IgnoreErrorf("[%s/%s] %s current cluster tier %s does not match the fallback cluster [%s/%s] %s tier %s", tc.GetNamespace(), tc.GetName(), memberType, tier, fallbackCluster.Cluster.Namespace, fallbackCluster.Cluster.Name, memberType, fallbackTier, ) } var nodeSelector map[string]string var fallbackNodeSelector map[string]string var fallbackClusterStatus *v1alpha1.FallbackClusterStatus var isScaling bool switch memberType { case v1alpha1.TiKVMemberType: nodeSelector = tc.Spec.TiKV.NodeSelector fallbackNodeSelector = fallbackTC.Spec.TiKV.NodeSelector fallbackClusterStatus = &fallbackTC.Status.TiKV.FallbackCluster isScaling = fallbackTC.TiKVScaling() default: return nil } if len(nodeSelector) > 0 && len(fallbackNodeSelector) > 0 { partition := nodeSelector[placement.PartitionLabel] fallbackPartition := fallbackNodeSelector[placement.PartitionLabel] if len(partition) > 0 && partition != fallbackPartition { return controller.IgnoreErrorf("[%s/%s] %s current cluster partition %s does not match the fallback cluster [%s/%s] %s partition %s", tc.GetNamespace(), tc.GetName(), memberType, partition, fallbackCluster.Cluster.Namespace, fallbackCluster.Cluster.Name, memberType, fallbackPartition, ) } zone := nodeSelector[placement.ZoneLabel] fallbackZone := fallbackNodeSelector[placement.ZoneLabel] if len(zone) > 0 && zone != fallbackZone { return controller.IgnoreErrorf("[%s/%s] %s current cluster zone %s does not match the fallback cluster [%s/%s] %s zone %s", tc.GetNamespace(), tc.GetName(), memberType, zone, fallbackCluster.Cluster.Namespace, fallbackCluster.Cluster.Name, memberType, fallbackZone, ) } } if isScaling { klog.Warningf("Fallback TidbCluster: [%s/%s] %s is scaling", fallbackTC.GetNamespace(), fallbackTC.GetName(), memberType) return nil } if fallbackClusterStatus.Active != active { fallbackClusterStatus.Active = active fallbackClusterStatus.InitialReplicas = replicas klog.Infof("update fallback cluster [%s/%s]'s %s, active %t replicas %d", fallbackTC.GetNamespace(), fallbackTC.GetName(), memberType.String(), active, replicas) if _, err := deps.TiDBClusterControl.Update(fallbackTC.DeepCopy()); err != nil { klog.Errorf("failed to update TidbCluster: [%s/%s], error: %v", fallbackTC.GetNamespace(), fallbackTC.GetName(), err) } } return nil } func getClusterTier(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType) (tier string, err error) { var storeLabels []string var nodeSelector map[string]string switch memberType { case v1alpha1.TiKVMemberType: storeLabels = tc.Spec.TiKV.StoreLabels nodeSelector = tc.Spec.TiKV.NodeSelector default: return "", nil } var hasTierLabel bool for _, storeLabel := range storeLabels { if storeLabel == placement.TierLabel { hasTierLabel = true break } } if !hasTierLabel { return "", controller.IgnoreErrorf("failed to find tier store label %s in TidbCluster [%s/%s] %s", placement.TierLabel, tc.GetNamespace(), tc.GetName(), memberType) } if len(nodeSelector) > 0 { tier = nodeSelector[placement.TierLabel] } if len(tier) == 0 { return "", controller.IgnoreErrorf("failed to find tier value %s in TidbCluster [%s/%s] %s", placement.TierLabel, tc.GetNamespace(), tc.GetName(), memberType) } return tier, nil } // ScalingType is the current scaling type type ScalingType string const ( // NonScalingType represents the non-scaling type. NonScalingType ScalingType = "NonScaling" // ScalingInType represents the scaling-in type. ScalingInType ScalingType = "ScalingIn" // ScalingOutType represents the scaling-out type. ScalingOutType ScalingType = "ScalingOut" ) func getStoreLimit(storeLimit *v1alpha1.StoreLimit, scalingType ScalingType) float64 { switch scalingType { case ScalingInType: return *storeLimit.ScalingInRate case ScalingOutType: return *storeLimit.ScalingOutRate default: return *storeLimit.NormalRate } } func isStoreLimitEnabled(storeLimit *v1alpha1.StoreLimit) bool { return storeLimit != nil && storeLimit.Enabled != nil && *storeLimit.Enabled } func syncStoreLimit(tc *v1alpha1.TidbCluster, component string, desiredReplicas, actualReplicas int32, storeRegionScores map[uint64]float64, storeLimit *v1alpha1.StoreLimit, superseding *v1alpha1.TidbClusterRef, deps *controller.Dependencies) { if !isStoreLimitEnabled(storeLimit) || len(storeRegionScores) == 0 { return } var scalingType ScalingType if desiredReplicas < actualReplicas { scalingType = ScalingInType } else if isSuperseding(superseding) { scalingType = ScalingOutType } else { scalingType = NonScalingType var storeRegionScoreSum float64 for _, storeRegionScore := range storeRegionScores { storeRegionScoreSum += storeRegionScore } var avgStoreRegionScore = storeRegionScoreSum / float64(len(storeRegionScores)) for _, storeRegionScore := range storeRegionScores { if storeRegionScore/avgStoreRegionScore < *storeLimit.UnbalanceThreshold { scalingType = ScalingOutType break } } } pdCli := controller.GetPDClient(deps.PDControl, tc) config, err := pdCli.GetConfig() if err != nil { klog.Errorf("syncStoreLimit: failed to get pd config, err %s", err.Error()) return } if config.Schedule.StoreLimitVersion != "" && config.Schedule.StoreLimitVersion != "v1" { klog.Warningf("syncStoreLimit: only support store limit v1") return } var targetRate = getStoreLimit(storeLimit, scalingType) for storeID := range storeRegionScores { var rate float64 if config.Schedule.StoreLimit != nil { rate = config.Schedule.StoreLimit[storeID].AddPeer } if rate != targetRate { err := pdCli.SetStoreLimit(storeID, targetRate) if err != nil { klog.Errorf("syncStoreLimit: failed to set [%s/%s]'s %s %d store limit from %.0f to %.0f, err %s", tc.GetNamespace(), tc.GetName(), component, storeID, rate, targetRate, err.Error()) return } klog.Infof("syncStoreLimit: set [%s/%s]'s %s %d store limit from %.0f to %.0f", tc.GetNamespace(), tc.GetName(), component, storeID, rate, targetRate) } } } func syncNodeScaleInProtectionStatus(tc *v1alpha1.TidbCluster, stores map[string]v1alpha1.TiKVStore, deps *controller.Dependencies) error { var hasNodeChange bool for id, store := range stores { nodeChange, _ := scaleInProtectionForNode(tc, store.PodName, true, &store, deps) if nodeChange { stores[id] = store hasNodeChange = true } } if hasNodeChange { if _, err := deps.TiDBClusterControl.Update(tc.DeepCopy()); err != nil { klog.Errorf("failed to update TidbCluster: [%s/%s], error: %v", tc.GetNamespace(), tc.GetName(), err) return nil } return controller.RequeueErrorf("TidbCluster: [%s/%s], waiting to sync node scale-in protection status", tc.GetNamespace(), tc.GetName()) } return nil } func scaleInProtectionForNode(tc *v1alpha1.TidbCluster, podName string, enabled bool, store *v1alpha1.TiKVStore, deps *controller.Dependencies) (nodeChange bool, err error) { pod, err := deps.PodLister.Pods(tc.GetNamespace()).Get(podName) if err != nil { return false, err } nodeName := pod.Spec.NodeName if enabled && store != nil { if store.NodeName != nodeName && nodeName != "" { err := updateNodeScaleInProtectionAnnotation(tc, podName, store.NodeName, false, deps) if err == nil { klog.Infof("scaleInProtectionForNode: schedule pod %s from [%s] to node %s", podName, store.NodeName, nodeName) store.NodeName = nodeName nodeChange = true } } } return nodeChange, updateNodeScaleInProtectionAnnotation(tc, podName, nodeName, enabled, deps) } func updateNodeScaleInProtectionAnnotation(tc *v1alpha1.TidbCluster, podName, nodeName string, enabled bool, deps *controller.Dependencies) error { if nodeName == "" { return nil } node, err := deps.NodeLister.Get(nodeName) if err != nil { return err } ann := node.GetAnnotations() if ann == nil { ann = map[string]string{} } var newAnnotations map[string]string var removeAnnotations []string if enabled { if ann[CAScaleDownDisabledAnnotation] == "true" { return nil } newAnnotations = map[string]string{} newAnnotations[CAScaleDownDisabledAnnotation] = "true" klog.Infof("updateNodeScaleInProtectionAnnotation: disable CA scale down for the node %s with the pod %s", nodeName, podName) } else { if _, ok := ann[CAScaleDownDisabledAnnotation]; !ok { return nil } removeAnnotations = []string{CAScaleDownDisabledAnnotation} klog.Infof("updateNodeScaleInProtectionAnnotation: enable CA scale down for the node %s with the pod %s", nodeName, podName) } _, err = deps.NodeControl.UpdateAnnotations(tc, node, newAnnotations, removeAnnotations) if err != nil { klog.Errorf("updateNodeScaleInProtectionAnnotation: failed to update the annotation for the node %s with the pod %s, err %s", nodeName, podName, err.Error()) } return err } func upgradeInNonReadyStates(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType, newSet, oldSet *apps.StatefulSet, coordinator coordinator.CoordinatorInterface, deps *controller.Dependencies) error { isUpgrading := !templateEqual(newSet, oldSet) || mngerutils.StatefulSetIsUpgrading(oldSet) if !isUpgrading { return nil } readyCount := 0 switch memberType { case v1alpha1.PDMemberType: for _, member := range tc.Status.PD.Members { if member.Health { readyCount++ } } case v1alpha1.TiKVMemberType: for _, store := range tc.Status.TiKV.Stores { if store.State == v1alpha1.TiKVStateUp { readyCount++ } } if readyCount > 0 && readyCount < len(tc.Status.TiKV.Stores) && tc.IsTiKVScaleInProtectionEnabled() && *newSet.Spec.Replicas < *oldSet.Spec.Replicas { // Suspend scaling-in if there are stores that are not ready. tc.Status.TiKV.Phase = v1alpha1.UpgradePhase resetReplicas(newSet, oldSet) return cancelDeleteAllStores(tc, memberType, deps) } case v1alpha1.TiFlashMemberType: for _, store := range tc.Status.TiFlash.Stores { if store.State == v1alpha1.TiKVStateUp { readyCount++ } } if readyCount > 0 && readyCount < len(tc.Status.TiFlash.Stores) && tc.IsTiFlashScaleInProtectionEnabled() && *newSet.Spec.Replicas < *oldSet.Spec.Replicas { // Suspend scaling-in if there are stores that are not ready. tc.Status.TiFlash.Phase = v1alpha1.UpgradePhase resetReplicas(newSet, oldSet) return cancelDeleteAllStores(tc, memberType, deps) } default: return nil } if readyCount == 0 { // Force upgrade if all members of the cluster are not ready. return forceUpgrade(tc, memberType, newSet, oldSet, coordinator, deps) } return nil } func cancelDeleteAllStores(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType, deps *controller.Dependencies) error { tcName := tc.GetName() ns := tc.GetNamespace() var stores map[string]v1alpha1.TiKVStore var pods = make(map[string]uint64) var stsActualReplicas int32 switch memberType { case v1alpha1.TiKVMemberType: stores = tc.Status.TiKV.Stores stsActualReplicas = tc.TiKVStsActualReplicas() case v1alpha1.TiFlashMemberType: stores = tc.Status.TiFlash.Stores stsActualReplicas = tc.TiFlashStsActualReplicas() default: return nil } pdCli := controller.GetPDClient(deps.PDControl, tc) for _, store := range stores { storeID, err := strconv.ParseUint(store.ID, 10, 64) if err != nil { return err } pods[store.PodName] = storeID if store.State == v1alpha1.TiKVStateOffline { err := pdCli.CancelDeleteStore(storeID) if err != nil { klog.Errorf("cancelDeleteAllStores: failed to cancel delete store %d of [%s/%s]'s %s, error: %v", storeID, ns, tcName, memberType, err) return err } klog.Infof("cancelDeleteAllStores: cancel delete store %d of [%s/%s]'s %s successfully", storeID, ns, tcName, memberType) } } klog.Infof("[%s/%s]'s %s stores %d replicas %d", ns, tcName, memberType, len(stores), stsActualReplicas) for ordinal := int32(0); ordinal < stsActualReplicas; ordinal++ { podName := ordinalPodName(memberType, tcName, ordinal) pod, err := deps.PodLister.Pods(tc.Namespace).Get(podName) if err != nil { klog.Errorf("cancelDeleteAllStores: failed to get pod %s for [%s/%s]'s %s, error: %s", podName, ns, tcName, memberType, err) return err } storeID, ok := pods[podName] if !ok { err = checkPodStoreIDLabel(tc, memberType, pod, deps) if err != nil { klog.Errorf("cancelDeleteAllStores: failed to check store id label for pod %s/%s in tc %s/%s, error: %s", ns, podName, ns, tcName, err) return err } storeIDStr := pod.Labels[label.StoreIDLabelKey] storeID, err = strconv.ParseUint(storeIDStr, 10, 64) if err != nil { klog.Errorf("cancelDeleteAllStores: failed to parse store id [%s] for pod %s/%s in tc %s/%s, error: %s", storeIDStr, ns, podName, ns, tcName, err) return err } } storeInfo, err := pdCli.GetStore(storeID) if err != nil { klog.Errorf("cancelDeleteAllStores: failed to get store %d info of [%s/%s]'s %s, error: %v", storeID, ns, tcName, memberType, err) return err } if storeInfo.Store.StateName != v1alpha1.TiKVStateTombstone { continue } // If the store is already Tombstone, delete its pvcs and pod to recreate store. klog.Infof("cancelDeleteAllStores: store [%d] of pod %s/%s in tc %s/%s became tombstone", storeID, ns, podName, ns, tcName) pvcs, err := util.ResolvePVCFromPod(pod, deps.PVCLister) if err != nil && !errors.IsNotFound(err) { klog.Errorf("cancelDeleteAllStores: failed to list pvcs for pod %s/%s in tc %s/%s err: %v", ns, podName, ns, tcName, err) return err } klog.Infof("cancelDeleteAllStores: list pvcs %d for pod %s/%s in tc %s/%s", len(pvcs), ns, podName, ns, tcName) var wg sync.WaitGroup var pvcErr error if len(pvcs) > 0 { wg.Add(1) go func() { defer wg.Done() for _, pvc := range pvcs { pvcName := pvc.Name pvcErr = deps.PVCControl.DeletePVC(tc, pvc) if pvcErr != nil { klog.Errorf("cancelDeleteAllStores: failed to delete pvc %s/%s, %v", ns, pvcName, pvcErr) } else { klog.Infof("cancelDeleteAllStores: delete pvc %s/%s successfully", ns, pvcName) } } klog.Infof("cancelDeleteAllStores: delete pvcs %d from pod %s/%s in tc %s/%s", len(pvcs), ns, podName, ns, tcName) }() } podErr := deps.PodControl.DeletePod(tc, pod) if podErr != nil { klog.Errorf("cancelDeleteAllStores: failed to delete pod %s/%s, %v", ns, podName, podErr) } else { klog.Infof("cancelDeleteAllStores: delete pod: %s/%s successfully", ns, podName) } wg.Wait() if podErr != nil { return podErr } if pvcErr != nil { return pvcErr } } return nil } func checkPodStoreIDLabel(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType, pod *v1.Pod, deps *controller.Dependencies) error { if len(pod.Labels[label.StoreIDLabelKey]) > 0 { return nil } tcName := tc.GetName() ns := tc.GetNamespace() pvcs, err := util.ResolvePVCFromPod(pod, deps.PVCLister) if err != nil { klog.Errorf("checkPodStoreIDLabel: failed to get pvcs for pod %s/%s in tc %s/%s, error: %s", ns, pod.Name, ns, tcName, err) return err } for _, pvc := range pvcs { if len(pvc.Labels) == 0 { continue } if len(pod.Labels[label.ClusterIDLabelKey]) == 0 { pod.Labels[label.ClusterIDLabelKey] = tc.Status.ClusterID } if pod.Labels[label.ClusterIDLabelKey] == pvc.Labels[label.ClusterIDLabelKey] && len(pvc.Labels[label.StoreIDLabelKey]) > 0 { pod.Labels[label.StoreIDLabelKey] = pvc.Labels[label.StoreIDLabelKey] klog.Warningf("checkPodStoreIDLabel: set store id label for %s %s/%s store %s from pvc %s", memberType, ns, pod.Name, pvc.Labels[label.StoreIDLabelKey], pvc.Name) updatedPod, err := deps.PodControl.UpdatePod(tc, pod) if err != nil { return err } pod = updatedPod break } } return nil } func forceUpgrade(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType, newSet, oldSet *apps.StatefulSet, coordinator coordinator.CoordinatorInterface, deps *controller.Dependencies) error { switch memberType { case v1alpha1.PDMemberType: tc.Status.PD.Phase = v1alpha1.UpgradePhase case v1alpha1.TiKVMemberType: tc.Status.TiKV.Phase = v1alpha1.UpgradePhase case v1alpha1.TiFlashMemberType: tc.Status.TiFlash.Phase = v1alpha1.UpgradePhase default: return nil } if *newSet.Spec.Replicas != *oldSet.Spec.Replicas { resetReplicas(newSet, oldSet) } if err := isReadyForUpgrade(tc, memberType, coordinator); err != nil { return err } mngerutils.SetUpgradePartition(newSet, 0) errSTS := mngerutils.UpdateStatefulSet(deps.StatefulSetControl, tc, newSet, oldSet) return controller.RequeueErrorf("tidbcluster: [%s/%s]'s %s force upgrade, %v", tc.GetNamespace(), tc.GetName(), memberType.String(), errSTS) } func isReadyForUpgrade(tc *v1alpha1.TidbCluster, memberType v1alpha1.MemberType, coordinator coordinator.CoordinatorInterface) error { var memberIPPattern string var enableUpgradeInNonReadyStates bool switch memberType { case v1alpha1.PDMemberType: memberIPPattern = pdIPPattern enableUpgradeInNonReadyStates = tc.IsPDUpgradeInNonReadyStatesEnabled() case v1alpha1.TiKVMemberType: memberIPPattern = tikvIPPattern enableUpgradeInNonReadyStates = tc.IsTiKVUpgradeInNonReadyStatesEnabled() case v1alpha1.TiFlashMemberType: memberIPPattern = tiflashIPPattern enableUpgradeInNonReadyStates = tc.IsTiFlashUpgradeInNonReadyStatesEnabled() default: return nil } if enableUpgradeInNonReadyStates { return nil } if !coordinator.IsReadyForPhase(tc, memberType, v1alpha1.UpgradePhase) { return controller.RequeueErrorf("tidbcluster: [%s/%s]'s %s is not ready to upgrade", tc.GetNamespace(), tc.GetName(), memberType.String()) } notReadyMembers := coordinator.NotReadyMembers(tc, memberType) notReadyExternalMembers, err := getExternalMembers(tc, memberIPPattern, notReadyMembers) if err != nil { return err } if len(notReadyExternalMembers) > 0 { return controller.RequeueErrorf("tidbcluster: [%s/%s]'s %s cannot be upgraded when other tc members %v in the same group are not ready", tc.GetNamespace(), tc.GetName(), memberType.String(), notReadyExternalMembers) } return nil } func getExternalMembers(tc *v1alpha1.TidbCluster, memberIPPattern string, members map[string]string) ([]string, error) { pattern, err := regexp.Compile(fmt.Sprintf(memberIPPattern, tc.Name, tc.Name, tc.Namespace, controller.FormatClusterDomainForRegex(tc.Spec.ClusterDomain))) if err != nil { return nil, err } var externalMembers []string for key, ip := range members { if !pattern.Match([]byte(ip)) { externalMembers = append(externalMembers, key) } } return externalMembers, nil } type PodMetrics struct { Metadata struct { Name string `json:"name"` Namespace string `json:"namespace"` Labels map[string]string `json:"labels"` CreationTimestamp time.Time `json:"creationTimestamp"` } `json:"metadata"` Timestamp time.Time `json:"timestamp"` Window string `json:"window"` Containers []struct { Name string `json:"name"` Usage struct { CPU string `json:"cpu"` Memory string `json:"memory"` } `json:"usage"` } `json:"containers"` } func parseCPUNano(cpu string) (int64, error) { length := len(cpu) if length < 2 { return 0, fmt.Errorf("invalid format %s", cpu) } i, err := strconv.ParseInt(cpu[:length-1], 10, 64) if err != nil { return 0, err } if cpu[length-1] == 'n' { return i, nil } if cpu[length-1] == 'u' { return i * 1000, nil } if cpu[length-1] == 'm' { return i * 1000 * 1000, nil } return 0, fmt.Errorf("invalid format %s", cpu) } type CPUMetrics struct { MilliCores uint64 UsedMilliCores uint64 Timestamp time.Time } func getPodCPUMetrics(namespace, podName string, deps *controller.Dependencies) (cpuMetrics *CPUMetrics, err error) { pod, err := deps.PodLister.Pods(namespace).Get(podName) if err != nil { return nil, err } nodeName := pod.Spec.NodeName node, err := deps.NodeLister.Get(nodeName) if err != nil { return nil, err } cpuMilliCores := node.Status.Capacity.Cpu().MilliValue() data, err := deps.KubeClientset.CoreV1().RESTClient().Get().AbsPath("apis/metrics.k8s.io/v1beta1/namespaces/" + namespace + "/pods/" + podName).DoRaw(context.Background()) if err != nil { return nil, err } var podMetrics PodMetrics err = json.Unmarshal(data, &podMetrics) if err != nil { return nil, err } var cpuUsedNano int64 for _, c := range podMetrics.Containers { containerCPU, err := parseCPUNano(c.Usage.CPU) if err != nil { return nil, err } cpuUsedNano += containerCPU } cpuUsedMilliCores := cpuUsedNano / 1000000 return &CPUMetrics{ MilliCores: uint64(cpuMilliCores), UsedMilliCores: uint64(cpuUsedMilliCores), Timestamp: podMetrics.Timestamp, }, nil }