When wikis have pending jobs, create a Kubernetes job to work through the items

CHANGELOG.md

@@ -1,5 +1,8 @@
 # api
 
+## 8x.12.0 - TBD
+- Poll wikis for pending MediaWiki jobs and create Kubernetes jobs to process them if needed
+
 ## 8x.11.1 - 18 April 2023
 - Do not disable elastic search on wikis after a failure
 

app/Console/Kernel.php

@@ -8,6 +8,7 @@
 use App\Jobs\PruneEventPageUpdatesTable;
 use App\Jobs\PruneQueryserviceBatchesTable;
 use App\Jobs\SandboxCleanupJob;
+use App\Jobs\PollForMediaWikiJobsJob;
 use Illuminate\Console\Scheduling\Schedule;
 use Illuminate\Foundation\Console\Kernel as ConsoleKernel;
 use App\Jobs\PlatformStatsSummaryJob;
@@ -43,6 +44,8 @@ protected function schedule(Schedule $schedule)
 
         // Schedule site stat updates for each wiki and platform-summary
         $schedule->command('schedule:stats')->daily();
+
+        $schedule->job(new PollForMediaWikiJobsJob)->everyMinute();
     }
 
     /**

app/Jobs/PollForMediaWikiJobsJob.php

@@ -0,0 +1,45 @@
+<?php
+
+namespace App\Jobs;
+
+use App\Wiki;
+use Illuminate\Support\Facades\Http;
+use Illuminate\Support\Facades\Log;
+
+class PollForMediaWikiJobsJob extends Job
+{
+    public function handle (): void
+    {
+        $allWikiDomains = Wiki::all()->pluck('domain');
+        foreach ($allWikiDomains as $wikiDomain) {
+            if ($this->hasPendingJobs($wikiDomain)) {
+                $this->enqueueWiki($wikiDomain);
+            }
+        }
+    }
+
+    private function hasPendingJobs (string $wikiDomain): bool
+    {
+        $response = Http::withHeaders([
+            'host' => $wikiDomain
+        ])->get(
+            getenv('PLATFORM_MW_BACKEND_HOST').'/w/api.php?action=query&meta=siteinfo&siprop=statistics&format=json'
+        );
+
+        if ($response->failed()) {
+            $this->job->markAsFailed();
+            Log::error(
+                'Failure polling wiki '.$wikiDomain.' for pending MediaWiki jobs: '.$response->clientError()
+            );
+            return false;
+        }
+
+        $pendingJobsCount = data_get($response->json(), 'query.statistics.jobs', 0);
+        return $pendingJobsCount > 0;
+    }
+
+    private function enqueueWiki (string $wikiDomain): void
+    {
+        dispatch(new ProcessMediaWikiJobsJob($wikiDomain));
+    }
+}

app/Jobs/ProcessMediaWikiJobsJob.php

@@ -0,0 +1,118 @@
+<?php
+
+namespace App\Jobs;
+
+use Illuminate\Bus\Queueable;
+use Illuminate\Contracts\Queue\ShouldQueue;
+use Illuminate\Contracts\Queue\ShouldBeUnique;
+use Illuminate\Queue\InteractsWithQueue;
+use Illuminate\Support\Facades\Log;
+use Maclof\Kubernetes\Client;
+use Maclof\Kubernetes\Models\Job as KubernetesJob;
+
+class ProcessMediaWikiJobsJob implements ShouldQueue, ShouldBeUnique
+{
+    use InteractsWithQueue, Queueable;
+
+    private string $wikiDomain;
+    private string $jobsKubernetesNamespace;
+
+    public function __construct (string $wikiDomain)
+    {
+        $this->wikiDomain = $wikiDomain;
+        $this->jobsKubernetesNamespace = env('API_JOB_NAMESPACE', 'api-jobs');
+    }
+
+    public function uniqueId(): string
+    {
+        return $this->wikiDomain;
+    }
+
+    public function handle (Client $kubernetesClient): void
+    {
+        $kubernetesClient->setNamespace('default');
+        $mediawikiPod = $kubernetesClient->pods()->setFieldSelector([
+            'status.phase' => 'Running'
+        ])->setLabelSelector([
+            'app.kubernetes.io/name' => 'mediawiki',
+            'app.kubernetes.io/component' => 'app-backend'
+        ])->first();
+
+        if ($mediawikiPod === null) {
+            $this->fail(
+                new \RuntimeException(
+                    'Unable to find a running MediaWiki pod in the cluster, '.
+                    'cannot continue.'
+                )
+            );
+            return;
+        }
+        $mediawikiPod = $mediawikiPod->toArray();
+
+        $kubernetesClient->setNamespace($this->jobsKubernetesNamespace);
+        $jobSpec = new KubernetesJob([
+            'metadata' => [
+                'name' => 'run-all-mw-jobs-'.hash('sha1', $this->wikiDomain),
+                'namespace' => $this->jobsKubernetesNamespace,
+                'labels' => [
+                    'app.kubernetes.io/instance' => $this->wikiDomain,
+                    'app.kubernetes.io/name' => 'run-all-mw-jobs'
+                ]
+            ],
+            'spec' => [
+                'ttlSecondsAfterFinished' => 0,
+                'template' => [
+                    'metadata' => [
+                        'name' => 'run-all-mw-jobs'
+                    ],
+                    'spec' => [
+                        'containers' => [
+                            0 => [
+                                'name' => 'run-all-mw-jobs',
+                                'image' => $mediawikiPod['spec']['containers'][0]['image'],
+                                'env' => array_merge(
+                                    $mediawikiPod['spec']['containers'][0]['env'],
+                                    [['name' => 'WBS_DOMAIN', 'value' => $this->wikiDomain]]
+                                ),
+                                'command' => [
+                                    0 => 'bash',
+                                    1 => '-c',
+                                    2 => <<<'CMD'
+                                    JOBS_TO_GO=1
+                                    while [ "$JOBS_TO_GO" != "0" ]
+                                    do
+                                        echo "Running 1000 jobs"
+                                        php w/maintenance/runJobs.php --maxjobs 1000
+                                        echo Waiting for 1 seconds...
+                                        sleep 1
+                                        JOBS_TO_GO=$(php w/maintenance/showJobs.php | tr -d '[:space:]')
+                                        echo $JOBS_TO_GO jobs to go
+                                    done
+                                    CMD
+                                ],
+                            ]
+                        ],
+                        'restartPolicy' => 'Never'
+                    ]
+                ]
+            ]
+        ]);
+
+        $job = $kubernetesClient->jobs()->apply($jobSpec);
+        $jobName = data_get($job, 'metadata.name');
+        if (!$jobName) {
+            // The k8s client does not fail reliably on 4xx responses, so checking the name
+            // currently serves as poor man's error handling.
+            $this->fail(
+                new \RuntimeException('Job creation for wiki "'.$this->wikiDomain.'" failed.')
+            );
+            return;
+        }
+        Log::info(
+            'MediaWiki Job for wiki "'.$this->wikiDomain.'" exists or was created with name "'.$jobName.'".'
+        );
+
+        return;
+    }
+
+}

config/queue.php

@@ -63,10 +63,9 @@
             'driver' => 'redis',
             'connection' => 'default',
             'queue' => env('REDIS_QUEUE', 'default'),
-            'retry_after' => 90,
+            'retry_after' => 100,
             'block_for' => null,
         ],
-
     ],
 
     /*

phpunit.xml

@@ -14,5 +14,6 @@
     <env name="APP_ENV" value="testing"/>
     <env name="CACHE_DRIVER" value="array"/>
     <env name="QUEUE_CONNECTION" value="sync"/>
+    <env name="PLATFORM_MW_BACKEND_HOST" value="mediawiki-139-app-backend.default.svc.cluster.default"/>
   </php>
 </phpunit>

tests/Jobs/PollForMediaWikiJobsJobTest.php

@@ -0,0 +1,94 @@
+<?php
+
+namespace Tests\Jobs;
+
+use App\Wiki;
+use App\Jobs\PollForMediaWikiJobsJob;
+use App\Jobs\ProcessMediaWikiJobsJob;
+use Tests\TestCase;
+use Illuminate\Contracts\Queue\Job;
+use Illuminate\Foundation\Testing\RefreshDatabase;
+use Illuminate\Support\Facades\Http;
+use Illuminate\Support\Facades\Bus;
+use Illuminate\Database\Eloquent\Model;
+
+class PollForMediaWikiJobsJobTest extends TestCase
+{
+
+    use RefreshDatabase;
+
+    private Model $wiki;
+
+    public function setUp(): void
+    {
+        parent::setUp();
+        $this->wiki = Wiki::factory()->create();
+    }
+
+    public function testNoJobs()
+    {
+        Http::fake([
+            getenv('PLATFORM_MW_BACKEND_HOST').'/w/api.php?action=query&meta=siteinfo&siprop=statistics&format=json' => Http::response([
+                'query' => [
+                    'statistics' => [
+                        'jobs' => 0
+                    ]
+                ]
+            ], 200)
+        ]);
+
+        Bus::fake();
+        $mockJob = $this->createMock(Job::class);
+        $job = new PollForMediaWikiJobsJob();
+        $job->setJob($mockJob);
+
+        $mockJob->expects($this->never())->method('fail');
+        $mockJob->expects($this->never())->method('markAsFailed');
+        $job->handle();
+        Bus::assertNothingDispatched();
+    }
+
+    public function testWithJobs()
+    {
+        Http::fake([
+            getenv('PLATFORM_MW_BACKEND_HOST').'/w/api.php?action=query&meta=siteinfo&siprop=statistics&format=json' => Http::response([
+                'query' => [
+                    'statistics' => [
+                        'jobs' => 3
+                    ]
+                ]
+            ], 200)
+        ]);
+        Bus::fake();
+
+        $mockJob = $this->createMock(Job::class);
+
+        $job = new PollForMediaWikiJobsJob();
+        $job->setJob($mockJob);
+
+        $mockJob->expects($this->never())->method('fail');
+        $mockJob->expects($this->never())->method('markAsFailed');
+        $job->handle();
+        Bus::assertDispatched(ProcessMediaWikiJobsJob::class);
+    }
+
+    public function testWithFailure()
+    {
+        Http::fake([
+            getenv('PLATFORM_MW_BACKEND_HOST').'/w/api.php?action=query&meta=siteinfo&siprop=statistics&format=json' => Http::response([
+                'error' => 'Something went wrong'
+            ], 500)
+        ]);
+        Bus::fake();
+
+        $mockJob = $this->createMock(Job::class);
+
+        $job = new PollForMediaWikiJobsJob();
+        $job->setJob($mockJob);
+
+        $mockJob->expects($this->once())->method('markAsFailed');
+        $mockJob->expects($this->never())->method('fail');
+        $job->handle();
+        Bus::assertNothingDispatched();
+    }
+}

tests/Jobs/ProcessMediaWikiJobsJobTest.php

@@ -0,0 +1,86 @@
+<?php
+
+namespace Tests\Jobs;
+
+use Illuminate\Foundation\Testing\RefreshDatabase;
+use Tests\TestCase;
+use Illuminate\Contracts\Queue\Job;
+use App\Jobs\ProcessMediaWikiJobsJob;
+use Maclof\Kubernetes\Client;
+use Http\Adapter\Guzzle6\Client as Guzzle6Client;
+use GuzzleHttp\HandlerStack;
+use GuzzleHttp\Handler\MockHandler;
+use GuzzleHttp\Psr7\Response;
+
+class ProcessMediaWikiJobsJobTest extends TestCase
+{
+    use RefreshDatabase;
+
+    public function testJobFailOnNoMediaWikiPod()
+    {
+        $mockJob = $this->createMock(Job::class);
+        $mockJob->expects($this->once())->method('fail');
+
+        $job = new ProcessMediaWikiJobsJob('test.wikibase.cloud');
+        $job->setJob($mockJob);
+
+        $mock = new MockHandler([
+            new Response(200, [], json_encode([ 'items' => [] ])),
+        ]);
+
+        $handlerStack = HandlerStack::create($mock);
+        $mockGuzzle = Guzzle6Client::createWithConfig([
+            'handler' => $handlerStack,
+            'verify' => '/var/run/secrets/kubernetes.io/serviceaccount/ca.crt',
+        ]);
+
+        $job->handle(new Client([
+            'master' => 'https://kubernetes.default.svc',
+            'token' => '/var/run/secrets/kubernetes.io/serviceaccount/token',
+        ], null, $mockGuzzle));
+    }
+
+    public function testJobDoesNotFail()
+    {
+        $mockJob = $this->createMock(Job::class);
+        $mockJob->expects($this->never())->method('fail');
+
+        $job = new ProcessMediaWikiJobsJob('test.wikibase.cloud');
+        $job->setJob($mockJob);
+
+        $mock = new MockHandler([
+            new Response(200, [], json_encode([ 'items' => [
+                [
+                    'kind' => 'Pod',
+                    'spec' => [
+                        'containers' => [
+                            [
+                                'image' => 'helloworld',
+                                'env' => [
+                                    'SOMETHING' => 'something'
+                                ]
+                            ]
+                        ]
+                    ]
+                ]
+            ]])),
+            new Response(200, [], json_encode([ 'items' => [] ])),
+            new Response(201, [], json_encode([
+                'metadata' => [
+                    'name' => 'some-job-name'
+                ]
+            ]))
+        ]);
+
+        $handlerStack = HandlerStack::create($mock);
+        $mockGuzzle = Guzzle6Client::createWithConfig([
+            'handler' => $handlerStack,
+            'verify' => '/var/run/secrets/kubernetes.io/serviceaccount/ca.crt',
+        ]);
+
+        $job->handle(new Client([
+            'master' => 'https://kubernetes.default.svc',
+            'token' => '/var/run/secrets/kubernetes.io/serviceaccount/token',
+        ], null, $mockGuzzle));
+    }
+}