Add log-reader abstractions and hooks into writer/replica lifecycles - #232
Open
the-mikedavis wants to merge 7 commits into
Open
Add log-reader abstractions and hooks into writer/replica lifecycles#232the-mikedavis wants to merge 7 commits into
the-mikedavis wants to merge 7 commits into
Conversation
This change introduces a behaviour `osiris_log_reader` which can be implemented externally to read from a stream at a given offset spec. This closes over the high-level reading operations `send_file/3` and `chunk_iterator/3`. `osiris:init_reader/4` selects the reader module based on application env, and then callers use `osiris_log_reader` to interact with the reader. By default all of these functions delegate to `osiris_log`. `osiris_log` doesn't need any meaningful changes this way. The only change is to expose the `header_map()` type.
This can be used flexibly to evaluate retention depending on the name or contents of index files. You pass in a function which returns a tuple with the index files split into two lists: to delete and to keep. This could be used as a way to truncate everything up to an offset or to guarantee that an offset (for example an uncommitted one) won't be truncated. Since these files are sorted, some retention functions could operate just on the names (deriving the offset of the segment with `erlang:binary_to_integer/1`).
This change refactors `parse_header/2` to take the chunk header binary and the position at which it was read and return a `header_map()`. This is useful for other readers - so that they do not need to duplicate the binary match code and `next_position` calculation.
When local retention on the writer node deletes a segment while the
`osiris_replica_reader` is reading it, `read_header_with_ra` skips
ahead to the next available segment using `first_chunk_id`. If the
replica has not yet received data up to that offset, `accept_chunk`
exits with `{accept_chunk_out_of_order, Received, Expected}`.
The replica recovers correctly either way — the stream coordinator
restarts it and `init_acceptor` resyncs to the writer's current
position. However, the unhandled exit floods the log with thousands
of `[error]` lines per event.
Catch the exit in `handle_incoming_data`, log a single `[warning]`,
and stop with `normal`. Recovery behavior is unchanged.
This adds hooks in the log which are executed for writers and acceptors that can be used in plugins. The plugin intercepts the config map in init and can modify values. Same for retention: a plugin can modify the retention specs when it is updated.
After `trigger_retention_eval` sets counters from local segment state, call the hook to let plugins override values. The tiered storage plugin uses this to correct `?C_FIRST_OFFSET` when the remote tier holds older data than the local tier.
This allows a more hermetic upgrade with no race between reading retention from its source (i.e. the metadata store in RabbitMQ) and updating it, in rabbitmq-stream-s3.
|
Tick the box to add this pull request to the merge queue (same as
|
Collaborator
|
Guilty 😬 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
This is a recreation of #196 - some git nonsense lead to the PR there auto-closing.
This is is three changes:
{'fun', fun()}retention spec which lets retention policies have fine-grained control over what prefix of the osiris_log is deleted.osiris_log_readerbehaviour and module. Consumers of Osiris use theosirisAPI to create an offset reader and thenosiris_log_readerto use the reader. The default implementation uses osiris_log as usual.osiris_log_hooksbehaviour and default (empty) implementation. Implementors can use this to hook into the lifecycles of Osiris around writer and acceptor creation, and retention updates. This one is super small.Together these things make a solid base for implementing #184 (fixes #184). We use it in https://github.com/amazon-mq/rabbitmq-stream-s3 to serve a tiered-storage approach that aggressively archives data to elastic blob storage (Amazon S3) which appears seemless from the outside. It should be generic to other approaches (e.g. other blob stores or HDFS, or maybe even offload to shared spinning disks / RAID).
See a bunch of prior history in #196 and #194 before that.