Implements system signal handling for archiving interruption / termination

core/CliMulti.php

@@ -95,6 +95,11 @@ class CliMulti
      */
     private $logger;
 
+    /**
+     * @var int|null
+     */
+    private $signal = null;
+
     public function __construct(LoggerInterface $logger = null)
     {
         $this->supportsAsync = $this->supportsAsync();
@@ -103,6 +108,26 @@ public function __construct(LoggerInterface $logger = null)
         $this->logger = $logger ?: new NullLogger();
     }
 
+    public function handleSignal(int $signal): void
+    {
+        $this->signal = $signal;
+
+        if (\SIGTERM !== $signal) {
+            return;
+        }
+
+        foreach ($this->processes as $process) {
+            if ($process instanceof ProcessSymfony) {
+                $this->logger->debug(
+                    'Aborting command: {command} [method = asyncCliSymfony]',
+                    ['command' => $process->getCommandLine()]
+                );
+
+                $process->stop(0);
+            }
+        }
+    }
+
     /**
      * It will request all given URLs in parallel (async) using the CLI and wait until all requests are finished.
      * If multi cli is not supported (eg windows) it will initiate an HTTP request instead (not async).
@@ -124,13 +149,21 @@ public function request(array $piwikUrls)
             }
         }
 
-        $chunks = array($piwikUrls);
+        $chunks = [$piwikUrls];
+
         if ($this->concurrentProcessesLimit) {
             $chunks = array_chunk($piwikUrls, $this->concurrentProcessesLimit);
         }
 
-        $results = array();
+        $results = [];
+
         foreach ($chunks as $urlsChunk) {
+            if (null !== $this->signal) {
+                $this->logSkippedRequests($urlsChunk);
+
+                continue;
+            }
+
             $results = array_merge($results, $this->requestUrls($urlsChunk));
         }
 
@@ -616,6 +649,16 @@ private function requestUrls(array $piwikUrls)
         return $results;
     }
 
+    private function logSkippedRequests(array $urls): void
+    {
+        foreach ($urls as $url) {
+            $this->logger->debug(
+                'Skipped climulti:request after abort signal received: {url}',
+                ['url' => $url]
+            );
+        }
+    }
+
     private static function getSuperUserTokenAuth()
     {
         return Piwik::requestTemporarySystemAuthToken('CliMultiNonAsyncArchive', 36);

core/CronArchive.php

@@ -36,6 +36,7 @@
 use Piwik\Plugins\UsersManager\API as APIUsersManager;
 use Piwik\Plugins\UsersManager\UserPreferences;
 use Piwik\Log\LoggerInterface;
+use Piwik\Scheduler\Scheduler;
 
 /**
  * ./console core:archive runs as a cron and is a useful tool for general maintenance,
@@ -230,6 +231,28 @@ class CronArchive
      */
     private $supportsAsync;
 
