• Permanent events log and replay functionality;
• Publishers/consumers topology visibility;
• Distributed tracing of events.

In the previous article I described how we organize messages contracts, communication-layer, and resilience in PHP microservices. The current article is more language-agnostic, as well as it is just a collection of ideas/approaches rather than call for a specific implementation.

Events history

In AMQP once a message is delivered (and optionally acknowledged), it is removed from a queue, and you will never see it again. What if you need to investigate what happened at that stage?

I can think of several options, for example:

• Enabling debug level of logging in message bus.
• RabbitMQ Firehose.
• Subscribing to all messages and storing/redirecting them to separate storage or message bus (might be done via custom subscriber or via tools like Apache Flink, Kafka RabbitMQ connector, etc.).

We have chosen the last option, which works the following way:

• There is a queue, which is subscribed to all messages in topic exchange via “#” binding key (which means matching any routing key);
• There are consumers, which constantly read messages from that queue and store it in the appropriate format to persistent storage.

I should also mention that the main disadvantage of such approach is that the replication/logging queue will receive duplicates of every message in the cluster, which might affect performance of this cluster. This might be not a big deal, but at least proper monitoring is required. You should be alerted immediately if some of the publishers go crazy and starts generating an enormous amount of messages (which in my opinion is a must-have anyway).

Having storage of all events, we got:

• Events history log, which can be used for errors investigation (what was in the message which triggered that error?);
• Events, which might be a source of data for your data warehouse and reporting;
• Ability to replay messages.

Events replay

The latter (replay functionality) is especially useful in situations when something breaks, and data for previous X minutes/days/etc. must be reprocessed.

Another example is: we start building a pricing application, which needs to have information about prices for all the time (what was offered?) and orders (what was actually sold?) so that it can learn, predict and advise on prices. No problem, we can replay all offers/orders messages for it!

Replay required custom implementation, though it is low-hanging fruit. Though you must be careful implementing such functionality because republishing messages towards all consumers of this message might be harmful, normally it should be done into a specific queue only. This is achievable by using direct exchange and publishing to a routing key that matches desired queue name.

Topology visibility

Normally configuration of publishers and consumers is standardized and enforced by a library that is used for RabbitMQ communication. In our case we had a slightly adjusted configuration of standard Symfony Messenger, which looks like this:

transports:
    amqp_product_category_v1:
        dsn: '%env(MESSENGER_TRANSPORT_AMQP_DSN)%'
        options:
            queue_name: '%queue_name.category.v1%'
            binding_keys: [ '%product.category.v1.routing_key%' ]
    amqp_product_v1:
        dsn: '%env(MESSENGER_TRANSPORT_AMQP_DSN)%'
        options:
            publish_routing_key: '%product.v1.routing_key%'
routing:
    # routing: Message -> Transports (multiple allowed)
    'App\Message\ProductMessage': [ 'amqp_product_v1' ]

One of our engineers implemented a nice tool that analyzes such configuration across multiple applications and aggregates this data into the possible topology of RabbitMQ communication, which might look so for example:

Tracing

Having visualized configuration is great, but this is “possible theoretical” topology, which does not reflect actual data flow (which might be different in different environments because of versions of software, or which might not reflect the fact that some of the publishers are not generating messages for whatever reason for example).

Additionally, data flow creates a dependent chain of events, for example, 2 products were damaged in a warehouse (published by warehouse application), which triggered stock level decrease by 2 (published by inventory application), which triggered a decrease in the quantity offered to customer (published by offer application), which … (and so on).

And now when you want to investigate why customer sees less available quantity, you should be able to trace this chain back to the event of products being damaged.

Both problems (real data flow and tracing back chain of events) might be solved by implementing distributed tracing, which however requires an ability to link messages to each other and identify producers and consumers.

This is where messages metadata gets very helpful:

When pushing this data to a tracing system, you would see the visualized chain of events that happened in your system. Google Trace is shown as an example in the screenshot below, but any other solution would work as well.

Which libraries do you use when building communication between PHP microservices via message bus? Though this task does not sound uncommon, we ended up with custom implementation on top of Symfony Messenger component. In this article I will try to share the most important concepts of our in-house solution.

What did we try to solve?

While building our ERP system (a kind of mix of supply chain management, order processing, and accounting) we followed Spotify team model, where a team is split into smaller squads and tribes, which aim to be as autonomous as possible. This (according to Conway’s law) led to implementation of technical solution trying to reflect independence of teams, such as:

• Microservices architecture, where each service has a single owner-team.
• Continuous delivery avoiding centralized deployment trains.
• Contract-driven communication between services (and teams in turn as well).
• Prevailing of asynchronous (message bus) over synchronous (RPC/REST/etc.) communication.

In addition to the directions given above, there were few more tasks to solve regarding asynchronous communication:

• RabbitMQ was already used, so it needed to be adapted first. But in general, we wanted to be as agnostic as possible in terms of using message buses. We ended up using our solution with at least AMQP, SQS and DB transports, so the goal was achieved.
• There must be a way to enforce contracts of messages with a framework for cross-team reviews of these contracts.
• Software must be resilient.

