Added cleanup commands

CHANGELOG.md

@@ -1,3 +1,6 @@
+# Version 5.5.0
+* Added cleanup commands
+
 # Version 5.4.1
 * Fix File Exceptions integration
 

README.md

@@ -243,7 +243,6 @@ implementing `DataflowTypeInterface`.
 
 Otherwise, manually add the tag `coderhapsodie.dataflow.type` in your dataflow type service configuration:
 
-```yaml
 ```yaml
 CodeRhapsodie\DataflowExemple\DataflowType\MyFirstDataflowType:
   tags:
@@ -598,6 +597,10 @@ the messenger component instead.
 
 `code-rhapsodie:dataflow:dump-schema` Generates schema create / update SQL queries
 
+`code-rhapsodie:dataflow:set_crashed` Jobs that have been in the "running" status for too long will be set in the "crashed" status.
+
+`code-rhapsodie:dataflow:job_cleanup` Remove old completed or crashed jobs
+
 ### Work with many databases
 
 All commands have a `--connection` option to define what Doctrine DBAL connection to use during execution.

src/Command/JobCleanupCommand.php

@@ -0,0 +1,32 @@
+<?php
+
+declare(strict_types=1);
+
+namespace CodeRhapsodie\DataflowBundle\Command;
+
+use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
+use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+
+#[AsCommand(name: 'code-rhapsodie:job_cleanup', description: 'Cleanup job history.')]
+class JobCleanupCommand extends Command
+{
+    public function __construct(private JobRepository $jobRepository, private int $retention)
+    {
+        parent::__construct();
+    }
+
+    protected function configure()
+    {
+        $this->setHelp('Job retention can be configured with the "job_history.retention" configuration.');
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $this->jobRepository->deleteOld($this->retention);
+
+        return Command::SUCCESS;
+    }
+}

src/Command/JobShowCommand.php

@@ -93,7 +93,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int
         $io->table(['Field', 'Value'], $display);
         if ($input->getOption('details')) {
             $io->section('Exceptions');
-            $exceptions = array_map(fn (string $exception) => substr($exception, 0, 900).'…', $job->getExceptions());
+            $exceptions = array_map(static fn (string $exception) => substr($exception, 0, 900).'…', $job->getExceptions());
 
             $io->write($exceptions);
         }

src/Command/SchemaCommand.php

@@ -38,7 +38,7 @@ protected function execute(InputInterface $input, OutputInterface $output): int
 
         // add -- before each keys
         $options = array_combine(
-            array_map(fn ($key) => '--'.$key, array_keys($options)),
+            array_map(static fn ($key) => '--'.$key, array_keys($options)),
             array_values($options)
         );
 

src/Command/SetCrashedCommand.php

@@ -0,0 +1,32 @@
+<?php
+
+declare(strict_types=1);
+
+namespace CodeRhapsodie\DataflowBundle\Command;
+
+use CodeRhapsodie\DataflowBundle\Repository\JobRepository;
+use Symfony\Component\Console\Attribute\AsCommand;
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+
+#[AsCommand(name: 'code-rhapsodie:set_crashed', description: 'Set long running jobs as crashed.')]
+class SetCrashedCommand extends Command
+{
+    public function __construct(private JobRepository $jobRepository, private int $crashedDelay)
+    {
+        parent::__construct();
+    }
+
+    protected function configure()
+    {
+        $this->setHelp('How long jobs have to run before they are set as crashed can be configured with the "job_history.crashed_delay" configuration.');
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $this->jobRepository->crashLongRunning($this->crashedDelay);
+
+        return Command::SUCCESS;
+    }
+}

src/DataflowType/Dataflow/Dataflow.php

@@ -67,7 +67,7 @@ public function setCustomExceptionIndex(callable $callable): self
      */
     public function setAfterItemProcessors(array $processors): self
     {
-        $this->afterItemProcessors = array_map(fn (callable $callable) => \Closure::fromCallable($callable), $processors);
+        $this->afterItemProcessors = array_map(static fn (callable $callable) => \Closure::fromCallable($callable), $processors);
 
         return $this;
     }

src/DependencyInjection/CodeRhapsodieDataflowExtension.php

@@ -30,6 +30,8 @@ public function load(array $configs, ContainerBuilder $container): void
         $container->setParameter('coderhapsodie.dataflow.dbal_default_connection', $config['dbal_default_connection']);
         $container->setParameter('coderhapsodie.dataflow.default_logger', $config['default_logger']);
         $container->setParameter('coderhapsodie.dataflow.exceptions_mode.type', $config['exceptions_mode']['type']);
