19. Logical Network Topological Nodes

Several applications support drawing network diagrams. There are lots of flavors of edges and nodes, but they share commonalities. For the sake of this chapter, we’ll choose a few icons necessary to represent the components we’ll be discussing. Google Drive makes many icons available using the drawing dialog.

Hosts, compute instances, and worker machines are all terms referring to a computer on a network. They can host server applications, implement mapreduce, host daemonized processes like crond, or host a queue listener such as an nsq or a kafka client as well as any number of other tasks. We’ll represent these with the icon shown in Figure 19.1, where “Hostname” represents the name of the host in the network.

The Internet is usually represented with a cloud icon, as shown in Figure 19.2. It represents any type of possible traffic. This has a number of implications. First, any sort of request could come through. If there are types of traffic you would like to accept or reject, the Internet icon serves as a reminder that input should be filtered to only the sort of traffic desired. Further, it comes with the implication that potentially malicious traffic could be encountered, so inputs should be sanitized. These are two basic building blocks and the minimum required for a complete public-facing application (though the application wouldn’t be terribly resilient). Let’s get into some components that offer more robustness.

There are several things to note here. First, the border around the nginx and api nodes indicates they are on the same network or behind the same gateway, which is the door to the Internet. The gateway is not pictured in Figure 19.3 since it’s out of the scope of our discussion. For our purposes, you can consider this boundary a convenient way to group nodes in your network.

Next, you should note there are two nginx hosts receiving requests from the Internet. Each of these has its own IP, but they also have a virtual IP, which is shared between them by CARP. This allows one to take over for the other in the event there is a node failure.

Finally, there are three api nodes, and each can receive requests from each nginx node. These are drawn as a full directional mesh, but it’s worth noting a lot of people prefer to draw an API layer to cut down on busy details like lots of arrows.

Most cloud providers make a reliable load balancer available to you. They take care of these little details so you don’t have to. It is still a good idea to know how to do them yourself.

There are a few great use cases for load balancers. The most common is the one pictured in Figure 19.3. A second one, though, is as a means of implementing multitenancy. Many applications are single-threaded (Python and Redis, for example). To make the most use of the CPU cores on larger machines, it’s common to run several of these processes at once, each listening on a different port.

Figure 19.4 shows how you might diagram multitenancy on a single API node.

The colon indicates what port each process is listening to on that host. Here it’s understood from the context that nginx is routing requests to/from the API processes on this machine. You could make it more explicit, though, by drawing arrows.

Caching in the application is appealing sometimes, especially when the application is small. In general, though, if you wanted to speak strictly, this violates the separation of concerns between processes. You should have one process for caching and another for handling requests. The 12-factor standard for application development supports this with the assertion that state should not be maintained in a 12-factor application. Something sitting in the cache represents the state of that cached object at that time.

If you insist on application-level caching, be aware of some important caveats: processes don’t typically share memory. This means a cache on one process will not be visible to another; hence, invalidating an object on write in that cache will not invalidate it in other application caches. Effective caching requires a memory management/eviction strategy. If you don’t have a process in place for removing elements from the cache, you have effectively created a memory leak. Each time your application reboots (normal deploys, for example), the origin of your cached resource will see burst traffic. This can become a bigger concern as your application becomes more popular. Shared resources between requests, especially those that require authentication, should be a big red flag. Making it possible to share state across requests opens your application to many dangers, including accidentally revealing an access token (or worse) for one user to another user.

Now, if you still insist on application-level caching, allow us to give two use cases, one of which is totally fine and another that can cause major headaches.

This is typically done by making a database call for the content through another API, joining those results to the recommender results, and sending the whole package back to the user. If you use an application cache, you can store the content mapped to each ID and avoid ever having to make the request. This would be a big time-saver. Since the content is pretty static, it’s no big deal if one application’s version goes stale for a while.

You might notice in this case content that is currently popular stays popular for a while and then makes way for new popular content. In this case, a “least recently used” eviction policy would be pretty appropriate. As content becomes less often used, it will fall off the end of the cache. This ensures the number of external API calls for the static content is at a minimum.

This seems sound at a glance. However, it assumes there is no shared state between requests. In app servers that allow for coroutines and other asynchronous code, these objects can still persist as other requests are being served since one request can be paused while waiting on IO and another request resumed. If, in your subroutines, you use a globally defined method to reference the request, you will inevitably end up with the wrong user information in the request.

In Figure 19.5 you can find a typical interaction between the API layer, the application cache, and the database. The cache usually serves two purposes in this context. The first is a response cache. The second is an object cache.

Caching responses that can be reused allows the API to save a ton of time. What might be a series of template computations and API and database calls turns into an O(1) lookup and a simple transfer of data.

Object caches allow the API to say “Have I requested this recently?” and in the case it has, it says, “Okay, I’ll take it from the cache instead of the database.”

One of the simplest ways to manage applications that are aggregating concurrently is to use the incr/decr methods made available by Redis. If the statistics are not business critical in the sense that they absolutely cannot disappear, a cache is a good choice to aggregate them and serve them from.

