Merge branch 'timeout' of gitlab.ricebook.net:platform/core into timeout

Author: CMGS

cluster/calcium/run_and_wait.go

@@ -3,6 +3,8 @@ package calcium
 import (
 	"bufio"
 	"fmt"
+	"sync"
+	"time"
 
 	log "github.com/Sirupsen/logrus"
 	enginetypes "github.com/docker/docker/api/types"
@@ -11,6 +13,11 @@ import (
 	"golang.org/x/net/context"
 )
 
+// TODO 这里还是有个问题啊
+// 如果在这个等待的过程中重启了core, 那岂不是又有容器没有回收的?
+// 有啥方法避免这个问题么?
+const defaultWaitTimeout = 1200
+
 func (c *calcium) RunAndWait(specs types.Specs, opts *types.DeployOptions) (chan *types.RunAndWaitMessage, error) {
 	ch := make(chan *types.RunAndWaitMessage)
 
@@ -19,35 +26,51 @@ func (c *calcium) RunAndWait(specs types.Specs, opts *types.DeployOptions) (chan
 	entry.LogConfig = "json-file"
 	specs.Entrypoints[opts.Entrypoint] = entry
 
+	// 默认给出1200秒的超时时间吧
+	// 没别的地方好传了, 不如放这里好了, 不需要用的就默认0或者不传
+	waitTimeout := entry.RunAndWaitTimeout
+	if waitTimeout == 0 {
+		waitTimeout = defaultWaitTimeout
+	}
+
+	// 创建容器, 有问题就gg
 	createChan, err := c.CreateContainer(specs, opts)
 	if err != nil {
-		log.Errorf("[RunAndWait] Create container error, %s", err.Error())
+		log.Errorf("[RunAndWait] Create container error, %v", err)
 		return ch, err
 	}
 
+	// 来个goroutine处理剩下的事情
+	// 基本上就是, attach拿日志写到channel, 以及等待容器结束后清理资源
 	go func() {
-		defer log.Info("[RunAndWait] Finish run and wait for containers")
-		defer close(ch)
 		logsOpts := enginetypes.ContainerLogsOptions{Follow: true, ShowStdout: true, ShowStderr: true}
+		wg := sync.WaitGroup{}
 
-		ids := map[string]*types.Node{}
 		for message := range createChan {
+			// 可能不成功, 可能没有容器id, 但是其实第一项足够判断了
+			// 不成功就无视掉
 			if !message.Success || message.ContainerID == "" {
 				log.Errorf("[RunAndWait] Create container error, %s", message.Error)
 				continue
 			}
 
+			// 找不到对应node也不管
+			// 理论上不会这样
 			node, err := c.store.GetNode(message.Podname, message.Nodename)
 			if err != nil {
-				log.Errorf("[RunAndWait] Can't find node, %s", err.Error())
+				log.Errorf("[RunAndWait] Can't find node, %v", err)
 				continue
 			}
 
-			ids[message.ContainerID] = node
+			// 加个task
+			wg.Add(1)
+
+			// goroutine attach日志然后处理了写回给channel
+			// 日志跟task无关, 不管wg
 			go func(node *types.Node, containerID string) {
 				resp, err := node.Engine.ContainerLogs(context.Background(), containerID, logsOpts)
 				if err != nil {
-					log.Errorf("[RunAndWait] Failed to get logs, %s", err.Error())
+					log.Errorf("[RunAndWait] Failed to get logs, %v", err)
 					return
 				}
 
@@ -60,25 +83,37 @@ func (c *calcium) RunAndWait(specs types.Specs, opts *types.DeployOptions) (chan
 				}
 
 				if err := scanner.Err(); err != nil {
-					log.Errorf("[RunAndWait] Parse log failed, %s", err.Error())
+					log.Errorf("[RunAndWait] Parse log failed, %v", err)
 					return
 				}
 			}(node, message.ContainerID)
-		}
 
-		rmids := []string{}
-		for id, node := range ids {
-			rmids = append(rmids, id)
-			code, err := node.Engine.ContainerWait(context.Background(), id)
-			exitData := []byte(fmt.Sprintf("[exitcode] %d", code))
-			if err != nil {
-				log.Errorf("%s run failed, %s", id[:12], err.Error())
-				exitData = []byte(fmt.Sprintf("[exitcode]unknown %s", err.Error()))
-			}
-			ch <- &types.RunAndWaitMessage{ContainerID: id, Data: exitData}
-			log.Infof("[RunAndWait] Container %s finished, remove", id[:12])
+			// goroutine 等待容器结束后删除掉
+			go func(node *types.Node, containerID string) {
+				// 无论如何一定要删掉这个容器并且回收资源
+				// 并且容器删除了才算是task完成了一个
+				defer wg.Done()
+				defer log.Infof("[RunAndWait] Container %s finished and removed", containerID[:12])
+				defer c.RemoveContainer([]string{containerID})
+
+				// 给超时时间, 不能等个没完没了
+				// 超时的时候返回的code会是1, 不过会被err给覆盖掉
+				ctx, _ := context.WithTimeout(context.Background(), time.Duration(waitTimeout)*time.Second)
+				code, err := node.Engine.ContainerWait(ctx, containerID)
+				exitData := []byte(fmt.Sprintf("[exitcode] %d", code))
+				if err != nil {
+					log.Errorf("%s run failed, %v", containerID[:12], err)
+					exitData = []byte(fmt.Sprintf("[exitcode]unknown %v", err))
+				}
+
+				ch <- &types.RunAndWaitMessage{ContainerID: containerID, Data: exitData}
+			}(node, message.ContainerID)
 		}
-		go c.RemoveContainer(rmids)
+
+		// 等待全部任务完成才可以关闭channel
+		wg.Wait()
+		log.Info("[RunAndWait] Finish run and wait for containers")
+		close(ch)
 	}()
 
 	return ch, nil

types/specs.go

@@ -33,6 +33,7 @@ type Entrypoint struct {
 	Privileged              string   `yaml:"privileged,omitempty"`
 	LogConfig               string   `yaml:"log_config,omitempty"`
 	WorkingDir              string   `yaml:"working_dir,omitempty"`
+	RunAndWaitTimeout       int      `yaml:"run_and_wait_timeout"`
 }
 
 // single bind