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.
This commit is contained in:
Jose Diaz-Gonzalez
135
plugins/common/parallel.go
Normal file
135
plugins/common/parallel.go
Normal file
@@ -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)
|
||||
}
|
||||
136
plugins/ps/ps.go
136
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}...)
|
||||
|
||||
@@ -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 == "" {
|
||||
|
||||
Reference in New Issue
Block a user