2

Say that you have a monitoring application that reports downtimes for a large number of remote systems (think IOT). A monitoring daemon polls the remote system and reports that status (on/off) via an event stream to an event processor. The aim of the event processor is to determine when and for how long the system was down and persist that information for end-users to access it. Think of it something like this:

enter image description here

The downtime table in the storage looks something like

uuid remote_system_id start end
asdf1 qwer2 2022-10-10 01:00:00 2022-10-10 01:05:00
asdf2 qwer2 2022-10-10 02:00:00 null

In the happy case the event processor just needs to create a new row when an "off" event comes in and store the timestamp the start column. Then, when an "on" event comes later, we just need to update the end column of the same row with the new timestamp. If the state polled has not changed, then we don't need to update the storage.

As long as event come in in order there's no problem, but for different reasons we can have out of order events. For example consider the chain of events

event number remote_system_id state timestamp
1 qwer2 off 1
2 qwer2 on 2
3 qwer2 on 4
4 qwer2 off 3

As you can see the third event has the same state as the second and is thus dropped, which results in one row for the downtime from t=1 to t=2 and one row for an ongoing downtime started at t=3. Has the events come in in order, then also the second row out have an end time set to t=4.

My question is then how to redesign the system in order to have an up to date view of the downtimes even with late and out of order arrival of events. One idea would be to keep track of the latest timestamp in a separate cache so that we can detect out of order events and trigger a reprocessing of the event stream. This means that:

  1. We need to persist the event stream indefinitely
  2. We need a way to pause the live processing during reprocessing and then resume it
  3. The storage needs to be able to handle a lot of sudden updates when rewriting the history

This seems like a reasonably standard problem and I wonder if there are any best practices to learn from.

1 Answer 1

3

tldr;

how to redesign the system in order to have an up to date view of the downtimes even with late and out of order arrival of events.

A simple solution to this is to store all the messages in an SQL database and use a View or a query that will always show you the current state. This is a simple gap and island analysis query that can be executed with a high degree of efficiency in an SQL database with the right indexing strategy.

We need a way to pause the live processing during reprocessing and then resume it

Usually you would not try to pause the live processing, that can easily raise chase conditions, it is better to have two flows for processing, one for the live processing and one for the stale or offline messages. These might flow through some of the same logic but are likely to have a few additional criteria that does not apply to live data.

That being said, to pause the processing should be as simple as stopping the processing services, you could however do this with some sort of system level flag or setting. Keep in mind that while the system is stopped, your ingestion endpoint will continue to queue up messages, on most platforms there are fixed limits to size of the pool before messages can either no longer be received, or the oldest messages start popping off the stack.

  • It is a good idea to implement a cache to determine the difference between what is live vs what is stale, however if your processing logic is distributed (scale out), then you will benefit from a distributed cache implementation.

The storage needs to be able to handle a lot of sudden updates when rewriting the history

If you are using a view, then there is no rewriting. in any case for a single message, at most that message will only ever affect 2 rows, the row before and the row after that point in time, never any more.

  • You should not be trying to maintain a specific sequence of data, that is the job of an indexer if you need one and databases have strategies for rebuilding indexes
  • Look into Materialized Views as an example of how databases can handle a lot of sudden updates. Some RDBMS have native support for MVs but its a simple enough concept to implement from first principals in any database.

Update - RE: Reprocessing the stream

I would never suggest re-processing the stream, but have done so once or twice after data down-stream was so corrupted that it needed to be re-processed. The other scenario where being able to re-process the stream is useful is for resetting test and development environments.

  • If re-processing to simulate a live test, try to make sure you re-transmit using the same relative time factor. Don't try to normalise the messages or dump them all at once, to effectively test for scale and spikes use a strategy that send the data at a randomised interval or an interval based on the original message timings.

Replay of the stream is not recommended as a solution for eventual consistency, for each stale message only 2 existing records can be affected, you might be re-processing millions of rows for the sake of 2 rows. If you are paying for consumption thats a lot of CPU ticks and bandwidth for a new message.

