Add fluxsvc flag to scale up job workers
@@ -53,6 +53,8 @@ func main() {
memcachedTimeout = fs.Duration("memcached-timeout", 100*time.Millisecond, "Maximum time to wait before giving up on memcached requests.")
memcachedService = fs.String("memcached-service", "memcached", "SRV service used to discover memcache servers.")
registryCacheExpiry = fs.Duration("registry-cache-expiry", 20*time.Minute, "Duration to keep cached registry tag info. Must be < 1 month.")
+ releaseJobWorkers = fs.Int(jobs.ReleaseJob+"-workers", 1, "Number of workers to process release jobs")
+ automatedInstanceJobWorkers = fs.Int(jobs.AutomatedInstanceJob+"-workers", 1, "Number of workers to process automated_instance jobs")
versionFlag = fs.Bool("version", false, "Get version number")
)
fs.Parse(os.Args)
@@ -191,13 +193,17 @@ func main() {
// release jobs can't interfere with slow automated service jobs, or vice
// versa. This is probably not optimal. Really all jobs should be quick and
// recoverable.
- for _, queue := range []string{
- jobs.DefaultQueue,
- jobs.ReleaseJob,
- jobs.AutomatedInstanceJob,
+ for queue, workers := range map[string]int{
+ jobs.DefaultQueue: 1, // Backwards compatibility...
+ jobs.ReleaseJob: *releaseJobWorkers,
+ jobs.AutomatedInstanceJob: *automatedInstanceJobWorkers,
} {
logger := log.NewContext(logger).With("component", "worker", "queues", fmt.Sprint([]string{queue}))
+ // create i workers for this queue
+ for i := 0; i < workers; i++ {
worker := jobs.NewWorker(jobStore, logger, []string{queue})
+
+ // All workers understand all job types, because I'm a lazy coder.
worker.Register(jobs.AutomatedInstanceJob, auto)
worker.Register(jobs.ReleaseJob, release.NewReleaser(instancer))
@@ -208,7 +214,7 @@ func main() {
}
}()
go worker.Work()
-
+ }
}
// Job GC cleaner
