Merge pull request #19 from aloware/feature/fair-queue-improvements

Feature | Fair-Queue Improvements
aloware · Jun 14, 2022 · 30b13ff · 30b13ff
2 parents c0feb15 + 65b64d9
commit 30b13ff
Show file tree

Hide file tree

Showing 8 changed files with 198 additions and 9 deletions.
diff --git a/public/app.js b/public/app.js
diff --git a/resources/js/screens/dashboard.vue b/resources/js/screens/dashboard.vue
@@ -182,6 +182,39 @@
                     </div>
                 </div>
             </div>
+            <div class="card-bg-secondary">
+                <div class="d-flex">
+                    <div class="w-50 border-right border-bottom">
+                        <div class="p-4">
+                            <small class="text-uppercase">Processed Jobs Past Minute</small>
+
+                            <h4 class="mt-4 mb-0">
+                                {{ stats.processedJobsInPastMinute ? stats.processedJobsInPastMinute.toLocaleString() : 0 }}
+                            </h4>
+                        </div>
+                    </div>
+
+                    <div class="w-50 border-right border-bottom">
+                        <div class="p-4">
+                            <small class="text-uppercase">Processed Jobs Past 20 Minutes</small>
+
+                            <h4 class="mt-4 mb-0">
+                                {{ stats.processedJobsInPast20Minutes ? stats.processedJobsInPast20Minutes.toLocaleString() : 0 }}
+                            </h4>
+                        </div>
+                    </div>
+
+                    <div class="w-50 border-right border-bottom">
+                        <div class="p-4">
+                            <small class="text-uppercase">Processed Jobs Past Hour</small>
+
+                            <h4 class="mt-4 mb-0">
+                                {{ stats.processedJobsInPastHour ? stats.processedJobsInPastHour.toLocaleString() : 0 }}
+                            </h4>
+                        </div>
+                    </div>
+                </div>
+            </div>
         </div>
 
         <div class="card mt-4" v-if="workload.length">

diff --git a/src/Commands/RefreshStats.php b/src/Commands/RefreshStats.php
@@ -0,0 +1,73 @@
+<?php
+
+namespace Aloware\FairQueue\Commands;
+
+use Aloware\FairQueue\Facades\FairQueue;
+use Aloware\FairQueue\Interfaces\RepositoryInterface;
+use Aloware\FairQueue\Repositories\RedisKeys;
+use Illuminate\Console\Command;
+
+class RefreshStats extends Command
+{
+    use RedisKeys;
+    /**
+     * The name and signature of the console command.
+     *
+     * @var string
+     */
+    protected $signature = 'fair-queue:refresh-stats';
+
+    /**
+     * The console command description.
+     *
+     * @var string
+     */
+    protected $description = 'Refresh Jobs Stats';
+
+    /**
+     * @var RepositoryInterface
+     */
+    private $repository;
+
+    /**
+     * Create a new command instance.
+     *
+     * @return void
+     */
+    public function __construct(RepositoryInterface $repository)
+    {
+        parent::__construct();
+
+        $this->repository = $repository;
+    }
+
+    /**
+     * Execute the console command.
+     *
+     * @return mixed
+     */
+    public function handle()
+    {
+        $redis = FairQueue::getConnection();
+
+        $this->refreshKeys($redis, $redis->keys($this->recentProcesedJobsPattern(1)), 1);
+        $this->refreshKeys($redis, $redis->keys($this->recentProcesedJobsPattern(20)), 20);
+        $this->refreshKeys($redis, $redis->keys($this->recentProcesedJobsPattern(60)), 60);
+
+        $this->info('Fair-Queue Stats Refreshed Successfully');
+    }
+
+    /**
+     * Refresh Redis Keys
+     *
+     * @return mixed
+     */
+    public function refreshKeys($redis, $keys, $minute = 1)
+    {
+        $timestamp = now()->subMinutes($minute)->getPreciseTimestamp(3);
+        foreach ($keys as $key)
+        {
+            $redis->zremrangebyscore($key, '-inf', "({$timestamp}");
+        }
+    }
+}
diff --git a/src/FairQueueServiceProvider.php b/src/FairQueueServiceProvider.php
@@ -7,15 +7,20 @@
 use Aloware\FairQueue\Commands\PurgeFailedJobs;
 use Aloware\FairQueue\Commands\RecoverLostJobs;
 use Aloware\FairQueue\Commands\RetryFailedJobs;
