Forem: Usama Ashraf

Using Events In Node.js The Right Way

Usama Ashraf — Sat, 03 Nov 2018 18:06:55 +0000

Before event-driven programming became popular, the standard way to communicate between different parts of an application was pretty straightforward: a component that wanted to send out a message to another one explicitly invoked a method on that component. But event-driven code is written to react rather than be called.

The Benefits Of Eventing

This approach causes our components to be much more decoupled. Basically, as we continue to write an application we’ll identify events along the way, fire them at the right time and attach one or more event listeners to each one. Extending functionality becomes much easier since we can just add on more listeners to a particular event without tampering with the existing listeners or the part of the application where the event was fired from. What we’re talking about is essentially the Observer pattern.

^{Source: https://www.dofactory.com/javascript/observer-design-pattern}

Designing An Event-Driven Architecture

Identifying events is pretty important since we don’t want to end up having to remove/replace existing events from the system, which might force us to delete/modify any number of listeners that were attached to the event. The general principle I use is to consider firing an event only when a unit of business logic finishes execution.
So say you want to send out a bunch of different emails after a user’s registration. Now, the registration process itself might involve many complicated steps, queries etc. But from a business point of view it is one step. And each of the emails to be sent out are individual steps as well. So it would make sense to fire an event as soon as registration finishes and have multiple listeners attached to it, each of which is responsible for sending out one type of email.

Node’s asynchronous, event-driven architecture has certain kinds of objects called “emitters” that emit named events which cause functions called "listeners" to be invoked. All objects that emit events are instances of the EventEmitter class. Using it we can create our own events.

An Example

Let’s use the built-in events module (which I encourage you to check out in detail) to gain access to EventEmitter.

// my_emitter.js

const EventEmitter = require('events');

const myEmitter = new EventEmitter();

module.exports = myEmitter;

This is the part of the application where our server receives an HTTP request, saves a new user and emits an event accordingly:

// registration_handler.js

const myEmitter = require('./my_emitter');

// Perform the registration steps

// Pass the new user object as the message passed through by this event.
myEmitter.emit('user-registered', user);

And a separate module where we attach a listener:

// listener.js

const myEmitter = require('./my_emitter');

myEmitter.on('user-registered', (user) => {
  // Send an email or whatever.
});

It’s a good practice to separate policy from implementation. In this case policy means which listeners are subscribed to which events and implementation means the listeners themselves.

// subscriptions.js

const myEmitter = require('./my_emitter');
const sendEmailOnRegistration = require('./send_email_on_registration');
const someOtherListener = require('./some_other_listener');


myEmitter.on('user-registered', sendEmailOnRegistration);
myEmitter.on('user-registered', someOtherListener);

// send_email_on_registration.js

module.exports = (user) => {
  // Send a welcome email or whatever.
}

This separation allows for the listener to become re-usable too i.e. it can be attached to other events that send out the same message (a user object). It’s also important to mention that when multiple listeners are attached to a single event, they will be executed synchronously and in the order that they were attached. Hence someOtherListener will run after sendEmailOnRegistration finishes execution.
However if you want your listeners to run asynchronously you can simply wrap their implementations with setImmediate like this:

// send_email_on_registration.js

module.exports = (user) => {
  setImmediate(() => {
    // Send a welcome email or whatever.
  });
}

Keep Your Listeners Clean

Stick to the Single Responsibility Principle when writing listeners: one listener should do one thing only and do it well. Avoid, for instance, writing too many conditionals within a listener that decide what to do depending on the data (message) that was transmitted by the event. It would be much more appropriate to use different events in that case:

// registration_handler.js

const myEmitter = require('./my_emitter');

// Perform the registration steps

// The application should react differently if the new user has been activated instantly.
if (user.activated) {
  myEmitter.emit('user-registered:activated', user);

} else {
  myEmitter.emit('user-registered', user);
}

// subscriptions.js

const myEmitter = require('./my_emitter');
const sendEmailOnRegistration = require('./send_email_on_registration');
const someOtherListener = require('./some_other_listener');
const doSomethingEntirelyDifferent = require('./do_something_entirely_different');


myEmitter.on('user-registered', sendEmailOnRegistration);
myEmitter.on('user-registered', someOtherListener);

myEmitter.on('user-registered:activated', doSomethingEntirelyDifferent);

Detaching Listeners Explicitly When Necessary

In the previous example our listeners were totally independent functions. But in cases where a listener is associated with an object (it’s a method), it has to be manually detached from the events it had subscribed to. Otherwise, the object will never be garbage-collected since a part of the object (the listener) will continue to be referenced by an external object (the emitter). Thus the possibility of a memory-leak.

For example if we’re building a chat application and we want that the responsibility for showing a notification when a new message arrives in a chat room that a user has connected to should lie within that user object itself, we might do this:

// chat_user.js

class ChatUser {

  displayNewMessageNotification(newMessage) {
    // Push an alert message or something.
  }

  // `chatroom` is an instance of EventEmitter.
  connectToChatroom(chatroom) {
    chatroom.on('message-received', this.displayNewMessageNotification);
  }

  disconnectFromChatroom(chatroom) {
    chatroom.removeListener('message-received', this.displayNewMessageNotification);
  }
}

When the user closes his/her tab or loses their internet connection for a while, naturally, we might want to fire a callback on the server-side that notifies the other users that one of them just went offline. At this point of course it doesn’t make any sense for displayNewMessageNotification to be invoked for the offline user, but it will continue to be called on new messages unless we remove it explicitly. If we don’t, aside from the unnecessary call, the user object will also stay in memory indefinitely. So be sure to call disconnectFromChatroom in your server-side callback that executes whenever a user goes offline.

Beware

The loose coupling in event-driven architectures can also lead to increased complexity if we’re not careful. It can be difficult to keep track of dependencies in our system i.e. which listeners end up executing on which events. Our application will become especially prone to this problem if we start emitting events from within listeners, possibly triggering chains of unexpected events.

Serialization in Node REST APIs

Usama Ashraf — Thu, 27 Sep 2018 03:00:17 +0000

One of the things I learned a while back was that database columns should not be mapped directly to the JSON responses an API serves (columns can change, backward compatibility etc).
There should be some layer of separation between the logic of forming the response and fetching/querying it.
In Rails we have Active Model Serializers and the new fast_jsonapi gem from Netflix. Is there a widely-used analogous package for Node or some best practices that large-scale organizations using Node (like Ebay, Paypal, Netflix etc) employ?
Assuming we're talking about an API built up on Express.

Juggling Multiple Languages Simultaneously

Usama Ashraf — Fri, 21 Sep 2018 16:54:29 +0000

I've been coding Javascript, Ruby and Go almost everyday for the past few months and I'm just wondering if in the long run it hurts my chances of learning expertise in either of them. I clearly feel the "switches" I have to make daily.
Plus, is the "jack of all trades, master of none" problem a real and significant one?

Playing With Apache Storm On Docker - Like A Boss

Usama Ashraf — Tue, 15 May 2018 09:49:46 +0000

This article is not the ultimate guide to Storm nor is it meant to be. Storm's pretty huge, and just one long-read probably can't do it justice anyways. Of course, any additions, feedback or constructive criticism will be greatly appreciated.
OK, now that that's out of the way, let's see what we'll be covering:

The necessity of Storm, the 'why' of it, what it is and what it isn't
A bird's eye view of how it works.
What a Storm topology roughly looks like in code (Java)
Setting up and playing with a production-worthy Storm cluster on Docker.
A few words on message processing reliability.

I'm also assuming that you're at least somewhat familiar with Docker and containerization.

Continuous streams of data are ubiquitous and becoming even more so with the increasing number of IoT devices being used. Of course this data is stored, processed and analyzed to provide predictive, actionable results. But petabytes take long to analyze, even with Hadoop (as good as MapReduce may be) or Spark (a remedy to the limitations of MapReduce). Secondly, very often we don't need to deduce patterns over long periods of time. Of the petabytes of incoming data collected over months, at any given moment, we might not need to take into account all of it, just a real-time snapshot. Perhaps we don't need to know the longest trending hashtag over five years, but just the one right now. This is what Storm is built for, to accept tons of data coming in extremely fast, possibly from various sources, analyze it and publish the real-time updates to a UI or some other place without storing any itself.

How It Works

The architecture of Storm can be compared to a network of roads connecting a set of checkpoints. Traffic begins at a certain checkpoint (called a spout) and passes through other checkpoints (called bolts). The traffic is of course the stream of data that is retrieved by the spout (from a data source, a public API for example) and routed to various bolts where the data is filtered, sanitized, aggregated, analyzed, sent to a UI for people to view or any other target. The network of spouts and bolts is called a topology, and the data flows in the form of tuples (list of values that may have different types).

^{Source: https://dzone.com/articles/apache-storm-architecture}

One important thing to talk about is the direction of the data traffic. Conventionally, we would have one or multiple spouts reading the data from an API, a Kafka topic or some other queuing system. The data would then flow one-way to one or multiple bolts which may forward it to other bolts and so on. Bolts may publish the analyzed data to a UI or to another bolt. But the traffic is almost always unidirectional, like a DAG. Although it is certainly possible to make cycles, we're unlikely to need such a convoluted topology.

Installing a Storm release involves a number of steps, which you're free to follow on your machine. But later on I'll be using Docker containers for a Storm cluster deployment and the images will take care of setting up everything we need.

Some Code

While Storm does offer support for other languages, most topologies are written in Java since it's the most efficient option we have.

A very basic spout, that just emits random digits, may look like this:

public class RandomDigitSpout extends BaseRichSpout 
{
  // To output tuples from spout to the next stage bolt
  SpoutOutputCollector collector;  

  public void nextTuple() 
  {
    int randomDigit = ThreadLocalRandom.current().nextInt(0, 10);

    // Emit the digit to the next stage bolt
    collector.emit(new Values(randomDigit));
  }

  public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer)
  {
    // Tell Storm the schema of the output tuple for this spout.
    // It consists of a single column called 'random-digit'.
    outputFieldsDeclarer.declare(new Fields("random-digit"));
  }
}

And a simple bolt that takes in the stream of random and just emits the even ones:

public class EvenDigitBolt extends BaseRichBolt 
{
  // To output tuples from this bolt to the next bolt.
  OutputCollector collector;

  public void execute(Tuple tuple) 
  {
    // Get the 1st column 'random-digit' from the tuple
    int randomDigit = tuple.getInt(0);

    if (randomDigit % 2 == 0) {
      collector.emit(new Values(randomDigit));
    }
  }

  public void declareOutputFields(OutputFieldsDeclarer declarer) 
  {
    // Tell Storm the schema of the output tuple for this bolt.
    // It consists of a single column called 'even-digit'
    declarer.declare(new Fields("even-digit"));
  }
}

Another simple bolt that'll receive the filtered stream from EvenDigitBolt, and just multiply each even digit by 10 and emit it forward:

public class MultiplyByTenBolt extends BaseRichBolt 
{
  OutputCollector collector;

  public void execute(Tuple tuple) 
  {
    // Get 'even-digit' from the tuple.
    int evenDigit = tuple.getInt(0);

    collector.emit(new Values(evenDigit * 10));
  }

  public void declareOutputFields(OutputFieldsDeclarer declarer) 
  {
    declarer.declare(new Fields("even-digit-multiplied-by-ten"));
  }
}

Putting them together to form our topology:

package packagename
// ...

public class OurSimpleTopology { 

  public static void main(String[] args) throws Exception
  {
    // Create the topology
    TopologyBuilder builder = new TopologyBuilder();

    // Attach the random digit spout to the topology.
    // Use just 1 thread for the spout.
    builder.setSpout("random-digit-spout", new RandomDigitSpout());

    // Connect the even digit bolt to our spout. 
    // The bolt will use 2 threads and the digits will be randomly
    // shuffled/distributed among the 2 threads.
    // The third parameter is formally called the parallelism hint.
    builder.setBolt("even-digit-bolt", new EvenDigitBolt(), 2)
           .shuffleGrouping("random-digit-spout");

    // Connect the multiply-by-10 bolt to our even digit bolt.
    // This bolt will use 4 threads, among which data from the
    // even digit bolt will be shuffled/distributed randomly.
    builder.setBolt("multiplied-by-ten-bolt", new MultiplyByTenBolt(), 4)
           .shuffleGrouping("even-digit-bolt");

    // Create a configuration object.
    Config conf = new Config();

    // The number of independent JVM processes this topology will use.
    conf.setNumWorkers(2);

    // Submit our topology with the configuration.
    StormSubmitter.submitTopology("our-simple-topology", conf, builder.createTopology());
  }
}

Parallelism In Storm Topologies

Fully understanding parallelism in Storm can be daunting, at least in my experience. A topology requires at least one process to operate on (obviously). Within this process we can parallelize the execution of our spouts and bolts using threads. In our example, RandomDigitSpout will launch just one thread, and the data spewed from that thread will be distributed among 2 threads of the EvenDigitBolt. But the way this distribution happens, referred to as the stream grouping, can be important. For example you may have a stream of temperature recordings from two cities, where the tuples emitted by the spout look like this:

// City name, temperature, time of recording

("Atlanta",       94, "2018-05-11 23:14")
("New York City", 75, "2018-05-11 23:15")
("New York City", 76, "2018-05-11 23:16")
("Atlanta",       96, "2018-05-11 23:15")
("New York City", 77, "2018-05-11 23:17")
("Atlanta",       95, "2018-05-11 23:16")
("New York City", 76, "2018-05-11 23:18")

Suppose we're attaching just one bolt whose job is to calculate the changing average temperature of each city. If we can reasonably expect that in any given time interval we'll get roughly an equal number of tuples from both the cities, it would make sense to dedicate 2 threads to our bolt and send the data for Atlanta to one of them and New York to the other. A fields grouping would serve our purpose, which partitions data among the threads by the value of the field specified in the grouping:

// The tuples with the same city name will go to the same thread.
builder.setBolt("avg-temp-bolt", new AvgTempBolt(), 2)
       .fieldsGrouping("temp-spout", new Fields("city_name"));

And of course there are other types of groupings as well. For most cases, though, the grouping probably won't matter much and you can just shuffle the data and throw it among the bolt threads randomly (shuffle grouping).
Now there's another important component to this: the number of worker processes that our topology will run on. The total number of threads that we specified will then be equally divided among the worker processes. So in our example random digit topology we had 1 spout thread, 2 even-digit bolt threads and 4 multiply-by-ten bolt threads (7 total). Each of the 2 worker processes would be responsible for running 2 multiply-by-ten bolt threads, 1 even-digit bolt and one of the processes will run the 1 spout thread.

Of course, the 2 worker processes will have their main threads, which in turn will launch the spout and bolt threads. So all in all we'll have 9 threads. These are collectively called executors.

It's important to realize that if you set a spout's parallelism hint > 1 (i.e. multiple executors), you can end up emitting the same data several times. Say, the spout reads from the public Twitter stream API and uses two executors. That means that the bolts receiving the data from the spout will get the same tweet twice. It is only after the spout emits the tuples that data parallelism comes into play, i.e. the tuples get divided among the bolts according to the specified stream grouping.

Running multiple workers on a single node would be fairly pointless. Later, however, we'll use a proper, distributed, multi-node cluster and see how workers are divided on different nodes.

Building Our Topology

Here's the directory structure I suggest:

yourproject/
            pom.xml  
            src/
                jvm/
                    packagename/
                           RandomDigitSpout.java
                           EvenDigitBolt.java
                           MultiplyByTenBolt.java
                           OurSimpleTopology.java

Maven is commonly used for building Storm topologies, and it requires a pom.xml file (The POM) that defines various configuration details, project dependencies etc. Getting into the nitty-gritty of the POM will probably be an overkill here.

First, we'll run mvn clean inside yourproject to clear any compiled files we may have, making sure to compile each module from scratch.
And then mvn package to compile our code and package it in an executable JAR file, inside a newly created target folder. This might take quite a few minutes the first time, especially if your topology has many dependencies.
To submit our topology: storm jar target/packagename-{version number}.jar packagename.OurSimpleTopology

Hopefully, by now the gap between concept and code in Storm has been somewhat bridged. However, no serious Storm deployment will be a single topology instance running on one server.

What A Storm Cluster Looks Like

To take full advantage of Storm's scalability and fault-tolerance, any production-grade topology would be submitted to a cluster of machines.

Storm distributions are installed on the master node (Nimbus) and all the slave nodes (Supervisors).
The master node runs the Storm Nimbus daemon and the Storm UI. The slave nodes run the Storm Supervisor daemons. A Zookeeper daemon on a separate node is used for coordination among the master node and the slave nodes. Zookeeper, by the way, is only used for cluster management and never any kind of message passing. It's not like spouts and bolts are sending data to each other through it or anything like that. The Nimbus daemon finds available Supervisors via ZooKeeper, to which the Supervisor daemons register themselves. And other managerial tasks, some of which will become clear shortly.

^{The Storm UI is a web interface used to manage the state of our cluster. We'll get to this later.}

Our topology is submitted to the Nimbus daemon on the master node and then distributed among the worker processes running on the slave/supervisor nodes. Because of Zookeeper, it doesn't matter how many slave/supervisor nodes you run initially, as you can always seamlessly add more and Storm will automatically integrate them into the cluster.

Whenever we start a Supervisor it allocates a certain number of worker processes (that we can configure) which can then be used by the submitted topology. So in the image above there are a total of 5 allocated workers. Remember this line:
conf.setNumWorkers(5)
This means that the topology will try to use a total of 5 workers. And since our two Supervisor nodes have a total of 5 allocated workers: each of the 5 allocated worker processes will run one instance of the topology. If we had done:
conf.setNumWorkers(4)
then one worker process would have remained idle/unused. If the number of specified workers was 6 and the total allocated workers were 5, then because of the limitation only 5 actual topology workers would've been functional.

Before we set this all up using Docker, a few important things to keep in mind regarding fault-tolerance:

If any worker on any slave node dies, the Supervisor daemon will have it restarted. If restarting repeatedly fails, the worker will be reassigned to another machine.
If an entire slave node dies, its share of the work will be given to another supervisor/slave node.
If the Nimbus goes down, the workers will remain unaffected. However, until the Nimbus is restored workers won't be reassigned to other slave nodes if, say, their node crashes.
The Nimbus & Supervisors are themselves stateless, but with Zookeeper, some state information is stored so that things can begin where they were left off if a node crashes or a daemon dies unexpectedly.
Nimbus, Supervisor & Zookeeper daemons are all fail-fast. This means that they themselves are not very tolerant to unexpected errors, and will shut down if they encounter one. For this reason they have to be run under supervision using a watchdog program that monitors them constantly and restarts them automatically if they ever crash. Supervisord is probably the most popular option for this (not to be confused with the Storm Supervisor daemon).

Note: In most Storm clusters, the Nimbus itself is never deployed as a single instance but as a cluster. If this fault-tolerance is not incorporated and our sole Nimbus goes down, we'll lose the ability to submit new topologies, gracefully kill running topologies, reassign work to other Supervisor nodes if one crashes etc. For simplicity, our illustrative cluster will use a single instance. Similarly, the Zookeeper is very often deployed as a cluster but we'll use just one.

Dockerizing The Cluster

Launching individual containers and all that goes along with them can be cumbersome, so I prefer to use Docker Compose. We'll be going with one Zookeeper node, one Nimbus node and one Supervisor node initially. They'll be defined as Compose services, all corresponding to one container each at the beginning. Later on, I'll use Compose scaling to add another Supervisor node (container). Here's our entire code & the project structure:

zookeeper/
         Dockerfile
storm-nimbus/
         Dockerfile
         storm.yaml
         code/
             pom.xml
             src/
                 jvm/
                     coincident_hashtags/
                                ExclamationTopology.java  
storm-supervisor/
         Dockerfile
         storm.yaml
docker-compose.yml

And our docker-compose.yml:

version: '3.2'

services:
    zookeeper:
        build: ./zookeeper
        # Keep it running.  
        tty: true

    storm-nimbus:
        build: ./storm-nimbus
        # Run this service after 'zookeeper' and make 'zookeeper' reference.
        links:
            - zookeeper
        tty: true
        # Map port 8080 of the host machine to 8080 of the container.
        # To access the Storm UI from our host machine.
        ports:
            - 8080:8080
        volumes:
            - './storm-nimbus:/theproject'

    storm-supervisor:
        build: ./storm-supervisor
        links:
            - zookeeper
            - storm-nimbus
        tty: true

# Host volume used to store our code on the master node (Nimbus).
volumes:
    storm-nimbus:

Feel free to explore the Dockerfiles. They basically just install the dependencies (Java 8, Storm, Maven, Zookeeper etc) on the relevant containers.
The storm.yaml files override certain default configurations for the Storm installations. The line ADD storm.yaml /conf inside the Nimbus and Supervisor Dockerfiles puts them inside the containers where Storm can read them.
storm-nimbus/storm.yaml:

# The Nimbus needs to know where the Zookeeper is. This specifies the list of the
# hosts in the Zookeeper cluster. We're using just one node, of course.
# 'zookeeper' is the Docker Compose network reference.
storm.zookeeper.servers:
  - "zookeeper"

storm-supervisor/storm.yaml:

# Telling the Supervisor where the Zookeeper is.
storm.zookeeper.servers:
  - "zookeeper"

# The worker nodes need to know which machine(s) are the candidate of master
# in order to download the topology jars.
nimbus.seeds : ["storm-nimbus"]

# For each Supervisor, we configure how many workers run on that machine. 
# Each worker uses a single port for receiving messages, and this setting 
# defines which ports are open for use. We define four ports here, so Storm will 
# allocate up to four workers to run on this node.
supervisor.slots.ports:
    - 6700
    - 6701
    - 6702
    - 6703

These options are adequate for our cluster. The more curious can check out all the default configurations here.

Run docker-compose up at the project root.

After all the images have been built and all the service started, open a new terminal, type docker ps and you'll see something like this:

Starting The Nimbus

Let's SSH into the Nimbus container using its name:
docker exec -it coincidenthashtagswithapachestorm_storm-nimbus_1 bash
and then start the Nimbus daemon:
storm nimbus

Starting The Storm UI

Similarly, open another terminal, SSH into the Nimbus again and launch the UI using storm ui:

Go to localhost:8080 on your browser and you'll see a nice overview of our cluster:

The Free slots in the Cluster Summary indicate how many total workers (on all Supervisor nodes) are available & waiting for a topology to consume them. Used Slots indicate how many of the total are currently busy with a topology. Since we haven't launched any Supervisors yet, they're both zero. We'll get to Executors and Tasks later. Also, as we can see, no topologies have been submitted yet.

Starting A Supervisor Node

SSH into the one Supervisor container and launch the Supervisor daemon:
docker exec -it coincidenthashtagswithapachestorm_storm-supervisor_1 bash
storm supervisor

Now let's go refresh our UI:

Note: Any changes in our cluster may take a few seconds to reflect on the UI.

We have a new running Supervisor which comes with four allocated workers. These four workers are the result of specifying four ports in our storm.yaml for the Supervisor node. Of course, they're all free (four Free slots). Let's submit a topology to the Nimbus and put'em to work.

Submitting A Topology To The Nimbus

SSH into the Nimbus on a new terminal. I've written the Dockerfile so that we land on our working (landing) directory /theproject. Inside this is code, where our topology resides. Our topology is pretty simple. It uses a spout that generates random words and a bolt that just appends three exclamation marks (!!!) to the words. Two of these bolts are added back-to-back and so at the end of the stream we'll get words with six exclamation marks. It also specifies that it needs three workers (conf.setNumWorkers(3)).

public static void main(String[] args) throws Exception
{
    TopologyBuilder builder = new TopologyBuilder();

    builder.setSpout("word", new TestWordSpout(), 10);

    builder.setBolt("exclaim1", new ExclamationBolt(), 3).shuffleGrouping("word");

    builder.setBolt("exclaim2", new ExclamationBolt(), 2).shuffleGrouping("exclaim1");

    Config conf = new Config();

    // Turn on  debugging mode
    conf.setDebug(true);

    conf.setNumWorkers(3);

    StormSubmitter.submitTopology("exclamation-topology", conf, builder.createTopology());
}

cd code
mvn clean
mvn package
storm jar target/coincident-hashtags-1.2.1.jar coincident_hashtags.ExclamationTopology

After the topology has been submitted successfully, refresh the UI:

As soon as we submitted the topology, the Zookeeper was notified. The Zookeeper in turn notified the Supervisor to download the code from the Nimbus. We now see our topology along with its three occupied workers, leaving just one free.
And 10 word spout threads + 3 exclaim1 bolt threads + 2 exclaim bolt threads + the 3 main threads from the workers = total of 18 executors. And you might've noticed something new: tasks.

WTF Are Tasks

Another concept in Storm's parallelism. But don't sweat it, a task is just an instance of a spout or bolt that an executor uses; what actually does the processing. By default the number of tasks is equal to the number of executors. In rare cases you might need each executor to instantiate more tasks.

// Each of the two executors (threads) of this bolt will instantiate
// two objects of this bolt (total 4 bolt objects/tasks).
builder.setBolt("even-digit-bolt", new EvenDigitBolt(), 2)
       .setNumTasks(4) 
       .shuffleGrouping("random-digit-spout");

This is a shortcoming on my part, but I can't think of a good use case where we'd need multiple tasks per executor. May be if we were adding some parallelism ourselves, like spawning a new thread within the bolt to handle a long running task, then the main executor thread won't block and will be able to continue processing using the other bolt. However this can make our topology hard to understand. If any one knows of scenarios where the performance gain from multiple tasks outweighs the added complexity, please post a comment.

Anyways, returning from that slight detour, let's see an overview of our topology. Click on the name under Topology Summary and scroll down to Worker Resources:

We can clearly see the division of our executors (threads) among the 3 workers. And of course all the 3 workers are on the same, single Supervisor node we're running.

Now, let's say scale out!

Add Another Supervisor

From the project root, let's add another Supervisor node/container
docker-compose scale storm-supervisor=2

SSH into the new container:
docker exec -it coincidenthashtagswithapachestorm_storm-supervisor_2 bash

And fire up:
storm supervisor

If you refresh the UI you'll see that we've successfully added another Supervisor and four more workers (total of 8 workers/slots). To really take advantage of the new Supervisor, let's increase the topology's workers.

First kill the running one: storm kill exclamation-topology
Change this line to: conf.setNumWorkers(6)
Change the project version number in your pom.xml. Try using a proper scheme, like semantic versioning. I'll just stick with 1.2.1.
Rebuild the topology: mvn package
Resubmit it: storm jar target/coincident-hashtags-1.2.1.jar coincident_hashtags.ExclamationTopology

Reload the UI:

You can now see the new Supervisor and the 6 busy workers out of a total of 8 available ones. Also important to note is that the 6 busy ones have been equallly divided among the two Supervisors. Again, click the topology name and scroll down.

We see two unique Supervisor IDs, both running on different nodes, and all our executors pretty evenly divided among them. This is great. But Storm comes with another nifty way of doing so while the topology is running. Something called rebalancing. On the Nimbus we'd run:
storm rebalance exclamation-topology -n 6 (go from 3 to 6 workers)
Or to change the number of executors for a particular component:
storm rebalance exclamation-topology -e even-digit-bolt=3

Reliable Message Processing

One question we haven't tackled is about what happens if a bolt fails to process a tuple. Well, Storm provides us a mechanism using which the originating spout (specifically the task) can replay the failed tuple. This processing guarantee doesn't just happen by itself, it's a conscious design choice and does add latency.
Spouts send out tuples to bolts, which emit tuples derived from the input tuples to other bolts and so on. That one, original tuple spurs an entire tree of tuples. If any child tuple, so to speak, of the original one fails then any remedial steps (rollbacks etc) may well have to be taken at multiple bolts. That could get pretty hairy, and so what Storm does is that it allows the original tuple to be emitted again right from the source (the spout). Consequentially, any operations performed by bolts that are a function of the incoming tuples should be idempotent. A tuple is considered "fully processed" when every tuple in its tree has been processed, and every tuple has to be explicitly acknowledged by the bolts. However, that's not all. There's another thing to be done explicitly: maintain a link between the original tuple and its child tuples. Storm will then be able to trace the origin of the child tuples and thus be able to replay the original tuple. This is called anchoring. And this has been done in our exclamation bolt:

// ExclamationBolt

// 'tuple' is the original one received from the test word spout.
// It's been anchored to/with the tuple going out.
_collector.emit(tuple, new Values(exclamatedWord.toString()));

// Explicitly acknowledge that the tuple has been processed.
_collector.ack(tuple);

The ack call will result in the ack method on the spout being called, if it has been implemented. So, say, you're reading the tuple data from some queue and you can only take it off the queue if the tuple has been fully processed. Well, the ack method is where you'd do that. You can also emit out tuples without anchoring: _collector.emit(new Values(exclamatedWord.toString())) and forgo reliability.

A tuple can fail two ways:
i) A bolt dies and a tuple times out. Or it times out for some other reason. The timeout is 30 seconds by default and can be changed using config.put(Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS, 60)
ii) The fail method is explicitly called on the tuple in a bolt: _collector.fail(tuple). You may do this in case of an exception.

