Chapter 4. Event stream processing with Amazon Kinesis

So far in this book, we have worked exclusively with Apache Kafka as our unified log. Because it is an open source technology, we have had to set up and configure Kafka and its dependencies (such as ZooKeeper) ourselves. This has given us great insight into how a unified log works “under the hood,” but some of you may be wondering whether there is an alternative that is not so operationally demanding. Can we outsource the operation of our unified log to a third party, without losing the great qualities of the unified log?

The answer is a qualified yes. This chapter introduces Amazon Kinesis (https://aws.amazon.com/kinesis/), a hosted unified log service available as part of Amazon Web Services. Developed internally at Amazon to solve its own challenges around log collection at scale, Kinesis has extremely similar semantics to Kafka—along with subtle differences that we will tease out in this chapter.

Before kicking off this chapter, you might want to jump to the appendix for a brief AWS primer that will get you up to speed on the Amazon Web Services platform, unless you already know your way into the AWS ecosystem. Once your AWS account is set up, this chapter will show you how to use the AWS command-line interface (CLI) tools to create your first event stream in Kinesis and write and read some events to it.

We will then dive into a new use case for the unified log: using it for systems monitoring. We will create a simple long-running agent in Python that emits a steady stream of readings from our server, writing these events to Kinesis by using the AWS Python SDK. Once we have these events in Kinesis, we will write another Python application that monitors our agent’s events, looking for potential problems. Again, this Python monitoring application will be built using the AWS Python SDK, also known as boto.

Please note that, as Amazon Kinesis Data Streams is not currently available in AWS Free Tier, the procedures in this book will necessarily involve creating live resources in your Amazon Web Services account, which can incur some charges.^[1] Don’t worry—I will tell you as soon as you can safely delete a given resource. In addition, you can set alerts on your spending in order to be notified whenever the charges go above a certain threshold.^[2]

¹You can find more information about the pricing of AWS Kinesis Data Streams at https://aws.amazon.com/kinesis/streams/pricing/.

²You can set up AWS billing notification alerts at https://docs.aws.amazon.com/awsaccountbilling/latest/aboutv2/billing-getting-started.html#d0e1069.

4.1. Writing events to Kinesis

Our AWS account is set up, and we have the AWS CLI primed for action: we’re ready to introduce the project we’ll be working through! In previous chapters, we focused on events related to end-user behavior. In this chapter, we’ll take a different tack and generate a simple stream of events related to systems monitoring.

4.1.1. Systems monitoring and the unified log

Let’s imagine that our company has a server that keeps running out of space. It hosts a particularly chatty application that keeps generating lots of log files. Our systems administrator wants to receive a warning whenever the server’s disk reaches 80% full, so that he can go in and manually archive and remove the excess log files.

Of course, there is a rich, mature ecosystem of systems monitoring tools that could meet this requirement. Simplifying somewhat,^[3] these tools typically use one of two monitoring architectures:

³A good article about push versus pull monitoring is at https://web.archive.org/web/20161004192212/ https://www.boxever.com/push-vs-pull-for-monitoring.

Figure 4.1 depicts both the push and pull approaches. Sometimes a systems monitoring tool will provide both approaches. For example, Zabbix and Prometheus are predominantly pull-based systems with some push support.

In the push-based model, we have agents generating events and submitting them to a centralized system; the centralized system then analyzes the event stream obtained from all agents. Does this sound familiar? We can transplant this approach directly into our unified log, as per figure 4.2. The building blocks are the same, as you saw in earlier chapters: event producers, a unified log, and a set of event consumers. But there are two main differences from our previous unified log experiences:

It looks, then, like it should be straightforward to meet our systems administrator’s monitoring requirements by using a unified log such as Apache Kafka or Amazon Kinesis. We will be using Kinesis for this chapter, so before we get started, we’ll take a brief look at the terminology differences between Kafka and Kinesis.

4.1.2. Terminology differences from Kafka

Amazon Kinesis has extremely similar semantics to Apache Kafka. But the two platforms diverge a little in the descriptive language that they use. Figure 4.3 sets out the key differences: essentially, Kinesis uses streams, whereas Kafka uses topics. Kinesis streams consist of one or more shards, whereas Kafka topics contain partitions. Personally, I prefer the Kinesis terms to Kafka’s: they are a little less ambiguous and have less “message queue” baggage.

Differences of language aside, the fact that Kinesis offers the same key building blocks as Kafka is encouraging: it suggests that almost everything we could do in Kafka, we can do in Kinesis. To be sure, we will come across differences of approach and capability through this chapter; I will make sure to highlight these as they come up. For now, let’s get started with Kinesis.

4.1.3. Setting up our stream

First, we need a Kinesis stream to send our systems monitoring events to. Most commands in the AWS CLI follow this format:

In our case, all of our commands will start with aws kinesis. You can find a full reference of all available AWS CLI commands for Kinesis here:

We can create our new stream by using the AWS CLI like so:

Press Enter, and then switch back to the AWS web interface, and click Amazon Kinesis. If you are quick enough, you should see the new stream listed with its status set to CREATING, and with a count of 0 shards, as in figure 4.4. After the stream is created, Kinesis will report the stream status as ACTIVE and display the correct number of shards. We can write events to and read events from only ACTIVE streams.

We created our stream with two shards; events that are sent to the event stream will be written to one or either of the two shards. Any stream processing apps that we write will have to make sure to read events from all shards. At the time of writing, Amazon enforces a few limits around shards:^[4]

⁴AWS Kinesis Data Streams limits are described at https://docs.aws.amazon.com/streams/latest/dev/service-sizes-and-limits.html.

Don’t worry—we won’t be hitting any of these limits; we could have happily made do with just one shard in our stream.

So, at this point, we have our Kinesis stream ready and waiting to receive events. In the next section, let’s model those events.

4.1.4. Modeling our events

Remember that our systems administrator wants to receive a warning whenever the troublesome server’s disk reaches 80% full. To support this monitoring, the agent running on the server will need to regularly read filesystem metrics and send those metrics into our unified log for further analysis. We can model these metrics readings as events by using the grammatical structure introduced in chapter 2:

Putting these together, we can sketch out the event model that we’ll need to assemble, as in figure 4.5.

Now that you know what our events should look like, let’s write our agent.

4.1.5. Writing our agent

We are going to write our agent in Python, making use of the excellent boto3 library, which is the official Python SDK for AWS.^[5] All the official language-specific SDKs for AWS support writing events to Kinesis, which seems fair: for Kinesis to be a truly unified log, we need to be able to send events to it from all our various client applications, whatever language they are written in.

⁵You can download the Python SDK for AWS from https://aws.amazon.com/sdk-for-python/.

Let’s get started. We are going to build up our systems monitoring agent piece by piece, using the Python interactive interpreter. Start it up by logging into your Vagrant guest and typing python at the command prompt:

First let’s define and test a function that gives us the filesystem metrics that we need. Paste the following into your Python interpreter, being careful to keep the white-space intact:

You should see something like the following output; the exact numbers will depend on your computer:

Good—now you know how to retrieve the information we need about the filesystem. Next, let’s create all the metadata that we need for our event. Paste in the following:

You should now see something a little like this:

Note that we are uniquely identifying each event by attaching a freshly minted version 4 UUID as its event ID.^[6] We will explore event IDs in much more detail in chapter 10.

⁶Wikipedia provides a good description of universally unique identifiers: https://en.wikipedia.org/wiki/Universally_unique_identifier.

Let’s put this all together with a function that creates our event as a Python dictionary. Type in the following at the interpreter:

It’s a little verbose, but the intent should be clear from the Python interpreter’s output, which should be something like this:

We have now constructed our first well-structured systems monitoring event! How do we send it to our Kinesis stream? It should be as simple as this:

The key method to understand here is conn.put_record, which takes three required arguments:

Now we just need to connect to Kinesis and try writing an event. This is as simple as the following:

Part of the reason that this code is so simple is that the AWS CLI tool that you configured earlier uses boto, the AWS SDK for Python, under the hood. Therefore, boto can access the AWS credentials that you set up earlier in the AWS CLI without any trouble.

Press Enter on the preceding code and you should be greeted with ... silence! Although, in this case, no news is good news, it would still be nice to get some visual feedback. This can be arranged: next, put the sending of our event into an infinite loop in the Python interpreter, like so:

Leave this running for a couple of minutes, and then head back into the Kinesis section of the AWS web interface, and click your events stream to bring up the Stream Details view. At the bottom of this view, in the Monitoring tab, you should be able to see the beginnings of lines on some of the charts, as in figure 4.6.

Unfortunately, this is the only visual confirmation we can get that we are successfully writing to our Kinesis stream—at least until we write some kind of stream consumer. We will do that soon, but first let’s wrap up our systems monitoring agent. We won’t need the Python interpreter anymore, so you can kill the infinite loop with Ctrl-C, and then exit the interpreter with Ctrl-D.

Let’s consolidate all of our work at the interpreter into a single file to run our agent’s core monitoring loop. Create a file called agent.py and populate it with the contents of the following listing.

Make the agent.py file executable and run it:

Our agent is running! Check out figure 4.7 for a visualization of what we have created.

Leave our Python agent running in a terminal. It will continue to write an Agent read filesystem metrics event every 10 seconds. Now we need to write our stream processing app to consume this stream of events and notify us if our server’s disk reaches the dreaded 80% full.

4.2. Reading from Kinesis

A variety of frameworks and SDKs can read events from a Kinesis stream. Let’s review these briefly, and then use two of these tools to implement basic monitoring of our event stream.

4.2.1. Kinesis frameworks and SDKs

For such a young platform, Kinesis already supports a slightly bewildering array of frameworks and SDKs for event stream processing. Here are the main ones:

A surprising omission from the preceding list is Apache Samza—surprising because, as you have seen, the semantics of Amazon Kinesis and Apache Kafka are extremely similar. Given how well Samza works with Kafka (see chapter 5), we might expect it to work well with Kinesis too.

4.2.2. Reading events with the AWS CLI

Let’s start with the “rawest” tool for processing a Kinesis stream: the AWS CLI. You would likely choose a higher-level framework to build a production application, but the AWS CLI will give you familiarity with the key building blocks underpinning those frameworks.

First, we’ll use the describe-stream command to review the exact contents of our Kinesis stream:

The response is a JSON structure containing an array of Shards, which in turn contains definitions for our stream’s two shards. Each shard is identified by a unique ShardId, and contains metadata:

Before we can read events from our stream, we need to retrieve what Amazon calls a shard iterator for each shard in the stream. A shard iterator is a slightly abstract concept. You can think of it as something like a short-lived (five-minute) file handle on a given shard. Each shard iterator defines a cursor position for a set of sequential reads from the shard, where the cursor position takes the form of the sequence number of the first event to read. Let’s create a shard iterator now:

We specified that the shard iterator should be of type TRIM_HORIZON. This is AWS jargon for the oldest events in the shard that have not yet been trimmed—expired for being too old. At the time of writing, records are trimmed from a Kinesis stream after a fixed period of 24 hours. This period can be increased up to 168 hours but will incur an additional cost. There are three other shard iterator types:

Figure 4.9 illustrates the various shard iterator options.

Now we are ready to try reading some records by using our new shard iterator:

An empty array of records! That’s okay; this simply means that this initial portion of the shard doesn’t contain any events. The AWS documentation warns us that we might have to cycle through a few shard iterators before we reach any events. Fortunately, the NextShardIterator gives us the next “file handle” to use, so let’s plug this into our next get-records call:

I have elided the output, but this time the AWS CLI returns 24 records, along with a NextShardIterator for us to fetch further events from our shard. Let’s just check that the Base64-encoded contents of an event are as we expect. Again, in the Python interpreter, type in the following:

And you should see this:

Good—we can finally confirm that our systems monitoring agent has faithfully recorded our event contents in Kinesis using boto.

Another thing to stress is that, just as in Kafka, each of these records is still stored in the Kinesis stream, available for other applications to consume. It’s not like the act of reading has “popped” these events off the shard forever. We can demonstrate this quickly by creating an all-new shard iterator, this time laser-focused on this same event:

We then request a single event (--limit=1) from our shard by using the new shard iterator:

This is the same PartitionKey (our event ID), SequenceNumber, and indeed Base64-encoded data as before; we have successfully retrieved the same event twice!

No doubt, this section has been a lot to take in. Let’s summarize before we move on:

Figure 4.10 illustrates this process.

4.2.3. Monitoring our stream with boto

Now that you understand the basics of reading events from a Kinesis stream, let’s return to the task at hand: monitoring our agent’s event stream in order to check whether our server is running low on disk space. After getting our hands dirty with the AWS CLI, we’re now going back to the AWS Python SDK, boto. The AWS CLI and boto expose the exact same primitives for stream processing, which is unsurprising, given that the AWS CLI is built on boto! The main difference is that the AWS Python SDK will let us use all the power of Python in our stream processing application.

The stream processing application we’ll build in this section is illustrated in figure 4.11. It follows a simple algorithm:

As before, we are going to build up our application piece by piece in the Python interpreter. This time, however, we’ll start with the stream reading mechanics and then work our way back to the business logic (the event monitoring).

First, we need to create a thread for each of the shards in our event stream. Although we know that our stream has two shards, this could change in the future, so we’ll use boto to check the number of shards our stream currently has. Type the following at your Python prompt:

Press Enter and you should see this:

This is all the information we need to create a thread for each shard:

Press Enter again and you should see this:

Your threads’ numbers may be different. This is the basic pattern for the rest of our monitoring application: we will start a thread to process each shard, and each thread will run our monitoring code. One thing to note is that our application would have to be restarted if more shards were added (or indeed taken away), to create threads for the new shards. This is a limitation of working with Kinesis at such a low level. If we were using one of the Kinesis Client Libraries, changes in our stream’s population of shards would be handled for us transparently.

Now we need a loop in each thread to handle reading events from each shard via shard iterators. The loop will be broadly the same as that of figure 4.11:

We’ll build this loop inside a Python class, imaginatively called ShardReader, which will run on its supplied thread. Type the following at the Python interpreter:

We’ll kick off the threads in a way similar to before:

Press Enter and, assuming that your agent from section 4.1 is still running in another console, you should see something like the following:

There will be a burst of events to start with. This is because both threads are catching up with the historic contents of each shard—all events dating from the so-called TRIM_HORIZON onward. After the backlog is cleared, the output should settle down to an event every 10 seconds or so, matching the rate at which our agent is producing readings. Note the Reader-shardId- prefixes on each of the output messages: these tell us which ShardReader each event is being read by. The allocation is random because each event is partitioned based on its event ID, which as a UUID is essentially random.

Let’s see what happens if we stop the loop and then paste the same code in to start processing again:

Compare the partition keys (remember, these are the event IDs) to the previous run, and you’ll see that processing has restarted from the beginning of the shard. Our processing app is a goldfish; it has no memory of what events it has read on previous runs. Again, this is something that higher-level frameworks like the Kinesis Client Library handle: they allow you to “checkpoint” your progress against a stream by using persistent storage such as Amazon’s DynamoDB.

Now that we have a stream processing framework in place, all that is left is to check the available disk space as reported in each event and detect if the available space drops below 20%. We need a function that takes a reading from our agent and generates an incident as required. Here is a function that does exactly that:

Our function checks whether the proportion of available disk space is 20% or less, and if so, it returns the server’s hostname and the proportion of available disk space in a tuple. If the check passes, we return a None, indicating that there is no action to take. We also tolerate any KeyError by returning a None, in case other event types are added to this stream in the future.

Let’s test our new function in the Python interpreter with a valid event and an empty event:

Good: the first call to detect_incident returned a (None, None) tuple, while the second call successfully detected that the server with hostname ulp has only 15% disk space available.

That’s all the code that we need for our monitoring application. In the following listing, we consolidate everything into a single file, monitor.py.

Make the monitor.py file executable and run it:

Our monitoring application is now running, reading each event in both shards of our events stream and reporting back on our server’s disk usage. Unless you have a disk drive that’s as cluttered as mine, chances are that the disk usage alert isn’t firing. We can change that by temporarily creating an arbitrarily large file on our hard drive, using the fallocate command.

A quick filesystem check in my Vagrant virtual machine suggests that I have around 35 gigabytes available:

I created a temporary file sized at 30 gigabytes:

And then I switch back to the terminal running monitor.py and see this:

Great—the alert is firing! Every new filesystem metrics event being emitted by our agent.py is now triggering an alert in monitor.py. We can switch this off just as easily:

And switch back to our other terminal:

So, this completes our systems monitoring example. To avoid incurring further AWS costs, you must now delete the events stream from the Kinesis screen in your AWS UI, as in figure 4.12.

To recap: we have written a simple systems monitoring agent in Python that emits a steady stream of filesystem metrics onto an Amazon Kinesis stream. We have then written another Python application, again using boto, the Amazon Python SDK, which monitors the agent’s event stream in Kinesis, looking for hard drives filling up. Although our systems monitoring example is a rather narrow one, hopefully you can see how this could be extended to a more generalized monitoring approach.

Working with a two-shard Kinesis stream at a low level by using the AWS CLI and boto has given you a good handle on how Kinesis is designed. In the coming chapters, we will work with higher-level tools such as the KCL and Apache Spark Streaming, and your understanding of how Kinesis is designed at this lower level will stand us in good stead for this.

Summary