Support tidb upgrade api (#2287)

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
+}
+

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()
 

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)
 		})
 	}
 

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()
 

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 {

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