Shopware 6 Canonical Log for Message Queue Tasks

framework.yaml

fig/packages/framework.yaml
framework:
  messenger:
    buses:
      messenger.bus.shopware:
        middleware:
          - "Shopware\\Production\\Messenger\\TaskLoggingMiddleware"


services:
- "Shopware\\Production\\Messenger\\TaskLoggingMiddleware": ~

TaskLoggingMiddleware.php

<?php

namespace Shopware\Production\Messenger;

use Symfony\Component\Messenger\Envelope;
use Symfony\Component\Messenger\Exception\HandlerFailedException;
use Symfony\Component\Messenger\Middleware\MiddlewareInterface;
use Symfony\Component\Messenger\Middleware\StackInterface;

use Shopware\Core\Framework\MessageQueue\ScheduledTask;
use Shopware\Core\Framework\DataAbstractionLayer\Indexing\EntityIndexingMessage;
use Shopware\Core\Content\ImportExport\Message\DeleteFileMessage as DeleteImportExportFile;
use Shopware\Core\Content\Media\Message\DeleteFileMessage;
use Shopware\Core\Content\Media\Message\GenerateThumbnailsMessage;
use Shopware\Storefront\Framework\Cache\CacheWarmer\WarmUpMessage;

class TaskLoggingMiddleware implements MiddlewareInterface
{
    public function handle(Envelope $envelope, StackInterface $stack): Envelope
    {
        $message = $envelope->getMessage();
        $taskName = $this->getTaskName($message);
        $args = $this->extractArgumentsFromMessage($message);

        $start = microtime(true);
        try {
            return $stack->next()->handle($envelope, $stack);
        } catch (HandlerFailedException $e) {
            $args = $this->addExceptionToArgs($e, $args);

            throw $e;
        } finally {
            $this->logTaskProcessing($taskName, $args, $start);
        }
    }

    private function getTaskName(object $message): string
    {
        if ($message instanceof ScheduledTask) {
            $taskName = $message->getTaskName();
        } else {
            $classParts = explode('\\', get_class($message));
            $taskName = end($classParts);

            if (substr($taskName, -7) === 'Message') {
                $taskName = substr($taskName, 0, -7);
            }
        }

        return $taskName;
    }

    private function extractArgumentsFromMessage($message)
    {
        if ($message instanceof EntityIndexingMessage) {
            $data = $message->getData();

            if (is_array($data)) {
                return ['data' => implode(',', $data)];
            }
        } else if ($message instanceof ScheduledTask) {
            return ['taskId' => $message->getTaskId()];
        } else if ($message instanceof DeleteImportExportFile || $message instanceof DeleteFileMessage) {
           return ['files' => implode(',', array_map(function ($f) { return basename($f); }, $message->getFiles()))];
        } else if ($message instanceof GenerateThumbnailsMessage) {
           return ['mediaIds' => implode(',', $message->getMediaIds())];
        } else if ($message instanceof WarmUpMessage) {
           return ['route' => $message->getRoute(), 'domain' => $message->getDomain(), 'cache_id' => $message->getCacheId()];
        }

        return [];
    }

    private function logTaskProcessing(string $taskName, array $args, $start): void
    {
        $args['duration'] = number_format(microtime(true) - $start, 3);

        $message = date('Y-m-d H:i:s', time()) . ' ' . $taskName . ' ';
        foreach ($args as $arg => $value) {
            $message .= $arg . '=' . $value . ' ';
        }

        file_put_contents("/tmp/worker.log", $message . PHP_EOL, FILE_APPEND);
    }

    private function addExceptionToArgs($e, array $args): array
    {
        $exceptions = $e->getNestedExceptions();
        $args['exception'] = get_class($exceptions[0]);
        $args['exception.msg'] = $exceptions[0]->getMessage();

        return $args;
    }
}