In both these cases, the fail method on the spout will be called, if it is implemented. And if we want the tuple to be replayed, it would have to be done explicitly in the fail method by calling emit, just like in nextTuple(). When tracking tuples, every one has to be acked or failed. Otherwise, the topology will eventually run out of memory.
It's also important to know that you have to do all of this yourself when writing custom spouts and bolts. But the Storm core can help. For example, a bolt implementing BaseBasicBolt does acking automatically. Or built-in spouts for popular data sources like Kafka take care of queuing and replay logic after acknowledgment and failure.

Parting Shots

Designing a Storm topology or cluster is always about tweaking the various knobs we have and settling where the result seems optimal. There are a few things that'll help in this process, like using a configuration file to read parallelism hints, number of workers etc so you don't have to edit and recompile your code repeatedly. Define your bolts logically, one per indivisible task, and keep them light and efficient. Similarly, your spouts' nextTuple() methods should be optimized.
Use the Storm UI effectively. By default it doesn't show us the complete picture, only 5% of the total tuples emitted. To monitor all of them use config.setStatsSampleRate(1.0d). Keep an eye on the Acks and Latency values for individual bolts and topologies via the UI, that's what you want to look at when turning the knobs.

Microservices & RabbitMQ On Docker

Usama Ashraf — Wed, 25 Apr 2018 23:59:44 +0000