+use Aloware\FairQueue\Commands\RefreshStats;
 use Aloware\FairQueue\Facades\FairQueue;
 use Aloware\FairQueue\Repositories\RedisRepository;
 use Aloware\FairQueue\Interfaces\RepositoryInterface;
+use Aloware\FairQueue\Repositories\RedisKeys;
 use Illuminate\Console\Scheduling\Schedule;
 use Illuminate\Support\ServiceProvider;
 use Illuminate\Support\Facades\Route;
+use Illuminate\Support\Facades\Queue;
+use Illuminate\Queue\Events\JobProcessed;
 
 class FairQueueServiceProvider extends ServiceProvider
 {
+    use RedisKeys;
     /**
      * Register the service provider.
      *
@@ -43,17 +48,21 @@ public function boot(): void
             GenerateSignal::class,
             RetryFailedJobs::class,
             PurgeFailedJobs::class,
+            RefreshStats::class,
         ]);
 
         $this->registerRoutes();
         $this->registerResources();
         $this->publishAssets();
+        $this->registerQueueEvents();
 
         $this->app->booted(function () {
             /** @var Schedule $schedule */
             $schedule = $this->app->make(Schedule::class);
 
             $schedule->command(RecoverLostJobs::class)->hourly();
+            $schedule->command(RecoverLostJobs::class)->hourly();
+            $schedule->command(RefreshStats::class)->everyMinute();
         });
     }
 
@@ -73,6 +82,30 @@ protected function registerRoutes(): void
         });
     }
 
+    /**
+     * Register the FairQueue Queue Events.
+     *
+     * @return void
+     */
+    protected function registerQueueEvents(): void
+    {
+        Queue::after(function (JobProcessed $event) {
+            $redis = FairQueue::getConnection();
+            $payload = $event->job->payload();
+            $command = unserialize($payload['data']['command']);
+            if(!$command instanceof FairSignalJob)
+                return;
+            $queue = $command->queue;
+            $partition = $command->partition;
+            $past_minute_key = $this->partitionProcessedJobsInPastMinutesKey($queue, $partition, 1);
+            $past_20minute_key = $this->partitionProcessedJobsInPastMinutesKey($queue, $partition, 20);
+            $past_60minute_key = $this->partitionProcessedJobsInPastMinutesKey($queue, $partition, 60);
+            $redis->zadd($past_minute_key, now()->getPreciseTimestamp(3), $payload['id']);
+            $redis->zadd($past_20minute_key, now()->getPreciseTimestamp(3), $payload['id']);
+            $redis->zadd($past_60minute_key, now()->getPreciseTimestamp(3), $payload['id']);
+        });
+    }
+
     /**
      * Register the FairQueue resources.
      *

diff --git a/src/Http/Controllers/DashboardStatsController.php b/src/Http/Controllers/DashboardStatsController.php
@@ -19,9 +19,12 @@ public function index()
         $totalFailedJobs = FairQueue::totalFailedJobsCount($failedQueues);
 
         return response()->json([
-            'totalJobs'  => $totalJobs,
-            'totalFailedJobs'  => $totalFailedJobs,
+            'totalJobs' => $totalJobs,
+            'totalFailedJobs' => $totalFailedJobs,
             'totalQueues' => count($queues),
+            'processedJobsInPastMinute' => FairQueue::processedJobsInPastMinutes($queues, 1),
+            'processedJobsInPast20Minutes' => FairQueue::processedJobsInPastMinutes($queues, 20),
+            'processedJobsInPastHour' => FairQueue::processedJobsInPastMinutes($queues, 60),
         ]);
     }
 

diff --git a/src/Interfaces/RepositoryInterface.php b/src/Interfaces/RepositoryInterface.php
@@ -37,6 +37,8 @@ public function job($queue, $partition, $index);
 
     public function totalJobsCount($queues);
 
+    public function processedJobsInPastMinutes($queues, $minutes);
+
     public function failedQueues();
 
     public function failedQueuesWithPartitions();
@@ -55,6 +57,8 @@ public function purgeFailedJobs();
 
     public function recoverLost($age = 300);
 
+    public function getConnection();
+
     /**
      * @throws SampleNotFoundException
      */