+        $container->setParameter('coderhapsodie.dataflow.job_history.retention', $config['job_history']['retention']);
+        $container->setParameter('coderhapsodie.dataflow.job_history.crashed_delay', $config['job_history']['crashed_delay']);
 
         if ($config['exceptions_mode']['type'] === 'file') {
             $container->setParameter('coderhapsodie.dataflow.flysystem_service', $config['exceptions_mode']['flysystem_service']);

src/DependencyInjection/Configuration.php

@@ -52,6 +52,20 @@ public function getConfigTreeBuilder(): TreeBuilder
                         ->thenInvalid('You need "league/flysystem" to use Dataflow file exception mode.')
                     ->end()
                 ->end()
+                ->arrayNode('job_history')
+                    ->addDefaultsIfNotSet()
+                    ->children()
+                        ->integerNode('retention')
+                            ->defaultValue(30)
+                            ->min(0)
+                            ->info('How many days completed and crashed jobs are kept when running the cleanup command.')
+                        ->end()
+                        ->integerNode('crashed_delay')
+                            ->defaultValue(24)
+                            ->min(24)
+                            ->info('Jobs running for more than this many hours will be set as crashed when running the cleanup command.')
+                        ->end()
+                    ->end()
             ->end()
         ;
 

src/Entity/Job.php

@@ -17,6 +17,7 @@ class Job
     public const STATUS_RUNNING = 1;
     public const STATUS_COMPLETED = 2;
     public const STATUS_QUEUED = 3;
+    public const STATUS_CRASHED = 4;
 
     private const KEYS = [
         'id',

src/Repository/JobRepository.php

@@ -151,6 +151,33 @@ public function createQueryBuilder($alias = null): QueryBuilder
         return $qb;
     }
 
+    public function crashLongRunning(int $hours): void
+    {
+        $qb = $this->connection->createQueryBuilder();
+        $qb->update(static::TABLE_NAME, 'j')
+            ->set('j.status', ':new_status')
+            ->set('j.end_time', ':now')
+            ->andWhere('j.status = :status')
+            ->andWhere('j.start_time < :date')
+            ->setParameter('status', Job::STATUS_RUNNING)
+            ->setParameter('date', new \DateTime("- {$hours} hours"), 'datetime')
+            ->setParameter('new_status', Job::STATUS_CRASHED)
+            ->setParameter('now', new \DateTime(), 'datetime')
+            ->executeStatement()
+        ;
+    }
+
+    public function deleteOld(int $days): void
+    {
+        $qb = $this->connection->createQueryBuilder();
+        $qb->delete(static::TABLE_NAME, 'j')
+            ->andWhere($qb->expr()->in('j.status', [Job::STATUS_COMPLETED, Job::STATUS_CRASHED]))
+            ->andWhere('j.end_time < :date')
+            ->setParameter('date', new \DateTime("- {$days} days"), 'datetime')
+            ->executeStatement()
+        ;
+    }
+
     private function returnFirstOrNull(QueryBuilder $qb): ?Job
     {
         $stmt = $qb->executeQuery();

src/Resources/config/services.yaml

@@ -3,7 +3,7 @@ services:
         public: false
 
     CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistryInterface: '@CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistry'
-    CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistry:
+    CodeRhapsodie\DataflowBundle\Registry\DataflowTypeRegistry: ~
 
     CodeRhapsodie\DataflowBundle\Command\AddScheduledDataflowCommand:
         arguments:
@@ -100,3 +100,15 @@ services:
         arguments:
             $repository: '@CodeRhapsodie\DataflowBundle\Repository\JobRepository'
             $exceptionHandler: '@CodeRhapsodie\DataflowBundle\ExceptionsHandler\ExceptionHandlerInterface'
+
+    CodeRhapsodie\DataflowBundle\Command\JobCleanupCommand:
+        arguments:
+            $jobRepository: '@CodeRhapsodie\DataflowBundle\Repository\JobRepository'
+            $retention: '%coderhapsodie.dataflow.job_history.retention%'
+        tags: ['console.command']
+
+    CodeRhapsodie\DataflowBundle\Command\SetCrashedCommand:
+        arguments:
+            $jobRepository: '@CodeRhapsodie\DataflowBundle\Repository\JobRepository'
+            $crashedDelay: '%coderhapsodie.dataflow.job_history.crashed_delay%'
+        tags: ['console.command']