Added logging of output from worker (this is an option you can enable for both successful and failed tasks)

Author: webdevvie

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();

DependencyInjection/Configuration.php

@@ -24,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;
     }

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

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'];
     }
 
     /**
@@ -232,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());
     }
@@ -245,15 +265,39 @@ 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
      *

Tests/Service/TaskQueueServiceTest.php

@@ -138,15 +138,15 @@ public function testMarkDone()
         $workPackage = new WorkPackage($taskEntity, $job, $exampleTask);
 
         $this->entityManager->shouldReceive('persist')->withArgs([$taskEntity]);
-        $this->entityManager->shouldReceive('flush');
+        $this->entityManager->shouldReceive('flush')->twice();
         $this->pheanStalkProxy->shouldReceive('delete');
         $service = new TaskQueueService(
             $this->entityManager,
             $this->pheanStalkProxy,
             $this->serializer,
             $this->params
         );
-        $service->markDone($workPackage);
+        $service->markDone($workPackage, 'log goes here');
     }
 
     /**
@@ -165,15 +165,15 @@ public function testMarkFailed()
         $workPackage = new WorkPackage($taskEntity, $job, $exampleTask);
 
         $this->entityManager->shouldReceive('persist')->withArgs([$taskEntity]);
-        $this->entityManager->shouldReceive('flush');
+        $this->entityManager->shouldReceive('flush')->twice();
         $this->pheanStalkProxy->shouldReceive('delete');
         $service = new TaskQueueService(
             $this->entityManager,
             $this->pheanStalkProxy,
             $this->serializer,
             $this->params
         );
-        $service->markDone($workPackage);
+        $service->markDone($workPackage, 'log goes here');
     }
 
     /**
@@ -348,6 +348,10 @@ public function setupPheanstalkProxyMock()
      */
     public function generateFakeParams()
     {
-        $this->params = ["default_tube" => "gtldtube"];
+        $this->params = array(
+            "default_tube" => "gtldtube",
+            "log_worker_output_on_success" => true,
+            "log_worker_output_on_failure" => true
+        );
     }
 }