RE: ARCHITECTURE

application/service/executor/executor.go

@@ -0,0 +1,41 @@
+package executor
+
+import (
+	"context"
+	"github.com/duck8823/duci/application"
+	"github.com/duck8823/duci/application/semaphore"
+	"github.com/duck8823/duci/domain/model/docker"
+	"github.com/duck8823/duci/domain/service/runner"
+	"github.com/labstack/gommon/random"
+)
+
+type JobExecutor struct {
+	runner.DockerRunner
+	StartFunc func(context.Context)
+	EndFunc   func(context.Context, error)
+}
+
+// Execute job
+func (r *JobExecutor) Execute(ctx context.Context, dir string, cmd ...string) error {
+	r.StartFunc(ctx)
+
+	errs := make(chan error, 1)
+
+	timeout, cancel := context.WithTimeout(ctx, application.Config.Timeout())
+	defer cancel()
+
+	go func() {
+		semaphore.Acquire()
+		errs <- r.DockerRunner.Run(timeout, dir, docker.Tag(random.String(16, random.Lowercase)), cmd)
+		semaphore.Release()
+	}()
+
+	select {
+	case <-timeout.Done():
+		r.EndFunc(ctx, timeout.Err())
+		return timeout.Err()
+	case err := <-errs:
+		r.EndFunc(ctx, err)
+		return err
+	}
+}

application/service/runner/helper.go

@@ -0,0 +1,62 @@
+package runner
+
+import (
+	"bytes"
+	"github.com/duck8823/duci/application/service/docker"
+	"github.com/duck8823/duci/infrastructure/archive/tar"
+	"github.com/pkg/errors"
+	"gopkg.in/yaml.v2"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+)
+
+// createTarball creates a tar archive
+func createTarball(workDir string) (*os.File, error) {
+	tarFilePath := filepath.Join(workDir, "duci.tar")
+	writeFile, err := os.OpenFile(tarFilePath, os.O_RDWR|os.O_CREATE, 0600)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+	defer writeFile.Close()
+
+	if err := tar.Create(workDir, writeFile); err != nil {
+		return nil, errors.WithStack(err)
+	}
+
+	readFile, _ := os.Open(tarFilePath)
+	return readFile, nil
+}
+
+// dockerfilePath returns a path to dockerfile for duci using
+func dockerfilePath(workDir string) docker.Dockerfile {
+	dockerfile := "./Dockerfile"
+	if exists(filepath.Join(workDir, ".duci/Dockerfile")) {
+		dockerfile = ".duci/Dockerfile"
+	}
+	return docker.Dockerfile(dockerfile)
+}
+
+// exists indicates whether the file exists
+func exists(name string) bool {
+	_, err := os.Stat(name)
+	return !os.IsNotExist(err)
+}
+
+// runtimeOptions parses a config.yml and returns a docker runtime options
+func runtimeOptions(workDir string) (docker.RuntimeOptions, error) {
+	var opts docker.RuntimeOptions
+
+	if !exists(filepath.Join(workDir, ".duci/config.yml")) {
+		return opts, nil
+	}
+	content, err := ioutil.ReadFile(filepath.Join(workDir, ".duci/config.yml"))
+	if err != nil {
+		return opts, errors.WithStack(err)
+	}
+	content = []byte(os.ExpandEnv(string(content)))
+	if err := yaml.NewDecoder(bytes.NewReader(content)).Decode(&opts); err != nil {
+		return opts, errors.WithStack(err)
+	}
+	return opts, nil
+}

application/service/runner/runner.go

