github/dokku
0
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat: add proper parallel support

Browse Source
This commit is contained in:
Jose Diaz-Gonzalez
2020-11-21 17:31:13 -05:00
parent ef083e6289
commit 28a36c4908
5 changed files with 121 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
plugins/ps/ps.go
View File

@@ -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++
}

8
plugins/ps/src/commands/commands.go
View File

@@ -19,14 +19,14 @@ Additional commands:`
helpContent = `
ps:inspect <app>, Displays a sanitized version of docker inspect for an app
ps:rebuild [--parallel count] [--serial] [--all|<app>], Rebuilds an app from source
ps:rebuild [--parallel count] [--all|<app>], Rebuilds an app from source
ps:report [<app>] [<flag>], Displays a process report for one or more apps
ps:restart [--parallel count] [--serial] [--all|<app>], Restart an app
ps:restart [--parallel count] [--all|<app>], Restart an app
ps:restore [<app>], Start previously running apps e.g. after reboot
ps:scale [--skip-deploy] <app> <proc>=<count> [<proc>=<count>...], Get/Set how many instances of a given process to run
ps:set <app> <key> <value>, Set or clear a ps property for an app
ps:start [--parallel count] [--serial] [--all|<app>], Start an app
ps:stop [--parallel count] [--serial] [--all|<app>], Stop an app
ps:start [--parallel count] [--all|<app>], Start an app
ps:stop [--parallel count] [--all|<app>], Stop an app
`
)

12
plugins/ps/src/subcommands/subcommands.go
View File

@@ -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))
}

16
plugins/ps/subcommands.go
View File

@@ -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 == "" {