cluster: integrate tispark

Makefile

@@ -1,4 +1,5 @@
-.PHONY: server
+.PHONY: cmd components server
+.DEFAULT_GOAL := default
 
 GOVER := $(shell go version)
 
@@ -24,11 +25,10 @@ FILES     := $$(find . -name "*.go")
 FAILPOINT_ENABLE  := $$(tools/bin/failpoint-ctl enable)
 FAILPOINT_DISABLE := $$(tools/bin/failpoint-ctl disable)
 
-default: build check
+default: check build
 
 include ./tests/Makefile
 
-
 # Build TiUP and all components
 build: tiup components
 
@@ -120,8 +120,6 @@ fmt:
 	@echo "goimports (if installed)"
 	$(shell gimports -w $(FILES) 2>/dev/null)
 
-.PHONY: cmd
-
 tools/bin/errcheck: tools/check/go.mod
 	cd tools/check; \
 	$(GO) build -o ../bin/errcheck github.com/kisielk/errcheck

components/cluster/command/check.go

@@ -167,6 +167,7 @@ func checkSystemInfo(s *cliutil.SSHConnectionProps, topo *spec.Specification, op
 					inst.OS(),
 					inst.Arch(),
 					insightVer,
+					"", // use default srcPath
 					inst.GetHost(),
 					task.CheckToolsPathDir,
 				).

components/cluster/command/deploy.go

@@ -146,6 +146,11 @@ func confirmTopology(clusterName, version string, topo *spec.Specification, patc
 		log.Errorf("    3. The component marked as `patched` has been replaced by previours patch command.")
 	}
 
+	if len(topo.TiSparkMasters) > 0 || len(topo.TiSparkWorkers) > 0 {
+		log.Warnf("There are TiSpark nodes defined in the topology, please note that you'll need to manually install Java Runtime Environment (JRE) 8 on the host, other wise the TiSpark nodes will fail to start.")
+		log.Warnf("You may read the OpenJDK doc for a reference: https://openjdk.java.net/install/")
+	}
+
 	return cliutil.PromptForConfirmOrAbortError("Do you want to continue? [y/N]: ")
 }
 
@@ -270,37 +275,50 @@ func deploy(clusterName, clusterVersion, topoFile string, opt deployOptions) err
 		// log dir will always be with values, but might not used by the component
 		logDir := clusterutil.Abs(globalOptions.User, inst.LogDir())
 		// Deploy component
+		// prepare deployment server
 		t := task.NewBuilder().
 			UserSSH(inst.GetHost(), inst.GetSSHPort(), globalOptions.User, gOpt.SSHTimeout).
 			Mkdir(globalOptions.User, inst.GetHost(),
 				deployDir, logDir,
 				filepath.Join(deployDir, "bin"),
 				filepath.Join(deployDir, "conf"),
 				filepath.Join(deployDir, "scripts")).
