From e6e5ede436dd42427003546ddca1a6b89d674f0e Mon Sep 17 00:00:00 2001 From: Jose Diaz-Gonzalez Date: Sat, 21 Nov 2020 19:36:13 -0500 Subject: [PATCH] refactor: make parallel commands available to all of Dokku There are other places in the codebase that currently act on all apps or could benefit from being able to do so. Being able to run `common.RunCommandAgainstAllApps()` and get that for free will be a boon to implementing this functionality elsewhere. --- plugins/common/parallel.go | 135 ++++++++++++++++++++++++++++++++++++ plugins/ps/ps.go | 136 ------------------------------------- plugins/ps/subcommands.go | 8 +-- 3 files changed, 139 insertions(+), 140 deletions(-) create mode 100644 plugins/common/parallel.go diff --git a/plugins/common/parallel.go b/plugins/common/parallel.go new file mode 100644 index 000000000..779bc4f90 --- /dev/null +++ b/plugins/common/parallel.go @@ -0,0 +1,135 @@ +package common + +import ( + "fmt" + "runtime" + "sync" +) + +// 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 { + Name string +} + +// ParallelCommandResult is the result of a parallel +// command run +type ParallelCommandResult struct { + Name string + Error error +} + +// RunCommandAgainstAllApps runs a given ParallelCommand against all apps +func RunCommandAgainstAllApps(command ParallelCommand, commandName string, parallelCount int) error { + runInSerial := false + + if parallelCount < -1 { + return fmt.Errorf("Invalid value %d for --parallel flag", parallelCount) + } + + if parallelCount == -1 { + cpuCount := runtime.NumCPU() + LogWarn(fmt.Sprintf("Setting --parallel=%d value to CPU count of %d", parallelCount, cpuCount)) + parallelCount = cpuCount + } + + if parallelCount == 0 || parallelCount == 1 { + LogWarn(fmt.Sprintf("Running %s in serial mode", commandName)) + runInSerial = true + } + + if runInSerial { + return runCommandAgainstAllAppsSerially(command, commandName) + } + + return runCommandAgainstAllAppsInParallel(command, commandName, parallelCount) +} + +func runCommandAgainstAllAppsInParallel(command ParallelCommand, commandName string, parallelCount int) error { + apps, err := DokkuApps() + if err != nil { + LogWarn(err.Error()) + return nil + } + + jobs := make(chan string, parallelCount) + results := make(chan ParallelCommandResult, len(apps)) + + go allocateJobs(apps, jobs) + done := make(chan error) + go aggregateResults(results, done) + createParallelWorkerPool(jobs, results, command, parallelCount) + err = <-done + + return err +} + +func runCommandAgainstAllAppsSerially(command ParallelCommand, commandName string) error { + apps, err := DokkuApps() + if err != nil { + LogWarn(err.Error()) + return nil + } + + errorCount := 0 + for _, appName := range apps { + LogInfo1(fmt.Sprintf("Running %s against app %s", commandName, appName)) + if err = command(appName); err != nil { + errorCount++ + } + } + + if errorCount > 0 { + return fmt.Errorf("%s command returned %d errors", commandName, errorCount) + } + + return nil +} + +func allocateJobs(input []string, jobs chan string) { + for _, job := range input { + jobs <- job + } + close(jobs) +} + +func aggregateResults(results chan ParallelCommandResult, done chan error) { + var parallelError error + errorCount := 0 + for result := range results { + if result.Error != nil { + LogWarn(fmt.Sprintf("Error running command against %s", result.Name)) + errorCount++ + } + } + if errorCount > 0 { + parallelError = fmt.Errorf("Encountered %d errors during parallel run", errorCount) + } + done <- parallelError +} + +func createParallelWorker(jobs chan string, results chan ParallelCommandResult, command ParallelCommand, wg *sync.WaitGroup, workerID int) { + for job := range jobs { + LogInfo1(fmt.Sprintf("Running command against %s", job)) + output := ParallelCommandResult{ + Name: job, + Error: command(job), + } + results <- output + } + wg.Done() +} + +func createParallelWorkerPool(jobs chan string, results chan ParallelCommandResult, command ParallelCommand, numberOfWorkers int) { + var wg sync.WaitGroup + for i := 0; i < numberOfWorkers; i++ { + wg.Add(1) + go createParallelWorker(jobs, results, command, &wg, i) + } + wg.Wait() + close(results) +} diff --git a/plugins/ps/ps.go b/plugins/ps/ps.go index 1a9818943..81c52e8c2 100644 --- a/plugins/ps/ps.go +++ b/plugins/ps/ps.go @@ -2,11 +2,8 @@ package ps import ( "fmt" - "runtime" "strconv" "strings" - "sync" - "time" "github.com/dokku/dokku/plugins/common" ) @@ -15,23 +12,6 @@ import ( // and defaults to -1 (false) 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{ @@ -39,122 +19,6 @@ var ( } ) -// RunCommandAgainstAllApps runs a given ParallelCommand against all apps -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 { - return RunCommandAgainstAllAppsSerially(command, commandName) - } - - return RunCommandAgainstAllAppsInParallel(command, commandName, parallelCount) -} - -// RunCommandAgainstAllAppsInParallel runs a given -// ParallelCommand against all apps in parallel -func RunCommandAgainstAllAppsInParallel(command ParallelCommand, commandName string, parallelCount int) error { - 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 -// ParallelCommand against all apps in serial -func RunCommandAgainstAllAppsSerially(command ParallelCommand, commandName string) error { - apps, err := common.DokkuApps() - if err != nil { - common.LogWarn(err.Error()) - return nil - } - - errorCount := 0 - for _, appName := range apps { - common.LogInfo1(fmt.Sprintf("Running %s against app %s", commandName, appName)) - if err = command(appName); err != nil { - errorCount++ - } - } - - if errorCount > 0 { - return fmt.Errorf("%s command returned %d errors", commandName, errorCount) - } - - return nil -} - // Rebuild rebuilds app from base image func Rebuild(appName string) error { return common.PlugnTrigger("receive-app", []string{appName}...) diff --git a/plugins/ps/subcommands.go b/plugins/ps/subcommands.go index d4103fb90..bda148ff8 100644 --- a/plugins/ps/subcommands.go +++ b/plugins/ps/subcommands.go @@ -27,7 +27,7 @@ func CommandInspect(appName string) error { // CommandRebuild rebuilds an app from source func CommandRebuild(appName string, allApps bool, parallelCount int) error { if allApps { - return RunCommandAgainstAllApps(Rebuild, "rebuild", parallelCount) + return common.RunCommandAgainstAllApps(Rebuild, "rebuild", parallelCount) } if appName == "" { @@ -67,7 +67,7 @@ func CommandReport(appName string, infoFlag string) error { // CommandRestart restarts an app func CommandRestart(appName string, allApps bool, parallelCount int) error { if allApps { - return RunCommandAgainstAllApps(Restart, "restart", parallelCount) + return common.RunCommandAgainstAllApps(Restart, "restart", parallelCount) } if appName == "" { @@ -150,7 +150,7 @@ func CommandSet(appName string, property string, value string) error { // CommandStart starts an app func CommandStart(appName string, allApps bool, parallelCount int) error { if allApps { - return RunCommandAgainstAllApps(Start, "start", parallelCount) + return common.RunCommandAgainstAllApps(Start, "start", parallelCount) } if appName == "" { @@ -167,7 +167,7 @@ func CommandStart(appName string, allApps bool, parallelCount int) error { // CommandStop stops an app func CommandStop(appName string, allApps bool, parallelCount int) error { if allApps { - return RunCommandAgainstAllApps(Stop, "stop", parallelCount) + return common.RunCommandAgainstAllApps(Stop, "stop", parallelCount) } if appName == "" {