A microservices-based architecture involves decomposing your monolith app into multiple, totally independently deployable and scalable services. Beyond this base definition, what constitutes a microservice can be somewhat subjective, though there are several battle-tested practices adopted by giants like Netflix and Uber that should always be considered. And I'll discuss some of them. Ultimately, we want to divide our app into smaller apps, each of which is a system apart & deals with only one aspect of the whole app and does it really well. This decomposition is a very consequential step and can be done on the basis of subdomains, which have to be identified correctly. The smaller apps are more modular & manageable with well-defined boundaries, can be written using different languages/frameworks, fail in isolation so that the entire app doesn't go down (no SPOF). Take a Cinema ticketing example:

^{Source: https://codeburst.io/build-a-nodejs-cinema-api-gateway-and-deploying-it-to-docker-part-4-703c2b0dd269}

Let's break down this bird's eye view:

i) The user app can be a mobile client, SPA etc or any client consuming our backend services.

ii) It's considered a bad practice to ask our clients to communicate with each of our services separately, for reasons I tried to explain here. This is what the API gateways are for: to receive client requests, call our service(s), return a response. Thus the client only has to talk to one server, giving the illusion of a monolith. Multiple gateways may be used for different kinds of clients (mobile apps, tablets, browsers etc.). They can and should be responsible for limited functionality like merging/joining responses from services, authentication, ACLs. In large applications, which need to scale and move dynamically, gateways also need access to a Service Registry which holds the locations of our microservice instances, databases etc.