To support re-play of the stream it is important to identify the fault boundaries. If the trigger to re-play is a stale message, identify the oldest possible stale message that you will allow into the system, this is your a time-based fault boundary.

  • It does not normally make good business sense to store all telemetry in it's raw state in a single flat repository, even if you need to replay many years for regulatory purposes, there is a point where really need to archive the older messages to a cold storage media.
  • In many cases the need to re-play messages can be mitigated through other design strategies. To reply on re-play is the laziest and generally most expensive option.

Many cloud message streaming service buses will have a degree of storage, the most that I've seen that is practical is 7 days, though Azure Event Hubs and offers a standard configuration with longer term storage services, I'm sure AWS would have something similar. If you need to re-play longer periods than your streaming platform supports, then you will need to persist these messages to a storage medium of some type. SQL is an option if you want to query the data or quickly re-organise the data, but can be expensive, blob storage is the cheapest form but is usually more appropriate for data you don't need to analyse at rest. Different vendors will have different products and services to assist you in this space.

Whatever you choose, it is best practice to setup an expiry policy to drop (or archive) data that goes outside of the fault boundary. You could also partition the messages either into multiples of the fault boundaries or using a natural partition key from within the message metadata, like the origin device or the system or site being monitored.

If you are going to re-process, then you need to make your down-stream storage and processing resilient to the fact that duplicate messages are coming through. That might involve clearing out storage first or making your processing streams only write or send notifications if those actions had not already been performed. Eitherway, the very nature of trying to support re-write will add a processing overhead that will affect every write operation.

  • If your processing raises alert notifications, you can't really take them back if the new stream doesn't have the same alerts, you also should not be bothering end users with notifications they have already received.

Original Response:

Many high availability solutions, like IoT that involve distributed real-time processing will have to deal with this issue, what you are describing is Eventual Consistency.

Even if you have highly efficient ingestion and processing capabilities, there are many issues that can lead to out of sequence receipt and processing of data. For high availability there may be multiple physical endpoints to receive messages, for high throughput you might implement parallel or batch processing or your message source might also implement batching or have an offline capability where telemetry is stored and sent when the connection is re-established.

In all systems there are ways to enforce sequential consistency of the event streams, but doing so reduces availability and response of the system. It is generally easier to embrace eventual consistency than to try and prevent it.

If the state polled has not changed, then we don't need to update the storage.

This is the first issue, for eventual consistency to work we cannot discard any messages, not until you can ensure that the data has reached a state that is consistent with the event source (or physical world). This is problematic to implement at the receiver end, you could have some elaborate checksums to do this verification but if you need to know when the data is consistent it is more common to specify an overall time tolerance of the system which can still lead to data inconsistencies if messages are received outside of this tolerance.

NOTE: When we are dealing with messages that are potentially received out of sequence it is important that the message contains the timestamp and that you do not rely on the time of receipt or processing to determine the actual point in time that the message describes... This is a crucial distinction that needs to me made.

In Systems that report telemetry on a polled or timed basis, rather than on change of value "events" it is usually more practical to aggregate the messages closer to the source and only send on change, the polling process can maintain a cache of the "current" state and only raise the message through the system if or when the state changes, thereby changing the message stream into an event stream.

  • This is simple enough to implement if there is a dedicated resource that performs the polling, if it is a distributed process, then there are standard products and implementations of distributed cache, Redis is my personal favourite.

One idea would be to keep track of the latest timestamp in a separate cache so that we can detect out of order events and trigger a reprocessing of the event stream.

Your analysis here is heading in the right direction, but caching introduces a new problem, the messages that arrive out of sequence will still be lost. Caching will ensure that the "current" state is correct, but doesn't help with the historic data stream. In fact, due to the nature of many event processing systems, even if we cache at the message source, the other steps along the chain can still end up processing the events out of sequence. Caching at any point in the chain can reduce the risk of messages being processed out of sequence, but each aggregation or processing step in the chain also increases this risk, in many cases we still need to account for eventual consistency!

