Tiered Storage Support for RabbitMQ Streams

[SimonUnge]

Overview

This design introduces tiered storage support in Osiris, allowing stream segment files and index files to be offloaded from local disk to alternative storage backends. It preserves the existing write-path semantics, ensuring that writes continue to operate as before. The read path is extended to support retrieving segments of data from multiple "sources," including both local disk and cold storage. The system is designed with a pluggable architecture, enabling integration with arbitrary cold storage backends.

Goals

• Enable storage tiering for stream segment and index files
• Minimize changes to existing Osiris segment file management, writing and reading logic
• Provide native support for tiered storage for local mounted drives
• Provide pluggable support for tiered storage for other storage backends (cloud storage, future extensions)
• Maintain a manifest that tracks segment (and index) file locations and tiers as well as metadata for file size and age, in a manifest file per stream
• Support more than one level of tiered storage via a simple tier index
• Writes and segment data replication between members remain unchanged

Terminology

• Drive: A pluggable module implementing storage access. Could represent local disk, cloud storage, mounted NAS, EBS, or any cold storage layers.
• Manifest: A binary Erlang term describing which drive each segment is located on, along with metadata such as offset, size, timestamp, and tier level.

Architecture

Manifest File

Each stream has a corresponding manifest file on disk, encoded as an Erlang term. It is stored and updated locally by the node and can be replicated using RabbitMQ’s existing distribution mechanisms.
Example structure:

#{
  <<"segment-00001">> =>
   #{offset => 0,
     drive => local,
     path => "/var/lib/rabbitmq/segments/segment-00001",
     size => 52428800,  %% in bytes
     timestamp => 1712500000, %% epoch seconds
     tier => 0},

  <<"segment-00002">> =>
   #{offset => 50000,
     drive => s3,
     path => "s3://rabbitmq-bucket/streams/segment-00002",
     size => 52428800,
     timestamp => 1712503600,
     tier => 1}
}.

Manifest Module

A new osiris_manifest module will:

• Load and cache the manifest on stream startup?
• Allow efficient lookups of segment metadata by name
• Provide functions to update metadata entries
• Serialize and flush the manifest to disk

API Sketch:

-spec get(SegmentId) -> SegmentMeta.
-spec update(SegmentId, SegmentMeta) -> ok.
-spec delete(SegmentId) -> ok.
-spec all() -> SegmentMeta.

Drive Abstraction

Drives implement the behavior for reading segments, given a manifest entry. Example behaviors:

-callback open(File, Opts) -> {ok, io_device()} | {error, term()}.
-callback read_file_info(File) -> {ok, FileInfo} | {error, Reason}.

This approach will allow Osiris to use the file module just like it does today, regardless of where the io_device fetches the data from.

Storage Policy

• New messages always write to Tier 0.
• Older, less-accessed segments migrate to higher tiers based on retention config
  • This can start out using the current max-age and max-length-bytes but move the files one tier up instead of delete the segment files.
• Reads check the manifest for the segment’s tier and retrieve data accordingly.

Integration with Osiris

• When Osiris opens a segment, it asks the manifest for the path and drive type.
• Osiris uses the drive module to open the file and read from it.
• Writing remains unchanged: writes go to local disk, and segments may be offloaded to higher tier drives based on retention settings.

Replication

• The manifest file is only updated by the stream leader for the stream
• Manifests are replicated (how?) to each member

Benefits

• Minimal disruption to existing Osiris write logic
• Extensible to any number of tier levels
• No impact on publishers; only affects stream consumers
• Supports heterogeneous clusters with multiple storage strategies
• Flexible manifest metadata including offset, size, timestamp, and tier level

Open Questions

• Instead of reading from a higher tier, we could ‘move’ segment files back to tier 0 storage when they are read from

Future Work

• UI/CLI support for viewing and managing segment tiers

[kjnilsson]

A couple of questions around the manifest?

Should it contain all (including the currently appending) segments or just those that have been moved from tier 0?

I think manifest replication could be done in the replica overview call, it can just grap the whole manifest and overwrite it's local one.

How would the manifest be maintained if there are more than 2 tiers, e.g. local, hdd, s3 where each tier moves segments based on some limits.

[kjnilsson]

Also I think we should consider local index caching of remote segments as operations such as timestamp scanning can be very chatty.

[SimonUnge]

Should it contain all (including the currently appending) segments or just those that have been moved from tier 0?

My thinking was that it should, but in my mind you'd have very few local segments. That said, you could have loads so perhaps it is redundant and even unnecessary to keep track of the tier 0 segments.

I think manifest replication could be done in the replica overview call, it can just grap the whole manifest and overwrite it's local one.

Perfect!

How would the manifest be maintained if there are more than 2 tiers, e.g. local, hdd, s3 where each tier moves segments based on some limits.

I was thinking the manifest only points out the tier level, type and path, i.e only one manifest file. Do you think that is a bad idea? We could also skip the complexity of more than 2 levels to start with, as long as we dont lock ourselves into a corner.

Also I think we should consider local index caching of remote segments as operations such as timestamp scanning can be very chatty.

Right, I thought the manifest file could contain some index related information (timestamp, offset perhaps) as part of each segment file, but to store it in memory as well is your thinking?

[the-mikedavis]

We (@SimonUnge & @kjnilsson and I) chatted earlier today a bunch about this and designing for an S3 remote tier. Here are some notes on what we talked about and some discussion about designs...

Apache Kafka tiered storage...

Kafka has a tiered storage feature which is very well documented in KIP-405.

KIP-405 talks about tiered storage in terms of two tiers: local and remote. The remote tier is assumed to have write-once-read-many semantics with S3 and HDFS as examples.

