diff -Naur a/vendor/magento/module-cron/Observer/ProcessCronQueueObserver.php b/vendor/magento/module-cron/Observer/ProcessCronQueueObserver.php --- a/vendor/magento/module-cron/Observer/ProcessCronQueueObserver.php +++ b/vendor/magento/module-cron/Observer/ProcessCronQueueObserver.php @@ -13,6 +13,8 @@ use Magento\Framework\App\State; use Magento\Framework\Console\Cli; use Magento\Framework\Event\ObserverInterface; use \Magento\Cron\Model\Schedule; +use Magento\Framework\Profiler\Driver\Standard\Stat; +use Magento\Framework\Profiler\Driver\Standard\StatFactory; /** * @SuppressWarnings(PHPMD.CouplingBetweenObjects) @@ -127,6 +129,11 @@ class ProcessCronQueueObserver implements ObserverInterface private $jobs; /** + * @var Stat + */ + private $statProfiler; + + /** * @param \Magento\Framework\ObjectManagerInterface $objectManager * @param \Magento\Cron\Model\ScheduleFactory $scheduleFactory * @param \Magento\Framework\App\CacheInterface $cache @@ -138,6 +145,7 @@ class ProcessCronQueueObserver implements ObserverInterface * @param \Magento\Framework\Process\PhpExecutableFinderFactory $phpExecutableFinderFactory * @param \Psr\Log\LoggerInterface $logger * @param \Magento\Framework\App\State $state + * @param StatFactory $statFactory * @SuppressWarnings(PHPMD.ExcessiveParameterList) */ public function __construct( @@ -151,7 +159,8 @@ class ProcessCronQueueObserver implements ObserverInterface \Magento\Framework\Stdlib\DateTime\DateTime $dateTime, \Magento\Framework\Process\PhpExecutableFinderFactory $phpExecutableFinderFactory, \Psr\Log\LoggerInterface $logger, - \Magento\Framework\App\State $state + \Magento\Framework\App\State $state, + StatFactory $statFactory ) { $this->_objectManager = $objectManager; $this->_scheduleFactory = $scheduleFactory; @@ -164,6 +173,7 @@ class ProcessCronQueueObserver implements ObserverInterface $this->phpExecutableFinder = $phpExecutableFinderFactory->create(); $this->logger = $logger; $this->state = $state; + $this->statProfiler = $statFactory->create(); } /** @@ -179,27 +189,26 @@ class ProcessCronQueueObserver implements ObserverInterface */ public function execute(\Magento\Framework\Event\Observer $observer) { - $pendingJobs = $this->_getPendingSchedules(); + $currentTime = $this->dateTime->gmtTimestamp(); $jobGroupsRoot = $this->_config->getJobs(); + // sort jobs groups to start from used in separated process + uksort( + $jobGroupsRoot, + function ($a, $b) { + return $this->getCronGroupConfigurationValue($b, 'use_separate_process') + - $this->getCronGroupConfigurationValue($a, 'use_separate_process'); + } + ); $phpPath = $this->phpExecutableFinder->find() ?: 'php'; foreach ($jobGroupsRoot as $groupId => $jobsRoot) { - $this->_cleanup($groupId); - $this->_generate($groupId); - if ($this->_request->getParam('group') !== null - && $this->_request->getParam('group') !== '\'' . ($groupId) . '\'' - && $this->_request->getParam('group') !== $groupId - ) { + if (!$this->isGroupInFilter($groupId)) { continue; } - if (($this->_request->getParam(self::STANDALONE_PROCESS_STARTED) !== '1') && ( - $this->_scopeConfig->getValue( - 'system/cron/' . $groupId . '/use_separate_process', - \Magento\Store\Model\ScopeInterface::SCOPE_STORE - ) == 1 - ) + if ($this->_request->getParam(self::STANDALONE_PROCESS_STARTED) !== '1' + && $this->getCronGroupConfigurationValue($groupId, 'use_separate_process') == 1 ) { $this->_shell->execute( $phpPath . ' %s cron:run --group=' . $groupId . ' --' . Cli::INPUT_KEY_BOOTSTRAP . '=' @@ -211,42 +220,9 @@ class ProcessCronQueueObserver implements ObserverInterface continue; } - /** @var \Magento\Cron\Model\Schedule $schedule */ - foreach ($pendingJobs as $schedule) { - $jobConfig = isset($jobsRoot[$schedule->getJobCode()]) ? $jobsRoot[$schedule->getJobCode()] : null; - if (!$jobConfig) { - continue; - } - - $scheduledTime = strtotime($schedule->getScheduledAt()); - if ($scheduledTime > $currentTime) { - continue; - } - - try { - if ($schedule->tryLockJob()) { - $this->_runJob($scheduledTime, $currentTime, $jobConfig, $schedule, $groupId); - } - } catch (\Exception $e) { - $schedule->setMessages($e->getMessage()); - if ($schedule->getStatus() === Schedule::STATUS_ERROR) { - $this->logger->critical($e); - } - if ($schedule->getStatus() === Schedule::STATUS_MISSED - && $this->state->getMode() === State::MODE_DEVELOPER - ) { - $this->logger->info( - sprintf( - "%s Schedule Id: %s Job Code: %s", - $schedule->getMessages(), - $schedule->getScheduleId(), - $schedule->getJobCode() - ) - ); - } - } - $schedule->save(); - } + $this->cleanupJobs($groupId, $currentTime); + $this->generateSchedules($groupId); + $this->processPendingJobs($groupId, $jobsRoot, $currentTime); } } @@ -263,24 +239,25 @@ class ProcessCronQueueObserver implements ObserverInterface */ protected function _runJob($scheduledTime, $currentTime, $jobConfig, $schedule, $groupId) { - $scheduleLifetime = (int)$this->_scopeConfig->getValue( - 'system/cron/' . $groupId . '/' . self::XML_PATH_SCHEDULE_LIFETIME, - \Magento\Store\Model\ScopeInterface::SCOPE_STORE - ); + $jobCode = $schedule->getJobCode(); + $scheduleLifetime = $this->getCronGroupConfigurationValue($groupId, self::XML_PATH_SCHEDULE_LIFETIME); $scheduleLifetime = $scheduleLifetime * self::SECONDS_IN_MINUTE; if ($scheduledTime < $currentTime - $scheduleLifetime) { $schedule->setStatus(Schedule::STATUS_MISSED); + $this->logger->info(sprintf('Cron Job %s is missed', $jobCode)); throw new \Exception('Too late for the schedule'); } if (!isset($jobConfig['instance'], $jobConfig['method'])) { $schedule->setStatus(Schedule::STATUS_ERROR); + $this->logger->error(sprintf('Cron Job %s has an error', $jobCode)); throw new \Exception('No callbacks found'); } $model = $this->_objectManager->create($jobConfig['instance']); $callback = [$model, $jobConfig['method']]; if (!is_callable($callback)) { $schedule->setStatus(Schedule::STATUS_ERROR); + $this->logger->error(sprintf('Cron Job %s has an error', $jobCode)); throw new \Exception( sprintf('Invalid callback: %s::%s can\'t be called', $jobConfig['instance'], $jobConfig['method']) ); @@ -288,17 +265,65 @@ class ProcessCronQueueObserver implements ObserverInterface $schedule->setExecutedAt(strftime('%Y-%m-%d %H:%M:%S', $this->dateTime->gmtTimestamp()))->save(); + $this->startProfiling(); try { + $this->logger->info(sprintf('Cron Job %s is run', $jobCode)); call_user_func_array($callback, [$schedule]); } catch (\Exception $e) { $schedule->setStatus(Schedule::STATUS_ERROR); + $this->logger->error(sprintf( + 'Cron Job %s has an error. Statistics: %s %s', + $jobCode, + $this->getProfilingStat(), $e->getMessage() + )); throw $e; + } finally { + $this->stopProfiling(); } $schedule->setStatus(Schedule::STATUS_SUCCESS)->setFinishedAt(strftime( '%Y-%m-%d %H:%M:%S', $this->dateTime->gmtTimestamp() )); + + $this->logger->info(sprintf( + 'Cron Job %s is successfully finished. Statistics: %s', + $jobCode, + $this->getProfilingStat() + )); + } + + /** + * Starts profiling + * + * @return void + */ + private function startProfiling() + { + $this->statProfiler->clear(); + $this->statProfiler->start('job', microtime(true), memory_get_usage(true), memory_get_usage()); + } + + /** + * Stops profiling + * + * @return void + */ + private function stopProfiling() + { + $this->statProfiler->stop('job', microtime(true), memory_get_usage(true), memory_get_usage()); + } + + /** + * Retrieves statistics in the JSON format + * + * @return string + */ + private function getProfilingStat() + { + $stat = $this->statProfiler->get('job'); + unset($stat[Stat::START]); + return json_encode($stat); } /** @@ -306,15 +331,13 @@ class ProcessCronQueueObserver implements ObserverInterface * * @return \Magento\Cron\Model\ResourceModel\Schedule\Collection */ - protected function _getPendingSchedules() + private function getPendingSchedules($groupId) { - if (!$this->_pendingSchedules) { - $this->_pendingSchedules = $this->_scheduleFactory->create()->getCollection()->addFieldToFilter( - 'status', - Schedule::STATUS_PENDING - )->load(); - } - return $this->_pendingSchedules; + $jobs = $this->getJobs(); + $pendingJobs = $this->_scheduleFactory->create()->getCollection(); + $pendingJobs->addFieldToFilter('status', Schedule::STATUS_PENDING); + $pendingJobs->addFieldToFilter('job_code', ['in' => array_keys($jobs[$groupId])]); + return $pendingJobs; } /** @@ -323,22 +346,32 @@ class ProcessCronQueueObserver implements ObserverInterface * @param string $groupId * @return $this */ - protected function _generate($groupId) + private function generateSchedules($groupId) { /** * check if schedule generation is needed */ $lastRun = (int)$this->_cache->load(self::CACHE_KEY_LAST_SCHEDULE_GENERATE_AT . $groupId); - $rawSchedulePeriod = (int)$this->_scopeConfig->getValue( - 'system/cron/' . $groupId . '/' . self::XML_PATH_SCHEDULE_GENERATE_EVERY, - \Magento\Store\Model\ScopeInterface::SCOPE_STORE + $rawSchedulePeriod = (int)$this->getCronGroupConfigurationValue( + $groupId, + self::XML_PATH_SCHEDULE_GENERATE_EVERY ); $schedulePeriod = $rawSchedulePeriod * self::SECONDS_IN_MINUTE; if ($lastRun > $this->dateTime->gmtTimestamp() - $schedulePeriod) { return $this; } - $schedules = $this->_getPendingSchedules(); + /** + * save time schedules generation was ran with no expiration + */ + $this->_cache->save( + $this->dateTime->gmtTimestamp(), + self::CACHE_KEY_LAST_SCHEDULE_GENERATE_AT . $groupId, + ['crontab'], + null + ); + + $schedules = $this->getPendingSchedules($groupId); $exists = []; /** @var Schedule $schedule */ foreach ($schedules as $schedule) { @@ -353,16 +386,6 @@ class ProcessCronQueueObserver implements ObserverInterface $this->_generateJobs($jobs[$groupId], $exists, $groupId); $this->cleanupScheduleMismatches(); - /** - * save time schedules generation was ran with no expiration - */ - $this->_cache->save( - $this->dateTime->gmtTimestamp(), - self::CACHE_KEY_LAST_SCHEDULE_GENERATE_AT . $groupId, - ['crontab'], - null - ); - return $this; } @@ -372,7 +395,7 @@ class ProcessCronQueueObserver implements ObserverInterface * @param array $jobs * @param array $exists * @param string $groupId - * @return $this + * @return void */ protected function _generateJobs($jobs, $exists, $groupId) { @@ -385,77 +408,60 @@ class ProcessCronQueueObserver implements ObserverInterface $timeInterval = $this->getScheduleTimeInterval($groupId); $this->saveSchedule($jobCode, $cronExpression, $timeInterval, $exists); } - return $this; } /** * Clean expired jobs * - * @param string $groupId - * @return $this + * @param $groupId + * @param $currentTime + * @return void */ - protected function _cleanup($groupId) + private function cleanupJobs($groupId, $currentTime) { - $this->cleanupDisabledJobs($groupId); - // check if history cleanup is needed $lastCleanup = (int)$this->_cache->load(self::CACHE_KEY_LAST_HISTORY_CLEANUP_AT . $groupId); - $historyCleanUp = (int)$this->_scopeConfig->getValue( - 'system/cron/' . $groupId . '/' . self::XML_PATH_HISTORY_CLEANUP_EVERY, - \Magento\Store\Model\ScopeInterface::SCOPE_STORE - ); + $historyCleanUp = (int)$this->getCronGroupConfigurationValue($groupId, self::XML_PATH_HISTORY_CLEANUP_EVERY); if ($lastCleanup > $this->dateTime->gmtTimestamp() - $historyCleanUp * self::SECONDS_IN_MINUTE) { return $this; } - - // check how long the record should stay unprocessed before marked as MISSED - $scheduleLifetime = (int)$this->_scopeConfig->getValue( - 'system/cron/' . $groupId . '/' . self::XML_PATH_SCHEDULE_LIFETIME, - \Magento\Store\Model\ScopeInterface::SCOPE_STORE + // save time history cleanup was ran with no expiration + $this->_cache->save( + $this->dateTime->gmtTimestamp(), + self::CACHE_KEY_LAST_HISTORY_CLEANUP_AT . $groupId, + ['crontab'], + null ); - $scheduleLifetime = $scheduleLifetime * self::SECONDS_IN_MINUTE; - /** - * @var \Magento\Cron\Model\ResourceModel\Schedule\Collection $history - */ - $history = $this->_scheduleFactory->create()->getCollection()->addFieldToFilter( - 'status', - ['in' => [Schedule::STATUS_SUCCESS, Schedule::STATUS_MISSED, Schedule::STATUS_ERROR]] - )->load(); + $this->cleanupDisabledJobs($groupId); - $historySuccess = (int)$this->_scopeConfig->getValue( - 'system/cron/' . $groupId . '/' . self::XML_PATH_HISTORY_SUCCESS, - \Magento\Store\Model\ScopeInterface::SCOPE_STORE - ); - $historyFailure = (int)$this->_scopeConfig->getValue( - 'system/cron/' . $groupId . '/' . self::XML_PATH_HISTORY_FAILURE, - \Magento\Store\Model\ScopeInterface::SCOPE_STORE - ); + $historySuccess = (int)$this->getCronGroupConfigurationValue($groupId, self::XML_PATH_HISTORY_SUCCESS); + $historyFailure = (int)$this->getCronGroupConfigurationValue($groupId, self::XML_PATH_HISTORY_FAILURE); $historyLifetimes = [ Schedule::STATUS_SUCCESS => $historySuccess * self::SECONDS_IN_MINUTE, Schedule::STATUS_MISSED => $historyFailure * self::SECONDS_IN_MINUTE, Schedule::STATUS_ERROR => $historyFailure * self::SECONDS_IN_MINUTE, + Schedule::STATUS_PENDING => max($historyFailure, $historySuccess) * self::SECONDS_IN_MINUTE, ]; - $now = $this->dateTime->gmtTimestamp(); - /** @var Schedule $record */ - foreach ($history as $record) { - $checkTime = $record->getExecutedAt() ? strtotime($record->getExecutedAt()) : - strtotime($record->getScheduledAt()) + $scheduleLifetime; - if ($checkTime < $now - $historyLifetimes[$record->getStatus()]) { - $record->delete(); - } + $jobs = $this->getJobs()[$groupId]; + $scheduleResource = $this->_scheduleFactory->create()->getResource(); + $connection = $scheduleResource->getConnection(); + $count = 0; + foreach ($historyLifetimes as $status => $time) { + $count += $connection->delete( + $scheduleResource->getMainTable(), + [ + 'status = ?' => $status, + 'job_code in (?)' => array_keys($jobs), + 'created_at < ?' => $connection->formatDate($currentTime - $time) + ] + ); } - // save time history cleanup was ran with no expiration - $this->_cache->save( - $this->dateTime->gmtTimestamp(), - self::CACHE_KEY_LAST_HISTORY_CLEANUP_AT . $groupId, - ['crontab'], - null - ); - - return $this; + if ($count) { + $this->logger->info(sprintf('%d cron jobs were cleaned', $count)); + } } /** @@ -486,7 +492,7 @@ class ProcessCronQueueObserver implements ObserverInterface for ($time = $currentTime; $time < $timeAhead; $time += self::SECONDS_IN_MINUTE) { $scheduledAt = strftime('%Y-%m-%d %H:%M:00', $time); $alreadyScheduled = !empty($exists[$jobCode . '/' . $scheduledAt]); - $schedule = $this->generateSchedule($jobCode, $cronExpression, $time); + $schedule = $this->createSchedule($jobCode, $cronExpression, $time); $valid = $schedule->trySchedule(); if (!$valid) { if ($alreadyScheduled) { @@ -510,7 +516,7 @@ class ProcessCronQueueObserver implements ObserverInterface * @param int $time * @return Schedule */ - protected function generateSchedule($jobCode, $cronExpression, $time) + protected function createSchedule($jobCode, $cronExpression, $time) { $schedule = $this->_scheduleFactory->create() ->setCronExpr($cronExpression) @@ -528,10 +534,7 @@ class ProcessCronQueueObserver implements ObserverInterface */ protected function getScheduleTimeInterval($groupId) { - $scheduleAheadFor = (int)$this->_scopeConfig->getValue( - 'system/cron/' . $groupId . '/' . self::XML_PATH_SCHEDULE_AHEAD_FOR, - \Magento\Store\Model\ScopeInterface::SCOPE_STORE - ); + $scheduleAheadFor = (int)$this->getCronGroupConfigurationValue($groupId, self::XML_PATH_SCHEDULE_AHEAD_FOR); $scheduleAheadFor = $scheduleAheadFor * self::SECONDS_IN_MINUTE; return $scheduleAheadFor; @@ -547,16 +550,26 @@ class ProcessCronQueueObserver implements ObserverInterface private function cleanupDisabledJobs($groupId) { $jobs = $this->getJobs(); + $jobsToCleanup = []; foreach ($jobs[$groupId] as $jobCode => $jobConfig) { if (!$this->getCronExpression($jobConfig)) { /** @var \Magento\Cron\Model\ResourceModel\Schedule $scheduleResource */ - $scheduleResource = $this->_scheduleFactory->create()->getResource(); - $scheduleResource->getConnection()->delete($scheduleResource->getMainTable(), [ - 'status=?' => Schedule::STATUS_PENDING, - 'job_code=?' => $jobCode, - ]); + $jobsToCleanup[] = $jobCode; } } + + if (count($jobsToCleanup) > 0) { + $scheduleResource = $this->_scheduleFactory->create()->getResource(); + $count = $scheduleResource->getConnection()->delete( + $scheduleResource->getMainTable(), + [ + 'status = ?' => Schedule::STATUS_PENDING, + 'job_code in (?)' => $jobsToCleanup, + ] + ); + + $this->logger->info(sprintf('%d cron jobs were cleaned', $count)); + } } /** @@ -586,12 +599,12 @@ class ProcessCronQueueObserver implements ObserverInterface */ private function cleanupScheduleMismatches() { + /** @var \Magento\Cron\Model\ResourceModel\Schedule $scheduleResource */ + $scheduleResource = $this->_scheduleFactory->create()->getResource(); foreach ($this->invalid as $jobCode => $scheduledAtList) { - /** @var \Magento\Cron\Model\ResourceModel\Schedule $scheduleResource */ - $scheduleResource = $this->_scheduleFactory->create()->getResource(); $scheduleResource->getConnection()->delete($scheduleResource->getMainTable(), [ - 'status=?' => Schedule::STATUS_PENDING, - 'job_code=?' => $jobCode, + 'status = ?' => Schedule::STATUS_PENDING, + 'job_code = ?' => $jobCode, 'scheduled_at in (?)' => $scheduledAtList, ]); } @@ -608,4 +621,87 @@ class ProcessCronQueueObserver implements ObserverInterface } return $this->jobs; } + + /** + * Get CronGroup Configuration Value + * + * @param $groupId + * @return int + */ + private function getCronGroupConfigurationValue($groupId, $path) + { + return $this->_scopeConfig->getValue( + 'system/cron/' . $groupId . '/' . $path, + \Magento\Store\Model\ScopeInterface::SCOPE_STORE + ); + return $scheduleLifetime; + } + + /** + * Is Group In Filter + * + * @param $groupId + * @return bool + */ + private function isGroupInFilter($groupId): bool + { + return !($this->_request->getParam('group') !== null + && trim($this->_request->getParam('group'), "'") !== $groupId); + } + + /** + * Process pending jobs + * + * @param $groupId + * @param $jobsRoot + * @param $currentTime + */ + private function processPendingJobs($groupId, $jobsRoot, $currentTime) + { + $procesedJobs = []; + $pendingJobs = $this->getPendingSchedules($groupId); + /** @var \Magento\Cron\Model\Schedule $schedule */ + foreach ($pendingJobs as $schedule) { + if (isset($procesedJobs[$schedule->getJobCode()])) { + // process only on job per run + continue; + } + $jobConfig = isset($jobsRoot[$schedule->getJobCode()]) ? $jobsRoot[$schedule->getJobCode()] : null; + if (!$jobConfig) { + continue; + } + + $scheduledTime = strtotime($schedule->getScheduledAt()); + if ($scheduledTime > $currentTime) { + continue; + } + + try { + if ($schedule->tryLockJob()) { + $this->_runJob($scheduledTime, $currentTime, $jobConfig, $schedule, $groupId); + } + } catch (\Exception $e) { + $schedule->setMessages($e->getMessage()); + if ($schedule->getStatus() === Schedule::STATUS_ERROR) { + $this->logger->critical($e); + } + if ($schedule->getStatus() === Schedule::STATUS_MISSED + && $this->state->getMode() === State::MODE_DEVELOPER + ) { + $this->logger->info( + sprintf( + "%s Schedule Id: %s Job Code: %s", + $schedule->getMessages(), + $schedule->getScheduleId(), + $schedule->getJobCode() + ) + ); + } + } + if ($schedule->getStatus() === Schedule::STATUS_SUCCESS) { + $procesedJobs[$schedule->getJobCode()] = true; + } + $schedule->save(); + } + } }