Merge branch 'master' into satisfy-yifu

Author: CMGS

VERSION

@@ -1 +1 @@
-0.7.22
+0.7.23

cluster/calcium/build_image.go

@@ -72,7 +72,9 @@ func getRandomNode(c *calcium, podname string) (*types.Node, error) {
 		return nil, err
 	}
 	if len(nodes) == 0 {
-		return nil, fmt.Errorf("No nodes available in pod %s", podname)
+		err = fmt.Errorf("No nodes available in pod %s", podname)
+		log.Debugf("Error during getRandomNode from %s: %v", podname, err)
+		return nil, err
 	}
 
 	nodemap := make(map[string]types.CPUMap)
@@ -81,6 +83,11 @@ func getRandomNode(c *calcium, podname string) (*types.Node, error) {
 	}
 	nodename, err := c.scheduler.RandomNode(nodemap)
 	if err != nil {
+		log.Debugf("Error during getRandomNode from %s: %v", podname, err)
+		return nil, err
+	}
+	if nodename == "" {
+		err = fmt.Errorf("Got empty node during getRandomNode from %s", podname)
 		return nil, err
 	}
 

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) {
@@ -38,14 +37,18 @@ func (c *calcium) createContainerWithCPUPeriod(specs types.Specs, opts *types.De
 		return ch, fmt.Errorf("Minimum memory limit allowed is 4MB")
 	}
 
+	log.Debugf("Deploy options: %v", opts)
+	log.Debugf("Deploy specs: %v", specs)
 	cpuandmem, _, err := c.getCPUAndMem(opts.Podname, opts.Nodename, 1.0)
 	if err != nil {
 		return ch, err
 	}
 	go utils.SendMemCap(cpuandmem, "before-alloc")
 	nodesInfo := utils.GetNodesInfo(cpuandmem)
 
+	log.Debugf("Input opts.CPUQuota: %f", opts.CPUQuota)
 	cpuQuota := int(opts.CPUQuota * float64(utils.CpuPeriodBase))
+	log.Debugf("Tranfered cpuQuota: %d", cpuQuota)
 	plan, err := utils.AllocContainerPlan(nodesInfo, cpuQuota, opts.Memory, opts.Count) // 还是以 Bytes 作单位， 不转换了
 
 	if err != nil {

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/meta.go

@@ -3,7 +3,10 @@ package calcium
 // All functions are just proxy to store, since I don't want store to be exported.
 // All these functions are meta data related.
 
-import "gitlab.ricebook.net/platform/core/types"
+import (
+	log "github.com/Sirupsen/logrus"
+	"gitlab.ricebook.net/platform/core/types"
+)
 
 func (c *calcium) ListPods() ([]*types.Pod, error) {
 	return c.store.GetAllPods()
@@ -50,14 +53,15 @@ func (c *calcium) ListPodNodes(podname string, all bool) ([]*types.Node, error)
 	var nodes []*types.Node
 	candidates, err := c.store.GetNodesByPod(podname)
 	if err != nil {
+		log.Debugf("Error during ListPodNodes from %s: %v", podname, err)
 		return nodes, err
 	}
 	for _, candidate := range candidates {
 		if candidate.Available || all {
 			nodes = append(nodes, candidate)
 		}
 	}
-	return nodes, err
+	return nodes, nil
 }
 
 func (c *calcium) GetContainer(id string) (*types.Container, error) {

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)