From 4f5d9e30688c4e6d625f03ab5042cc4d2d93c593 Mon Sep 17 00:00:00 2001 From: nexustar Date: Sun, 8 Oct 2023 19:15:22 +0800 Subject: [PATCH] Support tidb upgrade api (#2287) --- pkg/cluster/api/tidbapi.go | 78 ++++++++++++++++++++++++++++++++ pkg/cluster/manager/patch.go | 3 +- pkg/cluster/manager/reload.go | 2 +- pkg/cluster/manager/upgrade.go | 2 +- pkg/cluster/operation/upgrade.go | 32 ++++++++++++- pkg/tidbver/tidbver.go | 5 ++ 6 files changed, 118 insertions(+), 4 deletions(-) create mode 100644 pkg/cluster/api/tidbapi.go diff --git a/pkg/cluster/api/tidbapi.go b/pkg/cluster/api/tidbapi.go new file mode 100644 index 0000000000..ddb3a36403 --- /dev/null +++ b/pkg/cluster/api/tidbapi.go @@ -0,0 +1,78 @@ +// Copyright 2022 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 api + +import ( + "context" + "crypto/tls" + "fmt" + "time" + + "github.com/pingcap/tiup/pkg/utils" +) + +// TiDBClient is client for access TiKVCDC Open API +type TiDBClient struct { + urls []string + client *utils.HTTPClient + ctx context.Context +} + +// NewTiDBClient return a `TiDBClient` +func NewTiDBClient(ctx context.Context, addresses []string, timeout time.Duration, tlsConfig *tls.Config) *TiDBClient { + httpPrefix := "http" + if tlsConfig != nil { + httpPrefix = "https" + } + urls := make([]string, 0, len(addresses)) + for _, addr := range addresses { + urls = append(urls, fmt.Sprintf("%s://%s", httpPrefix, addr)) + } + + return &TiDBClient{ + urls: urls, + client: utils.NewHTTPClient(timeout, tlsConfig), + ctx: ctx, + } +} + +func (c *TiDBClient) getEndpoints(api string) (endpoints []string) { + for _, url := range c.urls { + endpoints = append(endpoints, fmt.Sprintf("%s%s", url, api)) + } + return endpoints +} + +// StartUpgrade sends the start upgrade message to the TiDB server +func (c *TiDBClient) StartUpgrade() error { + api := "/upgrade/start" + endpoints := c.getEndpoints(api) + _, err := tryURLs(endpoints, func(endpoint string) ([]byte, error) { + return c.client.Post(c.ctx, endpoint, nil) + }) + + return err +} + +// FinishUpgrade sends the finish upgrade message to the TiDB server +func (c *TiDBClient) FinishUpgrade() error { + api := "/upgrade/finish" + endpoints := c.getEndpoints(api) + _, err := tryURLs(endpoints, func(endpoint string) ([]byte, error) { + return c.client.Post(c.ctx, endpoint, nil) + }) + + return err +} + diff --git a/pkg/cluster/manager/patch.go b/pkg/cluster/manager/patch.go index 3babc36b82..087ded0ccb 100644 --- a/pkg/cluster/manager/patch.go +++ b/pkg/cluster/manager/patch.go @@ -104,7 +104,8 @@ func (m *Manager) Patch(name string, packagePath string, opt operator.Options, o if offline { return nil } - return operator.Upgrade(ctx, topo, opt, tlsCfg, base.Version) + // TBD: should patch be treated as an upgrade? + return operator.Upgrade(ctx, topo, opt, tlsCfg, base.Version, false) }). Build() diff --git a/pkg/cluster/manager/reload.go b/pkg/cluster/manager/reload.go index 2a78e3fe41..21ec4f6b7a 100644 --- a/pkg/cluster/manager/reload.go +++ b/pkg/cluster/manager/reload.go @@ -124,7 +124,7 @@ func (m *Manager) Reload(name string, gOpt operator.Options, skipRestart, skipCo return err } b.Func("Upgrade Cluster", func(ctx context.Context) error { - return operator.Upgrade(ctx, topo, gOpt, tlsCfg, base.Version) + return operator.Upgrade(ctx, topo, gOpt, tlsCfg, base.Version, false) }) } diff --git a/pkg/cluster/manager/upgrade.go b/pkg/cluster/manager/upgrade.go index 06a668b999..fe76ff5ac5 100644 --- a/pkg/cluster/manager/upgrade.go +++ b/pkg/cluster/manager/upgrade.go @@ -237,7 +237,7 @@ Do you want to continue? [y/N]:`, if offline { return nil } - return operator.Upgrade(ctx, topo, opt, tlsCfg, base.Version) + return operator.Upgrade(ctx, topo, opt, tlsCfg, base.Version, true) }). Build() diff --git a/pkg/cluster/operation/upgrade.go b/pkg/cluster/operation/upgrade.go index a34731a9fb..80ee1679b2 100644 --- a/pkg/cluster/operation/upgrade.go +++ b/pkg/cluster/operation/upgrade.go @@ -39,13 +39,14 @@ var ( increaseLimitPoint = checkpoint.Register() ) -// Upgrade the cluster. +// Upgrade the cluster. (actually, it's rolling restart) func Upgrade( ctx context.Context, topo spec.Topology, options Options, tlsCfg *tls.Config, currentVersion string, + isUpgrade bool, ) error { roleFilter := set.NewStringSet(options.Roles...) nodeFilter := set.NewStringSet(options.Nodes...) @@ -70,6 +71,7 @@ func Upgrade( var origRegionScheduleLimit int var err error + var tidbClient *api.TiDBClient var pdEndpoints []string forcePDEndpoints := os.Getenv(EnvNamePDEndpointOverwrite) // custom set PD endpoint list @@ -100,6 +102,21 @@ func Upgrade( } }() } + case spec.ComponentTiDB: + dbs := topo.(*spec.Specification).TiDBServers + endpoints := []string{} + for _, db := range dbs { + endpoints = append(endpoints, utils.JoinHostPort(db.GetManageHost(), db.StatusPort)) + } + + if isUpgrade && tidbver.TiDBSupportUpgradeAPI(currentVersion) { + tidbClient = api.NewTiDBClient(ctx, endpoints, 10*time.Second, tlsCfg) + err = tidbClient.StartUpgrade() + if err != nil { + return err + } + } + default: // do nothing, kept for future usage with other components } @@ -178,6 +195,19 @@ func Upgrade( return err } } + + switch component.Name() { + case spec.ComponentTiDB: + if isUpgrade && tidbver.TiDBSupportUpgradeAPI(currentVersion) { + err = tidbClient.FinishUpgrade() + if err != nil { + return err + } + } + + default: + // do nothing, kept for future usage with other components + } } if topo.GetMonitoredOptions() == nil { diff --git a/pkg/tidbver/tidbver.go b/pkg/tidbver/tidbver.go index bfa169899e..18cc53e74f 100644 --- a/pkg/tidbver/tidbver.go +++ b/pkg/tidbver/tidbver.go @@ -26,6 +26,11 @@ func TiDBSupportSecureBoot(version string) bool { return semver.Compare(version, "v5.3.0") >= 0 || strings.Contains(version, "nightly") } +// TiDBSupportUpgradeAPI return if given version of TiDB support upgrade API +func TiDBSupportUpgradeAPI(version string) bool { + return semver.Compare(version, "v7.4.0") >= 0 || strings.Contains(version, "nightly") +} + // TiKVSupportAdvertiseStatusAddr return if given version of TiKV support --advertise-status-addr func TiKVSupportAdvertiseStatusAddr(version string) bool { // TiKV support --advertise-status-addr since v4.0.1