Skip to content

Commit 916a11b

Browse files
DuodenumLDuodenumL
committed
move selfmon
1 parent 6e331fc commit 916a11b

File tree

12 files changed

+80
-71
lines changed

12 files changed

+80
-71
lines changed

‎cluster/calcium/calcium.go

+3-20
Original file line numberDiff line numberDiff line change
@@ -15,8 +15,6 @@ import (
1515
"github.com/projecteru2/core/source/github"
1616
"github.com/projecteru2/core/source/gitlab"
1717
"github.com/projecteru2/core/store"
18-
"github.com/projecteru2/core/store/etcdv3"
19-
"github.com/projecteru2/core/store/redis"
2018
"github.com/projecteru2/core/types"
2119

2220
"github.com/pkg/errors"
@@ -31,27 +29,16 @@ type Calcium struct {
3129
watcher discovery.Service
3230
wal *WAL
3331
identifier string
34-
selfmon *NodeStatusWatcher
3532
}
3633

3734
// New returns a new cluster config
3835
func New(config types.Config, t *testing.T) (*Calcium, error) {
3936
logger := log.WithField("Calcium", "New").WithField("config", config)
4037

4138
// set store
42-
var store store.Store
43-
var err error
44-
switch config.Store {
45-
case types.Redis:
46-
store, err = redis.New(config, t)
47-
if err != nil {
48-
return nil, logger.Err(context.TODO(), errors.WithStack(err))
49-
}
50-
default:
51-
store, err = etcdv3.New(config, t)
52-
if err != nil {
53-
return nil, logger.Err(context.TODO(), errors.WithStack(err))
54-
}
39+
store, err := store.NewStore(config, t)
40+
if err != nil {
41+
return nil, logger.Err(context.TODO(), errors.WithStack(err))
5542
}
5643

5744
// set scheduler
@@ -83,10 +70,6 @@ func New(config types.Config, t *testing.T) (*Calcium, error) {
8370
cal := &Calcium{store: store, config: config, scheduler: potassium, source: scm, watcher: watcher}
8471
cal.wal, err = newCalciumWAL(cal)
8572
cal.identifier = config.Identifier()
86-
cal.selfmon = NewNodeStatusWatcher(cal)
87-
88-
// start node status watcher
89-
go cal.selfmon.run()
9073

9174
return cal, logger.Err(nil, errors.WithStack(err)) //nolint
9275
}

‎cluster/calcium/dissociate.go

+2-3
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,6 @@ import (
44
"context"
55

66
"github.com/projecteru2/core/log"
7-
"github.com/projecteru2/core/store"
87
"github.com/projecteru2/core/types"
98
"github.com/projecteru2/core/utils"
109

@@ -34,7 +33,7 @@ func (c *Calcium) DissociateWorkload(ctx context.Context, ids []string) (chan *t
3433
ctx,
3534
// if
3635
func(ctx context.Context) (err error) {
37-
if err = c.store.UpdateNodeResource(ctx, node, &workload.ResourceMeta, store.ActionIncr); err == nil {
36+
if err = c.store.UpdateNodeResource(ctx, node, &workload.ResourceMeta, types.ActionIncr); err == nil {
3837
log.Infof(ctx, "[DissociateWorkload] Workload %s dissociated", workload.ID)
3938
}
4039
return errors.WithStack(err)
@@ -48,7 +47,7 @@ func (c *Calcium) DissociateWorkload(ctx context.Context, ids []string) (chan *t
4847
if failedByCond {
4948
return nil
5049
}
51-
return errors.WithStack(c.store.UpdateNodeResource(ctx, node, &workload.ResourceMeta, store.ActionDecr))
50+
return errors.WithStack(c.store.UpdateNodeResource(ctx, node, &workload.ResourceMeta, types.ActionDecr))
5251
},
5352
c.config.GlobalTimeout,
5453
)

‎cluster/calcium/remove.go

+2-3
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,6 @@ import (
66
"sync"
77

88
"github.com/projecteru2/core/log"
9-
"github.com/projecteru2/core/store"
109
"github.com/projecteru2/core/types"
1110
"github.com/projecteru2/core/utils"
1211

@@ -42,7 +41,7 @@ func (c *Calcium) RemoveWorkload(ctx context.Context, ids []string, force bool,
4241
ctx,
4342
// if
4443
func(ctx context.Context) error {
45-
return errors.WithStack(c.store.UpdateNodeResource(ctx, node, &workload.ResourceMeta, store.ActionIncr))
44+
return errors.WithStack(c.store.UpdateNodeResource(ctx, node, &workload.ResourceMeta, types.ActionIncr))
4645
},
4746
// then
4847
func(ctx context.Context) (err error) {
@@ -56,7 +55,7 @@ func (c *Calcium) RemoveWorkload(ctx context.Context, ids []string, force bool,
5655
if failedByCond {
5756
return nil
5857
}
59-
return errors.WithStack(c.store.UpdateNodeResource(ctx, node, &workload.ResourceMeta, store.ActionDecr))
58+
return errors.WithStack(c.store.UpdateNodeResource(ctx, node, &workload.ResourceMeta, types.ActionDecr))
6059
},
6160
c.config.GlobalTimeout,
6261
)

‎core.go

+4
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ import (
1818
"github.com/projecteru2/core/metrics"
1919
"github.com/projecteru2/core/rpc"
2020
pb "github.com/projecteru2/core/rpc/gen"
21+
"github.com/projecteru2/core/selfmon"
2122
"github.com/projecteru2/core/utils"
2223
"github.com/projecteru2/core/version"
2324

@@ -133,6 +134,9 @@ func serve(c *cli.Context) error {
133134
ctx, cancel := signal.NotifyContext(c.Context, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
134135
defer cancel()
135136

137+
// start node status checker
138+
go selfmon.RunNodeStatusWatcher(ctx, config, cluster, t)
139+
136140
// start engine cache checker
137141
go factory.EngineCacheChecker(ctx, config.ConnectionTimeout)
138142

‎engine/factory/factory.go

+1-1
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,7 @@ var (
3333
)
3434

3535
func getEngineCacheKey(endpoint, ca, cert, key string) string {
36-
return endpoint + "-" + utils.SHA256(fmt.Sprintf(":%v:%v:%v", ca, cert, key))[:8]
36+
return fmt.Sprintf("%v-%v", endpoint, utils.SHA256(fmt.Sprintf(":%v:%v:%v", ca, cert, key))[:8])
3737
}
3838

3939
// EngineCacheChecker checks if the engine in cache is available

‎cluster/calcium/selfmon.go ‎selfmon/selfmon.go

+38-25
Original file line numberDiff line numberDiff line change
@@ -1,16 +1,19 @@
1-
package calcium
1+
package selfmon
22

33
import (
44
"context"
55
"math/rand"
66
"os/signal"
77
"syscall"
8+
"testing"
89
"time"
910

1011
"github.com/pkg/errors"
1112

13+
"github.com/projecteru2/core/cluster"
1214
"github.com/projecteru2/core/log"
13-
coretypes "github.com/projecteru2/core/types"
15+
"github.com/projecteru2/core/store"
16+
"github.com/projecteru2/core/types"
1417
"github.com/projecteru2/core/utils"
1518
)
1619

@@ -19,22 +22,32 @@ const ActiveKey = "/selfmon/active"
1922

2023
// NodeStatusWatcher monitors the changes of node status
2124
type NodeStatusWatcher struct {
22-
id int64
23-
cal *Calcium
25+
id int64
26+
config types.Config
27+
cluster cluster.Cluster
28+
store store.Store
2429
}
2530

26-
// NewNodeStatusWatcher .
27-
func NewNodeStatusWatcher(cal *Calcium) *NodeStatusWatcher {
31+
// RunNodeStatusWatcher .
32+
func RunNodeStatusWatcher(ctx context.Context, config types.Config, cluster cluster.Cluster, t *testing.T) {
2833
rand.Seed(time.Now().UnixNano())
2934
id := rand.Int63n(10000) // nolint
30-
return &NodeStatusWatcher{
31-
id: id,
32-
cal: cal,
35+
store, err := store.NewStore(config, t)
36+
if err != nil {
37+
log.Errorf(context.TODO(), "[RunNodeStatusWatcher] %v failed to create store, err: %v", id, err)
38+
return
39+
}
40+
41+
watcher := &NodeStatusWatcher{
42+
id: id,
43+
config: config,
44+
store: store,
45+
cluster: cluster,
3346
}
47+
watcher.run(ctx)
3448
}
3549

36-
func (n *NodeStatusWatcher) run() {
37-
ctx := n.getSignalContext(context.TODO())
50+
func (n *NodeStatusWatcher) run(ctx context.Context) {
3851
for {
3952
select {
4053
case <-ctx.Done():
@@ -45,7 +58,7 @@ func (n *NodeStatusWatcher) run() {
4558
log.Errorf(ctx, "[NodeStatusWatcher] %v stops watching, err: %v", n.id, err)
4659
}
4760
})
48-
time.Sleep(n.cal.config.ConnectionTimeout)
61+
time.Sleep(n.config.ConnectionTimeout)
4962
}
5063
}
5164
}
@@ -90,7 +103,7 @@ func (n *NodeStatusWatcher) withActiveLock(parentCtx context.Context, f func(ctx
90103
if errors.Is(err, context.Canceled) {
91104
log.Info("[Register] context canceled")
92105
return
93-
} else if !errors.Is(err, coretypes.ErrKeyExists) {
106+
} else if !errors.Is(err, types.ErrKeyExists) {
94107
log.Errorf(ctx, "[Register] failed to re-register: %v", err)
95108
time.Sleep(time.Second)
96109
continue
@@ -126,20 +139,20 @@ func (n *NodeStatusWatcher) withActiveLock(parentCtx context.Context, f func(ctx
126139
}
127140

128141
func (n *NodeStatusWatcher) register(ctx context.Context) (<-chan struct{}, func(), error) {
129-
return n.cal.store.StartEphemeral(ctx, ActiveKey, n.cal.config.HAKeepaliveInterval)
142+
return n.store.StartEphemeral(ctx, ActiveKey, n.config.HAKeepaliveInterval)
130143
}
131144

132145
func (n *NodeStatusWatcher) initNodeStatus(ctx context.Context) {
133146
log.Debug(ctx, "[NodeStatusWatcher] init node status started")
134-
nodes := make(chan *coretypes.Node)
147+
nodes := make(chan *types.Node)
135148

136149
go func() {
137150
defer close(nodes)
138151
// Get all nodes which are active status, and regardless of pod.
139152
var err error
140-
var ch <-chan *coretypes.Node
141-
utils.WithTimeout(ctx, n.cal.config.GlobalTimeout, func(ctx context.Context) {
142-
ch, err = n.cal.ListPodNodes(ctx, &coretypes.ListNodesOptions{
153+
var ch <-chan *types.Node
154+
utils.WithTimeout(ctx, n.config.GlobalTimeout, func(ctx context.Context) {
155+
ch, err = n.cluster.ListPodNodes(ctx, &types.ListNodesOptions{
143156
Podname: "",
144157
Labels: nil,
145158
All: true,
@@ -161,9 +174,9 @@ func (n *NodeStatusWatcher) initNodeStatus(ctx context.Context) {
161174
}()
162175

163176
for node := range nodes {
164-
status, err := n.cal.GetNodeStatus(ctx, node.Name)
177+
status, err := n.cluster.GetNodeStatus(ctx, node.Name)
165178
if err != nil {
166-
status = &coretypes.NodeStatus{
179+
status = &types.NodeStatus{
167180
Nodename: node.Name,
168181
Podname: node.Podname,
169182
Alive: false,
@@ -178,15 +191,15 @@ func (n *NodeStatusWatcher) monitor(ctx context.Context) error {
178191
go n.initNodeStatus(ctx)
179192

180193
// monitor node status
181-
messageChan := n.cal.NodeStatusStream(ctx)
194+
messageChan := n.cluster.NodeStatusStream(ctx)
182195
log.Infof(ctx, "[NodeStatusWatcher] %v watch node status started", n.id)
183196
defer log.Infof(ctx, "[NodeStatusWatcher] %v stop watching node status", n.id)
184197

185198
for {
186199
select {
187200
case message, ok := <-messageChan:
188201
if !ok {
189-
return coretypes.ErrMessageChanClosed
202+
return types.ErrMessageChanClosed
190203
}
191204
go n.dealNodeStatusMessage(ctx, message)
192205
case <-ctx.Done():
@@ -195,18 +208,18 @@ func (n *NodeStatusWatcher) monitor(ctx context.Context) error {
195208
}
196209
}
197210

198-
func (n *NodeStatusWatcher) dealNodeStatusMessage(ctx context.Context, message *coretypes.NodeStatus) {
211+
func (n *NodeStatusWatcher) dealNodeStatusMessage(ctx context.Context, message *types.NodeStatus) {
199212
if message.Error != nil {
200213
log.Errorf(ctx, "[NodeStatusWatcher] deal with node status stream message failed %+v", message)
201214
return
202215
}
203216

204217
// TODO maybe we need a distributed lock to control concurrency
205-
opts := &coretypes.SetNodeOptions{
218+
opts := &types.SetNodeOptions{
206219
Nodename: message.Nodename,
207220
WorkloadsDown: !message.Alive,
208221
}
209-
if _, err := n.cal.SetNode(ctx, opts); err != nil {
222+
if _, err := n.cluster.SetNode(ctx, opts); err != nil {
210223
log.Errorf(ctx, "[NodeStatusWatcher] set node %s failed %v", message.Nodename, err)
211224
return
212225
}

‎store/etcdv3/node.go

+2-3
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@ import (
1717
"github.com/projecteru2/core/engine/fake"
1818
"github.com/projecteru2/core/log"
1919
"github.com/projecteru2/core/metrics"
20-
"github.com/projecteru2/core/store"
2120
"github.com/projecteru2/core/types"
2221
"github.com/projecteru2/core/utils"
2322
)
@@ -178,9 +177,9 @@ func (m *Mercury) UpdateNodes(ctx context.Context, nodes ...*types.Node) error {
178177
// UpdateNodeResource update cpu and memory on a node, either add or subtract
179178
func (m *Mercury) UpdateNodeResource(ctx context.Context, node *types.Node, resource *types.ResourceMeta, action string) error {
180179
switch action {
181-
case store.ActionIncr:
180+
case types.ActionIncr:
182181
node.RecycleResources(resource)
183-
case store.ActionDecr:
182+
case types.ActionDecr:
184183
node.PreserveResources(resource)
185184
default:
186185
return types.ErrUnknownControlType

‎store/etcdv3/node_test.go

+2-3
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,6 @@ import (
77
"testing"
88
"time"
99

10-
"github.com/projecteru2/core/store"
1110
"github.com/projecteru2/core/types"
1211

1312
"github.com/stretchr/testify/assert"
@@ -223,8 +222,8 @@ func TestUpdateNodeResource(t *testing.T) {
223222
assert.NoError(t, err)
224223
assert.Equal(t, node.Name, "test")
225224
assert.Error(t, m.UpdateNodeResource(ctx, node, nil, "wtf"))
226-
assert.NoError(t, m.UpdateNodeResource(ctx, node, &types.ResourceMeta{CPU: map[string]int64{"0": 100}}, store.ActionIncr))
227-
assert.NoError(t, m.UpdateNodeResource(ctx, node, &types.ResourceMeta{CPU: map[string]int64{"0": 100}}, store.ActionDecr))
225+
assert.NoError(t, m.UpdateNodeResource(ctx, node, &types.ResourceMeta{CPU: map[string]int64{"0": 100}}, types.ActionIncr))
226+
assert.NoError(t, m.UpdateNodeResource(ctx, node, &types.ResourceMeta{CPU: map[string]int64{"0": 100}}, types.ActionDecr))
228227
}
229228

230229
func TestExtractNodename(t *testing.T) {

‎store/redis/node.go

+2-3
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,6 @@ import (
1414
"github.com/projecteru2/core/engine/fake"
1515
"github.com/projecteru2/core/log"
1616
"github.com/projecteru2/core/metrics"
17-
"github.com/projecteru2/core/store"
1817
"github.com/projecteru2/core/types"
1918
"github.com/projecteru2/core/utils"
2019

@@ -169,9 +168,9 @@ func (r *Rediaron) UpdateNodes(ctx context.Context, nodes ...*types.Node) error
169168
// UpdateNodeResource update cpu and memory on a node, either add or subtract
170169
func (r *Rediaron) UpdateNodeResource(ctx context.Context, node *types.Node, resource *types.ResourceMeta, action string) error {
171170
switch action {
172-
case store.ActionIncr:
171+
case types.ActionIncr:
173172
node.RecycleResources(resource)
174-
case store.ActionDecr:
173+
case types.ActionDecr:
175174
node.PreserveResources(resource)
176175
default:
177176
return types.ErrUnknownControlType

‎store/redis/node_test.go

+2-3
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,6 @@ import (
66
"path/filepath"
77
"time"
88

9-
"github.com/projecteru2/core/store"
109
"github.com/projecteru2/core/types"
1110
)
1211

@@ -213,8 +212,8 @@ func (s *RediaronTestSuite) TestUpdateNodeResource() {
213212
s.NoError(err)
214213
s.Equal(node.Name, "test")
215214
s.Error(s.rediaron.UpdateNodeResource(ctx, node, nil, "wtf"))
216-
s.NoError(s.rediaron.UpdateNodeResource(ctx, node, &types.ResourceMeta{CPU: map[string]int64{"0": 100}}, store.ActionIncr))
217-
s.NoError(s.rediaron.UpdateNodeResource(ctx, node, &types.ResourceMeta{CPU: map[string]int64{"0": 100}}, store.ActionDecr))
215+
s.NoError(s.rediaron.UpdateNodeResource(ctx, node, &types.ResourceMeta{CPU: map[string]int64{"0": 100}}, types.ActionIncr))
216+
s.NoError(s.rediaron.UpdateNodeResource(ctx, node, &types.ResourceMeta{CPU: map[string]int64{"0": 100}}, types.ActionDecr))
218217
}
219218

220219
func (s *RediaronTestSuite) TestExtractNodename() {

0 commit comments

Comments
 (0)