Fix imported pump & drainer start script

pkg/cluster/api/binlog.go

@@ -84,12 +84,20 @@ type NodeStatus struct {
 }
 
 // IsPumpTombstone check if drainer is tombstone.
-func (c *BinlogClient) IsPumpTombstone(nodeID string) (bool, error) {
+func (c *BinlogClient) IsPumpTombstone(addr string) (bool, error) {
+	nodeID, err := c.nodeID(addr, "pumps")
+	if err != nil {
+		return false, err
+	}
 	return c.isTombstone("pumps", nodeID)
 }
 
 // IsDrainerTombstone check if drainer is tombstone.
-func (c *BinlogClient) IsDrainerTombstone(nodeID string) (bool, error) {
+func (c *BinlogClient) IsDrainerTombstone(addr string) (bool, error) {
+	nodeID, err := c.nodeID(addr, "drainers")
+	if err != nil {
+		return false, err
+	}
 	return c.isTombstone("drainer", nodeID)
 }
 
@@ -121,13 +129,36 @@ func (c *BinlogClient) drainerNodeStatus() (status []*NodeStatus, err error) {
 	return c.nodeStatus("drainers")
 }
 
+func (c *BinlogClient) nodeID(addr, ty string) (string, error) {
+	nodes, err := c.nodeStatus(ty)
+	if err != nil {
+		return "", err
+	}
+
+	for _, node := range nodes {
+		if addr == node.Addr {
+			return node.NodeID, nil
+		}
+	}
+
+	return "", errors.Errorf("pump node id for address %s not found", addr)
+}
+
 // UpdateDrainerState update the specify state as the specified state.
-func (c *BinlogClient) UpdateDrainerState(nodeID string, state string) error {
+func (c *BinlogClient) UpdateDrainerState(addr string, state string) error {
+	nodeID, err := c.nodeID(addr, "drainers")
+	if err != nil {
+		return err
+	}
 	return c.updateStatus("drainers", nodeID, state)
 }
 
 // UpdatePumpState update the specify state as the specified state.
-func (c *BinlogClient) UpdatePumpState(nodeID string, state string) error {
+func (c *BinlogClient) UpdatePumpState(addr string, state string) error {
+	nodeID, err := c.nodeID(addr, "pumps")
+	if err != nil {
+		return err
+	}
 	return c.updateStatus("pumps", nodeID, state)
 }
 
@@ -228,11 +259,19 @@ func (c *BinlogClient) offline(addr string, nodeID string) error {
 }
 
 // OfflinePump offline a pump.
-func (c *BinlogClient) OfflinePump(addr string, nodeID string) error {
+func (c *BinlogClient) OfflinePump(addr string) error {
+	nodeID, err := c.nodeID(addr, "pumps")
+	if err != nil {
+		return err
+	}
 	return c.offline(addr, nodeID)
 }
 
 // OfflineDrainer offline a drainer.
-func (c *BinlogClient) OfflineDrainer(addr string, nodeID string) error {
+func (c *BinlogClient) OfflineDrainer(addr string) error {
+	nodeID, err := c.nodeID(addr, "drainers")
+	if err != nil {
+		return err
+	}
 	return c.offline(addr, nodeID)
 }

pkg/cluster/manager.go

@@ -611,7 +611,7 @@ func (m *Manager) Display(clusterName string, opt operator.Options) error {
 		// Check if there is some instance in tombstone state
 		nodes, _ := operator.DestroyTombstone(ctx, t, true /* returnNodesOnly */, opt, tlsCfg)
 		if len(nodes) != 0 {
-			color.Green("There are some nodes in state: `Tombstone`\n\tNodes: %+v\n\tYou can destroy them with the command: `tiup cluster prune %s`", nodes, clusterName)
+			color.Green("There are some nodes can be pruned: `Tombstone`\n\tNodes: %+v\n\tYou can destroy them with the command: `tiup cluster prune %s`", nodes, clusterName)
 		}
 	}
 

pkg/cluster/operation/scale_in.go

@@ -315,13 +315,13 @@ func deleteMember(
 		}
 	case spec.ComponentDrainer:
 		addr := instance.GetHost() + ":" + strconv.Itoa(instance.GetPort())
-		err := binlogClient.OfflineDrainer(addr, addr)
+		err := binlogClient.OfflineDrainer(addr)
 		if err != nil {
-			return errors.AddStack(err)
+			return err
 		}
 	case spec.ComponentPump:
 		addr := instance.GetHost() + ":" + strconv.Itoa(instance.GetPort())
-		err := binlogClient.OfflinePump(addr, addr)
+		err := binlogClient.OfflinePump(addr)
 		if err != nil {
 			return errors.AddStack(err)
 		}

pkg/cluster/spec/drainer.go

@@ -146,8 +146,13 @@ func (i *DrainerInstance) InitConfig(
 
 	enableTLS := i.topo.GlobalOptions.TLSEnabled
 	spec := i.InstanceSpec.(DrainerSpec)
+	nodeID := i.GetHost() + ":" + strconv.Itoa(i.GetPort())
+	// keep origin node id if is imported
+	if i.IsImported() {
+		nodeID = ""
+	}
 	cfg := scripts.NewDrainerScript(
-		i.GetHost()+":"+strconv.Itoa(i.GetPort()),
+		nodeID,
 		i.GetHost(),
 		paths.Deploy,
 		paths.Data[0],

pkg/cluster/spec/pump.go

@@ -145,8 +145,13 @@ func (i *PumpInstance) InitConfig(
 
 	enableTLS := i.topo.GlobalOptions.TLSEnabled
 	spec := i.InstanceSpec.(PumpSpec)
+	nodeID := i.GetHost() + ":" + strconv.Itoa(i.GetPort())
+	// keep origin node id if is imported
+	if i.IsImported() {
+		nodeID = ""
+	}
 	cfg := scripts.NewPumpScript(
-		i.GetHost()+":"+strconv.Itoa(i.GetPort()),
+		nodeID,
 		i.GetHost(),
 		paths.Deploy,
 		paths.Data[0],

templates/scripts/run_drainer.sh.tpl

@@ -22,7 +22,9 @@ exec numactl --cpunodebind={{.NumaNode}} --membind={{.NumaNode}} bin/drainer \
 {{- else}}
 exec bin/drainer \
 {{- end}}
+{{- if .NodeID}}
     --node-id="{{.NodeID}}" \
+{{- end}}
     --addr="{{.IP}}:{{.Port}}" \
     --pd-urls="{{template "PDList" .Endpoints}}" \
     --data-dir="{{.DataDir}}" \

templates/scripts/run_pump.sh.tpl

@@ -22,7 +22,9 @@ exec numactl --cpunodebind={{.NumaNode}} --membind={{.NumaNode}} bin/pump \
 {{- else}}
 exec bin/pump \
 {{- end}}
+{{- if .NodeID}}
     --node-id="{{.NodeID}}" \
+{{- end}}
     --addr="0.0.0.0:{{.Port}}" \
     --advertise-addr="{{.Host}}:{{.Port}}" \
     --pd-urls="{{template "PDList" .Endpoints}}" \