support unlimited containers

VERSION

@@ -1 +1 @@
-20.03.01
+20.03.02

agent.go

@@ -22,6 +22,7 @@ func setupLogLevel(l string) error {
 		return err
 	}
 	log.SetLevel(level)
+	log.SetOutput(os.Stdout)
 	return nil
 }
 

engine/attach.go

@@ -22,6 +22,9 @@ import (
 
 func (e *Engine) attach(container *types.Container) {
 	transfer := e.forwards.Get(container.ID, 0)
+	if transfer == "" {
+		transfer = logs.Discard
+	}
 	writer, err := logs.NewWriter(transfer, e.config.Log.Stdout)
 	if err != nil {
 		log.Errorf("[attach] Create log forward failed %s", err)

engine/engine.go

@@ -15,6 +15,7 @@ import (
 	coretypes "github.com/projecteru2/core/types"
 	coreutils "github.com/projecteru2/core/utils"
 	"github.com/shirou/gopsutil/cpu"
+	"github.com/shirou/gopsutil/mem"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -25,6 +26,7 @@ type Engine struct {
 	docker  *engineapi.Client
 	node    *coretypes.Node
 	cpuCore float64 // 因为到时候要乘以 float64 所以就直接转换成 float64 吧
+	memory  int64
 
 	transfers *utils.HashBackends
 	forwards  *utils.HashBackends
@@ -64,7 +66,13 @@ func NewEngine(config *types.Config) (*Engine, error) {
 		return nil, err
 	}
 	log.Infof("[NewEngine] Host has %d cpus", len(cpus))
+	memory, err := mem.VirtualMemory()
+	if err != nil {
+		return nil, err
+	}
+	log.Infof("[NewEngine] Host has %d memory", memory.Total)
 	engine.cpuCore = float64(len(cpus))
+	engine.memory = int64(memory.Total)
 	engine.transfers = utils.NewHashBackends(config.Metrics.Transfers)
 	engine.forwards = utils.NewHashBackends(config.Log.Forwards)
 	return engine, nil

engine/helper.go

@@ -56,6 +56,9 @@ func (e *Engine) detectContainer(ID string) (*types.Container, error) {
 	}
 	// 计算容器用了多少 CPU
 	container = status.CalcuateCPUNum(container, c, e.cpuCore)
+	if container.Memory == 0 {
+		container.Memory = e.memory
+	}
 	// 活着才有发布必要
 	if c.NetworkSettings != nil && container.Running {
 		networks := map[string]string{}

engine/logs/writer.go

@@ -12,6 +12,9 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
+// Discard .
+const Discard = "__discard__"
+
 // ErrConnecting means writer is in connecting status, waiting to be connected
 var ErrConnecting = errors.New("Connecting")
 
@@ -40,7 +43,7 @@ func (d discard) Close() error {
 
 // NewWriter return writer
 func NewWriter(addr string, stdout bool) (*Writer, error) {
-	if addr == "__discard__" {
+	if addr == Discard {
 		return &Writer{
 			enc: NewStreamEncoder(discard{}),
 		}, nil

go.mod

@@ -19,9 +19,9 @@ require (
 	github.com/grpc-ecosystem/go-grpc-prometheus v1.2.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway v1.8.5 // indirect
 	github.com/jinzhu/configor v1.1.1
-	github.com/projecteru2/core v0.0.0-20191211114916-9678b49b98a5
+	github.com/projecteru2/core v0.0.0-20200313031343-262c693ab0bc
 	github.com/prometheus/client_golang v0.9.3
-	github.com/shirou/gopsutil v2.19.11+incompatible
+	github.com/shirou/gopsutil v2.20.2+incompatible
 	github.com/sirupsen/logrus v1.4.2
 	github.com/soheilhy/cmux v0.1.4 // indirect
 	github.com/stretchr/testify v1.3.0

store/core/node.go

@@ -17,7 +17,7 @@ func (c *CoreStore) GetNode(nodename string) (*types.Node, error) {
 
 	cpus := types.CPUMap{}
 	for k, v := range resp.Cpu {
-		cpus[k] = int(v)
+		cpus[k] = int64(v)
 	}
 
 	node := &types.Node{

utils/hash.go

@@ -15,6 +15,9 @@ func NewHashBackends(data []string) *HashBackends {
 
 // Get get a backend
 func (s *HashBackends) Get(v string, offset int) string {
+	if s.length == 0 {
+		return ""
+	}
 	h := fnv.New32a()
 	h.Write([]byte(v))
 	return s.data[(h.Sum32()+uint32(offset))%s.length]