<commit_before>package calcium

import (
	"bufio"
	"fmt"
	"sync"
	"time"

	log "github.com/Sirupsen/logrus"
	enginetypes "github.com/docker/docker/api/types"
	"gitlab.ricebook.net/platform/core/types"
	"gitlab.ricebook.net/platform/core/utils"
	"golang.org/x/net/context"
)

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

	// 默认给出1200秒的超时时间吧
	// 没别的地方好传了, 不如放这里好了, 不需要用的就默认0或者不传
	waitTimeout := entry.RunAndWaitTimeout
	if waitTimeout == 0 {
		waitTimeout = c.config.RunAndWaitTimeout
	}

	// 创建容器, 有问题就gg
	createChan, err := c.CreateContainer(specs, opts)
	if err != nil {
		log.Errorf("[RunAndWait] Create container error, %v", err)
		return ch, err
	}

	// 来个goroutine处理剩下的事情
	// 基本上就是, attach拿日志写到channel, 以及等待容器结束后清理资源
	go func() {
		defer close(ch)
		logsOpts := enginetypes.ContainerLogsOptions{Follow: true, ShowStdout: true, ShowStderr: true}
		wg := sync.WaitGroup{}

		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, %v", err)
				continue
			}

			// 加个task
			wg.Add(1)

			// goroutine attach日志然后处理了写回给channel
			// 日志跟task无关, 不管wg
			go func(node *types.Node, containerID string) {
				defer wg.Done()
				defer log.Infof("[RunAndWait] Container %s finished and removed", containerID[:12])
				defer c.removeContainerSync([]string{containerID})

				ctx, _ := context.WithTimeout(context.Background(), time.Duration(waitTimeout)*time.Second)
				resp, err := node.Engine.ContainerLogs(ctx, containerID, logsOpts)
				if err != nil {
					log.Errorf("[RunAndWait] Failed to get logs, %v", err)
					ch <- &types.RunAndWaitMessage{ContainerID: containerID,
						Data: []byte(fmt.Sprintf("[exitcode] unknown %v", err))}
					return
				}

				stream := utils.FuckDockerStream(resp)
				scanner := bufio.NewScanner(stream)
				for scanner.Scan() {
					data := scanner.Bytes()
					ch <- &types.RunAndWaitMessage{ContainerID: containerID, Data: data}
					log.Debugf("[RunAndWait] %s %s", containerID[:12], data)
				}
<commit_msg>fix invalid problem<commit_after>package calcium

import (
	"bufio"
	"fmt"
	"sync"
	"time"

	log "github.com/Sirupsen/logrus"
	enginetypes "github.com/docker/docker/api/types"
	"gitlab.ricebook.net/platform/core/types"
	"gitlab.ricebook.net/platform/core/utils"
	"golang.org/x/net/context"
)

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 = "journald"
	specs.Entrypoints[opts.Entrypoint] = entry

	// 默认给出1200秒的超时时间吧
	// 没别的地方好传了, 不如放这里好了, 不需要用的就默认0或者不传
	waitTimeout := entry.RunAndWaitTimeout
	if waitTimeout == 0 {
		waitTimeout = c.config.RunAndWaitTimeout
	}

	// 创建容器, 有问题就gg
	createChan, err := c.CreateContainer(specs, opts)
	if err != nil {
		log.Errorf("[RunAndWait] Create container error, %v", err)
		return ch, err
	}

	// 来个goroutine处理剩下的事情
	// 基本上就是, attach拿日志写到channel, 以及等待容器结束后清理资源
	go func() {
		defer close(ch)
		logsOpts := enginetypes.ContainerLogsOptions{Follow: true, ShowStdout: true, ShowStderr: true}
		wg := sync.WaitGroup{}

		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, %v", err)
				continue
			}

			// 加个task
			wg.Add(1)

			// goroutine attach日志然后处理了写回给channel
			// 日志跟task无关, 不管wg
			go func(node *types.Node, containerID string) {
				defer wg.Done()
				defer log.Infof("[RunAndWait] Container %s finished and removed", containerID[:12])
				defer c.removeContainerSync([]string{containerID})

				ctx, cancel := context.WithTimeout(context.Background(), time.Duration(waitTimeout)*time.Second)
				defer cancel()
				resp, err := node.Engine.ContainerLogs(ctx, containerID, logsOpts)
				if err != nil {
					log.Errorf("[RunAndWait] Failed to get logs, %v", err)
					ch <- &types.RunAndWaitMessage{ContainerID: containerID,
						Data: []byte(fmt.Sprintf("[exitcode] unknown %v", err))}
					return
				}

				stream := utils.FuckDockerStream(resp)
				scanner := bufio.NewScanner(stream)
				for scanner.Scan() {
					data := scanner.Bytes()
					ch <- &types.RunAndWaitMessage{ContainerID: containerID, Data: data}
					log.Debugf("[RunAndWait] %s %s", containerID[:12], data)
				}

{'fraction_non_alphanumeric': 0.13834146341463416, 'fraction_numerical': 0.004292682926829269, 'mean_word_length': 3.799625468164794, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '903609', 'n_tokens_mistral': 2278, 'n_tokens_neox': 2064, 'n_words': 373}