func (c *Cluster) AddExecutor(execInfo *core.ExecutorInfo) *core.Result {
	if execInfo.Worker.Cluster != c.ClusterId {
		return core.FailedResult(-1001, "not match cluster")
	}
	c.executorInfosLocker.Lock()
	defer c.executorInfosLocker.Unlock()
	c.ExecutorInfos[execInfo.TaskConfig.TaskID] = execInfo
	return core.SuccessResult()
}
AddExecutor add executor info into executor list
package cluster
import (
	"errors"
	"fmt"
	"github.com/devfeel/mapper"
	"github.com/devfeel/rockman/config"
	"github.com/devfeel/rockman/core"
	"github.com/devfeel/rockman/logger"
	"github.com/devfeel/rockman/registry"
	"github.com/devfeel/rockman/rpc/client"
	"github.com/devfeel/rockman/scheduler"
	_json "github.com/devfeel/rockman/util/json"
	"github.com/hashicorp/consul/api"
	"strconv"
	"sync"
	"time"
)
const (
	MinQueryResourceInterval = 60
)
type (
	Cluster struct {
		ClusterId              string
		Registry               *registry.Registry
		LeaderKey              string
		LeaderServer           string
		leaderLastIndex        uint64
		lastGetLeaderTime      time.Time
		Nodes                  map[string]*core.NodeInfo
		nodesLastIndex         uint64
		nodesLocker            *sync.RWMutex
		lastLoadNodesTime      time.Time
		ExecutorInfos          map[string]*core.ExecutorInfo
		executorInfosLastIndex uint64
		executorInfosLocker    *sync.RWMutex
		lastLoadExecutorsTime  time.Time
		rpcClients             map[string]*client.RpcClient
		rpcClientLocker        *sync.RWMutex
		Scheduler              scheduler.Scheduler
		config                 *config.Profile
		isSTW                  bool  
		OnNodesChange          WatchChangeHandle
		OnNodeOffline          NodeOfflineHandle
		OnLeaderChange         WatchChangeHandle
		OnLeaderChangeFailed   WatchFailedHandle
	}
	WatchChangeHandle func()
	WatchFailedHandle func()
	NodeOfflineHandle func(info *core.NodeInfo)
)
func NewCluster(profile *config.Profile, registry *registry.Registry) *Cluster {
	cluster := new(Cluster)
	cluster.config = profile
	cluster.ClusterId = profile.Cluster.ClusterId
	cluster.LeaderKey = getLeaderKey(profile.Cluster.ClusterId)
	cluster.Nodes = make(map[string]*core.NodeInfo)
	cluster.nodesLocker = new(sync.RWMutex)
	cluster.ExecutorInfos = make(map[string]*core.ExecutorInfo)
	cluster.executorInfosLocker = new(sync.RWMutex)
	cluster.rpcClients = make(map[string]*client.RpcClient)
	cluster.rpcClientLocker = new(sync.RWMutex)
	cluster.Registry = registry
	cluster.Scheduler = scheduler.NewScheduler()
	logger.Node().Debug("Cluster {" + cluster.ClusterId + "} init success.")
	return cluster
}
func (c *Cluster) Start() error {
	logger.Default().Debug("Cluster start...")
	c.watchLeader()
	return nil
}
func (c *Cluster) Stop() error {
	logger.Default().Debug("Cluster Stop.")
	c.isSTW = true
	return nil
}
func (c *Cluster) ElectionLeader(leaderServer string) error {
	lt := "Cluster.ElectionLeader "
	opts := &api.LockOptions{
		Key:   c.LeaderKey,
		Value: []byte(leaderServer),
		SessionOpts: &api.SessionEntry{
			Name:     leaderServer,
			TTL:      "10s",
			Behavior: "delete",
		},
	}
	locker, err := c.Registry.CreateLockerOpts(opts)
	if err != nil {
		return err
	}
	_, err = locker.Locker.Lock(nil)
	if err != nil {
		return err
	}
	logger.Cluster().Debug(lt + "success.")
	c.loadOnlineNodes()
	c.watchOnlineNodes()
	c.queryExecutorsFromWorkers()
	c.cycleQueryWorkerResource()
	return nil
}
func (c *Cluster) GetLeaderInfo() (string, error) {
	if c.LeaderServer != "" {
		return c.LeaderServer, nil
	}
	kvPair, meta, err := c.Registry.Get(c.LeaderKey, nil)
	if err != nil {
		return "", err
	} else {
		if kvPair.Session == "" {
			return "", errors.New("lock session is nil")
		} else {
			c.LeaderServer = string(kvPair.Value)
			c.leaderLastIndex = meta.LastIndex
			c.lastGetLeaderTime = time.Now()
			return c.LeaderServer, nil
		}
	}
}
func (c *Cluster) AddNodeInfo(nodeInfo *core.NodeInfo) *core.Result {
	if nodeInfo.Cluster != c.ClusterId {
		return core.FailedResult(-1001, "not match cluster")
	}
	key := nodeInfo.EndPoint()
	resource, result := c.QueryNodeResource(key)
	if result.Error != nil {
		resource = nodeInfo.GetEmptyResource()
		logger.Cluster().Warn("AddNodeInfo.QueryResource[" + key + "] error: " + result.Error.Error())
		logger.Cluster().Error(result.Error, "AddNodeInfo.QueryResource["+key+"] error")
	} else {
		if !result.IsSuccess() {
			resource = nodeInfo.GetEmptyResource()
			logger.Cluster().Warn("AddNodeInfo.QueryResource[" + key + "] failed: " + result.Message())
		}
	}
	c.nodesLocker.Lock()
	defer c.nodesLocker.Unlock()
	c.Scheduler.SetResource(resource)
	c.Nodes[key] = nodeInfo
	logger.Cluster().DebugF("Cluster add node [%v] [%v] [Master: %v] [Wroker: %v]", nodeInfo.NodeID, nodeInfo.EndPoint(), nodeInfo.IsMaster, nodeInfo.IsWorker)
	return core.SuccessResult()
}