|
| 1 | +// Copyright 2024 PingCAP, Inc. |
| 2 | +// |
| 3 | +// Licensed under the Apache License, Version 2.0 (the "License"); |
| 4 | +// you may not use this file except in compliance with the License. |
| 5 | +// You may obtain a copy of the License at |
| 6 | +// |
| 7 | +// http://www.apache.org/licenses/LICENSE-2.0 |
| 8 | +// |
| 9 | +// Unless required by applicable law or agreed to in writing, software |
| 10 | +// distributed under the License is distributed on an "AS IS" BASIS, |
| 11 | +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 12 | +// See the License for the specific language governing permissions and |
| 13 | +// limitations under the License. |
| 14 | + |
| 15 | +package tasks |
| 16 | + |
| 17 | +import ( |
| 18 | + "context" |
| 19 | + "crypto/tls" |
| 20 | + "strings" |
| 21 | + "time" |
| 22 | + |
| 23 | + "github.com/go-logr/logr" |
| 24 | + |
| 25 | + "github.com/pingcap/tidb-operator/api/v2/core/v1alpha1" |
| 26 | + "github.com/pingcap/tidb-operator/v2/pkg/apicall" |
| 27 | + coreutil "github.com/pingcap/tidb-operator/v2/pkg/apiutil/core/v1alpha1" |
| 28 | + "github.com/pingcap/tidb-operator/v2/pkg/client" |
| 29 | + "github.com/pingcap/tidb-operator/v2/pkg/resourcemanagerapi" |
| 30 | + "github.com/pingcap/tidb-operator/v2/pkg/runtime/scope" |
| 31 | + pdm "github.com/pingcap/tidb-operator/v2/pkg/timanager/pd" |
| 32 | + |
| 33 | + ctrlclient "sigs.k8s.io/controller-runtime/pkg/client" |
| 34 | +) |
| 35 | + |
| 36 | +const ( |
| 37 | + resourceManagerServiceName = "resource_manager" |
| 38 | + |
| 39 | + defaultAPITimeout = 5 * time.Second |
| 40 | +) |
| 41 | + |
| 42 | +var ( |
| 43 | + newRMClient = resourcemanagerapi.NewClient |
| 44 | +) |
| 45 | + |
| 46 | +func transferPrimaryIfNeeded( |
| 47 | + ctx context.Context, |
| 48 | + logger logr.Logger, |
| 49 | + c client.Client, |
| 50 | + cluster *v1alpha1.Cluster, |
| 51 | + rm *v1alpha1.ResourceManager, |
| 52 | + pdClient pdm.PDClient, |
| 53 | +) (bool, error) { |
| 54 | + groupName := rm.Labels[v1alpha1.LabelKeyGroup] |
| 55 | + if groupName == "" { |
| 56 | + // Not managed by ResourceManagerGroup, skip. |
| 57 | + return false, nil |
| 58 | + } |
| 59 | + |
| 60 | + if cluster == nil || strings.TrimSpace(cluster.Status.PD) == "" { |
| 61 | + return false, nil |
| 62 | + } |
| 63 | + if pdClient == nil || pdClient.Underlay() == nil { |
| 64 | + return false, nil |
| 65 | + } |
| 66 | + |
| 67 | + peers, err := listGroupResourceManagers(ctx, c, cluster, rm, groupName) |
| 68 | + if err != nil { |
| 69 | + return false, err |
| 70 | + } |
| 71 | + if len(peers) <= 1 { |
| 72 | + return false, nil |
| 73 | + } |
| 74 | + |
| 75 | + var tlsConfig *tls.Config |
| 76 | + if coreutil.IsTLSClusterEnabled(cluster) { |
| 77 | + cfg, err := apicall.GetClientTLSConfig(ctx, c, cluster) |
| 78 | + if err != nil { |
| 79 | + return false, err |
| 80 | + } |
| 81 | + tlsConfig = cfg |
| 82 | + } |
| 83 | + |
| 84 | + primaryAddr, err := pdClient.Underlay().GetMicroServicePrimary(ctx, resourceManagerServiceName) |
| 85 | + if err != nil { |
| 86 | + return false, err |
| 87 | + } |
| 88 | + if primaryAddr == "" { |
| 89 | + return false, nil |
| 90 | + } |
| 91 | + |
| 92 | + primaryAddr = strings.TrimRight(strings.TrimSpace(primaryAddr), "/") |
| 93 | + myAddr := coreutil.InstanceAdvertiseURL[scope.ResourceManager](cluster, rm, coreutil.ResourceManagerClientPort(rm)) |
| 94 | + if myAddr == "" || normalizeMicroserviceAddr(primaryAddr) != normalizeMicroserviceAddr(myAddr) { |
| 95 | + return false, nil |
| 96 | + } |
| 97 | + |
| 98 | + transferee := coreutil.LongestReadyPeer[scope.ResourceManager](rm, peers) |
| 99 | + if transferee == nil { |
| 100 | + logger.Info("no healthy transferee available for resource manager primary transfer", "name", rm.Name) |
| 101 | + return false, nil |
| 102 | + } |
| 103 | + |
| 104 | + logger.Info("try to transfer resource manager primary", "from", rm.Name, "to", transferee.Name) |
| 105 | + |
| 106 | + rmClient := newRMClient(ensureAddrScheme(primaryAddr, coreutil.IsTLSClusterEnabled(cluster)), defaultAPITimeout, tlsConfig) |
| 107 | + if err := rmClient.TransferPrimary(ctx, transferee.Name); err != nil { |
| 108 | + return false, err |
| 109 | + } |
| 110 | + |
| 111 | + return true, nil |
| 112 | +} |
| 113 | + |
| 114 | +func listGroupResourceManagers( |
| 115 | + ctx context.Context, |
| 116 | + c client.Client, |
| 117 | + cluster *v1alpha1.Cluster, |
| 118 | + rm *v1alpha1.ResourceManager, |
| 119 | + groupName string, |
| 120 | +) ([]*v1alpha1.ResourceManager, error) { |
| 121 | + var list v1alpha1.ResourceManagerList |
| 122 | + selector := map[string]string{ |
| 123 | + v1alpha1.LabelKeyComponent: v1alpha1.LabelValComponentResourceManager, |
| 124 | + v1alpha1.LabelKeyCluster: cluster.Name, |
| 125 | + v1alpha1.LabelKeyGroup: groupName, |
| 126 | + } |
| 127 | + if err := c.List(ctx, &list, ctrlclient.InNamespace(rm.Namespace), ctrlclient.MatchingLabels(selector)); err != nil { |
| 128 | + return nil, err |
| 129 | + } |
| 130 | + peers := make([]*v1alpha1.ResourceManager, 0, len(list.Items)) |
| 131 | + for i := range list.Items { |
| 132 | + peer := &list.Items[i] |
| 133 | + peers = append(peers, peer) |
| 134 | + } |
| 135 | + return peers, nil |
| 136 | +} |
| 137 | + |
| 138 | +func normalizeMicroserviceAddr(addr string) string { |
| 139 | + addr = strings.TrimRight(strings.TrimSpace(addr), "/") |
| 140 | + addr = strings.TrimPrefix(addr, "http://") |
| 141 | + addr = strings.TrimPrefix(addr, "https://") |
| 142 | + return addr |
| 143 | +} |
| 144 | + |
| 145 | +func ensureAddrScheme(addr string, tlsEnabled bool) string { |
| 146 | + addr = strings.TrimRight(strings.TrimSpace(addr), "/") |
| 147 | + if strings.HasPrefix(addr, "http://") || strings.HasPrefix(addr, "https://") { |
| 148 | + return addr |
| 149 | + } |
| 150 | + scheme := "http" |
| 151 | + if tlsEnabled { |
| 152 | + scheme = "https" |
| 153 | + } |
| 154 | + return scheme + "://" + addr |
| 155 | +} |
0 commit comments