-			Mkdir(globalOptions.User, inst.GetHost(), dataDirs...).
-			CopyComponent(
+			Mkdir(globalOptions.User, inst.GetHost(), dataDirs...)
+
+		// copy dependency component if needed
+		switch inst.ComponentName() {
+		case spec.ComponentTiSpark:
+			t = t.DeploySpark(inst, version, "" /* default srcPath */, deployDir)
+		default:
+			t = t.CopyComponent(
 				inst.ComponentName(),
 				inst.OS(),
 				inst.Arch(),
 				version,
+				"", // use default srcPath
 				inst.GetHost(),
 				deployDir,
-			).
-			InitConfig(
-				clusterName,
-				clusterVersion,
-				inst,
-				globalOptions.User,
-				opt.ignoreConfigCheck,
-				meta.DirPaths{
-					Deploy: deployDir,
-					Data:   dataDirs,
-					Log:    logDir,
-					Cache:  spec.ClusterPath(clusterName, spec.TempConfigPath),
-				},
-			).
-			BuildAsStep(fmt.Sprintf("  - Copy %s -> %s", inst.ComponentName(), inst.GetHost()))
-		deployCompTasks = append(deployCompTasks, t)
+			)
+		}
+
+		// generate configs for the component
+		t = t.InitConfig(
+			clusterName,
+			clusterVersion,
+			inst,
+			globalOptions.User,
+			opt.ignoreConfigCheck,
+			meta.DirPaths{
+				Deploy: deployDir,
+				Data:   dataDirs,
+				Log:    logDir,
+				Cache:  spec.ClusterPath(clusterName, spec.TempConfigPath),
+			},
+		)
+
+		deployCompTasks = append(deployCompTasks,
+			t.BuildAsStep(fmt.Sprintf("  - Copy %s -> %s", inst.ComponentName(), inst.GetHost())),
+		)
 	})
 
 	nodeInfoTask := task.NewBuilder().Func("Check status", func(ctx *task.Context) error {
@@ -373,6 +391,11 @@ func buildMonitoredDeployTask(
 		version := spec.ComponentVersion(comp, version)
 
 		for host, info := range uniqueHosts {
+			// FIXME: as the uniqueHosts list is built with os-arch as part of the key,
+			// for platform independent packages, it will be downloaded multiple times
+			// and be saved with different file names in the packages dir, the tarballs
+			// are identical and the only difference is platform in filename.
+
 			// populate unique os/arch set
 			key := fmt.Sprintf("%s-%s-%s", comp, info.os, info.arch)
 			if _, found := uniqueCompOSArch[key]; !found {
@@ -404,6 +427,7 @@ func buildMonitoredDeployTask(
 					info.os,
 					info.arch,
 					version,
+					"", // use default srcPath
 					host,
 					deployDir,
 				).

components/cluster/command/destroy.go

@@ -69,7 +69,8 @@ You can retain some nodes and roles data when destroy cluster, eg:
 
 			logger.EnableAuditLog()
 			metadata, err := spec.ClusterMetadata(clusterName)
-			if err != nil && !errors.Is(perrs.Cause(err), meta.ErrValidate) {
+			if err != nil && !errors.Is(perrs.Cause(err), meta.ErrValidate) &&
+				!errors.Is(perrs.Cause(err), spec.ErrNoTiSparkMaster) {
 				return err
 			}
 

components/cluster/command/display.go

@@ -72,7 +72,8 @@ func newDisplayCmd() *cobra.Command {
 			}
 
 			metadata, err := spec.ClusterMetadata(clusterName)
-			if err != nil && !errors.Is(perrs.Cause(err), meta.ErrValidate) {
+			if err != nil && !errors.Is(perrs.Cause(err), meta.ErrValidate) &&
+				!errors.Is(perrs.Cause(err), spec.ErrNoTiSparkMaster) {
 				return perrs.AddStack(err)
 			}
 			return destroyTombstoneIfNeed(clusterName, metadata, gOpt)
@@ -88,7 +89,8 @@ func newDisplayCmd() *cobra.Command {
 
 func displayDashboardInfo(clusterName string) error {
 	metadata, err := spec.ClusterMetadata(clusterName)
-	if err != nil && !errors.Is(perrs.Cause(err), meta.ErrValidate) {
+	if err != nil && !errors.Is(perrs.Cause(err), meta.ErrValidate) &&
+		!errors.Is(perrs.Cause(err), spec.ErrNoTiSparkMaster) {
 		return err
 	}
 
@@ -121,7 +123,8 @@ func displayDashboardInfo(clusterName string) error {
 
 func displayClusterMeta(clusterName string, opt *operator.Options) error {
 	clsMeta, err := spec.ClusterMetadata(clusterName)
-	if err != nil && !errors.Is(perrs.Cause(err), meta.ErrValidate) {
+	if err != nil && !errors.Is(perrs.Cause(err), meta.ErrValidate) &&
+		!errors.Is(perrs.Cause(err), spec.ErrNoTiSparkMaster) {
 		return err
 	}
 
@@ -175,7 +178,8 @@ func destroyTombstoneIfNeed(clusterName string, metadata *spec.ClusterMeta, opt
 
 func displayClusterTopology(clusterName string, opt *operator.Options) error {
 	metadata, err := spec.ClusterMetadata(clusterName)
-	if err != nil && !errors.Is(perrs.Cause(err), meta.ErrValidate) {
+	if err != nil && !errors.Is(perrs.Cause(err), meta.ErrValidate) &&
+		!errors.Is(perrs.Cause(err), spec.ErrNoTiSparkMaster) {
 		return err
 	}
 

components/cluster/command/reload.go

@@ -112,7 +112,15 @@ func buildReloadTask(
 			case spec.ComponentGrafana, spec.ComponentPrometheus, spec.ComponentAlertManager:
 				version := spec.ComponentVersion(compName, metadata.Version)
 				tb.Download(compName, inst.OS(), inst.Arch(), version).
-					CopyComponent(compName, inst.OS(), inst.Arch(), version, inst.GetHost(), deployDir)
+					CopyComponent(
+						compName,
+						inst.OS(),
+						inst.Arch(),
+						version,
+						"", // use default srcPath
+						inst.GetHost(),
+						deployDir,
+					)
 			}
 			hasImported = true
 		}

components/cluster/command/scale_in.go

@@ -115,7 +115,15 @@ func scaleIn(clusterName string, options operator.Options) error {
 				case spec.ComponentGrafana, spec.ComponentPrometheus, spec.ComponentAlertManager:
 					version := spec.ComponentVersion(compName, metadata.Version)
 					tb.Download(compName, instance.OS(), instance.Arch(), version).
-						CopyComponent(compName, instance.OS(), instance.Arch(), version, instance.GetHost(), deployDir)
+						CopyComponent(
+							compName,
+							instance.OS(),
+							instance.Arch(),
+							version,
+							"", // use default srcPath
+							instance.GetHost(),
+							deployDir,
+						)
 				}
 				hasImported = true
 			}

components/cluster/command/scale_out.go

@@ -15,12 +15,13 @@ package command
 
 import (
 	"context"
+	"errors"
 	"io/ioutil"
 	"path/filepath"
 	"strings"
 
 	"github.com/joomcode/errorx"
-	"github.com/pingcap/errors"
+	perrs "github.com/pingcap/errors"
 	"github.com/pingcap/tiup/pkg/cliutil"
 	"github.com/pingcap/tiup/pkg/cliutil/prepare"
 	"github.com/pingcap/tiup/pkg/cluster/clusterutil"
@@ -72,11 +73,11 @@ func newScaleOutCmd() *cobra.Command {
 func scaleOut(clusterName, topoFile string, opt scaleOutOptions) error {
 	exist, err := tidbSpec.Exist(clusterName)
 	if err != nil {
-		return errors.AddStack(err)
+		return perrs.AddStack(err)
 	}
 
 	if !exist {
-		return errors.Errorf("cannot scale-out non-exists cluster %s", clusterName)
+		return perrs.Errorf("cannot scale-out non-exists cluster %s", clusterName)
 	}
 
 	metadata, err := spec.ClusterMetadata(clusterName)
@@ -91,7 +92,12 @@ func scaleOut(clusterName, topoFile string, opt scaleOutOptions) error {
 		MonitoredOptions: metadata.Topology.MonitoredOptions,
 		ServerConfigs:    metadata.Topology.ServerConfigs,
 	}
-	if err := clusterutil.ParseTopologyYaml(topoFile, &newPart); err != nil {
+
+	// The no tispark master error is ignored, as if the tispark master is removed from the topology
+	// file for some reason (manual edit, for example), it is still possible to scale-out it to make
+	// the whole topology back to normal state.
+	if err := clusterutil.ParseTopologyYaml(topoFile, &newPart); err != nil &&
+		!errors.Is(perrs.Cause(err), spec.ErrNoTiSparkMaster) {
 		return err
 	}
 
@@ -146,7 +152,7 @@ func scaleOut(clusterName, topoFile string, opt scaleOutOptions) error {
 			// FIXME: Map possible task errors and give suggestions.
 			return err
 		}
-		return errors.Trace(err)
+		return perrs.Trace(err)
 	}
 
 	log.Infof("Scaled cluster `%s` out successfully", clusterName)
@@ -247,11 +253,28 @@ func buildScaleOutTask(
 				filepath.Join(deployDir, "conf"),
 				filepath.Join(deployDir, "scripts")).
 			Mkdir(metadata.User, inst.GetHost(), dataDirs...)
+
+		srcPath := ""
 		if patchedComponents.Exist(inst.ComponentName()) {
-			tb.InstallPackage(spec.ClusterPath(clusterName, spec.PatchDirName, inst.ComponentName()+".tar.gz"), inst.GetHost(), deployDir)
-		} else {
-			tb.CopyComponent(inst.ComponentName(), inst.OS(), inst.Arch(), version, inst.GetHost(), deployDir)
+			srcPath = spec.ClusterPath(clusterName, spec.PatchDirName, inst.ComponentName()+".tar.gz")
 		}
+
+		// copy dependency component if needed
+		switch inst.ComponentName() {
+		case spec.ComponentTiSpark:
+			tb = tb.DeploySpark(inst, version, srcPath, deployDir)
+		default:
+			tb.CopyComponent(
+				inst.ComponentName(),
+				inst.OS(),
+				inst.Arch(),
+				version,
+				srcPath,
+				inst.GetHost(),
+				deployDir,
+			)
+		}
+
 		t := tb.ScaleConfig(clusterName,
 			metadata.Version,
 			metadata.Topology,
@@ -282,7 +305,15 @@ func buildScaleOutTask(
 			case spec.ComponentGrafana, spec.ComponentPrometheus, spec.ComponentAlertManager:
 				version := spec.ComponentVersion(compName, metadata.Version)
 				tb.Download(compName, inst.OS(), inst.Arch(), version).
-					CopyComponent(compName, inst.OS(), inst.Arch(), version, inst.GetHost(), deployDir)
+					CopyComponent(
+						compName,
+						inst.OS(),
+						inst.Arch(),
+						version,
+						"", // use default srcPath
+						inst.GetHost(),
+						deployDir,
+					)
 			}
 			hasImported = true
 		}
@@ -367,7 +398,7 @@ func validateNewTopo(topo *spec.Specification) (err error) {
 	topo.IterInstance(func(instance spec.Instance) {
 		// check for "imported" parameter, it can not be true when scaling out
 		if instance.IsImported() {
-			err = errors.New(
+			err = perrs.New(
 				"'imported' is set to 'true' for new instance, this is only used " +
 					"for instances imported from tidb-ansible and make no sense when " +
 					"scaling out, please delete the line or set it to 'false' for new instances")

components/cluster/command/upgrade.go

@@ -131,16 +131,38 @@ func upgrade(clusterName, clusterVersion string, opt operator.Options) error {
 			if inst.IsImported() {
 				switch inst.ComponentName() {
 				case spec.ComponentPrometheus, spec.ComponentGrafana, spec.ComponentAlertManager:
-					tb.CopyComponent(inst.ComponentName(), inst.OS(), inst.Arch(), version, inst.GetHost(), deployDir)
-				default:
-					tb.BackupComponent(inst.ComponentName(), metadata.Version, inst.GetHost(), deployDir).
-						CopyComponent(inst.ComponentName(), inst.OS(), inst.Arch(), version, inst.GetHost(), deployDir)
+					tb.CopyComponent(
+						inst.ComponentName(),
+						inst.OS(),
+						inst.Arch(),
+						version,
+						"", // use default srcPath
+						inst.GetHost(),
+						deployDir,
+					)
 				}
 				hasImported = true
-			} else {
-				tb.BackupComponent(inst.ComponentName(), metadata.Version, inst.GetHost(), deployDir).
-					CopyComponent(inst.ComponentName(), inst.OS(), inst.Arch(), version, inst.GetHost(), deployDir)
 			}
+
+			// backup files of the old version
+			tb = tb.BackupComponent(inst.ComponentName(), metadata.Version, inst.GetHost(), deployDir)
+
+			// copy dependency component if needed
+			switch inst.ComponentName() {
+			case spec.ComponentTiSpark:
+				tb = tb.DeploySpark(inst, version, "" /* default srcPath */, deployDir)
+			default:
+				tb = tb.CopyComponent(
+					inst.ComponentName(),
+					inst.OS(),
+					inst.Arch(),
+					version,
+					"", // use default srcPath
+					inst.GetHost(),
+					deployDir,
+				)
+			}
+
 			tb.InitConfig(
 				clusterName,
 				clusterVersion,

components/dm/command/deploy.go

@@ -244,6 +244,7 @@ func deploy(clusterName, clusterVersion, topoFile string, opt deployOptions) err
 				inst.OS(),
 				inst.Arch(),
 				version,
+				"", // use default srcPath
 				inst.GetHost(),
 				deployDir,
 			).