Ultimately the endpoint, in this case your database, needs to account for eventual consistency and so it, or a tightly coupled cache before it, needs to keep track of all messages so that it can detect and react to messages arriving out of sequence.

Traditionally this is the difference between your transactional/operational database and your Data Warehouse. The operational database collects all messages, then at regular intervals the data is aggregated and transferred to the DW. At a smaller scale, you can do this between tables in the same database, one table recording all the history of messages, one table recording the aggregate of the messages, in this case the downtime events.

Lets call the history table system_history, it logs all messages that are received.
I have added in message sequence only for illustration purposes

event number remote_system_id state timestamp message sequence
1 qwer2 off 2022-10-10 01:00:00 1
2 qwer2 on 2022-10-10 01:05:00 2

The process that writes to the system_history table, also writes to the downtime table. In an SQL based system you might even choose to use an AFTER INSERT TRIGGER to update the downtime table. This is generally the best approach to provide transactional consistency if you are using the database as the cache.

At this point, downtime is correct, according to the information that is available:

uuid remote_system_id start end
asdf1 qwer2 2022-10-10 01:00:00 2022-10-10 01:05:00

The next message received is sequence number 4, which states that the system is on, this is appended into the system_history table, but does not yet affect the downtime table because the state has not yet changed:

event number remote_system_id state timestamp
1 qwer2 off 2022-10-10 01:00:00
2 qwer2 on 2022-10-10 01:05:00
3 qwer2 on 2022-10-10 02:20:00

What is important here to note is the source of truth. In this current model the history table is the source of truth, the aggregate table is just that, an arbitrary point in time aggragation that is deliberately omitting data.

To determine the "current" state of the system, we need to query the history table in the sequence of the message timestamps, not the sequence that the messages were received. However, in this case the concept of "current" has a different reference point, we do not use the current time or "now" we use the timestamp of the message that we are recieving. So we are actually looking for the state of the system at a specific point in time

That might look like this:

DECLARE @sysId CHAR(5) = 'qwer2';
DECLARE @msgTime DATETIME = '2022-10-10 02:20:00';
SELECT TOP 1 state 
FROM system_history 
WHERE remote_system_id = @sysId
  AND timestamp < @msgTime
ORDER BY timestamp DESC

NOTE: I'm using less than so that this query can be executed after the insert of the new message has occurred. Adjust this comparison logic to suit your order of operations
The order is significant, we want the last record so sort descending.

If the state does not match the messasge, then we know that a change of state has occurred and the downtime may need to be updated. If there is no change of state, then no change may need to occur

  • In reality there is a little bit more to this, if we receive multiple off messages out of sync then the downtime.start may need to be updated. But this is already a massive post, I only hope to explain the concept ;)

The next message received is sequence number 3, which states that the system is off, this is appended into the system_history table:

event number remote_system_id state timestamp
1 qwer2 off 2022-10-10 01:00:00
2 qwer2 on 2022-10-10 01:05:00
3 qwer2 on 2022-10-10 02:20:00
4 qwer2 off 2022-10-10 02:00:00

At this point in time (2022-10-10 02:00:00), the previous state was on, from message 2, NOT message 4. So technically we would be correct in injecting this row into the downtime table:

uuid remote_system_id start end
asdf2 qwer2 2022-10-10 02:00:00 null

But we can do a quick check to determine if there is a corresponding change of state in the future:

DECLARE @sysId CHAR(5) = 'qwer2';
DECLARE @msgTime DATETIME = '2022-10-10 02:00:00';
DECLARE @state CHAR(3) = 'off';
SELECT TOP 1 state 
FROM system_history 
WHERE remote_system_id = @sysId
  AND timestamp > @msgTime
  AND state <> @state
ORDER BY timestamp ASC

In this case we order in ascending order to get the first record with a changed state.

In this case we find that there is an on state in the future, so we can update the downtime record, resulting in this:

