Skip to content

Use worker pool for parallel jobs #4

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Robert-Furth wants to merge 11 commits into
base: master
Choose a base branch
from
Open

Use worker pool for parallel jobs #4

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
54711cc
Stop job containers once they finish running a job
Robert-Furth Jul 20, 2021
20215fc
Give each workflow run a UUID
Robert-Furth Jul 20, 2021
abbe7b5
Do not update remote actions once they are cached
Robert-Furth Jul 20, 2021
4e5b24f
Merge branch 'master' of https://github.com/Robert-Furth/act
Robert-Furth Jul 21, 2021
0672119
Artifact support (#1)
Robert-Furth Aug 23, 2021
7eb5279
Merge branch 'nektos:master' into master
Robert-Furth Aug 23, 2021
59c7adf
Put scripts in /tmp instead of the working directory (#2)
Robert-Furth Aug 27, 2021
e83b460
Make matrix 'include' section work as intented
Robert-Furth Sep 6, 2021
9549f7c
Evaluate checkout ref when determining if a checkout is local
Robert-Furth Sep 8, 2021
3ef3420
Security fix: do not bind /var/run/docker.sock
Robert-Furth Sep 9, 2021
4592601
Use worker pool approach for parallel jobs
Robert-Furth Sep 14, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions cmd/root.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import (

"github.com/AlecAivazis/survey/v2"
"github.com/andreaskoch/go-fswatch"
"github.com/google/uuid"
"github.com/joho/godotenv"
"github.com/mitchellh/go-homedir"
gitignore "github.com/sabhiram/go-gitignore"
Expand Down Expand Up @@ -250,6 +251,9 @@ func newRunCommand(ctx context.Context, input *Input) func(*cobra.Command, []str
}
}

// Generate a new UUID
runid := strings.ReplaceAll(uuid.New().String(), "-", "")

// run the plan
config := &runner.Config{
Actor: input.actor,
Expand All @@ -274,6 +278,7 @@ func newRunCommand(ctx context.Context, input *Input) func(*cobra.Command, []str
ContainerCapAdd: input.containerCapAdd,
ContainerCapDrop: input.containerCapDrop,
AutoRemove: input.autoRemove,
RunID: runid,
}
r, err := runner.New(config)
if err != nil {
Expand Down
1 change: 1 addition & 0 deletions go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ require (
github.com/go-git/go-git/v5 v5.2.0
github.com/go-ini/ini v1.62.0
github.com/golang/protobuf v1.4.3 // indirect
github.com/google/uuid v1.3.0 // indirect
github.com/gopherjs/gopherjs v0.0.0-20181103185306-d547d1d9531e // indirect
github.com/imdario/mergo v0.3.11 // indirect
github.com/joho/godotenv v1.3.0
Expand Down
2 changes: 2 additions & 0 deletions go.sum
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -466,6 +466,8 @@ github.com/google/subcommands v1.0.1/go.mod h1:ZjhPrFU+Olkh9WazFPsl27BQ4UPiG37m3
github.com/google/uuid v1.0.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.1.2/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I=
github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/google/wire v0.3.0/go.mod h1:i1DMg/Lu8Sz5yYl25iOdmc5CT5qusaa+zmRWs16741s=
github.com/google/wire v0.4.0/go.mod h1:ngWDr9Qvq3yZA10YrxfyGELY/AFWGVpy9c1LTRi1EoU=
github.com/googleapis/gax-go v2.0.0+incompatible/go.mod h1:SFVmujtThgffbyetf+mdk2eWhX2bMyUtNHzFKcPA9HY=
Expand Down
22 changes: 16 additions & 6 deletions pkg/common/executor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,20 +90,30 @@ func NewErrorExecutor(err error) Executor {
}
}

func parallelExecutorWorker(executorChan <-chan Executor, errChan chan error, ctx *context.Context) {
for executor := range executorChan {
err := executor.ChannelError(errChan)(*ctx)
if err != nil {
log.Fatal(err)
}
}
}

// NewParallelExecutor creates a new executor from a parallel of other executors
func NewParallelExecutor(executors ...Executor) Executor {
return func(ctx context.Context) error {
maxJobs := 4
errChan := make(chan error)
executorChan := make(chan Executor, len(executors))

for i := 0; i < maxJobs; i++ {
go parallelExecutorWorker(executorChan, errChan, &ctx)
}
for _, executor := range executors {
e := executor
go func() {
err := e.ChannelError(errChan)(ctx)
if err != nil {
log.Fatal(err)
}
}()
executorChan <- e
}
close(executorChan)

// Executor waits all executors to cleanup these resources.
var firstErr error
Expand Down
5 changes: 5 additions & 0 deletions pkg/common/git.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -285,6 +285,11 @@ func NewGitCloneExecutor(input NewGitCloneExecutorInput) Executor {
defer cloneLock.Unlock()

refName := plumbing.ReferenceName(fmt.Sprintf("refs/heads/%s", input.Ref))

if _, err := git.PlainOpen(input.Dir); err == nil {
return nil
}

r, err := CloneIfRequired(ctx, refName, input, logger)
if err != nil {
return err
Expand Down
41 changes: 27 additions & 14 deletions pkg/model/workflow.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -226,21 +226,11 @@ func (j *Job) GetMatrixes() []map[string]interface{} {
case []interface{}:
for _, i := range t {
i := i.(map[string]interface{})
for k := range i {
if _, ok := m[k]; ok {
includes = append(includes, i)
break
}
}
includes = append(includes, i)
}
case interface{}:
v := v.(map[string]interface{})
for k := range v {
if _, ok := m[k]; ok {
includes = append(includes, v)
break
}
}
includes = append(includes, v)
}
}
delete(m, "include")
Expand Down Expand Up @@ -271,8 +261,20 @@ func (j *Job) GetMatrixes() []map[string]interface{} {
matrixes = append(matrixes, matrix)
}
for _, include := range includes {
log.Debugf("Adding include '%v'", include)
matrixes = append(matrixes, include)
notAdded := true
for _, matrix := range matrixProduct {
if hasKeysInCommon(matrix, include) && commonKeysMatch(matrix, include) {
log.Debugf("Applying include '%v' to matrix entry '%v'", include, matrix)
for k, v := range include {
matrix[k] = v
}
notAdded = false
}
}
if notAdded {
log.Debugf("Appending include '%v' to matrix", include)
matrixes = append(matrixes, include)
}
}
} else {
matrixes = append(matrixes, make(map[string]interface{}))
Expand All @@ -292,6 +294,17 @@ func commonKeysMatch(a map[string]interface{}, b map[string]interface{}) bool {
return true
}

func hasKeysInCommon(a map[string]interface{}, b map[string]interface{}) bool {
for aKey := range a {
for bKey := range b {
if aKey == bKey {
return true
}
}
}
return false
}

// ContainerSpec is the specification of the container to use for the job
type ContainerSpec struct {
Image string `yaml:"image"`
Expand Down
42 changes: 19 additions & 23 deletions pkg/runner/run_context.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,7 +62,8 @@ func (rc *RunContext) GetEnv() map[string]string {
}

func (rc *RunContext) jobContainerName() string {
return createContainerName("act", rc.String())
name := createContainerName("act", rc.Config.RunID, rc.String())
return name
}

// Returns the binds and mounts for the container, resolving paths as appopriate
Expand All @@ -74,11 +75,15 @@ func (rc *RunContext) GetBindsAndMounts() ([]string, map[string]string) {
}

binds := []string{
fmt.Sprintf("%s:%s", rc.Config.ContainerDaemonSocket, "/var/run/docker.sock"),
// ***DO NOT BIND '/var/run/docker.sock'!*** If you do, then you're giving the runner image, running code
// scraped from the internet, access to docker ON THE HOST MACHINE.
// (I've already come across a workflow that runs 'docker image prune -af'. That was a nasty shock. --Robert)
// fmt.Sprintf("%s:%s", rc.Config.ContainerDaemonSocket, "/var/run/docker.sock"),
}

mounts := map[string]string{
"act-toolcache": "/toolcache",
"act-artifacts-" + rc.Config.RunID: "/artifacts",
}

if rc.Config.BindWorkdir {
Expand Down Expand Up @@ -224,7 +229,9 @@ func (rc *RunContext) Executor() common.Executor {
}
steps = append(steps, rc.stopJobContainer())

return common.NewPipelineExecutor(steps...).If(rc.isEnabled)
return common.NewPipelineExecutor(steps...).
Finally(rc.stopJobContainer().IfBool(rc.Config.AutoRemove)).
If(rc.isEnabled)
}

func (rc *RunContext) newStepExecutor(step *model.Step) common.Executor {
Expand Down Expand Up @@ -403,22 +410,8 @@ func mergeMaps(maps ...map[string]string) map[string]string {
func createContainerName(parts ...string) string {
name := make([]string, 0)
pattern := regexp.MustCompile("[^a-zA-Z0-9]")
partLen := (30 / len(parts)) - 1
for i, part := range parts {
if i == len(parts)-1 {
name = append(name, pattern.ReplaceAllString(part, "-"))
} else {
// If any part has a '-<number>' on the end it is likely part of a matrix job.
// Let's preserve the number to prevent clashes in container names.
re := regexp.MustCompile("-[0-9]+$")
num := re.FindStringSubmatch(part)
if len(num) > 0 {
name = append(name, trimToLen(pattern.ReplaceAllString(part, "-"), partLen-len(num[0])))
name = append(name, num[0])
} else {
name = append(name, trimToLen(pattern.ReplaceAllString(part, "-"), partLen))
}
}
for _, part := range parts {
name = append(name, pattern.ReplaceAllString(part, "-"))
}
return strings.ReplaceAll(strings.Trim(strings.Join(name, "-"), "-"), "--", "-")
}
Expand Down Expand Up @@ -585,7 +578,7 @@ func (rc *RunContext) getGithubContext() *githubContext {
return ghc
}

func (ghc *githubContext) isLocalCheckout(step *model.Step) bool {
func (ghc *githubContext) isLocalCheckout(step *model.Step, ee ExpressionEvaluator) bool {
if step.Type() == model.StepTypeInvalid {
// This will be errored out by the executor later, we need this here to avoid a null panic though
return false
Expand All @@ -605,8 +598,11 @@ func (ghc *githubContext) isLocalCheckout(step *model.Step) bool {
if repository, ok := step.With["repository"]; ok && repository != ghc.Repository {
return false
}
if repository, ok := step.With["ref"]; ok && repository != ghc.Ref {
return false
if ref, ok := step.With["ref"]; ok {
interp, ok2 := ee.InterpolateWithStringCheck(ref)
if ok2 && interp != ghc.Ref {
return false
}
}
return true
}
Expand Down Expand Up @@ -721,7 +717,7 @@ func (rc *RunContext) withGithubEnv(env map[string]string) map[string]string {
func (rc *RunContext) localCheckoutPath() (string, bool) {
ghContext := rc.getGithubContext()
for _, step := range rc.Run.Job().Steps {
if ghContext.isLocalCheckout(step) {
if ghContext.isLocalCheckout(step, rc.ExprEval) {
return step.With["path"], true
}
}
Expand Down
16 changes: 15 additions & 1 deletion pkg/runner/runner.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ import (
"strings"

"github.com/nektos/act/pkg/common"
"github.com/nektos/act/pkg/container"
"github.com/nektos/act/pkg/model"
log "github.com/sirupsen/logrus"
)
Expand Down Expand Up @@ -43,6 +44,7 @@ type Config struct {
ContainerCapAdd []string // list of kernel capabilities to add to the containers
ContainerCapDrop []string // list of kernel capabilities to remove from the containers
AutoRemove bool // controls if the container is automatically removed upon workflow completion
RunID string // A unique ID for the current workflow run
}

// Resolves the equivalent host path inside the container
Expand Down Expand Up @@ -149,7 +151,19 @@ func (runner *runnerImpl) NewPlanExecutor(plan *model.Plan) common.Executor {
pipeline = append(pipeline, common.NewParallelExecutor(stageExecutor...))
}

return common.NewPipelineExecutor(pipeline...)
return common.NewPipelineExecutor(pipeline...).
Finally(func(ctx context.Context) error {
if !runner.config.AutoRemove {
return nil
}
artifactVolume := "act-artifacts-" + runner.config.RunID
log.Infof("Cleaning up artifacts volume \"%s\"", artifactVolume)
err := container.NewDockerVolumeRemoveExecutor(artifactVolume, false)(ctx)
if err != nil {
log.Errorf("Error cleaning up volume \"%s\": %v", artifactVolume, err)
}
return err
})
}

func (runner *runnerImpl) newRunContext(run *model.Run, matrix map[string]interface{}) *RunContext {
Expand Down
6 changes: 3 additions & 3 deletions pkg/runner/step_context.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ func (sc *StepContext) Executor() common.Executor {
remoteAction.URL = rc.Config.GitHubInstance

github := rc.getGithubContext()
if remoteAction.IsCheckout() && github.isLocalCheckout(step) {
if remoteAction.IsCheckout() && github.isLocalCheckout(step, rc.ExprEval) {
return func(ctx context.Context) error {
common.Logger(ctx).Debugf("Skipping local actions/checkout because workdir was already copied")
return nil
Expand Down Expand Up @@ -227,7 +227,7 @@ func (sc *StepContext) setupShellCommand() common.Executor {
run = runPrepend + "\n" + run + "\n" + runAppend

log.Debugf("Wrote command '%s' to '%s'", run, scriptName)
scriptPath := fmt.Sprintf("%s/%s", rc.Config.ContainerWorkdir(), scriptName)
scriptPath := fmt.Sprintf("/tmp/%s", scriptName)

if step.Shell == "" {
step.Shell = rc.Run.Job().Defaults.Run.Shell
Expand All @@ -252,7 +252,7 @@ func (sc *StepContext) setupShellCommand() common.Executor {

sc.Cmd = finalCMD

return rc.JobContainer.Copy(rc.Config.ContainerWorkdir(), &container.FileEntry{
return rc.JobContainer.Copy("/tmp/", &container.FileEntry{
Name: scriptName,
Mode: 0755,
Body: script.String(),
Expand Down