petr-korobeinikov.com

How we built a universal event bus that needs no knowledge of Kafka or other brokers

· go, kafka, events, architecture, cloudevents, asyncapi

Universal event bus for microservices

Hi! I’m Petr Korobeinikov, tech lead for the DBaaS for Redis team at #CloudMTS. Some time ago I set out to build a common toolkit for our development teams. The goal was simple: a developer doesn’t waste time digging into how a specific tool works, takes the ready-made instructions, and simply does the job – writes code. A standard environment helps people move from team to team and adapt quickly, and makes it easier for newcomers to get started.

Today I want to talk about one element of that standard environment, one that lets you quickly get started working with message brokers, even if you’ve never laid eyes on Kafka or any other broker before. I’ll be talking about the event bus (EventBus), and how we set up its code generation.

What is an event bus and why do we need it?

In our work we deal with a microservice architecture. For asynchronous communication between services we use an event bus. Right now, under the hood, it’s Kafka (we wrote about it here), but it could just as easily be NSQ, Pulsar, or even Pub/Sub in Redis. And that’s the whole point: for a developer setting up message exchange, what’s underneath shouldn’t matter.

Overall, the setup works like this:

  • The developer describes the message/event following a strict protobuf specification. They spell out what should be sent, from where, and to whom. If the developer writes something that doesn’t match the required form, the code simply won’t compile.

  • Based on the resulting specification, the developer runs automatic code generation for producers and consumers.

  • The developer doesn’t have to rack their brains over how to format a message, and doesn’t waste time hand-writing producers and consumers. Profit!

Now let’s talk about all of this in more detail.

The specification

We built the strict specification on top of Protocol Buffers (protobuf). Even though we’ve been working with this technology for a long time, mostly in the context of gRPC, we still decided to take a look at other options.

Let me briefly explain why we didn’t go with the tools below:

AsyncAPI. A decent alternative that lets you use any delivery mechanism you like. But the spec would have to be written as a sprawling yaml file, very close to the OpenAPI (Swagger) specification. That’s yet another case of “programming in yaml,” which always breeds a certain dislike among developers. There are tools similar to Swagger that can generate a spec from code that’s already been written. But that’s also a big pain – any spec file generated from, say, Swagger annotations is always imprecise. We instead go by one principle – specification first.

On top of that, subjectively, AsyncAPI’s documentation feels incomplete and lags behind reality, even though the developer community holds public meetups fairly often. You can watch recordings on their official channel.

CloudEvents. It also supports numerous message delivery methods: AMQP (RabbitMQ), Kafka, MQTT, NATS, and others.

The main problem is the lack of a schema for the message itself – a schema for the payload: by default, data can be written in an arbitrary format with no way to specify its structure. Overall, CloudEvents is more about who the producer is, who the message needs to be delivered to, and when it was sent.

oneof data {

    // Binary data
    bytes binary_data = 2;

    // String data
    string text_data = 3;

    // Protobuf Message data
    google.protobuf.Any proto_data = 4;
}

Source

On top of that, it seemed like the barrier to entry for CloudEvents was pretty high. And based on our past experience adapting it, it became clear that, both then and now, it’s overkill.

In the end we wrote a separate Go plugin for protoc (protoc-gen-go-eventbus). The plugin supports an option where we specify the topic for events – the place events get published to, and where readers will read them from.

To form the topic name we use the following format: first the project’s namespace, then the project name, and then the entity or action being performed.

syntax = "proto3";

package eventbus;

option go_package = "./;eventbus";

import "cloud.mts.ru/protobuf/eventbus/annotation.proto";
import "dbaas-redis.redis-hub-gateway.types.proto";

message ApplianceCreateClaimedEvent {
 option (cloud.mts.ru.protobuf.eventbus.topic) = "dbaas-redis.redis-hub-gateway.appliance-create-claimed";

 message Resource {
   int32 total_cpu = 1;
   IntegerUnit total_ram = 2;
   IntegerUnit disk_size = 3;
   string disk_policy = 4;
 }

 message ApplianceVersion {
   string id = 1;
   string name = 2;
 }

 string appliance_id = 1;
 string project_id = 2;
 // …
}

Example specification

Both the resulting protoc-gen-go-eventbus plugin and protoc itself are baked into a separate Docker image. That way we no longer need to install protoc on developers’ machines or keep its version up to date: the current version in use, along with all the necessary plugins and annotations, lives in an image available from our internal registry.

You can read more about this approach to baking tools into Docker images on my blog.

Automatic code generation

Based on the data in the specification, the producer instances themselves get created automatically. Every one of them has a Produce method, which takes ctx as its first argument, and as the second, the event generated from protobuf using the specification above. In this form, the event is sent from the producer into the event bus.

if err := s.applianceCreateEventProducer.Produce(ctx, applianceRequest); err != nil {
  logging.LoggerFromContext(ctx).
     With(zap.String("applianceID", applianceRequest.ApplianceId)).
     Error("create appliance failed", zap.Error(&errCreateAppliance{err: err}))

  return err
}

Example of a producer call

The consumer has an API similar to the producer’s. When the consumer instance is generated, a consumption method gets created too. It takes ctx as the first argument, and as the second – a function that incoming events get fed into. In the example below we handle the cluster creation event.

If processing completes without errors, we return nil.

grp.Go(func() error {
  return applianceCreateClaimedEventConsumer.Consume(ctx, func(ctx context.Context, event *eventbus.ApplianceCreateClaimedEvent) error {
     // do some work
     return err
  })
})

Example of a consumer call

If we return an error, the message will need to be read again.

A bit about monitoring

None of our solutions run without monitoring. And the event bus is no exception. The metrics-collection pattern that suited us best in this situation turned out to be RED (Rate, Errors, Duration). All metrics are collected automatically inside the generated code. The developer doesn’t need to write anything themselves.

From there, the collected metrics go through the well-known mechanism:

  1. They land on the service’s /metrics endpoint.
  2. From there, on the next “scrape,” they end up in VictoriaMetrics.
  3. And they show up on a dashboard in Grafana.

We also have a set of tools for working with Grafana that let you debug dashboards described with grafonnet locally, right on your own laptop. Under the hood are publicly available open-source tools, and we provide a simple, clear command-line interface on top of them. I’ll definitely talk about this in one of my future articles.

What we end up with

Despite being relatively simple (building the tool itself took 4 days, and rolling it out took one sprint), the event bus brings a lot of value both right now and down the line.

  • Low barrier to entry for developers. A developer doesn’t need to know Kafka or any other message broker. They might not even know what’s underneath. We have an event bus that makes it obvious how to send events automatically – no registration, no SMS required.

  • Universality. In this setup, Kafka can be swapped out for anything.

  • Protection from human error. Thanks to the schema’s typing.

  • Simpler unit tests. The typed specification lets us reproduce various scenarios (including error scenarios) using mocks. In doing so we don’t call any real code, and no data gets written to the bus.

  • Simple API. The developer doesn’t need to spend a long time figuring out how to use all this: describe it, generate it, and go.

That’s all for now. Join the constructive discussion in the comments. May the source be with you.