feat(input): add cache input for consuming items from a cache resource - #941
feat(input): add cache input for consuming items from a cache resource#941ecordell wants to merge 2 commits into
Conversation
3359578 to
acae310
Compare
Adds a cache input that reads all items stored within a cache resource supporting the optional ListableCache interface, emitting each item as a message with its key in the cache_key metadata field and shutting down once the contents of the cache have been fully consumed. Keys are captured once on connect and values are then read individually, so items deleted mid-stream are skipped and only the key list is held in memory.
acae310 to
faae99c
Compare
gregfurman
left a comment
gregfurman
left a comment
There was a problem hiding this comment.
I've rebase this to include the latest changes. It's looking great!
| ` | ||
| input: | ||
| cache: | ||
| resource: source |
There was a problem hiding this comment.
Should this be mirroring the output destination with the target field being used to specify the cache resource? e.g
input:
cache:
target: source|
|
||
| msg := service.NewMessage(value) | ||
| msg.MetaSetMut("cache_key", key) | ||
| return msg, func(context.Context, error) error { return nil }, nil |
There was a problem hiding this comment.
Q: should we not be deleting the key from the cache on successful delivery? Or perhaps an option that allows for this would be a good idea like we do with the aws_s3 input/processor's delete_objects field.
| listErr = fmt.Errorf("cache resource '%v' does not support listing keys", c.cacheName) | ||
| return | ||
| } | ||
| for key, err := range lister.Keys(ctx) { |
There was a problem hiding this comment.
suggestion: Wondering if this should be converted into a pull-style iterator (via iter.Pull) that gets pulled from each time Read is called instead of this approach that'll only iterate over what we have on startup.
Otherwise, I think we should consider using a channel here that gets populated and read off-of in the read call. That way, we could potentially even run this listing logic within a goroutine that loops around constantly giving us new keys for the Read to use.
| "github.com/warpstreamlabs/bento/public/service" | ||
| ) | ||
|
|
||
| const ciFieldResource = "resource" |
There was a problem hiding this comment.
I think some filtering here to include/exclude certain keys from the iteration could be a good idea e.g
cache:
resource: source
exclude_prefixes:
- "system_"
- "foo_"That way, we can save ourselves a processor call to filter out unwanted data from within the component instead of only after a Get call has been made.
| Beta(). | ||
| Categories("Utility"). | ||
| Version("1.21.0"). | ||
| Summary("Reads items stored within a [cache resource](/docs/components/caches/about), emitting each item as a message and shutting down once the contents of the cache have been fully consumed."). |
There was a problem hiding this comment.
shutting down once the contents of the cache have been fully consumed.
We could the https://warpstreamlabs.github.io/bento/docs/components/inputs/read_until to restart the input when exhausted. Think an example of this would also be great for those wanting to constantly poll from the input.
| Description("The [`cache` resource](/docs/components/caches/about) to read from."), | ||
| service.NewAutoRetryNacksToggleField(), | ||
| ). | ||
| Example( |
There was a problem hiding this comment.
Can we add another example here using the https://warpstreamlabs.github.io/bento/docs/components/inputs/sequence/ ?
Think it's a really good use-case for this component to drain a cache prior to the next one starting!
2 participants
Description
Stacked on #940 — only the last commit (
feat(input): add cache input...) is new to this PR; the diff will reduce to it once #940 merges.Adds a
cacheinput that reads all items stored within a cache resource supporting the optionalListableCacheinterface from #940, emitting one message per item with the item's key in thecache_keymetadata field, and shutting down once the contents of the cache have been fully consumed:Keys are captured once on connect and values are then read individually, so only the key list is held in memory rather than the full cache contents. Items deleted after the keys were captured are skipped. Attempting to consume from a cache that doesn't support key listing surfaces as a clear connection error.
Because the input terminates once exhausted, it composes with
sequencefor seed-then-follow patterns (replay stored state, then consume a live source).Testing
Unit tests covering read-all, skip-deleted and missing-resource behaviour, plus a stream-level test through
StreamBuilderbacked by amemorycache. Component docs generated withbento_docs_genand lint-checked.