Summary

This article explains the mechanics of how an instance of the RStreams Bus works by looking at the operations that cause data to flow through the bus.

What is the “Bus”?

It’s an AWS-native implementation of an event streaming and general purpose messaging system built using AWS services.

Event Streaming

The bus is a way for a client application (usually a bot) to…

• Push vast numbers of data events into a named queue quickly, typically using the Node SDK, and keep them in order
• Retrieve vast numbers of data events from a named queue quickly, retrieving them in order so they can be processed in order
• Read from one RStreams queue while simultaneously processing data already retrieved and optionally pushing that data to another queue(s)

General Purpose Messaging

The following are the messaging models supported by the RStreams Bus:

• Push events to consumer : Bot will be invoked when events are pushed into a given queue and to continue pulling events from the queue until there are no more events or the Bot shuts down and then restarted and invoked again if/when there are more events
• Consumer pulls events : Bot will be invoked on a cron and pull events from a given queue
• Consumer reduces events : Bot will read events from a queue and aggregate/reduce them in some manner, turning N events into 1 event and pushing the new event to another queue
• Producer pushes to a single queue - 1 to 1 : Bot pushes data events to a single queue
• Producer pushes to multiple queues - 1 to many : Bot reads data events from a queue and then pushes data events to multiple queues
• Multiple producers push to a single queue - many to 1 : Multiple bots read data from various queues and write data events to a single queue

The Bus Architecture Diagram

Here’s a diagram of an RStreams bus.

The Bus Architecture - 1/2 RStreams Bus is DynamoDB and S3

The RStreams Bus is mostly bookkeeping. The three DynamoDB tables listed above - Bot Table, Event Table, Stream Table, are the workhorse of RStreams and where the vast majority of its functionality stems from. The Event table stores a definition of every queue in the RStreams instance. The Bot Table stores a definition of every bot in the RStreams instance.

A bot is code that is registered with RStreams to read from a queue or a “system”. A “system” is a special producer of data that can have a different icon in Botmon and can track checksums. For example, Mongo might be a system in Botmon. Bots can read directly from Mongo since Mongo has been registered as a supported “system.” Additional “systems” might include ElasticSearch, MySQL, Oracle, Postgres, Redshift, MS Sql Server, etc.

The Stream table stores every event in every queue, storing either the actual payload of events gzipped or an S3 path to the file that contains the payload of events. Each record may contain multiple contiguous gzipped events and each S3 file referenced may contain multiple contiguous events.

 1"checkpoints": {
 2  "read": {
 3    "system:mongo-weather": {
 4      "checkpoint": "z/2021/06/11/14/42/01/002:MT:826...0004", // eid of last read 
 5                                                               // event
 6      "ended_timestamp": 1623422523673, // millisecond timestamp of last read event
 7                                        // for last chunk 
 8      "records": 1, // number of records for the last chunk
 9      "source_timestamp": 1623422523673, // millisecond event source timestamp of
10                                         // last read event for last chunk
11      "started_timestamp": 1623422523673 // millisecond timestamp of first read
12                                         // event for last chunk
13    }
14  },
15  "write": {
16    "queue:weather": {
17      "checkpoint": "z/2021/06/11/14/40/1623422416226-0000000", // eid of last
18                                                                // written event
19      "ended_timestamp": 1623422416226,
20      "records": 1,
21      "source_timestamp": 1623422414001,
22      "started_timestamp": 1623422416880
23    }
24  }
25}

1{ 
2  "missing":{"M":{"count":{"N":"0"}}},
3  "correct":{"M":{"count":{"N":"638"}}},
4  "incorrect":{"M":{"count":{"N":"0"}}},
5  "percent":{"S":"100.00"}
6}

 1"instances": {
 2  "0": {
 3    "invokeTime": 1623432243253, // millisecond timestamp of invocation
 4    "token": 1623432242544, // invocation token
 5    "maxDuration": 899999, // number of milliseconds the bot is allowed to
 6                           // run (timeout duration)
 7    "requestId": "723d9f0c-7a66-4dff-9220-f2da19e3fd51", // id of the current
 8                                                         // invocation
 9    "startTime": 1623432243415, // millisecond timestamp of when the bot
10                                // started running
11    "status": "complete", // status of last invocation (complete|error)
12}}

