implement running status check for tiflash

pkg/cluster/operation/action.go

@@ -471,8 +471,13 @@ func StartComponent(ctx context.Context, instances []spec.Instance, noAgentHosts
 	// eg: PD has more strict restrictions on the capacity expansion process,
 	// that is, there should be only one node in the peer-join stage at most
 	// ref https://github.com/tikv/pd/blob/d38b36714ccee70480c39e07126e3456b5fb292d/server/join/join.go#L179-L191
-	if options.Operation == ScaleOutOperation && (name == spec.ComponentPD || name == spec.ComponentDMMaster) {
-		return serialStartInstances(ctx, instances, options, tlsCfg)
+	if options.Operation == ScaleOutOperation {
+		switch name {
+		case spec.ComponentPD,
+			spec.ComponentTiFlash,
+			spec.ComponentDMMaster:
+			return serialStartInstances(ctx, instances, options, tlsCfg)
+		}
 	}
 
 	errg, _ := errgroup.WithContext(ctx)

pkg/cluster/spec/tiflash.go

@@ -19,12 +19,15 @@ import (
 	"crypto/tls"
 	"encoding/json"
 	"fmt"
+	"io"
+	"net/http"
 	"os"
 	"path/filepath"
 	"sort"
 	"strings"
 	"time"
 
+	"github.com/pingcap/errors"
 	perrs "github.com/pingcap/errors"
 	"github.com/pingcap/tiup/pkg/cluster/api"
 	"github.com/pingcap/tiup/pkg/cluster/ctxt"
@@ -250,6 +253,11 @@ func (i *TiFlashInstance) GetServicePort() int {
 	return i.InstanceSpec.(*TiFlashSpec).FlashServicePort
 }
 
+// GetStatusPort returns the status port of TiFlash
+func (i *TiFlashInstance) GetStatusPort() int {
+	return i.InstanceSpec.(*TiFlashSpec).FlashProxyStatusPort
+}
+
 // checkIncorrectKey checks TiFlash's key should not be set in config
 func (i *TiFlashInstance) checkIncorrectKey(key string) error {
 	errMsg := "NOTE: TiFlash `%s` should NOT be set in topo's \"%s\" config, its value will be ignored, you should set `data_dir` in each host instead, please check your topology"
@@ -730,3 +738,54 @@ func (i *TiFlashInstance) PrepareStart(ctx context.Context, tlsCfg *tls.Config)
 	pdClient := api.NewPDClient(endpoints, 10*time.Second, tlsCfg)
 	return pdClient.UpdateReplicateConfig(bytes.NewBuffer(enablePlacementRules))
 }
+
+// Ready implements Instance interface
+func (i *TiFlashInstance) Ready(ctx context.Context, e ctxt.Executor, timeout uint64) error {
+	if err := PortStarted(ctx, e, i.Port, timeout); err != nil {
+		return err
+	}
+
+	scheme := "http"
+	if i.topo.BaseTopo().GlobalOptions.TLSEnabled {
+		scheme = "https"
+		// TODO: implement tls config for tiflash instances
+		// (we don't support tiflash instance in tls enabled cluster yet)
+	}
+	addr := fmt.Sprintf("%s://%s:%d/tiflash/store-status", scheme, i.Host, i.GetStatusPort())
+	req, err := http.NewRequest("GET", addr, nil)
+	if err != nil {
+		return err
+	}
+	req = req.WithContext(ctx)
+
+	retryOpt := utils.RetryOption{
+		Delay:   time.Second,
+		Timeout: time.Second * time.Duration(timeout),
+	}
+	var queryErr error
+	if err := utils.Retry(func() error {
+		client := utils.NewHTTPClient(statusQueryTimeout, nil)
+		res, err := client.Client().Do(req)
+		if err != nil {
+			queryErr = err
+			return err
+		}
+		defer res.Body.Close()
+		body, err := io.ReadAll(res.Body)
+		if err != nil {
+			queryErr = err
+			return err
+		}
+		if res.StatusCode == http.StatusNotFound || string(body) == "Running" {
+			return nil
+		}
+
+		err = fmt.Errorf("tiflash store status '%s' not ready", string(body))
+		queryErr = err
+		return err
+	}, retryOpt); err != nil {
+		return errors.Annotatef(queryErr, "timed out waiting for tiflash %s:%d to be ready after %ds",
+			i.Host, i.Port, timeout)
+	}
+	return nil
+}