After analysis, we came up to the conclusion that Symfony Messenger gives the most flexibility in terms of being message bus agnostic. It provides a unified wrapper, where transports (Redis, AMQP, DB, AWS SQS, etc.) are pluggable as separate dependencies.

Though there are also significant limitations of Symfony Messenger (when I write this article, Symfony 5.3 is the latest stable version):

• The component is not really adopted to work with messages published and consumed by different applications, see #33912 for example.
• There are gaps in resilience behavior, e.g. see open #36870, already resolved #36864, and a few more.

Therefore, we decided to build our custom transport (as a private composer package), which I am going to describe below.

Implementation

This transport is easy to use, one needs only to install the private package and specify a custom prefix in transport DSN instead of standard “amqp://”.

The transport was built on top of symfony/amqp-messenger, so it reuses connection classes, while sender and receiver are extended.

#1 Deserialization and contracts

Having custom transport allows introducing additional configuration options without breaking already existing options for standard “amqp-messenger”.

dsn: 'custom-amqp://...'
options:
    serialization:
        deserialize_to: 'App\Message\ProductMessage'
    validation:
        schema_path: 'product_message_schema.json'
        validate_on_consume: true

First, we needed to create a way to support multiple applications’ communication, therefore we introduced “deserialize_to” option, which tells the library which class must be used as a message object. In native Symfony Messenger deserialization happens based on message header, which contains full class name (including namespace) of initial message — which is of course hardly usable when publisher and subscriber are different applications.

Second, we wanted to enforce contract validation (we use JSON messages, therefore JSON-schema is a contract). Just having a contract is not enough, messages should be actually validated against this contract. But when should you apply it: on publish or on consume? Our reasoning was:

• Validation is obligatory on publishing, because publisher is the owner of a contract — they should avoid publishing invalid messages at any cost.
• However, validation is optional on consumption and is disabled by default.

#2 Contracts organization

All right, we have enforced teams to use contracts in messages. Now no message can be published without a contract. We all also agreed that contracts should be maintained in a backward-compatible manner: if a change is incompatible, new major version or just another contract should be introduced.

But now we have multiple messages contracts and JSON files circulating through multiple applications with unclear versioning and vague ownership and maintenance process.

In addition to that many of our JSON-schema contracts contained sub-schemas like this:

"price": {
    "description": "Promotion price",
    "$ref": "../common/money.json"
},

Where “money.json” contains an object consisting of currency and amount, which is reused and copied multiple times.

We came up with the following solution:

1. All contracts are stored in a single repository:

• This allows introducing a process of cross-team review for any contract changes, e.g. just an agreement or enforcement of merge-request reviews.
• This allows setting up notifications for any contract changes(e.g. if notifications are sent to a dedicated Slack channel).
• This solves a problem of sub-schemas, which now are described once and referenced multiple times. No copy-pasting is needed anymore.

2. Every time a change happens, contracts are automatically packed into a new version of PHP package (Symfony bundle in this case), which then can be installed/updated in publishers/consumers applications.

• This solves versioning problem: applications rely on specific contracts, which might be updated only explicitly.
• There is no limitation towards a specific technology, we might build similar package for Java/JavaScript/Python, etc.

3. Contracts also contain some metadata, e.g. description of what this message is about and destination (in terminology of RabbitMQ this is equal to “routing key”).

Additionally, such approach allowed auto-generation of documentation pages with a human-readable catalog of all data flowing through message bus in the system.

Symfony bundle containing contracts exposes parameters, which can be referenced as schema path or routing key, so no path or routing keys hardcoding is needed anymore, no manual contracts copying is needed as well.

options:
    serialization:
        deserialize_to: 'App\Message\ProductMessage'
    validation:
        schema_path: '%contracts.schema.product.path%'
    publish_routing_key: '%contracts.product.v1.routing_key%'

#3 Messages metadata

We found it very useful to inject metadata into each message, because it greatly helps with logging, debugging, and tracing.

This might be a subject for a separate article since the topic is not too small.

#4 Expected unexpected

Okay, what do we do when failures happen? Symfony Messenger already has a built-in retries mechanism and failure transports concept, which can be configured for consumers.

What about producers? What happens if RabbitMQ is temporary not available, or message schema validation fails? Our goal is to avoid losing data at any cost. Unfortunately, there is nothing available out of the box, so we had to implement adding to failed transport on publishing as well. Symfony Messenger has a great concept of message stamps, so we distinguish failed on publishing and failed on consuming using custom stamps added to envelopes.

It was important for us to continuously monitor failed messages (the aim is to have zero of them), because some of them might be republished/reconsumed automatically, while others need manual resolution. We monitor number of failed messages via Prometheus and Grafana, where applications expose monitoring endpoint, which contains multiple metrics including failed messages.

Finally, what should engineers do with failed messages? Symfony Messenger has CLI commands to manage them (list/view/reconsume/delete), which we extended with “republish” action. But CLI has a disadvantage as well: it means engineers should get SSH/kubectl exec/whatever else permission in production, which is kind of complicated when you work with sensitive information (e.g., customer orders info). So, we additionally implemented HTTP endpoints and UI with proper permissions restrictions for failed messages management.

See also second part: Observability with RabbitMQ and microservices