+    /**
+     * @var null|int
+     */
+    private $signal = null;
+
+    /**
+     * @var CliMulti|null
+     */
+    private $cliMultiHandler = null;
+
+    /**
+     * @var Scheduler|null
+     */
+    private $scheduler = null;
+
+    private $step = 0;
+
+    private const STEP_INIT = 1;
+    private const STEP_ARCHIVING = 2;
+    private const STEP_SCHEDULED_TASKS = 3;
+    private const STEP_FINISH = 4;
+
     /**
      * Constructor.
      *
@@ -272,16 +295,50 @@ public function main()
         $self = $this;
         Access::doAsSuperUser(function () use ($self) {
             try {
+                $this->step = self::STEP_INIT;
                 $self->init();
+
+                $this->step = self::STEP_ARCHIVING;
                 $self->run();
+
+                $this->step = self::STEP_SCHEDULED_TASKS;
                 $self->runScheduledTasks();
+
+                $this->step = self::STEP_FINISH;
                 $self->end();
             } catch (StopArchiverException $e) {
                 $this->logger->info("Archiving stopped by stop archiver exception" . $e->getMessage());
             }
         });
     }
 
+    public function handleSignal(int $signal): void
+    {
+        $this->logger->info('Received system signal to stop archiving: ' . $signal);
+
+        $this->signal = $signal;
+
+        // initialisation and finishing can be stopped directly.
+        if (in_array($this->step, [self::STEP_INIT, self::STEP_FINISH])) {
+            $this->logger->info('Archiving stopped');
+            exit;
+        }
+
+        // stop archiving
+        if (!empty($this->cliMultiHandler)) {
+            $this->logger->info('Trying to stop running cli processes...');
+            $this->cliMultiHandler->handleSignal($signal);
+        }
+
+        // stop scheduled tasks
+        if (!empty($this->scheduler)) {
+            $this->logger->info('Trying to stop running tasks...');
+            $this->scheduler->handleSignal($signal);
+        }
+
+        // Note: finishing the archiving process will be handled in `run()`
+    }
+
     public function init()
     {
         $this->segmentArchiving = StaticContainer::get(SegmentArchiving::class);
@@ -386,6 +443,11 @@ public function run()
         $queueConsumer->setMaxSitesToProcess($this->maxSitesToProcess);
 
         while (true) {
+            if (null !== $this->signal) {
+                $this->logger->info("Archiving will stop now because signal to abort received");
+                return;
+            }
+
             if ($this->isMaintenanceModeEnabled()) {
                 $this->logger->info("Archiving will stop now because maintenance mode is enabled");
                 return;
@@ -498,18 +560,29 @@ private function launchArchivingFor($archives, QueueConsumer $queueConsumer)
             return 0; // all URLs had no visits and were using the tracker
         }
 
-        $cliMulti = $this->makeCliMulti();
-        $cliMulti->timeRequests();
+        $this->cliMultiHandler = $this->makeCliMulti();
+        $this->cliMultiHandler->timeRequests();
 
-        $responses = $cliMulti->request($urls);
+        $responses = $this->cliMultiHandler->request($urls);
 
         $this->disconnectDb();
 
-        $timers = $cliMulti->getTimers();
+        $timers = $this->cliMultiHandler->getTimers();
         $successCount = 0;
 
         foreach ($urls as $index => $url) {
             $content = array_key_exists($index, $responses) ? $responses[$index] : null;
+
+            if (null !== $this->signal && empty($content)) {
+                // processes killed by system
+                $idinvalidation = $archivesBeingQueried[$index]['idinvalidation'];
+
+                $this->model->releaseInProgressInvalidation($idinvalidation);
+                $this->logger->info('Archiving process killed, reset invalidation with id ' . $idinvalidation);
+
+                continue;
+            }
+
             $checkInvalid = $this->checkResponse($content, $url);
 
             $stats = json_decode($content, $assoc = true);
@@ -527,7 +600,6 @@ private function launchArchivingFor($archives, QueueConsumer $queueConsumer)
 
             $visitsForPeriod = $this->getVisitsFromApiResponse($stats);
 
-
             $this->logArchiveJobFinished(
                 $url,
                 $timers[$index],
@@ -537,7 +609,6 @@ private function launchArchivingFor($archives, QueueConsumer $queueConsumer)
                 !$checkInvalid
             );
 
-
             $this->deleteInvalidatedArchives($archivesBeingQueried[$index]);
 
             $this->repairInvalidationsIfNeeded($archivesBeingQueried[$index]);
@@ -618,6 +689,11 @@ public function getErrors()
      */
     public function end()
     {
+        if (null !== $this->signal) {
+            // Skip if abort signal has been received
+            return;
+        }
+
         /**
          * This event is triggered after archiving.
          *
@@ -650,6 +726,11 @@ public function logFatalError($m)
 
     public function runScheduledTasks()
     {
+        if (null !== $this->signal) {
+            // Skip running scheduled task if abort signal has been received
+            return;
+        }
+
         $this->logSection("SCHEDULED TASKS");
 
         if ($this->disableScheduledTasks) {
@@ -673,7 +754,8 @@ public function runScheduledTasks()
         //         enable/disable the task
         Rules::$disablePureOutdatedArchive = true;
 
-        CoreAdminHomeAPI::getInstance()->runScheduledTasks();
+        $this->scheduler = StaticContainer::get(Scheduler::class);
+        $this->scheduler->run();
 
         $this->logSection("");
     }

core/Plugin/ConsoleCommand.php

@@ -10,6 +10,7 @@
 namespace Piwik\Plugin;
 
 use Symfony\Component\Console\Command\Command as SymfonyCommand;
+use Symfony\Component\Console\Command\SignalableCommandInterface;
 use Symfony\Component\Console\Exception\LogicException;
 use Symfony\Component\Console\Helper\ProgressBar;
 use Symfony\Component\Console\Helper\QuestionHelper;
@@ -28,7 +29,7 @@
  *
  * @api
  */
-class ConsoleCommand extends SymfonyCommand
+class ConsoleCommand extends SymfonyCommand implements SignalableCommandInterface
 {
     /**
      * @var ProgressBar|null
@@ -124,6 +125,43 @@ final public function run(InputInterface $input, OutputInterface $output): int
         return parent::run($input, $output);
     }
 
+    /**
+     * Method is final to make it impossible to overwrite it in plugin commands
+     * use getSystemSignalsToHandle() instead.
+     *
+     * @return array<int>
+     */
+    final public function getSubscribedSignals(): array
+    {
+        return $this->getSystemSignalsToHandle();
+    }
+
+    /**
+     * Method is final to make it impossible to overwrite it in plugin commands
+     * use handleSystemSignal() instead.
+     */
+    final public function handleSignal(int $signal): void
+    {
+        $this->handleSystemSignal($signal);
+    }
+
+    /**
+     * Returns the list of system signals to subscribe.
+     *
+     * @return array<int>
+     */
+    public function getSystemSignalsToHandle(): array
+    {
+        return [];
+    }
+
+    /**
+     * The method will be called when the application is signaled.
+     */
+    public function handleSystemSignal(int $signal): void
+    {
+    }
+
     /**
      * Adds a negatable option (e.g. --ansi / --no-ansi)
      *

core/Scheduler/Scheduler.php

@@ -81,6 +81,11 @@ class Scheduler
      */
     private $lock;
 
+    /**
+     * @var int|null
+     */
+    private $signal = null;
+
     public function __construct(TaskLoader $loader, LoggerInterface $logger, ScheduledTaskLock $lock)
     {
         $this->timetable = new Timetable();
@@ -89,6 +94,11 @@ public function __construct(TaskLoader $loader, LoggerInterface $logger, Schedul
         $this->lock = $lock;
     }
 
+    public function handleSignal(int $signal): void
+    {
+        $this->signal = $signal;
+    }
+
     /**
      * Executes tasks that are scheduled to run, then reschedules them.
      *
@@ -121,6 +131,11 @@ public function run()
 
             // loop through each task
             foreach ($tasks as $task) {
+                if (in_array($this->signal, [\SIGINT, \SIGTERM], true)) {
+                    $this->logger->info("Scheduler: Aborting due to received signal");
+                    return $executionResults;
+                }
+
                 // if the task does not have the current priority level, don't execute it yet
                 if ($task->getPriority() != $priority) {
                     continue;