CloudWatch Insights - monitoring async workers traffic distribution

Symfony Framework Messenger allows to run parts of your system's processes asynchronously. To gain a better understanding of how your application really works, you can use the following queries. 

These queries filter all logs that comes from Symfony Messenger Workers and aggregate them to provide broader perspective. These also prepare a summary of all asynchronous execution units in your system, giving you greater awareness of the overall complexity and the building blocks that make up the whole.

So, at the very beginning let me explain the data from the logs in my application. This way you can apply these queries to your system, keeping in mind that my log structures might differ partially or entirely.

• message - with value MESSENGER_WORKER_MESSAGE_FINISH indicates Symfony Messenger Message that was successfully handled, 
• extra.messengerWorker.class - event/command class that was handled by Worker
• extra.messengerWorker.executionId - represents each message that was handled by Messenger Worker 
• extra.requestDuration - whole Worker duration, 
• context.messengerWorker.processingDuration - execution of a single Message duration
• context.messengerWorker.memoryUsageMb - whole Worker memory usage

📊 List all unique Workers 

Thanks to that query you can easily check how many different async-workers do you have in your system. dedup function helps a lot with removing duplicated message class names.    

fields @timestamp, @message, @logStream, @log
| filter message = "MESSENGER_WORKER_MESSAGE_FINISH"
| display extra.messengerWorker.class 
| dedup extra.messengerWorker.class 

All unique Messenger events/commands that are handel asynchronously.
This summary gathers all async execution units (that have worked in a given period of time)
of you application in one place. 

📊 Reveal traffic distribution per Worker 

That query shows the busiest workers in your system.     

fields @timestamp, @message, @logStream, @log
| filter message = "MESSENGER_WORKER_MESSAGE_FINISH"
| stats count(*) as executionCount by extra.messengerWorker.class  
| sort by executionCount desc

It's a simple list which execution count for each individual worker that was handled in a given period of time.

The wider time period you'll set for the query the more (probably) unique worker type you might get as a results. 

Changing the time period of the query to only one day would reveal less worker types
but returned metrics might be more useful. 

📊 More detailed Workers data 

Each row symbolizes a single worker per instances, how many messages they handled, when the first and last messages were handled, and, of course, the message type they handled. Take look at interesting usage of  latest(fieldName) function which allows to present not aggregated data in this summary.      

fields @timestamp, @message, @logStream, @log
| parse @logStream "/php/*" as log_stream_id
| filter message = "MESSENGER_WORKER_MESSAGE_FINISH"
| stats latest(substr(log_stream_id, 0, 6)) as instance,
        count(extra.messengerWorker.executionId) as executions, 
        min(@timestamp) as start,
        max(@timestamp) as end,
        max(extra.requestDuration / 60) as duration_in_minutes,
        latest(extra.messengerWorker.class) as message_class
  by extra.executionId as worker_exectuion_id

You can use that query when you need to check a specific worker performance.

📊 The longest Workers executions

To identify the longest workers execution in you application just use the query bellow. These spots could be the first candidates for optimization or potential bottlenecks that slow down your application.       

fields @timestamp, @message, @logStream, @log
| filter message like /(?i)(MESSENGER_WORKER_MESSAGE_FINISH)/
| display datetime, 
          context.messengerWorker.processingDuration / 60,
          context.messengerWorker.memoryUsageMb,
          extra.messengerWorker.class
| sort context.messengerWorker.processingDuration desc