1 | <?php |
||
2 | namespace App\BxConsole; |
||
3 | |||
4 | use Doctrine\Common\Annotations\AnnotationReader; |
||
5 | use Symfony\Component\Console\Helper\Table; |
||
6 | use Symfony\Component\Console\Helper\TableCell; |
||
7 | use Symfony\Component\Console\Helper\TableSeparator; |
||
8 | use Symfony\Component\Console\Input\ArrayInput; |
||
9 | use Symfony\Component\Console\Input\InputInterface; |
||
10 | use Symfony\Component\Console\Input\InputOption; |
||
11 | use Symfony\Component\Console\Output\ConsoleOutputInterface; |
||
12 | use Symfony\Component\Console\Output\OutputInterface; |
||
13 | use App\BxConsole\Annotations\Agent; |
||
14 | |||
15 | class Cron extends BxCommand { |
||
16 | |||
17 | use LockableTrait; |
||
18 | |||
19 | const EXEC_STATUS_SUCCESS = 'SUCCESS'; |
||
20 | const EXEC_STATUS_ERROR = 'ERROR'; |
||
21 | const EXEC_STATUS_WORK = 'WORK'; |
||
22 | |||
23 | const RESTART_TIME = 3600; |
||
24 | |||
25 | const SORT_NAME = 'name'; |
||
26 | const SORT_TIME = 'time'; |
||
27 | |||
28 | protected function configure() { |
||
29 | |||
30 | $this->setName('system:cron') |
||
31 | ->setDescription('Job scheduler for application commands') |
||
32 | ->addOption('status', 's', InputOption::VALUE_NONE, 'Show BX_CRONTAB status table') |
||
33 | ->addOption('bytime', 't', InputOption::VALUE_NONE, 'Sort status table by exec time desc') |
||
34 | ->addOption('clean', 'c', InputOption::VALUE_REQUIRED, 'Command to be clean crontab data (status, last exec)') |
||
35 | ->addOption('all', 'a', InputOption::VALUE_NONE, 'Command to be clean all crontab data (status, last exec)'); |
||
36 | } |
||
37 | |||
38 | protected function execute(InputInterface $input, OutputInterface $output): int |
||
39 | { |
||
40 | set_time_limit(EnvHelper::getCrontabTimeout()); |
||
41 | |||
42 | $logger = EnvHelper::getLogger('bx_cron'); |
||
43 | if($logger) { |
||
44 | $this->setLogger($logger); |
||
45 | } |
||
46 | |||
47 | $showStatus = $input->getOption('status'); |
||
48 | $byTime = $input->getOption('bytime'); |
||
49 | if($showStatus) { |
||
50 | $sort = ($byTime ? self::SORT_TIME : self::SORT_NAME); |
||
51 | $this->showStatus($output, $sort); |
||
52 | return 0; |
||
53 | } |
||
54 | |||
55 | if(EnvHelper::getSwitch('BX_CRONTAB_RUN', EnvHelper::SWITCH_STATE_OFF)) { |
||
56 | if($this->logger) { |
||
57 | $this->logger->alert('BxCron switch off'); |
||
58 | } |
||
59 | return 0; |
||
60 | } |
||
61 | |||
62 | if(!$this->lock()) { |
||
63 | $msg = 'The command is already running in another process.'; |
||
64 | $output->writeln($msg); |
||
65 | if($this->logger) { |
||
66 | $this->logger->warning($msg); |
||
67 | } |
||
68 | return 0; |
||
69 | } |
||
70 | |||
71 | if($sleepInterval = EnvHelper::checkSleepInterval()) { |
||
72 | $msg = sprintf("Sleep in interval %s", $sleepInterval); |
||
73 | $output->writeln($msg); |
||
74 | if($this->logger) { |
||
75 | $this->logger->warning($msg); |
||
76 | } |
||
77 | return 0; |
||
78 | } |
||
79 | |||
80 | $clean = $input->getOption('clean'); |
||
81 | if($clean) { |
||
82 | $command = $this->getApplication()->find($clean); |
||
83 | $this->cleanJob($command->getName()); |
||
84 | $output->writeln($command->getName() . " will be executed now"); |
||
85 | return 0; |
||
86 | } |
||
87 | |||
88 | $cleanAll = $input->getOption('all'); |
||
89 | if($cleanAll) { |
||
90 | $this->cleanJob(); |
||
91 | $output->writeln("All commands will be executed now"); |
||
92 | return 0; |
||
93 | } |
||
94 | |||
95 | $this->executeJobs($output); |
||
96 | |||
97 | $this->release(); |
||
98 | |||
99 | return 0; |
||
100 | } |
||
101 | |||
102 | protected function showStatus(OutputInterface $output, $sort) { |
||
103 | |||
104 | $table = new Table($output); |
||
105 | $table->setStyle('box-double'); |
||
106 | |||
107 | $isSwitchOff = EnvHelper::getSwitch('BX_CRONTAB_RUN', EnvHelper::SWITCH_STATE_OFF); |
||
108 | |||
109 | $jobs = $this->getCronJobs(); |
||
110 | $this->sortCronTab($jobs, $sort); |
||
111 | $lastExec = 0; |
||
112 | $hasError = false; |
||
113 | |||
114 | foreach($jobs as $cmd => $job) { |
||
115 | $execTime = $job['last_exec']; |
||
116 | if($execTime > $lastExec) $lastExec = $execTime; |
||
117 | if(!empty($job['error'])) { |
||
118 | $hasError = true; |
||
119 | } |
||
120 | } |
||
121 | |||
122 | $headStr = sprintf( |
||
123 | "BX_CRONTAB_RUN: %s; LAST_EXEC: %s; AGENTS_COUNT: %d", |
||
124 | ($isSwitchOff ? 'OFF' : 'ON'), |
||
125 | ($lastExec ? date("d.m.Y H:i:s", $lastExec) : 'NONE'), |
||
126 | count($jobs), |
||
127 | ); |
||
128 | |||
129 | $header = [ |
||
130 | 'Command', |
||
131 | 'Period', |
||
132 | 'Last Exec', |
||
133 | 'Status', |
||
134 | ]; |
||
135 | |||
136 | if($hasError) { |
||
137 | $header[] = 'Error'; |
||
138 | } |
||
139 | |||
140 | $table->setHeaders([ |
||
141 | [new TableCell($headStr, ['colspan' => ($hasError ? 5 : 4)])], |
||
142 | $header, |
||
143 | ]); |
||
144 | |||
145 | $cnt = 1; |
||
146 | foreach($jobs as $cmd => $job) { |
||
147 | if($cnt > 1) $table->addRow(new TableSeparator()); |
||
148 | $row = [ |
||
149 | $cmd, |
||
150 | $job['period'], |
||
151 | ($job['last_exec'] ? date("d.m.Y H:i:s", $job['last_exec']) : 'NONE'), |
||
152 | $job['status'], |
||
153 | ]; |
||
154 | if($hasError) { |
||
155 | $row[] = $job['error']; |
||
156 | } |
||
157 | $table->addRow($row); |
||
158 | $cnt++; |
||
159 | } |
||
160 | |||
161 | $table->render(); |
||
162 | } |
||
163 | |||
164 | protected function cleanJob($command = false) { |
||
165 | |||
166 | $crontab = []; |
||
167 | |||
168 | if($command) { |
||
169 | $crontab = $this->getCronTab(); |
||
170 | if($crontab === false) { |
||
0 ignored issues
–
show
introduced
by
![]() |
|||
171 | return false; |
||
172 | } |
||
173 | unset($crontab[$command]); |
||
174 | } |
||
175 | |||
176 | $this->setCronTab($crontab); |
||
177 | } |
||
178 | |||
179 | protected function executeJobs(OutputInterface $output) { |
||
180 | |||
181 | $jobs = $this->getCronJobs(); |
||
182 | $allTimeout = EnvHelper::getCrontabTimeout(); |
||
183 | $workTime = 0; |
||
184 | |||
185 | if(!empty($jobs)) { |
||
186 | |||
187 | foreach($jobs as $cmd => $job) { |
||
188 | |||
189 | $job['cmd'] = $cmd; |
||
190 | if($this->isActualJob($job)) { |
||
191 | |||
192 | $job['status'] = self::EXEC_STATUS_WORK; |
||
193 | $job['start_time'] = time(); |
||
194 | $this->updateJob($cmd, $job); |
||
195 | |||
196 | $command = $this->getApplication()->find($cmd); |
||
197 | $cmdInput = new ArrayInput(['command' => $cmd]); |
||
198 | $timeStart = microtime(true); |
||
199 | $execTime = 0; |
||
200 | try { |
||
201 | |||
202 | $returnCode = $command->run($cmdInput, $output); |
||
203 | $execTime = microtime(true) - $timeStart; |
||
204 | |||
205 | if(!$returnCode) { |
||
206 | |||
207 | $job['status'] = self::EXEC_STATUS_SUCCESS; |
||
208 | |||
209 | $msg = sprintf("%s: SUCCESS [%.2f s]", $cmd, $execTime); |
||
210 | if($this->logger) { |
||
211 | $this->logger->alert($msg); |
||
212 | } |
||
213 | $output->writeln(PHP_EOL . $msg); |
||
214 | |||
215 | } else { |
||
216 | |||
217 | $job['status'] = self::EXEC_STATUS_ERROR; |
||
218 | $job['error_code'] = $returnCode; |
||
219 | |||
220 | $msg = sprintf("%s: ERROR [%.2f s]", $cmd, $execTime); |
||
221 | if($this->logger) { |
||
222 | $this->logger->alert($msg); |
||
223 | } |
||
224 | $output->writeln(PHP_EOL . $msg); |
||
225 | } |
||
226 | |||
227 | } catch (\Exception $e) { |
||
228 | |||
229 | $job['status'] = self::EXEC_STATUS_ERROR; |
||
230 | $job['error'] = $e->getMessage(); |
||
231 | |||
232 | |||
233 | if($this->logger) { |
||
234 | $this->logger->error($e, ['command' => $cmd]); |
||
235 | } |
||
236 | $output->writeln(PHP_EOL . 'ERROR: ' . $e->getMessage()); |
||
237 | |||
238 | } finally { |
||
239 | |||
240 | if(!$execTime) { |
||
241 | $execTime = microtime(true) - $timeStart; |
||
242 | } |
||
243 | $job['last_exec'] = time(); |
||
244 | $job['exec_time'] = round($execTime, 1); |
||
245 | } |
||
246 | |||
247 | $this->updateJob($cmd, $job); |
||
248 | |||
249 | $workTime += $execTime; |
||
250 | if($workTime * 2 > $allTimeout) { |
||
251 | break; |
||
252 | } |
||
253 | /* |
||
254 | * Let's do just one task |
||
255 | */ |
||
256 | //break; |
||
257 | } |
||
258 | } // foreach($jobs as $cmd => $job) |
||
259 | } // if(!empty($jobs)) |
||
260 | } |
||
261 | |||
262 | protected function isActualJob(&$job): bool |
||
263 | { |
||
264 | $actual = false; |
||
265 | |||
266 | if($job['status'] == self::EXEC_STATUS_WORK) { |
||
267 | if($job['start_time'] && $job['start_time'] < (time() - self::RESTART_TIME)) { |
||
268 | $actual = true; |
||
269 | } |
||
270 | } |
||
271 | |||
272 | $period = intval($job['period']); |
||
273 | |||
274 | if($period > 0) { |
||
275 | if(time() - $job['last_exec'] >= $period) { |
||
276 | $actual = true; |
||
277 | } |
||
278 | } else if(!empty($job['times'])) { |
||
279 | //TODO: |
||
280 | } |
||
281 | |||
282 | if($actual && !empty($job['interval'])) { |
||
283 | $times = explode('-', $job['interval']); |
||
284 | if(count($times) == 2) { |
||
285 | $minTime = Time24::validateTimeString($times[0]); |
||
286 | $maxTime = Time24::validateTimeString($times[1]); |
||
287 | if($minTime && $maxTime) { |
||
288 | if(!Time24::inInterval($minTime, $maxTime)) { |
||
289 | $msg = sprintf("%s not in interval %s", $job['cmd'], $job['interval']); |
||
290 | if($this->logger) { |
||
291 | $this->logger->alert($msg); |
||
292 | } |
||
293 | return false; |
||
294 | } |
||
295 | } |
||
296 | } |
||
297 | } |
||
298 | return $actual; |
||
299 | } |
||
300 | |||
301 | public function getCronJobs(): array |
||
302 | { |
||
303 | $crontab = $this->getCronTab(); |
||
304 | if($crontab === false) { |
||
0 ignored issues
–
show
|
|||
305 | return []; |
||
306 | } |
||
307 | |||
308 | /** @var Application $app */ |
||
309 | $app = $this->getApplication(); |
||
310 | |||
311 | $commands = $app->all(); |
||
312 | |||
313 | $selfCommands = []; |
||
314 | foreach($commands as $command) { |
||
315 | /** @var BxCommand $command */ |
||
316 | if($command instanceof BxCommand) { |
||
317 | $name = $command->getName(); |
||
318 | $selfCommands[$name] = [ |
||
319 | 'object' => $command, |
||
320 | ]; |
||
321 | } |
||
322 | } |
||
323 | |||
324 | $agents = []; |
||
325 | $reader = new AnnotationReader(); |
||
326 | foreach($selfCommands as $cmd => $selfCommand) { |
||
327 | $reflectionClass = new \ReflectionClass($selfCommand['object']); |
||
328 | $annotations = $reader->getClassAnnotations($reflectionClass); |
||
329 | |||
330 | foreach($annotations as $annotation) { |
||
331 | if($annotation instanceof Agent) { |
||
332 | $agents[$cmd] = $annotation->toArray(); |
||
333 | } |
||
334 | } |
||
335 | } |
||
336 | |||
337 | foreach($crontab as $cmd => $job) { |
||
338 | if(is_array($job) && isset($agents[$cmd])) { |
||
339 | $agents[$cmd] = array_merge($job, $agents[$cmd]); |
||
340 | } |
||
341 | } |
||
342 | |||
343 | $this->setCronTab($agents); |
||
344 | |||
345 | return $agents; |
||
346 | } |
||
347 | |||
348 | protected function updateJob($cmd, $job) { |
||
349 | |||
350 | return $this->updateCronTab([$cmd => $job]); |
||
351 | } |
||
352 | |||
353 | protected function updateCronTab(array $changedAgents) { |
||
354 | |||
355 | $crontab = $this->getCronTab(); |
||
356 | |||
357 | if($crontab === false) { |
||
0 ignored issues
–
show
|
|||
358 | return false; |
||
359 | } else { |
||
360 | $crontab = array_merge($crontab, $changedAgents); |
||
361 | return $this->setCronTab($crontab); |
||
362 | } |
||
363 | } |
||
364 | |||
365 | protected function setCronTab(array $agents) { |
||
366 | |||
367 | $isSuccess = true; |
||
368 | $this->sortCronTab($agents); |
||
369 | |||
370 | $filename = EnvHelper::getCrontabFile(); |
||
371 | |||
372 | $fh = fopen($filename, 'c'); |
||
373 | if (flock($fh, LOCK_EX)) { |
||
374 | ftruncate($fh, 0); |
||
375 | if(!fwrite($fh, json_encode($agents, JSON_PRETTY_PRINT))) { |
||
376 | throw new \Exception('Unable to write BX_CRONTAB : ' . $filename); |
||
377 | } |
||
378 | } else { |
||
379 | $isSuccess = false; |
||
380 | } |
||
381 | flock($fh, LOCK_UN); |
||
382 | fclose($fh); |
||
383 | |||
384 | return $isSuccess; |
||
385 | } |
||
386 | |||
387 | /** |
||
388 | * @return array|false|mixed |
||
389 | */ |
||
390 | protected function getCronTab() { |
||
391 | |||
392 | $filename = EnvHelper::getCrontabFile(); |
||
393 | |||
394 | $fh = fopen($filename, 'r'); |
||
395 | if(!$fh) { |
||
0 ignored issues
–
show
|
|||
396 | return false; |
||
397 | } |
||
398 | if(flock($fh, LOCK_SH)) { |
||
399 | $cronTab = []; |
||
400 | clearstatcache(); |
||
401 | $filesize = (int) filesize($filename); |
||
402 | if($filesize > 0 && $data = fread($fh, $filesize)) { |
||
403 | $decoded = json_decode($data, true); |
||
404 | if(is_array($decoded)) { |
||
405 | $cronTab = $decoded; |
||
406 | } else { |
||
407 | throw new \Exception("Unable to parse cronTab"); |
||
408 | } |
||
409 | } |
||
410 | } else { |
||
411 | $cronTab = false; |
||
412 | } |
||
413 | flock($fh, LOCK_UN); |
||
414 | fclose($fh); |
||
415 | |||
416 | return $cronTab; |
||
417 | } |
||
418 | |||
419 | protected function sortCronTab(array &$crontab, $sort = self::SORT_NAME) { |
||
420 | |||
421 | if($sort == self::SORT_TIME) { |
||
422 | $sorting = []; |
||
423 | foreach($crontab as $cmd => $data) { |
||
424 | $sorting[$cmd] = $data['last_exec']; |
||
425 | } |
||
426 | arsort($sorting, SORT_NUMERIC); |
||
427 | $sorted = []; |
||
428 | foreach($sorting as $cmd => $time) { |
||
429 | $sorted[$cmd] = $crontab[$cmd]; |
||
430 | } |
||
431 | $crontab = $sorted; |
||
432 | } else { |
||
433 | ksort($crontab, SORT_STRING); |
||
434 | } |
||
435 | } |
||
436 | } |