Redirect stdout loggers to stderr preventing json response polutions

Author: josbeir

src/Command/SubprocessJobRunnerCommand.php

@@ -20,12 +20,15 @@
 use Cake\Console\Arguments;
 use Cake\Console\ConsoleIo;
 use Cake\Core\ContainerInterface;
+use Cake\Log\Engine\ConsoleLog;
+use Cake\Log\Log;
 use Cake\Queue\Job\Message;
 use Cake\Queue\Queue\Processor;
 use Enqueue\Null\NullConnectionFactory;
 use Enqueue\Null\NullMessage;
 use Interop\Queue\Message as QueueMessage;
 use Interop\Queue\Processor as InteropProcessor;
+use Psr\Log\LoggerInterface;
 use Psr\Log\NullLogger;
 use RuntimeException;
 use Throwable;
@@ -160,8 +163,10 @@ protected function executeJob(array $data): string
             }
         }
 
+        $logger = $this->configureLogging($data);
+
         $message = new Message($queueMessage, $context, $this->container);
-        $processor = new Processor(new NullLogger(), $this->container);
+        $processor = new Processor($logger, $this->container);
 
         $result = $processor->processMessage($message);
 
@@ -170,6 +175,34 @@ protected function executeJob(array $data): string
         return is_string($result) ? $result : (string)$result;
     }
 
