From 28a36c490856514f0c2b0f57b891bc859861ec01 Mon Sep 17 00:00:00 2001 From: Jose Diaz-Gonzalez Date: Sat, 21 Nov 2020 17:31:13 -0500 Subject: [PATCH] feat: add proper parallel support --- docs/deployment/process-management.md | 20 ++--- plugins/ps/ps.go | 101 ++++++++++++++++++++-- plugins/ps/src/commands/commands.go | 8 +- plugins/ps/src/subcommands/subcommands.go | 12 +-- plugins/ps/subcommands.go | 16 ++-- 5 files changed, 121 insertions(+), 36 deletions(-) diff --git a/docs/deployment/process-management.md b/docs/deployment/process-management.md index 82576c672..79323ef54 100644 --- a/docs/deployment/process-management.md +++ b/docs/deployment/process-management.md @@ -4,14 +4,14 @@ ``` ps:inspect # Displays a sanitized version of docker inspect for an app -ps:rebuild [--parallel count] [--serial] [--all|] # Rebuilds an app from source +ps:rebuild [--parallel count] [--all|] # Rebuilds an app from source ps:report [] [] # Displays a process report for one or more apps -ps:restart [--parallel count] [--serial] [--all|] # Restart an app +ps:restart [--parallel count] [--all|] # Restart an app ps:restore [] # Start previously running apps e.g. after reboot ps:scale [--skip-deploy] = [=...] # Get/Set how many instances of a given process to run ps:set # Set or clear a ps property for an app -ps:start [--parallel count] [--serial] [--all|] # Start an app -ps:stop [--parallel count] [--serial] [--all|] # Stop an app +ps:start [--parallel count] [--all|] # Start an app +ps:stop [--parallel count] [--all|] # Stop an app ``` By default, Dokku will only start a single `web` process - if defined - though process scaling can be managed by the `ps` plugin or [via a custom `DOKKU_SCALE` file](/docs/deployment/process-management.md#manually-managing-process-scaling). @@ -43,10 +43,10 @@ dokku ps:rebuild node-js-app You may also rebuild all applications at once, which is useful when enabling/disabling a plugin that modifies all applications: ```shell -dokku ps:rebuildall +dokku ps:rebuild --all ``` -> The `ps:rebuild` and `ps:rebuildall` commands only work for applications for which there is a source, and thus +> The `ps:rebuild` command only works for applications for which there is a source, and thus > will only always work deterministically for Git-deployed application. Please see > the [images documentation](/docs/deployment/methods/images.md) and [tar documentation](/docs/deployment/methods/tar.md) > in for more information concerning rebuilding those applications. @@ -62,7 +62,7 @@ dokku ps:restart node-js-app You may also trigger a restart on all applications at one time: ```shell -dokku ps:restartall +dokku ps:restart --all ``` ### `ps:scale` command @@ -104,7 +104,7 @@ dokku ps:stop node-js-app You may also stop all applications at once: ```shell -dokku ps:stopall +dokku ps:stop --all ``` ### Starting applications @@ -117,12 +117,12 @@ dokku ps:start node-js-app ### Starting all applications -In some cases, it may be necessary to start all applications from scratch - e.g. if all Docker containers have been manually stopped. This can be executed via the `ps:startall` command, which supports parallelism in the same manner `ps:rebuildall`, `ps:restartall`, and `ps:stopall` do. +In some cases, it may be necessary to start all applications from scratch - e.g. if all Docker containers have been manually stopped. This can be executed via the `ps:start` command with the `--all` flag, which supports parallelism - via the `--parallel` flag - in the same manner `ps:rebuild`, `ps:restart`, and `ps:stop` do. Be aware that no action will be taken if the application containers are running. ```shell -dokku ps:startall +dokku ps:start --all ``` ## Restart policies diff --git a/plugins/ps/ps.go b/plugins/ps/ps.go index d32fa2968..1a9818943 100644 --- a/plugins/ps/ps.go +++ b/plugins/ps/ps.go @@ -2,20 +2,36 @@ package ps import ( "fmt" + "runtime" "strconv" "strings" + "sync" + "time" "github.com/dokku/dokku/plugins/common" ) // RunInSerial is the default value for whether to run a command in parallel or not // and defaults to -1 (false) -const RunInSerial = -1 +const RunInSerial = 0 // ParallelCommand is a type that declares functions // the ps plugins can execute in parallel type ParallelCommand func(string) error +// ParallelCommandRun contains all the arguments +// necessary for a parallel command run +type ParallelCommandRun struct { + AppName string +} + +// ParallelCommandResult is the result of a parallel +// command run +type ParallelCommandResult struct { + AppName string + Error error +} + var ( // DefaultProperties is a map of all valid ps properties with corresponding default property values DefaultProperties = map[string]string{ @@ -24,9 +40,18 @@ var ( ) // RunCommandAgainstAllApps runs a given ParallelCommand against all apps -func RunCommandAgainstAllApps(command ParallelCommand, commandName string, runInSerial bool, parallelCount int) error { - if runInSerial && parallelCount != RunInSerial { - common.LogWarn("Ignoring --parallel value and running in serial mode") +func RunCommandAgainstAllApps(command ParallelCommand, commandName string, parallelCount int) error { + runInSerial := false + + if parallelCount == -1 { + cpuCount := runtime.NumCPU() + common.LogWarn(fmt.Sprintf("Setting --parallel=%d value to CPU count of %d", parallelCount, cpuCount)) + parallelCount = cpuCount + } + + if parallelCount == 0 || parallelCount == 1 { + common.LogWarn(fmt.Sprintf("Running %s in serial mode", commandName)) + runInSerial = true } if runInSerial { @@ -38,9 +63,72 @@ func RunCommandAgainstAllApps(command ParallelCommand, commandName string, runIn // RunCommandAgainstAllAppsInParallel runs a given // ParallelCommand against all apps in parallel -// TODO: implement me func RunCommandAgainstAllAppsInParallel(command ParallelCommand, commandName string, parallelCount int) error { - return nil + apps, err := common.DokkuApps() + if err != nil { + common.LogWarn(err.Error()) + return nil + } + + jobs := make(chan string, parallelCount) + results := make(chan ParallelCommandResult, len(apps)) + + allocate := func() { + for _, appName := range apps { + jobs <- appName + } + close(jobs) + } + + result := func(done chan error) { + var parallelError error + errorCount := 0 + for result := range results { + if result.Error != nil { + common.LogWarn(fmt.Sprintf("Error running %s against app %s", commandName, result.AppName)) + errorCount++ + } + } + if errorCount > 0 { + parallelError = fmt.Errorf("Encountered %d errors during parallel run", errorCount) + } + done <- parallelError + } + + worker := func(wg *sync.WaitGroup, workerID int) { + for appName := range jobs { + common.LogInfo1(fmt.Sprintf("Running %s against app %s", commandName, appName)) + output := ParallelCommandResult{ + AppName: appName, + Error: command(appName), + } + results <- output + } + wg.Done() + } + + createParallelWorkerPool := func(numberOfWorkers int) { + var wg sync.WaitGroup + for i := 0; i < numberOfWorkers; i++ { + wg.Add(1) + go worker(&wg, i) + } + wg.Wait() + close(results) + } + + startTime := time.Now() + go allocate() + done := make(chan error) + go result(done) + createParallelWorkerPool(parallelCount) + err = <-done + endTime := time.Now() + diff := endTime.Sub(startTime) + + common.LogInfo2Quiet(fmt.Sprintf("Total time taken: %.2f seconds", diff.Seconds())) + + return err } // RunCommandAgainstAllAppsSerially runs a given @@ -54,6 +142,7 @@ func RunCommandAgainstAllAppsSerially(command ParallelCommand, commandName strin errorCount := 0 for _, appName := range apps { + common.LogInfo1(fmt.Sprintf("Running %s against app %s", commandName, appName)) if err = command(appName); err != nil { errorCount++ } diff --git a/plugins/ps/src/commands/commands.go b/plugins/ps/src/commands/commands.go index 7672b4082..d578af0e7 100644 --- a/plugins/ps/src/commands/commands.go +++ b/plugins/ps/src/commands/commands.go @@ -19,14 +19,14 @@ Additional commands:` helpContent = ` ps:inspect , Displays a sanitized version of docker inspect for an app - ps:rebuild [--parallel count] [--serial] [--all|], Rebuilds an app from source + ps:rebuild [--parallel count] [--all|], Rebuilds an app from source ps:report [] [], Displays a process report for one or more apps - ps:restart [--parallel count] [--serial] [--all|], Restart an app + ps:restart [--parallel count] [--all|], Restart an app ps:restore [], Start previously running apps e.g. after reboot ps:scale [--skip-deploy] = [=...], Get/Set how many instances of a given process to run ps:set , Set or clear a ps property for an app - ps:start [--parallel count] [--serial] [--all|], Start an app - ps:stop [--parallel count] [--serial] [--all|], Stop an app + ps:start [--parallel count] [--all|], Start an app + ps:stop [--parallel count] [--all|], Stop an app ` ) diff --git a/plugins/ps/src/subcommands/subcommands.go b/plugins/ps/src/subcommands/subcommands.go index a8e976129..280d08350 100644 --- a/plugins/ps/src/subcommands/subcommands.go +++ b/plugins/ps/src/subcommands/subcommands.go @@ -26,10 +26,9 @@ func main() { args := flag.NewFlagSet("ps:rebuild", flag.ExitOnError) allApps := args.Bool("all", false, "--all: restart all apps") parallelCount := args.Int("parallel", ps.RunInSerial, "--parallel: number of apps to restart in parallel, -1 to match cpu count") - runInSerial := args.Bool("serial", false, "--serial: restart apps serially") args.Parse(os.Args[2:]) appName := args.Arg(0) - err = ps.CommandRebuild(appName, *allApps, *runInSerial, *parallelCount) + err = ps.CommandRebuild(appName, *allApps, *parallelCount) case "report": args := flag.NewFlagSet("ps:report", flag.ExitOnError) args.Parse(os.Args[2:]) @@ -40,10 +39,9 @@ func main() { args := flag.NewFlagSet("ps:restart", flag.ExitOnError) allApps := args.Bool("all", false, "--all: restart all apps") parallelCount := args.Int("parallel", ps.RunInSerial, "--parallel: number of apps to restart in parallel, -1 to match cpu count") - runInSerial := args.Bool("serial", false, "--serial: restart apps serially") args.Parse(os.Args[2:]) appName := args.Arg(0) - err = ps.CommandRestart(appName, *allApps, *runInSerial, *parallelCount) + err = ps.CommandRestart(appName, *allApps, *parallelCount) case "restore": args := flag.NewFlagSet("ps:restore", flag.ExitOnError) args.Parse(os.Args[2:]) @@ -71,18 +69,16 @@ func main() { args := flag.NewFlagSet("ps:start", flag.ExitOnError) allApps := args.Bool("all", false, "--all: restart all apps") parallelCount := args.Int("parallel", ps.RunInSerial, "--parallel: number of apps to restart in parallel, -1 to match cpu count") - runInSerial := args.Bool("serial", false, "--serial: restart apps serially") args.Parse(os.Args[2:]) appName := args.Arg(0) - err = ps.CommandStart(appName, *allApps, *runInSerial, *parallelCount) + err = ps.CommandStart(appName, *allApps, *parallelCount) case "stop": args := flag.NewFlagSet("ps:stop", flag.ExitOnError) allApps := args.Bool("all", false, "--all: restart all apps") parallelCount := args.Int("parallel", ps.RunInSerial, "--parallel: number of apps to restart in parallel, -1 to match cpu count") - runInSerial := args.Bool("serial", false, "--serial: restart apps serially") args.Parse(os.Args[2:]) appName := args.Arg(0) - err = ps.CommandStop(appName, *allApps, *runInSerial, *parallelCount) + err = ps.CommandStop(appName, *allApps, *parallelCount) default: common.LogFail(fmt.Sprintf("Invalid plugin subcommand call: %s", subcommand)) } diff --git a/plugins/ps/subcommands.go b/plugins/ps/subcommands.go index 065951a62..d4103fb90 100644 --- a/plugins/ps/subcommands.go +++ b/plugins/ps/subcommands.go @@ -25,9 +25,9 @@ func CommandInspect(appName string) error { } // CommandRebuild rebuilds an app from source -func CommandRebuild(appName string, allApps bool, runInSerial bool, parallelCount int) error { +func CommandRebuild(appName string, allApps bool, parallelCount int) error { if allApps { - return RunCommandAgainstAllApps(Rebuild, "rebuild", runInSerial, parallelCount) + return RunCommandAgainstAllApps(Rebuild, "rebuild", parallelCount) } if appName == "" { @@ -65,9 +65,9 @@ func CommandReport(appName string, infoFlag string) error { } // CommandRestart restarts an app -func CommandRestart(appName string, allApps bool, runInSerial bool, parallelCount int) error { +func CommandRestart(appName string, allApps bool, parallelCount int) error { if allApps { - return RunCommandAgainstAllApps(Restart, "restart", runInSerial, parallelCount) + return RunCommandAgainstAllApps(Restart, "restart", parallelCount) } if appName == "" { @@ -148,9 +148,9 @@ func CommandSet(appName string, property string, value string) error { } // CommandStart starts an app -func CommandStart(appName string, allApps bool, runInSerial bool, parallelCount int) error { +func CommandStart(appName string, allApps bool, parallelCount int) error { if allApps { - return RunCommandAgainstAllApps(Start, "start", runInSerial, parallelCount) + return RunCommandAgainstAllApps(Start, "start", parallelCount) } if appName == "" { @@ -165,9 +165,9 @@ func CommandStart(appName string, allApps bool, runInSerial bool, parallelCount } // CommandStop stops an app -func CommandStop(appName string, allApps bool, runInSerial bool, parallelCount int) error { +func CommandStop(appName string, allApps bool, parallelCount int) error { if allApps { - return RunCommandAgainstAllApps(Stop, "stop", runInSerial, parallelCount) + return RunCommandAgainstAllApps(Stop, "stop", parallelCount) } if appName == "" {