Merge branch 'master' of github.com:webdevvie/pheanstalk-task-queue-bundle

webdevvie · webdevvie · commit 62893e361f5c · 2014-04-20T15:37:00.000+02:00
diff --git a/Command/AbstractWorker.php b/Command/AbstractWorker.php
@@ -94,7 +94,7 @@ abstract class AbstractWorker extends ContainerAwareCommand
      */
     protected function addDefaultConfiguration()
     {
-        $this->addOption('use-tube', null, InputOption::VALUE_OPTIONAL, 'What tube to send to');
+        $this->addOption('use-tube', null, InputOption::VALUE_OPTIONAL, 'What tube to work with');
     }
 
     /**
diff --git a/Command/StopWorkerTenderCommand.php b/Command/StopWorkerTenderCommand.php
@@ -0,0 +1,93 @@
+<?php
+
+namespace Webdevvie\PheanstalkTaskQueueBundle\Command;
+
+use Webdevvie\PheanstalkTaskQueueBundle\Command\AbstractWorker;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
+use Symfony\Component\Console\Output\OutputInterface;
+use Webdevvie\PheanstalkTaskQueueBundle\Entity\Task;
+
+/**
+ * Class StopWorkerTenderCommand
+ * Sends a signal to the worker tender to stop
+ *
+ * @package Webdevvie\PheanstalkTaskQueueBundle\Command
+ * @author John Bakker <me@johnbakker.name>
+ */
+class StopWorkerTenderCommand extends AbstractWorker
+{
+    /**
+     * {@inheritDoc}
+     *
+     * @return void
+     */
+    protected function configure()
+    {
+        $this->setName('taskqueue:stop-worker-tender')
+            ->setDescription('Signals the worker tender to stop and waits for it to stop');
+        $this->addDefaultConfiguration();
+    }
+
+    /**
+     * {@inheritDoc}
+     *
+     * @param InputInterface $input
+     * @param OutputInterface $output
+     * @return void
+     * @throws \InvalidArgumentException
+     */
+    protected function execute(InputInterface $input, OutputInterface $output)
+    {
+        $this->initialiseWorker($input, $output);
+        $processes = $this->findWorkerTenderProcessesForTube($this->tube);
+        if (count($processes) == 0) {
+            $output->writeln("<info>Found no processes</info>");
+            return;
+        }
+        $output->writeln('<info>Found ' . count($processes) . ' process(es)');
+        foreach ($processes as $process) {
+            $output->writeln('<info>sending kill to:</info> ' . $process);
+            exec("kill $process");
+        }
+        $output->write('<info>Waiting on processes to stop</info> ');
+        while (count($processes) > 0) {
+            $processes = $this->findWorkerTenderProcessesForTube($this->tube);
+            $output->write(".");
+            sleep(1);
+
+        }
+        $output->writeln("Done!");
+    }
+
+    /**
+     * Returns a list of processes that match the worker-tender
+     *
+     * @param string $tube
+     * @return array
+     */
+    private function findWorkerTenderProcessesForTube($tube)
+    {
+        $processes = array();
+        $command = "ps ax ";
+        $command .= "|grep php ";
+        $command .= "|grep -v stop ";
+        $command .= "|grep worker-tender ";
+        if ($tube !== '') {
+            $command .= "|grep " . escapeshellarg('use-tube=' . $tube);
+        }
+        $data = exec($command);
+        $lines = explode("\n", $data);
+        foreach ($lines as $line) {
+            $line = trim($line);
+            $parts = explode(" ", $line);
+            if ($parts[0] === '') {
+                continue;
+            }
+            $processes[] = intval($parts[0]);
+        }
+
+        return $processes;
+
+    }
+}
diff --git a/Command/Tender/ChildProcessContainer.php b/Command/Tender/ChildProcessContainer.php
@@ -197,7 +197,7 @@ public function isCleanedUp()
      */
     public function getOutputLinesAvailable()
     {
-        $outputLines = [];
+        $outputLines = array();
         return $outputLines;
     }
 
