successfully able to run simple workflows

Signed-off-by: Casey Lee <cplee@nektos.com>

pkg/container/docker_build.go

@@ -1,6 +1,7 @@
 package container
 
 import (
+	"context"
 	"io"
 	"os"
 	"path/filepath"
@@ -16,16 +17,16 @@ import (
 
 // NewDockerBuildExecutorInput the input for the NewDockerBuildExecutor function
 type NewDockerBuildExecutorInput struct {
-	DockerExecutorInput
 	ContextDir string
 	ImageTag   string
 }
 
 // NewDockerBuildExecutor function to create a run executor for the container
 func NewDockerBuildExecutor(input NewDockerBuildExecutorInput) common.Executor {
-	return func() error {
-		input.Logger.Infof("docker build -t %s %s", input.ImageTag, input.ContextDir)
-		if input.Dryrun {
+	return func(ctx context.Context) error {
+		logger := common.Logger(ctx)
+		logger.Infof("docker build -t %s %s", input.ImageTag, input.ContextDir)
+		if common.Dryrun(ctx) {
 			return nil
 		}
 
@@ -33,9 +34,9 @@ func NewDockerBuildExecutor(input NewDockerBuildExecutorInput) common.Executor {
 		if err != nil {
 			return err
 		}
-		cli.NegotiateAPIVersion(input.Ctx)
+		cli.NegotiateAPIVersion(ctx)
 
-		input.Logger.Debugf("Building image from '%v'", input.ContextDir)
+		logger.Debugf("Building image from '%v'", input.ContextDir)
 
 		tags := []string{input.ImageTag}
 		options := types.ImageBuildOptions{
@@ -49,10 +50,10 @@ func NewDockerBuildExecutor(input NewDockerBuildExecutorInput) common.Executor {
 
 		defer buildContext.Close()
 
-		input.Logger.Debugf("Creating image from context dir '%s' with tag '%s'", input.ContextDir, input.ImageTag)
-		resp, err := cli.ImageBuild(input.Ctx, buildContext, options)
+		logger.Debugf("Creating image from context dir '%s' with tag '%s'", input.ContextDir, input.ImageTag)
+		resp, err := cli.ImageBuild(ctx, buildContext, options)
 
-		err = input.logDockerResponse(resp.Body, err != nil)
+		err = logDockerResponse(logger, resp.Body, err != nil)
 		if err != nil {
 			return err
 		}

pkg/container/docker_common.go

@@ -1,115 +0,0 @@
-package container
-
-import (
-	"bufio"
-	"context"
-	"encoding/json"
-	"errors"
-	"fmt"
-	"github.com/docker/docker/pkg/stdcopy"
-	"io"
-	"os"
-
-	"github.com/sirupsen/logrus"
-)
-
-// DockerExecutorInput common input params
-type DockerExecutorInput struct {
-	Ctx    context.Context
-	Logger *logrus.Entry
-	Dryrun bool
-}
-
-type dockerMessage struct {
-	ID          string `json:"id"`
-	Stream      string `json:"stream"`
-	Error       string `json:"error"`
-	ErrorDetail struct {
-		Message string
-	}
-	Status   string `json:"status"`
-	Progress string `json:"progress"`
-}
-
-func (i *DockerExecutorInput) logDockerOutput(dockerResponse io.Reader) {
-	w := i.Logger.Writer()
-	_, err := stdcopy.StdCopy(w, w, dockerResponse)
-	if err != nil {
-		i.Logger.Error(err)
-	}
-}
-
-func (i *DockerExecutorInput) streamDockerOutput(dockerResponse io.Reader) {
-	out := os.Stdout
-	go func() {
-		<-i.Ctx.Done()
-		fmt.Println()
-	}()
-
-	_, err := io.Copy(out, dockerResponse)
-	if err != nil {
-		i.Logger.Error(err)
-	}
-}
-
-func (i *DockerExecutorInput) writeLog(isError bool, format string, args ...interface{}) {
-	if i.Logger == nil {
-		return
-	}
-	if isError {
-		i.Logger.Errorf(format, args...)
-	} else {
-		i.Logger.Debugf(format, args...)
-	}
-
-}
-
-func (i *DockerExecutorInput) logDockerResponse(dockerResponse io.ReadCloser, isError bool) error {
-	if dockerResponse == nil {
-		return nil
-	}
-	defer dockerResponse.Close()
-
-	scanner := bufio.NewScanner(dockerResponse)
-	msg := dockerMessage{}
-
-	for scanner.Scan() {
-		line := scanner.Bytes()
-
-		msg.ID = ""
-		msg.Stream = ""
-		msg.Error = ""
-		msg.ErrorDetail.Message = ""
-		msg.Status = ""
-		msg.Progress = ""
-
-		if err := json.Unmarshal(line, &msg); err != nil {
-			i.writeLog(false, "Unable to unmarshal line [%s] ==> %v", string(line), err)
-			continue
-		}
-
-		if msg.Error != "" {
-			i.writeLog(isError, "%s", msg.Error)
-			return errors.New(msg.Error)
-		}
-
-		if msg.ErrorDetail.Message != "" {
-			i.writeLog(isError, "%s", msg.ErrorDetail.Message)
-			return errors.New(msg.Error)
-		}
-
-		if msg.Status != "" {
-			if msg.Progress != "" {
-				i.writeLog(isError, "%s :: %s :: %s\n", msg.Status, msg.ID, msg.Progress)
-			} else {
-				i.writeLog(isError, "%s :: %s\n", msg.Status, msg.ID)
-			}
-		} else if msg.Stream != "" {
-			i.writeLog(isError, msg.Stream)
-		} else {
-			i.writeLog(false, "Unable to handle line: %s", string(line))
-		}
-	}
-
-	return nil
-}

pkg/container/docker_logger.go

@@ -0,0 +1,117 @@
+package container
+
+import (
+	"bufio"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+
+	"github.com/nektos/act/pkg/common"
+	"github.com/sirupsen/logrus"
+
+	"github.com/docker/docker/pkg/stdcopy"
+)
+
+type dockerMessage struct {
+	ID          string `json:"id"`
+	Stream      string `json:"stream"`
+	Error       string `json:"error"`
+	ErrorDetail struct {
+		Message string
+	}
+	Status   string `json:"status"`
+	Progress string `json:"progress"`
+}
+
+func logDockerOutput(ctx context.Context, dockerResponse io.Reader) {
+	logger := common.Logger(ctx)
+	if entry, ok := logger.(*logrus.Entry); ok {
+		w := entry.Writer()
+		_, err := stdcopy.StdCopy(w, w, dockerResponse)
+		if err != nil {
+			logrus.Error(err)
+		}
+	} else if lgr, ok := logger.(*logrus.Logger); ok {
+		w := lgr.Writer()
+		_, err := stdcopy.StdCopy(w, w, dockerResponse)
+		if err != nil {
+			logrus.Error(err)
+		}
+	} else {
+		logrus.Errorf("Unable to get writer from logger (type=%T)", logger)
+	}
+}
+
+func streamDockerOutput(ctx context.Context, dockerResponse io.Reader) {
+	out := os.Stdout
+	go func() {
+		<-ctx.Done()
+		fmt.Println()
+	}()
+
+	_, err := io.Copy(out, dockerResponse)
+	if err != nil {
+		logrus.Error(err)
+	}
+}
+
+func logDockerResponse(logger logrus.FieldLogger, dockerResponse io.ReadCloser, isError bool) error {
+	if dockerResponse == nil {
+		return nil
+	}
+	defer dockerResponse.Close()
+
+	scanner := bufio.NewScanner(dockerResponse)
+	msg := dockerMessage{}
+
+	for scanner.Scan() {
+		line := scanner.Bytes()
+
+		msg.ID = ""
+		msg.Stream = ""
+		msg.Error = ""
+		msg.ErrorDetail.Message = ""
+		msg.Status = ""
+		msg.Progress = ""
+
+		if err := json.Unmarshal(line, &msg); err != nil {
+			writeLog(logger, false, "Unable to unmarshal line [%s] ==> %v", string(line), err)
+			continue
+		}
+
+		if msg.Error != "" {
+			writeLog(logger, isError, "%s", msg.Error)
+			return errors.New(msg.Error)
+		}
+
+		if msg.ErrorDetail.Message != "" {
+			writeLog(logger, isError, "%s", msg.ErrorDetail.Message)
+			return errors.New(msg.Error)
+		}
+
+		if msg.Status != "" {
+			if msg.Progress != "" {
+				writeLog(logger, isError, "%s :: %s :: %s\n", msg.Status, msg.ID, msg.Progress)
+			} else {
+				writeLog(logger, isError, "%s :: %s\n", msg.Status, msg.ID)
+			}
+		} else if msg.Stream != "" {
+			writeLog(logger, isError, msg.Stream)
+		} else {
+			writeLog(logger, false, "Unable to handle line: %s", string(line))
+		}
+	}
+
+	return nil
+}
+
+func writeLog(logger logrus.FieldLogger, isError bool, format string, args ...interface{}) {
+	if isError {
+		logger.Errorf(format, args...)
+	} else {
+		logger.Debugf(format, args...)
+	}
+}

pkg/container/docker_pull.go

@@ -1,40 +1,60 @@
 package container
 
 import (
+	"context"
 	"fmt"
 	"strings"
 
 	"github.com/docker/docker/api/types"
 	"github.com/docker/docker/client"
 	"github.com/nektos/act/pkg/common"
+	log "github.com/sirupsen/logrus"
 )
 
 // NewDockerPullExecutorInput the input for the NewDockerPullExecutor function
 type NewDockerPullExecutorInput struct {
-	DockerExecutorInput
-	Image string
+	Image     string
+	ForcePull bool
 }
 
 // NewDockerPullExecutor function to create a run executor for the container
 func NewDockerPullExecutor(input NewDockerPullExecutorInput) common.Executor {
-	return func() error {
-		input.Logger.Infof("docker pull %v", input.Image)
+	return func(ctx context.Context) error {
+		logger := common.Logger(ctx)
+		logger.Infof("docker pull %v", input.Image)
 
-		if input.Dryrun {
+		if common.Dryrun(ctx) {
+			return nil
+		}
+
+		pull := input.ForcePull
+		if !pull {
+			imageExists, err := ImageExistsLocally(ctx, input.Image)
+			log.Debugf("Image exists? %v", imageExists)
+			if err != nil {
+				return fmt.Errorf("unable to determine if image already exists for image %q", input.Image)
+			}
+
+			if !imageExists {
+				pull = true
+			}
+		}
+
+		if !pull {
 			return nil
 		}
 
 		imageRef := cleanImage(input.Image)
-		input.Logger.Debugf("pulling image '%v'", imageRef)
+		logger.Debugf("pulling image '%v'", imageRef)
 
 		cli, err := client.NewClientWithOpts(client.FromEnv)
 		if err != nil {
 			return err
 		}
-		cli.NegotiateAPIVersion(input.Ctx)
+		cli.NegotiateAPIVersion(ctx)
 
-		reader, err := cli.ImagePull(input.Ctx, imageRef, types.ImagePullOptions{})
-		_ = input.logDockerResponse(reader, err != nil)
+		reader, err := cli.ImagePull(ctx, imageRef, types.ImagePullOptions{})
+		_ = logDockerResponse(logger, reader, err != nil)
 		if err != nil {
 			return err
 		}

pkg/container/docker_run.go

@@ -10,12 +10,12 @@ import (
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/client"
 	"github.com/nektos/act/pkg/common"
+	"github.com/pkg/errors"
 	"golang.org/x/crypto/ssh/terminal"
 )
 
 // NewDockerRunExecutorInput the input for the NewDockerRunExecutor function
 type NewDockerRunExecutorInput struct {
-	DockerExecutorInput
 	Image           string
 	Entrypoint      []string
 	Cmd             []string
@@ -30,182 +30,198 @@ type NewDockerRunExecutorInput struct {
 
 // NewDockerRunExecutor function to create a run executor for the container
 func NewDockerRunExecutor(input NewDockerRunExecutorInput) common.Executor {
-	return func() error {
+	cr := new(containerReference)
+	cr.input = input
 
-		input.Logger.Infof("docker run image=%s entrypoint=%+q cmd=%+q", input.Image, input.Entrypoint, input.Cmd)
-		if input.Dryrun {
+	return common.
+		NewInfoExecutor("docker run image=%s entrypoint=%+q cmd=%+q", input.Image, input.Entrypoint, input.Cmd).
+		Then(
+			common.NewPipelineExecutor(
+				cr.connect(),
+				cr.find(),
+				cr.remove().IfBool(!input.ReuseContainers),
+				cr.create(),
+				cr.copyContent(),
+				cr.attach(),
+				cr.start(),
+				cr.wait(),
+			).Finally(
+				cr.remove().IfBool(!input.ReuseContainers),
+			).IfNot(common.Dryrun),
+		)
+}
+
+type containerReference struct {
+	input NewDockerRunExecutorInput
+	cli   *client.Client
+	id    string
+}
+
+func (cr *containerReference) connect() common.Executor {
+	return func(ctx context.Context) error {
+		cli, err := client.NewClientWithOpts(client.FromEnv)
+		if err != nil {
+			return errors.WithStack(err)
+		}
+		cli.NegotiateAPIVersion(ctx)
+		cr.cli = cli
+		return nil
+	}
+}
+
+func (cr *containerReference) find() common.Executor {
+	return func(ctx context.Context) error {
+		containers, err := cr.cli.ContainerList(ctx, types.ContainerListOptions{
+			All: true,
+		})
+		if err != nil {
+			return errors.WithStack(err)
+		}
+
+		for _, container := range containers {
+			for _, name := range container.Names {
+				if name[1:] == cr.input.Name {
+					cr.id = container.ID
+					return nil
+				}
+			}
+		}
+
+		cr.id = ""
+		return nil
+	}
+}
+
+func (cr *containerReference) remove() common.Executor {
+	return func(ctx context.Context) error {
+		if cr.id == "" {
 			return nil
 		}
 
-		cli, err := client.NewClientWithOpts(client.FromEnv)
+		logger := common.Logger(ctx)
+		err := cr.cli.ContainerRemove(context.Background(), cr.id, types.ContainerRemoveOptions{
+			RemoveVolumes: true,
+			Force:         true,
+		})
 		if err != nil {
-			return err
+			return errors.WithStack(err)
 		}
-		cli.NegotiateAPIVersion(input.Ctx)
+		cr.id = ""
 
-		// check if container exists
-		containerID, err := findContainer(input, cli, input.Name)
-		if err != nil {
-			return err
-		}
-
-		// if we have an old container and we aren't reusing, remove it!
-		if !input.ReuseContainers && containerID != "" {
-			input.Logger.Debugf("Found existing container for %s...removing", input.Name)
-			removeContainer(input, cli, containerID)
-			containerID = ""
-		}
-
-		// create a new container if we don't have one to reuse
-		if containerID == "" {
-			containerID, err = createContainer(input, cli)
-			if err != nil {
-				return err
-			}
-		}
-
-		// be sure to cleanup container if we aren't reusing
-		if !input.ReuseContainers {
-			defer removeContainer(input, cli, containerID)
-		}
-
-		executor := common.NewPipelineExecutor(
-			func() error {
-				return copyContentToContainer(input, cli, containerID)
-			}, func() error {
-				return attachContainer(input, cli, containerID)
-			}, func() error {
-				return startContainer(input, cli, containerID)
-			}, func() error {
-				return waitContainer(input, cli, containerID)
-			},
-		)
-		return executor()
-	}
-
-}
-
-func createContainer(input NewDockerRunExecutorInput, cli *client.Client) (string, error) {
-	isTerminal := terminal.IsTerminal(int(os.Stdout.Fd()))
-
-	config := &container.Config{
-		Image:      input.Image,
-		Cmd:        input.Cmd,
-		Entrypoint: input.Entrypoint,
-		WorkingDir: input.WorkingDir,
-		Env:        input.Env,
-		Tty:        isTerminal,
-	}
-
-	if len(input.Volumes) > 0 {
-		config.Volumes = make(map[string]struct{})
-		for _, vol := range input.Volumes {
-			config.Volumes[vol] = struct{}{}
-		}
-	}
-
-	resp, err := cli.ContainerCreate(input.Ctx, config, &container.HostConfig{
-		Binds: input.Binds,
-	}, nil, input.Name)
-	if err != nil {
-		return "", err
-	}
-	input.Logger.Debugf("Created container name=%s id=%v from image %v", input.Name, resp.ID, input.Image)
-	input.Logger.Debugf("ENV ==> %v", input.Env)
-
-	return resp.ID, nil
-}
-
-func findContainer(input NewDockerRunExecutorInput, cli *client.Client, containerName string) (string, error) {
-	containers, err := cli.ContainerList(input.Ctx, types.ContainerListOptions{
-		All: true,
-	})
-	if err != nil {
-		return "", err
-	}
-
-	for _, container := range containers {
-		for _, name := range container.Names {
-			if name[1:] == containerName {
-				return container.ID, nil
-			}
-		}
-	}
-
-	return "", nil
-}
-
-func removeContainer(input NewDockerRunExecutorInput, cli *client.Client, containerID string) {
-	err := cli.ContainerRemove(context.Background(), containerID, types.ContainerRemoveOptions{
-		RemoveVolumes: true,
-		Force:         true,
-	})
-	if err != nil {
-		input.Logger.Errorf("%v", err)
-	}
-
-	input.Logger.Debugf("Removed container: %v", containerID)
-}
-
-func copyContentToContainer(input NewDockerRunExecutorInput, cli *client.Client, containerID string) error {
-	for dstPath, srcReader := range input.Content {
-		input.Logger.Debugf("Extracting content to '%s'", dstPath)
-		err := cli.CopyToContainer(input.Ctx, containerID, dstPath, srcReader, types.CopyToContainerOptions{})
-		if err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func attachContainer(input NewDockerRunExecutorInput, cli *client.Client, containerID string) error {
-	out, err := cli.ContainerAttach(input.Ctx, containerID, types.ContainerAttachOptions{
-		Stream: true,
-		Stdout: true,
-		Stderr: true,
-	})
-	if err != nil {
-		return err
-	}
-	isTerminal := terminal.IsTerminal(int(os.Stdout.Fd()))
-	if !isTerminal || os.Getenv("NORAW") != "" {
-		go input.logDockerOutput(out.Reader)
-	} else {
-		go input.streamDockerOutput(out.Reader)
-	}
-	return nil
-}
-
-func startContainer(input NewDockerRunExecutorInput, cli *client.Client, containerID string) error {
-	input.Logger.Debugf("STARTING image=%s entrypoint=%s cmd=%v", input.Image, input.Entrypoint, input.Cmd)
-
-	if err := cli.ContainerStart(input.Ctx, containerID, types.ContainerStartOptions{}); err != nil {
-		return err
-	}
-
-	input.Logger.Debugf("Started container: %v", containerID)
-	return nil
-}
-
-func waitContainer(input NewDockerRunExecutorInput, cli *client.Client, containerID string) error {
-	statusCh, errCh := cli.ContainerWait(input.Ctx, containerID, container.WaitConditionNotRunning)
-	var statusCode int64
-	select {
-	case err := <-errCh:
-		if err != nil {
-			return err
-		}
-	case status := <-statusCh:
-		statusCode = status.StatusCode
-	}
-
-	input.Logger.Debugf("Return status: %v", statusCode)
-
-	if statusCode == 0 {
+		logger.Debugf("Removed container: %v", cr.id)
 		return nil
-	} else if statusCode == 78 {
-		return fmt.Errorf("exit with `NEUTRAL`: 78")
 	}
-
-	return fmt.Errorf("exit with `FAILURE`: %v", statusCode)
+}
+
+func (cr *containerReference) create() common.Executor {
+	return func(ctx context.Context) error {
+		if cr.id != "" {
+			return nil
+		}
+		logger := common.Logger(ctx)
+		isTerminal := terminal.IsTerminal(int(os.Stdout.Fd()))
+
+		input := cr.input
+		config := &container.Config{
+			Image:      input.Image,
+			Cmd:        input.Cmd,
+			Entrypoint: input.Entrypoint,
+			WorkingDir: input.WorkingDir,
+			Env:        input.Env,
+			Tty:        isTerminal,
+		}
+
+		if len(input.Volumes) > 0 {
+			config.Volumes = make(map[string]struct{})
+			for _, vol := range input.Volumes {
+				config.Volumes[vol] = struct{}{}
+			}
+		}
+
+		resp, err := cr.cli.ContainerCreate(ctx, config, &container.HostConfig{
+			Binds: input.Binds,
+		}, nil, input.Name)
+		if err != nil {
+			return errors.WithStack(err)
+		}
+		logger.Debugf("Created container name=%s id=%v from image %v", input.Name, resp.ID, input.Image)
+		logger.Debugf("ENV ==> %v", input.Env)
+
+		cr.id = resp.ID
+		return nil
+	}
+}
+
+func (cr *containerReference) copyContent() common.Executor {
+	return func(ctx context.Context) error {
+		logger := common.Logger(ctx)
+		for dstPath, srcReader := range cr.input.Content {
+			logger.Debugf("Extracting content to '%s'", dstPath)
+			err := cr.cli.CopyToContainer(ctx, cr.id, dstPath, srcReader, types.CopyToContainerOptions{})
+			if err != nil {
+				return errors.WithStack(err)
+			}
+		}
+		return nil
+	}
+}
+
+func (cr *containerReference) attach() common.Executor {
+	return func(ctx context.Context) error {
+		out, err := cr.cli.ContainerAttach(ctx, cr.id, types.ContainerAttachOptions{
+			Stream: true,
+			Stdout: true,
+			Stderr: true,
+		})
+		if err != nil {
+			return errors.WithStack(err)
+		}
+		isTerminal := terminal.IsTerminal(int(os.Stdout.Fd()))
+		if !isTerminal || os.Getenv("NORAW") != "" {
+			go logDockerOutput(ctx, out.Reader)
+		} else {
+			go streamDockerOutput(ctx, out.Reader)
+		}
+		return nil
+	}
+}
+
+func (cr *containerReference) start() common.Executor {
+	return func(ctx context.Context) error {
+		logger := common.Logger(ctx)
+		logger.Debugf("STARTING image=%s entrypoint=%s cmd=%v", cr.input.Image, cr.input.Entrypoint, cr.input.Cmd)
+
+		if err := cr.cli.ContainerStart(ctx, cr.id, types.ContainerStartOptions{}); err != nil {
+			return errors.WithStack(err)
+		}
+
+		logger.Debugf("Started container: %v", cr.id)
+		return nil
+	}
+}
+
+func (cr *containerReference) wait() common.Executor {
+	return func(ctx context.Context) error {
+		logger := common.Logger(ctx)
+		statusCh, errCh := cr.cli.ContainerWait(ctx, cr.id, container.WaitConditionNotRunning)
+		var statusCode int64
+		select {
+		case err := <-errCh:
+			if err != nil {
+				return errors.WithStack(err)
+			}
+		case status := <-statusCh:
+			statusCode = status.StatusCode
+		}
+
+		logger.Debugf("Return status: %v", statusCode)
+
+		if statusCode == 0 {
+			return nil
+		} else if statusCode == 78 {
+			return fmt.Errorf("exit with `NEUTRAL`: 78")
+		}
+
+		return fmt.Errorf("exit with `FAILURE`: %v", statusCode)
+	}
 }

pkg/container/docker_run_test.go

@@ -30,27 +30,21 @@ func TestNewDockerRunExecutor(t *testing.T) {
 	logger.SetOutput(buf)
 	logger.SetFormatter(&rawFormatter{})
 
+	ctx := common.WithLogger(context.Background(), logger)
+
 	runner := NewDockerRunExecutor(NewDockerRunExecutorInput{
-		DockerExecutorInput: DockerExecutorInput{
-			Ctx:    context.TODO(),
-			Logger: logrus.NewEntry(logger),
-		},
 		Image: "hello-world",
 	})
 
 	puller := NewDockerPullExecutor(NewDockerPullExecutorInput{
-		DockerExecutorInput: DockerExecutorInput{
-			Ctx:    context.TODO(),
-			Logger: logrus.NewEntry(noopLogger),
-		},
 		Image: "hello-world",
 	})
 
 	pipeline := common.NewPipelineExecutor(puller, runner)
-	err := pipeline()
+	err := pipeline(ctx)
 	assert.NoError(t, err)
 
-	expected := `docker run image=hello-world entrypoint=[] cmd=[]Hello from Docker!`
+	expected := `docker pull hello-worlddocker run image=hello-world entrypoint=[] cmd=[]Hello from Docker!`
 	actual := buf.String()
 	assert.Equal(t, expected, actual[:len(expected)])
 }