uuid remote_system_id start end
asdf1 qwer2 2022-10-10 01:00:00 2022-10-10 01:05:00
asdf2 qwer2 2022-10-10 02:00:00 2022-10-10 02:20:00

This is a start but over time the history table will grow and performance of the queries for the "current" values will significantly degrade. So from a pure SQL point of view there are other efficiencies that can be gained. If the read traffic on the downtime table is minimal, you could replace this table with a view, that would negate the need to maintain the table as part of the insert operation at all.

Another possibility is that if the reads are frequent, but do not need to have information that is accurate to exactly the point in time of the request, then you could materialize the view, so periodically (or on demand) store the results of the view in a table.

  • Materialized Views are the most basic form of Data Warehousing
  • Even for a table based aggregate approach, it is a good idea to define a view that can be used to reconcile or rebuild the aggregate table.

There is still a problem we have not yet discussed, what exactly does your event processor do?

If your event processor does anything like raising notifications to other systems or to people directly, like SMS, Email, post on twitter or initiate any other communication streams, then the concept of caching becomes important again, but for a slightly different reason.

In most cases for notifications into the real-world, or atleast to the next system it is only relevant to raise "current" events so that the external system (or user) can respond.

So, for the case of our message 4 (on) being processed before 3 (off) it is not normally helpful to the external system to be notified that the system was off 20 minutes ago, because it is now on. You might want to raise a specific message in this case, but it would normally be a lower priority message because the system has already recovered, where as an off notification would conceivably be more urgent for the recipient to react to.

You can still use the database as the cache, and follow the same process, this time adding a third table, lets call it system_state as it represents the current state of the system:

remote_system_id state timestamp
qwer2 on 2022-10-10 02:20:00

Let us assume we've rolled back to before processing message 3. For the purpose of raising the external notification, we only want to know the 'Current' state so we check two things against the current record in the system_state table:

  1. Is this new message more recent than the current record?
  2. If the new message is newer, then has the state changed? (Does a notification need to be sent?)

At the end of the comparison, even if the state has not changed, if the message is newer, then the current timestamp needs to be reset to match the new message.

You can easily write this into a single atomic SQL statement without too much issue. The point is that there are times when you need a cache, and times when you need access to the underlying history or source of absolute truth and your current solution involves a database that can be engineered to provide both for you and is one of the most stable mechanisms that can be used to ensure transactional consistency, eventually :)

5
  • Thanks for the very extensive answer. Your main assumption is that I can keep the history in the same database and use that to recalculate the start and end times. We're considering this to cover the case of out of order on a short timespan, but that is not enough for two reasons:
    – bangnab
    Oct 21, 2022 at 7:09
  • depending on the volume of data, its usually not practical or even economical to keep the raw telemetry in the same database as your line of business data, I was just trying to convey the concept, sow the seed. You can use a traditional SQL database for this purpose but when you hit certain scales of data you will have to start to apply more and more data warehousing techniques. I use an approach similar to this in a production IoT solution that tolerates offline telemetry submitted up to 3 days late. Oct 21, 2022 at 10:01
  • Data older than that is received through a different channel. Where possible I try to avoid re-processing, in almost all scenarios data out of sync does not raise further events. I'll update the post with a response to be able to handle a lot of sudden updates Oct 21, 2022 at 10:04
  • I really appreciate the effort, but in my case we really have a lot of data and we can't keep everything in an SQL database. Therefore I need to sometimes reprocess the stream indefinitely long back in time (for a given remote_system_id) and need a strategy to orchestrate that. In the end I figured that as long as I'm able to retrieve from the stream (or a copy of it) only the messages just preceding and following the late messages (for the same remote_system_id), then I should be able to correctly update the view.
    – bangnab
    Oct 24, 2022 at 13:04
  • @bangnab how long back are we talking? I'm ingesting 14M rows of telemetry per week and running no problems in SQL Server running on Azure. You need to partition the storage, so pick the outside period of the offline/stale data but you would never dream of re-processing the stream, its just too intensive. That's like tape backups, we've evolved from there. Oct 25, 2022 at 12:05

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.