@@ -10,7 +10,6 @@ import (
 	"github.com/duck8823/duci/application/service/github"
 	"github.com/duck8823/duci/application/service/logstore"
 	"github.com/duck8823/duci/data/model"
-	"github.com/duck8823/duci/infrastructure/archive/tar"
 	"github.com/duck8823/duci/infrastructure/logger"
 	"github.com/labstack/gommon/random"
 	"github.com/pkg/errors"
@@ -116,30 +115,6 @@ func (r *DockerRunner) dockerBuild(ctx context.Context, dir string, repo github.
 	return nil
 }
 
-func createTarball(workDir string) (*os.File, error) {
-	tarFilePath := filepath.Join(workDir, "duci.tar")
-	writeFile, err := os.OpenFile(tarFilePath, os.O_RDWR|os.O_CREATE, 0600)
-	if err != nil {
-		return nil, errors.WithStack(err)
-	}
-	defer writeFile.Close()
-
-	if err := tar.Create(workDir, writeFile); err != nil {
-		return nil, errors.WithStack(err)
-	}
-
-	readFile, _ := os.Open(tarFilePath)
-	return readFile, nil
-}
-
-func dockerfilePath(workDir string) docker.Dockerfile {
-	dockerfile := "./Dockerfile"
-	if exists(filepath.Join(workDir, ".duci/Dockerfile")) {
-		dockerfile = ".duci/Dockerfile"
-	}
-	return docker.Dockerfile(dockerfile)
-}
-
 func (r *DockerRunner) dockerRun(ctx context.Context, dir string, repo github.Repository, cmd docker.Command) (docker.ContainerID, error) {
 	opts, err := runtimeOpts(dir)
 	if err != nil {
@@ -211,8 +186,3 @@ func (r *DockerRunner) finish(ctx context.Context, src *github.TargetSource, err
 	}
 	r.LogStore.Finish(ctx.UUID())
 }
-
-func exists(name string) bool {
-	_, err := os.Stat(name)
-	return !os.IsNotExist(err)
-}

application/service/runner/runner_test.go

@@ -768,6 +768,9 @@ func TestRunnerImpl_Run_NonNormal(t *testing.T) {
 
 		// and
 		application.Config.Job.Timeout = 1
+		defer func() {
+			application.Config.Job.Timeout = 600
+		}()
 
 		mockDocker := mock_docker.NewMockService(ctrl)
 		mockDocker.EXPECT().

domain/model/docker/log.go

@@ -0,0 +1,92 @@
+package docker
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"github.com/duck8823/duci/domain/model/job"
+	"github.com/pkg/errors"
+	"io"
+	"time"
+)
+
+var now = time.Now
+
+type buildLogger struct {
+	reader *bufio.Reader
+}
+
+// NewBuildLog return a instance of Log.
+func NewBuildLog(r io.Reader) *buildLogger {
+	return &buildLogger{bufio.NewReader(r)}
+}
+
+// ReadLine returns LogLine.
+func (l *buildLogger) ReadLine() (*job.LogLine, error) {
+	for {
+		line, _, readErr := l.reader.ReadLine()
+		msg := extractMessage(line)
+		if readErr == io.EOF {
+			return &job.LogLine{Timestamp: now(), Message: msg}, readErr
+		}
+		if readErr != nil {
+			return nil, errors.WithStack(readErr)
+		}
+
+		if len(msg) == 0 {
+			continue
+		}
+
+		return &job.LogLine{Timestamp: now(), Message: msg}, readErr
+	}
+}
+
+type runLogger struct {
+	reader *bufio.Reader
+}
+
+// NewRunLog returns a instance of Log
+func NewRunLog(r io.Reader) *runLogger {
+	return &runLogger{bufio.NewReader(r)}
+}
+
+// ReadLine returns LogLine.
+func (l *runLogger) ReadLine() (*job.LogLine, error) {
+	for {
+		line, _, readErr := l.reader.ReadLine()
+		if readErr != nil && readErr != io.EOF {
+			return nil, errors.WithStack(readErr)
+		}
+
+		messages, err := trimPrefix(line)
+		if err != nil {
+			return nil, errors.WithStack(err)
+		}
+
+		// prevent to CR
+		progress := bytes.Split(messages, []byte{'\r'})
+		return &job.LogLine{Timestamp: now(), Message: string(progress[0])}, readErr
+	}
+}
+
+func extractMessage(line []byte) string {
+	s := &struct {
+		Stream string `json:"stream"`
+	}{}
+	json.NewDecoder(bytes.NewReader(line)).Decode(s)
+	return s.Stream
+}
+
+func trimPrefix(line []byte) ([]byte, error) {
+	if len(line) < 8 {
+		return []byte{}, nil
+	}
+
+	// detect logstore prefix
+	// see https://godoc.org/github.com/docker/docker/client#Client.ContainerLogs
+	if !((line[0] == 1 || line[0] == 2) && (line[1] == 0 && line[2] == 0 && line[3] == 0)) {
+		return nil, fmt.Errorf("invalid logstore prefix: %+v", line[:7])
+	}
+	return line[8:], nil
+}

domain/model/docker/options.go

@@ -0,0 +1,37 @@
+package docker
+
+import (
+	"fmt"
+	"strings"
+)
+
+// RuntimeOptions is a docker options.
+type RuntimeOptions struct {
+	Environments Environments
+	Volumes      Volumes
+}
+
+// Environments represents a docker `-e` option.
+type Environments map[string]interface{}
+
+// ToArray returns string array of environments
+func (e Environments) ToArray() []string {
+	var a []string
+	for key, val := range e {
+		a = append(a, fmt.Sprintf("%s=%v", key, val))
+	}
+	return a
+}
+
+// Volumes represents a docker `-v` option.
+type Volumes []string
+
+// ToMap returns map of volumes.
+func (v Volumes) ToMap() map[string]struct{} {
+	m := make(map[string]struct{})
+	for _, volume := range v {
+		key := strings.Split(volume, ":")[0]
+		m[key] = struct{}{}
+	}
+	return m
+}

domain/model/docker/types.go

@@ -0,0 +1,41 @@
+package docker
+
+// Tag describes a docker tag
+type Tag string
+
+// ToString return string value
+func (t Tag) ToString() string {
+	return string(t)
+}
+
+// Command describes a docker CMD
+type Command []string
+
+// ToSlice returns slice values
+func (c Command) ToSlice() []string {
+	return []string(c)
+}
+
+// Dockerfile represents a path to dockerfile
+type Dockerfile string
+
+// ToString returns string value
+func (d Dockerfile) ToString() string {
+	return string(d)
+}
+
+// ContainerID describes a container id of docker
+type ContainerID string
+
+// ToString returns string value
+func (c ContainerID) ToString() string {
+	return string(c)
+}
+
+// ExitCode describes a exit code
+type ExitCode int64
+
+// IsFailure returns whether failure code or not
+func (c ExitCode) IsFailure() bool {
+	return c != 1
+}

domain/model/job/job.go

@@ -0,0 +1,50 @@
+package job
+
+import (
+	"bytes"
+	"encoding/json"
+	"github.com/google/uuid"
+	"github.com/pkg/errors"
+)
+
+// Job represents a task
+type Job struct {
+	Finished bool      `json:"finished"`
+	Stream   []LogLine `json:"stream"`
+}
+
+// AppendLog append log line to stream
+func (j *Job) AppendLog(line LogLine) {
+	j.Stream = append(j.Stream, line)
+}
+
+// Finish set true to Finished
+func (j *Job) Finish() {
+	j.Finished = true
+}
+
+// ToBytes returns marshal byte slice
+func (j *Job) ToBytes() ([]byte, error) {
+	data, err := json.Marshal(j)
+	if err != nil {
+		return nil, errors.WithStack(err)
+	}
+	return data, nil
+}
+
+// NewJob returns unmarshal Job instance
+func NewJob(data []byte) (*Job, error) {
+	job := &Job{}
+	if err := json.NewDecoder(bytes.NewReader(data)).Decode(job); err != nil {
+		return nil, errors.WithStack(err)
+	}
+	return job, nil
+}
+
+// ID is the identifier of job
+type ID uuid.UUID
+
+// ToSlice returns slice value
+func (i ID) ToSlice() []byte {
+	return []byte(i[:])
+}