iii) Each service has its own storage. This is a key point and ensures loose coupling. Some queries will then need to join data that is owned by multiple services. To avoid this major performance hit, data may be replicated and sharded. This principle is not just tolerated in microservices, but encouraged.

iv) The REST calls made to our API gateway are passed to the services, which in turn talk to other services, return a result to the gateway which, perhaps, compiles it and responds with it to the client. Communication among services like this on one client request to the app should not happen. Otherwise we'll be sacrificing performance, on account of another HTTP round-trip, for the newly introduced modularity.

A single request should, ideally, only invoke one service to fetch the response. This means any synchronous requests between the services should be minimized, and that's not always possible; mechanisms like gRPC, Thrift or even simple HTTP (as in our example) are commonly employed when necessary. As you may have guessed, this implies that data will have to be replicated across our services. Say, the GET /catalog/<<cityId>> endpoint is also supposed to return the premieres at each cinema in the city at that time. With our new strategy, the premieres will have to be stored in the database for the Cinema Catalog service as well. Hence, point iii).

Communicating Asynchronously Between The Services

So, say, the premieres change as a result of some CRUD operation on the Movies service. To keep the data in sync, that update event will have to be emitted and applied to the Cinema Catalog service as well. Try to picture our microservices as a cluster of state machines where updates in states may have to be communicated across the cluster to achieve eventual consistency. Of course, we should never expect our end-users to have to wait longer for requests to finish and sacrifice their time for modularity to our benefit. Thereby, all of this communication has to be non-blocking. And that's where RabbitMQ comes in.

RabbitMQ is a very powerful message broker that implements the AMQP messaging protocol. Here's the abstract: first, you install a RabbitMQ server instance (broker) on a system. Then a publisher/producer program connects to this server and sends out a message. RabbitMQ queues that message and siphons it off to a single or multiple subscriber/consumer programs that are out there listening on the RabbitMQ server.

Before I get to the crux of this article, I want to explicitly declare that microservices are way more complex and we won't be covering critical topics like fault tolerance because of the intricacy of distributed systems, the full role of the API gateway, Service Discovery, data consistency patterns like Sagas, preventing service failure cascading using Circuit Breakers, health checks and architectural patterns like CQRS. Not to mention how to decide whether microservices will work for you or not.

How RabbitMQ Works

More specifically, the messages are published to an exchange inside the RabbitMQ broker. The exchange then distributes copies of that message to queues on the basis of certain developer-defined rules called bindings. This part of the messages' journey is called routing. And this indirection is of course what makes for the non-blocking message transmissions. Consumers listening on those queues that got the message will receive it. Pretty simple, right?

