Syncano · 23doors · Jul 8, 2020 · Jul 8, 2020
diff --git a/app/docker/manager.go b/app/docker/manager.go
@@ -58,6 +58,10 @@ type Constraints struct {
 	StorageLimit    string
 }
 
+func (c *Constraints) MCPU() uint32 {
+	return uint32(c.CPULimit / 1e6)
+}
+
 const (
 	containerWorkdir = "/tmp"
 	defaultTimeout   = 30 * time.Second

diff --git a/app/lb/server.go b/app/lb/server.go
@@ -46,7 +46,9 @@ type ServerOptions struct {
 	WorkerRetry          int
 	WorkerKeepalive      time.Duration
 	WorkerMinReady       int
-	WorkerErrorThreshold uint32
+	WorkerErrorThreshold uint
+
+	DefaultScriptMcpu uint
 
 	// Limiter
 	LimiterOptions limiter.Options
@@ -56,7 +58,10 @@ type ServerOptions struct {
 var DefaultOptions = &ServerOptions{
 	WorkerRetry:          3,
 	WorkerKeepalive:      30 * time.Second,
+	WorkerMinReady:       1,
 	WorkerErrorThreshold: 2,
+
+	DefaultScriptMcpu: 125,
 }
 
 const (
@@ -171,6 +176,15 @@ func (s *Server) Run(stream pb.ScriptRunner_RunServer) error {
 		return common.ErrInvalidArgument
 	}
 
+	// Set defaults
+	if scriptMeta.Options == nil {
+		scriptMeta.Options = &scriptpb.RunMeta_Options{}
+	}
+
+	if scriptMeta.Options.Mcpu == 0 {
+		scriptMeta.Options.Mcpu = uint32(s.options.DefaultScriptMcpu)
+	}
+
 	return s.processRun(ctx, logger, stream, meta, scriptMeta)
 }
 
@@ -274,6 +288,7 @@ func (s *Server) processRun(ctx context.Context, logger logrus.FieldLogger, stre
 
 	if response != nil && response.Cached {
 		// Add container to worker cache if we got any response.
+		cont.ID = response.ContainerId
 		s.addCache(cont, def)
 	}
 
@@ -306,16 +321,21 @@ func (s *Server) addCache(cont *WorkerContainer, def *script.Definition) {
 	s.mu.Unlock()
 }
 
-func (s *Server) removeCache(w *Worker, def *script.Definition, id string) {
+func (s *Server) removeCache(w *Worker, def *script.Definition, containerID string) bool {
 	idx := def.Index.Hash()
 	defHash := def.Hash()
+	removed := true
 
 	s.mu.Lock()
 
-	if w.RemoveCache(def, id) {
+	if w.RemoveCache(def, containerID) {
 		// Remove worker container with specified definition from cache map.
 		if m, ok := s.workerContainersCached[defHash]; ok {
-			delete(m, id)
+			if _, ok = m[containerID]; ok {
+				delete(m, containerID)
+			} else {
+				removed = false
+			}
 
 			if len(m) == 0 {
 				delete(s.workerContainersCached, defHash)
@@ -330,6 +350,8 @@ func (s *Server) removeCache(w *Worker, def *script.Definition, id string) {
 				} else {
 					m[w.ID]--
 				}
+			} else {
+				removed = false
 			}
 
 			if len(m) == 0 {
@@ -339,6 +361,8 @@ func (s *Server) removeCache(w *Worker, def *script.Definition, id string) {
 	}
 
 	s.mu.Unlock()
+
+	return removed
 }
 
 func (s *Server) relayReponse(logger logrus.FieldLogger, stream pb.ScriptRunner_RunServer, cont *WorkerContainer, resCh <-chan interface{}) (*scriptpb.RunResponse, error) {
@@ -491,7 +515,11 @@ func (s *Server) ContainerRemoved(ctx context.Context, in *pb.ContainerRemovedRe
 		return nil, ErrUnknownWorkerID
 	}
 
-	s.removeCache(cur.(*Worker), def, in.ContainerId)
+	if s.removeCache(cur.(*Worker), def, in.ContainerId) {
+		logger.Info("Removed container from worker cache")
+	} else {
+		logger.Warn("Container not found in worker cache!")
+	}
 
 	return &pb.ContainerRemovedResponse{}, nil
 }

diff --git a/app/lb/server_test.go b/app/lb/server_test.go
@@ -410,6 +410,7 @@ func TestServerMethods(t *testing.T) {
 			handler := http.HandlerFunc(s.ReadyHandler)
 
 			Convey("writes 200 if worker count is satisfied", func() {
+				s.options.WorkerMinReady = 0
 				handler.ServeHTTP(rr, req)
 				So(rr.Code, ShouldEqual, http.StatusOK)
 			})

diff --git a/app/lb/worker.go b/app/lb/worker.go
@@ -149,8 +149,8 @@ func (w *Worker) release(mCPU, conns uint32) {
 }
 
 // IncreaseErrorCount increases error count of worker.
-func (w *Worker) IncreaseErrorCount() uint32 {
-	return atomic.AddUint32(&w.errorCount, 1)
+func (w *Worker) IncreaseErrorCount() uint {
+	return uint(atomic.AddUint32(&w.errorCount, 1))
 }
 
 // ResetErrorCount resets error count of worker.

diff --git a/app/script/runner.go b/app/script/runner.go
@@ -403,7 +403,7 @@ func (r *DockerRunner) Run(ctx context.Context, logger logrus.FieldLogger, reque
 	}
 
 	if def.MCPU == 0 {
-		def.MCPU = uint32(r.options.Constraints.CPULimit / 1e6)
+		def.MCPU = r.options.Constraints.MCPU()
 	}
 
 	if options.Weight == 0 {

diff --git a/cmd/lb.go b/cmd/lb.go
@@ -69,8 +69,16 @@ As there is no authentication, always run it in a private network.`,
 			EnvVars: []string{"WORKER_RETRY"}, Value: lb.DefaultOptions.WorkerRetry, Destination: &lbOptions.WorkerRetry,
 		},
 		&cli.IntFlag{
-			Name: "worker-min-ready", Usage: "number of retries on failed worker run",
-			EnvVars: []string{"WORKER_MIN_READY"}, Destination: &lbOptions.WorkerMinReady,
+			Name: "worker-min-ready", Usage: "number of required workers available",
+			EnvVars: []string{"WORKER_MIN_READY"}, Value: lb.DefaultOptions.WorkerMinReady, Destination: &lbOptions.WorkerMinReady,
+		},
+		&cli.UintFlag{
+			Name: "worker-error-threshold", Usage: "worker error max threshold",
+			EnvVars: []string{"WORKER_ERROR_THRESHOLD"}, Value: lb.DefaultOptions.WorkerErrorThreshold, Destination: &lbOptions.WorkerErrorThreshold,
+		},
+		&cli.UintFlag{
+			Name: "default-script-mcpu", Usage: "default script run mcpu",
+			EnvVars: []string{"DEFAULT_SCRIPT_MCPU"}, Value: lb.DefaultOptions.DefaultScriptMcpu, Destination: &lbOptions.DefaultScriptMcpu,
 		},
 		// LB Limiter options.
 		&cli.DurationFlag{

diff --git a/deploy/env/eu1.env b/deploy/env/eu1.env
@@ -1,6 +1,8 @@
 BROKER_MIN=2
 BROKER_MAX=8
+MCPU=2000
 CONCURRENCY=15
+DEFAULT_SCRIPT_MCPU=125
 DEBUG=0
 DOCKER_DEVICE=/dev/sdb
 DOCKER_DNS=169.254.169.254

diff --git a/deploy/env/staging.env b/deploy/env/staging.env
@@ -1,6 +1,8 @@
 BROKER_MIN=1
 BROKER_MAX=8
+MCPU=2000
 CONCURRENCY=15
+DEFAULT_SCRIPT_MCPU=125
 DEBUG=1
 DOCKER_DEVICE=/dev/sdb
 DOCKER_DNS=169.254.169.254