Protocols

It is crucial for any [distributed] system to precisely define the protocols that the components use to communicate. Having precise protocols allows us to be precise about the compatibility of the different microservices and to precisely explain to our clients the messages that the system produces. Moreover, we can use these protocol definitions to accelerate the implementation of the different microservices: if each microservice knows the protocol, it can trivially validate its inputs and it can generate synthetic outputs. This allows us to build a walking skeleton: a complete implementation of all the microservices and message paths, without having to spend the time to implement the functionality of each of the microservices.

A good protocol definition gives us the flexibility to maintain compatibility even as we add and remove fields. There are a number of protocol languages and tools; however, the mature ones aren’t simply languages to describe protocols. Mature protocol tooling generates implementations for many target languages and runtimes, and the generated code fits well into the target language and runtime. It should also be possible to use the protocol tooling to derive as much value as possible: think documentation, tests, naive implementations, and many more.

syntax = "proto3";
package com.reactivearchitecturecookbook.ingest.v1m0;

message IngestedImage {
  string mime_type = 1;
  bytes content = 2;
}

syntax = "proto3";
package com.reactivearchitecturecookbook;

import "google/protobuf/any.proto";

message Envelope {
  string correlation_id = 1;
  google.protobuf.Any payload = 4;
}

include(FindProtobuf)

