adds autoScalingStrategy option

laravel · Mar 4, 2023 · ceaaa58 · ceaaa58
1 parent 4ce5d9a
commit ceaaa58
Show file tree

Hide file tree

Showing 6 changed files with 60 additions and 8 deletions.
diff --git a/src/AutoScaler.php b/src/AutoScaler.php
@@ -96,11 +96,17 @@ protected function timeToClearPerQueue(Supervisor $supervisor, Collection $pools
     protected function numberOfWorkersPerQueue(Supervisor $supervisor, Collection $queues)
     {
         $timeToClearAll = $queues->sum('time');
+        $totalJobs = $queues->sum('size');
 
-        return $queues->mapWithKeys(function ($timeToClear, $queue) use ($supervisor, $timeToClearAll) {
+        return $queues->mapWithKeys(function ($timeToClear, $queue) use ($supervisor, $timeToClearAll, $totalJobs) {
             if ($timeToClearAll > 0 &&
                 $supervisor->options->autoScaling()) {
-                return [$queue => (($timeToClear['time'] / $timeToClearAll) * $supervisor->options->maxProcesses)];
+                $numberOfProcesses = $supervisor->options->autoScaleByNumberOfJobs()
+                    ? ($timeToClear['size'] / $totalJobs)
+                    : ($timeToClear['time'] / $timeToClearAll);
+                $numberOfProcesses *= $supervisor->options->maxProcesses;
+
+                return [$queue => $numberOfProcesses];
             } elseif ($timeToClearAll == 0 &&
                       $supervisor->options->autoScaling()) {
                 return [

diff --git a/src/Console/SupervisorCommand.php b/src/Console/SupervisorCommand.php
@@ -36,7 +36,8 @@ class SupervisorCommand extends Command
                             {--balance-max-shift=1 : The maximum number of processes to increase or decrease per one scaling}
                             {--workers-name=default : The name that should be assigned to the workers}
                             {--parent-id=0 : The parent process ID}
-                            {--rest=0 : Number of seconds to rest between jobs}';
+                            {--rest=0 : Number of seconds to rest between jobs}
+                            {--auto-scaling-strategy=runtime : If supervisor should scale by jobs or time to complete}';
 
     /**
      * The console command description.
@@ -111,12 +112,15 @@ protected function supervisorOptions()
                     ? $this->option('backoff')
                     : $this->option('delay');
 
+        $balance = $this->option('balance');
+        $autoScalingStrategy = $balance === 'auto' ? $this->option('auto-scaling-strategy') : null;
+
         return new SupervisorOptions(
             $this->argument('name'),
             $this->argument('connection'),
             $this->getQueue($this->argument('connection')),
             $this->option('workers-name'),
-            $this->option('balance'),
+            $balance,
             $backoff,
             $this->option('max-time'),
             $this->option('max-jobs'),
@@ -131,7 +135,8 @@ protected function supervisorOptions()
             $this->option('balance-cooldown'),
             $this->option('balance-max-shift'),
             $this->option('parent-id'),
-            $this->option('rest')
+            $this->option('rest'),
+            $autoScalingStrategy
         );
     }
 

diff --git a/src/QueueCommandString.php b/src/QueueCommandString.php
@@ -27,7 +27,7 @@ public static function toWorkerOptionsString(SupervisorOptions $options)
      */
     public static function toSupervisorOptionsString(SupervisorOptions $options)
     {
-        return sprintf('--workers-name=%s --balance=%s --max-processes=%s --min-processes=%s --nice=%s --balance-cooldown=%s --balance-max-shift=%s --parent-id=%s %s',
+        return sprintf('--workers-name=%s --balance=%s --max-processes=%s --min-processes=%s --nice=%s --balance-cooldown=%s --balance-max-shift=%s --parent-id=%s --auto-scaling-strategy=%s %s',
             $options->workersName,
             $options->balance,
             $options->maxProcesses,
@@ -36,6 +36,7 @@ public static function toSupervisorOptionsString(SupervisorOptions $options)
             $options->balanceCooldown,
             $options->balanceMaxShift,
             $options->parentId,
+            $options->autoScalingStrategy,
             static::toOptionsString($options)
         );
     }

diff --git a/src/SupervisorOptions.php b/src/SupervisorOptions.php
@@ -158,6 +158,13 @@ class SupervisorOptions
      */
     public $autoScale;
 
+    /**
+     * Indicates auto-scaling strategy should use runtime (time-to-complete) or size (total count of jobs).
+     *
+     * @var string|null
+     */
+    public $autoScalingStrategy = null;
+
     /**
      * Create a new worker options instance.
      *
@@ -181,6 +188,7 @@ class SupervisorOptions
      * @param  int  $balanceMaxShift
      * @param  int  $parentId
      * @param  int  $rest
+     * @param  string|null  $autoScalingStrategy
      */
     public function __construct($name,
                                 $connection,
@@ -201,7 +209,9 @@ public function __construct($name,
                                 $balanceCooldown = 3,
                                 $balanceMaxShift = 1,
                                 $parentId = 0,
-                                $rest = 0)
+                                $rest = 0,
+                                $autoScalingStrategy = 'runtime'
+    )
     {
         $this->name = $name;
         $this->connection = $connection;
@@ -223,6 +233,7 @@ public function __construct($name,
         $this->balanceMaxShift = $balanceMaxShift;
         $this->parentId = $parentId;
         $this->rest = $rest;
+        $this->autoScalingStrategy = $autoScalingStrategy;
     }
 
     /**
@@ -258,6 +269,16 @@ public function autoScaling()
         return $this->balance === 'auto';
     }
 
+    /**
+     * Determine if auto-scaling should be based count of jobs per queue.
+     *
+     * @return bool
+     */
+    public function autoScaleByNumberOfJobs()
+    {
+        return $this->autoScalingStrategy === 'size';
+    }
+
     /**
      * Get the command-line representation of the options for a supervisor.
      *
@@ -316,6 +337,7 @@ public function toArray()
             'balanceMaxShift' => $this->balanceMaxShift,
             'parentId' => $this->parentId,
             'rest' => $this->rest,
+            'autoScalingStrategy' => $this->autoScalingStrategy,
         ];
     }
 

diff --git a/tests/Feature/AddSupervisorTest.php b/tests/Feature/AddSupervisorTest.php
@@ -28,7 +28,7 @@ public function test_add_supervisor_command_creates_new_supervisor_on_master_pro
         $this->assertCount(1, $master->supervisors);
 
         $this->assertSame(
-            'exec '.$phpBinary.' artisan horizon:supervisor my-supervisor redis --workers-name=default --balance=off --max-processes=1 --min-processes=1 --nice=0 --balance-cooldown=3 --balance-max-shift=1 --parent-id=0 --backoff=0 --max-time=0 --max-jobs=0 --memory=128 --queue="default" --sleep=3 --timeout=60 --tries=0 --rest=0',
+            'exec '.$phpBinary.' artisan horizon:supervisor my-supervisor redis --workers-name=default --balance=off --max-processes=1 --min-processes=1 --nice=0 --balance-cooldown=3 --balance-max-shift=1 --parent-id=0 --auto-scaling-strategy=time --backoff=0 --max-time=0 --max-jobs=0 --memory=128 --queue="default" --sleep=3 --timeout=60 --tries=0 --rest=0',
             $master->supervisors->first()->process->getCommandLine()
         );
     }

diff --git a/tests/Feature/AutoScalerTest.php b/tests/Feature/AutoScalerTest.php
@@ -240,4 +240,22 @@ public function test_scaler_does_not_permit_going_to_zero_processes_despite_exce
         $this->assertSame(14, $supervisor->processPools['first']->totalProcessCount());
         $this->assertSame(1, $supervisor->processPools['second']->totalProcessCount());
     }
+
+    public function test_scaler_assigns_more_processes_to_queue_with_more_jobs_when_using_size_strategy()
+    {
+        [$scaler, $supervisor] = $this->with_scaling_scenario(100, [
+            'first' => ['current' => 50, 'size' => 1000, 'runtime' => 10],
+            'second' => ['current' => 50, 'size' => 500, 'runtime' => 1000],
+        ], ['autoScalingStrategy' => 'size']);
+
+        $scaler->scale($supervisor);
+
+        $this->assertSame(51, $supervisor->processPools['first']->totalProcessCount());
+        $this->assertSame(49, $supervisor->processPools['second']->totalProcessCount());
+
+        $scaler->scale($supervisor);
+
+        $this->assertSame(52, $supervisor->processPools['first']->totalProcessCount());
+        $this->assertSame(48, $supervisor->processPools['second']->totalProcessCount());
+    }
 }