@@ -301,9 +301,9 @@ private function sendTERMSignal()
      */
     public function getReadyForBed()
     {
-        $busyStatusses = [ChildProcessContainer::STATUS_SLEEPY,
+        $busyStatusses = array(ChildProcessContainer::STATUS_SLEEPY,
             ChildProcessContainer::STATUS_BUSY_BUT_SLEEPY
-        ];
+        );
         if (in_array($this->status, $busyStatusses)) {
             //ready for bed
             return;
diff --git a/Command/WorkerCommand.php b/Command/WorkerCommand.php
@@ -87,8 +87,9 @@ private function runTask(WorkPackage $taskObject)
         try {
             $command = $this->taskCommandGenerator->generate($taskObject);
         } catch (TaskCommandGeneratorException $exception) {
-            $this->verboseOutput("Invalid command supplied in " . get_class($taskObject));
-            $this->taskQueueService->markFailed($taskObject);
+            $logMessage = "Invalid command supplied in " . get_class($taskObject);
+            $this->verboseOutput($logMessage);
+            $this->taskQueueService->markFailed($taskObject, $logMessage);
             return;
         }
         $fullCommand = 'exec app/console ' . $command;
@@ -100,6 +101,7 @@ private function runTask(WorkPackage $taskObject)
         $hasReceivedSignal = false;
         $exitCode = 0;
         $exitCodeText = '';
+        $childOutput = '';
         while ($runningCommand) {
             $this->handleRunningProcess(
                 $process,
@@ -111,15 +113,16 @@ private function runTask(WorkPackage $taskObject)
                 $exitCodeText
             );
             $incOut = $process->getIncrementalOutput();
+            $childOutput .= $incOut;
             if (strlen($incOut) > 0) {
                 $this->output->write($incOut);
             }
         }
         if ($failed) {
             $this->verboseOutput("FAILED CHILD: " . $exitCode . "::" . $exitCodeText);
-            $this->taskQueueService->markFailed($taskObject);
+            $this->taskQueueService->markFailed($taskObject, $childOutput);
         } else {
-            $this->taskQueueService->markDone($taskObject);
+            $this->taskQueueService->markDone($taskObject, $childOutput);
         }
         if ($process->isRunning()) {
             $process->stop();
diff --git a/Command/WorkerTenderCommand.php b/Command/WorkerTenderCommand.php
@@ -18,7 +18,7 @@
  * one time per "tube"* You can supply several parameters to this command. Please refer to those parameters
  * for more information about customising your worker-tender experience
  *
- * @author John Bakker <me@johnbakker.name
+ * @author John Bakker <me@johnbakker.name>
  */
 class WorkerTenderCommand extends AbstractWorker
 {
@@ -66,7 +66,7 @@ protected function configure()
                 'Max time to keep a worker in seconds',
                 6000000
             );
-        $this->addDefaultConfiguration();
+        $this->addOption('use-tube', null, InputOption::VALUE_REQUIRED, 'What tube to work with');
     }
 
     /**
@@ -235,8 +235,10 @@ public function moreOrLess($total, $busy, $available)
      */
     private function findDisposableWorkers()
     {
+        $disposableStatusses = array(ChildProcessContainer::STATUS_READY, ChildProcessContainer::STATUS_ALIVE);
         foreach ($this->family as &$child) {
-            if (in_array($child->status, array(ChildProcessContainer::STATUS_READY, ChildProcessContainer::STATUS_ALIVE))) {
+
+            if (in_array($child->status, $disposableStatusses)) {
                 if ($child->getAge() < 10) {
                     //less than ten seconds old keep it alive a bit to let its do its job
                     continue;
diff --git a/DependencyInjection/Configuration.php b/DependencyInjection/Configuration.php
@@ -7,8 +7,6 @@
 
 /**
  * This is the class that validates and merges configuration from your app/config files
- *
- * To learn more see {@link http://symfony.com/doc/current/cookbook/bundles/extension.html#cookbook-bundles-extension-config-class}
  */
 class Configuration implements ConfigurationInterface
 {
@@ -26,6 +24,8 @@ public function getConfigTreeBuilder()
         $rootNode
             ->children()
             ->scalarNode('default_tube')->isRequired()->end()
+            ->booleanNode('log_worker_output_on_failure')->defaultTrue()->end()
+            ->booleanNode('log_worker_output_on_success')->defaultFalse()->end()
             ->end();
         return $treeBuilder;
     }
diff --git a/Entity/Task.php b/Entity/Task.php
@@ -74,6 +74,14 @@ class Task
      */
     private $data;
 
+    /**
+     * The output supplied by the command
+     * @var string
+     *
+     * @ORM\Column(name="log", type="text", nullable=true)
+     */
+    private $log;
+
     /**
      * The tube this task was sent to
      * @var string
@@ -171,6 +179,19 @@ public function setStatus($status)
         return $this;
     }
 
+    /**
+     * Set log
+     *
+     * @param string $status
+     * @return task
+     */
+    public function setLog($log)
+    {
+        $this->log = $log;
+        $this->modified = new \DateTime();
+        return $this;
+    }
+
     /**
      * Get status
      *
@@ -181,7 +202,15 @@ public function getStatus()
         return $this->status;
     }
 
-
+    /**
+     * Get log
+     *
+     * @return string
+     */
+    public function getLog()
+    {
+        return $this->status;
+    }
 
     /**
      * Get created
diff --git a/Service/TaskQueueService.php b/Service/TaskQueueService.php
@@ -37,6 +37,20 @@ class TaskQueueService
      */
     private $defaultTube;
 
+    /**
+     * Log the worker output on success
+     *
+     * @var boolean
+     */
+    private $logWorkerOutputOnSuccess;
+
+    /**
+     * Log the worker output on failure
+     *
+     * @var boolean
+     */
+    private $logWorkerOutputOnFailure;
+
     /**
      * @var \Doctrine\ORM\EntityManager
      */
@@ -65,6 +79,8 @@ public function __construct(
         $this->entityManager = $entityManager;
         $this->taskRepo = $this->entityManager->getRepository('WebdevviePheanstalkTaskQueueBundle:Task');
         $this->defaultTube = $params['default_tube'];
+        $this->logWorkerOutputOnSuccess = $params['log_worker_output_on_success'];
+        $this->logWorkerOutputOnFailure = $params['log_worker_output_on_failure'];
     }
 
     /**
@@ -86,8 +102,7 @@ public function getDefaultTube()
     public function getStatusOfTaskWithId($taskId)
     {
         $task = $this->taskRepo->find($taskId);
-        if(!($task instanceof Task))
-        {
+        if (!($task instanceof Task)) {
             //the task was not found
             return Task::STATUS_GONE;
         }
@@ -169,7 +184,7 @@ public function cleanUpTasks($timePeriod)
             t.created <= :older'
         );
         $query->setParameter('status', array(Task::STATUS_DONE));
-        $query->setParameter('older', date("Y-m-d H:i:s", time()-$timePeriod));
+        $query->setParameter('older', date("Y-m-d H:i:s", time() - $timePeriod));
         $query->execute();
     }
 
@@ -206,7 +221,9 @@ public function reserveTask($tube = null)
             throw new TaskQueueServiceException("Invalid format in TaskQueue {$tube} ");
         } catch (\ReflectionException $exception) {
             $this->beanstalk->delete($inTask);
-            throw new TaskQueueServiceException("Invalid format in TaskQueue {$tube} class ".$parts[0].' is unknown');
+            throw new TaskQueueServiceException(
+                "Invalid format in TaskQueue {$tube} class " . $parts[0] . ' is unknown'
+            );
         }
         if (!($taskObject instanceof TaskDescriptionInterface)) {
             $this->beanstalk->delete($inTask);
@@ -231,11 +248,15 @@ public function reserveTask($tube = null)
      * Deletes a task from the queue
      *
      * @param WorkPackage $task
+     * @param string $log
      * @throws TaskQueueServiceException
      * @return void
      */
-    public function markDone(WorkPackage $task)
+    public function markDone(WorkPackage $task, $log)
     {
+        if ($this->logWorkerOutputOnSuccess) {
+            $this->updateTaskLog($task, $log);
+        }
         $this->updateTaskStatus($task, Task::STATUS_DONE);
         $this->beanstalk->delete($task->getPheanstalkJob());
     }
@@ -244,16 +265,42 @@ public function markDone(WorkPackage $task)
      * Marks a job as failed and deletes it from the beanstalk tube
      *
      * @param WorkPackage $task
+     * @param string $log
      * @throws TaskQueueServiceException
      * @return void
      */
-    public function markFailed(WorkPackage $task)
+    public function markFailed(WorkPackage $task, $log)
     {
+        if ($this->logWorkerOutputOnFailure) {
+            $this->updateTaskLog($task, $log);
+        }
         $this->updateTaskStatus($task, Task::STATUS_FAILED);
         $this->beanstalk->delete($task->getPheanstalkJob());
     }
 
     /**
+     * Writes the log to the Task entity
+     *
+     * @param WorkPackage $task
+     * @param string $log
+     * @return void
+     * @throws TaskQueueServiceException
+     */
+    private function updateTaskLog(WorkPackage $task, $log)
+    {
+        $taskEntity = $task->getTaskEntity();
+        if ($taskEntity instanceof Task) {
+            $taskEntity->setLog($log);
+            //make sure it is stored...
+            $this->entityManager->flush($taskEntity);
+        } else {
+            throw new TaskQueueServiceException("Entity is not of type Task");
+        }
+    }
+
+    /**
+     * Updates the task status
+     *
      * @param WorkPackage $task
      * @param string $status
      * @return void
diff --git a/Tests/Service/TaskQueueServiceTest.php b/Tests/Service/TaskQueueServiceTest.php

Original file line number	Diff line number	Diff line change
`@@ -94,7 +94,7 @@ abstract class AbstractWorker extends ContainerAwareCommand`
`94`	`94`	`*/`
`95`	`95`	`protected function addDefaultConfiguration()`
`96`	`96`	`{`
`97`		`- $this->addOption('use-tube', null, InputOption::VALUE_OPTIONAL, 'What tube to send to');`
	`97`	`+ $this->addOption('use-tube', null, InputOption::VALUE_OPTIONAL, 'What tube to work with');`
`98`	`98`	`}`
`99`	`99`
`100`	`100`	`/**`
Original file line number	Diff line number	Diff line change
`@@ -197,7 +197,7 @@ public function isCleanedUp()`
`197`	`197`	`*/`
`198`	`198`	`public function getOutputLinesAvailable()`
`199`	`199`	`{`
`200`		`- $outputLines = [];`
	`200`	`+ $outputLines = array();`
`201`	`201`	`return $outputLines;`
`202`	`202`	`}`
`203`	`203`
`@@ -301,9 +301,9 @@ private function sendTERMSignal()`
`301`	`301`	`*/`
`302`	`302`	`public function getReadyForBed()`
`303`	`303`	`{`
`304`		`- $busyStatusses = [ChildProcessContainer::STATUS_SLEEPY,`
	`304`	`+ $busyStatusses = array(ChildProcessContainer::STATUS_SLEEPY,`
`305`	`305`	`ChildProcessContainer::STATUS_BUSY_BUT_SLEEPY`
`306`		`- ];`
	`306`	`+ );`
`307`	`307`	`if (in_array($this->status, $busyStatusses)) {`
`308`	`308`	`//ready for bed`
`309`	`309`	`return;`