Refactor a bit, combine node resouce and pod resouce calculation (#264)

* support get all nodes by ListPodNodes API

* cleanup comments

* support store status forever

* decrease service put frequency

* remove useless

* combine node resource and pod resource

cluster/calcium/build.go

@@ -43,7 +43,7 @@ func (c *Calcium) BuildImage(ctx context.Context, opts *types.BuildOptions) (cha
 
 func (c *Calcium) selectBuildNode(ctx context.Context) (*types.Node, error) {
 	// get pod from config
-	// TODO VM BRANCH conside vm build machines.
+	// TODO can choose multiple pod here for other engine support
 	if c.config.Docker.BuildPod == "" {
 		return nil, types.ErrNoBuildPod
 	}

cluster/calcium/build_test.go

@@ -21,7 +21,6 @@ const (
 	repo = "https://test/repo.git"
 )
 
-// TODO test errors
 // test no tags
 func TestBuild(t *testing.T) {
 	c := NewTestCluster()

cluster/calcium/calcium.go

@@ -17,7 +17,7 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-//Calcium implement the cluster
+// Calcium implement the cluster
 type Calcium struct {
 	config    types.Config
 	store     store.Store

cluster/calcium/control.go

@@ -65,7 +65,7 @@ func (c *Calcium) doStartContainer(ctx context.Context, container *types.Contain
 	if err = container.Start(ctx); err != nil {
 		return message, err
 	}
-	// TODO healthcheck
+	// TODO healthcheck first
 	if container.Hook != nil && len(container.Hook.AfterStart) > 0 {
 		message, err = c.doHook(
 			ctx,

cluster/calcium/create.go

@@ -19,8 +19,6 @@ import (
 
 // CreateContainer use options to create containers
 func (c *Calcium) CreateContainer(ctx context.Context, opts *types.DeployOptions) (chan *types.CreateContainerMessage, error) {
-	// TODO: new way to normalize
-	//opts.Normalize()
 	opts.ProcessIdent = utils.RandomString(16)
 	log.Infof("[CreateContainer %s] Creating container with options:", opts.ProcessIdent)
 	litter.Dump(opts)
@@ -134,7 +132,6 @@ func (c *Calcium) doDeployWorkloads(ctx context.Context, ch chan *types.CreateCo
 				err = e
 				rollbackMap[nodename] = indices
 			}
-			//process
 		}(nodename, deploy, seq)
 		seq += deploy
 	}

cluster/calcium/pod.go

@@ -15,7 +15,7 @@ func (c *Calcium) AddPod(ctx context.Context, podname, desc string) (*types.Pod,
 func (c *Calcium) RemovePod(ctx context.Context, podname string) error {
 	return c.withNodesLocked(ctx, podname, []string{}, nil, true, func(nodes map[string]*types.Node) error {
 		// TODO dissociate container to node
-		// TODO remove node first
+		// TODO should remove node first
 		return c.store.RemovePod(ctx, podname)
 	})
 }

cluster/calcium/resource.go

@@ -3,7 +3,6 @@ package calcium
 import (
 	"context"
 	"fmt"
-	"strings"
 
 	"github.com/pkg/errors"
 	log "github.com/sirupsen/logrus"
@@ -22,25 +21,15 @@ func (c *Calcium) PodResource(ctx context.Context, podname string) (*types.PodRe
 		return nil, err
 	}
 	r := &types.PodResource{
-		Name:            podname,
-		CPUPercents:     map[string]float64{},
-		MemoryPercents:  map[string]float64{},
-		StoragePercents: map[string]float64{},
-		VolumePercents:  map[string]float64{},
-		Verifications:   map[string]bool{},
-		Details:         map[string]string{},
+		Name:          podname,
+		NodesResource: []*types.NodeResource{},
 	}
 	for _, node := range nodes {
-		nodeDetail, err := c.doGetNodeResource(ctx, node.Name, false)
+		nodeResource, err := c.doGetNodeResource(ctx, node.Name, false)
 		if err != nil {
 			return nil, err
 		}
-		r.CPUPercents[node.Name] = nodeDetail.CPUPercent
-		r.MemoryPercents[node.Name] = nodeDetail.MemoryPercent
-		r.StoragePercents[node.Name] = nodeDetail.StoragePercent
-		r.VolumePercents[node.Name] = nodeDetail.VolumePercent
-		r.Verifications[node.Name] = nodeDetail.Verification
-		r.Details[node.Name] = strings.Join(nodeDetail.Details, "\n")
+		r.NodesResource = append(r.NodesResource, nodeResource)
 	}
 	return r, nil
 }
@@ -53,8 +42,7 @@ func (c *Calcium) NodeResource(ctx context.Context, nodename string, fix bool) (
 	}
 	for _, container := range nr.Containers {
 		if _, err := container.Inspect(ctx); err != nil { // 用于探测节点上容器是否存在
-			nr.Verification = false
-			nr.Details = append(nr.Details, fmt.Sprintf("container %s inspect failed %v \n", container.ID, err))
+			nr.Diffs = append(nr.Diffs, fmt.Sprintf("container %s inspect failed %v \n", container.ID, err))
 			continue
 		}
 	}
@@ -70,7 +58,7 @@ func (c *Calcium) doGetNodeResource(ctx context.Context, nodename string, fix bo
 		}
 		nr = &types.NodeResource{
 			Name: node.Name, CPU: node.CPU, MemCap: node.MemCap, StorageCap: node.StorageCap,
-			Containers: containers, Verification: true, Details: []string{},
+			Containers: containers, Diffs: []string{},
 		}
 
 		cpus := 0.0
@@ -93,33 +81,29 @@ func (c *Calcium) doGetNodeResource(ctx context.Context, nodename string, fix bo
 			}
 		}
 		if cpus != node.CPUUsed {
-			nr.Verification = false
-			nr.Details = append(nr.Details, fmt.Sprintf("cpus used: %f diff: %f", node.CPUUsed, cpus))
+			nr.Diffs = append(nr.Diffs, fmt.Sprintf("cpus used: %f diff: %f", node.CPUUsed, cpus))
 		}
 		node.CPU.Add(cpumap)
 		for i, v := range node.CPU {
 			if node.InitCPU[i] != v {
-				nr.Verification = false
-				nr.Details = append(nr.Details, fmt.Sprintf("cpu %s diff %d", i, node.InitCPU[i]-v))
+				nr.Diffs = append(nr.Diffs, fmt.Sprintf("cpu %s diff %d", i, node.InitCPU[i]-v))
 			}
 		}
 
 		if memory+node.MemCap != node.InitMemCap {
-			nr.Verification = false
-			nr.Details = append(nr.Details, fmt.Sprintf("memory used: %d, diff %d", node.MemCap, node.InitMemCap-(memory+node.MemCap)))
+			nr.Diffs = append(nr.Diffs, fmt.Sprintf("memory used: %d, diff %d", node.MemCap, node.InitMemCap-(memory+node.MemCap)))
 		}
 
 		nr.StoragePercent = 0
 		if node.InitStorageCap != 0 {
 			nr.StoragePercent = float64(storage) / float64(node.InitStorageCap)
 			if storage+node.StorageCap != node.InitStorageCap {
-				nr.Verification = false
-				nr.Details = append(nr.Details, fmt.Sprintf("storage used: %d, diff %d", node.StorageCap, node.InitStorageCap-(storage+node.StorageCap)))
+				nr.Diffs = append(nr.Diffs, fmt.Sprintf("storage used: %d, diff %d", node.StorageCap, node.InitStorageCap-(storage+node.StorageCap)))
 			}
 		}
 
 		if err := node.Engine.ResourceValidate(ctx, cpus, cpumap, memory, storage); err != nil {
-			nr.Details = append(nr.Details, err.Error())
+			nr.Diffs = append(nr.Diffs, err.Error())
 		}
 
 		if fix {

cluster/calcium/resource_test.go

@@ -80,11 +80,10 @@ func TestPodResource(t *testing.T) {
 	// success
 	r, err := c.PodResource(ctx, podname)
 	assert.NoError(t, err)
-	assert.Len(t, r.CPUPercents, 1)
-	assert.Len(t, r.MemoryPercents, 1)
-	assert.Len(t, r.StoragePercents, 1)
-	assert.False(t, r.Verifications[nodename])
-	assert.NotEmpty(t, r.Details[nodename])
+	assert.Equal(t, r.NodesResource[0].CPUPercent, 0.9)
+	assert.Equal(t, r.NodesResource[0].MemoryPercent, 0.5)
+	assert.Equal(t, r.NodesResource[0].StoragePercent, 0.1)
+	assert.NotEmpty(t, r.NodesResource[0].Diffs)
 }
 
 func TestNodeResource(t *testing.T) {
@@ -146,9 +145,8 @@ func TestNodeResource(t *testing.T) {
 	nr, err := c.NodeResource(ctx, nodename, true)
 	assert.NoError(t, err)
 	assert.Equal(t, nr.Name, nodename)
-	assert.NotEmpty(t, nr.Details)
-	assert.False(t, nr.Verification)
-	details := strings.Join(nr.Details, ",")
+	assert.NotEmpty(t, nr.Diffs)
+	details := strings.Join(nr.Diffs, ",")
 	assert.Contains(t, details, "inspect failed")
 }
 

discovery/helium/helium.go

@@ -68,11 +68,9 @@ func (h *Helium) start(ctx context.Context) {
 					Addresses: addresses,
 					Interval:  h.config.ServiceDiscoveryPushInterval * 2,
 				}
-				h.dispatch(latestStatus)
-
 			case <-timer.C:
-				h.dispatch(latestStatus)
 			}
+			h.dispatch(latestStatus)
 			timer.Stop()
 			timer.Reset(h.config.ServiceDiscoveryPushInterval)
 		}

engine/docker/build.go

@@ -34,8 +34,8 @@ WORKDIR {{.Dir}}{{ end }}
 {{ if .StopSignal }}STOPSIGNAL {{.StopSignal}} {{ end }}`
 	copyTmpl = "COPY --from=%s %s %s"
 	runTmpl  = "RUN %s"
-	//TODO consider work dir privilege
-	//Add user manually
+	// TODO consider work dir privilege
+	// Add user manually
 	userTmpl = `RUN echo "{{.User}}::{{.UID}}:{{.UID}}:{{.User}}:/dev/null:/sbin/nologin" >> /etc/passwd && \
 echo "{{.User}}:x:{{.UID}}:" >> /etc/group && \
 echo "{{.User}}:!::0:::::" >> /etc/shadow

engine/docker/container.go

@@ -331,11 +331,8 @@ func (e *Engine) VirtualizationUpdateResource(ctx context.Context, ID string, op
 	}
 
 	memory := opts.Memory
-	//softLimit := opts.SoftLimit
-	// unlimited memory
 	if memory == 0 {
 		memory = maxMemory
-		//softLimit = false
 	}
 
 	quota := opts.Quota

engine/docker/helper.go

@@ -119,9 +119,6 @@ func makeResourceSetting(cpu float64, memory int64, cpuMap map[string]int64, num
 		// numaNode will empty or numaNode
 		resource.CpusetMems = numaNode
 	}
-	//if softlimit {
-	//	resource.MemoryReservation = memory
-	//} else {
 	resource.Memory = memory
 	resource.MemorySwap = memory
 	resource.MemoryReservation = memory / 2

go.mod

@@ -41,7 +41,6 @@ require (
 	golang.org/x/net v0.0.0-20200813134508-3edf25e44fcc
 	golang.org/x/sys v0.0.0-20200812155832-6a926be9bd1d // indirect
 	google.golang.org/grpc v1.28.0
-	google.golang.org/protobuf v1.23.0
 	gopkg.in/alexcesaro/statsd.v2 v2.0.0 // indirect
 	gotest.tools v2.2.0+incompatible // indirect
 )

rpc/counter.go

@@ -29,7 +29,7 @@ func (v *Vibranium) taskDone(name string, verbose bool) {
 	v.TaskNum--
 }
 
-//Wait for all tasks done
+// Wait for all tasks done
 // 会在外面graceful之后调用.
 // 不完成不给退出进程.
 func (v *Vibranium) Wait() {