The design's main component is the "RemoteLogManager" (RLM) which creates tasks for leader and follower replicas. The leader is in charge of uploading any segment files which have rolled over. Followers keep track of metadata changes and can serve data from the remote tier. Also notable, the RLM keeps a bounded cache (possibly LRU) of index files.

Their design separates the storage interface and metadata management interface. There's extra discussion of this here. Ultimately the metadata part is pluggable and the default implementation uses an internal topic. This part is in charge of storing and fetching metadata about segments on the remote tier (and caching that info to make it relatively cheap).

---

Moving data between brokers cheaply...

We're looking for a design for this feature that lets us store minimal significant data on any individual broker. Keeping around extra data to make tip reads fast, for example, is fine. But we want to upload to the remote tier very aggressively. The reason we want this is that it lets us move replicas between brokers cheaply and restart faster. For one workflow, if you have a three node cluster with lots of streams on all three nodes and expand to a larger 7 node cluster, we want to be able to re-balance the stream replicas so that they use the cluster resources move evenly.

Moving replicas between brokers is expensive with regular on-disk streams because you need to replicate the entire stream. For a remote tier with write-once-read-many semantics, though, you only have to replicate any data which is not yet in the remote tier to new replicas, so it becomes feasible and relatively cheap.

With a design like KIP-405 above, if we uploaded to the remote tier after rolling over a segment then we're dealing with a rough upper bound of the max segment size of data which needs to be replicated to new replicas. In a high-scale workload like moving 1000 streams this is a rough upper bound of 500 GB with the current 500 MB default max segment size - too much data.

Reducing the segment size reduces the upper bound on data transfer needed to move streams. The current max segment file size (500 MB) is fairly large and it could probably be reduced without severely impacting throughput. A scaling problem with decreasing max segment file size is the number of index files. An order-of-magnitude decrease in max segment file size can increase the number of index files written for the same amount of data by an order of magnitude. At a high publish rate it would become expensive to track these index files in any sort of manifest (like described in the issue description) because the index files would be produced at a high frequency.

---

Remote tier manifest design (mainly designing for S3)...

We need a manifest (as described in the issue description) tracking which segment/index files are in the remote tier. Ideally this should be cheap to truncate at the beginning for retention and append at the end for new segments/indexes. But it also needs the ability to find the correct segment/index for an arbitrary offset or timestamp fairly quickly. The latency of an attach at an arbitrary point can be somewhat high but it shouldn't mean a linear scan of the remote tier: we should be able to attach with a low expected number of reads from the remote tier. Side note: tip reads (the common case) should be served by the local tier so they shouldn't need the manifest and should be similarly as fast as today.

One way to design the manifest which would allow for a smaller max segment file size could be to sparsely track index files. A manifest could be updated once every X segment file roll-overs to track that Xth segment and metadata like the amount of bytes added since the last update. Then to find an exact segment file later, you would find approximately the correct index file given metadata in the manifest and then move to the next index file until the correct index file is found.

To make this possible we need to add information to the index file. Index file entries contain the offset of each chunk written but not the number of records, so currently, finding the offset of the next segment/index file in the log means looking into the end of the segment file. We can add a new version of index files which contains this information.

Updating the manifest for a subset of index files allows us to keep the update frequency low in higher throughput scenarios while still allowing a relatively low expected number of reads from the remote tier to find an exact position in a segment file. To make repeated attaches around the same areas cheaper we can keep LRU caches of parts of the manifest and index files which were downloaded for scanning.

We can also try to break up the manifest into a hierarchy. A smaller file could track metadata about a max number of index files (or a relatively large segment-file byte count like 5 GB). This smaller metadata file would then roll over like a segment file and be added to the overall manifest. So the overall manifest for a stream would update very infrequently and could be cached longer. Lookup would mean first downloading and caching the overall manifest and finding the correct metadata file, then downloading that to find the approximately correct index file, and following next-offset pointers to find the fully correct index file.

I think the manifest could still use some extra thinking. Maybe we can make a disk-based data structure that can reach our goals here without being too complicated. We want the manifest to be stored on the remote tier so that it doesn't have a replication cost on the writer when adding new reader replicas.

---

Interface design for osiris...

The remote tier manifest described above would take the place of places in osiris_log where we list the stream's data directory and operate on the sorted index and segment file names. Ideally we want flexibility in the manifest design so that we can specialize it for S3, but it should also be possible to write a manifest that works well locally (possibly with a different design) or on a different "cold" (rather than remote) tier like an attached HDD on each replica.

To make this pluggable we should factor out parts of osiris_log which hard-code list_dir operations to instead run callbacks of a behaviour for operations like finding the correct index/segment file for an offset spec, determining where in the stream replication should start and (maybe) evaluating retention. A default implementation would use the current logic and an S3 plugin could check the remote manifest. The plugins will also need a way to be notified of events like a stream being created or deleted, a segment file being rolled over, and the commit offset moving up. While this interface is at a higher level than some of the prototype work with the io-device interface, I don't think it should be a horrific change to osiris_log. The rest of Osiris should have no idea that tiering is happening.

I'll work on some exploratory prototyping for something along these lines.

---

Appendix: S3 consistency model and tools...

• Strong consistency - added Dec 1, 2020
  • KIP-405 talks about how S3 is eventually consistent. Note that the v1 of that KIP was written in 2018 and strong consistency was added in 2020.
  • Docs: Amazon S3 data consistency model
• Conditional writes - added Aug 20, 2024
• Compare-and-swap writes - added Nov 25, 2024

---