/
WorkerMgr.go
76 lines (64 loc) · 1.5 KB
/
WorkerMgr.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
package master
import (
"context"
"github.com/Goular/crontab/common"
"go.etcd.io/etcd/clientv3"
"go.etcd.io/etcd/mvcc/mvccpb"
"time"
)
// /cron/workers/
type WorkerMgr struct {
client *clientv3.Client
kv clientv3.KV
lease clientv3.Lease
}
var (
G_workerMgr *WorkerMgr
)
// 获取在线worker列表
func (workerMgr *WorkerMgr) ListWorkers() (workerArr []string, err error) {
var (
getResp *clientv3.GetResponse
kv *mvccpb.KeyValue
workerIP string
)
// 初始化数组
workerArr = make([]string, 0)
// 获取目录下所有Kv
if getResp, err = workerMgr.kv.Get(context.TODO(), common.JOB_WORKER_DIR, clientv3.WithPrefix()); err != nil {
return
}
// 解析每个节点的IP
for _, kv = range getResp.Kvs {
// kv.Key : /cron/workers/192.168.2.1
workerIP = common.ExtractWorkerIP(string(kv.Key))
workerArr = append(workerArr, workerIP)
}
return
}
func InitWorkerMgr() (err error) {
var (
config clientv3.Config
client *clientv3.Client
kv clientv3.KV
lease clientv3.Lease
)
// 初始化配置
config = clientv3.Config{
Endpoints: G_config.EtcdEndpoints, // 集群地址
DialTimeout: time.Duration(G_config.EtcdDialTimeout) * time.Millisecond, // 连接超时
}
// 建立连接
if client, err = clientv3.New(config); err != nil {
return
}
// 得到KV和Lease的API子集
kv = clientv3.NewKV(client)
lease = clientv3.NewLease(client)
G_workerMgr = &WorkerMgr{
client: client,
kv: kv,
lease: lease,
}
return
}