diff --git a/src/Repositories/RedisKeys.php b/src/Repositories/RedisKeys.php
@@ -15,6 +15,16 @@ private function partitionKey($queue, $partition, $prefix = '')
         );
     }
 
+    private function queueKey($queue, $prefix = '')
+    {
+        return sprintf(
+            '%s%s:*:%s',
+            $this->fairQueueKeyPrefix(),
+            $prefix,
+            $queue
+        );
+    }
+
     private function failedPartitionKey($queue, $partition)
     {
         return $this->partitionKey($queue, $partition, '-failed');
@@ -25,6 +35,26 @@ private function partitionProcessedCountJobKey($queue, $partition)
         return $this->partitionKey($queue, $partition, '-internal') . ':processed';
     }
 
+    private function queueProcessedJobsInPastMinutesKey($queue, $minutes = 1)
+    {
+        $queue_key = $this->queueKey($queue, '-internal');
+        return "{$queue_key}:processed:{$minutes}min";
+    }
+
+    private function partitionProcessedJobsInPastMinutesKey($queue, $partition, $minutes = 1)
+    {
+        $partition_key = $this->partitionKey($queue, $partition, '-internal');
+        return "{$partition_key}:processed:{$minutes}min";
+    }
+
+    private function recentProcesedJobsPattern($minute = 1)
+    {
+        return sprintf(
+            '%s-internal:*:processed:' . $minute . 'min*',
+            $this->fairQueueKeyPrefix()
+        );
+    }
+
     private function partitionPerSecKey($queue, $partition)
     {
         return $this->partitionKey($queue, $partition, '-internal') . ':persec';

diff --git a/src/Repositories/RedisRepository.php b/src/Repositories/RedisRepository.php
@@ -53,6 +53,19 @@ public function totalJobsCount($queues)
         return $this->totalJobsCountPrivate($queues, 'partitions', 'partitionKey');
     }
 
+    public function processedJobsInPastMinutes($queues, $minutes)
+    {
+        $redis = $this->getConnection();
+
+        $total = 0;
+        foreach($queues as $queue)
+        {
+            $queue_key = $this->queueProcessedJobsInPastMinutesKey($queue, $minutes);
+            $total += $redis->zcard($queue_key);
+        }
+        return $total;
+    }
+
     public function failedPartitions($queue)
     {
         return $this->partitionsPrivate(
@@ -202,7 +215,7 @@ public function recoverLost($age = 300)
         $redis = $this->getConnection();
 
         $pattern = $this->inProgressJobsPattern();
-        $keys    = $redis->keys($pattern);
+        $keys = $redis->keys($pattern);
 
         $count = 0;
 
@@ -317,7 +330,7 @@ private function partitionsWithCountPrivate(
 
         $pattern = $this->$queuePartitionListPatternResolver($queue);
 
-        $keys       = $redis->keys($pattern);
+        $keys = $redis->keys($pattern);
         $partitions = [];
 
         foreach ($keys as $key) {
@@ -346,8 +359,8 @@ private function partitionsWithCountPrivate(
 
     private function jobsPrivate($queue, $partition, $partitionKeyResolver = 'partitionKey')
     {
-        $redis      = $this->getConnection();
-        $perPage    = request('limit', 25);
+        $redis = $this->getConnection();
+        $perPage = request('limit', 25);
         $startingAt = request('starting_at', 0);
 
         $partitionKey = $this->$partitionKeyResolver($queue, $partition);
@@ -433,9 +446,9 @@ private function popPrivate($queue, $partition, $partitionKeyResolver = 'partiti
         return $redis->lpop($partitionKey);
     }
 
-    private function getConnection()
+    public function getConnection()
     {
         $database = config('fair-queue.database');
         return Redis::connection($database);
     }
-}
+}