Summary
This is a proposal of a new implementation of the ROR message bus. Today's solution is directly tied to RabbitMQ and exposes the users (both the developers of ROR, and third party developers of Microservices) for a lot of complexity. This proposal describes a new message bus package that abstracts away the complexity of the message bus infrastructure implementation. It also proposes a helper package, resourceupdate, that provides a set of helper functions to quickly get resource updates. This will decouple the code in microservices from the underlying implementation of the messagebus, today, RabbitMQ.
Motivation
The motivation behind this proposal is that we as Developers of Microservices want a simpler way to get messages about resource updates from the ROR-api.
Background
Todays solution utilizes RabbitMq directly and the user must be familiar with both RabbitMq and message bus architecture. We want to achieve two things with this design, decouple the message bus implementation from the ror code, and make it easier for the user to interact with the message bus. Although the design is simpler than working with a message bus library directly, it is still a generic library. Writing libraries on top of this framework, to handle specific cases e.g writing a library that provides types for specific messages related to resourceUpdates, Is encouraged.
Proposed Solution
The design has gone through a few iterations already, but it is not fully complete. The reason for this is that we want to involve the core ROR team in the discussion early on. No specific implementation is proposed. We want to test out a few message bus implementations, at least RabbitMq and NATS, and that will take time. Thanks to Kjetil Sigvartsen, Kevin Vatn, Sigurd Skogmo, Morten Ottestad and Tor Halvor Frivold, for providing feedback on different iterations of the design.
This proposal only describe a receive once system, where each message is consumed and not stored on the queue. If we want to support this we should create a separate Stream proposal to keep this proposal focused and small.
The solution as a whole is a set of interfaces that define a publisher, subscriber model, with a topic exchange (rabbitMq term used to clarify a bit what we mean) behind the scenes. The reason we decided on a topic exchange design, is that we use both agents and microservices to manage our resources and they each have different granularity of concern when it comes to resources. Agents care about single instances of resources, while microservices, in most cases, care about every instance within a kind. This difference in granularity makes it difficult to use single queues per resource kind, as it would make for complex handling logic for agents.
Publishing messages
The philosophy of this system is that producers don't ask for things to be handled. They instead announce that something has happened, the api does not ask the microservice to change the resource, it publishes a message telling the bus that the resource status has changed, then it's up to the microservice to act upon that message. The sender does not expect an answer back from the microservice that it has started working. When the microservice has changed the resource to allign with the status the api will publish a message saying the resource spec has changed, and then other consumers may act upon that message.
Why not simple queues
When making the design we came up with three scenarios when using simple queues.
If we were to use simple queues we would quickly need to have a queue for each instance of a resource which would not scale well.
This is where topic exchanges comes in. They allow us to dynamically create queues of the granularity we need, initiated from the clients side. Using a simple routing scheme, we can use the Type, kind and instance id of a resource (or other type of event) to create queues. This means that agents will get queues that only contain messages for a specific resource instance, and microservices will get queues that contain all messages for a resource kind. All of this logic needs to be supported in the provider, so it limits us to providers that support topic exchange like functionality. This will result in less queues than having a queue per instance of a resource, but significantly less because where we have a lot of instances of resources we use microservices to handle those anyways. e.g we have a thousands of backup runs/jobs, while we have a few hundred clusters.
Another problem is that we don't want to expose our message bus outside of the cluster directly. Instead we want to forward messages via SSE, this means that our SSE server needs to satisfy the interfaces so it can act as a propper message bus client.
The full flow is described in the figure below.
Simple datatypes
The design builds on a few simple datatypes and three main datatypes, the publish interface, the subscribe interface, and the message struct.
type messageId string
type messageRunId string //usefull for connecting events together
type MessageType string
type MessageKind string
type MessageInstance string
type Route string
type Handler func Handle(ctx context.Context, message messageBus.Message) error
Message type, kind, Instance and route
The routing of messages rely on the Type, Kind and Instance hierarchy. The route looks like this "Type:Kind:Instance". Instance and kind can be omitted to act as a wildcard, essentially meaning match all of that type.
e.g "ResourceUpdate:Vm" means give me all resourceUpdate messages that have kind vm.
Handler
the handler defines the function that the subscriber interface expects. Im not quite certain what the best way to handle errors from these handlers are.
Publish
package Messagebus
Type Publisher interface{
Publish(ctx context.Context, message Message) error
}The publisher interface is dead simple, you have a message and want to publish it. Notice the lack of destination or route, that is all contained in the message and should be decoded by the implementation. see the message object for more details.
Subscribe
package Messagebus
Type Subscriber interface{
Subscribe(ctx conetxt.Context, handler Handler, route Route) error
}The subscriber interface is slightly more complicated. we need to define a handler for the messages we receive and a route we want to listen to.
Message
package Messagebus
Type Message struct{
// Message metadata, do we need extra fields for acking, more time metadata?
Id Id
RunId RunId
TimeSent time.Time //k8s time to keep it consistent with the rest of the API?
// Routing info
Type MessageType //eg. ResourceUpdate, Event. The types can be defined as constants in the messagebus
Kind MessageKind //Kind the subtype of type, eg. vm.
Instance MessageInstance //instance of kind eg. vm.id
// Payload
Payload any //Metadata beyond the type and kind
}The message object carries metadata about the message, the routing information, and the payload. Some duplicate information is to be expected e.g the instance will often be the id of a resource which will also be found in the paylaod. But keeping this info on the message itself also, states explicitly what requirements needs to be presents for a message to make it through the bus.
The message object can be split into more objects if needed, such as metadata and routing, but for simplicity its kept in one.
Example: old way vs new way
Below is an example from the ROR code how the message bus is used today, and then the same code but with our proposed solution.
// No explicit function requirements, this is a function that relies on the application state and global variables.
// when calling this function its not apparently clear what it needs to operate. Only by examening it will you see that it will fail if
// the RabbitMQConnection is not initialized in the apiconnections module.
func StartListeningRabbitMQ() {
// Here we need to declare a rabbitMq queue, if for whatever reasone someone else would want to listen
// to this queue they would need to know how to declare it. which isn't documented. This can result in
// multiple simmilar queues for the same purpose
SSEventsQueueName := fmt.Sprintf("%s-%s", SSEventsQueueNamePrefix, uuid.New().String())
go func() {
// we need to know rabbitMq relatively well to use this listener.
// this config is very specific for rabbitMq and if we were to know what
// each paramter means to avoid making errors or to get or messages properly
// should we autoAck this queue? what does that imply for others who might be listening?
// this config should not be set by the user, but by the implementer of the message bus.
// this does mean that the user looses some flexibility, but most of the time the user just
// wants to get a message and not have to think about what "fanout" means.
// if for whatever reason they need some very specific feature, they can work together with the
// maintainers of the messagebus to find a solution.
config := rabbitmqhandler.RabbitMQListnerConfig{
Client: apiconnections.RabbitMQConnection,
QueueName: SSEventsQueueName,
Consumer: "",
AutoAck: false,
QueueAutoDelete: true,
Exclusive: false,
NoLocal: false,
NoWait: false,
Args: nil,
Exchange: SSEventsExchange,
ExcahngeKind: "fanout",
ExchangeAutoDelete: true,
ExcahngeDurable: true,
}
// The rest of this code is fairly straight forward and the only issue is that it uses
// rabbitMq directly.
rabbithandler := rabbitmqhandler.New(config, ssemessagehandler{})
// this referes to global application state and is hard to write tests for. It also drops errors.
_ = apiconnections.RabbitMQConnection.RegisterHandler(rabbithandler)
}()
}
type ssemessagehandler struct {
}
func (amh ssemessagehandler) HandleMessage(ctx context.Context, message amqp091.Delivery) error {
switch message.RoutingKey {
case SSERouteBroadcast:
err := HandleSSEEvent(ctx, message)
if err != nil {
rlog.Error("could not handle event", err)
return err
}
default:
rlog.Debugc(ctx, "could not handle message")
}
return nil
}
func HandleSSEEvent(ctx context.Context, message amqp091.Delivery) error {
if message.Body == nil {
return errors.New("message.body is nil")
}
var sseEvent SseEvent
err := json.Unmarshal(message.Body, &sseEvent)
if err != nil {
return err
}
Server.Message <- EventMessage{
Clients: Server.Clients.GetBroadcast(),
SseEvent: sseEvent,
}
return nil
}// Explicit function parameter: messageBus.Watcher
// This allows us to pass any messageBus client that implements the watcher interface to the function and run
// it, no need to setup global state first for this function to work. This allows us to crate mock clients for
// testing. It also communicates explicitly what this function requires to work, no hidden state.
// Since this will be calling a function that starts a goroutine we need to bring with us a context.
func StartListeningRabbitMQ(ctx context.Context, messageBusSubscriber messageBus.Subscriber) {
// We no longer have to define the config for a rabbitMq queue here
// nor do we have to start a new goroutine, all is handled by the messagebus implementation
// passed to this function.
// the rabbitMq handler has been switched out with a messagebus handle
// notice that using the messagebus directly means that we have to construct
// the routing string manually. Creating a package for resource update will allow us to
// supply more user friendly inputs, and construct the routing string for them
err := messageBusSubscriber.Subscribe(ctx, handle, "ResourceUpdate:cluster:<cluster-id>")
if err != nil {
[...]
}
}
// NOTE: sample handler, the handler interface was not fully developed when this example was created
func handle(ctx context.Context, message messageBus.Message) error {
[...]
}
Summary
This is a proposal of a new implementation of the ROR message bus. Today's solution is directly tied to RabbitMQ and exposes the users (both the developers of ROR, and third party developers of Microservices) for a lot of complexity. This proposal describes a new message bus package that abstracts away the complexity of the message bus infrastructure implementation. It also proposes a helper package, resourceupdate, that provides a set of helper functions to quickly get resource updates. This will decouple the code in microservices from the underlying implementation of the messagebus, today, RabbitMQ.
Motivation
The motivation behind this proposal is that we as Developers of Microservices want a simpler way to get messages about resource updates from the ROR-api.
Background
Todays solution utilizes RabbitMq directly and the user must be familiar with both RabbitMq and message bus architecture. We want to achieve two things with this design, decouple the message bus implementation from the ror code, and make it easier for the user to interact with the message bus. Although the design is simpler than working with a message bus library directly, it is still a generic library. Writing libraries on top of this framework, to handle specific cases e.g writing a library that provides types for specific messages related to resourceUpdates, Is encouraged.
Proposed Solution
The design has gone through a few iterations already, but it is not fully complete. The reason for this is that we want to involve the core ROR team in the discussion early on. No specific implementation is proposed. We want to test out a few message bus implementations, at least RabbitMq and NATS, and that will take time. Thanks to Kjetil Sigvartsen, Kevin Vatn, Sigurd Skogmo, Morten Ottestad and Tor Halvor Frivold, for providing feedback on different iterations of the design.
This proposal only describe a receive once system, where each message is consumed and not stored on the queue. If we want to support this we should create a separate Stream proposal to keep this proposal focused and small.
The solution as a whole is a set of interfaces that define a publisher, subscriber model, with a topic exchange (rabbitMq term used to clarify a bit what we mean) behind the scenes. The reason we decided on a topic exchange design, is that we use both agents and microservices to manage our resources and they each have different granularity of concern when it comes to resources. Agents care about single instances of resources, while microservices, in most cases, care about every instance within a kind. This difference in granularity makes it difficult to use single queues per resource kind, as it would make for complex handling logic for agents.
Publishing messages
The philosophy of this system is that producers don't ask for things to be handled. They instead announce that something has happened, the api does not ask the microservice to change the resource, it publishes a message telling the bus that the resource status has changed, then it's up to the microservice to act upon that message. The sender does not expect an answer back from the microservice that it has started working. When the microservice has changed the resource to allign with the status the api will publish a message saying the resource spec has changed, and then other consumers may act upon that message.
Why not simple queues
When making the design we came up with three scenarios when using simple queues.
If we were to use simple queues we would quickly need to have a queue for each instance of a resource which would not scale well.
This is where topic exchanges comes in. They allow us to dynamically create queues of the granularity we need, initiated from the clients side. Using a simple routing scheme, we can use the Type, kind and instance id of a resource (or other type of event) to create queues. This means that agents will get queues that only contain messages for a specific resource instance, and microservices will get queues that contain all messages for a resource kind. All of this logic needs to be supported in the provider, so it limits us to providers that support topic exchange like functionality. This will result in less queues than having a queue per instance of a resource, but significantly less because where we have a lot of instances of resources we use microservices to handle those anyways. e.g we have a thousands of backup runs/jobs, while we have a few hundred clusters.
Another problem is that we don't want to expose our message bus outside of the cluster directly. Instead we want to forward messages via SSE, this means that our SSE server needs to satisfy the interfaces so it can act as a propper message bus client.
The full flow is described in the figure below.
Simple datatypes
The design builds on a few simple datatypes and three main datatypes, the publish interface, the subscribe interface, and the message struct.
Message type, kind, Instance and route
The routing of messages rely on the Type, Kind and Instance hierarchy. The route looks like this "Type:Kind:Instance". Instance and kind can be omitted to act as a wildcard, essentially meaning match all of that type.
e.g "ResourceUpdate:Vm" means give me all resourceUpdate messages that have kind vm.
Handler
the handler defines the function that the subscriber interface expects. Im not quite certain what the best way to handle errors from these handlers are.
Publish
The publisher interface is dead simple, you have a message and want to publish it. Notice the lack of destination or route, that is all contained in the message and should be decoded by the implementation. see the message object for more details.
Subscribe
The subscriber interface is slightly more complicated. we need to define a handler for the messages we receive and a route we want to listen to.
Message
The message object carries metadata about the message, the routing information, and the payload. Some duplicate information is to be expected e.g the instance will often be the id of a resource which will also be found in the paylaod. But keeping this info on the message itself also, states explicitly what requirements needs to be presents for a message to make it through the bus.
The message object can be split into more objects if needed, such as metadata and routing, but for simplicity its kept in one.
Example: old way vs new way
Below is an example from the ROR code how the message bus is used today, and then the same code but with our proposed solution.