The Bus Architecture - Writing, the Other Half of the Diagram

Everything to the left of the three DynamoDB tables in the diagram involves getting data onto the bus, which means into a queue, fast while maintaining total/partial ordering, as necessary.

Every time an engineer uses the RStreams Node SDK to write data into a queue, even from a bot that is reading from an upstream queue that then turns around and writes to a downstream queue, what’s really happening is that SDK is writing to Kinesis, in one of three ways.

Direct to Kinesis

The SDK is simply writing the events directly to Kinesis with the data event payload gzipped and embedded directly on the event sent to Kinesis. This is the fastest method with the lowest latency.

Direct to Kinesis with S3 Payloads

The SDK is writing a file of one or more actual data events to S3 and then writing a Kinesis event that references back S3 file with the actual payload. This can introduce some downstream latency when reading the S3 files. The SDK will automatically choose this method for a single event when it’s too big for Kinesis to accept - bigger than 1mb. Engineers may choose this method via code when slightly increased read latency warrants making ingestion faster due to the massive number of expected events coming continuously. The read penalty payed for grabbing an S3 file is pretty minimal if the S3 file has many, many events in it. Read performance degrades only when downstream processes have to read many S3 files to work through a relatively small number of events. Read performance actually improves when there is a small number of very large S3 event files. However, the stars have to align correctly for this to make sense. How often is so much data delivered all at once that stuffing a large number of events into an S3 file doesn’t introduce traditional batch-like waiting and latencies? It happens but it doesn’t apply to all use cases.

Kinesis by way of Kinesis Firehose

Sometimes slightly higher ingestion latency, on the order of a 90 seconds, is OK and there is so much data coming through a single bot that batching up over this period makes sense. Firehose makes the most sense, however, when multiple bots are feeding data at massive scale into as single queue. When that’s the case, Kinesis Firehose is the way to go. Kinesis Firehose writes its events directly to S3 for us where RStreams uses the Direct to Kinesis with S3 Payloads described above to drop a single event into Kinesis that references the S3 file of events.

Data Movement through the Bus

Once data hits a queue, what happens? How does it move? How does the bot get the data from the upstream queue? How does data flow to the downstream queue?

Bots register themselves with RStreams, indicating whether RStreams should push events to the bot when invoked or whether the bot will pull events from the upstream queue manually itself on invocation. During registration, bots can also designate micro-batch sizes and other config about how often and when to invoke bots as events arrive in the upstream queue, perhaps favoring more invocations and less latency at the cost of economies of scale of working on a micro-batch of events in a single invocation. Bots can also register and tell RStreams that they are to be invoked on a schedule dictated by the cron statement that accompanies the registration.

When bots are invoked and have data pushed to them or they pull data from the upstream queue, it’s up to the bot what happens next. The bot may save the data into a database or external system and that’s it, the end of the stream. The bot may choose to use the SDK to write to another queue after changing the data it received from the upstream queue somehow. Well, using the SDK to write to a queue is “writing to the bus” which we just went over with the three approaches that use Kinesis. The data gets ingested through kinesis and makes its way into the Stream table. Writing to the Stream DynamoDB table means “new events arrived in the queue.” RStreams will detect this and cause any will at the appropriate time, based on bot config, invoke any registered bots since there are new events in the queue.

 1{
 2  "id": "id_of_bot",
 3  "event": "id_of_the_queue",
 4  "event_source_timestamp": 1614185828000, // timestamp of the initial event
 5  "timestamp": 1614185828000, // timestamp of this event
 6  "correlation_id": { // information to track what created this event
 7    "source": "previous_queue_id",
 8    "start": "z/2021/02/24/16/50/1614185855405-0000000", // first event id
 9                                                         // included
10    "end": "", // last event id included (optional)
11    "units": 1 // number of records included (optional, defaults to 1)
12  },
13  "eid": "z/2021/02/24/16/57/1614185855405-0000000", // Unique ID for event
14  "payload": {} // your custom data
15}