add -maxextractworker parameter/config setting using remeh/sizedwaitg…

…roup #79 #77 #76

README.md

@@ -65,6 +65,10 @@ cd /tmp/$RDIR/ ; time r10k puppetfile install
 You can just grab the most recent stable release here:
 https://github.com/xorpaul/g10k/releases
 
+- Before using g10k with a large Puppet setup with many modules, be sure to increase the amount of open file handles (nfiles) and number of child processes (nproc), see limits.conf(5) for details.
+- If you are using a private Git or Forge server think about adjusting the `-maxworker` parameter/config setting before DOSing your own infrastructure ;) (default 50)
+- To protect your local machine use `-maxextractworker` parameter/config setting with wich you can limit the number of Goroutines that are allowed to run in parallel for local Git and Forge module extracting processes (git clone, untar and gunzip) (default 20)
+
 
 ## Usage Docs
 ```
@@ -87,6 +91,8 @@ Usage of ./g10k:
         purge the Puppet environment directory and do a full sync
   -info
         log info output, defaults to false
+  -maxextractworker int
+        how many Goroutines are allowed to run in parallel for local Git and Forge module extracting processes (git clone, untar and gunzip) (default 20)
   -maxworker int
         how many Goroutines are allowed to run in parallel for Git and Forge module resolving (default 50)
   -module string

config.go

@@ -68,12 +68,21 @@ func readConfigfile(configFile string) ConfigSettings {
 	}
 
 	// set default max Go routines for Forge and Git module resolution if none is given
-	if maxworker == 0 {
-		config.Maxworker = 50
-	} else {
+	if !(config.Maxworker > 0) {
+		config.Maxworker = maxworker
+	}
+	if maxworker != 50 {
 		config.Maxworker = maxworker
 	}
 
+	// set default max Go routines for Forge and Git module extracting
+	if !(config.MaxExtractworker > 0) {
+		config.MaxExtractworker = maxExtractworker
+	}
+	if maxExtractworker != 20 {
+		config.MaxExtractworker = maxExtractworker
+	}
+
 	return config
 }
 

forge.go

@@ -168,9 +168,8 @@ func queryForgeAPI(name string, file string, fm ForgeModule) ForgeResult {
 			Warnf("Forge API error, trying to use cache for module " + fm.author + "/" + name)
 			_ = getLatestCachedModule(fm)
 			return ForgeResult{false, "", "", 0}
-		} else {
-			Fatalf("queryForgeAPI(): Error while issuing the HTTP request to " + url + " Error: " + err.Error())
 		}
+		Fatalf("queryForgeAPI(): Error while issuing the HTTP request to " + url + " Error: " + err.Error())
 	}
 	duration := time.Since(before).Seconds()
 	Verbosef("Querying Forge API " + url + " took " + strconv.FormatFloat(duration, 'f', 5, 64) + "s")

g10k.go

@@ -29,7 +29,6 @@ var (
 	moduleParam            string
 	configFile             string
 	config                 ConfigSettings
-	wg                     sync.WaitGroup
 	mutex                  sync.Mutex
 	empty                  struct{}
 	syncGitCount           int
@@ -47,6 +46,7 @@ var (
 	uniqueForgeModules     map[string]ForgeModule
 	latestForgeModules     LatestForgeModules
 	maxworker              int
+	maxExtractworker       int
 )
 
 type LatestForgeModules struct {
@@ -65,7 +65,8 @@ type ConfigSettings struct {
 	Sources                  map[string]Source
 	Timeout                  int  `yaml:"timeout"`
 	IgnoreUnreachableModules bool `yaml:"ignore_unreachable_modules"`
-	Maxworker                int
+	Maxworker                int  `yaml:"maxworker"`
+	MaxExtractworker         int  `yaml:"maxextractworker"`
 	UseCacheFallback         bool `yaml:"use_cache_fallback"`
 }
 
@@ -152,6 +153,7 @@ func main() {
 	flag.StringVar(&moduleDirParam, "moduledir", "", "allows overriding of Puppetfile specific moduledir setting, the folder in which Puppet modules will be extracted")
 	flag.StringVar(&cacheDirParam, "cachedir", "", "allows overriding of the g10k config file cachedir setting, the folder in which g10k will download git repositories and Forge modules")
 	flag.IntVar(&maxworker, "maxworker", 50, "how many Goroutines are allowed to run in parallel for Git and Forge module resolving")
+	flag.IntVar(&maxExtractworker, "maxextractworker", 20, "how many Goroutines are allowed to run in parallel for local Git and Forge module extracting processes (git clone, untar and gunzip)")
 	flag.BoolVar(&pfMode, "puppetfile", false, "install all modules from Puppetfile in cwd")
 	flag.StringVar(&pfLocation, "puppetfilelocation", "./Puppetfile", "which Puppetfile to use in -puppetfile mode")
 	flag.BoolVar(&force, "force", false, "purge the Puppet environment directory and do a full sync")
@@ -170,7 +172,7 @@ func main() {
 	version := *versionFlag
 
 	if version {
-		fmt.Println("g10k Version 1.0 Build time:", buildtime, "UTC")
+		fmt.Println("g10k Version 0.4 Build time:", buildtime, "UTC")
 		os.Exit(0)
 	}
 
@@ -222,7 +224,7 @@ func main() {
 			}
 			//config = ConfigSettings{CacheDir: cachedir, ForgeCacheDir: cachedir, ModulesCacheDir: cachedir, EnvCacheDir: cachedir, Forge:{Baseurl: "https://forgeapi.puppetlabs.com"}, Sources: sm}
 			forgeDefaultSettings := Forge{Baseurl: "https://forgeapi.puppetlabs.com"}
-			config = ConfigSettings{CacheDir: cachedir, ForgeCacheDir: cachedir, ModulesCacheDir: cachedir, EnvCacheDir: cachedir, Sources: sm, Forge: forgeDefaultSettings, Maxworker: maxworker, UseCacheFallback: usecacheFallback}
+			config = ConfigSettings{CacheDir: cachedir, ForgeCacheDir: cachedir, ModulesCacheDir: cachedir, EnvCacheDir: cachedir, Sources: sm, Forge: forgeDefaultSettings, Maxworker: maxworker, UseCacheFallback: usecacheFallback, MaxExtractworker: maxExtractworker}
 			target = pfLocation
 			puppetfile := readPuppetfile(target, "", "cmdlineparam", false)
 			puppetfile.workDir = "."
@@ -254,7 +256,7 @@ func main() {
 	Debugf("Forge modules metadata.json parsing took " + strconv.FormatFloat(metadataJsonParseTime, 'f', 4, 64) + " seconds")
 
 	if !check4update && !quiet {
-		fmt.Println("Synced", target, "with", syncGitCount, "git repositories and", syncForgeCount, "Forge modules in "+strconv.FormatFloat(time.Since(before).Seconds(), 'f', 1, 64)+"s with git ("+strconv.FormatFloat(syncGitTime, 'f', 1, 64)+"s sync, I/O", strconv.FormatFloat(ioGitTime, 'f', 1, 64)+"s) and Forge ("+strconv.FormatFloat(syncForgeTime, 'f', 1, 64)+"s query+download, I/O", strconv.FormatFloat(ioForgeTime, 'f', 1, 64)+"s) using", strconv.Itoa(config.Maxworker), "workers")
+		fmt.Println("Synced", target, "with", syncGitCount, "git repositories and", syncForgeCount, "Forge modules in "+strconv.FormatFloat(time.Since(before).Seconds(), 'f', 1, 64)+"s with git ("+strconv.FormatFloat(syncGitTime, 'f', 1, 64)+"s sync, I/O", strconv.FormatFloat(ioGitTime, 'f', 1, 64)+"s) and Forge ("+strconv.FormatFloat(syncForgeTime, 'f', 1, 64)+"s query+download, I/O", strconv.FormatFloat(ioForgeTime, 'f', 1, 64)+"s) using", strconv.Itoa(config.Maxworker), "resolv and", strconv.Itoa(config.MaxExtractworker), "extract workers")
 	}
 	if dryRun && (needSyncForgeCount > 0 || needSyncGitCount > 0) {
 		os.Exit(1)

puppetfile.go

@@ -8,6 +8,7 @@ import (
 	"sync"
 
 	"github.com/henvic/uiprogress"
+	"github.com/remeh/sizedwaitgroup"
 )
 
 // sourceSanityCheck is a validation function that checks if the given source has all neccessary attributes (basedir, remote, SSH key exists if given)
@@ -26,9 +27,10 @@ func sourceSanityCheck(source string, sa Source) {
 }
 
 func resolvePuppetEnvironment(envBranch string) {
+	wg := sizedwaitgroup.New(config.MaxExtractworker + 1)
 	allPuppetfiles := make(map[string]Puppetfile)
 	for source, sa := range config.Sources {
-		wg.Add(1)
+		wg.Add()
 		go func(source string, sa Source) {
 			defer wg.Done()
 			if force {
@@ -65,7 +67,7 @@ func resolvePuppetEnvironment(envBranch string) {
 						foundBranch = true
 					}
 
-					wg.Add(1)
+					wg.Add()
 
 					go func(branch string) {
 						defer wg.Done()
@@ -94,6 +96,7 @@ func resolvePuppetEnvironment(envBranch string) {
 					}(branch)
 
 				}
+
 				if sa.WarnMissingBranch && !foundBranch {
 					Warnf("WARNING: Couldn't find specified branch '" + envBranch + "' anywhere in source '" + source + "' (" + sa.Remote + ")")
 				}
@@ -124,7 +127,7 @@ func resolvePuppetEnvironment(envBranch string) {
 }
 
 func resolvePuppetfile(allPuppetfiles map[string]Puppetfile) {
-	var wg sync.WaitGroup
+	wg := sizedwaitgroup.New(config.MaxExtractworker)
 	exisitingModuleDirs := make(map[string]struct{})
 	uniqueGitModules := make(map[string]GitModule)
 	uniqueForgeModules := make(map[string]ForgeModule)
@@ -205,7 +208,7 @@ func resolvePuppetfile(allPuppetfiles map[string]Puppetfile) {
 			mutex.Unlock()
 		}
 		for gitName, gitModule := range pf.gitModules {
-			wg.Add(1)
+			wg.Add()
 			go func(gitName string, gitModule GitModule) {
 				defer wg.Done()
 				targetDir := moduleDir + gitName + "/"
@@ -260,7 +263,7 @@ func resolvePuppetfile(allPuppetfiles map[string]Puppetfile) {
 						}
 					}
 				} else {
-					success = syncToModuleDir(moduleCacheDir, targetDir, tree, gitModule.ignoreUnreachable, gitModule.ignoreUnreachable)
+					syncToModuleDir(moduleCacheDir, targetDir, tree, gitModule.ignoreUnreachable, gitModule.ignoreUnreachable)
 				}
 
 				// remove this module from the exisitingModuleDirs map
@@ -272,7 +275,7 @@ func resolvePuppetfile(allPuppetfiles map[string]Puppetfile) {
 			}(gitName, gitModule)
 		}
 		for forgeModuleName, fm := range pf.forgeModules {
-			wg.Add(1)
+			wg.Add()
 			go func(forgeModuleName string, fm ForgeModule) {
 				defer wg.Done()
 				syncForgeToModuleDir(forgeModuleName, fm, moduleDir)