file(GLOB_RECURSE PROTOS ${CMAKE_CURRENT_SOURCE_DIR}/../protocol/*.proto)
protobuf_generate_cpp(PROTO_SRC PROTO_HEADER ${PROTOS})
set(CMAKE_INCLUDE_CURRENT_DIR TRUE)
include_directories(${PROTOBUF_INCLUDE_DIR})

scalaVersion := "2.12.1"
libraryDependencies += "com.trueaccord.scalapb"
                    %% "scalapb-json4s"
                     % "0.1.6"

libraryDependencies += "com.trueaccord.scalapb"
                    %% "scalapb-runtime"
                     % com.trueaccord.scalapb.compiler.Version.scalapbVersion
                     % "protobuf"

PB.includePaths in Compile ++= Seq(file("../protocol"))
PB.protoSources in Compile := Seq(file("../protocol"))

PB.targets in Compile := Seq(
  scalapb.gen(flatPackage = true) -> (sourceManaged in Compile).value
)

protocol
  com
    reactivearchitecturecookbook
      envelope.proto
      ingest-v1m0.proto
      faceextract-v1m0.proto
      faceextract-v1m1.proto
      ...
      faceextract-v2m0.proto

// Given arbitrary gen instance...
const ingest::v1m0::IngestedImage gen;

// ...we expect the following to hold.
ingest::v1m0::IngestedImage ser;
ser.ParseFromString(gen.SerializeAsString());
ASSERT(ser.content() == gen.content());
ASSERT(ser.mime_type() == gen.mime_type());

#include <gtest/gtest.h>
#include <rapidcheck.h>
#include <rapidcheck/gtest.h>
#include "protobuf_gen.h"
#include <ingest-v1m0.pb.h>

using namespace com::reactivearchitecturecookbook;

RC_GTEST_PROP(main_test, handle_extract_face,
              (const ingest::v1m0::IngestedImage &gen)) {
    ingest::v1m0::IngestedImage ser;
    ser.ParseFromString(gen.SerializeAsString());

    RC_ASSERT(ser.content() == gen.content());
    RC_ASSERT(ser.mime_type() == gen.mime_type());
}

mime_type = , content = *jazdfwDRTERVE GFD BHDF
mime_type = .+*-<,7$%9*>:>0)+, content = 	

aE@TEVD BF
mime_type = )< ?3992,#//(#%/08),/<<3=#7.<-4), content = 13ZXVMADSEW^
...

Authentication and authorisation

Following the diagram on Figure 2-4, we need to make sure that we only accept authorized requests. More specifically, we need to make sure that each service is able to decide whether it is authorized to process the consumed message. The authorisation mechanism needs to give us more information that simple Boolean. Consider the output of our processing pipeline; all that we need to do is to make HTTP POSTs to a URL that belongs to the client (we mean someone interested in collecting the output our system produces). That sounds simple enough, but how do we compute the URL where the POSTs should be sent? We certainly need to identify the client in all messages, all the way from the ingestion microservice to this microservice.

The easiest approach would be to require every request hitting the ingestion microservice to include the URL where the responses should be pushed. While trivial, this is a terrible design choice: it allows anyone to direct the results to a URL of their choice. If this system processed sensitive information (think security camera images, identity and travel documents, or indeed performed some biometric processing), the ability to specify arbitrary URL in the request for the delivery of the results will result in leaking of such sensitive data; even if privacy did not concern you, the ability to specify arbitrary URL will result in attackers using this system to perform DOS-style attacks on the given URL.

The second approach would be to include a client identifier in each request—and message—then add a database to the push microservice, which would perform mapping from the client identifier to the URL. This would remove the DOS attack security hole, but would still leave us exposed to leaking data to the wrong client. This is a typical identity management, authentication, and authorisation scenario.

Let’s assume we have identity management and authentication services, and explore the authorisation approaches. In monolithic applications, we typically relied on server-side session. A client would authenticate and upon success, we stored the details of the authentication in the server’s memory³ under a unique identifier. The client typically stored this identifier in a cookie, which it presented on every request. This allowed the code on the server to look up the authentication in the session; the authorisation value was used to authorise the requests. This represents a reference-based authentication: in order to access the authentication value, we need a client-side value (the cookie) and a service, which can resolve the reference to the authentication value. In a distributed system, we need to move to a value-based authentication, where the client-side value is the same as the server-side value, and can be directly used for authorisation.

We build a token whose payload encodes the entire authentication detail and include a mechanism to verify the token’s authenticity without the need of any further services—the answer to the question “have we issued this exact token?” We require it to be passed to the ingestion service, and include it in every message in the system. This token can be used to authenticate the requested operations or access to the requested resources. The verification mechanism can use digital signature to verify that the token is indeed a valid token. While this allows us to verify that no-one has tampered with the token, it allows everyone to examine the token. An alternative is to use asymmetric encryption, where we encrypt the token using a public key, decrypt using a private key. A successful decryption means that a matching public key was used to encrypt it; consequently, that the token has not been tampered with. However, every microservice that needs to decrypt the token must have access to the matching private key⁴.

Adding the token to our messages is a great showcase of how important it is to have well-defined protocols, and how important it is for the protocol tooling to have good support for all the languages that we use in our system. The Envelope with the added token field is shown in Example 2-10.

syntax = "proto3";
package com.reactivearchitecturecookbook;

import "google/protobuf/any.proto";

message Envelope {
  string correlation_id = 1;
  string token = 2;
  google.protobuf.Any payload = 3;
}

The ingestion microservice extracts the value for the token field from the Authorization HTTP header (using the Bearer schema), and is the first service to verify that the token is indeed valid and that it contains the authorisation to access the ingestion service. We use the JSON Web Token defined in RFC7519 [Link to Come], but explained in a much more user-friendly way at https://jwt.io.

The JSON Web Token allows us to define any number of claims; think of each claim as the bearer claims to have authorisation to do x, where x is a value that the each microservice understands. In the system we’re building, we use a simple naming convention for the claims: if the bearer is authorized to use the 1.x version of the faceextract microservice, the token contains the faceextract-1.* claim; if the bearer is authorized to use any version of the ocr microservice, the token contains the ocr-* claim. The value of these claims is specific to each microservice. Version 1.0 of the faceextract service does not need any further information about a claim, a simple Boolean is sufficient; the latest version of the OCR microservice needs complex configuration for the OCR features the bearer is authorized to use. This is a very important aspect of using token-based authorisation: the authorisation can contain very specific details and settings.

All microservices in the Image Processing System use encrypted JSON Web Tokens, which adds the complexity of good key management system (the source code includes the private and public keys as files, but that is certainly not a good practice for highly secure systems), but prevents the clients from examining the payload in the token. This system allows the end devices (think mobiles, IoT cameras, etc) to perform their processing to improve the user experience, but it does not trust the conclusions of any classification code on the devices; the final conclusions are computed entirely within this system. Again, the JWT libraries are available for C++ as well as Scala / Java.

Event-sourcing

Services that maintain state (particularly in-flight state), but are also able to recover from failures by restarting and reaching the same state as the failed instance, the services need to be able to re-process messages starting from the last known good message.

Depending on the communication mechanism between the microservices, we either have to provide each micoservice with its own event journal, of—if the message broker supports it—we can use the broker as the event journal⁵. Regardless of the event-sourcing mechanism (the message broker or each microservice’s event journal), the flow of processing events and writing snapshots and offsets into the journal remains the same.

Figure 2-5. Event-sourcing using message broker

Upon start, the microservice loads the offset of the last message from where it needs to consume messages in order to arrive at a well-known state. In this case, the microservice needs to consume three messages before it can produce one output message; upon producing the output message, its state is empty. (This is a special case of a consistent state, which can be persisted as a snapshot.)

The service subscribes to receive messages from the broker starting from the loaded offset,

The broker delivers the messages to the microservice; if the service fails during the processing of the messages, its new instance will start again from step <1>. Your runtime infrastructure should detect and prevent process thrashing, where the service keeps restarting, because the crash is triggered by one of the messages.

The service has processed all three input messages, its state now allows it to produce one output message; the broker acknowledges receipt of the message.

When the output message is successfully placed on the output, the microservice can write the offset 0x98 to its offsets store.

If there is a very large difference in processing complexity between consuming and validating messages and acting on the consumed messages, or if there are great variations in the velocity of the incoming messages, it will be necessary to split the microservice into the write and read sides. The write side treats the incoming messages as commands. The write side validates the command and turns in into an event, which is appended to the journal. The write side should not contain any logic responsible for dealing with the events: its responsibility is to turn the incoming command into an event in the journal as quickly as possible. The read side consumes the events the write side has appended to the journal (automatically with some delay or explicitly by asking for updates), and performs its logic. Importantly, the read side cannot append to the journal: it is responsible for acting on the events. Splitting the microservice into the two parts allows us to scale each part differently, though the exact rule for scaling depends on the exact nature of the microservice, though your aim is to balance the throughputs of the read and write sides.

val broker = Broker()
broker.subscribe("topic-1") { 
  case (messages, offsets) => 
    handle(messages) 

    broker.confirm(offsets) 
}

Partitioning and replication

The nature of offset store means that it is a good approach to divide the broker into several topics, the values in the offsets store refer to individual topics. Even in modestly large systems, the messages in one topic would not fit in one node (fit refers to the I/O load and the storage space to keep the payloads, even if old messages are regularly garbage-collected), so a topic has to be partitioned. Each topic partition has to fit on one node (think durable storage space for the messages until they are garbage-collected); a convenient consequence is that because the messages in a topic partition are on one node, there can be deterministic order of the messages.

Partitioning helps us with distributing the messages (and the associated I/O and storage load) on multiple broker nodes; we can also have as many consumers as we have partitions, allowing us to distribute the processing load. However, partitioning doesn’t help us with data resilience. With partitioning alone, we cannot recover catastrophic failures of partitions. To do so, we need to replicate the topic partitions: replication ensures that we have copies of the messages in the partitions on multiple nodes. The price of partitioning is loss of total order; the price of replication is that we have to think about the balance of consistency, availability, and partition tolerance. We would like to have all three, of course, but we can only have two strong properties. Fortunately, the three properties of distributed systems are not binary: there are multiple levels of consistency, which influences the degree of availability and partition tolerance. The complexity of selecting the right CAP values is a decision for the engineering and business teams; it is a game of trade-offs. Nevertheless, with appropriate partitioning and replication, we can provide elasticity, resilience, and responsiveness.

Limiting impact of failures

Good protocols, value-based semantics, event-sourcing (and, where applicable, CQRS) all serve to allow the services to remain available even in failure conditions. Let’s tackle failures that might seem insurmountable, but with careful technical and business consideration, we can define graceful degradation strategies.

Let’s consider failure catastrophic failure in the database that contains the identities for the authentication service to use. If the business requires that users are able to log in regardless of how degraded the system as a whole becomes, you should consider allowing the authentiction service to issue the authentiction token for a typical authentication details without actually performing any authentication checks. The same applies to the authorisation service: if its dependencies fail, consider issuing very restricted allow typical usage token, regardless of what was passed in. The risk we are taking on here is that the system grant access to users that should not have been allowed in, but the damage to the business would be greater if legitimate users were not allowed to use the system.

The failure of outward-facing systems is easiest to describe, but there can be just as severe internal failures that our microservices can tolerate and where the impact on the business is well-understandable. The event-sourced microservice can tolerate failures of its offsets store. The business decision will drive the length of time it can tolerate the failure for, and what it does if the failure is persistent or permanent. If the microservice is running, and the offset store becomes unavailable, we risk having to re-process growing number of messages in case of failure or scaling events.

The business impact is defined by the computational cost to re-process already successfully processed messages (remember, we could not write the latest offset to the offset store!), and the impact on the customers who will receive duplicate messages. Similarly, if the offset store is unavailable during the microservice’s startup, the business decision might be to start processing at offset defined as offset_last-100, or even zero. The business risk is that some messages might not be processed, or that there will be too many needless messages re-processed. Nevertheless, both might be perfectly acceptable compared to the service being unavilable.

Good example of limiting the impact of failures is the summary microservice, which tolerates temporary failures in its offset store and the push microservice, which tolerates temporary failures in its journal. The authorisation microservice tolerates total failures of its dependencies: in that case, the service issues allow every idempotent workload tokens to the clients, which the business deemed to be a good graceful degradation strategy. The clients can still submit images to be processed—these idempotent requests—but non-idempotent requests are not authorized. The tokens have short expiry date with random time delta added to each one. The clients to refresh the tokens on expiry, but the random time deltas in the expiry dates spread the load on the authorisation service once it recovers.

Back-pressure

Without back-pressure, the information flows in the system only in one direction, from source to sink; the source sets the pace and the sinks just have to cope. If the system is balanced, then the sinks can cope with the data the source produces, but a spike in usage will disturb this balance. Let’s explore what happens when the upstream service accepts requests from external systems and is capable of handling much greater load than the downstream service in Figure 2-6.

Figure 2-6. Back-pressure

In happy-days scenario, the upstream service receives requests from external systems at the rate that the downstream service can process. Without back-pressure, the system happens to work, because the load is within the predicted range. With back-pressure, the system also works well, because the downstream service tells the upstream component how much load it is ready to consume.

When there is a spike in the incoming load, the situation changes dramatically. Without back-pressure, the upstream accepts all requests, then overwhelms the downstream service. With back-pressure, the upstream service knows that downstream services can only process 10 messages, so it rejects most of the incoming requests, even though it could handle all the extra load. Replying with errors is not great, but it is better than causing failures that spread through the entire system.

The cascade of failures spreads from the downstream service, which can no longer handle the load pushed to it. This might cause failures in the message broker (its partitions grow too large for the nodes that host them), which ultimately causes failures in the upstream component. The result is that the system becomes unavailable (or unresponsive) for everyone. In case of REST API, this is the difference between receiving HTTP 503 (over capacity) immediately, or waiting for a long time and receiving HTTP 500 (internal server error) or no response at all.

Systems with back-pressure can flex their capacity if there are intermittent capacity problems or transient failures; the upstream services remain available and responsive, even if that means rejecting requests that would take the system over its immediate capacity.

It isn’t possible to dismiss back-pressure as irrelevant simply because the system uses a message broker to connect two services. While it is true that the broker can provide short-term buffer for systems without back-pressure; nevertheless, if the downstream messages do not keep up with the incoming message rate in the broker, the time for a specific message to travel through the system will increase. If the increase in load is not just a momentary spike, the downstream services will be busy dealing with queued messages and starting to breach the performance service levels, while newer messages are being queued up; their performance service level is breached even before they reach the downstream service. Put bluntly, you will have a system whose services are running at full capacity, without failures, but the result is nothing but timeouts. Before we conculde, it is important to point out that, even though the back-pressure portion of Figure 2-6 does not explicitly show it, the back-pressure reporting must be asynchronous. If the upstream service blocks on the poll operation to find out how much the downstream component can accept, then the upstream service becomes unresponsive.

All microservices in the Image Processing System use back-pressure in the libraries that connect them to the message broker; the ingestion and push microservices rely on Akka HTTP to manage back-pressure using the underlying asynchrnous I/O that handles the incoming and outgoing network connections.

External interfaces

Within our own system, we can implement back-pressure from start to end, but we have to take into account external systems. We already know that we don’t control the load we receive from the external systems, so it is important to reject requests that would result in overloading our system. Even though our system is in the position of some external system for our integration clients, we should do our best to not cause a denial-of-service “attack” on our clients.

Because we can’t assume any details about the client systems, we have to start from the position of no back-pressure reporting from the client. We have to take cues from the underlying I/O to deduce the acceptable load⁶. When we detect that we are over-loading the client (because we are beginning to get timeouts or error status codes), it is important not to make things worse by aggressively re-trying the outgoing requests. This means that we have to consider some type of dead-letter store, where we keep the failed requests to retry later. Moreover, a push-style interface requires the clients to have a public-facing endpoint. This endpoint should require secure transport, which is trivial to implement, but it also means that our system is responsible for checking that the connection is indeed secure. And so we have to verify the endpoint’s certificate, which means that we are now responsible for maintaining the certificate chains, checking revocations, etc; and it takes only one client to request that we handle self-signed certificate or clear-text connections to significantly increase the complexity of the push service.

Firehose

An alternative approach is to have a firehose API, where the clients open a long-running HTTP POSTs to our systems, indicating the last-successful offset (and other parameters such as maximum data rate and any filters to be applied). Our system then sends response chunks that represent batches of messages (the same ones that the push microservice would send), which means that our clients don’t have to do as much work to implement public-facing endpoint, and they are ultimately in control of the data rate: the client’s ingestion system can drop the connection whenever it wants.

In both cases, the push and firehose services need to maintain their own journals. Relying on the broker’s message journal would work well if we had well-known and fixed number of clients (allowing us to structure the queues / partitions appropriately), but we hope that the number of clients will grow. This means that we should maintain our own journal, disconnected from the broker’s journal, which means message duplication (a message is stored in the broker’s journal as well as the microservice’s journal; moreover, the services cannot share their data stores, which means that the same message is in the firehose and push journals). However, we gain finer control over partitioning and message grouping. Figure 2-7 shows the final architecture.

Figure 2-7. Loosely coupled architecture

We are now ready to explore the important implementation details in each of the services.

Ingestion microservice

The ingestion microservice is responsible for accepting the requests containing the images from the clients, checking that the client is authorized to make the request, then checking that the requests contain valid data (i.e. is the HTTP POST body really an image?); finally, the ingestion microservice packs the validated payload into our envelope and publishes it to an outgoing queue. Once all of this work is done, the microservice replies to the caller with a simple acknowledgement response. If the client receives HTTP 200, we have validated and ingested the request, and we will complete its processing at some point in the future. As far as the client (think another system, web, or mobile application using our REST API) is concerned, it will not receive any further communication.

trait IngestionRoute extends Directives with GenericUnmarshallers { 
  def authorizeAndExtract(token: String): Try[JWTClaimsSet] 
  def extractIngestImage(request: HttpRequest)
                        (implicit ec: ExecutionContext): Future[IngestedImage] 
  def publishIngestedImage(claimsSet: JWTClaimsSet, token: String,
                           transactionId: String)
                          (ingestedImage: IngestedImage)
                          (implicit ec: ExecutionContext): Future[Unit] 

  private def authorizedRouteHandler(requestContext: RequestContext,
                                      token: String, transactionId: String)
                                    (claimsSet: JWTClaimsSet)
                                    (implicit ec: ExecutionContext): Future[RouteResult] = { 
    extractIngestImage(requestContext.request)
      .flatMap(publishIngestedImage(claimsSet, token, transactionId))
      .flatMap { _ => requestContext.complete("") }
      .recoverWith { case x => requestContext.fail(x) }
  }

  def ingestionRoute(implicit ec: ExecutionContext): Route = 
    path(RemainingPath) { tid => 
      post { 
        extractCredentials { 
          case Some(credentials) if credentials.scheme() == "Bearer" =>
            request: RequestContext => 
            authorizeAndExtract(credentials.token())
              .map(authorizedRouteHandler(request, credentials.token(), tid.toString()))
              .getOrElse(request.reject(AuthorizationFailedRejection))
          case _ => _.reject(AuthorizationFailedRejection)
        }
      }
    }

}

We mixin the IngestionRoute trait to the main class and implement the steps in the ingestion pipeline in the abstract methods; notably, they all return Futures, implying non-blocking implementations.

The authorizeAndExtract function takes the String that claims to be the JSON Web Token the authorization service issued. The implementation is expected to parse it, validate that it was indeed the authorization service that issued it, and return the JWTClaimsSet wrapped in Try.

The extractIngestImage function takes the HttpRequest and is responsible for checking that the request body does indeed contain bytes that are a valid image. It returns a Future[IngestedImage], allowing the implementations to perform complex asynchronous validation, if needed.

The publishIngestedImage is responsible for taking—amongst others—the IngestedImage and placing it on the output topic. It returns the Future[Unit] once it receives confirmation from Kafka that the message has been placed on the requested topic. (We say Future[Unit] because we aren’t interested in the value of the result; we only want to be able to tell the difference between Success(()) and Failure(()).)

The authorizedRouteHandler implements the processing pipeline that is applied to the incoming HTTP requests.

A Route is a function mapping RequestContext to Future[RouteResult]; we construct this function by using Akka HTTP’s DSL.

The path directive matches the request path on its parameter; in this case, we want the entire remaining path.

We only accept the HTTP POST method, so we use he post directive.

To extract the credentials supplied with the request (in the Authorization HTTP header), we apply the extractCredentials directive and then match on its optional result.

We finally use the steps in our processing pipeline by checking the authorization token first, then mapping the successful value through the authorizedRouteHandler; failures result in the appropriate HTTP rejection

It is important to notice the asynchrnous nature of the entire processing pipeline. The ingestionRoute returns a function that Akka HTTP uses in its low-level actor runtime to service the incoming HTTP requests. Akka HTTP handles not just the incoming route, but also bakes in back-pressure management and reporting.

Vision microservices

The vision microservices each perform different computer vision task on the same input data. Because the computer vision code is implemented in C++ & CUDA, and because we want to keep the complexity of the microservice to a minimum, we are not going to create wrappers that make the native code accessible from Java or Scala code. After all, one of the points of microservices is that we should be able to use the most appropriate language or toolkit for each microservice.

Regardless of the language, the microservice fit into our message-driven infrastructure. The computer vision work results in zero or more output messages that go on the output Kafka topic. Before we consider the implication of this design, let’s review the core principles of Kafka messaging: we need to understand topics, partitions, and consumer groups.

A topic is a concept of a basic pipe, identified by some name, that transports the bytes that make up the payloads of the messages from producers to consumers. Kafka parallelises the message delivery by partitioning the topic.

A partition is essentially a directory with journal files for the messages; a partition has to fit on one node in terms of the size of the journal and the I/O performance. (If you have one partition on a particular topic, that topic will not scale on multiple nodes, even if they are available; if you have a topic with 1000 partitions, you could potentially make use of the processing and I/O power of 1000 nodes.) Because one partition is fully contained in one node, each partition is totally ordered; however, a topic with more than one partition is not totally ordered. The partition count therefore defines the consumer parallelism: we can consume as many batches of messages as there are consumer groups. We can grow the number of partitions, but the existing messages in the topic will not be repartitioned.

The consumption of messages is handled by a consumer group, which is a set of threads that receive the messages in the partitions. Kafka selects exactly one thread to deliver it a batch of messages from a partition.

To help us think about the messaging in Kafka, and to help us understand topics, partitions and consumer groups, we say that:

• topics are divided into partitions; the number of partitions defines the maximum consuming parallelism,

• each consumer group receives the same partitions (containing the same messages),

• if there are more consumer threads than there are partitions on a topic, some threads will never receive a message,

• if there are more partitions than threads, some threads will receive data from multiple partitions,

• if a consumer thread subscribes to multiple partitions there is no guarantee about the order the received messages, other than that within the partition the offsets will be sequential. For example, the thread may receive 5 messages from partition 10 and 6 from partition 11, then 5 more from partition 10 followed by 5 more from partition 10 even if partition 11 has data available,

• adding or removing the consumer group threads will cause Kafka to re-balance, possibly changing the assignment of a partition to a thread,

• in high-level consumers, Zookeeper maintains the offsets of the delivered messages for each partition in a consumer group,

• in low-level consumers, the consumers are responsible for maintaining the offset stores, allowing them to control the offset from which they want to subscribe.

Figure 2-8 illustrates three consumers subscribed to one partition within a consumer group.

Figure 2-8. Kafka messaging

Let’s explore how the messages are consumed:

We are starting when two new messages in a topic partition arrived after the last offset,

Kafka selects one consumer thread (from a known set of consumer threads it obtained when re-balancing the partitions) and delivers it a batch of messages, the consumer processes the batch of messages and,

confirms the processed offsets within the configured timeout (which should be as small as possible),

the new offset is moved, effectively marking the messages in the journal as processed;

when new messages arrive, the process repeats.

Confirming the consumption not at the point of message arrival, but after successful processing explains how we can guarantee at-least-once delivery semantics. In the happy-days scenario, we complete all our processing within the time Kafka gives us to confirm the offsets of the received batch of messages. If the batch of messages is lost on the way from Kafka to the service, or if the service receives the messages, but Kafka does not receive the confirmation within a configured timeout (because the service failed or because the confirmation was lost), Kafka will repartition the topic and select a different consumer thread to handle the batch starting from the last known confirmed offset.

Consumer consumer;
Producer producer;
consumer.subscribe(std::vector<std::string>{"in_topic"});
while (true) {
    auto in_message = consumer.consume();

    auto out_message = process(in_message);
    producer.produce("out_topic", out_message);
    consumer.commit(in_message);
}

using commit_opaque = std::tuple<
    std::shared_ptr<std::atomic_long>,
    std::shared_ptr<RdKafka::TopicPartition>>;

class commit_dr_cb : public RdKafka::DeliveryReportCb { 
private:
    std::shared_ptr<RdKafka::KafkaConsumer> consumer_; 
public:
    commit_dr_cb(std::shared_ptr<RdKafka::KafkaConsumer> consumer) :
      consumer_(consumer) {
    };

    void dr_cb(RdKafka::Message &message) override {
      commit_opaque *co = static_cast<commit_opaque *>(message.msg_opaque());
      auto ctr = std::get<0>(*co);
      if (std::atomic_fetch_sub(ctr.get(), 1L) == 1L) {
          auto tp = std::get<1>(*co);
          if (message.err() == RdKafka::ERR_NO_ERROR)
              consumer_->commitAsync(std::vector<RdKafka::TopicPartition *>{tp.get()}); 
          delete co; 
      }
    }
};
std::atomic_bool run(true);
const std::string in_topic_name = "ingest-1";
const std::string out_topic_name = "vision-1";
const std::string key = "key";
auto conf = std::unique_ptr<RdKafka::Conf>(RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL));
auto consumer = std::shared_ptr(RdKafka::KafkaConsumer::create(conf, ...));
commit_dr_cb dr_cb(consumer);
conf->set("dr_cb", &dr_cb, err_str); 
auto producer = std::unique_ptr(RdKafka::Producer::create(conf, ...));
auto out_topic = std::unique_ptr(RdKafka::Topic::create(producer.get(), out_topic_name, ...));
consumer->subscribe(std::vector<std::string>({ in_topic_name })); 
while (run) {
    auto in_message = std::unique_ptr<RdKafka::Message>(consumer->consume(10)); 
    if (in_message->err() != RdKafka::ERR_NO_ERROR) continue;
    std::string out_payload = ...; 
    commit_opaque* opaque = new commit_opaque(
        std::make_shared<std::atomic_long>(1L),
        std::shared_ptr<RdKafka::TopicPartition>(
          RdKafka::TopicPartition::create(
            message->topic_name(), message->partition(), message->offset()
          )
        )
    ); 
    const auto resp = producer->produce(out_topic.get(), partition,
        RdKafka::Producer::RK_MSG_COPY,
        const_cast<char *>(out_payload.c_str()), out_payload.size(),
        &key, opaque); 
    producer->poll(0); 
    if (resp != RdKafka::ERR_NO_ERROR) delete opaque; 
    
}
consumer->close(); 
producer->flush(1000); 

namespace fe = faceextract::v1m0;
namespace in = ingest::v1m0;

template<typename T, typename U, typename Handler>
std::vector<Envelope> handle_sync(
    const std::unique_ptr<RdKafka::Message> &message, Handler handler) {
    ...
}
const auto partition = RdKafka::Topic::PARTITION_UA;
auto message = std::unique_ptr<RdKafka::Message>(consumer->consume(10));
if (message->err() != RdKafka::ERR_NO_ERROR) continue;
const auto key = message->key();
const auto out_envelopes = handle_sync<in::IngestedImage, fe::ExtractedFace>(
    message,
    [](const auto &, const auto &) { return std::vector<fe::ExtractedFace>(...) });
    commit_opaque* opaque = new commit_opaque(
        std::make_shared<std::atomic_long>(out_envelopes.size()),
        std::shared_ptr<RdKafka::TopicPartition>(RdKafka::TopicPartition::create(
            message->topic_name(), message->partition(), message->offset()
        )
    )
);
for (const auto &out_envelope : out_envelopes) {
    const auto out_payload = out_envelope.SerializeAsString();
    const auto resp =
        producer->produce(out_topic.get(), partition, RdKafka::Producer::RK_MSG_COPY,
                          const_cast<char *>(out_payload.c_str()), out_payload.size(),
                          &key, opaque);
}

producer->poll(0);

Push microservices

We construct the token using the standard fields, but add the push string claim, which holds the URL that the push microservice should use. For simplicity, we say that if the token does not include the push claim, then the caller is not authorized to use the push microservice. It seems that all there is left for us to do is to subscribe to the appropriate Kafka topics, consume the messages, and make the appropriate push requests in Example 2-16.

class PushActor(jwtDecrypter: JWEDecrypter) extends Actor {
  implicit val materializer = ActorMaterializer()
  private val pool = Http(context.system).superPool[Unit]()
  private[this] val kafkaConsumerActor = context.actorOf(...)

  private def push(record: ConsumerRecord[String, Envelope]): Unit = {
    (for {
      jwt <- Try(EncryptedJWT.parse(record.value().token)) 
      _ <- Try(jwt.decrypt(jwtDecrypter)) 
      uri <- Try(Uri(jwt.getJWTClaimsSet.getStringClaim("push-1"))) 
      entity = HttpEntity(ContentTypes.`application/json`,
                          JsonFormat.toJsonString(record.value().payload)) 

      rq = HttpRequest(method = HttpMethods.POST, uri = uri, entity = entity) 
    } yield (rq, ())).foreach(x => Source.single(x).via(pool).runWith(...)) 
  }

  override def receive: Receive = {
    case MessageBatch(consumerRecords) =>
      consumerRecords.recordsList.foreach(push) 
      kafkaConsumerActor ! Confirm(consumerRecords.offsets)
    case (Success(resp@HttpResponse(_, _, entity, _)), _) => 
      resp.discardEntityBytes()
    case (Failure(_), _) => 
      // bad request
  }

}

When a batch of messages arrives from Kafka, we apply the push function to each message; this is an asynchronous operation, so we are able to confirm the delivered offsets immediately,

we must handle HTTP responses from the client, at the bare minimum, we must ensure that we consume the bytes that the client sent to allow the asynchronous I/O to complete; we must do this even if we do not care about the content of the response,

we handle the failure in a no-op operation,

the processing of each ConsumerRecord[String, Envelope] starts by checking whether the supplied token is valid,

given a valid token, we verify that the push-1 claim is present and that it can be parsed into a Uri,

the entity that will be sent out is a JSON representation of the message’s payload,

the request in our simple implementation is a HTTP POST with the given entity,

finally, we use Akka’s pooled, cached, and back-pressure aware HTTP client to begin handling the request

This code is similar to Example 2-20, but it is clearly not handling catastrophic failures well. The underlying Akka HTTP machinery handles back-pressure from the client, with so that if we flood it with requests it cannot handle, we are notified in the actor’s behaviour, but there is nothing we actually do about the errors. Moreover, if the node hosting this microservice crashes after confirming the offsets to Kafka, but before it can receive responses (successful or not) from the client, we never go back to the messages we failed to send. (We have already confirmed the offsets, and we’re in auto-partitioning mode.) “Not a problem,” you say, “we’ll just use the auto-partitioning mode with manual offsets, and we’ll persist the offsets.” This would work well if we had the as many (or more) partitions as clients; recall that we use the client identity as the partitioning key. If we have fewer partitions than clients, the latest sent offsets will include messages for multiple clients. While the clients no doubt understand that there could be duplicate messages, we should limit the duplicates to an absolute minimum.

Resilient push service

To implement a push service that is “nice” to our customers (it backs off in case the client’s endpoint starts failing; it tries to send the failed requests later), but it does not lose any requests before it delivers them to the client, we could use the broker-backed event-sourcing approach. We would create a topic with as many partitions as we have clients and then fall back on the auto-partitioning with manual offsets mode of operation. However, this approach will not scale well if the number of clients is difficult to predict⁷. So, we need to use our own journal.

The PushActor becomes PersistentActor, which gives it the ability to write events to a journal. (The type of journal and details of how to connect to it are configurable.) Having persistence changes the flow of processing to one where as soon as we have validated the messages, persisted them as events in our journal, we confirm the consumed offsets to Kafka. We perform the actual HTTP push operations at some later point. We have in effect divided the microservice into write and read sides: we have a CQRS/ES microservice! (Remember that the only motivation for this complexity is because we want to implement service that can recover its state in case of failures, and because the velocity of the messages arriving might be completely different than the velocity with which we are able to output the messages.)

Figure 2-9. CQRS/ES

Even though CQRS/ES implementation shown in [Link to Come] is a complex piece of code, Akka handles most of this complexity.

class PushActor(...) extends PersistentActor {
  ...
  override def receiveCommand: Receive = {
    case PushActor.extractor(consumerRecords) =>
      val requests = consumerRecords.recordsList.flatMap { ... } 

      persistAll(requests) { _ => 
        kafkaConsumerActor ! Confirm(consumerRecords.offsets) 
      }
  }
}

object PushView {

  def apply(tag: String, redisClientPool: redisClientPool)
           (implicit system: ActorSystem, materializer: ActorMaterializer) = ... 

}

class PushView(tag: String, offset: Offset, redisClientPool: redisClientPool)
              (implicit system: ActorSystem, materializer: ActorMaterializer) {
  private val pool = Http(system).superPool[Offset]() 
  private val readJournal: RedisReadJournal =
    PersistenceQuery(system).readJournalFor[RedisReadJournal](RedisReadJournal.Identifier) 
  private val source: Source[EventEnvelope2, NotUsed] = readJournal.eventsByTag(tag, offset) 
  private def eventToRequestPair(event: Any): (HttpRequest, Offset) = ... 
  private def commitLatestOffset(result: Seq[(Try[HttpResponse], Offset)]): Future[Unit] = ... 

  source
    .map(e => eventToRequestPair(e.event))
    .via(pool)
    .groupedWithin(10, 60.seconds)
    .mapAsync(1)(commitLatestOffset)
    .runWith(Sink.ignore) 

}

Summary service

Let’s make integration easier and bring additional value to our clients by implementing a message summary microservice. This microservice can combine the output of the vision microservices—and using our knowledge of the vision processes—produce useful high-level summaries of multiple ingested messages.

The summary microservice combines multiple messages to produce one, but the input messages arrive at arbitrary points in time (potentially seconds apart) and in arbitrary order: the summary microservice has to maintain state that allows it to recover this state in case of failure. Because the summary microservice need to maintain their state over a number of seconds, we can no longer simply leave the offsets of the incoming messages unconfirmed in Kafka (only confirming when the summarisation is done and we’ve sent a response). To do this, we would have to significantly increase the confirmation timeout, which would result in larger batches of delivered messages, and it would delay our ability to detect failures. Remember the smart endpoints, dumb pipes tenet of microservices: we should not abuse Kafka to avoid having to implement smart endpoints. This means that each grouping microservice has to maintain its own persistent state that must enable it to replay the messages it has failed to process. Figure 2-10 reminds us where the summary service fits.

Figure 2-10. Summary service

Notice that the data store belongs to the summary microservice; the data in that store are not shared with any other microservice in the system. All that we have to resolve is exactly what data the service needs to persist to be able to recover from failures, and to maintain our message delivery guarantees. Let’s imagine we only process messages for one transaction, and suppose that we need to consume 3 incoming messages before we can produce one summary response. Figure 2-11 shows that the microservice consumes one message at a time from Kafka.

Figure 2-11. Auto partitioning with manual offsets

When Kafka assigns the topic and partition to the listener in the summary microservice, the microservice loads the offsets where it wishes to start consuming messages from a database (0x97),

Kafka then starts the subscription from the indicated offset,

the microservice consumes a batch with one message starting at offset 0x98; it immediately confirms the offset 0x98: this informs Kafka that the consumer thread is healthy, and it delivers the next batch of messages to it,

the microservice consumes a batch with one message starting at offset 0x99 and confirms the offset,

the same flow happens for batch starting at 0x9a, but the message—being the third number in the count of messages⁸—allows us to produce the summary message,

the microservice delivers the summary message the output topic, and it receives a confirmation of successful delivery,

the microservice writes the offset 0x9a to its offset database.

This flow looks just like the event-sourcing we described in “Event-sourcing”. We are using Kafka as the event journal, so all that the summary microservice has to do is to remember is the offset that allows it to recover its state in case of failure. In order for the summary microservice to be as resilient as possible, it should tolerate as many failures in its dependencies as possible. Clearly, it cannot tolerate Kafka becoming inaccessible, but can it continue working or can it start if its offsets database is inaccessible? The answer dependes on how much re-processing the summary microservice is prepared to do if the offsets database is not accessible. If it is ready to re-process all messages, it can completely ignore failures in writes and on failures of reads, it simply assumes zero offset. This sounds attractive, but if this microservice started processing messages from offset 0, it could generate load that could cause downstream microservices to fail; bringing a ripple of failures caused by extreme spike in messages. A better approach is to tolerate failures in offset writes for a certain period of time (during which we expect the offsets database to recover).

The summary state machine

The diagram in Figure 2-12 shows the in-flight states the transaction goes through; this will help us to implement the core of its logic.

Figure 2-12. States in the summary service

Example 2-19 shows the summarisation code, with the core state machine in the Summary trait and the Summary.Incomplete and Summary.Complete subtypes. We then define the Summaries case class that holds a collection of individual Summary instances, and applies the incoming Kafka messages (in the ConsumerRecord collection) to the Summary instances it holds, returning the new version of itself together with the completed outcomes and offsets of the topic partitions that can be saved to the microservice’s database when the summary messages are successfully placed on the outgoing topic.

Example 2-19. Summary code

sealed trait Summary {
  def next(envelope: Envelope): Summary
}

object Summary {
  case class Incomplete private(events: List[Envelope]) extends Summary {
    override def next(envelope: Envelope): Summary = ...
  }

  case class Complete(outcome: Outcome) extends Summary {
    override def next(envelope: Envelope): Summary = this
  }
}

case class Summaries() {
  import Summaries._

  def appending(consumerRecords: List[ConsumerRecord[String, Envelope]]):
    (Summaries, Map[String, Outcome], Offsets) = ...

The state machine in code in Example 2-19 implements the logic of the summary service, but we’re still missing the machinery that implements the event-sourcing service.

class SummariesActor(consumerConf: KafkaConsumer.Conf[String, Envelope],
                     consumerActorConf: KafkaConsumerActor.Conf,
                     producerConf: KafkaProducer.Conf[String, Envelope],
                     redisClientPool: RedisClientPool,
                     topicNames: List[String]) extends Actor {

  private[this] val kafkaConsumerActor = context.actorOf(...) 
  private[this] val kafkaProducer = KafkaProducer(producerConf)
  private[this] var summaries: Summaries = Summaries.empty

  import scala.concurrent.duration._

  override def supervisorStrategy: SupervisorStrategy = 
    OneForOneStrategy(maxNrOfRetries = 3, withinTimeRange = 3.seconds) {
      case _ => SupervisorStrategy.Restart
    }

  override def preStart(): Unit = { 
    kafkaConsumerActor ! Subscribe.AutoPartitionWithManualOffset(topicNames)
  }

  override def receive: Receive = { 
    case AssignedListener(partitions: List[TopicPartition]) => 
      val offsetsMap = redisClientPool.withClient { ... }
      sender() ! Offsets(offsetsMap) 

    case SummariesActor.extractor(consumerRecords) => 
      val (ns, outcomes, offsets) = summaries.appending(consumerRecords.recordsList) 
      summaries = ns
      kafkaConsumerActor ! KafkaConsumerActor.Confirm(consumerRecords.offsets) 

      if (outcomes.nonEmpty) { 
        val sent = outcomes.map { case (transactionId, outcome) =>
          val out = Envelope(...)
          kafkaProducer.send(KafkaProducerRecord("summary-1", transactionId, out)) 
        } 
        import context.dispatcher
        Future.sequence(sent).onComplete { 
          case Success(recordMetadata) => persistOffsets(offsets) 
          case Failure(ex) => self ! Kill 
        }
      }
  }

  private def persistOffsets(offsets: Offsets): Unit = { 
    redisClientPool.withClient { ... }
  }

}

Tooling

Each member of the development team can clone the project and build any of the microservices. However, it would be difficult for one engineer to build every microservice that the system needs, and to configure its dependencies; though it is something that the engineer might need to work on one of the microservices. The situation becomes even more complex when the engineer needs to run large-scale testing and training, which is particularly applicable to computer vision & machine learning.

Machine learning training and the appropriate testing is taking us away from entirely deterministic world, but that is nowhere near the unpredictability of chaos engineering. In chaos engineering, we construct tests that verify the system’s behavior while we inject faults into the running system. The faults can be as obvious as complete node failures and network partitions; more subtle faults triggered by garbage-collection pauses and increased I/O latencies and packet drops; all the way to very subtle faults such as successful system calls that nevertheless yield the wrong result—think socket or file read operations that indicate successful result, but that fill the userspace buffers with the wrong bytes; or concurrency problems exposed by the [hardware] power and performance management running different cores of modern CPUs at different speeds.

To have confidence in the stability of a distributed microservices system, we must verify that the system produces expected results under load and in presence of chaos.

FROM ubuntu:16.04

ENV CUDA_VERSION 8.0
ENV CUDA_PKG_VERSION 8-0=8.0.44-1

RUN apt-get update && apt-get install -y --no-install-recommends 
        cuda-core-$CUDA_PKG_VERSION 
        ...
        git 
        g++ 
        cmake 
        && 
    ln -s cuda-$CUDA_VERSION /usr/local/cuda && 
    rm -rf /var/lib/apt/lists/*
...

RUN curl -L -o opencv3.zip https://github.com/opencv/opencv/archive/3.2.0.zip && 
    unzip opencv3.zip && 
    ...
    cmake .. && 
    make -j8 install

RUN echo "/usr/local/cuda/lib64" >> /etc/ld.so.conf.d/cuda.conf && 
    ldconfig

FROM oreilly/rac-native-devel:latest

RUN apt-get update && apt-get remove -y --no-install-recommends 
        git 
        gcc 
        g++ 
        cmake 
        && 
      ln -s cuda-$CUDA_VERSION /usr/local/cuda && 
      rm -rf /var/lib/apt/lists/*

...

~/ips/faceextract $ racc build
-- The C compiler identification is GNU 5.4.0
-- The CXX compiler identification is GNU 5.4.0
-- Check for working C compiler: /usr/bin/cc
-- Check for working C compiler: /usr/bin/cc -- works
...
[ 93%] Building CXX object rapidcheck-build/.../detail/Recipe.cpp.o
[ 94%] Building CXX object rapidcheck-build/.../detail/ScaleInteger.cpp.o
[ 96%] Linking CXX executable main
[ 96%] Built target main
[ 98%] Linking CXX static library librapidcheck.a
[ 98%] Built target rapidcheck
Scanning dependencies of target main-test
[100%] Linking CXX executable main-test
[100%] Built target main-test
Test project /var/src/module/target
    Start 1: all
1/2 Test #1: all ..............................   Passed    0.01 sec
    Start 2: all
2/2 Test #2: all ..............................   Passed    0.01 sec

100% tests passed, 0 tests failed out of 2

Total Test time (real) =   0.12 sec