Redis to the Rescue: Upgrading Our Notification System
Current Dynamics and Challenges:
In our system, whenever a crash happens in any product, an event is triggered. The product team needs to be alerted about these events. For example, if 1,000 such events occur across all users within 5 minutes, an alert should be triggered. We also need to send daily or hourly alerts with information about the events received during those periods.
Our system used Java’s built-in data structures like TreeMap and LinkedList to handle this data and manage notifications, keeping only the latest 2 days of data. However, as the volume of events increased, we started experiencing Java heap memory issues.
We realized this setup isn’t scalable. Simply increasing Java heap space is not an ideal solution, nor is regularly restarting the application to prevent heap space from growing too much.
We need a more robust and scalable approach to handle the growing amount of data and ensure efficient alerting without running into memory issues.
Redis: A Promising Solution, but Challenges Arise
We thought Redis could be a great replacement since we already had data structured in TreeMap with key-value pairs, and all we needed was to move this data outside of internal storage. Redis seemed like an excellent choice — it was fast and reliable.
Initially, we seamlessly transitioned to Redis, creating key-value pairs and storing our data. Everything seemed to work well at first. However, we soon realized our mistake. Managing too many keys became overwhelming. The system became bloated, and accumulating keys based on time thresholds became a significant issue, especially under heavy production loads.
After facing these challenges, we had to reassess our approach. Ultimately, we decided to revert to the previous method of using Java’s internal memory, despite its shortcomings. While this solution wasn’t perfect, it at least functioned reliably. However, the process was slow and cumbersome, prompting us to reconsider our entire approach.
Redis Streams: A Potential Solution Emerges
As we delved into the issue, we stumbled upon Redis Streams — a feature designed precisely for tackling similar challenges. It sparked our interest, prompting us to go deeper into this new territory.
After some research, we realized that Redis Streams could indeed address our needs perfectly. However, we encountered a minor setback: Redis Streams lacked Time-To-Live (TTL) functionality. Despite this limitation, we remained undeterred. We believed that we could overcome this obstacle later. Our priority was to implement the solution effectively. We were confident that we could devise a workaround for the TTL issue once our system was up and running.
Concepts:
As we worked to improve our system, Redis Streams emerged as a promising solution. Here are some key concepts about Redis Streams that helped us build a more efficient system:
What is Redis Stream?
Redis Stream is a powerful data structure introduced in Redis 5.0 that allows you to manage and process streams of data efficiently. It is designed to handle high-throughput scenarios, making it ideal for applications like ours that require real-time data processing and alerting.
Redis Streams can be thought of as a message queue(Redis stream as an append-only data structure, however Redis streams are represented in memory, so we are also able to delete entries), where each entry consists of a unique ID and associated fields with values. It supports appending new entries, querying specific ranges of entries. Streams are inherently ordered, meaning the entries are stored and accessed in the order they were added.
Example:
Redis Streams store data as a series of entries in a key, each identified by a unique ID. The unique ID consists of two parts: a millisecond timestamp and a sequence number to ensure uniqueness even for multiple entries added in the same millisecond. The format of the unique ID is timestamp-sequence
"stream-key":
"1692632086370-0":
"name"
"John"
"1692632094485-0":
"name"
"Alice"XADD is the command used to add new entries to a Redis Stream. The time complexity of XADD is O(1), meaning it operates in constant time, making it very efficient even with large volumes of data.
XADD key [NOMKSTREAM] [<MAXLEN | MINID> [= | ~] threshold
[LIMIT count]] <* | id> field value [field value ...]
XADD stream-key * name John
XADD stream-key * name Alice
XADD stream-key * name BobEach XADD command adds a new entry to the stream stream-key. The * symbol tells Redis to generate a unique ID automatically.
[Official Documentation of XADD]
XRANGE:
XRANGE allows us to retrieve a range of entries from a Redis Stream. This command helps in querying specific segments of the stream. The time complexity of XRANGE is O(N), where N is the number of elements in the stream.
XRANGE key start end [COUNT count]
XRANGE somestream 1526985054069 1526985055069 // specifc entries
XRANGE somestream - + // all entries[Official Documentation of XRANGE]
XDEL:
Get Sunandanbose’s stories in your inbox
Join Medium for free to get updates from this writer.
XDEL is used to delete specific entries from a Redis Stream, and returns the number of entries deleted. This is useful for managing and maintaining the stream’s data. The time complexity of XDEL is O(1) for each entry deleted, making it efficient for stream maintenance.
XDEL key id [id ...]Let’s Discuss the Code
Our Event service is built as a Spring Boot application that exposes an API to fetch events. We use the Spring Redis starter dependencies for integrating with Redis, but you can also use other Redis clients. For instance, I’ve used Jedis before, and it’s an excellent option too.
By using Spring Boot and Spring Redis starter, we can easily manage our Redis connections and operations within our application, ensuring a smooth and efficient workflow. If you prefer, Jedis is also a great choice for interacting with Redis directly.
EventService:
@Service
public class EventService {
@Autowired
private StringRedisTemplate redisTemplate;
public void createEvent(Event event) {
ObjectRecord<String, String> record = StreamRecords.newRecord()
.ofObject(event.toString())
.withStreamKey(getStreamKey(event.getProduct()));
redisTemplate.opsForStream().add(record);
}
private String getStreamKey(String product) {
return "crash-events-" + product.toLowerCase();
}
}Building the Alert Service:
We successfully inserted events into our application. Now, let’s work on the Notification Service.
Note: It’s best to use the Observer Design Pattern here, as it will make the service more efficient.
In our implementation, I’ve added a scheduler that runs every 10 seconds. I chose to use a scheduler because the actual application has much more functionality and constraints. For simplicity, I’ve removed those additional constraints in this example.
AlertService:
@Configuration
@EnableScheduling
public class AlertService {
@Scheduled(fixedRate = 10000)
public void crashThresholdAlert() {
List<String> products = getProducts();
for (String product : products) {
String prefix = "alert.threshold." + product.toLowerCase() + ".";
Integer threshold = //Fetch Threshold limit from the config files.
int duration = //Fetch Duration time from the config files.
String unit = //Fetch time metric unit from the config files.
ChronoUnit chronoUnit = ChronoUnit.valueOf(unit.toUpperCase());
Integer eventsCount = eventService.getEventsCount(duration, chronoUnit, product);
print(product + " : " + eventsCount);
synchronized (pauseAlertOnProducts) {
print("ALERT stopped for products: " + pauseAlertOnProducts);
if (eventsCount >= threshold && !stopAlertOnProducts.contains(product)) {
print("ALERT: " + product + " has crossed the threshold of " + threshold + " in last " + duration + " " + unit);
// TODO: Send actual alert instead of printing
pauseAlertOnProducts.add(product);
scheduler.schedule(() -> {
synchronized (pauseAlertOnProducts) {
print("Removing alert for " + product);
pauseAlertOnProducts.remove(product);
}
}, duration, TimeUnit.valueOf(unit));
}
}
}
}
}The crashThresholdAlert method checks if any product has exceeded its crash threshold within a specified time frame. For each product, it fetches the alert threshold, duration, and unit from a property service and retrieves the event count. If the count exceeds the threshold and alert hasn't been paused for the product, it sends an alert message and haults alert for the product for a given time duration.
EventService:
@Service
public class EventService {
private StringRedisTemplate redisTemplate;
public Integer getEventsCount(Integer time, ChronoUnit unit, String product) {
Instant now = Instant.now();
String epochTime = String.valueOf(now.minus(time, unit).toEpochMilli());
return getEventsCount(epochTime, "+", product);
}
public Integer getEventsCount(String from, String now, String product) {
Range<String> range = Range.closed(from, now);
List<MapRecord<String, Object, Object>> records =
redisTemplate.opsForStream().range(getStreamKey(product), range);
return records != null ? records.size() : 0;
}
}I’m only explaining the CrashThreshold here. Other types of alerts will work in a similar way to this threshold alert.
Implementing TTL Functionality in Redis Streams
Now, let’s address how to implement TTL (Time-To-Live) functionality in Redis Streams.
We discovered a useful command called XTRIM. Here's a bit about how it works:
- XTRIM: This command trims the stream to keep it within specified limits.
- MAXLEN: Trims the stream to a maximum length by removing the oldest entries. For example,
XTRIM mystream MAXLEN 1000keeps only the latest 1000 entries. - MINID: Trims entries with IDs less than a specified minimum ID. For example,
XTRIM mystream MINID 1653660588123-0removes all entries with IDs less than1653660588123-0.
Since Redis streams store data sequentially by timestamp, we can use XTRIM to delete stream IDs below a certain timestamp, effectively implementing a TTL-like functionality. The catch is that this needs to be run manually by us, unlike TTL which Redis applies automatically.
Remove Id from Stream:
private void removeExpiredEvents(String product) {
executorService.submit(() -> {
try {
Instant now = Instant.now();
String epochTime = String.valueOf(now.minus(2, ChronoUnit.DAYS).toEpochMilli());
List<RecordId> recordIds = redisTemplate.opsForStream().range(getStreamKey(product), Range.closed("-", epochTime))
.stream()
.map(MapRecord::getId)
.collect(Collectors.toList());
logger.info("Removing expired events: " + recordIds.size());
if (!recordIds.isEmpty()) {
redisTemplate.opsForStream().delete(getStreamKey(product), recordIds.toArray(new RecordId[0]));
}
} catch (Exception e) {
e.printStackTrace();
}
});
executorService.shutdown();
}After setting everything up, we found that the Spring Redis dependency doesn’t support the MINID functionality with XTRIM. It only allows setting MAXLEN.
To work around this, we extracted all the record IDs using XRANGE and then deleted the IDs using XDEL. While XTRIM with MINID would have been ideal, this approach also works well for our needs.
Note: If you’re using Jedis, you can use XTRIM MINID directly.
Conclusion: Reflecting on the Alert and Notification Service Journey
With the completion of the Alert and Notification Service, it feels like the solution was simple. However, the journey of discovering this solution, going back and forth between learning and implementing, was truly enjoyable.
Throughout this process, I’ve learned a lot, and I hope to continue encountering such challenges in the future. It’s these challenges that push us to grow and improve.
If you have any feedback or would like to connect, feel free to reach out to me on LinkedIn. I’m always open to learning and collaborating with others.
Thank you for joining me on this journey!
