Merge branch 'run-and-wait' into 'master'

run and wait

See merge request !69

Author: CMGS

cluster/calcium/create_container.go

@@ -26,9 +26,8 @@ func (c *calcium) CreateContainer(specs types.Specs, opts *types.DeployOptions)
 	pod, _ := c.store.GetPod(opts.Podname)
 	if pod.Scheduler == "CPU" {
 		return c.createContainerWithScheduler(specs, opts)
-	} else {
-		return c.createContainerWithCPUPeriod(specs, opts)
 	}
+	return c.createContainerWithCPUPeriod(specs, opts)
 }
 
 func (c *calcium) createContainerWithCPUPeriod(specs types.Specs, opts *types.DeployOptions) (chan *types.CreateContainerMessage, error) {

cluster/calcium/helper.go

@@ -140,7 +140,7 @@ func makeMountPaths(specs types.Specs, config types.Config) ([]string, map[strin
 func runExec(client *engineapi.Client, container enginetypes.ContainerJSON, label string) error {
 	cmd, ok := container.Config.Labels[label]
 	if !ok || cmd == "" {
-		log.Debug("No %s found in container %s", label, container.ID)
+		log.Debugf("No %s found in container %s", label, container.ID)
 		return nil
 	}
 

cluster/calcium/run_and_wait.go

@@ -0,0 +1,94 @@
+package calcium
+
+import (
+	"bufio"
+	"fmt"
+	"sync"
+
+	log "github.com/Sirupsen/logrus"
+	enginetypes "github.com/docker/docker/api/types"
+	"gitlab.ricebook.net/platform/core/types"
+	"golang.org/x/net/context"
+)
+
+// FUCK DOCKER
+const PREFIXLEN int = 8
+
+func (c *calcium) RunAndWait(specs types.Specs, opts *types.DeployOptions) (chan *types.RunAndWaitMessage, error) {
+	ch := make(chan *types.RunAndWaitMessage)
+
+	// 强制为 json-file 输出
+	entry, _ := specs.Entrypoints[opts.Entrypoint]
+	entry.LogConfig = "json-file"
+	specs.Entrypoints[opts.Entrypoint] = entry
+
+	createChan, err := c.CreateContainer(specs, opts)
+	if err != nil {
+		log.Errorf("[RunAndWait] Create container error, %s", err.Error())
+		return ch, err
+	}
+
+	go func() {
+		wg := &sync.WaitGroup{}
+		defer log.Info("[RunAndWait] Finish run and wait for containers")
+		defer close(ch)
+		defer wg.Wait()
+		logsOpts := enginetypes.ContainerLogsOptions{Follow: true, ShowStdout: true, ShowStderr: true}
+
+		for message := range createChan {
+			wg.Add(1)
+			if message.ContainerID == "" {
+				log.Errorf("[RunAndWait] Can't find container id %s", err.Error())
+				continue
+			}
+
+			node, err := c.store.GetNode(message.Podname, message.Nodename)
+			if err != nil {
+				log.Errorf("[RunAndWait] Can't find node, %s", err.Error())
+				continue
+			}
+
+			go func(node *types.Node, message *types.CreateContainerMessage) {
+				defer wg.Done()
+				resp, err := node.Engine.ContainerLogs(context.Background(), message.ContainerID, logsOpts)
+				if err != nil {
+					data := fmt.Sprintf("[RunAndWait] Failed to get logs, %s", err.Error())
+					ch <- &types.RunAndWaitMessage{ContainerID: message.ContainerID, Data: data}
+					return
+				}
+
+				scanner := bufio.NewScanner(resp)
+				for scanner.Scan() {
+					data := scanner.Bytes()[PREFIXLEN:]
+					log.Debugf("[RunAndWait] %s %s", message.ContainerID[:12], data)
+					m := &types.RunAndWaitMessage{ContainerID: message.ContainerID, Data: string(data)}
+					ch <- m
+				}
+
+				if err := scanner.Err(); err != nil {
+					data := fmt.Sprintf("[RunAndWait] Parse log failed, %s", err.Error())
+					ch <- &types.RunAndWaitMessage{ContainerID: message.ContainerID, Data: data}
+					return
+				}
+
+				container, err := c.GetContainer(message.ContainerID)
+				if err != nil {
+					data := fmt.Sprintf("[RunAndWait] Container not found, %s", err.Error())
+					ch <- &types.RunAndWaitMessage{ContainerID: message.ContainerID, Data: data}
+					return
+				}
+				defer c.removeOneContainer(container)
+
+				containerJSON, err := container.Inspect()
+				if err == nil {
+					ch <- &types.RunAndWaitMessage{ContainerID: message.ContainerID, Data: fmt.Sprintf("[exitcode] %d", containerJSON.State.ExitCode)}
+				} else {
+					ch <- &types.RunAndWaitMessage{ContainerID: message.ContainerID, Data: fmt.Sprintf("[exitcode]unknown %s", err.Error())}
+				}
+				log.Infof("[RunAndWait] Container %s finished, remove", message.ContainerID)
+			}(node, message)
+		}
+	}()
+
+	return ch, nil
+}

cluster/cluster.go

@@ -21,6 +21,7 @@ type Cluster interface {
 	// cluster methods
 	BuildImage(repository, version, uid, artifact string) (chan *types.BuildImageMessage, error)
 	CreateContainer(specs types.Specs, opts *types.DeployOptions) (chan *types.CreateContainerMessage, error)
+	RunAndWait(specs types.Specs, opts *types.DeployOptions) (chan *types.RunAndWaitMessage, error)
 	UpgradeContainer(ids []string, image string) (chan *types.UpgradeContainerMessage, error)
 	RemoveContainer(ids []string) (chan *types.RemoveContainerMessage, error)
 	RemoveImage(podname, nodename string, images []string) (chan *types.RemoveImageMessage, error)