+    /**
+     * Configure logging to use STDERR to prevent job logs from contaminating STDOUT.
+     * Reconfigures all CakePHP loggers to write to STDERR with no additional formatting.
+     *
+     * @param array<string, mixed> $data Job data
+     * @return \Psr\Log\LoggerInterface
+     */
+    protected function configureLogging(array $data): LoggerInterface
+    {
+        // Drop all existing loggers to prevent duplicate logging
+        foreach (Log::configured() as $loggerName) {
+            Log::drop($loggerName);
+        }
+
+        // Configure a single stderr logger
+        Log::setConfig('default', [
+            'className' => ConsoleLog::class,
+            'stream' => 'php://stderr',
+        ]);
+
+        $logger = Log::engine('default');
+        if (!$logger instanceof LoggerInterface) {
+            $logger = new NullLogger();
+        }
+
+        return $logger;
+    }
+
     /**
      * Output result as JSON to STDOUT.
      *

src/Command/WorkerCommand.php

@@ -172,6 +172,21 @@ protected function getLogger(Arguments $args): LoggerInterface
         return $logger ?? new NullLogger();
     }
 
+    /**
+     * Get logger for subprocess output.
+     * Always returns a real logger for subprocess mode to show job logs.
+     *
+     * @param \Cake\Console\Arguments $args Arguments
+     * @return \Psr\Log\LoggerInterface
+     */
+    protected function getSubprocessLogger(Arguments $args): LoggerInterface
+    {
+        $loggerName = (string)$args->getOption('logger');
+        $logger = Log::engine($loggerName);
+
+        return $logger ?? new NullLogger();
+    }
+
     /**
      * Creates and returns a Processor object
      *
@@ -201,15 +216,20 @@ protected function getProcessor(Arguments $args, ConsoleIo $io, LoggerInterface
         if ($args->getOption('subprocess') || ($config['subprocess']['enabled'] ?? false)) {
             $subprocessConfig = array_merge(
                 $config['subprocess'] ?? [],
-                ['enabled' => true],
+                [
+                    'enabled' => true,
+                    'logger' => $config['logger'] ?? (string)$args->getOption('logger'),
+                ],
             );
 
             if ($processorClass !== Processor::class && !is_subclass_of($processorClass, Processor::class)) {
                 $io->error('Subprocess mode is only supported with the default Processor class');
                 $this->abort();
             }
 
-            $processor = new SubprocessProcessor($logger, $subprocessConfig, $this->container);
+            // Use a real logger for subprocess output so logs are visible
+            $subprocessLogger = $this->getSubprocessLogger($args);
+            $processor = new SubprocessProcessor($subprocessLogger, $subprocessConfig, $this->container);
         } else {
             $processor = new $processorClass($logger, $this->container);
         }

src/Queue/SubprocessProcessor.php

@@ -185,6 +185,7 @@ protected function prepareJobData(QueueMessage $message): array
             'messageClass' => get_class($message),
             'body' => $body,
             'properties' => $properties,
+            'logger' => $this->config['logger'] ?? 'stderr',
         ];
     }
 
@@ -281,6 +282,11 @@ protected function executeInSubprocess(array $jobData): array
                         }
 
                         $errorOutput .= $chunk;
+                        // Stream subprocess logs to parent's stderr in real-time
+                        // Skip in PHPUnit test context to avoid test framework issues
+                        if (!defined('PHPUNIT_COMPOSER_INSTALL') && !defined('__PHPUNIT_PHAR__')) {
+                            fwrite(STDERR, $chunk);
+                        }
                     }
                 }
 

tests/TestCase/Command/SubprocessJobRunnerCommandTest.php

@@ -24,6 +24,7 @@
 use Enqueue\Null\NullMessage;
 use Interop\Queue\Processor;
 use ReflectionClass;
+use TestApp\Job\MultilineLogJob;
 use TestApp\TestProcessor;
 
 class SubprocessJobRunnerCommandTest extends TestCase
@@ -252,4 +253,67 @@ public function testOutputResult(): void
 
         $method->invoke($command, $io, ['success' => true, 'result' => 'ack']);
     }
+
+    /**
+     * Test that subprocess jobs with multiple log lines properly separate logs from JSON output
+     */
+    public function testLogsRedirectedToStderr(): void
+    {
+        $jobData = [
+            'messageClass' => NullMessage::class,
+            'body' => [
+                'class' => [MultilineLogJob::class, 'execute'],
+                'args' => [],
+            ],
+            'properties' => [],
+            'logger' => 'debug',
+        ];
+
+        $command = 'php ' . ROOT . 'bin/cake.php queue subprocess-runner';
+
+        $descriptors = [
+            0 => ['pipe', 'r'],
+            1 => ['pipe', 'w'],
+            2 => ['pipe', 'w'],
+        ];
+
+        $process = proc_open($command, $descriptors, $pipes);
+        $this->assertIsResource($process);
+
+        // Write job data to STDIN
+        $jobDataJson = json_encode($jobData);
+        if ($jobDataJson !== false) {
+            fwrite($pipes[0], $jobDataJson);
+        }
+
+        fclose($pipes[0]);
+
+        // Read STDOUT and STDERR
+        $stdout = stream_get_contents($pipes[1]);
+        fclose($pipes[1]);
+
+        $stderr = stream_get_contents($pipes[2]);
+        fclose($pipes[2]);
+
+        proc_close($process);
+
+        // STDOUT should be valid JSON without any log messages
+        $result = json_decode($stdout, true);
+        $this->assertIsArray($result, 'STDOUT should contain valid JSON: ' . $stdout);
+        $this->assertArrayHasKey('success', $result);
+        $this->assertTrue($result['success']);
+        $this->assertSame(Processor::ACK, $result['result']);
+
+        // STDOUT should not contain any job log messages
+        $this->assertStringNotContainsString('Job execution started', $stdout);
+        $this->assertStringNotContainsString('Processing step', $stdout);
+        $this->assertStringNotContainsString('Job execution finished', $stdout);
+
+        // All log messages should be in STDERR
+        $this->assertStringContainsString('Job execution started', $stderr);
+        $this->assertStringContainsString('Processing step 1 completed', $stderr);
+        $this->assertStringContainsString('Processing step 2 completed', $stderr);
+        $this->assertStringContainsString('Processing step 3 completed', $stderr);
+        $this->assertStringContainsString('Job execution finished', $stderr);
+    }
 }

tests/test_app/src/Job/MultilineLogJob.php

@@ -0,0 +1,26 @@
+<?php
+declare(strict_types=1);
+
+namespace TestApp\Job;
+
+use Cake\Log\LogTrait;
+use Cake\Queue\Job\JobInterface;
+use Cake\Queue\Job\Message;
+use Interop\Queue\Processor;
+use Psr\Log\LogLevel;
+
+class MultilineLogJob implements JobInterface
+{
+    use LogTrait;
+
+    public function execute(Message $message): ?string
+    {
+        $this->log('Job execution started', LogLevel::INFO);
+        $this->log('Processing step 1 completed', LogLevel::INFO);
+        $this->log('Processing step 2 completed', LogLevel::INFO);
+        $this->log('Processing step 3 completed', LogLevel::INFO);
+        $this->log('Job execution finished', LogLevel::INFO);
+
+        return Processor::ACK;
+    }
+}