fix selfmon related known issues

cluster/calcium/node.go

@@ -67,8 +67,10 @@ func (c *Calcium) SetNode(ctx context.Context, opts *types.SetNodeOptions) (*typ
 		litter.Dump(opts)
 		opts.Normalize(node)
 		n = node
+
 		n.Available = (opts.StatusOpt == types.TriTrue) || (opts.StatusOpt == types.TriKeep && n.Available)
-		if !n.Available {
+		n.Bypass = (opts.BypassOpt == types.TriTrue) || (opts.BypassOpt == types.TriKeep && n.Bypass)
+		if n.IsDown() {
 			logger.Errorf(ctx, "[SetNodeAvailable] node marked down: %s", opts.Nodename)
 		}
 		if opts.WorkloadsDown {

rpc/gen/core.proto

@@ -155,6 +155,7 @@ message Node {
     map<string, int64> volume = 19;
     int64 volume_used = 20;
     map<string, int64> init_numa_memory = 21;
+    bool bypass = 22;
 }
 
 message Nodes {
@@ -178,6 +179,7 @@ message SetNodeOptions {
     map<string, int64> delta_volume = 9;
     bool workloads_down = 10;
     string endpoint = 11;
+    TriOpt bypass_opt = 12;
 }
 
 message SetNodeStatusOptions {

rpc/rpc.go

@@ -180,9 +180,21 @@ func (v *Vibranium) ListPodNodes(ctx context.Context, opts *pb.ListNodesOptions)
 	ctx, cancel := context.WithTimeout(ctx, v.config.GlobalTimeout)
 	defer cancel()
 	nodes := []*pb.Node{}
+	nodeChan := make(chan *pb.Node, len(ns))
+	wg := &sync.WaitGroup{}
 	for _, n := range ns {
-		nodes = append(nodes, toRPCNode(ctx, n))
+		wg.Add(1)
+		go func(node *types.Node) {
+			defer wg.Done()
+			nodeChan <- toRPCNode(ctx, node)
+		}(n)
 	}
+	wg.Wait()
+	close(nodeChan)
+	for node := range nodeChan {
+		nodes = append(nodes, node)
+	}
+
 	return &pb.Nodes{Nodes: nodes}, nil
 }
 

rpc/transform.go

@@ -79,6 +79,7 @@ func toRPCNode(ctx context.Context, n *types.Node) *pb.Node {
 		Numa:           n.NUMA,
 		NumaMemory:     n.NUMAMemory,
 		InitNumaMemory: n.InitNUMAMemory,
+		Bypass:         n.Bypass,
 	}
 }
 
@@ -153,6 +154,7 @@ func toCoreSetNodeOptions(b *pb.SetNodeOptions) (*types.SetNodeOptions, error) {
 		DeltaVolume:     b.DeltaVolume,
 		NUMA:            b.Numa,
 		Labels:          b.Labels,
+		BypassOpt:       types.TriOptions(b.BypassOpt),
 	}
 	for cpuID, cpuShare := range b.DeltaCpu {
 		r.DeltaCPU[cpuID] = int64(cpuShare)

store/etcdv3/node.go

@@ -241,6 +241,7 @@ func (m *Mercury) doAddNode(ctx context.Context, name, endpoint, podname, ca, ce
 			NUMAMemory:     numaMemory,
 		},
 		Available: true,
+		Bypass:    false,
 	}
 
 	bytes, err := json.Marshal(node)
@@ -291,7 +292,7 @@ func (m *Mercury) doGetNodes(ctx context.Context, kvs []*mvccpb.KeyValue, labels
 			return nil, err
 		}
 		node.Init()
-		if (node.Available || all) && utils.FilterWorkload(node.Labels, labels) {
+		if (!node.IsDown() || all) && utils.FilterWorkload(node.Labels, labels) {
 			engine, err := m.makeClient(ctx, node, false)
 			if err != nil {
 				return nil, err

store/redis/node.go

@@ -280,7 +280,7 @@ func (r *Rediaron) doGetNodes(ctx context.Context, kvs map[string]string, labels
 			return nil, err
 		}
 		node.Init()
-		if (node.Available || all) && utils.FilterWorkload(node.Labels, labels) {
+		if (!node.IsDown() || all) && utils.FilterWorkload(node.Labels, labels) {
 			engine, err := r.makeClient(ctx, node, false)
 			if err != nil {
 				return nil, err

types/node.go

@@ -59,6 +59,8 @@ type Node struct {
 	CPUUsed    float64 `json:"cpuused"`
 	VolumeUsed int64   `json:"volumeused"`
 
+	// Bypass if bypass is true, it will not participate in future scheduling
+	Bypass    bool       `json:"bypass,omitempty"`
 	Available bool       `json:"available"`
 	Engine    engine.API `json:"-"`
 }
@@ -208,6 +210,13 @@ func (n *Node) PreserveResources(resource *ResourceMeta) {
 	}
 }
 
+// IsDown returns if the node is marked as down.
+func (n *Node) IsDown() bool {
+	// If `bypass` is true, then even if the node is still healthy, the node will be regarded as `down`.
+	// Currently `bypass` will only be set when the cli calls the `up` and `down` commands.
+	return n.Bypass || !n.Available
+}
+
 // NodeMetrics used for metrics collecting
 type NodeMetrics struct {
 	Name        string

types/options.go

@@ -182,6 +182,7 @@ type SetNodeOptions struct {
 	DeltaVolume     VolumeMap
 	NUMA            map[string]string
 	Labels          map[string]string
+	BypassOpt       TriOptions
 }
 
 // Validate checks options