apparently workers run all jobs

This commit is contained in:
Andre Medeiros 2022-11-02 01:06:08 -04:00
parent 166111e558
commit bf84f75be4
2 changed files with 12 additions and 34 deletions

View file

@ -24,25 +24,20 @@ var (
func WorkerCmd(ctx context.Context) *cobra.Command { func WorkerCmd(ctx context.Context) *cobra.Command {
var consumers int var consumers int
var queueID string
cmd := &cobra.Command{ cmd := &cobra.Command{
Use: "worker", Use: "worker",
Args: cobra.ExactArgs(0), Args: cobra.ExactArgs(0),
Short: "Work through job queues.", Short: "Work through job queues.",
RunE: func(cmd *cobra.Command, args []string) error { RunE: func(cmd *cobra.Command, args []string) error {
if queueID == "" {
return fmt.Errorf("need a queue to work on")
}
runtime.SetBlockProfileRate(1) runtime.SetBlockProfileRate(1)
runtime.SetMutexProfileFraction(1) runtime.SetMutexProfileFraction(1)
svc := fmt.Sprintf("worker: %s", queueID) svc := fmt.Sprintf("worker")
logger := cmdutil.NewLogger(svc) logger := cmdutil.NewLogger(svc)
defer func() { _ = logger.Sync() }() defer func() { _ = logger.Sync() }()
tag := fmt.Sprintf("worker:%s", queueID) tag := fmt.Sprintf("worker")
statsd, err := cmdutil.NewStatsdClient(tag) statsd, err := cmdutil.NewStatsdClient(tag)
if err != nil { if err != nil {
return err return err
@ -63,21 +58,18 @@ func WorkerCmd(ctx context.Context) *cobra.Command {
} }
defer redis.Close() defer redis.Close()
workerFn, ok := queues[queueID]
if !ok {
return fmt.Errorf("queue does not exist: %s", queueID)
}
worker := workerFn(ctx, logger, statsd, db, redis, consumers)
mgr := faktoryworker.NewManager() mgr := faktoryworker.NewManager()
mgr.Concurrency = consumers mgr.Concurrency = consumers
mgr.Register(queueID, worker.Process) for queue, workerFn := range queues {
worker := workerFn(ctx, logger, statsd, db, redis, consumers)
mgr.Register(queue, worker.Process)
}
return mgr.RunWithContext(ctx) return mgr.RunWithContext(ctx)
}, },
} }
cmd.Flags().IntVar(&consumers, "consumers", runtime.NumCPU()*64, "The consumers to run") cmd.Flags().IntVar(&consumers, "consumers", runtime.NumCPU()*64, "The consumers to run")
cmd.Flags().StringVar(&queueID, "queue", "", "The queue to work on")
return cmd return cmd
} }

View file

@ -105,7 +105,7 @@ services:
maxInstances: 30 maxInstances: 30
targetCPUPercent: 80 targetCPUPercent: 80
buildCommand: go install github.com/bugsnag/panic-monitor@latest && go build ./cmd/apollo buildCommand: go install github.com/bugsnag/panic-monitor@latest && go build ./cmd/apollo
startCommand: panic-monitor ./apollo worker --queue notifications --consumers 256 startCommand: panic-monitor ./apollo worker --queue NotificationCheckJob --consumers 256
# Stuck Notifications Checker # Stuck Notifications Checker
- type: worker - type: worker
@ -123,7 +123,7 @@ services:
maxInstances: 10 maxInstances: 10
targetCPUPercent: 80 targetCPUPercent: 80
buildCommand: go install github.com/bugsnag/panic-monitor@latest && go build ./cmd/apollo buildCommand: go install github.com/bugsnag/panic-monitor@latest && go build ./cmd/apollo
startCommand: panic-monitor ./apollo worker --queue stuck-notifications --consumers 64 startCommand: panic-monitor ./apollo worker --queue StuckNotificationsJob --consumers 64
# Scheduler # Scheduler
- type: worker - type: worker
@ -137,20 +137,6 @@ services:
buildCommand: go install github.com/bugsnag/panic-monitor@latest && go build ./cmd/apollo buildCommand: go install github.com/bugsnag/panic-monitor@latest && go build ./cmd/apollo
startCommand: panic-monitor ./apollo scheduler startCommand: panic-monitor ./apollo scheduler
# User Watcher
- type: worker
name: worker.watcher.users
env: go
plan: starter
envVars:
- fromGroup: env-settings
- key: BUGSNAG_APP_TYPE
value: worker
- key: BUGSNAG_METADATA_QUEUE
value: users
buildCommand: go install github.com/bugsnag/panic-monitor@latest && go build ./cmd/apollo
startCommand: panic-monitor ./apollo worker --queue users
# Subreddit Watcher # Subreddit Watcher
- type: worker - type: worker
name: worker.watcher.subreddits name: worker.watcher.subreddits
@ -167,7 +153,7 @@ services:
maxInstances: 10 maxInstances: 10
targetCPUPercent: 80 targetCPUPercent: 80
buildCommand: go install github.com/bugsnag/panic-monitor@latest && go build ./cmd/apollo buildCommand: go install github.com/bugsnag/panic-monitor@latest && go build ./cmd/apollo
startCommand: panic-monitor ./apollo worker --queue subreddits startCommand: panic-monitor ./apollo worker --queue SubredditWatcherJob
# Trending Posts Watcher # Trending Posts Watcher
- type: worker - type: worker
@ -185,7 +171,7 @@ services:
maxInstances: 10 maxInstances: 10
targetCPUPercent: 80 targetCPUPercent: 80
buildCommand: go install github.com/bugsnag/panic-monitor@latest && go build ./cmd/apollo buildCommand: go install github.com/bugsnag/panic-monitor@latest && go build ./cmd/apollo
startCommand: panic-monitor ./apollo worker --queue trending startCommand: panic-monitor ./apollo worker --queue SubredditTrendingJob
# Live Activities # Live Activities
- type: worker - type: worker
@ -203,7 +189,7 @@ services:
maxInstances: 10 maxInstances: 10
targetCPUPercent: 80 targetCPUPercent: 80
buildCommand: go install github.com/bugsnag/panic-monitor@latest && go build ./cmd/apollo buildCommand: go install github.com/bugsnag/panic-monitor@latest && go build ./cmd/apollo
startCommand: panic-monitor ./apollo worker --queue live-activities startCommand: panic-monitor ./apollo worker --queue LiveActivityJob
envVarGroups: envVarGroups:
# Environment # Environment