Skip to content

Commit

Permalink
Browse files Browse the repository at this point in the history
lock node without podname
  • Loading branch information
CMGS committed Dec 10, 2019
1 parent 647ecb1 commit 9c6eaf4
Show file tree
Hide file tree
Showing 13 changed files with 434 additions and 452 deletions.
2 changes: 1 addition & 1 deletion VERSION
@@ -1 +1 @@
19.11.06
19.11.07
2 changes: 1 addition & 1 deletion cluster/calcium/create.go
Expand Up @@ -67,7 +67,7 @@ func (c *Calcium) doCreateContainer(ctx context.Context, opts *types.DeployOptio
for i, m := range messages {
ch <- m
if m.Error != nil && m.ContainerID == "" {
if err := c.withNodeLocked(ctx, opts.Podname, nodeInfo.Name, func(node *types.Node) error {
if err := c.withNodeLocked(ctx, nodeInfo.Name, func(node *types.Node) error {
return c.store.UpdateNodeResource(ctx, node, m.CPU, opts.CPUQuota, opts.Memory, opts.Storage, store.ActionIncr)
}); err != nil {
log.Errorf("[doCreateContainer] Reset node %s failed %v", nodeInfo.Name, err)
Expand Down
2 changes: 1 addition & 1 deletion cluster/calcium/dissociate.go
Expand Up @@ -15,7 +15,7 @@ func (c *Calcium) DissociateContainer(ctx context.Context, IDs []string) (chan *
defer close(ch)
for _, ID := range IDs {
err := c.withContainerLocked(ctx, ID, func(container *types.Container) error {
return c.withNodeLocked(ctx, container.Podname, container.Nodename, func(node *types.Node) (err error) {
return c.withNodeLocked(ctx, container.Nodename, func(node *types.Node) (err error) {
if err := c.store.RemoveContainer(ctx, container); err != nil {
return err
}
Expand Down
4 changes: 2 additions & 2 deletions cluster/calcium/lock.go
Expand Up @@ -43,8 +43,8 @@ func (c *Calcium) withContainerLocked(ctx context.Context, ID string, f func(con
})
}

func (c *Calcium) withNodeLocked(ctx context.Context, podname, nodename string, f func(node *types.Node) error) error {
return c.withNodesLocked(ctx, podname, nodename, nil, true, func(nodes map[string]*types.Node) error {
func (c *Calcium) withNodeLocked(ctx context.Context, nodename string, f func(node *types.Node) error) error {
return c.withNodesLocked(ctx, "", nodename, nil, true, func(nodes map[string]*types.Node) error {
return f(nodes[nodename])
})
}
Expand Down
4 changes: 2 additions & 2 deletions cluster/calcium/lock_test.go
Expand Up @@ -189,11 +189,11 @@ func TestWithNodeLocked(t *testing.T) {
lock.On("Lock", mock.Anything).Return(nil)
// failed by get locked node
store.On("GetNode", mock.Anything, mock.Anything).Return(nil, types.ErrNoETCD).Once()
err := c.withNodeLocked(ctx, "test", "test", func(node *types.Node) error { return nil })
err := c.withNodeLocked(ctx, "test", func(node *types.Node) error { return nil })
assert.Error(t, err)
store.On("GetNode", mock.Anything, mock.Anything).Return(node1, nil)
// success
err = c.withNodeLocked(ctx, "test", "test", func(node *types.Node) error {
err = c.withNodeLocked(ctx, "test", func(node *types.Node) error {
assert.Equal(t, node.Name, node1.Name)
return nil
})
Expand Down
10 changes: 5 additions & 5 deletions cluster/calcium/node.go
Expand Up @@ -18,7 +18,7 @@ func (c *Calcium) AddNode(ctx context.Context, nodename, endpoint, podname, ca,

// RemoveNode remove a node
func (c *Calcium) RemoveNode(ctx context.Context, podname, nodename string) error {
return c.withNodeLocked(ctx, podname, nodename, func(node *types.Node) error {
return c.withNodeLocked(ctx, nodename, func(node *types.Node) error {
return c.store.RemoveNode(ctx, node)
})
}
Expand All @@ -37,20 +37,20 @@ func (c *Calcium) GetNode(ctx context.Context, nodename string) (*types.Node, er
func (c *Calcium) GetNodes(ctx context.Context, podname, nodename string, labels map[string]string, all bool) ([]*types.Node, error) {
var ns []*types.Node
var err error
if nodename == "" {
ns, err = c.ListPodNodes(ctx, podname, labels, all)
} else {
if nodename != "" {
var node *types.Node
node, err = c.GetNode(ctx, nodename)
ns = []*types.Node{node}
} else {
ns, err = c.ListPodNodes(ctx, podname, labels, all)
}
return ns, err
}

// SetNode set node available or not
func (c *Calcium) SetNode(ctx context.Context, opts *types.SetNodeOptions) (*types.Node, error) {
var n *types.Node
return n, c.withNodeLocked(ctx, opts.Podname, opts.Nodename, func(node *types.Node) error {
return n, c.withNodeLocked(ctx, opts.Nodename, func(node *types.Node) error {
n = node
litter.Dump(opts)
// status
Expand Down
2 changes: 1 addition & 1 deletion cluster/calcium/realloc.go
Expand Up @@ -100,7 +100,7 @@ func (c *Calcium) doReallocContainer(
for newCPU, memNodesContainers := range cpuMemNodeContainersInfo {
for newMemory, nodesContainers := range memNodesContainers {
for nodename, containers := range nodesContainers {
if err := c.withNodeLocked(ctx, pod.Name, nodename, func(node *types.Node) error {
if err := c.withNodeLocked(ctx, nodename, func(node *types.Node) error {
// 把记录的 CPU 还回去,变成新的可用资源
// 把记录的 Memory 还回去,变成新的可用资源
containerWithCPUBind := 0
Expand Down
2 changes: 1 addition & 1 deletion cluster/calcium/remove.go
Expand Up @@ -26,7 +26,7 @@ func (c *Calcium) RemoveContainer(ctx context.Context, IDs []string, force bool,
output := []*bytes.Buffer{}
success := false
if err := c.withContainerLocked(ctx, ID, func(container *types.Container) error {
return c.withNodeLocked(ctx, container.Podname, container.Nodename, func(node *types.Node) (err error) {
return c.withNodeLocked(ctx, container.Nodename, func(node *types.Node) (err error) {
if err = c.doRemoveContainer(ctx, container, force); err != nil {
return err
}
Expand Down

0 comments on commit 9c6eaf4

Please sign in to comment.