JobQueue: Create a debugging queue

As a part of creating the kafka-based JobQueue implementation,
we need to transition to the new queue smoothly and be able to
first debug the event production side, then try executing a no-op
job in parallel with each of the real jobs (to be able to check
deduplication and all the jpb management code) and only then
try switching jobs one-by-one to the new backend queue.

In order to achieve that at first we need to temporary produce
both to the old job pipeline and to the new one - thus creating
a queue 'splitter' - it delegates all the methods to the main
queue while the produce requests also go to the EventBus.

This class it temporary, so it's marked deprecated right away.
It will not be needed once the queue transition is over.

Bug: T163380
Change-Id: I483b8f00f1c593ff3af6ba6826868b553339525f

diff --git a/autoload.php b/autoload.php
index a6128a4..e0cab4c 100644 (file)
--- a/autoload.php
+++ b/autoload.php
@@ -679,6 +679,7 @@ $wgAutoloadLocalClasses = [
        'JobQueueMemory' => __DIR__ . '/includes/jobqueue/JobQueueMemory.php',
        'JobQueueReadOnlyError' => __DIR__ . '/includes/jobqueue/JobQueue.php',
        'JobQueueRedis' => __DIR__ . '/includes/jobqueue/JobQueueRedis.php',
+       'JobQueueSecondTestQueue' => __DIR__ . '/includes/jobqueue/JobQueueSecondTestQueue.php',
        'JobRunner' => __DIR__ . '/includes/jobqueue/JobRunner.php',
        'JobSpecification' => __DIR__ . '/includes/jobqueue/JobSpecification.php',
        'JpegHandler' => __DIR__ . '/includes/media/Jpeg.php',

diff --git a/includes/jobqueue/JobQueueFederated.php b/includes/jobqueue/JobQueueFederated.php
index 7fdd617..44721d9 100644 (file)
--- a/includes/jobqueue/JobQueueFederated.php
+++ b/includes/jobqueue/JobQueueFederated.php
@@ -28,7 +28,7 @@
  * For example, one can set $wgJobTypeConf['refreshLinks'] to point to a
  * JobQueueFederated instance, which itself would consist of three JobQueueRedis
  * instances, each using their own redis server. This would allow for the jobs
- * to be split (evenly or based on weights) accross multiple servers if a single
+ * to be split (evenly or based on weights) across multiple servers if a single
  * server becomes impractical or expensive. Different JobQueue classes can be mixed.
  *
  * The basic queue configuration (e.g. "order", "claimTTL") of a federated queue

diff --git a/includes/jobqueue/JobQueueSecondTestQueue.php b/includes/jobqueue/JobQueueSecondTestQueue.php
new file mode 100644 (file)
index 0000000..a1935df
--- /dev/null
+++ b/includes/jobqueue/JobQueueSecondTestQueue.php
@@ -0,0 +1,280 @@
+<?php
+
+/**
+ * A wrapper for the JobQueue that delegates all the method calls to a single,
+ * main queue, and also pushes all the jobs to a second job queue that's being
+ * debugged.
+ *
+ * This class was temporary added to test transitioning to the JobQueueEventBus
+ * and will removed after the transition is completed. This code is only needed
+ * while we are testing the new infrastructure to be able to compare the results
+ * between the queue implementations and make sure that they process the same jobs,
+ * deduplicate correctly, compare the delays, backlogs and make sure no jobs are lost.
+ * When the new infrastructure is well tested this will not be needed any more.
+ *
+ * @deprecated since 1.30
+ * @since 1.30
+ */
+class JobQueueSecondTestQueue extends JobQueue {
+
+       /**
+        * @var JobQueue
+        */
+       private $mainQueue;
+
+       /**
+        * @var JobQueue
+        */
+       private $debugQueue;
+
+       protected function __construct( array $params ) {
+               if ( !isset( $params['mainqueue'] ) ) {
+                       throw new MWException( "mainqueue parameter must be provided to the debug queue" );
+               }
+
+               if ( !isset( $params['debugqueue'] ) ) {
+                       throw new MWException( "debugqueue parameter must be provided to the debug queue" );
+               }
+
+               $conf = [ 'wiki' => $params['wiki'], 'type' => $params['type'] ];
+               $this->mainQueue = JobQueue::factory( $params['mainqueue'] + $conf );
+               $this->debugQueue = JobQueue::factory( $params['debugqueue'] + $conf );
+
+               // We need to construct parent after creating the main and debug queue
+               // because super constructor calls some methods we delegate to the main queue.
+               parent::__construct( $params );
+       }
+
+       /**
+        * Get the allowed queue orders for configuration validation
+        *
+        * @return array Subset of (random, timestamp, fifo, undefined)
+        */
+       protected function supportedOrders() {
+               return $this->mainQueue->supportedOrders();
+       }
+
+       /**
+        * Get the default queue order to use if configuration does not specify one
+        *
+        * @return string One of (random, timestamp, fifo, undefined)
+        */
+       protected function optimalOrder() {
+               return $this->mainQueue->optimalOrder();
+       }
+
+       /**
+        * Find out if delayed jobs are supported for configuration validation
+        *
+        * @return bool Whether delayed jobs are supported
+        */
+       protected function supportsDelayedJobs() {
+               return $this->mainQueue->supportsDelayedJobs();
+       }
+
+       /**
+        * @see JobQueue::isEmpty()
+        * @return bool
+        */
+       protected function doIsEmpty() {
+               return $this->mainQueue->doIsEmpty();
+       }
+
+       /**
+        * @see JobQueue::getSize()
+        * @return int
+        */
+       protected function doGetSize() {
+               return $this->mainQueue->doGetSize();
+       }
+
+       /**
+        * @see JobQueue::getAcquiredCount()
+        * @return int
+        */
+       protected function doGetAcquiredCount() {
+               return $this->mainQueue->doGetAcquiredCount();
+       }
+
+       /**
+        * @see JobQueue::getDelayedCount()
+        * @return int
+        */
+       protected function doGetDelayedCount() {
+               return $this->mainQueue->doGetDelayedCount();
+       }
+
+       /**
+        * @see JobQueue::getAbandonedCount()
+        * @return int
+        */
+       protected function doGetAbandonedCount() {
+               return $this->mainQueue->doGetAbandonedCount();
+       }
+
+       /**
+        * @see JobQueue::batchPush()
+        * @param IJobSpecification[] $jobs
+        * @param int $flags
+        */
+       protected function doBatchPush( array $jobs, $flags ) {
+               $this->mainQueue->doBatchPush( $jobs, $flags );
+
+               try {
+                       $this->debugQueue->doBatchPush( $jobs, $flags );
+               } catch ( Exception $exception ) {
+                       MWExceptionHandler::logException( $exception );
+               }
+       }
+
+       /**
+        * @see JobQueue::pop()
+        * @return Job|bool
+        */
+       protected function doPop() {
+               return $this->mainQueue->doPop();
+       }
+
+       /**
+        * @see JobQueue::ack()
+        * @param Job $job
+        */
+       protected function doAck( Job $job ) {
+               return $this->mainQueue->doAck( $job );
+       }
+
+       /**
+        * @see JobQueue::deduplicateRootJob()
+        * @param IJobSpecification $job
+        * @throws MWException
+        * @return bool
+        */
+       protected function doDeduplicateRootJob( IJobSpecification $job ) {
+               return $this->mainQueue->doDeduplicateRootJob( $job );
+       }
+
+       /**
+        * @see JobQueue::isRootJobOldDuplicate()
+        * @param Job $job
+        * @return bool
+        */
+       protected function doIsRootJobOldDuplicate( Job $job ) {
+               return $this->mainQueue->doIsRootJobOldDuplicate( $job );
+       }
+
+       /**
+        * @param string $signature Hash identifier of the root job
+        * @return string
+        */
+       protected function getRootJobCacheKey( $signature ) {
+               return $this->mainQueue->getRootJobCacheKey( $signature );
+       }
+
+       /**
+        * @see JobQueue::delete()
+        * @throws MWException
+        */
+       protected function doDelete() {
+               return $this->mainQueue->doDelete();
+       }
+
+       /**
+        * @see JobQueue::waitForBackups()
+        * @return void
+        */
+       protected function doWaitForBackups() {
+               $this->mainQueue->doWaitForBackups();
+       }
+
+       /**
+        * @see JobQueue::flushCaches()
+        * @return void
+        */
+       protected function doFlushCaches() {
+               $this->mainQueue->doFlushCaches();
+       }
+
+       /**
+        * Get an iterator to traverse over all available jobs in this queue.
+        * This does not include jobs that are currently acquired or delayed.
+        * Note: results may be stale if the queue is concurrently modified.
+        *
+        * @return Iterator
+        * @throws JobQueueError
+        */
+       public function getAllQueuedJobs() {
+               return $this->mainQueue->getAllQueuedJobs();
+       }
+
+       /**
+        * Get an iterator to traverse over all delayed jobs in this queue.
+        * Note: results may be stale if the queue is concurrently modified.
+        *
+        * @return Iterator
+        * @throws JobQueueError
+        * @since 1.22
+        */
+       public function getAllDelayedJobs() {
+               return $this->mainQueue->getAllDelayedJobs();
+       }
+
+       /**
+        * Get an iterator to traverse over all claimed jobs in this queue
+        *
+        * Callers should be quick to iterator over it or few results
+        * will be returned due to jobs being acknowledged and deleted
+        *
+        * @return Iterator
+        * @throws JobQueueError
+        * @since 1.26
+        */
+       public function getAllAcquiredJobs() {
+               return $this->mainQueue->getAllAcquiredJobs();
+       }
+
+       /**
+        * Get an iterator to traverse over all abandoned jobs in this queue
+        *
+        * @return Iterator
+        * @throws JobQueueError
+        * @since 1.25
+        */
+       public function getAllAbandonedJobs() {
+               return $this->mainQueue->getAllAbandonedJobs();
+       }
+
+       /**
+        * Do not use this function outside of JobQueue/JobQueueGroup
+        *
+        * @return string
+        * @since 1.22
+        */
+       public function getCoalesceLocationInternal() {
+               return $this->mainQueue->getCoalesceLocationInternal();
+       }
+
+       /**
+        * @see JobQueue::getSiblingQueuesWithJobs()
+        * @param array $types List of queues types
+        * @return array|null (list of queue types) or null if unsupported
+        */
+       protected function doGetSiblingQueuesWithJobs( array $types ) {
+               return $this->mainQueue->doGetSiblingQueuesWithJobs( $types );
+       }
+
+       /**
+        * @see JobQueue::getSiblingQueuesSize()
+        * @param array $types List of queues types
+        * @return array|null (list of queue types) or null if unsupported
+        */
+       protected function doGetSiblingQueueSizes( array $types ) {
+               return $this->mainQueue->doGetSiblingQueueSizes( $types );
+       }
+
+       /**
+        * @throws JobQueueReadOnlyError
+        */
+       protected function assertNotReadOnly() {
+               $this->mainQueue->assertNotReadOnly();
+       }
+}