When you do need that persistence layer, a database will usually do. If the rate of incoming updates becomes too high, though, you may consider techniques for batching them. There are two approaches to using a write-through cache to do this that come to mind.

The first involves a write-through cache in the application. As your taskworker receives messages from your queueing system, it saves them for a moment. If it encounters more than one update for a single record, it combines them. When those updates are successfully communicated to the database, the taskworker signals it has completed the update.

The second approach to managing a write-through cache takes into account that a user will continue sending events related to their session until the session times out. Imagine taskworkers each register a user session in the cache upon receipt of the first message from a given user. If the session already exists, the taskworker can increment its reference count. If a taskworker already has an active session (meaning they have incremented the session by 1), each additional message for the session is simply written to the cache as it arrives. This allows you to track how many taskworkers have an active reference to a given session at any time during the session. When that initial create session or increment operation occurs, a timer can be started on the taskworker. A new message coming in extends the time. No messages coming in allows the clock to run out. When the clock runs out for a given taskworker, it decrements the reference count to the session. The last taskworker to decrement its reference count deletes the session from the cache and writes the aggregate data to the database.

The gist of it is that only one node accepts database writes. That node is responsible for communicating the data it receives to the replica, secondary, or slave as you prefer. Usually this is done by transferring a binary log of the writes to the secondary node. This allows the two to stay in sync for the most part.

Replica lag is the phenomenon associated with the latency between the primary and replica nodes. It takes a bit of time to sync the binlogs from the primary to the replica, so data on the replica isn’t immediately available after writing to the master. Common replica lag is in milliseconds.

This is an important caveat to take into account. If you need a strong guarantee the value you’re reading is completely up-to-date, you will have to do that read on the primary node. A common method for scaling at the early stage, though, is reading from the replica wherever possible. This phenomenon is what decides what reads can be done where.

Naturally with this configuration your writes go to one node, and to the best of your ability you send the reads to another. If your application is more read-heavy, you can keep adding replica nodes with little consequence in terms of syncing binlogs. But what if it’s more write-heavy?

But it simplifies reading from your database to have them write to the same replica.

In the first approach, clients can choose what primary node to write to at random and distribute objects pretty effectively. There are some important caveats to this approach. As long as reads are sent to the replica, it’s simple to return a complete result. If reads are sent to either primary node, the results will be missing the data stored on the parallel primary node. The second important caveat involves avoiding conflicting data on the replica. If your keys are set to auto-increment on both master nodes, two objects sent to the replica can and will have matching primary keys, which violates the unique constraint of a primary key. A simple solution is to increment on both masters by 2 and give one an offset of 1.

This approach makes a lot of sense in terms of storage. If you have a 100-node cluster, for example, each containing a complete record of your database, that is a lot of space! By copying each shard twice, you save a lot of room. Many databases take this approach to replicating data. Elasticsearch, MongoDB, and Cassandra are a few.

Figure 19.9 shows a slightly more complicated approach to sharding. Elasticsearch manages its data in this way. If one node disappears for whatever reason, it begins copying the missing copy of whichever shards were on that node to another available host. If two nodes disappear with the only copy of a shard, that data is lost forever.

Why might you use a queue instead of just running the tasks synchronously, one after the other? I’m glad you asked. There is a single case where it makes sense to do that. That is in the case your dependency has no branches, as shown in Figure 19.10. We start with the event, A. The result of A is sent to process B, and the result of B is sent to process C. Executing this on one node would have about the same benefit as using a queue since they have to execute serially.

When you have multiple tasks dependent on a single result, the case for a queue becomes clear. By having each dependent process subscribe to the queue for the process on which it depends, they can all be relayed the result to process independently. Further, if there are tasks dependent on them, they can be triggered in the same way. In Figure 19.11, B and C both depend on the result of A. D, E, and F depend on the result of B, while G depends on the result of C.

The benefit of this approach is parallelization. Assuming there exist four processes to do the execution, D, E, F, and G can all be executed at the same time. Distributing workload in this way ensures computational resources are used as efficiently as possible since those processes that are not busy are listening for work to do.

One great example of this is when a user signs up to a social network and you want to find all their friends on that network and make it available for them to interact with those friends. In this case, you’ll send an event indicating contacts should be synced and then allow the user to proceed to the next dialog.

This could look something like what is pictured in Figure 19.12.

In other words, the client signs up for a service to the API. The API sends a message to a queue indicating contacts should be synced for that user. A taskworker, listening to that queue, receives the message and queries a third party. It then relays those results to the API upon receipt, making them available to the user.

Queues provide a great way to space out incoming requests to a rate the API can handle. If writes are written to a horizontally scaled queue, they can be inserted as the API is available. Taskworkers listen for those writes and execute them just as quickly as they can, alerting the client to their completion through push or some other server client protocol.

As you build out your architectural designs, diagramming and referring to those diagrams during discussions makes for well-informed, well-motivated conversations. There are many flavors of these diagrams, but the most important thing to remember is consistency.