Not exactly. There are four different types of exchanges and each, along with the bindings, defines a routing algorithm. "Routing algorithm" means, essentially, how the messages are distributed among the queues. Going into the details about each type might be an overkill here, so I'll just expand on the one we'll be using: the topic exchange:

For an exchange to push a message onto a queue, that queue must be bound to the exchange. We can create multiple exchanges with unique names, explicitly. However, when you deploy RabbitMQ it comes with a default, nameless exchange. And every queue we create will be automatically bound to this exchange. To be descriptive, I'll be creating a named exchange manually and then bind a queue to it. This binding is defined by a binding key. The exact way a binding key works, again, depends on the type of the exchange. Here's how it works with a topic exchange:

A queue is bound to an exchange using a string pattern (the binding key)
The published message is delivered to the exchange along with a routing key
The exchange checks which queues match the routing key based on the binding key pattern defined before.

^{* can substitute for exactly one word. # can substitute for zero or more words.

^{Source: https://www.rabbitmq.com/tutorials/tutorial-five-python.html}}

Any message with a routing key "quick.orange.rabbit" will be delivered to both queues. However, messages with "lazy.brown.fox" will only reach Q2. Those with a routing key not matching any pattern will be lost.

For some perspective, let's just skim over two other exchange types:

Fanout exchange: Messages sent to this kind of exchange will be sent to ALL the queues bound to it. The routing key, if provided, will be completely ignored. This can be used, for example, for broadcasting global configuration updates across a distributed system.
Direct exchange (simplest): Sends the message to the queue whose binding key is exactly equal to the given routing key. If there are multiple consumers listening on the queue, then the messages will be load-balanced among them, hence, it is commonly used to distribute tasks between multiple workers in a round robin manner.

My illustration will be very simple: a Python Flask app with a single POST endpoint, which, when called, will purport to update a user's info, emit a message to the RabbitMQ broker (non-blocking of course) and return a 201. A separate Go service will be listening for the message from the broker and hence have the chance to update its data accordingly. All three will be hosted on separate containers.

Setting Up Our Containerized Microservices & Broker Using Docker Compose

Provisioning a bunch of containers and all that goes along with them can be a pain, so I always rely on Docker Compose.

Here's the entire code. We're declaring three services that will be used for the three containers. The two volumes are needed for putting our code inside the containers:

# docker-compose.yml

version: "3.2"
services:
    rabbitmq-server:
        build: ./rabbitmq-server

    python-service:
        build: ./python-service
        # 'rabbitmq-server' will be available as a network reference inside this service 
        # and this service will start only after the RabbitMQ service does.
        depends_on:
            - rabbitmq-server
        # Keep it running.  
        tty: true
        # Map port 3000 on the host machine to port 3000 of the container.
        # This will be used to receive HTTP requests made to the service.
        ports:
            - "3000:3000"
        volumes:
            - './python-service:/python-service'

    go-service:
        build: ./go-service
        depends_on:
            - rabbitmq-server
        tty: true
        volumes:
            - './go-service:/go-service'

# Host volumes used to store code.
volumes:
    python-service:
    go-service:

The Dockerfiles are pretty much the standard ones from Docker Hub, to which I've added:

/go-service working directory in the Go service container.
/python-service working directory in the Python service container.
Go's RabbitMQ client library called amqp
Python's RabbitMQ client Pika & Flask

Our Flask app has just one endpoint that receives a user_id and a full_name, which will be used to update the user's profile. A message indicating this update will then be sent to the RabbitMQ broker.

# main.py

from flask import Flask
from flask import request
from flask import jsonify
from services.user_event_handler import emit_user_profile_update

app = Flask(__name__)

@app.route('/users/<int:user_id>', methods=['POST'])
def update(user_id):
    new_name = request.form['full_name']

    # Update the user in the datastore using a local transaction...

    emit_user_profile_update(user_id, {'full_name': new_name})

    return jsonify({'full_name': new_name}), 201

The logic for emitting events to other services should always be separate from the rest of the app, so I've extracted it to a module. The exchange should be explicitly checked for and created by the publisher and consumer both, because we can't know (nor should we rely on) which service starts first. This is a good practice, which most RabbitMQ client libraries facilitate seamlessly:

# services/user_event_handler.py

import pika
import json

def emit_user_profile_update(user_id, new_data):
    # 'rabbitmq-server' is the network reference we have to the broker, 
    # thanks to Docker Compose.
    connection = pika.BlockingConnection(pika.ConnectionParameters(host='rabbitmq-server'))
    channel    = connection.channel()

    exchange_name = 'user_updates'
    routing_key   = 'user.profile.update'

    # This will create the exchange if it doesn't already exist.
    channel.exchange_declare(exchange=exchange_name, exchange_type='topic', durable=True)

    new_data['id'] = user_id

    channel.basic_publish(exchange=exchange_name,
                          routing_key=routing_key,
                          body=json.dumps(new_data),
                          # Delivery mode 2 makes the broker save the message to disk.
                          # This will ensure that the message be restored on reboot even  
                          # if RabbitMQ crashes before having forwarded the message.
                          properties=pika.BasicProperties(
                            delivery_mode = 2,
                        ))

    print("%r sent to exchange %r with data: %r" % (routing_key, exchange_name, new_data))
    connection.close()

Don't get confused by channel. A channel is just a virtual, lightweight connection within the TCP connection that is meant to prevent opening multiple, expensive TCP connections. Especially in multithreaded environments.

The durable parameter ensures that the exchange is persisted to the disk and can be restored if the broker crashes or goes offline for any reason. The publisher (Python service) creates an exchange named user_updates and sends it the user's updated data with user.profile.update as the routing key. This will be matched with a user.profile.* binding key, which our Go service will define:

// main.go

package main

import (
    "fmt"
    "log"
    "github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
    if err != nil {
        // Exit the program.
        panic(fmt.Sprintf("%s: %s", msg, err))
    }
}

func main() {
    // 'rabbitmq-server' is the network reference we have to the broker, 
    // thanks to Docker Compose.
    conn, err := amqp.Dial("amqp://guest:guest@rabbitmq-server:5672/")
    failOnError(err, "Error connecting to the broker")
    // Make sure we close the connection whenever the program is about to exit.
    defer conn.Close()

    ch, err := conn.Channel()
    failOnError(err, "Failed to open a channel")
    // Make sure we close the channel whenever the program is about to exit.
    defer ch.Close()

    exchangeName := "user_updates"
    bindingKey   := "user.profile.*"

    // Create the exchange if it doesn't already exist.
    err = ch.ExchangeDeclare(
            exchangeName,   // name
            "topic",        // type
            true,           // durable
            false,
            false,
            false,
            nil,
    )
    failOnError(err, "Error creating the exchange")

    // Create the queue if it doesn't already exist.
    // This does not need to be done in the publisher because the
    // queue is only relevant to the consumer, which subscribes to it.
    // Like the exchange, let's make it durable (saved to disk) too.
    q, err := ch.QueueDeclare(
            "",    // name - empty means a random, unique name will be assigned
            true,  // durable
            false, // delete when the last consumer unsubscribes
            false, 
            false, 
            nil,   
    )
    failOnError(err, "Error creating the queue")

    // Bind the queue to the exchange based on a string pattern (binding key).
    err = ch.QueueBind(
            q.Name,       // queue name
            bindingKey,   // binding key
            exchangeName, // exchange
            false,
            nil,
    )
    failOnError(err, "Error binding the queue")

    // Subscribe to the queue.
    msgs, err := ch.Consume(
            q.Name, // queue
            "",     // consumer id - empty means a random, unique id will be assigned
            false,  // auto acknowledgement of message delivery
            false,  
            false,  
            false,  
            nil,
    )
    failOnError(err, "Failed to register as a consumer")


    forever := make(chan bool)

    go func() {
        for d := range msgs {
            log.Printf("Received message: %s", d.Body)

            // Update the user's data on the service's 
            // associated datastore using a local transaction...

            // The 'false' indicates the success of a single delivery, 'true' would
            // mean that this delivery and all prior unacknowledged deliveries on this
            // channel will be acknowledged, which I find no reason for in this example.
            d.Ack(false)
        }
    }()

    fmt.Println("Service listening for events...")

    // Block until 'forever' receives a value, which will never happen.
    <-forever
}

RabbitMQ uses port 5672 by default for non-TLS connections and "guest" as the username & password. You can study the plethora of configuration options available and how to use them with Pika and Go amqp.

You might be wondering what this line is for: d.Ack(false)

This tells the broker that the message has been delivered, processed successfully and can be deleted. By default, these acknowledgments happen automatically. But we specified so otherwise when we subscribed to the queue: ch.Consume.

Now, if the Go service crashes (for any unforeseeable reason), the acknowledgment won't be sent and this will cause the broker to re-queue the message so that it may be given another chance to be processed.

Launching The Microservices

Alright, let's fire'em up:

Run docker-compose up

When the three services have been built (will take at least a few minutes the first time), check for their names using docker ps:

Open two new terminals, SSH into the Python and Go containers using the respective container names and start the servers:

docker exec -it microservicesusingrabbitmq_python-service_1 bash
FLASK_APP=main.py python -m flask run --port 3000 --host 0.0.0.0

docker exec -it microservicesusingrabbitmq_go-service_1 bash
go run main.go

Open a third terminal to send the POST request. I'll use Curl:

curl -d "full_name=usama" -X POST http://localhost:3000/users/1

And you'll see the transmission:

At any point, you may also SSH into the RabbitMQ container and just look around:

rabbitmqctl list_exchanges (list all the exchanges on this broker node)
rabbitmqctl list_queues (list all the queues on this broker node)
rabbitmqctl list_bindings (list all the bindings on this broker node)
rabbitmqctl list_queues name messages_ready messages_unacknowledged (list all the queues with the number of messages each has that are ready to be delivered to clients but not yet delivered and those delivered but not yet acknowledged)

As I mentioned at the start, this is by no means a deep dive into microservices. There are many questions to be asked and I'll try to answer an important one: how do we make this communication transactional? So what happens if our Go service (consumer) throws an exception while updating the state on its end and we need to make sure that the update event rolls back across all the services that were affected by it?. Imagine how complicated this could get when we have several microservices and thousands of such "update events". Essentially, we'll need to incorporate separate events that perform the rollback.

In our case, if the Go service throws an exception while updating the data, it will have to send a message back to the Python service telling it to rollback the update. It's also important to note that in the case of these kinds of errors, the message delivery will have to be acknowledged (even though the processing wasn't successful), so that the message doesn't get re-queued by the broker. When writing our consumer, we'll have to decide which errors mean that the message should be re-queued (tried again) and which mean that the message should not be re-queued and just rolled back.

But how do we specify which update event to rollback and how exactly would the rollback happen? The Saga pattern along with event sourcing is widely used to ensure such data consistency.

A Few Words On Designing The Broker

Consider two things: the types of exchanges to use and how to group the exchanges.

If you need to broadcast certain kinds of messages to all the services in your system, have a look at the fanout exchange type. Then, one way to group the exchanges could be on the basis of events e.g. three fanout exchanges named user.profile.updated, user.profile.deleted, user.profile.added. This might not be what you want all the time since you might end up with too many exchanges and won’t be able to filter messages for specific consumers without creating a new exchange, queue and binding.

Another way could be to create topic exchanges in terms of entities in your system. So in our first example, user, movie, cinema etc could be entities and, say, queues bound to user could use binding keys like user.created (get message when a user is created), user.login (get message when a user has just logged in), user.roles.grant (get message telling that the user has been given an authorization role), user.notify (send the user a notification) etc.

Always use routing to filter and deliver messages to specific consumers. Writing code to discard certain incoming messages while accepting others is an anti-pattern. Consumers should only receive messages that they require.

Finally, if your needs are complex and you require that messages be filtered down to certain consumers on the basis of multiple properties, use a headers exchange.

Enjoy!

Service Workers & Rails Middleware

Usama Ashraf — Mon, 09 Apr 2018 11:39:51 +0000

One of the most promising features of the HTML5 API are Web Workers. JavaScript is of course single-threaded and thus will block its event loop while waiting on any long-running, synchronous operation. On the browser this could mean a frozen interface or worse. But when using Web Workers, this doesn't apply since they're entirely isolated browser threads running in the background, in an entirely different context from the web page and therefore have no access to the DOM, the window or document object.

Service Workers are simply special kinds of Web Workers that are installed by an origin server, run in the background and allow requests made to that server to be intercepted. So they kind of sit in between the browser and the server. Service Workers are a core part of modern Progressive Web Apps (PWAs).

Now, say, your internet is down. Since there's a running process that can intercept requests made to your site, we can write that process to display a custom, previously cached page rather than a miserable, grey page of death. It's cosmetic. Sure. But trust me, your end-users will notice. Don't believe me? Turn off your Wifi and reload this page. Looks pretty cool.

Admittedly, this article is meant more to be a leisurely excursion into two relatively advanced topics than a perfect solution to a critical problem (which it isn't). And the reason is...well why not.

Adding Our Service Worker To The Browser

On the Rails side I'll follow this directory structure to store the service worker script and the offline assets:

app/
   ...
   public/
       ...
       service-worker.js
       offline/
               offline.html
               offline.css
               offline.jpg

Thus the relevant URIs would be /service-worker.js, /offline/offline.html and so on.

The following script will register a given service worker in your browser. On Rails, just dumping it inside the manifest application.js (or a custom manifest file at another location) will work. Albeit it'd be much better to use a separate file:

// app/assets/javascripts/application.js
navigator.serviceWorker.register('/service-worker.js')
        .then(function(reg) {
            console.log('Service worker registration succeeded!');
        }).catch(function(error) {
            console.log('Service worker registration failed: ' + error);
        });

After which you can see the following in the Chrome DevTools:

Registration might fail for the following reasons:

Your site is not running on HTTPS (security concern). localhost works as well since it's considered secure but if you're developing locally on something other than localhost, say, a Vagrant-provisioned VM, just forward traffic on localhost to the VM IP. Here's how to create that proxy on Windows if our site in dev is running on 192.168.99.100:3000:

netsh interface portproxy add v4tov4 listenport=80 listenaddress=localhost connectport=3000 connectaddress=192.168.99.100

The service worker file is on a different origin to that of your app (security concern, can't have a service worker mess with requests made to other sites!).
The path to your service worker file is not right. It has to be relative to the origin.

Some out-dated browsers don't support Service Workers and since I'll be employing ES6 as well as the Cache API (for storing the offline page), let's check for their availability first:

// application.js

// Use feature-detection to check for ES6 support.
function browserSupportsES6() {
    try { eval("var foo = (x)=>x+1"); }
    catch (e) { return false; }
    return true;
}

// Use service workers only if the browser supports ES6, 
// the Cache API and of course Service Workers themselves.
if (browserSupportsES6() && ('caches' in window) && ('serviceWorker' in navigator)) {
    navigator.serviceWorker.register('/service-worker.js')
        .then(function(reg) {
            console.log('Service worker registration succeeded!');
        }).catch(function(error) {
            console.log('Service worker registration failed: ' + error);
        });
}

Alright, let's get to the meat of the story: the service worker itself.
After registration, the browser will attempt to install and then activate the service worker. The difference between these two will become clear shortly. For now, it's enough to know that the install event only happens once (for the same service worker script) and is a great place to hook into and fetch & cache our offline page.
As I mentioned earlier, the service worker can intercept requests made to the server. This is achieved by hooking into the fetch event.

// service-worker.js

// Path is relative to the origin.
const OFFLINE_PAGE_URL = 'offline/offline.html';
// We'll add more URIs to this array later.
const ASSETS_TO_BE_CACHED = [OFFLINE_PAGE_URL];

self.addEventListener('install', event => {
    event.waitUntil(
        // The Cache API is domain specific and allows an app to create & name 
        // various caches it'll use. This allows for better data organization.
        // Under each named cache, we'll add our key-value pairs.
        caches.open('my-service-worker-cache-name').then((cache) => {
            // addAll() hits (GET request) all the URIs in the array and caches 
            // the results, with the URIs as the keys.
            cache.addAll(ASSETS_TO_BE_CACHED)
                .then(() => console.log('Assets added to cache.'))
                .catch(err => console.log('Error while fetching assets', err));
            })
    );
});

self.addEventListener('fetch', (e) => {
    // All requests made to the server will pass through here.
    let response = fetch(e.request)
        .then((response) => response)
        // If one fails, return the offline page from the cache.
        // caches.match doesn't require the name of the specific
        // cache in which the key is located. It just traverses all created
        // by the current domain and fetches the first one.
        .catch(() => caches.match(OFFLINE_PAGE_URL));

    e.respondWith(response);
});

The promise passed to event.waitUntil() lets the browser know when your install completes, and if it was successful.
It's important to know that "failure" of a request passed through the fetch handler, using the Fetch API, doesn't mean a 4xx or 5xx response, unlike other AJAX implementations. That promise only fails in case of a network error or wrong CORS config on the server. Which is exactly what we want.

So our service worker in its current state will work. Except...

What about the assets used by our offline page (CSS, JS, images etc)? Shouldn't they be cached just like the HTML of the offline page?
What if we want to change any of the cached assets or the page itself? How would we update the cache on the browser accordingly?
What if we change our service worker script? How will the update happen on the browser?

All valid questions. Let's take'em on one by one:

Yes. Just add the asset URIs to the ASSETS_TO_BE_CACHED array and they'll be cached. In the fetch handler add a simple check for these assets, pull them from the cache and only hit the server if they're not found in the cache (ideally this will never happen).
Version your cache. Whenever you change the offline page or its assets, rename the cache created by the service worker and delete all other caches that have a different name (which will include the previously stored cache).
Every time you refresh your page the browser will perform a byte-by-byte comparison with the installed service worker script and if there's a difference it'll re-install it, immediately. This comparison also occurs every 24 hours or so automatic update checks. Of course, as you may have guessed, the browser will cache any assets that your Rails app serves by way of the Cache-Control header. If our service worker remains cached for a long time, well, that's bad. We want immediate updates. That's where we'll use a custom Rails middleware to skip the header for our service worker(s).

Note: The new worker won't be activated, i.e. in running state, until the previous worker is removed. This requires either a new session (new tab) or a manual update (the button for which is visible in the Chrome console picture above). We don't need to worry about this for the most part.

// service-worker.js

// Changing the cache version will cause existing cached resources to be
// deleted the next time the service worker is re-installed and re-activated.
const CACHE_VERSION = 1;
const CURRENT_CACHE = `your-app-name-cache-v-${CACHE_VERSION}`;
const OFFLINE_PAGE_URL = 'offline/offline.html';
const ASSETS_TO_BE_CACHED = ['offline/offline.css', 'offline/offline.jpg', OFFLINE_PAGE_URL];

self.addEventListener('install', event => {
    event.waitUntil(
        caches.open(CURRENT_CACHE).then((cache) => {
            // addAll() hits all the URIs in the array and caches 
            // the results, with the URIs as the keys.
            cache.addAll(ASSETS_TO_BE_CACHED)
                .then(() => console.log('Assets added to cache'))
                .catch(err => console.log('Error while fetching assets', err));
            })
    );
});

self.addEventListener('activate', event => {
    // Delete all caches except for CURRENT_CACHE, thus deleting the previous cache
    event.waitUntil(
        caches.keys().then(cacheNames => {
            return Promise.all(
                cacheNames.map(cacheName => {
                    if (cacheName !== CURRENT_CACHE) {
                        console.log('Deleting out of date cache:', cacheName);
                        return caches.delete(cacheName);
                    }
                })
            );
        })
    );
});

self.addEventListener('fetch', (e) => {
    const request = e.request;
    // If it's a request for an asset of the offline page.
    if (ASSETS_TO_BE_CACHED.some(uri => request.url.includes(uri))) {
        return e.respondWith(
            caches.match(request).then((response) => {
                // Pull from cache, otherwise fetch from the server.
                return response || fetch(request);
            })
        );
    }

    let response = fetch(request)
        .then((response) => response)
        .catch(() => caches.match(OFFLINE_PAGE_URL));

    e.respondWith(response);
});

You may ask why we purged the previously cached assets in activate and not install. This is done because if we were to purge the out-dated cache in install, the currently running worker at that time, still the old one because the new one hasn't yet been activated, might crash because of any dependency on the previous cache. Now, in our case there is no such strong dependency. But it's generally a good practice so I chose to follow it.

OK, But Where Does Rails Middleware Come In? And By The Way, WTF Is Middleware?

As I mentioned in point number 3 above, ideally service workers shouldn't be cacheable because we may require the clients' browsers to update them ASAP. This is our ultimate objective and middleware is just a means to that end. There could be better ways of doing this. Say, serving assets via Nginx and configuring it to not make service workers cacheable. But let's do it this way:

Middleware is basically any piece of code that wraps HTTP requests and responses made to and received from a server. It's very similar to the concept in Express. When a request passes through the entire middleware stack, it hits the app and then bounces back up the stack as the complete HTTP response.
As far as we're concerned, it's some code that a response generated by our Rails app has to go through before being returned to the browser. Rails is a Rack-based framework and this is what allows us to use this construct. Getting into the details of Rack here would probably be an overkill though.

Source: https://philsturgeon.uk

Rails middleware specification is really simple: it just has to implement a call method. Let's create a new directory app/middleware and put a new class in:

# app/middleware/service_worker_manager.rb
class ServiceWorkerManager
  # We'll pass 'service_workers' when we register this middleware.
  def initialize(app, service_workers)
    @app = app
    @service_workers = service_workers
  end

  def call(env)
    # Let the next middleware classes & app do their thing first...
    status, headers, response = @app.call(env)

    dont_cache = @service_workers.any? { |worker_name| env['REQUEST_PATH'].include?(worker_name) }

    # ...and modify the response if a service worker was fetched.
    if dont_cache
      headers['Cache-Control'] = 'no-cache'
    end

    [status, headers, response]
  end
end

We'll need to register our middleware:

# config/environments/development.rb

# ...

# Add our own middleware before the ActionDispatch::Static middleware
# and pass it an array of service worker URIs as a parameter.
config.middleware.insert_before ActionDispatch::Static, ServiceWorkerManager, ['service-worker.js']

If you run rake middleware you'll get a list of all the middleware being used by your app:

As we specified, ServiceWorkerManager has been added right before ActionDispatch::Static. The reason we added it at this particular position is because ActionDispatch::Static is used to serve static files from the public directory and automatically sets the Cache-Control header as well. In fact, static file requests don't even go through the rest of the middleware stack or the app and are returned immediately. So by placing our middleware before it, the outgoing response will have to pass through our middleware.

Here are two alternative ways to use your own middleware classes:

# Add our middleware after another one in the stack.
config.middleware.insert_after SomeOtherMiddleware, MyMiddleware, params...

# Appending our middleware to the end of the stack.
config.middleware.use MyMiddleware, params...

And we're done! Load your site, shut down your local server and try refreshing the page to see offline.html.

I've intentionally left out using the asset pipeline to transpile, minify our worker script and offline assets as that is not straightforward: fingerprinting (dynamic URIs that always change with the asset's content) & long-lived caching headers (which we just got rid of) cause problems with service workers. These are explained & solved very well here: a much better way to do all that we just did IMO :)

Offline page caching was just sort of a case study I used. Service Workers can be utilized to implement much richer features like background syncing, push notifications & others. And so can middleware: custom request logging, tracking, throttling etc. Hopefully this was a good learning experience and has motivated you to explore the two subjects further.

Scaling Out With Docker and Nginx

Usama Ashraf — Wed, 28 Mar 2018 14:10:28 +0000

Number 10 of the 12 Commandments states that our development environment should be as similar as possible to what we have in production. Not doing so can lead to many predicaments when debugging critical, live issues.

One important step in this regard would be to mimic a standard distributed production setup: a load-balancer sitting in front of multiple backend server instances, dividing incoming HTTP traffic among them.

This article is not an introduction to Docker, Compose or Nginx. But rather a guide to setting up the distributed system described above, assuming we’ve installed Docker and are familiar with images, containers etc. I’ll try to provide enough information about Compose and Nginx so we can get our hands dirty without succumbing to the noise that normally goes with them.

Each of our backend server instances (simple Node.js servers) and the Nginx load-balancer will be hosted inside Docker-based Linux containers. If we stick to 3 backend servers and 1 load-balancer we’ll need to manage/provision 4 Docker containers, for which Compose is a great tool.

What We Want

Here’s our directory structure:

docker-nginx/
       backend/
              src/
                 index.js
                 package-lock.json  
              Dockerfile
       load-balancer/
              nginx.conf
              Dockerfile
       docker-compose.yml

The src directory will contain our server-side code, in this case a simple Hello World Node (Express) app (of course your backend can be anything):

// index.js
const express = require('express');
const app = express();

app.get('/', (req, res) =>  {      
   console.log('I just received a GET request on port 3000!');
   res.send('Hello World!');
});

app.listen(3000, () => console.log('I just connected on port 3000!'));

The package-lock.json has nothing but an Express dependency .

Setting Up Our Dockerfiles and Compose Config

backend/Dockerfile will be used to build an image, which will then be used by Compose to provision 3 identical containers:

# Use one of the standard Node images from Docker Hub 
FROM node:boron
# The Dockerfile's author
LABEL Usama Ashraf
# Create a directory in the container where the code will be placed
RUN mkdir -p /backend-dir-inside-container
# Set this as the default working directory.
# We'll land here when we SSH into the container.
WORKDIR /backend-dir-inside-container
# Our Nginx container will forward HTTP traffic to containers of 
# this image via port 3000. For this, 3000 needs to be 'open'.
EXPOSE 3000

load-balancer/Dockerfile:

# Use the standard Nginx image from Docker Hub
FROM nginx
# The Dockerfile's author
LABEL Usama Ashraf
# Copy the configuration file from the current directory and paste 
# it inside the container to use it as Nginx's default config.
COPY nginx.conf /etc/nginx/nginx.conf
# Port 8080 of the container will be exposed and then mapped to port
# 8080 of our host machine via Compose. This way we'll be able to 
# access the server via localhost:8080 on our host.
EXPOSE 8080

# Start Nginx when the container has provisioned.
CMD ["nginx", "-g", "daemon off;"]

load-balancer/nginx.conf:

http {

 events { worker_connections 1024; }
 upstream localhost {
    # These are references to our backend containers, facilitated by
    # Compose, as defined in docker-compose.yml   
    server backend1:3000;
    server backend2:3000;
    server backend3:3000;
 }
 server {
    listen 8080;
    server_name localhost;
    location / {
       proxy_pass http://localhost;
       proxy_set_header Host $host;
    }
  }
}

This is a bare-bones Nginx configuration file. If anyone would like help with more advanced options please do post a comment and I’ll be happy to assist.

docker-compose.yml:

version: '3.2'
services:
  backend1:
      build: ./backend
      tty: true
      volumes:
        - './backend/src:/backend-dir-inside-container'
  backend2:
      build: ./backend
      tty: true
      volumes:
        - './backend/src:/backend-dir-inside-container'
  backend3:
      build: ./backend
      tty: true
      volumes:
        - './backend/src:/backend-dir-inside-container'
  loadbalancer:
      build: ./load-balancer
      tty: true
      links:
          - backend1
          - backend2
          - backend3
      ports:
          - '8080:8080'
volumes:
  backend:

Without going into details, here’s some insight into our Compose config:

A single Compose service generally uses one image, defined by a Dockerfile. When we build our services, the images are built & provisioned as containers.
If you’re new to Docker but familiar with VMs then may be this analogy will help: an ISO file for an OS (image) is used by VirtualBox (Compose) to launch a running VM (container). A service is made up of at least one running container.

build tells Compose where to look for a Dockerfile to build the image for the service.
tty just tells the container to keep running even when there’s no daemon specified via CMD in the Dockerfile. Otherwise, it’ll shut down right after provisioning (sounds strange, I know).
volumes in our case defines where to put the server-side code in the container (oversimplification). Volumes are a storage mechanism within containers and not a trivial feature.
links does two things: makes sure theloadbalancer service doesn’t start unless the backend services have started. And it allows backend1 , backend2 and backend3 to be used as references within loadbalancer, which we did in our nginx.conf.
ports specifies a mapping between a host port and a container port. 8080 of the container will receive client requests made to localhost:8080 on the host.

Launch

Run sudo docker-compose up --build inside docker-nginx (or whatever your project root is). After all the services have started, run sudo docker ps and you’ll see something like the following, a list of all the containers just launched:

Let’s SSH into one of the backend containers and start our Node server. After hitting localhost:8080 from our browser 5 times, this is what we get:

Of course the browser is hitting 8080 on the host machine, which has been mapped to 8080 on the Nginx container, which in turn forwards the request to port 3000 of the backend container. Hence the log shows requests on port 3000.

Open two new terminals on your host machine, SSH into the other two backend containers and start the Node servers within both:

sudo docker exec -it dockernginx_backend1_1 bash
node index.js

sudo docker exec -it dockernginx_backend3_1 bash
node index.js

Hit localhost:8080 on your browser multiple times (fast) and you’ll see that the requests are being divided among the 3 servers!
Now you can simulate stuff like session, cache persistence across multiple servers, concurrency issues or even get a rough idea about the throughput increase we can achieve when we scale out (e.g. by using wrk).

Here it is all in one place.

To deploy a similar Compose setup in production I recommend this; and you might enjoy kompose if you need to export it to Kubernetes.

Note: The more experienced might bring up Compose’s ability to scale a running service and ask whether it is possible to configure Nginx so as to auto-scale at run-time: sort of. The Nginx daemon will have to be restarted (downtime) and of course we’ll need a way to dynamically edit and add to the upstream servers’ group, which is certainly possible, but a fruitless hassle in this case IMO. If more server instances are needed, add more services and rebuild the images.

N+1 Queries, Batch Loading & Active Model Serializers in Rails

Usama Ashraf — Fri, 16 Mar 2018 11:16:15 +0000

The n+1 query problem is one of the most common scalability bottlenecks. If you’re comfortable with Rails, Active Model Serializers and already have a good idea about what our problem is going to be, then may be you can just jump straight into the code here.

Say you’re fetching an array of Post objects at a GET endpoint and you also want to load the respective authors of the posts, embedding an author object within each of the post objects. Here’s a naive way of doing it:

class PostsController < ApplicationController  
  def index    
    posts = Post.all     

    render json: posts  
  end
end

class Post
  belongs_to :author, class_name: 'User'
end

class PostSerializer < ActiveModel::Serializer  
  attributes :id, :title, :details
  belongs_to :author 
end

For each of the n Post objects being rendered a query will run to fetch the corresponding User object, hence we’ll run a total of n+1 queries. This is disastrous. And here’s how you fix it by eager loading the User object:

class PostsController < ApplicationController  
  def index    
    # Runs a SQL join with the users table.
    posts = Post.includes(:author).all     

    render json: posts  
  end
end

Until now there’s absolutely nothing new for veterans.

But let’s complicate this. Let’s assume that the site’s users are not being stored in the same RDBMS as the posts are, rather, the users are documents stored in MongoDB (for whatever reason). How do we modify our Post serializer to fetch the user now, optimally? This would be going back to square one:

class PostSerializer < ActiveModel::Serializer  
  attributes :id, :title, :details, :author

  # Will run n Mongo queries for n posts being rendered.
  def author
    User.find(object.author_id)
  end
end

# This is now a Mongoid document, not an ActiveRecord model.
class User  
  include Mongoid::Document  
  include Mongoid::Timestamps  
  # ...
end

The predicament that our users now reside in a Mongo database can be substituted with, say, calling a 3rd party HTTP service for fetching the users or storing them in a completely different RDBMS. Our essential problem remains that there’s no way to ‘join’ the users datastore with the posts table and get the response we want in a single query.

Of course, we can do better. We can fetch the entire response in two queries:

Fetch all the posts without the author attribute (1 SQL query).
Fetch all the corresponding authors by running a where-in query with the user IDs plucked from the array of posts (1 Mongo query with an IN clause).

posts      = Post.all
author_ids = posts.pluck(:author_id)
authors    = User.where(:_id.in => author_ids)

# Somehow pass the author objects to the post serializer and
# map them to the correct post objects. Can't imagine what 
# exactly that would look like, but probably not pretty.
render json: posts, pass_some_parameter_maybe: authors

So our original optimization problem has been reduced to “how do we make this code readable and maintainable”. The folks at Universe have come up with an absolute gem (too obvious?). Batch Loader has been incredibly helpful to me recently.

gem 'batch-loader'

bundle install

class PostSerializer < ActiveModel::Serializer  
  attributes :id, :title, :details, :author

  def author
    object.get_author_lazily
  end
end

class Post
  def get_author_lazily
    # The current post object is added to the batch here,
    # which is eventually processed when the block executes.   
    BatchLoader.for(self).batch do |posts, batch_loader|          
      author_ids = posts.pluck(:author_id)  
      User.where(:_id.in => author_ids).each do |user|
        post = posts.detect { |p| p.author_id == user._id.to_s }
        #'Assign' the user object to the right post.
        batch_loader.call(post, user)      
      end    
    end  
  end
end

If you’re familiar with Javascript Promises, think of the get_author_lazily method as returning a Promise which is evaluated later. That’s a decent analogy, I think, since BatchLoader uses lazy Ruby objects.
One caveat: BatchLoader caches the loaded values and to keep the responses up-to-date you should add this to your config/application.rb:

config.middleware.use BatchLoader::Middleware

That’s basically it! We’ve solved an advanced version of the n+1 queries problem while keeping our code clean and using Active Model Serializers the right way.

One problem though. If you have a User serializer (Active Model Serializers work with Mongoid as well), that won’t be called for the lazily loaded author objects, unlike before. To fix this we can use a Ruby block and serialize the author objects before they’re ‘assigned’ to the posts.

class PostSerializer < ActiveModel::Serializer  
  attributes :id, :title, :details, :author
  def author
    object.get_author_lazily do |author|
      # Serialize the author after it has been loaded.     
      ActiveModelSerializers::SerializableResource
                             .new(author)
                             .as_json[:user]
    end
  end
end

class Post
  def get_author_lazily
    # The current post object is added to the batch here,
    # which is eventually processed when the block executes.   
    BatchLoader.for(self).batch do |posts, batch_loader|
      author_ids = posts.pluck(:author_id)
      User.where(:_id.in => author_ids).each do |user|
        modified_user = block_given? ? yield(user) : user
        post = posts.detect { |p| p.author_id == user._id.to_s }  
        # 'Assign' the user object to the right post.
        batch_loader.call(post, modified_user)      
      end    
    end  
  end
end

Here’s the entire code. Enjoy!