From 3d31f0c20ef4c2581424adccd4c339ab86ff0d0e Mon Sep 17 00:00:00 2001 From: lublak <44057030+lublak@users.noreply.github.com> Date: Tue, 10 Mar 2026 16:11:16 +0100 Subject: [PATCH 01/24] add cache collector --- .../impl/pure/processor_cache_collector.go | 346 ++++++++++++++++++ public/service/message.go | 21 ++ public/service/message_batch_blobl.go | 27 ++ 3 files changed, 394 insertions(+) create mode 100644 internal/impl/pure/processor_cache_collector.go diff --git a/internal/impl/pure/processor_cache_collector.go b/internal/impl/pure/processor_cache_collector.go new file mode 100644 index 0000000000..4962b9cce6 --- /dev/null +++ b/internal/impl/pure/processor_cache_collector.go @@ -0,0 +1,346 @@ +package pure + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/warpstreamlabs/bento/public/bloblang" + "github.com/warpstreamlabs/bento/public/service" +) + +const ( + cacheCollectorPFieldResource = "resource" + cacheCollectorPFieldKey = "key" + cacheCollectorPFieldInit = "init" + cacheCollectorPFieldAppendCheck = "append_check" + cacheCollectorPFieldAppendMap = "append_map" + cacheCollectorPFieldFlushCheck = "flush_check" + cacheCollectorPFieldFlushDeletes = "flush_deletes" + cacheCollectorPFieldFlushMap = "flush_map" + cacheCollectorPFieldTTL = "ttl" +) + +func cacheCollectorSpec() *service.ConfigSpec { + return service.NewConfigSpec(). + Categories("Mapping"). + Stable(). + Summary("Accumulates messages across batch boundaries using a cache resource, allowing you to build up state before emitting a final result as structed data."). + Description(` +This processor works by storing an accumulated value in a cache, which is updated on each message based on bloblang expressions. It supports three phases: + +1. `+"`init`"+`: When the cache key doesn't exist, this bloblang expression is used to initialize the value. +2. `+"`append_check`"+`: For each message, if this expression evaluates to true, the value is updated using `+"`append_map`"+`. +3. `+"`flush_check`"+`: When this expression evaluates to true, the accumulated value is emitted as a new message and the cache is optionally cleared. + +The `+"`append_map`"+` bloblang expression can access both the current cached value as `+"`this.old`"+` and the current message as `+"`this.new`"+`.`). + Fields( + service.NewStringField(cacheCollectorPFieldResource). + Description("The [`cache` resource](/docs/components/caches/about) to use for storing accumulated state."), + service.NewInterpolatedStringField(cacheCollectorPFieldKey). + Description("A key for the cache entry. This should be consistent across messages that should be grouped together."), + service.NewBloblangField(cacheCollectorPFieldInit). + Description("Bloblang expression to initialize the value when the cache key doesn't exist. Defaults to an empty array."). + Default("root = []"). + Examples("root = []", `root = {"items": []}`, `root = {"count": 0, "total": 0}`), + service.NewBloblangField(cacheCollectorPFieldAppendCheck). + Description("Bloblang expression that must evaluate to `true` for a message to be appended to the accumulated value."). + Examples(`this.process == "work"`, `batch_index() < 100`), + service.NewBloblangField(cacheCollectorPFieldAppendMap). + Description("Bloblang expression used to update the accumulated value. It receives the current value as `this.old` and the new message as `this.new`."). + Examples(`root = this.old.append(this.new)`, `root = {"count": this.old.count + 1, "total": this.old.total + this.new.value}`), + service.NewBloblangField(cacheCollectorPFieldFlushCheck). + Description("Bloblang expression that must evaluate to `true` to emit the accumulated value and potentially clear the cache."). + Examples(`this.process == "end"`, `batch_index() == 0 && message_index() > 0 && batch_index() % 100 == 0`), + service.NewBoolField(cacheCollectorPFieldFlushDeletes). + Description("When `true`, the cache entry is deleted after flushing. Defaults to `false`."). + Default(false), + service.NewBloblangField(cacheCollectorPFieldFlushMap). + Description("Bloblang expression to transform the accumulated value before emitting. Defaults to `root`."). + Default("root = this"). + Examples(`root = this`, `root = {"result": this}`, `root.items = this`), + service.NewInterpolatedStringField(cacheCollectorPFieldTTL). + Description("The TTL of each individual item as a duration string. After this period an item will be eligible for removal during the next compaction. Not all caches support per-key TTLs, those that do will have a configuration field `default_ttl`, and those that do not will fall back to their generally configured TTL setting."). + Examples("60s", "5m", "36h"). + Advanced(). + Optional(), + ) +} + +type cacheCollector struct { + cacheName string + + key *service.InterpolatedString + init *bloblang.Executor + appendCheck *bloblang.Executor + appendMap *bloblang.Executor + flushCheck *bloblang.Executor + flushDeletes bool + flushMap *bloblang.Executor + ttl *service.InterpolatedString + + mgr *service.Resources +} + +func init() { + err := service.RegisterBatchProcessor( + "cache_collector", cacheCollectorSpec(), + func(conf *service.ParsedConfig, mgr *service.Resources) (service.BatchProcessor, error) { + return newCacheCollector(conf, mgr) + }) + if err != nil { + panic(err) + } +} + +func newCacheCollector(conf *service.ParsedConfig, mgr *service.Resources) (*cacheCollector, error) { + resource, err := conf.FieldString(cacheCollectorPFieldResource) + if err != nil { + return nil, err + } + + key, err := conf.FieldInterpolatedString(cacheCollectorPFieldKey) + if err != nil { + return nil, err + } + + init, err := conf.FieldBloblang(cacheCollectorPFieldInit) + if err != nil { + return nil, err + } + + appendCheck, err := conf.FieldBloblang(cacheCollectorPFieldAppendCheck) + if err != nil { + return nil, err + } + + appendMap, err := conf.FieldBloblang(cacheCollectorPFieldAppendMap) + if err != nil { + return nil, err + } + + flushCheck, err := conf.FieldBloblang(cacheCollectorPFieldFlushCheck) + if err != nil { + return nil, err + } + + flushDeletes, err := conf.FieldBool(cacheCollectorPFieldFlushDeletes) + if err != nil { + return nil, err + } + + flushMap, err := conf.FieldBloblang(cacheCollectorPFieldFlushMap) + if err != nil { + return nil, err + } + + ttl, err := conf.FieldInterpolatedString(cacheCollectorPFieldTTL) + if err != nil { + return nil, err + } + + return &cacheCollector{ + key: key, + init: init, + appendCheck: appendCheck, + appendMap: appendMap, + flushCheck: flushCheck, + flushMap: flushMap, + + cacheName: resource, + flushDeletes: flushDeletes, + + ttl: ttl, + + mgr: mgr, + }, nil +} + +type cacheCollectorAppendMessageData struct { + cached json.RawMessage + current json.RawMessage +} + +func (cc *cacheCollector) ProcessBatch(ctx context.Context, batch service.MessageBatch) ([]service.MessageBatch, error) { + var newMsgs []*service.Message + + var keyInterp *service.MessageBatchInterpolationExecutor + if cc.key != nil { + keyInterp = batch.InterpolationExecutor(cc.key) + } + + var ttlInterp *service.MessageBatchInterpolationExecutor + if cc.ttl != nil { + keyInterp = batch.InterpolationExecutor(cc.ttl) + } + + var appendCheck *service.MessageBatchBloblangExecutor + if cc.appendCheck != nil { + appendCheck = batch.BloblangExecutor(cc.appendCheck) + } + + var flushCheck *service.MessageBatchBloblangExecutor + if cc.flushCheck != nil { + flushCheck = batch.BloblangExecutor(cc.flushCheck) + } + + var init *service.MessageBatchBloblangExecutor + if cc.init != nil { + init = batch.BloblangExecutor(cc.init) + } + + for i, msg := range batch { + key, err := keyInterp.TryString(i) + if err != nil { + return nil, fmt.Errorf("key evaluation error: %w", err) + } + + ttls, err := ttlInterp.TryString(i) + if err != nil { + return nil, err + } + + var ttl *time.Duration + + if ttls != "" { + td, err := time.ParseDuration(ttls) + if err != nil { + return nil, fmt.Errorf("ttl must be a duration: %w", err) + } + ttl = &td + } + + var processAppend bool + var processFlush bool + + processAppend, err = appendCheck.QueryBool(i) + if err != nil { + return nil, fmt.Errorf("append_check evaluation error: %w", err) + } + + processFlush, err = flushCheck.QueryBool(i) + if err != nil { + return nil, fmt.Errorf("flush_check evaluation error: %w", err) + } + + if processAppend || processFlush { + var cachedValue []byte + if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { + cachedValue, err = cache.Get(ctx, key) + if err != nil { + if errors.Is(err, service.ErrKeyNotFound) { + cachedValue = nil + } else { + err = fmt.Errorf("failed to get cache key '%s': %v", key, err) + } + } + }); cerr != nil { + err = cerr + } + + if err != nil { + return nil, err + } + + if processAppend { + if cachedValue == nil { + initMsg, err := init.Query(i) + if err != nil { + return nil, fmt.Errorf("init evaluation error: %w", err) + } + initData, err := initMsg.AsBytes() + if err != nil { + return nil, fmt.Errorf("init data error: %w", err) + } + cachedValue = initData + } + + currentValue, err := msg.AsBytes() + if err != nil { + return nil, err + } + + appendMsg := service.NewMessage(nil) + appendMsg.SetStructured(cacheCollectorAppendMessageData{ + cached: cachedValue, + current: currentValue, + }) + + appendResult, err := appendMsg.BloblangQuery(cc.appendMap) + if err != nil { + return nil, err + } + + cachedValue, err = appendResult.AsBytes() + if err != nil { + return nil, err + } + + if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { + err = cache.Set(ctx, key, cachedValue, ttl) + if err != nil { + err = fmt.Errorf("failed to set cache key '%s': %v", key, err) + } + }); cerr != nil { + err = cerr + } + + if err != nil { + return nil, err + } + } + + if processFlush { + var cachedValue []byte + if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { + cachedValue, err = cache.Get(ctx, key) + if err != nil { + if errors.Is(err, service.ErrKeyNotFound) { + cachedValue = nil + } else { + err = fmt.Errorf("failed to get cache key '%s': %v", key, err) + } + } + }); cerr != nil { + err = cerr + } + + if err != nil { + return nil, err + } + + flushMsg := service.NewMessage(cachedValue) + + flushResult, err := flushMsg.BloblangQuery(cc.appendMap) + if err != nil { + return nil, err + } + + if cc.flushDeletes { + if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { + err = cache.Delete(ctx, key) + if err != nil { + err = fmt.Errorf("failed to delete cache key '%s': %v", key, err) + } + }); cerr != nil { + err = cerr + } + + if err != nil { + return nil, err + } + } + + newMsgs = append(newMsgs, flushResult) + } + } + } + + return []service.MessageBatch{newMsgs}, nil +} + +func (cc *cacheCollector) Close(ctx context.Context) error { + return nil +} diff --git a/public/service/message.go b/public/service/message.go index bcf3afd839..d80b1a957c 100644 --- a/public/service/message.go +++ b/public/service/message.go @@ -367,6 +367,27 @@ func (m *Message) BloblangQuery(blobl *bloblang.Executor) (*Message, error) { return nil, nil } +// BloblangQueryBool executes a parsed Bloblang mapping on a message and returns a +// bool back or an error if the mapping fails, the message gets deleted or value is not a bool. +func (m *Message) BloblangQueryBool(blobl *bloblang.Executor) (bool, error) { + newPart, err := m.BloblangQuery(blobl) + if err != nil { + return false, err + } + if newPart == nil { + return false, errors.New("query mapping resulted in deleted message, expected a bool value") + } + newValue, err := newPart.AsStructured() + if err != nil { + return false, err + } + if b, ok := newValue.(bool); ok { + return b, nil + } + + return false, value.NewTypeErrorFrom("mapping", newValue, value.TBool) +} + // BloblangQueryValue executes a parsed Bloblang mapping on a message and // returns the raw value result, or an error if either the mapping fails. // The error bloblang.ErrRootDeleted is returned if the root of the mapping diff --git a/public/service/message_batch_blobl.go b/public/service/message_batch_blobl.go index 9b030141f1..0cd42f80ca 100644 --- a/public/service/message_batch_blobl.go +++ b/public/service/message_batch_blobl.go @@ -1,6 +1,8 @@ package service import ( + "errors" + "github.com/warpstreamlabs/bento/internal/bloblang/field" "github.com/warpstreamlabs/bento/internal/bloblang/mapping" "github.com/warpstreamlabs/bento/internal/bloblang/query" @@ -56,6 +58,31 @@ func (b MessageBatchBloblangExecutor) Query(index int) (*Message, error) { return nil, nil } +// Query executes a parsed Bloblang mapping on a message batch, from the +// perspective of a particular message index, and returns a bool back or an +// error if the mapping fails, the message gets deleted or value is not a bool. +// +// This method allows mappings to perform windowed aggregations across message +// batches. +func (b MessageBatchBloblangExecutor) QueryBool(index int) (bool, error) { + newPart, err := b.Query(index) + if err != nil { + return false, err + } + if newPart == nil { + return false, errors.New("query mapping resulted in deleted message, expected a bool value") + } + newValue, err := newPart.AsStructured() + if err != nil { + return false, err + } + if b, ok := newValue.(bool); ok { + return b, nil + } + + return false, value.NewTypeErrorFrom("mapping", newValue, value.TBool) +} + // QueryValue executes a parsed Bloblang mapping on a message batch, // from the perspective of a particular message index, and returns the raw value // result or an error if the mapping fails. The error bloblang.ErrRootDeleted is From 4d1dd261de01eca3b8964b63cf2c0d9bf1956353 Mon Sep 17 00:00:00 2001 From: lublak Date: Tue, 10 Mar 2026 20:35:20 +0100 Subject: [PATCH 02/24] fix documentation --- internal/impl/pure/processor_cache_collector.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/internal/impl/pure/processor_cache_collector.go b/internal/impl/pure/processor_cache_collector.go index 4962b9cce6..3664593960 100644 --- a/internal/impl/pure/processor_cache_collector.go +++ b/internal/impl/pure/processor_cache_collector.go @@ -35,7 +35,7 @@ This processor works by storing an accumulated value in a cache, which is update 2. `+"`append_check`"+`: For each message, if this expression evaluates to true, the value is updated using `+"`append_map`"+`. 3. `+"`flush_check`"+`: When this expression evaluates to true, the accumulated value is emitted as a new message and the cache is optionally cleared. -The `+"`append_map`"+` bloblang expression can access both the current cached value as `+"`this.old`"+` and the current message as `+"`this.new`"+`.`). +The `+"`append_map`"+` bloblang expression can access both the current cached value as `+"`this.cached`"+` and the current message as `+"`this.current`"+`.`). Fields( service.NewStringField(cacheCollectorPFieldResource). Description("The [`cache` resource](/docs/components/caches/about) to use for storing accumulated state."), @@ -49,8 +49,8 @@ The `+"`append_map`"+` bloblang expression can access both the current cached va Description("Bloblang expression that must evaluate to `true` for a message to be appended to the accumulated value."). Examples(`this.process == "work"`, `batch_index() < 100`), service.NewBloblangField(cacheCollectorPFieldAppendMap). - Description("Bloblang expression used to update the accumulated value. It receives the current value as `this.old` and the new message as `this.new`."). - Examples(`root = this.old.append(this.new)`, `root = {"count": this.old.count + 1, "total": this.old.total + this.new.value}`), + Description("Bloblang expression used to update the accumulated value. It receives the current value as `this.cached` and the new message as `this.current`."). + Examples(`root = this.cached.append(this.current)`, `root = {"count": this.cached.count + 1, "total": this.cached.total + this.current.value}`), service.NewBloblangField(cacheCollectorPFieldFlushCheck). Description("Bloblang expression that must evaluate to `true` to emit the accumulated value and potentially clear the cache."). Examples(`this.process == "end"`, `batch_index() == 0 && message_index() > 0 && batch_index() % 100 == 0`), From 8dbb5a997e3adb2e9d88789d468d341ec7be6e1f Mon Sep 17 00:00:00 2001 From: lublak Date: Tue, 10 Mar 2026 22:49:40 +0100 Subject: [PATCH 03/24] improve cached value init implementation --- .../impl/pure/processor_cache_collector.go | 40 +++++-------------- 1 file changed, 11 insertions(+), 29 deletions(-) diff --git a/internal/impl/pure/processor_cache_collector.go b/internal/impl/pure/processor_cache_collector.go index 3664593960..3c1e8d5aaf 100644 --- a/internal/impl/pure/processor_cache_collector.go +++ b/internal/impl/pure/processor_cache_collector.go @@ -244,19 +244,19 @@ func (cc *cacheCollector) ProcessBatch(ctx context.Context, batch service.Messag return nil, err } - if processAppend { - if cachedValue == nil { - initMsg, err := init.Query(i) - if err != nil { - return nil, fmt.Errorf("init evaluation error: %w", err) - } - initData, err := initMsg.AsBytes() - if err != nil { - return nil, fmt.Errorf("init data error: %w", err) - } - cachedValue = initData + if cachedValue == nil { + initMsg, err := init.Query(i) + if err != nil { + return nil, fmt.Errorf("init evaluation error: %w", err) } + initData, err := initMsg.AsBytes() + if err != nil { + return nil, fmt.Errorf("init data error: %w", err) + } + cachedValue = initData + } + if processAppend { currentValue, err := msg.AsBytes() if err != nil { return nil, err @@ -293,24 +293,6 @@ func (cc *cacheCollector) ProcessBatch(ctx context.Context, batch service.Messag } if processFlush { - var cachedValue []byte - if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { - cachedValue, err = cache.Get(ctx, key) - if err != nil { - if errors.Is(err, service.ErrKeyNotFound) { - cachedValue = nil - } else { - err = fmt.Errorf("failed to get cache key '%s': %v", key, err) - } - } - }); cerr != nil { - err = cerr - } - - if err != nil { - return nil, err - } - flushMsg := service.NewMessage(cachedValue) flushResult, err := flushMsg.BloblangQuery(cc.appendMap) From 4311311432773e8fa316f229439ed024ee26a3c3 Mon Sep 17 00:00:00 2001 From: lublak Date: Tue, 10 Mar 2026 22:55:34 +0100 Subject: [PATCH 04/24] make docs --- .../impl/pure/processor_cache_collector.go | 656 +++++++++--------- .../components/processors/cache_collector.md | 195 ++++++ 2 files changed, 523 insertions(+), 328 deletions(-) create mode 100644 website/docs/components/processors/cache_collector.md diff --git a/internal/impl/pure/processor_cache_collector.go b/internal/impl/pure/processor_cache_collector.go index 3c1e8d5aaf..1ff0c7a76b 100644 --- a/internal/impl/pure/processor_cache_collector.go +++ b/internal/impl/pure/processor_cache_collector.go @@ -1,328 +1,328 @@ -package pure - -import ( - "context" - "encoding/json" - "errors" - "fmt" - "time" - - "github.com/warpstreamlabs/bento/public/bloblang" - "github.com/warpstreamlabs/bento/public/service" -) - -const ( - cacheCollectorPFieldResource = "resource" - cacheCollectorPFieldKey = "key" - cacheCollectorPFieldInit = "init" - cacheCollectorPFieldAppendCheck = "append_check" - cacheCollectorPFieldAppendMap = "append_map" - cacheCollectorPFieldFlushCheck = "flush_check" - cacheCollectorPFieldFlushDeletes = "flush_deletes" - cacheCollectorPFieldFlushMap = "flush_map" - cacheCollectorPFieldTTL = "ttl" -) - -func cacheCollectorSpec() *service.ConfigSpec { - return service.NewConfigSpec(). - Categories("Mapping"). - Stable(). - Summary("Accumulates messages across batch boundaries using a cache resource, allowing you to build up state before emitting a final result as structed data."). - Description(` -This processor works by storing an accumulated value in a cache, which is updated on each message based on bloblang expressions. It supports three phases: - -1. `+"`init`"+`: When the cache key doesn't exist, this bloblang expression is used to initialize the value. -2. `+"`append_check`"+`: For each message, if this expression evaluates to true, the value is updated using `+"`append_map`"+`. -3. `+"`flush_check`"+`: When this expression evaluates to true, the accumulated value is emitted as a new message and the cache is optionally cleared. - -The `+"`append_map`"+` bloblang expression can access both the current cached value as `+"`this.cached`"+` and the current message as `+"`this.current`"+`.`). - Fields( - service.NewStringField(cacheCollectorPFieldResource). - Description("The [`cache` resource](/docs/components/caches/about) to use for storing accumulated state."), - service.NewInterpolatedStringField(cacheCollectorPFieldKey). - Description("A key for the cache entry. This should be consistent across messages that should be grouped together."), - service.NewBloblangField(cacheCollectorPFieldInit). - Description("Bloblang expression to initialize the value when the cache key doesn't exist. Defaults to an empty array."). - Default("root = []"). - Examples("root = []", `root = {"items": []}`, `root = {"count": 0, "total": 0}`), - service.NewBloblangField(cacheCollectorPFieldAppendCheck). - Description("Bloblang expression that must evaluate to `true` for a message to be appended to the accumulated value."). - Examples(`this.process == "work"`, `batch_index() < 100`), - service.NewBloblangField(cacheCollectorPFieldAppendMap). - Description("Bloblang expression used to update the accumulated value. It receives the current value as `this.cached` and the new message as `this.current`."). - Examples(`root = this.cached.append(this.current)`, `root = {"count": this.cached.count + 1, "total": this.cached.total + this.current.value}`), - service.NewBloblangField(cacheCollectorPFieldFlushCheck). - Description("Bloblang expression that must evaluate to `true` to emit the accumulated value and potentially clear the cache."). - Examples(`this.process == "end"`, `batch_index() == 0 && message_index() > 0 && batch_index() % 100 == 0`), - service.NewBoolField(cacheCollectorPFieldFlushDeletes). - Description("When `true`, the cache entry is deleted after flushing. Defaults to `false`."). - Default(false), - service.NewBloblangField(cacheCollectorPFieldFlushMap). - Description("Bloblang expression to transform the accumulated value before emitting. Defaults to `root`."). - Default("root = this"). - Examples(`root = this`, `root = {"result": this}`, `root.items = this`), - service.NewInterpolatedStringField(cacheCollectorPFieldTTL). - Description("The TTL of each individual item as a duration string. After this period an item will be eligible for removal during the next compaction. Not all caches support per-key TTLs, those that do will have a configuration field `default_ttl`, and those that do not will fall back to their generally configured TTL setting."). - Examples("60s", "5m", "36h"). - Advanced(). - Optional(), - ) -} - -type cacheCollector struct { - cacheName string - - key *service.InterpolatedString - init *bloblang.Executor - appendCheck *bloblang.Executor - appendMap *bloblang.Executor - flushCheck *bloblang.Executor - flushDeletes bool - flushMap *bloblang.Executor - ttl *service.InterpolatedString - - mgr *service.Resources -} - -func init() { - err := service.RegisterBatchProcessor( - "cache_collector", cacheCollectorSpec(), - func(conf *service.ParsedConfig, mgr *service.Resources) (service.BatchProcessor, error) { - return newCacheCollector(conf, mgr) - }) - if err != nil { - panic(err) - } -} - -func newCacheCollector(conf *service.ParsedConfig, mgr *service.Resources) (*cacheCollector, error) { - resource, err := conf.FieldString(cacheCollectorPFieldResource) - if err != nil { - return nil, err - } - - key, err := conf.FieldInterpolatedString(cacheCollectorPFieldKey) - if err != nil { - return nil, err - } - - init, err := conf.FieldBloblang(cacheCollectorPFieldInit) - if err != nil { - return nil, err - } - - appendCheck, err := conf.FieldBloblang(cacheCollectorPFieldAppendCheck) - if err != nil { - return nil, err - } - - appendMap, err := conf.FieldBloblang(cacheCollectorPFieldAppendMap) - if err != nil { - return nil, err - } - - flushCheck, err := conf.FieldBloblang(cacheCollectorPFieldFlushCheck) - if err != nil { - return nil, err - } - - flushDeletes, err := conf.FieldBool(cacheCollectorPFieldFlushDeletes) - if err != nil { - return nil, err - } - - flushMap, err := conf.FieldBloblang(cacheCollectorPFieldFlushMap) - if err != nil { - return nil, err - } - - ttl, err := conf.FieldInterpolatedString(cacheCollectorPFieldTTL) - if err != nil { - return nil, err - } - - return &cacheCollector{ - key: key, - init: init, - appendCheck: appendCheck, - appendMap: appendMap, - flushCheck: flushCheck, - flushMap: flushMap, - - cacheName: resource, - flushDeletes: flushDeletes, - - ttl: ttl, - - mgr: mgr, - }, nil -} - -type cacheCollectorAppendMessageData struct { - cached json.RawMessage - current json.RawMessage -} - -func (cc *cacheCollector) ProcessBatch(ctx context.Context, batch service.MessageBatch) ([]service.MessageBatch, error) { - var newMsgs []*service.Message - - var keyInterp *service.MessageBatchInterpolationExecutor - if cc.key != nil { - keyInterp = batch.InterpolationExecutor(cc.key) - } - - var ttlInterp *service.MessageBatchInterpolationExecutor - if cc.ttl != nil { - keyInterp = batch.InterpolationExecutor(cc.ttl) - } - - var appendCheck *service.MessageBatchBloblangExecutor - if cc.appendCheck != nil { - appendCheck = batch.BloblangExecutor(cc.appendCheck) - } - - var flushCheck *service.MessageBatchBloblangExecutor - if cc.flushCheck != nil { - flushCheck = batch.BloblangExecutor(cc.flushCheck) - } - - var init *service.MessageBatchBloblangExecutor - if cc.init != nil { - init = batch.BloblangExecutor(cc.init) - } - - for i, msg := range batch { - key, err := keyInterp.TryString(i) - if err != nil { - return nil, fmt.Errorf("key evaluation error: %w", err) - } - - ttls, err := ttlInterp.TryString(i) - if err != nil { - return nil, err - } - - var ttl *time.Duration - - if ttls != "" { - td, err := time.ParseDuration(ttls) - if err != nil { - return nil, fmt.Errorf("ttl must be a duration: %w", err) - } - ttl = &td - } - - var processAppend bool - var processFlush bool - - processAppend, err = appendCheck.QueryBool(i) - if err != nil { - return nil, fmt.Errorf("append_check evaluation error: %w", err) - } - - processFlush, err = flushCheck.QueryBool(i) - if err != nil { - return nil, fmt.Errorf("flush_check evaluation error: %w", err) - } - - if processAppend || processFlush { - var cachedValue []byte - if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { - cachedValue, err = cache.Get(ctx, key) - if err != nil { - if errors.Is(err, service.ErrKeyNotFound) { - cachedValue = nil - } else { - err = fmt.Errorf("failed to get cache key '%s': %v", key, err) - } - } - }); cerr != nil { - err = cerr - } - - if err != nil { - return nil, err - } - - if cachedValue == nil { - initMsg, err := init.Query(i) - if err != nil { - return nil, fmt.Errorf("init evaluation error: %w", err) - } - initData, err := initMsg.AsBytes() - if err != nil { - return nil, fmt.Errorf("init data error: %w", err) - } - cachedValue = initData - } - - if processAppend { - currentValue, err := msg.AsBytes() - if err != nil { - return nil, err - } - - appendMsg := service.NewMessage(nil) - appendMsg.SetStructured(cacheCollectorAppendMessageData{ - cached: cachedValue, - current: currentValue, - }) - - appendResult, err := appendMsg.BloblangQuery(cc.appendMap) - if err != nil { - return nil, err - } - - cachedValue, err = appendResult.AsBytes() - if err != nil { - return nil, err - } - - if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { - err = cache.Set(ctx, key, cachedValue, ttl) - if err != nil { - err = fmt.Errorf("failed to set cache key '%s': %v", key, err) - } - }); cerr != nil { - err = cerr - } - - if err != nil { - return nil, err - } - } - - if processFlush { - flushMsg := service.NewMessage(cachedValue) - - flushResult, err := flushMsg.BloblangQuery(cc.appendMap) - if err != nil { - return nil, err - } - - if cc.flushDeletes { - if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { - err = cache.Delete(ctx, key) - if err != nil { - err = fmt.Errorf("failed to delete cache key '%s': %v", key, err) - } - }); cerr != nil { - err = cerr - } - - if err != nil { - return nil, err - } - } - - newMsgs = append(newMsgs, flushResult) - } - } - } - - return []service.MessageBatch{newMsgs}, nil -} - -func (cc *cacheCollector) Close(ctx context.Context) error { - return nil -} +package pure + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/warpstreamlabs/bento/public/bloblang" + "github.com/warpstreamlabs/bento/public/service" +) + +const ( + cacheCollectorPFieldResource = "resource" + cacheCollectorPFieldKey = "key" + cacheCollectorPFieldInit = "init" + cacheCollectorPFieldAppendCheck = "append_check" + cacheCollectorPFieldAppendMap = "append_map" + cacheCollectorPFieldFlushCheck = "flush_check" + cacheCollectorPFieldFlushDeletes = "flush_deletes" + cacheCollectorPFieldFlushMap = "flush_map" + cacheCollectorPFieldTTL = "ttl" +) + +func cacheCollectorSpec() *service.ConfigSpec { + return service.NewConfigSpec(). + Categories("Mapping"). + Stable(). + Summary("Accumulates messages across batch boundaries using a cache resource, allowing you to build up state before emitting a final result as structed data."). + Description(` +This processor works by storing an accumulated value in a cache, which is updated on each message based on bloblang expressions. It supports three phases: + +1. `+"`init`"+`: When the cache key doesn't exist, this bloblang expression is used to initialize the value. +2. `+"`append_check`"+`: For each message, if this expression evaluates to true, the value is updated using `+"`append_map`"+`. +3. `+"`flush_check`"+`: When this expression evaluates to true, the accumulated value is emitted as a new message and the cache is optionally cleared. + +The `+"`append_map`"+` bloblang expression can access both the current cached value as `+"`this.cached`"+` and the current message as `+"`this.current`"+`.`). + Fields( + service.NewStringField(cacheCollectorPFieldResource). + Description("The [`cache` resource](/docs/components/caches/about) to use for storing accumulated state."), + service.NewInterpolatedStringField(cacheCollectorPFieldKey). + Description("A key for the cache entry. This should be consistent across messages that should be grouped together."), + service.NewBloblangField(cacheCollectorPFieldInit). + Description("Bloblang expression to initialize the value when the cache key doesn't exist. Defaults to an empty array."). + Default("root = []"). + Examples("root = []", `root = {"items": []}`, `root = {"count": 0, "total": 0}`), + service.NewBloblangField(cacheCollectorPFieldAppendCheck). + Description("Bloblang expression that must evaluate to `true` for a message to be appended to the accumulated value."). + Examples(`this.process == "work"`, `batch_index() < 100`), + service.NewBloblangField(cacheCollectorPFieldAppendMap). + Description("Bloblang expression used to update the accumulated value. It receives the current value as `this.cached` and the new message as `this.current`."). + Examples(`root = this.cached.append(this.current)`, `root = {"count": this.cached.count + 1, "total": this.cached.total + this.current.value}`), + service.NewBloblangField(cacheCollectorPFieldFlushCheck). + Description("Bloblang expression that must evaluate to `true` to emit the accumulated value and potentially clear the cache."). + Examples(`this.process == "end"`, `batch_index() == 0 && message_index() > 0 && batch_index() % 100 == 0`), + service.NewBoolField(cacheCollectorPFieldFlushDeletes). + Description("When `true`, the cache entry is deleted after flushing. Defaults to `false`."). + Default(false), + service.NewBloblangField(cacheCollectorPFieldFlushMap). + Description("Bloblang expression to transform the accumulated value before emitting. Defaults to `root`."). + Default("root = this"). + Examples(`root = this`, `root = {"result": this}`, `root.items = this`), + service.NewInterpolatedStringField(cacheCollectorPFieldTTL). + Description("The TTL of each individual item as a duration string. After this period an item will be eligible for removal during the next compaction. Not all caches support per-key TTLs, those that do will have a configuration field `default_ttl`, and those that do not will fall back to their generally configured TTL setting."). + Examples("60s", "5m", "36h"). + Advanced(). + Optional(), + ) +} + +type cacheCollector struct { + cacheName string + + key *service.InterpolatedString + init *bloblang.Executor + appendCheck *bloblang.Executor + appendMap *bloblang.Executor + flushCheck *bloblang.Executor + flushDeletes bool + flushMap *bloblang.Executor + ttl *service.InterpolatedString + + mgr *service.Resources +} + +func init() { + err := service.RegisterBatchProcessor( + "cache_collector", cacheCollectorSpec(), + func(conf *service.ParsedConfig, mgr *service.Resources) (service.BatchProcessor, error) { + return newCacheCollector(conf, mgr) + }) + if err != nil { + panic(err) + } +} + +func newCacheCollector(conf *service.ParsedConfig, mgr *service.Resources) (*cacheCollector, error) { + resource, err := conf.FieldString(cacheCollectorPFieldResource) + if err != nil { + return nil, err + } + + key, err := conf.FieldInterpolatedString(cacheCollectorPFieldKey) + if err != nil { + return nil, err + } + + init, err := conf.FieldBloblang(cacheCollectorPFieldInit) + if err != nil { + return nil, err + } + + appendCheck, err := conf.FieldBloblang(cacheCollectorPFieldAppendCheck) + if err != nil { + return nil, err + } + + appendMap, err := conf.FieldBloblang(cacheCollectorPFieldAppendMap) + if err != nil { + return nil, err + } + + flushCheck, err := conf.FieldBloblang(cacheCollectorPFieldFlushCheck) + if err != nil { + return nil, err + } + + flushDeletes, err := conf.FieldBool(cacheCollectorPFieldFlushDeletes) + if err != nil { + return nil, err + } + + flushMap, err := conf.FieldBloblang(cacheCollectorPFieldFlushMap) + if err != nil { + return nil, err + } + + ttl, err := conf.FieldInterpolatedString(cacheCollectorPFieldTTL) + if err != nil { + return nil, err + } + + return &cacheCollector{ + key: key, + init: init, + appendCheck: appendCheck, + appendMap: appendMap, + flushCheck: flushCheck, + flushMap: flushMap, + + cacheName: resource, + flushDeletes: flushDeletes, + + ttl: ttl, + + mgr: mgr, + }, nil +} + +type cacheCollectorAppendMessageData struct { + cached json.RawMessage + current json.RawMessage +} + +func (cc *cacheCollector) ProcessBatch(ctx context.Context, batch service.MessageBatch) ([]service.MessageBatch, error) { + var newMsgs []*service.Message + + var keyInterp *service.MessageBatchInterpolationExecutor + if cc.key != nil { + keyInterp = batch.InterpolationExecutor(cc.key) + } + + var ttlInterp *service.MessageBatchInterpolationExecutor + if cc.ttl != nil { + keyInterp = batch.InterpolationExecutor(cc.ttl) + } + + var appendCheck *service.MessageBatchBloblangExecutor + if cc.appendCheck != nil { + appendCheck = batch.BloblangExecutor(cc.appendCheck) + } + + var flushCheck *service.MessageBatchBloblangExecutor + if cc.flushCheck != nil { + flushCheck = batch.BloblangExecutor(cc.flushCheck) + } + + var init *service.MessageBatchBloblangExecutor + if cc.init != nil { + init = batch.BloblangExecutor(cc.init) + } + + for i, msg := range batch { + key, err := keyInterp.TryString(i) + if err != nil { + return nil, fmt.Errorf("key evaluation error: %w", err) + } + + ttls, err := ttlInterp.TryString(i) + if err != nil { + return nil, err + } + + var ttl *time.Duration + + if ttls != "" { + td, err := time.ParseDuration(ttls) + if err != nil { + return nil, fmt.Errorf("ttl must be a duration: %w", err) + } + ttl = &td + } + + var processAppend bool + var processFlush bool + + processAppend, err = appendCheck.QueryBool(i) + if err != nil { + return nil, fmt.Errorf("append_check evaluation error: %w", err) + } + + processFlush, err = flushCheck.QueryBool(i) + if err != nil { + return nil, fmt.Errorf("flush_check evaluation error: %w", err) + } + + if processAppend || processFlush { + var cachedValue []byte + if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { + cachedValue, err = cache.Get(ctx, key) + if err != nil { + if errors.Is(err, service.ErrKeyNotFound) { + cachedValue = nil + } else { + err = fmt.Errorf("failed to get cache key '%s': %v", key, err) + } + } + }); cerr != nil { + err = cerr + } + + if err != nil { + return nil, err + } + + if cachedValue == nil { + initMsg, err := init.Query(i) + if err != nil { + return nil, fmt.Errorf("init evaluation error: %w", err) + } + initData, err := initMsg.AsBytes() + if err != nil { + return nil, fmt.Errorf("init data error: %w", err) + } + cachedValue = initData + } + + if processAppend { + currentValue, err := msg.AsBytes() + if err != nil { + return nil, err + } + + appendMsg := service.NewMessage(nil) + appendMsg.SetStructured(cacheCollectorAppendMessageData{ + cached: cachedValue, + current: currentValue, + }) + + appendResult, err := appendMsg.BloblangQuery(cc.appendMap) + if err != nil { + return nil, err + } + + cachedValue, err = appendResult.AsBytes() + if err != nil { + return nil, err + } + + if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { + err = cache.Set(ctx, key, cachedValue, ttl) + if err != nil { + err = fmt.Errorf("failed to set cache key '%s': %v", key, err) + } + }); cerr != nil { + err = cerr + } + + if err != nil { + return nil, err + } + } + + if processFlush { + flushMsg := service.NewMessage(cachedValue) + + flushResult, err := flushMsg.BloblangQuery(cc.appendMap) + if err != nil { + return nil, err + } + + if cc.flushDeletes { + if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { + err = cache.Delete(ctx, key) + if err != nil { + err = fmt.Errorf("failed to delete cache key '%s': %v", key, err) + } + }); cerr != nil { + err = cerr + } + + if err != nil { + return nil, err + } + } + + newMsgs = append(newMsgs, flushResult) + } + } + } + + return []service.MessageBatch{newMsgs}, nil +} + +func (cc *cacheCollector) Close(ctx context.Context) error { + return nil +} diff --git a/website/docs/components/processors/cache_collector.md b/website/docs/components/processors/cache_collector.md new file mode 100644 index 0000000000..154c5f04c1 --- /dev/null +++ b/website/docs/components/processors/cache_collector.md @@ -0,0 +1,195 @@ +--- +title: cache_collector +slug: cache_collector +type: processor +status: stable +categories: ["Mapping"] +--- + + + +import Tabs from '@theme/Tabs'; +import TabItem from '@theme/TabItem'; + +Accumulates messages across batch boundaries using a cache resource, allowing you to build up state before emitting a final result as structed data. + + + + + + +```yml +# Common config fields, showing default values +label: "" +cache_collector: + resource: "" # No default (required) + key: "" # No default (required) + init: root = [] + append_check: this.process == "work" # No default (required) + append_map: root = this.cached.append(this.current) # No default (required) + flush_check: this.process == "end" # No default (required) + flush_deletes: false + flush_map: root = this +``` + + + + +```yml +# All config fields, showing default values +label: "" +cache_collector: + resource: "" # No default (required) + key: "" # No default (required) + init: root = [] + append_check: this.process == "work" # No default (required) + append_map: root = this.cached.append(this.current) # No default (required) + flush_check: this.process == "end" # No default (required) + flush_deletes: false + flush_map: root = this + ttl: 60s # No default (optional) +``` + + + + +This processor works by storing an accumulated value in a cache, which is updated on each message based on bloblang expressions. It supports three phases: + +1. `init`: When the cache key doesn't exist, this bloblang expression is used to initialize the value. +2. `append_check`: For each message, if this expression evaluates to true, the value is updated using `append_map`. +3. `flush_check`: When this expression evaluates to true, the accumulated value is emitted as a new message and the cache is optionally cleared. + +The `append_map` bloblang expression can access both the current cached value as `this.cached` and the current message as `this.current`. + +## Fields + +### `resource` + +The [`cache` resource](/docs/components/caches/about) to use for storing accumulated state. + + +Type: `string` + +### `key` + +A key for the cache entry. This should be consistent across messages that should be grouped together. +This field supports [interpolation functions](/docs/configuration/interpolation#bloblang-queries). + + +Type: `string` + +### `init` + +Bloblang expression to initialize the value when the cache key doesn't exist. Defaults to an empty array. + + +Type: `string` +Default: `"root = []"` + +```yml +# Examples + +init: root = [] + +init: 'root = {"items": []}' + +init: 'root = {"count": 0, "total": 0}' +``` + +### `append_check` + +Bloblang expression that must evaluate to `true` for a message to be appended to the accumulated value. + + +Type: `string` + +```yml +# Examples + +append_check: this.process == "work" + +append_check: batch_index() < 100 +``` + +### `append_map` + +Bloblang expression used to update the accumulated value. It receives the current value as `this.cached` and the new message as `this.current`. + + +Type: `string` + +```yml +# Examples + +append_map: root = this.cached.append(this.current) + +append_map: 'root = {"count": this.cached.count + 1, "total": this.cached.total + this.current.value}' +``` + +### `flush_check` + +Bloblang expression that must evaluate to `true` to emit the accumulated value and potentially clear the cache. + + +Type: `string` + +```yml +# Examples + +flush_check: this.process == "end" + +flush_check: batch_index() == 0 && message_index() > 0 && batch_index() % 100 == 0 +``` + +### `flush_deletes` + +When `true`, the cache entry is deleted after flushing. Defaults to `false`. + + +Type: `bool` +Default: `false` + +### `flush_map` + +Bloblang expression to transform the accumulated value before emitting. Defaults to `root`. + + +Type: `string` +Default: `"root = this"` + +```yml +# Examples + +flush_map: root = this + +flush_map: 'root = {"result": this}' + +flush_map: root.items = this +``` + +### `ttl` + +The TTL of each individual item as a duration string. After this period an item will be eligible for removal during the next compaction. Not all caches support per-key TTLs, those that do will have a configuration field `default_ttl`, and those that do not will fall back to their generally configured TTL setting. +This field supports [interpolation functions](/docs/configuration/interpolation#bloblang-queries). + + +Type: `string` + +```yml +# Examples + +ttl: 60s + +ttl: 5m + +ttl: 36h +``` + + From 5b8dd31156b4855f2b857d965583f7173dd79776 Mon Sep 17 00:00:00 2001 From: lublak Date: Tue, 10 Mar 2026 23:46:15 +0100 Subject: [PATCH 05/24] if new messages are empty, just return nil --- internal/impl/pure/processor_cache_collector.go | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/internal/impl/pure/processor_cache_collector.go b/internal/impl/pure/processor_cache_collector.go index 1ff0c7a76b..092bab2821 100644 --- a/internal/impl/pure/processor_cache_collector.go +++ b/internal/impl/pure/processor_cache_collector.go @@ -320,7 +320,11 @@ func (cc *cacheCollector) ProcessBatch(ctx context.Context, batch service.Messag } } - return []service.MessageBatch{newMsgs}, nil + if len(newMsgs) > 0 { + return []service.MessageBatch{newMsgs}, nil + } + + return nil, nil } func (cc *cacheCollector) Close(ctx context.Context) error { From 0ca5ab2420d55483dac12573fe29e12a9f0eb091 Mon Sep 17 00:00:00 2001 From: lublak <44057030+lublak@users.noreply.github.com> Date: Wed, 11 Mar 2026 14:35:50 +0100 Subject: [PATCH 06/24] fix multiple implementation issues --- .../impl/pure/processor_cache_collector.go | 55 +++++++++++++------ 1 file changed, 37 insertions(+), 18 deletions(-) diff --git a/internal/impl/pure/processor_cache_collector.go b/internal/impl/pure/processor_cache_collector.go index 092bab2821..665c9f4e15 100644 --- a/internal/impl/pure/processor_cache_collector.go +++ b/internal/impl/pure/processor_cache_collector.go @@ -136,10 +136,7 @@ func newCacheCollector(conf *service.ParsedConfig, mgr *service.Resources) (*cac return nil, err } - ttl, err := conf.FieldInterpolatedString(cacheCollectorPFieldTTL) - if err != nil { - return nil, err - } + ttl, _ := conf.FieldInterpolatedString(cacheCollectorPFieldTTL) return &cacheCollector{ key: key, @@ -159,8 +156,8 @@ func newCacheCollector(conf *service.ParsedConfig, mgr *service.Resources) (*cac } type cacheCollectorAppendMessageData struct { - cached json.RawMessage - current json.RawMessage + Cached json.RawMessage `json:"cached"` + Current json.RawMessage `json:"current"` } func (cc *cacheCollector) ProcessBatch(ctx context.Context, batch service.MessageBatch) ([]service.MessageBatch, error) { @@ -173,7 +170,7 @@ func (cc *cacheCollector) ProcessBatch(ctx context.Context, batch service.Messag var ttlInterp *service.MessageBatchInterpolationExecutor if cc.ttl != nil { - keyInterp = batch.InterpolationExecutor(cc.ttl) + ttlInterp = batch.InterpolationExecutor(cc.ttl) } var appendCheck *service.MessageBatchBloblangExecutor @@ -181,11 +178,21 @@ func (cc *cacheCollector) ProcessBatch(ctx context.Context, batch service.Messag appendCheck = batch.BloblangExecutor(cc.appendCheck) } + var appendMap *service.MessageBatchBloblangExecutor + if cc.appendMap != nil { + appendMap = batch.BloblangExecutor(cc.appendMap) + } + var flushCheck *service.MessageBatchBloblangExecutor if cc.flushCheck != nil { flushCheck = batch.BloblangExecutor(cc.flushCheck) } + var flushMap *service.MessageBatchBloblangExecutor + if cc.flushMap != nil { + flushMap = batch.BloblangExecutor(cc.flushMap) + } + var init *service.MessageBatchBloblangExecutor if cc.init != nil { init = batch.BloblangExecutor(cc.init) @@ -197,9 +204,12 @@ func (cc *cacheCollector) ProcessBatch(ctx context.Context, batch service.Messag return nil, fmt.Errorf("key evaluation error: %w", err) } - ttls, err := ttlInterp.TryString(i) - if err != nil { - return nil, err + var ttls string + if ttlInterp != nil { + ttls, err = ttlInterp.TryString(i) + if err != nil { + return nil, err + } } var ttl *time.Duration @@ -232,6 +242,7 @@ func (cc *cacheCollector) ProcessBatch(ctx context.Context, batch service.Messag if err != nil { if errors.Is(err, service.ErrKeyNotFound) { cachedValue = nil + err = nil } else { err = fmt.Errorf("failed to get cache key '%s': %v", key, err) } @@ -262,13 +273,21 @@ func (cc *cacheCollector) ProcessBatch(ctx context.Context, batch service.Messag return nil, err } - appendMsg := service.NewMessage(nil) - appendMsg.SetStructured(cacheCollectorAppendMessageData{ - cached: cachedValue, - current: currentValue, + appendMsgJson, err := json.Marshal(cacheCollectorAppendMessageData{ + Cached: json.RawMessage(cachedValue), + Current: json.RawMessage(currentValue), }) - appendResult, err := appendMsg.BloblangQuery(cc.appendMap) + if err != nil { + return nil, err + } + + msg.SetBytes(appendMsgJson) + + appendResult, err := appendMap.Query(i) + + msg.SetBytes(currentValue) + if err != nil { return nil, err } @@ -293,9 +312,9 @@ func (cc *cacheCollector) ProcessBatch(ctx context.Context, batch service.Messag } if processFlush { - flushMsg := service.NewMessage(cachedValue) + msg.SetBytes(cachedValue) - flushResult, err := flushMsg.BloblangQuery(cc.appendMap) + msg, err = flushMap.Query(i) if err != nil { return nil, err } @@ -315,7 +334,7 @@ func (cc *cacheCollector) ProcessBatch(ctx context.Context, batch service.Messag } } - newMsgs = append(newMsgs, flushResult) + newMsgs = append(newMsgs, msg) } } } From 8ae36a2e4e9d9fbc2e52d3031ce8216bba0ca905 Mon Sep 17 00:00:00 2001 From: lublak <44057030+lublak@users.noreply.github.com> Date: Wed, 11 Mar 2026 14:35:55 +0100 Subject: [PATCH 07/24] add tests --- .../pure/processor_cache_collector_test.go | 180 ++++++++++++++++++ 1 file changed, 180 insertions(+) create mode 100644 internal/impl/pure/processor_cache_collector_test.go diff --git a/internal/impl/pure/processor_cache_collector_test.go b/internal/impl/pure/processor_cache_collector_test.go new file mode 100644 index 0000000000..6d36b81bb7 --- /dev/null +++ b/internal/impl/pure/processor_cache_collector_test.go @@ -0,0 +1,180 @@ +package pure + +import ( + "testing" + + "github.com/stretchr/testify/assert" + + "github.com/stretchr/testify/require" + + "github.com/warpstreamlabs/bento/public/service" +) + +func cacheCollectorProc(confStr string) (*cacheCollector, error) { + pConf, err := cacheCollectorSpec().ParseYAML(confStr, nil) + if err != nil { + return nil, err + } + return newCacheCollector(pConf, service.MockResources(service.MockResourcesOptAddCache("test_cache"))) + +} + +func TestCacheCollectorBasicAccumulation(t *testing.T) { + t.Run("basic_with_delete", func(t *testing.T) { + spec := ` +resource: test_cache +key: test_key +init: root = [{"id":"0"}] +append_check: true +append_map: | + root = this.cached.append(this.current) +flush_check: "batch_index() == 2" +flush_map: | + root.result = this + root.count = this.length() +flush_deletes: true +` + processor, err := cacheCollectorProc(spec) + require.NoError(t, err) + batch := service.MessageBatch{ + service.NewMessage([]byte(`{"id":"1"}`)), + service.NewMessage([]byte(`{"id":"2"}`)), + service.NewMessage([]byte(`{"id":"3"}`)), + } + + results, err := processor.ProcessBatch(t.Context(), batch) + + require.NoError(t, err) + require.Len(t, results, 1) + require.Len(t, results[0], 1) + + result, err := results[0][0].AsBytes() + require.NoError(t, err) + expected := `{"result":[{"id":"0"},{"id":"1"},{"id":"2"},{"id":"3"}],"count":4}` + assert.JSONEq(t, expected, string(result)) + + processor.mgr.AccessCache(t.Context(), "test_cache", func(c service.Cache) { + _, err = c.Get(t.Context(), "test_key") + }) + + require.Error(t, err) + }) + + t.Run("basic_without_delete", func(t *testing.T) { + spec := ` +resource: test_cache +key: test_key +init: root = [{"id":"0"}] +append_check: true +append_map: | + root = this.cached.append(this.current) +flush_check: "batch_index() == 2" +flush_map: | + root.result = this + root.count = this.length() +flush_deletes: false +` + processor, err := cacheCollectorProc(spec) + require.NoError(t, err) + batch := service.MessageBatch{ + service.NewMessage([]byte(`{"id":"1"}`)), + service.NewMessage([]byte(`{"id":"2"}`)), + service.NewMessage([]byte(`{"id":"3"}`)), + } + + results, err := processor.ProcessBatch(t.Context(), batch) + + require.NoError(t, err) + require.Len(t, results, 1) + require.Len(t, results[0], 1) + + result, err := results[0][0].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `{"result":[{"id":"0"},{"id":"1"},{"id":"2"},{"id":"3"}],"count":4}`, string(result)) + + var data []byte + + processor.mgr.AccessCache(t.Context(), "test_cache", func(c service.Cache) { + data, err = c.Get(t.Context(), "test_key") + }) + require.NoError(t, err) + + assert.JSONEq(t, `[{"id":"0"},{"id":"1"},{"id":"2"},{"id":"3"}]`, string(data)) + }) +} + +func TestCacheCollectorWithDifferentKeys(t *testing.T) { + t.Run("test_with_different_keys", func(t *testing.T) { + spec := ` +resource: test_cache +key: test_${! this.group } +init: root = [] +append_check: root = this.end == false +append_map: root = this.cached.append(this.current.data) +flush_check: this.end +flush_deletes: true +` + processor, err := cacheCollectorProc(spec) + require.NoError(t, err) + + batch := service.MessageBatch{ + service.NewMessage([]byte(`{"group":"a","data":{"id":"a1"},"end":false}`)), + service.NewMessage([]byte(`{"group":"a","data":{"id":"a2"},"end":false}`)), + service.NewMessage([]byte(`{"group":"b","data":{"id":"b1"},"end":false}`)), + service.NewMessage([]byte(`{"group":"a","data":{"id":"a3"},"end":false}`)), + service.NewMessage([]byte(`{"group":"b","data":{"id":"b2"},"end":false}`)), + service.NewMessage([]byte(`{"group":"b","data":{"id":"b3"},"end":false}`)), + + // end + service.NewMessage([]byte(`{"group":"a","end":true}`)), + service.NewMessage([]byte(`{"group":"b","end":true}`)), + } + + results, err := processor.ProcessBatch(t.Context(), batch) + + require.NoError(t, err) + require.Len(t, results, 1) + require.Len(t, results[0], 2) + + resultGroupA, err := results[0][0].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `[{"id":"a1"},{"id":"a2"},{"id":"a3"}]`, string(resultGroupA)) + + resultGroupB, err := results[0][1].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `[{"id":"b1"},{"id":"b2"},{"id":"b3"}]`, string(resultGroupB)) + }) +} + +func TestCacheCollectorWithMeta(t *testing.T) { + t.Run("access_meta", func(t *testing.T) { + spec := ` +resource: test_cache +key: test_key +init: root = [] +append_check: true +append_map: | + root = this.cached.append(metadata("meta_data")) +flush_check: true +flush_deletes: false +` + processor, err := cacheCollectorProc(spec) + require.NoError(t, err) + msg := service.NewMessage([]byte(`{"meta":true}`)) + msg.MetaSet("meta_data", "meta_value") + + batch := service.MessageBatch{ + msg, + } + + results, err := processor.ProcessBatch(t.Context(), batch) + + require.NoError(t, err) + require.Len(t, results, 1) + require.Len(t, results[0], 1) + + result, err := results[0][0].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `["meta_value"]`, string(result)) + }) +} From 59f7e77e0bc55c42f8aee1b862b9463ca4aebb92 Mon Sep 17 00:00:00 2001 From: lublak <44057030+lublak@users.noreply.github.com> Date: Wed, 11 Mar 2026 15:02:37 +0100 Subject: [PATCH 08/24] fix linting issues --- internal/impl/pure/processor_cache_collector_test.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/internal/impl/pure/processor_cache_collector_test.go b/internal/impl/pure/processor_cache_collector_test.go index 6d36b81bb7..0d0b3b50f3 100644 --- a/internal/impl/pure/processor_cache_collector_test.go +++ b/internal/impl/pure/processor_cache_collector_test.go @@ -53,10 +53,11 @@ flush_deletes: true expected := `{"result":[{"id":"0"},{"id":"1"},{"id":"2"},{"id":"3"}],"count":4}` assert.JSONEq(t, expected, string(result)) - processor.mgr.AccessCache(t.Context(), "test_cache", func(c service.Cache) { + cerr := processor.mgr.AccessCache(t.Context(), "test_cache", func(c service.Cache) { _, err = c.Get(t.Context(), "test_key") }) + require.Error(t, cerr) require.Error(t, err) }) @@ -94,9 +95,10 @@ flush_deletes: false var data []byte - processor.mgr.AccessCache(t.Context(), "test_cache", func(c service.Cache) { + cerr := processor.mgr.AccessCache(t.Context(), "test_cache", func(c service.Cache) { data, err = c.Get(t.Context(), "test_key") }) + require.NoError(t, cerr) require.NoError(t, err) assert.JSONEq(t, `[{"id":"0"},{"id":"1"},{"id":"2"},{"id":"3"}]`, string(data)) From 517e5d6fed32e618b4dd3b0a2a7a29186f5ec9b6 Mon Sep 17 00:00:00 2001 From: lublak <44057030+lublak@users.noreply.github.com> Date: Wed, 11 Mar 2026 15:06:45 +0100 Subject: [PATCH 09/24] fix cli linting --- internal/impl/pure/processor_cache_collector_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/impl/pure/processor_cache_collector_test.go b/internal/impl/pure/processor_cache_collector_test.go index 0d0b3b50f3..bd72b2368c 100644 --- a/internal/impl/pure/processor_cache_collector_test.go +++ b/internal/impl/pure/processor_cache_collector_test.go @@ -1,4 +1,4 @@ -package pure +package pure_test import ( "testing" From fc80d9d1d8b2db6c067498a0f3b06c5d4bb821f0 Mon Sep 17 00:00:00 2001 From: lublak Date: Wed, 11 Mar 2026 19:08:30 +0100 Subject: [PATCH 10/24] clean up code and tests --- .../impl/pure/processor_cache_collector.go | 18 +- .../pure/processor_cache_collector_test.go | 365 +++++++++--------- 2 files changed, 190 insertions(+), 193 deletions(-) diff --git a/internal/impl/pure/processor_cache_collector.go b/internal/impl/pure/processor_cache_collector.go index 665c9f4e15..7fd963724e 100644 --- a/internal/impl/pure/processor_cache_collector.go +++ b/internal/impl/pure/processor_cache_collector.go @@ -23,7 +23,7 @@ const ( cacheCollectorPFieldTTL = "ttl" ) -func cacheCollectorSpec() *service.ConfigSpec { +func CacheCollectorProcessorSpec() *service.ConfigSpec { return service.NewConfigSpec(). Categories("Mapping"). Stable(). @@ -69,7 +69,7 @@ The `+"`append_map`"+` bloblang expression can access both the current cached va ) } -type cacheCollector struct { +type cacheCollectorProcessor struct { cacheName string key *service.InterpolatedString @@ -85,17 +85,13 @@ type cacheCollector struct { } func init() { - err := service.RegisterBatchProcessor( - "cache_collector", cacheCollectorSpec(), - func(conf *service.ParsedConfig, mgr *service.Resources) (service.BatchProcessor, error) { - return newCacheCollector(conf, mgr) - }) + err := service.RegisterBatchProcessor("cache_collector", CacheCollectorProcessorSpec(), NewCacheCollectorFromConfig) if err != nil { panic(err) } } -func newCacheCollector(conf *service.ParsedConfig, mgr *service.Resources) (*cacheCollector, error) { +func NewCacheCollectorFromConfig(conf *service.ParsedConfig, mgr *service.Resources) (service.BatchProcessor, error) { resource, err := conf.FieldString(cacheCollectorPFieldResource) if err != nil { return nil, err @@ -138,7 +134,7 @@ func newCacheCollector(conf *service.ParsedConfig, mgr *service.Resources) (*cac ttl, _ := conf.FieldInterpolatedString(cacheCollectorPFieldTTL) - return &cacheCollector{ + return &cacheCollectorProcessor{ key: key, init: init, appendCheck: appendCheck, @@ -160,7 +156,7 @@ type cacheCollectorAppendMessageData struct { Current json.RawMessage `json:"current"` } -func (cc *cacheCollector) ProcessBatch(ctx context.Context, batch service.MessageBatch) ([]service.MessageBatch, error) { +func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch service.MessageBatch) ([]service.MessageBatch, error) { var newMsgs []*service.Message var keyInterp *service.MessageBatchInterpolationExecutor @@ -346,6 +342,6 @@ func (cc *cacheCollector) ProcessBatch(ctx context.Context, batch service.Messag return nil, nil } -func (cc *cacheCollector) Close(ctx context.Context) error { +func (cc *cacheCollectorProcessor) Close(ctx context.Context) error { return nil } diff --git a/internal/impl/pure/processor_cache_collector_test.go b/internal/impl/pure/processor_cache_collector_test.go index bd72b2368c..f4002ad6bd 100644 --- a/internal/impl/pure/processor_cache_collector_test.go +++ b/internal/impl/pure/processor_cache_collector_test.go @@ -1,182 +1,183 @@ -package pure_test - -import ( - "testing" - - "github.com/stretchr/testify/assert" - - "github.com/stretchr/testify/require" - - "github.com/warpstreamlabs/bento/public/service" -) - -func cacheCollectorProc(confStr string) (*cacheCollector, error) { - pConf, err := cacheCollectorSpec().ParseYAML(confStr, nil) - if err != nil { - return nil, err - } - return newCacheCollector(pConf, service.MockResources(service.MockResourcesOptAddCache("test_cache"))) - -} - -func TestCacheCollectorBasicAccumulation(t *testing.T) { - t.Run("basic_with_delete", func(t *testing.T) { - spec := ` -resource: test_cache -key: test_key -init: root = [{"id":"0"}] -append_check: true -append_map: | - root = this.cached.append(this.current) -flush_check: "batch_index() == 2" -flush_map: | - root.result = this - root.count = this.length() -flush_deletes: true -` - processor, err := cacheCollectorProc(spec) - require.NoError(t, err) - batch := service.MessageBatch{ - service.NewMessage([]byte(`{"id":"1"}`)), - service.NewMessage([]byte(`{"id":"2"}`)), - service.NewMessage([]byte(`{"id":"3"}`)), - } - - results, err := processor.ProcessBatch(t.Context(), batch) - - require.NoError(t, err) - require.Len(t, results, 1) - require.Len(t, results[0], 1) - - result, err := results[0][0].AsBytes() - require.NoError(t, err) - expected := `{"result":[{"id":"0"},{"id":"1"},{"id":"2"},{"id":"3"}],"count":4}` - assert.JSONEq(t, expected, string(result)) - - cerr := processor.mgr.AccessCache(t.Context(), "test_cache", func(c service.Cache) { - _, err = c.Get(t.Context(), "test_key") - }) - - require.Error(t, cerr) - require.Error(t, err) - }) - - t.Run("basic_without_delete", func(t *testing.T) { - spec := ` -resource: test_cache -key: test_key -init: root = [{"id":"0"}] -append_check: true -append_map: | - root = this.cached.append(this.current) -flush_check: "batch_index() == 2" -flush_map: | - root.result = this - root.count = this.length() -flush_deletes: false -` - processor, err := cacheCollectorProc(spec) - require.NoError(t, err) - batch := service.MessageBatch{ - service.NewMessage([]byte(`{"id":"1"}`)), - service.NewMessage([]byte(`{"id":"2"}`)), - service.NewMessage([]byte(`{"id":"3"}`)), - } - - results, err := processor.ProcessBatch(t.Context(), batch) - - require.NoError(t, err) - require.Len(t, results, 1) - require.Len(t, results[0], 1) - - result, err := results[0][0].AsBytes() - require.NoError(t, err) - assert.JSONEq(t, `{"result":[{"id":"0"},{"id":"1"},{"id":"2"},{"id":"3"}],"count":4}`, string(result)) - - var data []byte - - cerr := processor.mgr.AccessCache(t.Context(), "test_cache", func(c service.Cache) { - data, err = c.Get(t.Context(), "test_key") - }) - require.NoError(t, cerr) - require.NoError(t, err) - - assert.JSONEq(t, `[{"id":"0"},{"id":"1"},{"id":"2"},{"id":"3"}]`, string(data)) - }) -} - -func TestCacheCollectorWithDifferentKeys(t *testing.T) { - t.Run("test_with_different_keys", func(t *testing.T) { - spec := ` -resource: test_cache -key: test_${! this.group } -init: root = [] -append_check: root = this.end == false -append_map: root = this.cached.append(this.current.data) -flush_check: this.end -flush_deletes: true -` - processor, err := cacheCollectorProc(spec) - require.NoError(t, err) - - batch := service.MessageBatch{ - service.NewMessage([]byte(`{"group":"a","data":{"id":"a1"},"end":false}`)), - service.NewMessage([]byte(`{"group":"a","data":{"id":"a2"},"end":false}`)), - service.NewMessage([]byte(`{"group":"b","data":{"id":"b1"},"end":false}`)), - service.NewMessage([]byte(`{"group":"a","data":{"id":"a3"},"end":false}`)), - service.NewMessage([]byte(`{"group":"b","data":{"id":"b2"},"end":false}`)), - service.NewMessage([]byte(`{"group":"b","data":{"id":"b3"},"end":false}`)), - - // end - service.NewMessage([]byte(`{"group":"a","end":true}`)), - service.NewMessage([]byte(`{"group":"b","end":true}`)), - } - - results, err := processor.ProcessBatch(t.Context(), batch) - - require.NoError(t, err) - require.Len(t, results, 1) - require.Len(t, results[0], 2) - - resultGroupA, err := results[0][0].AsBytes() - require.NoError(t, err) - assert.JSONEq(t, `[{"id":"a1"},{"id":"a2"},{"id":"a3"}]`, string(resultGroupA)) - - resultGroupB, err := results[0][1].AsBytes() - require.NoError(t, err) - assert.JSONEq(t, `[{"id":"b1"},{"id":"b2"},{"id":"b3"}]`, string(resultGroupB)) - }) -} - -func TestCacheCollectorWithMeta(t *testing.T) { - t.Run("access_meta", func(t *testing.T) { - spec := ` -resource: test_cache -key: test_key -init: root = [] -append_check: true -append_map: | - root = this.cached.append(metadata("meta_data")) -flush_check: true -flush_deletes: false -` - processor, err := cacheCollectorProc(spec) - require.NoError(t, err) - msg := service.NewMessage([]byte(`{"meta":true}`)) - msg.MetaSet("meta_data", "meta_value") - - batch := service.MessageBatch{ - msg, - } - - results, err := processor.ProcessBatch(t.Context(), batch) - - require.NoError(t, err) - require.Len(t, results, 1) - require.Len(t, results[0], 1) - - result, err := results[0][0].AsBytes() - require.NoError(t, err) - assert.JSONEq(t, `["meta_value"]`, string(result)) - }) -} +package pure_test + +import ( + "testing" + + "github.com/stretchr/testify/assert" + + "github.com/stretchr/testify/require" + + "github.com/warpstreamlabs/bento/public/service" + + ipure "github.com/warpstreamlabs/bento/internal/impl/pure" +) + +func cacheCollectorProc(confStr string) (service.BatchProcessor, *service.Resources, error) { + pConf, err := ipure.CacheCollectorProcessorSpec().ParseYAML(confStr, nil) + if err != nil { + return nil, nil, err + } + mgr := service.MockResources(service.MockResourcesOptAddCache("test_cache")) + proc, err := ipure.NewCacheCollectorFromConfig(pConf, mgr) + return proc, mgr, err +} + +func TestCacheCollectorBasicAccumulationDeleteCache(t *testing.T) { + spec := ` +resource: test_cache +key: test_key +init: root = [{"id":"0"}] +append_check: true +append_map: | + root = this.cached.append(this.current) +flush_check: "batch_index() == 2" +flush_map: | + root.result = this + root.count = this.length() +flush_deletes: true +` + processor, mgr, err := cacheCollectorProc(spec) + require.NoError(t, err) + batch := service.MessageBatch{ + service.NewMessage([]byte(`{"id":"1"}`)), + service.NewMessage([]byte(`{"id":"2"}`)), + service.NewMessage([]byte(`{"id":"3"}`)), + } + + results, err := processor.ProcessBatch(t.Context(), batch) + + require.NoError(t, err) + require.Len(t, results, 1) + require.Len(t, results[0], 1) + + result, err := results[0][0].AsBytes() + require.NoError(t, err) + expected := `{"result":[{"id":"0"},{"id":"1"},{"id":"2"},{"id":"3"}],"count":4}` + assert.JSONEq(t, expected, string(result)) + + cerr := mgr.AccessCache(t.Context(), "test_cache", func(c service.Cache) { + _, err = c.Get(t.Context(), "test_key") + }) + + require.NoError(t, cerr) + require.Error(t, err) +} + +func TestCacheCollectorBasicAccumulatioKeepCache(t *testing.T) { + spec := ` +resource: test_cache +key: test_key +init: root = [{"id":"0"}] +append_check: true +append_map: | + root = this.cached.append(this.current) +flush_check: "batch_index() == 2" +flush_map: | + root.result = this + root.count = this.length() +flush_deletes: false +` + processor, mgr, err := cacheCollectorProc(spec) + require.NoError(t, err) + batch := service.MessageBatch{ + service.NewMessage([]byte(`{"id":"1"}`)), + service.NewMessage([]byte(`{"id":"2"}`)), + service.NewMessage([]byte(`{"id":"3"}`)), + } + + results, err := processor.ProcessBatch(t.Context(), batch) + + require.NoError(t, err) + require.Len(t, results, 1) + require.Len(t, results[0], 1) + + result, err := results[0][0].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `{"result":[{"id":"0"},{"id":"1"},{"id":"2"},{"id":"3"}],"count":4}`, string(result)) + + var data []byte + + cerr := mgr.AccessCache(t.Context(), "test_cache", func(c service.Cache) { + data, err = c.Get(t.Context(), "test_key") + }) + require.NoError(t, cerr) + require.NoError(t, err) + + assert.JSONEq(t, `[{"id":"0"},{"id":"1"},{"id":"2"},{"id":"3"}]`, string(data)) +} + +func TestCacheCollectorWithDifferentKeys(t *testing.T) { + t.Run("test_with_different_keys", func(t *testing.T) { + spec := ` +resource: test_cache +key: test_${! this.group } +init: root = [] +append_check: root = this.end == false +append_map: root = this.cached.append(this.current.data) +flush_check: this.end +flush_deletes: true +` + processor, _, err := cacheCollectorProc(spec) + require.NoError(t, err) + + batch := service.MessageBatch{ + service.NewMessage([]byte(`{"group":"a","data":{"id":"a1"},"end":false}`)), + service.NewMessage([]byte(`{"group":"a","data":{"id":"a2"},"end":false}`)), + service.NewMessage([]byte(`{"group":"b","data":{"id":"b1"},"end":false}`)), + service.NewMessage([]byte(`{"group":"a","data":{"id":"a3"},"end":false}`)), + service.NewMessage([]byte(`{"group":"b","data":{"id":"b2"},"end":false}`)), + service.NewMessage([]byte(`{"group":"b","data":{"id":"b3"},"end":false}`)), + + // end + service.NewMessage([]byte(`{"group":"a","end":true}`)), + service.NewMessage([]byte(`{"group":"b","end":true}`)), + } + + results, err := processor.ProcessBatch(t.Context(), batch) + + require.NoError(t, err) + require.Len(t, results, 1) + require.Len(t, results[0], 2) + + resultGroupA, err := results[0][0].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `[{"id":"a1"},{"id":"a2"},{"id":"a3"}]`, string(resultGroupA)) + + resultGroupB, err := results[0][1].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `[{"id":"b1"},{"id":"b2"},{"id":"b3"}]`, string(resultGroupB)) + }) +} + +func TestCacheCollectorWithMeta(t *testing.T) { + t.Run("access_meta", func(t *testing.T) { + spec := ` +resource: test_cache +key: test_key +init: root = [] +append_check: true +append_map: | + root = this.cached.append(metadata("meta_data")) +flush_check: true +flush_deletes: false +` + processor, _, err := cacheCollectorProc(spec) + require.NoError(t, err) + msg := service.NewMessage([]byte(`{"meta":true}`)) + msg.MetaSet("meta_data", "meta_value") + + batch := service.MessageBatch{ + msg, + } + + results, err := processor.ProcessBatch(t.Context(), batch) + + require.NoError(t, err) + require.Len(t, results, 1) + require.Len(t, results[0], 1) + + result, err := results[0][0].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `["meta_value"]`, string(result)) + }) +} From 5dc63796c4bd3cbc356a75baa32886a96926c122 Mon Sep 17 00:00:00 2001 From: lublak Date: Wed, 11 Mar 2026 19:10:09 +0100 Subject: [PATCH 11/24] use better testing names --- internal/impl/pure/processor_cache_collector_test.go | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/internal/impl/pure/processor_cache_collector_test.go b/internal/impl/pure/processor_cache_collector_test.go index f4002ad6bd..376ca20c95 100644 --- a/internal/impl/pure/processor_cache_collector_test.go +++ b/internal/impl/pure/processor_cache_collector_test.go @@ -22,7 +22,7 @@ func cacheCollectorProc(confStr string) (service.BatchProcessor, *service.Resour return proc, mgr, err } -func TestCacheCollectorBasicAccumulationDeleteCache(t *testing.T) { +func TestCacheCollectorProcessor_BasicAccumulationDeleteCache(t *testing.T) { spec := ` resource: test_cache key: test_key @@ -63,7 +63,7 @@ flush_deletes: true require.Error(t, err) } -func TestCacheCollectorBasicAccumulatioKeepCache(t *testing.T) { +func TestCacheCollectorProcessor_BasicAccumulatioKeepCache(t *testing.T) { spec := ` resource: test_cache key: test_key @@ -106,7 +106,7 @@ flush_deletes: false assert.JSONEq(t, `[{"id":"0"},{"id":"1"},{"id":"2"},{"id":"3"}]`, string(data)) } -func TestCacheCollectorWithDifferentKeys(t *testing.T) { +func TestCacheCollectorProcessor_WithDifferentKeys(t *testing.T) { t.Run("test_with_different_keys", func(t *testing.T) { spec := ` resource: test_cache @@ -149,7 +149,7 @@ flush_deletes: true }) } -func TestCacheCollectorWithMeta(t *testing.T) { +func TestCacheCollectorProcessor_WithMeta(t *testing.T) { t.Run("access_meta", func(t *testing.T) { spec := ` resource: test_cache @@ -159,7 +159,7 @@ append_check: true append_map: | root = this.cached.append(metadata("meta_data")) flush_check: true -flush_deletes: false +flush_deletes: true ` processor, _, err := cacheCollectorProc(spec) require.NoError(t, err) From c64aa22df73c3d1c4a02921462efb78d5acce0d5 Mon Sep 17 00:00:00 2001 From: lublak Date: Wed, 11 Mar 2026 19:39:43 +0100 Subject: [PATCH 12/24] add cleanup function --- internal/impl/pure/processor_cache_collector_test.go | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/internal/impl/pure/processor_cache_collector_test.go b/internal/impl/pure/processor_cache_collector_test.go index 376ca20c95..7daf83d1a7 100644 --- a/internal/impl/pure/processor_cache_collector_test.go +++ b/internal/impl/pure/processor_cache_collector_test.go @@ -38,6 +38,9 @@ flush_deletes: true ` processor, mgr, err := cacheCollectorProc(spec) require.NoError(t, err) + + t.Cleanup(func() { require.NoError(t, processor.Close(t.Context())) }) + batch := service.MessageBatch{ service.NewMessage([]byte(`{"id":"1"}`)), service.NewMessage([]byte(`{"id":"2"}`)), @@ -79,6 +82,9 @@ flush_deletes: false ` processor, mgr, err := cacheCollectorProc(spec) require.NoError(t, err) + + t.Cleanup(func() { require.NoError(t, processor.Close(t.Context())) }) + batch := service.MessageBatch{ service.NewMessage([]byte(`{"id":"1"}`)), service.NewMessage([]byte(`{"id":"2"}`)), @@ -163,6 +169,9 @@ flush_deletes: true ` processor, _, err := cacheCollectorProc(spec) require.NoError(t, err) + + t.Cleanup(func() { require.NoError(t, processor.Close(t.Context())) }) + msg := service.NewMessage([]byte(`{"meta":true}`)) msg.MetaSet("meta_data", "meta_value") From 7222828e6dcb81ed1dc709b95abf70f102470224 Mon Sep 17 00:00:00 2001 From: lublak <44057030+lublak@users.noreply.github.com> Date: Thu, 12 Mar 2026 13:09:50 +0100 Subject: [PATCH 13/24] add support for init_check and filter_untreated --- .../impl/pure/processor_cache_collector.go | 227 ++++++++++++------ .../pure/processor_cache_collector_test.go | 114 ++++++++- .../components/processors/cache_collector.md | 49 +++- 3 files changed, 298 insertions(+), 92 deletions(-) diff --git a/internal/impl/pure/processor_cache_collector.go b/internal/impl/pure/processor_cache_collector.go index 7fd963724e..0e77f6e0be 100644 --- a/internal/impl/pure/processor_cache_collector.go +++ b/internal/impl/pure/processor_cache_collector.go @@ -12,28 +12,30 @@ import ( ) const ( - cacheCollectorPFieldResource = "resource" - cacheCollectorPFieldKey = "key" - cacheCollectorPFieldInit = "init" - cacheCollectorPFieldAppendCheck = "append_check" - cacheCollectorPFieldAppendMap = "append_map" - cacheCollectorPFieldFlushCheck = "flush_check" - cacheCollectorPFieldFlushDeletes = "flush_deletes" - cacheCollectorPFieldFlushMap = "flush_map" - cacheCollectorPFieldTTL = "ttl" + cacheCollectorPFieldResource = "resource" + cacheCollectorPFieldKey = "key" + cacheCollectorPFieldInitCheck = "init_check" + cacheCollectorPFieldInitMap = "init_map" + cacheCollectorPFieldAppendCheck = "append_check" + cacheCollectorPFieldAppendMap = "append_map" + cacheCollectorPFieldFlushCheck = "flush_check" + cacheCollectorPFieldFlushDeletes = "flush_deletes" + cacheCollectorPFieldFlushMap = "flush_map" + cacheCollectorPFieldFilterUntreated = "filter_untreated" + cacheCollectorPFieldTTL = "ttl" ) func CacheCollectorProcessorSpec() *service.ConfigSpec { return service.NewConfigSpec(). Categories("Mapping"). - Stable(). + Beta(). Summary("Accumulates messages across batch boundaries using a cache resource, allowing you to build up state before emitting a final result as structed data."). Description(` This processor works by storing an accumulated value in a cache, which is updated on each message based on bloblang expressions. It supports three phases: -1. `+"`init`"+`: When the cache key doesn't exist, this bloblang expression is used to initialize the value. -2. `+"`append_check`"+`: For each message, if this expression evaluates to true, the value is updated using `+"`append_map`"+`. -3. `+"`flush_check`"+`: When this expression evaluates to true, the accumulated value is emitted as a new message and the cache is optionally cleared. +1. `+"`init_check`"+`: When the cache key doesn't exist, if this expression evaluates to true, the value is initialize using `+"`init_map`"+`. +2. `+"`append_check`"+`: For each message, if this expression evaluates to true and the cache was initialized, the value is updated using `+"`append_map`"+`. +3. `+"`flush_check`"+`: When this expression evaluates to true and the cache was initialized, the accumulated value is emitted as a new message and the cache is optionally cleared. The `+"`append_map`"+` bloblang expression can access both the current cached value as `+"`this.cached`"+` and the current message as `+"`this.current`"+`.`). Fields( @@ -41,7 +43,11 @@ The `+"`append_map`"+` bloblang expression can access both the current cached va Description("The [`cache` resource](/docs/components/caches/about) to use for storing accumulated state."), service.NewInterpolatedStringField(cacheCollectorPFieldKey). Description("A key for the cache entry. This should be consistent across messages that should be grouped together."), - service.NewBloblangField(cacheCollectorPFieldInit). + service.NewBloblangField(cacheCollectorPFieldInitCheck). + Description("Bloblang expression that must evaluate to `true` for a message to initialize the cache."). + Default("true"). + Examples(`this.process == "start"`), + service.NewBloblangField(cacheCollectorPFieldInitMap). Description("Bloblang expression to initialize the value when the cache key doesn't exist. Defaults to an empty array."). Default("root = []"). Examples("root = []", `root = {"items": []}`, `root = {"count": 0, "total": 0}`), @@ -61,6 +67,9 @@ The `+"`append_map`"+` bloblang expression can access both the current cached va Description("Bloblang expression to transform the accumulated value before emitting. Defaults to `root`."). Default("root = this"). Examples(`root = this`, `root = {"result": this}`, `root.items = this`), + service.NewBoolField(cacheCollectorPFieldFilterUntreated). + Description("When `true`, messages that have not been collected are automatically filtered. Defaults to `false`."). + Default(false), service.NewInterpolatedStringField(cacheCollectorPFieldTTL). Description("The TTL of each individual item as a duration string. After this period an item will be eligible for removal during the next compaction. Not all caches support per-key TTLs, those that do will have a configuration field `default_ttl`, and those that do not will fall back to their generally configured TTL setting."). Examples("60s", "5m", "36h"). @@ -73,13 +82,17 @@ type cacheCollectorProcessor struct { cacheName string key *service.InterpolatedString - init *bloblang.Executor + initCheck *bloblang.Executor + initMap *bloblang.Executor appendCheck *bloblang.Executor appendMap *bloblang.Executor flushCheck *bloblang.Executor flushDeletes bool flushMap *bloblang.Executor - ttl *service.InterpolatedString + + filterUntreated bool + + ttl *service.InterpolatedString mgr *service.Resources } @@ -102,7 +115,12 @@ func NewCacheCollectorFromConfig(conf *service.ParsedConfig, mgr *service.Resour return nil, err } - init, err := conf.FieldBloblang(cacheCollectorPFieldInit) + initCheck, err := conf.FieldBloblang(cacheCollectorPFieldInitCheck) + if err != nil { + return nil, err + } + + initMap, err := conf.FieldBloblang(cacheCollectorPFieldInitMap) if err != nil { return nil, err } @@ -132,16 +150,24 @@ func NewCacheCollectorFromConfig(conf *service.ParsedConfig, mgr *service.Resour return nil, err } + filterUntreated, err := conf.FieldBool(cacheCollectorPFieldFilterUntreated) + if err != nil { + return nil, err + } + ttl, _ := conf.FieldInterpolatedString(cacheCollectorPFieldTTL) return &cacheCollectorProcessor{ key: key, - init: init, + initCheck: initCheck, + initMap: initMap, appendCheck: appendCheck, appendMap: appendMap, flushCheck: flushCheck, flushMap: flushMap, + filterUntreated: filterUntreated, + cacheName: resource, flushDeletes: flushDeletes, @@ -151,7 +177,7 @@ func NewCacheCollectorFromConfig(conf *service.ParsedConfig, mgr *service.Resour }, nil } -type cacheCollectorAppendMessageData struct { +type cacheCollectorMessageData struct { Cached json.RawMessage `json:"cached"` Current json.RawMessage `json:"current"` } @@ -189,9 +215,14 @@ func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch servi flushMap = batch.BloblangExecutor(cc.flushMap) } - var init *service.MessageBatchBloblangExecutor - if cc.init != nil { - init = batch.BloblangExecutor(cc.init) + var initCheck *service.MessageBatchBloblangExecutor + if cc.initCheck != nil { + initCheck = batch.BloblangExecutor(cc.initCheck) + } + + var initMap *service.MessageBatchBloblangExecutor + if cc.initMap != nil { + initMap = batch.BloblangExecutor(cc.initMap) } for i, msg := range batch { @@ -218,24 +249,29 @@ func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch servi ttl = &td } - var processAppend bool - var processFlush bool + processInit, err := initCheck.QueryBool(i) + if err != nil { + return nil, fmt.Errorf("init_check evaluation error: %w", err) + } - processAppend, err = appendCheck.QueryBool(i) + processAppend, err := appendCheck.QueryBool(i) if err != nil { return nil, fmt.Errorf("append_check evaluation error: %w", err) } - processFlush, err = flushCheck.QueryBool(i) + processFlush, err := flushCheck.QueryBool(i) if err != nil { return nil, fmt.Errorf("flush_check evaluation error: %w", err) } if processAppend || processFlush { var cachedValue []byte + cacheValueExists := true + if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { cachedValue, err = cache.Get(ctx, key) if err != nil { + cacheValueExists = false if errors.Is(err, service.ErrKeyNotFound) { cachedValue = nil err = nil @@ -251,75 +287,56 @@ func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch servi return nil, err } - if cachedValue == nil { - initMsg, err := init.Query(i) - if err != nil { - return nil, fmt.Errorf("init evaluation error: %w", err) - } - initData, err := initMsg.AsBytes() - if err != nil { - return nil, fmt.Errorf("init data error: %w", err) + if !cacheValueExists { + if processInit { + initMsg, err := initMap.Query(i) + if err != nil { + return nil, fmt.Errorf("init_map evaluation error: %w", err) + } + initData, err := initMsg.AsBytes() + if err != nil { + return nil, fmt.Errorf("init data error: %w", err) + } + cachedValue = initData + cacheValueExists = true } - cachedValue = initData } - if processAppend { + if cacheValueExists { currentValue, err := msg.AsBytes() if err != nil { return nil, err } - appendMsgJson, err := json.Marshal(cacheCollectorAppendMessageData{ - Cached: json.RawMessage(cachedValue), - Current: json.RawMessage(currentValue), - }) + if processAppend { + appendMsgJson, err := json.Marshal(cacheCollectorMessageData{ + Cached: json.RawMessage(cachedValue), + Current: json.RawMessage(currentValue), + }) - if err != nil { - return nil, err - } + if err != nil { + return nil, err + } - msg.SetBytes(appendMsgJson) + msg.SetBytes(appendMsgJson) - appendResult, err := appendMap.Query(i) + appendResult, err := appendMap.Query(i) - msg.SetBytes(currentValue) + msg.SetBytes(currentValue) - if err != nil { - return nil, err - } - - cachedValue, err = appendResult.AsBytes() - if err != nil { - return nil, err - } - - if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { - err = cache.Set(ctx, key, cachedValue, ttl) if err != nil { - err = fmt.Errorf("failed to set cache key '%s': %v", key, err) + return nil, err } - }); cerr != nil { - err = cerr - } - if err != nil { - return nil, err - } - } - - if processFlush { - msg.SetBytes(cachedValue) - - msg, err = flushMap.Query(i) - if err != nil { - return nil, err - } + cachedValue, err = appendResult.AsBytes() + if err != nil { + return nil, err + } - if cc.flushDeletes { if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { - err = cache.Delete(ctx, key) + err = cache.Set(ctx, key, cachedValue, ttl) if err != nil { - err = fmt.Errorf("failed to delete cache key '%s': %v", key, err) + err = fmt.Errorf("failed to set cache key '%s': %v", key, err) } }); cerr != nil { err = cerr @@ -330,8 +347,66 @@ func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch servi } } + if processFlush { + msg.SetBytes(cachedValue) + + msg, err = flushMap.Query(i) + if err != nil { + return nil, err + } + + if cc.flushDeletes { + if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { + err = cache.Delete(ctx, key) + if err != nil { + err = fmt.Errorf("failed to delete cache key '%s': %v", key, err) + } + }); cerr != nil { + err = cerr + } + + if err != nil { + return nil, err + } + } + + fmt.Println("write flush") + + newMsgs = append(newMsgs, msg) + } + } else if !cc.filterUntreated { + fmt.Println("write raw") newMsgs = append(newMsgs, msg) } + } else if processInit { + initMsg, err := initMap.Query(i) + if err != nil { + return nil, fmt.Errorf("init_map evaluation error: %w", err) + } + initData, err := initMsg.AsBytes() + if err != nil { + return nil, fmt.Errorf("init data error: %w", err) + } + + if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { + err = cache.Add(ctx, key, initData, ttl) + if err != nil { + if errors.Is(err, service.ErrKeyAlreadyExists) { + err = nil + } else { + err = fmt.Errorf("failed to set cache key '%s': %v", key, err) + } + } + }); cerr != nil { + err = cerr + } + + if err != nil { + return nil, err + } + } else if !cc.filterUntreated { + fmt.Println("write raw") + newMsgs = append(newMsgs, msg) } } diff --git a/internal/impl/pure/processor_cache_collector_test.go b/internal/impl/pure/processor_cache_collector_test.go index 7daf83d1a7..1d12df8ace 100644 --- a/internal/impl/pure/processor_cache_collector_test.go +++ b/internal/impl/pure/processor_cache_collector_test.go @@ -26,7 +26,7 @@ func TestCacheCollectorProcessor_BasicAccumulationDeleteCache(t *testing.T) { spec := ` resource: test_cache key: test_key -init: root = [{"id":"0"}] +init_map: root = [{"id":"0"}] append_check: true append_map: | root = this.cached.append(this.current) @@ -55,8 +55,7 @@ flush_deletes: true result, err := results[0][0].AsBytes() require.NoError(t, err) - expected := `{"result":[{"id":"0"},{"id":"1"},{"id":"2"},{"id":"3"}],"count":4}` - assert.JSONEq(t, expected, string(result)) + assert.JSONEq(t, `{"result":[{"id":"0"},{"id":"1"},{"id":"2"},{"id":"3"}],"count":4}`, string(result)) cerr := mgr.AccessCache(t.Context(), "test_cache", func(c service.Cache) { _, err = c.Get(t.Context(), "test_key") @@ -70,7 +69,7 @@ func TestCacheCollectorProcessor_BasicAccumulatioKeepCache(t *testing.T) { spec := ` resource: test_cache key: test_key -init: root = [{"id":"0"}] +init_map: root = [{"id":"0"}] append_check: true append_map: | root = this.cached.append(this.current) @@ -117,7 +116,7 @@ func TestCacheCollectorProcessor_WithDifferentKeys(t *testing.T) { spec := ` resource: test_cache key: test_${! this.group } -init: root = [] +init_map: root = [] append_check: root = this.end == false append_map: root = this.cached.append(this.current.data) flush_check: this.end @@ -160,7 +159,7 @@ func TestCacheCollectorProcessor_WithMeta(t *testing.T) { spec := ` resource: test_cache key: test_key -init: root = [] +init_map: root = [] append_check: true append_map: | root = this.cached.append(metadata("meta_data")) @@ -190,3 +189,106 @@ flush_deletes: true assert.JSONEq(t, `["meta_value"]`, string(result)) }) } + +func TestCacheCollectorProcessor_WithInitCheck(t *testing.T) { + spec := ` +resource: test_cache +key: test_key +init_check: this.type == "start" +init_map: root = [] +append_check: this.type == "work" +append_map: | + root = this.cached.append(this.current) +flush_check: this.type == "end" +flush_deletes: true +` + processor, _, err := cacheCollectorProc(spec) + require.NoError(t, err) + + t.Cleanup(func() { require.NoError(t, processor.Close(t.Context())) }) + + batch := service.MessageBatch{ + service.NewMessage([]byte(`{"type":"work","id":"1"}`)), + + // start + service.NewMessage([]byte(`{"type":"start","id":"2"}`)), + service.NewMessage([]byte(`{"type":"work","id":"3"}`)), + service.NewMessage([]byte(`{"type":"work","id":"4"}`)), + service.NewMessage([]byte(`{"type":"ignore","id":"5"}`)), + service.NewMessage([]byte(`{"type":"end","id":"6"}`)), + // end + + service.NewMessage([]byte(`{"type":"work","id":"7"}`)), + service.NewMessage([]byte(`{"type":"end","id":"8"}`)), + } + + results, err := processor.ProcessBatch(t.Context(), batch) + + require.NoError(t, err) + require.Len(t, results, 1) + require.Len(t, results[0], 5) + + result, err := results[0][0].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `{"type":"work","id":"1"}`, string(result)) + + result, err = results[0][1].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `{"type":"ignore","id":"5"}`, string(result)) + + result, err = results[0][2].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `[{"type":"work","id":"3"},{"type":"work","id":"4"}]`, string(result)) + + result, err = results[0][3].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `{"type":"work","id":"7"}`, string(result)) + + result, err = results[0][4].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `{"type":"end","id":"8"}`, string(result)) +} + +func TestCacheCollectorProcessor_WithInitCheckAndFilter(t *testing.T) { + spec := ` +resource: test_cache +key: test_key +init_check: this.type == "start" +init_map: root = [] +append_check: this.type == "work" +append_map: | + root = this.cached.append(this.current) +flush_check: this.type == "end" +flush_deletes: true +filter_untreated: true +` + processor, _, err := cacheCollectorProc(spec) + require.NoError(t, err) + + t.Cleanup(func() { require.NoError(t, processor.Close(t.Context())) }) + + batch := service.MessageBatch{ + service.NewMessage([]byte(`{"type":"work","id":"1"}`)), + + // start + service.NewMessage([]byte(`{"type":"start","id":"2"}`)), + service.NewMessage([]byte(`{"type":"work","id":"3"}`)), + service.NewMessage([]byte(`{"type":"work","id":"4"}`)), + service.NewMessage([]byte(`{"type":"ignore","id":"5"}`)), + service.NewMessage([]byte(`{"type":"end","id":"6"}`)), + // end + + service.NewMessage([]byte(`{"type":"work","id":"7"}`)), + service.NewMessage([]byte(`{"type":"end","id":"8"}`)), + } + + results, err := processor.ProcessBatch(t.Context(), batch) + + require.NoError(t, err) + require.Len(t, results, 1) + require.Len(t, results[0], 1) + + result, err := results[0][0].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `[{"type":"work","id":"3"},{"type":"work","id":"4"}]`, string(result)) +} diff --git a/website/docs/components/processors/cache_collector.md b/website/docs/components/processors/cache_collector.md index 154c5f04c1..df06b30ee7 100644 --- a/website/docs/components/processors/cache_collector.md +++ b/website/docs/components/processors/cache_collector.md @@ -2,7 +2,7 @@ title: cache_collector slug: cache_collector type: processor -status: stable +status: beta categories: ["Mapping"] --- @@ -15,6 +15,9 @@ categories: ["Mapping"] import Tabs from '@theme/Tabs'; import TabItem from '@theme/TabItem'; +:::caution BETA +This component is mostly stable but breaking changes could still be made outside of major version releases if a fundamental problem with the component is found. +::: Accumulates messages across batch boundaries using a cache resource, allowing you to build up state before emitting a final result as structed data. @@ -31,12 +34,14 @@ label: "" cache_collector: resource: "" # No default (required) key: "" # No default (required) - init: root = [] + init_check: "true" + init_map: root = [] append_check: this.process == "work" # No default (required) append_map: root = this.cached.append(this.current) # No default (required) flush_check: this.process == "end" # No default (required) flush_deletes: false flush_map: root = this + filter_untreated: false ``` @@ -48,12 +53,14 @@ label: "" cache_collector: resource: "" # No default (required) key: "" # No default (required) - init: root = [] + init_check: "true" + init_map: root = [] append_check: this.process == "work" # No default (required) append_map: root = this.cached.append(this.current) # No default (required) flush_check: this.process == "end" # No default (required) flush_deletes: false flush_map: root = this + filter_untreated: false ttl: 60s # No default (optional) ``` @@ -62,9 +69,9 @@ cache_collector: This processor works by storing an accumulated value in a cache, which is updated on each message based on bloblang expressions. It supports three phases: -1. `init`: When the cache key doesn't exist, this bloblang expression is used to initialize the value. -2. `append_check`: For each message, if this expression evaluates to true, the value is updated using `append_map`. -3. `flush_check`: When this expression evaluates to true, the accumulated value is emitted as a new message and the cache is optionally cleared. +1. `init_check`: When the cache key doesn't exist, if this expression evaluates to true, the value is initialize using `init_map`. +2. `append_check`: For each message, if this expression evaluates to true and the cache was initialized, the value is updated using `append_map`. +3. `flush_check`: When this expression evaluates to true and the cache was initialized, the accumulated value is emitted as a new message and the cache is optionally cleared. The `append_map` bloblang expression can access both the current cached value as `this.cached` and the current message as `this.current`. @@ -85,7 +92,21 @@ This field supports [interpolation functions](/docs/configuration/interpolation# Type: `string` -### `init` +### `init_check` + +Bloblang expression that must evaluate to `true` for a message to initialize the cache. + + +Type: `string` +Default: `"true"` + +```yml +# Examples + +init_check: this.process == "start" +``` + +### `init_map` Bloblang expression to initialize the value when the cache key doesn't exist. Defaults to an empty array. @@ -96,11 +117,11 @@ Default: `"root = []"` ```yml # Examples -init: root = [] +init_map: root = [] -init: 'root = {"items": []}' +init_map: 'root = {"items": []}' -init: 'root = {"count": 0, "total": 0}' +init_map: 'root = {"count": 0, "total": 0}' ``` ### `append_check` @@ -174,6 +195,14 @@ flush_map: 'root = {"result": this}' flush_map: root.items = this ``` +### `filter_untreated` + +When `true`, messages that have not been collected are automatically filtered. Defaults to `false`. + + +Type: `bool` +Default: `false` + ### `ttl` The TTL of each individual item as a duration string. After this period an item will be eligible for removal during the next compaction. Not all caches support per-key TTLs, those that do will have a configuration field `default_ttl`, and those that do not will fall back to their generally configured TTL setting. From 2e6769e33600af8f93b8e57d5897eb3aa586954a Mon Sep 17 00:00:00 2001 From: lublak <44057030+lublak@users.noreply.github.com> Date: Thu, 12 Mar 2026 13:15:47 +0100 Subject: [PATCH 14/24] remove prints --- internal/impl/pure/processor_cache_collector.go | 4 ---- 1 file changed, 4 deletions(-) diff --git a/internal/impl/pure/processor_cache_collector.go b/internal/impl/pure/processor_cache_collector.go index 0e77f6e0be..5718091c08 100644 --- a/internal/impl/pure/processor_cache_collector.go +++ b/internal/impl/pure/processor_cache_collector.go @@ -370,12 +370,9 @@ func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch servi } } - fmt.Println("write flush") - newMsgs = append(newMsgs, msg) } } else if !cc.filterUntreated { - fmt.Println("write raw") newMsgs = append(newMsgs, msg) } } else if processInit { @@ -405,7 +402,6 @@ func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch servi return nil, err } } else if !cc.filterUntreated { - fmt.Println("write raw") newMsgs = append(newMsgs, msg) } } From 82c628f926e1b5ae0c1e22b62f114659c1f6b88d Mon Sep 17 00:00:00 2001 From: lublak <44057030+lublak@users.noreply.github.com> Date: Thu, 12 Mar 2026 14:38:28 +0100 Subject: [PATCH 15/24] use the same cached current design for flush --- .../impl/pure/processor_cache_collector.go | 18 ++++++++++++++---- .../pure/processor_cache_collector_test.go | 8 ++++---- .../components/processors/cache_collector.md | 14 +++++++------- 3 files changed, 25 insertions(+), 15 deletions(-) diff --git a/internal/impl/pure/processor_cache_collector.go b/internal/impl/pure/processor_cache_collector.go index 5718091c08..3eedf10fcc 100644 --- a/internal/impl/pure/processor_cache_collector.go +++ b/internal/impl/pure/processor_cache_collector.go @@ -64,9 +64,9 @@ The `+"`append_map`"+` bloblang expression can access both the current cached va Description("When `true`, the cache entry is deleted after flushing. Defaults to `false`."). Default(false), service.NewBloblangField(cacheCollectorPFieldFlushMap). - Description("Bloblang expression to transform the accumulated value before emitting. Defaults to `root`."). - Default("root = this"). - Examples(`root = this`, `root = {"result": this}`, `root.items = this`), + Description("Bloblang expression to transform the accumulated value before emitting. It receives the current value as `this.cached` and the new message as `this.current`. Defaults to `this.cached`."). + Default("root = this.cached"). + Examples(`root = this.cached`, `root = {"result": this.current.append(this.cached)}`, `root.items = this.cached`), service.NewBoolField(cacheCollectorPFieldFilterUntreated). Description("When `true`, messages that have not been collected are automatically filtered. Defaults to `false`."). Default(false), @@ -348,9 +348,19 @@ func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch servi } if processFlush { - msg.SetBytes(cachedValue) + flushMsgJson, err := json.Marshal(cacheCollectorMessageData{ + Cached: json.RawMessage(cachedValue), + Current: json.RawMessage(currentValue), + }) + + if err != nil { + return nil, err + } + + msg.SetBytes(flushMsgJson) msg, err = flushMap.Query(i) + if err != nil { return nil, err } diff --git a/internal/impl/pure/processor_cache_collector_test.go b/internal/impl/pure/processor_cache_collector_test.go index 1d12df8ace..e7bd6f4f04 100644 --- a/internal/impl/pure/processor_cache_collector_test.go +++ b/internal/impl/pure/processor_cache_collector_test.go @@ -32,8 +32,8 @@ append_map: | root = this.cached.append(this.current) flush_check: "batch_index() == 2" flush_map: | - root.result = this - root.count = this.length() + root.result = this.cached + root.count = this.cached.length() flush_deletes: true ` processor, mgr, err := cacheCollectorProc(spec) @@ -75,8 +75,8 @@ append_map: | root = this.cached.append(this.current) flush_check: "batch_index() == 2" flush_map: | - root.result = this - root.count = this.length() + root.result = this.cached + root.count = this.cached.length() flush_deletes: false ` processor, mgr, err := cacheCollectorProc(spec) diff --git a/website/docs/components/processors/cache_collector.md b/website/docs/components/processors/cache_collector.md index df06b30ee7..4e1151d661 100644 --- a/website/docs/components/processors/cache_collector.md +++ b/website/docs/components/processors/cache_collector.md @@ -40,7 +40,7 @@ cache_collector: append_map: root = this.cached.append(this.current) # No default (required) flush_check: this.process == "end" # No default (required) flush_deletes: false - flush_map: root = this + flush_map: root = this.cached filter_untreated: false ``` @@ -59,7 +59,7 @@ cache_collector: append_map: root = this.cached.append(this.current) # No default (required) flush_check: this.process == "end" # No default (required) flush_deletes: false - flush_map: root = this + flush_map: root = this.cached filter_untreated: false ttl: 60s # No default (optional) ``` @@ -179,20 +179,20 @@ Default: `false` ### `flush_map` -Bloblang expression to transform the accumulated value before emitting. Defaults to `root`. +Bloblang expression to transform the accumulated value before emitting. It receives the current value as `this.cached` and the new message as `this.current`. Defaults to `this.cached`. Type: `string` -Default: `"root = this"` +Default: `"root = this.cached"` ```yml # Examples -flush_map: root = this +flush_map: root = this.cached -flush_map: 'root = {"result": this}' +flush_map: 'root = {"result": this.current.append(this.cached)}' -flush_map: root.items = this +flush_map: root.items = this.cached ``` ### `filter_untreated` From 6dbf7c5812b9bb2b701ff43293c04dfa03662421 Mon Sep 17 00:00:00 2001 From: lublak <44057030+lublak@users.noreply.github.com> Date: Thu, 12 Mar 2026 15:20:21 +0100 Subject: [PATCH 16/24] fix deleted flush messages --- .../impl/pure/processor_cache_collector.go | 4 ++- .../pure/processor_cache_collector_test.go | 32 +++++++++++++++++++ 2 files changed, 35 insertions(+), 1 deletion(-) diff --git a/internal/impl/pure/processor_cache_collector.go b/internal/impl/pure/processor_cache_collector.go index 3eedf10fcc..01a9b7997e 100644 --- a/internal/impl/pure/processor_cache_collector.go +++ b/internal/impl/pure/processor_cache_collector.go @@ -380,7 +380,9 @@ func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch servi } } - newMsgs = append(newMsgs, msg) + if msg != nil { + newMsgs = append(newMsgs, msg) + } } } else if !cc.filterUntreated { newMsgs = append(newMsgs, msg) diff --git a/internal/impl/pure/processor_cache_collector_test.go b/internal/impl/pure/processor_cache_collector_test.go index e7bd6f4f04..5b20ead6d4 100644 --- a/internal/impl/pure/processor_cache_collector_test.go +++ b/internal/impl/pure/processor_cache_collector_test.go @@ -292,3 +292,35 @@ filter_untreated: true require.NoError(t, err) assert.JSONEq(t, `[{"type":"work","id":"3"},{"type":"work","id":"4"}]`, string(result)) } + +func TestCacheCollectorProcessor_FlushDeleted(t *testing.T) { + spec := ` +resource: test_cache +key: test_key +init_check: this.type == "start" +init_map: root = [] +append_check: this.type == "work" +append_map: | + root = this.cached.append(this.current) +flush_check: this.type == "end" +flush_map: | + root = deleted() +flush_deletes: true +filter_untreated: true +` + processor, _, err := cacheCollectorProc(spec) + require.NoError(t, err) + + t.Cleanup(func() { require.NoError(t, processor.Close(t.Context())) }) + + batch := service.MessageBatch{ + service.NewMessage([]byte(`{"type":"start","id":"1"}`)), + service.NewMessage([]byte(`{"type":"work","id":"2"}`)), + service.NewMessage([]byte(`{"type":"end","id":"3"}`)), + } + + results, err := processor.ProcessBatch(t.Context(), batch) + + require.NoError(t, err) + require.Len(t, results, 0) +} From a4828afcad5e85d911235e077f27b79785fe6b02 Mon Sep 17 00:00:00 2001 From: lublak <44057030+lublak@users.noreply.github.com> Date: Thu, 12 Mar 2026 15:42:59 +0100 Subject: [PATCH 17/24] linting fix --- internal/impl/pure/processor_cache_collector_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/impl/pure/processor_cache_collector_test.go b/internal/impl/pure/processor_cache_collector_test.go index 5b20ead6d4..3dddbb3243 100644 --- a/internal/impl/pure/processor_cache_collector_test.go +++ b/internal/impl/pure/processor_cache_collector_test.go @@ -322,5 +322,5 @@ filter_untreated: true results, err := processor.ProcessBatch(t.Context(), batch) require.NoError(t, err) - require.Len(t, results, 0) + require.Empty(t, results) } From b870767b7e16b0250ec7054178d8f5c7c9310bf7 Mon Sep 17 00:00:00 2001 From: lublak Date: Mon, 16 Mar 2026 22:58:41 +0100 Subject: [PATCH 18/24] implement init_mode, flush_keep_cache and passthrough_mode --- .../impl/pure/processor_cache_collector.go | 182 +++++++++++++----- .../pure/processor_cache_collector_test.go | 2 +- .../components/processors/cache_collector.md | 39 ++-- 3 files changed, 164 insertions(+), 59 deletions(-) diff --git a/internal/impl/pure/processor_cache_collector.go b/internal/impl/pure/processor_cache_collector.go index 01a9b7997e..f5dcd13981 100644 --- a/internal/impl/pure/processor_cache_collector.go +++ b/internal/impl/pure/processor_cache_collector.go @@ -16,12 +16,13 @@ const ( cacheCollectorPFieldKey = "key" cacheCollectorPFieldInitCheck = "init_check" cacheCollectorPFieldInitMap = "init_map" + cacheCollectorPFieldInitMode = "init_mode" cacheCollectorPFieldAppendCheck = "append_check" cacheCollectorPFieldAppendMap = "append_map" cacheCollectorPFieldFlushCheck = "flush_check" - cacheCollectorPFieldFlushDeletes = "flush_deletes" + cacheCollectorPFieldFlushKeepCache = "flush_keep_cache" cacheCollectorPFieldFlushMap = "flush_map" - cacheCollectorPFieldFilterUntreated = "filter_untreated" + cacheCollectorPFieldPassthroughMode = "passthrough_mode" cacheCollectorPFieldTTL = "ttl" ) @@ -51,6 +52,10 @@ The `+"`append_map`"+` bloblang expression can access both the current cached va Description("Bloblang expression to initialize the value when the cache key doesn't exist. Defaults to an empty array."). Default("root = []"). Examples("root = []", `root = {"items": []}`, `root = {"count": 0, "total": 0}`), + service.NewInterpolatedStringEnumField(cacheCollectorPFieldInitMode, "check", "ignore", "replace"). + Description("Option to change the behavior of the initialisation. `check` will check if the cache key already exists and returns to an error, `ignore` just ignores the current message and keeps the cached value and `replace` replaces the cached value."). + Advanced(). + Default("check"), service.NewBloblangField(cacheCollectorPFieldAppendCheck). Description("Bloblang expression that must evaluate to `true` for a message to be appended to the accumulated value."). Examples(`this.process == "work"`, `batch_index() < 100`), @@ -60,16 +65,18 @@ The `+"`append_map`"+` bloblang expression can access both the current cached va service.NewBloblangField(cacheCollectorPFieldFlushCheck). Description("Bloblang expression that must evaluate to `true` to emit the accumulated value and potentially clear the cache."). Examples(`this.process == "end"`, `batch_index() == 0 && message_index() > 0 && batch_index() % 100 == 0`), - service.NewBoolField(cacheCollectorPFieldFlushDeletes). - Description("When `true`, the cache entry is deleted after flushing. Defaults to `false`."). - Default(false), service.NewBloblangField(cacheCollectorPFieldFlushMap). Description("Bloblang expression to transform the accumulated value before emitting. It receives the current value as `this.cached` and the new message as `this.current`. Defaults to `this.cached`."). Default("root = this.cached"). Examples(`root = this.cached`, `root = {"result": this.current.append(this.cached)}`, `root.items = this.cached`), - service.NewBoolField(cacheCollectorPFieldFilterUntreated). - Description("When `true`, messages that have not been collected are automatically filtered. Defaults to `false`."). + service.NewBoolField(cacheCollectorPFieldFlushKeepCache). + Description("When `true`, the cache entry is not deleted after flushing. Defaults to `false`."). + Advanced(). Default(false), + service.NewInterpolatedStringEnumField(cacheCollectorPFieldPassthroughMode, "none", "unprocessed", "processed", "all"). + Description("Option to change the behavior of the messages after handled (flush always creates a new message). `none` will just filter messages alle messages out, `unprocessed` will passtrought messages not handled by init or append, `processed` will passtrought only messages handled by init or append and `all` will passtrought all messages."). + Advanced(). + Default("none"), service.NewInterpolatedStringField(cacheCollectorPFieldTTL). Description("The TTL of each individual item as a duration string. After this period an item will be eligible for removal during the next compaction. Not all caches support per-key TTLs, those that do will have a configuration field `default_ttl`, and those that do not will fall back to their generally configured TTL setting."). Examples("60s", "5m", "36h"). @@ -81,16 +88,16 @@ The `+"`append_map`"+` bloblang expression can access both the current cached va type cacheCollectorProcessor struct { cacheName string - key *service.InterpolatedString - initCheck *bloblang.Executor - initMap *bloblang.Executor - appendCheck *bloblang.Executor - appendMap *bloblang.Executor - flushCheck *bloblang.Executor - flushDeletes bool - flushMap *bloblang.Executor - - filterUntreated bool + key *service.InterpolatedString + initCheck *bloblang.Executor + initMap *bloblang.Executor + initMode *service.InterpolatedString + appendCheck *bloblang.Executor + appendMap *bloblang.Executor + flushCheck *bloblang.Executor + flushKeepCache bool + flushMap *bloblang.Executor + passthroughMode *service.InterpolatedString ttl *service.InterpolatedString @@ -125,6 +132,11 @@ func NewCacheCollectorFromConfig(conf *service.ParsedConfig, mgr *service.Resour return nil, err } + initMode, err := conf.FieldInterpolatedString(cacheCollectorPFieldInitMode) + if err != nil { + return nil, err + } + appendCheck, err := conf.FieldBloblang(cacheCollectorPFieldAppendCheck) if err != nil { return nil, err @@ -140,7 +152,7 @@ func NewCacheCollectorFromConfig(conf *service.ParsedConfig, mgr *service.Resour return nil, err } - flushDeletes, err := conf.FieldBool(cacheCollectorPFieldFlushDeletes) + flushKeepCache, err := conf.FieldBool(cacheCollectorPFieldFlushKeepCache) if err != nil { return nil, err } @@ -150,7 +162,7 @@ func NewCacheCollectorFromConfig(conf *service.ParsedConfig, mgr *service.Resour return nil, err } - filterUntreated, err := conf.FieldBool(cacheCollectorPFieldFilterUntreated) + passthroughMode, err := conf.FieldInterpolatedString(cacheCollectorPFieldPassthroughMode) if err != nil { return nil, err } @@ -161,15 +173,16 @@ func NewCacheCollectorFromConfig(conf *service.ParsedConfig, mgr *service.Resour key: key, initCheck: initCheck, initMap: initMap, + initMode: initMode, appendCheck: appendCheck, appendMap: appendMap, flushCheck: flushCheck, flushMap: flushMap, - filterUntreated: filterUntreated, + passthroughMode: passthroughMode, - cacheName: resource, - flushDeletes: flushDeletes, + cacheName: resource, + flushKeepCache: flushKeepCache, ttl: ttl, @@ -195,6 +208,21 @@ func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch servi ttlInterp = batch.InterpolationExecutor(cc.ttl) } + var initCheck *service.MessageBatchBloblangExecutor + if cc.initCheck != nil { + initCheck = batch.BloblangExecutor(cc.initCheck) + } + + var initMap *service.MessageBatchBloblangExecutor + if cc.initMap != nil { + initMap = batch.BloblangExecutor(cc.initMap) + } + + var initModeInterp *service.MessageBatchInterpolationExecutor + if cc.key != nil { + initModeInterp = batch.InterpolationExecutor(cc.initMode) + } + var appendCheck *service.MessageBatchBloblangExecutor if cc.appendCheck != nil { appendCheck = batch.BloblangExecutor(cc.appendCheck) @@ -215,14 +243,9 @@ func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch servi flushMap = batch.BloblangExecutor(cc.flushMap) } - var initCheck *service.MessageBatchBloblangExecutor - if cc.initCheck != nil { - initCheck = batch.BloblangExecutor(cc.initCheck) - } - - var initMap *service.MessageBatchBloblangExecutor - if cc.initMap != nil { - initMap = batch.BloblangExecutor(cc.initMap) + var passthroughModeInterp *service.MessageBatchInterpolationExecutor + if cc.key != nil { + passthroughModeInterp = batch.InterpolationExecutor(cc.passthroughMode) } for i, msg := range batch { @@ -231,6 +254,11 @@ func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch servi return nil, fmt.Errorf("key evaluation error: %w", err) } + passthroughMode, err := passthroughModeInterp.TryString(i) + if err != nil { + return nil, fmt.Errorf("passthrough evaluation error: %w", err) + } + var ttls string if ttlInterp != nil { ttls, err = ttlInterp.TryString(i) @@ -287,8 +315,32 @@ func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch servi return nil, err } - if !cacheValueExists { - if processInit { + if processInit { + if cacheValueExists { + initMode, err := initModeInterp.TryString(i) + if err != nil { + return nil, fmt.Errorf("init_mode evaluation error: %w", err) + } + + switch initMode { + case "check": + return nil, fmt.Errorf("failed to set cache key '%s': %v", key, service.ErrKeyAlreadyExists) + case "ignore": + // ignore the init message + case "replace": + initMsg, err := initMap.Query(i) + if err != nil { + return nil, fmt.Errorf("init_map evaluation error: %w", err) + } + initData, err := initMsg.AsBytes() + if err != nil { + return nil, fmt.Errorf("init data error: %w", err) + } + cachedValue = initData + default: + return nil, fmt.Errorf("init_mode unsupported mode %s", initMode) + } + } else { initMsg, err := initMap.Query(i) if err != nil { return nil, fmt.Errorf("init_map evaluation error: %w", err) @@ -365,7 +417,7 @@ func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch servi return nil, err } - if cc.flushDeletes { + if !cc.flushKeepCache { if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { err = cache.Delete(ctx, key) if err != nil { @@ -383,11 +435,18 @@ func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch servi if msg != nil { newMsgs = append(newMsgs, msg) } + } else if passthroughMode == "processed" || passthroughMode == "all" { + newMsgs = append(newMsgs, msg) } - } else if !cc.filterUntreated { + } else if passthroughMode == "unprocessed" || passthroughMode == "all" { newMsgs = append(newMsgs, msg) } } else if processInit { + initMode, err := initModeInterp.TryString(i) + if err != nil { + return nil, fmt.Errorf("init_mode evaluation error: %w", err) + } + initMsg, err := initMap.Query(i) if err != nil { return nil, fmt.Errorf("init_map evaluation error: %w", err) @@ -397,23 +456,58 @@ func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch servi return nil, fmt.Errorf("init data error: %w", err) } - if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { - err = cache.Add(ctx, key, initData, ttl) + switch initMode { + case "check": + if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { + err = cache.Add(ctx, key, initData, ttl) + if err != nil { + err = fmt.Errorf("failed to set cache key '%s': %v", key, err) + } + }); cerr != nil { + err = cerr + } + if err != nil { - if errors.Is(err, service.ErrKeyAlreadyExists) { - err = nil - } else { + return nil, err + } + case "ignore": + if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { + err = cache.Add(ctx, key, initData, ttl) + if err != nil { + if errors.Is(err, service.ErrKeyAlreadyExists) { + err = nil + } else { + err = fmt.Errorf("failed to set cache key '%s': %v", key, err) + } + } + }); cerr != nil { + err = cerr + } + + if err != nil { + return nil, err + } + case "replace": + if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { + err = cache.Set(ctx, key, initData, ttl) + if err != nil { err = fmt.Errorf("failed to set cache key '%s': %v", key, err) } + }); cerr != nil { + err = cerr } - }); cerr != nil { - err = cerr + + if err != nil { + return nil, err + } + default: + return nil, fmt.Errorf("init_mode unsupported mode %s", initMode) } - if err != nil { - return nil, err + if passthroughMode == "processed" || passthroughMode == "all" { + newMsgs = append(newMsgs, msg) } - } else if !cc.filterUntreated { + } else if passthroughMode == "unprocessed" || passthroughMode == "all" { newMsgs = append(newMsgs, msg) } } diff --git a/internal/impl/pure/processor_cache_collector_test.go b/internal/impl/pure/processor_cache_collector_test.go index 3dddbb3243..587bdbb5f7 100644 --- a/internal/impl/pure/processor_cache_collector_test.go +++ b/internal/impl/pure/processor_cache_collector_test.go @@ -22,7 +22,7 @@ func cacheCollectorProc(confStr string) (service.BatchProcessor, *service.Resour return proc, mgr, err } -func TestCacheCollectorProcessor_BasicAccumulationDeleteCache(t *testing.T) { +func TestCacheCollectorProcessor_BasicAccumulation(t *testing.T) { spec := ` resource: test_cache key: test_key diff --git a/website/docs/components/processors/cache_collector.md b/website/docs/components/processors/cache_collector.md index 4e1151d661..20636199cb 100644 --- a/website/docs/components/processors/cache_collector.md +++ b/website/docs/components/processors/cache_collector.md @@ -39,9 +39,7 @@ cache_collector: append_check: this.process == "work" # No default (required) append_map: root = this.cached.append(this.current) # No default (required) flush_check: this.process == "end" # No default (required) - flush_deletes: false flush_map: root = this.cached - filter_untreated: false ``` @@ -55,12 +53,13 @@ cache_collector: key: "" # No default (required) init_check: "true" init_map: root = [] + init_mode: check append_check: this.process == "work" # No default (required) append_map: root = this.cached.append(this.current) # No default (required) flush_check: this.process == "end" # No default (required) - flush_deletes: false flush_map: root = this.cached - filter_untreated: false + flush_keep_cache: false + passthrough_mode: none ttl: 60s # No default (optional) ``` @@ -124,6 +123,16 @@ init_map: 'root = {"items": []}' init_map: 'root = {"count": 0, "total": 0}' ``` +### `init_mode` + +Option to change the behavior of the initialisation. `check` will check if the cache key already exists and returns to an error, `ignore` just ignores the current message and keeps the cached value and `replace` replaces the cached value. +This field supports [interpolation functions](/docs/configuration/interpolation#bloblang-queries). + + +Type: `string` +Default: `"check"` +Options: `check`, `ignore`, `replace`. + ### `append_check` Bloblang expression that must evaluate to `true` for a message to be appended to the accumulated value. @@ -169,14 +178,6 @@ flush_check: this.process == "end" flush_check: batch_index() == 0 && message_index() > 0 && batch_index() % 100 == 0 ``` -### `flush_deletes` - -When `true`, the cache entry is deleted after flushing. Defaults to `false`. - - -Type: `bool` -Default: `false` - ### `flush_map` Bloblang expression to transform the accumulated value before emitting. It receives the current value as `this.cached` and the new message as `this.current`. Defaults to `this.cached`. @@ -195,14 +196,24 @@ flush_map: 'root = {"result": this.current.append(this.cached)}' flush_map: root.items = this.cached ``` -### `filter_untreated` +### `flush_keep_cache` -When `true`, messages that have not been collected are automatically filtered. Defaults to `false`. +When `true`, the cache entry is not deleted after flushing. Defaults to `false`. Type: `bool` Default: `false` +### `passthrough_mode` + +Option to change the behavior of the messages after handled (flush always creates a new message). `none` will just filter messages alle messages out, `unprocessed` will passtrought messages not handled by init or append, `processed` will passtrought only messages handled by init or append and `all` will passtrought all messages. +This field supports [interpolation functions](/docs/configuration/interpolation#bloblang-queries). + + +Type: `string` +Default: `"none"` +Options: `none`, `unprocessed`, `processed`, `all`. + ### `ttl` The TTL of each individual item as a duration string. After this period an item will be eligible for removal during the next compaction. Not all caches support per-key TTLs, those that do will have a configuration field `default_ttl`, and those that do not will fall back to their generally configured TTL setting. From 40e0aa775b0ed1c30156186643d9e002e630231c Mon Sep 17 00:00:00 2001 From: lublak Date: Mon, 16 Mar 2026 23:02:03 +0100 Subject: [PATCH 19/24] add some missing docs --- internal/impl/pure/processor_cache_collector.go | 2 +- website/docs/components/processors/cache_collector.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/internal/impl/pure/processor_cache_collector.go b/internal/impl/pure/processor_cache_collector.go index f5dcd13981..513ff4f760 100644 --- a/internal/impl/pure/processor_cache_collector.go +++ b/internal/impl/pure/processor_cache_collector.go @@ -38,7 +38,7 @@ This processor works by storing an accumulated value in a cache, which is update 2. `+"`append_check`"+`: For each message, if this expression evaluates to true and the cache was initialized, the value is updated using `+"`append_map`"+`. 3. `+"`flush_check`"+`: When this expression evaluates to true and the cache was initialized, the accumulated value is emitted as a new message and the cache is optionally cleared. -The `+"`append_map`"+` bloblang expression can access both the current cached value as `+"`this.cached`"+` and the current message as `+"`this.current`"+`.`). +The `+"`append_map`"+` and `+"`flush_map`"+` bloblang expressions can access both the current cached value as `+"`this.cached`"+` and the current message as `+"`this.current`"+`.`). Fields( service.NewStringField(cacheCollectorPFieldResource). Description("The [`cache` resource](/docs/components/caches/about) to use for storing accumulated state."), diff --git a/website/docs/components/processors/cache_collector.md b/website/docs/components/processors/cache_collector.md index 20636199cb..99f0c30e0a 100644 --- a/website/docs/components/processors/cache_collector.md +++ b/website/docs/components/processors/cache_collector.md @@ -72,7 +72,7 @@ This processor works by storing an accumulated value in a cache, which is update 2. `append_check`: For each message, if this expression evaluates to true and the cache was initialized, the value is updated using `append_map`. 3. `flush_check`: When this expression evaluates to true and the cache was initialized, the accumulated value is emitted as a new message and the cache is optionally cleared. -The `append_map` bloblang expression can access both the current cached value as `this.cached` and the current message as `this.current`. +The `append_map` and `flush_map` bloblang expressions can access both the current cached value as `this.cached` and the current message as `this.current`. ## Fields From 31026c38ba720dfac06c6e5461c2640262861bc5 Mon Sep 17 00:00:00 2001 From: lublak Date: Wed, 22 Apr 2026 22:43:54 +0200 Subject: [PATCH 20/24] adjust tests --- .../pure/processor_cache_collector_test.go | 18 +++++++++++------- 1 file changed, 11 insertions(+), 7 deletions(-) diff --git a/internal/impl/pure/processor_cache_collector_test.go b/internal/impl/pure/processor_cache_collector_test.go index 587bdbb5f7..c2096ad10a 100644 --- a/internal/impl/pure/processor_cache_collector_test.go +++ b/internal/impl/pure/processor_cache_collector_test.go @@ -27,6 +27,7 @@ func TestCacheCollectorProcessor_BasicAccumulation(t *testing.T) { resource: test_cache key: test_key init_map: root = [{"id":"0"}] +init_mode: ignore append_check: true append_map: | root = this.cached.append(this.current) @@ -34,7 +35,7 @@ flush_check: "batch_index() == 2" flush_map: | root.result = this.cached root.count = this.cached.length() -flush_deletes: true +flush_keep_cache: false ` processor, mgr, err := cacheCollectorProc(spec) require.NoError(t, err) @@ -70,6 +71,7 @@ func TestCacheCollectorProcessor_BasicAccumulatioKeepCache(t *testing.T) { resource: test_cache key: test_key init_map: root = [{"id":"0"}] +init_mode: ignore append_check: true append_map: | root = this.cached.append(this.current) @@ -77,7 +79,7 @@ flush_check: "batch_index() == 2" flush_map: | root.result = this.cached root.count = this.cached.length() -flush_deletes: false +flush_keep_cache: true ` processor, mgr, err := cacheCollectorProc(spec) require.NoError(t, err) @@ -117,10 +119,11 @@ func TestCacheCollectorProcessor_WithDifferentKeys(t *testing.T) { resource: test_cache key: test_${! this.group } init_map: root = [] +init_mode: ignore append_check: root = this.end == false append_map: root = this.cached.append(this.current.data) flush_check: this.end -flush_deletes: true +flush_keep_cache: false ` processor, _, err := cacheCollectorProc(spec) require.NoError(t, err) @@ -164,7 +167,7 @@ append_check: true append_map: | root = this.cached.append(metadata("meta_data")) flush_check: true -flush_deletes: true +flush_keep_cache: false ` processor, _, err := cacheCollectorProc(spec) require.NoError(t, err) @@ -200,7 +203,8 @@ append_check: this.type == "work" append_map: | root = this.cached.append(this.current) flush_check: this.type == "end" -flush_deletes: true +flush_keep_cache: false +passthrough_mode: unprocessed ` processor, _, err := cacheCollectorProc(spec) require.NoError(t, err) @@ -259,7 +263,7 @@ append_check: this.type == "work" append_map: | root = this.cached.append(this.current) flush_check: this.type == "end" -flush_deletes: true +flush_keep_cache: false filter_untreated: true ` processor, _, err := cacheCollectorProc(spec) @@ -305,7 +309,7 @@ append_map: | flush_check: this.type == "end" flush_map: | root = deleted() -flush_deletes: true +flush_keep_cache: false filter_untreated: true ` processor, _, err := cacheCollectorProc(spec) From 461d2c44086f475ed1a3f40a5887731db894e969 Mon Sep 17 00:00:00 2001 From: lublak Date: Wed, 22 Apr 2026 22:44:49 +0200 Subject: [PATCH 21/24] remove unsued keys --- internal/impl/pure/processor_cache_collector_test.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/internal/impl/pure/processor_cache_collector_test.go b/internal/impl/pure/processor_cache_collector_test.go index c2096ad10a..3d63c10e73 100644 --- a/internal/impl/pure/processor_cache_collector_test.go +++ b/internal/impl/pure/processor_cache_collector_test.go @@ -264,7 +264,6 @@ append_map: | root = this.cached.append(this.current) flush_check: this.type == "end" flush_keep_cache: false -filter_untreated: true ` processor, _, err := cacheCollectorProc(spec) require.NoError(t, err) @@ -310,7 +309,6 @@ flush_check: this.type == "end" flush_map: | root = deleted() flush_keep_cache: false -filter_untreated: true ` processor, _, err := cacheCollectorProc(spec) require.NoError(t, err) From 349568b38c5a92085210cab439dca5aa26f3ea70 Mon Sep 17 00:00:00 2001 From: lublak <44057030+lublak@users.noreply.github.com> Date: Mon, 29 Jun 2026 14:35:10 +0200 Subject: [PATCH 22/24] add new init mode flush, add more tests arround init mode --- .../impl/pure/processor_cache_collector.go | 100 +++++++- .../pure/processor_cache_collector_test.go | 230 ++++++++++++++++++ 2 files changed, 323 insertions(+), 7 deletions(-) diff --git a/internal/impl/pure/processor_cache_collector.go b/internal/impl/pure/processor_cache_collector.go index 513ff4f760..7d7c5c99d6 100644 --- a/internal/impl/pure/processor_cache_collector.go +++ b/internal/impl/pure/processor_cache_collector.go @@ -52,8 +52,8 @@ The `+"`append_map`"+` and `+"`flush_map`"+` bloblang expressions can access bot Description("Bloblang expression to initialize the value when the cache key doesn't exist. Defaults to an empty array."). Default("root = []"). Examples("root = []", `root = {"items": []}`, `root = {"count": 0, "total": 0}`), - service.NewInterpolatedStringEnumField(cacheCollectorPFieldInitMode, "check", "ignore", "replace"). - Description("Option to change the behavior of the initialisation. `check` will check if the cache key already exists and returns to an error, `ignore` just ignores the current message and keeps the cached value and `replace` replaces the cached value."). + service.NewInterpolatedStringEnumField(cacheCollectorPFieldInitMode, "check", "ignore", "replace", "flush"). + Description("Option to change the behavior of the initialisation. `check` will check if the cache key already exists and returns to an error, `ignore` just ignores the current message and keeps the cached value, `replace` replaces the cached value and `flush` flushes the cached value which gets replaced."). Advanced(). Default("check"), service.NewBloblangField(cacheCollectorPFieldAppendCheck). @@ -293,6 +293,11 @@ func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch servi } if processAppend || processFlush { + currentValue, err := msg.AsBytes() + if err != nil { + return nil, err + } + var cachedValue []byte cacheValueExists := true @@ -328,6 +333,38 @@ func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch servi case "ignore": // ignore the init message case "replace": + initMsg, err := initMap.Query(i) + if err != nil { + return nil, fmt.Errorf("init_map evaluation error: %w", err) + } + initData, err := initMsg.AsBytes() + if err != nil { + return nil, fmt.Errorf("init data error: %w", err) + } + cachedValue = initData + case "flush": + flushMsgJson, err := json.Marshal(cacheCollectorMessageData{ + Cached: json.RawMessage(cachedValue), + Current: json.RawMessage(currentValue), + }) + + if err != nil { + return nil, err + } + + msg.SetBytes(flushMsgJson) + + var fMsg *service.Message + fMsg, err = flushMap.Query(i) + + msg.SetBytes(currentValue) + + if err != nil { + return nil, err + } + + newMsgs = append(newMsgs, fMsg) + initMsg, err := initMap.Query(i) if err != nil { return nil, fmt.Errorf("init_map evaluation error: %w", err) @@ -355,11 +392,6 @@ func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch servi } if cacheValueExists { - currentValue, err := msg.AsBytes() - if err != nil { - return nil, err - } - if processAppend { appendMsgJson, err := json.Marshal(cacheCollectorMessageData{ Cached: json.RawMessage(cachedValue), @@ -497,6 +529,60 @@ func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch servi err = cerr } + if err != nil { + return nil, err + } + case "flush": + if cerr := cc.mgr.AccessCache(ctx, cc.cacheName, func(cache service.Cache) { + var cachedValue []byte + cachedValue, err = cache.Get(ctx, key) + if err != nil { + if errors.Is(err, service.ErrKeyNotFound) { + err = cache.Set(ctx, key, initData, ttl) + if err != nil { + err = fmt.Errorf("failed to set cache key '%s': %v", key, err) + } + } + return + } + + var currentValue []byte + currentValue, err = msg.AsBytes() + if err != nil { + return + } + + var flushMsgJson []byte + flushMsgJson, err = json.Marshal(cacheCollectorMessageData{ + Cached: json.RawMessage(cachedValue), + Current: json.RawMessage(currentValue), + }) + + if err != nil { + return + } + + msg.SetBytes(flushMsgJson) + + var fMsg *service.Message + fMsg, err = flushMap.Query(i) + + msg.SetBytes(currentValue) + + if err != nil { + return + } + + newMsgs = append(newMsgs, fMsg) + + err = cache.Set(ctx, key, initData, ttl) + if err != nil { + err = fmt.Errorf("failed to set cache key '%s': %v", key, err) + } + }); cerr != nil { + err = cerr + } + if err != nil { return nil, err } diff --git a/internal/impl/pure/processor_cache_collector_test.go b/internal/impl/pure/processor_cache_collector_test.go index 3d63c10e73..a6cf77862a 100644 --- a/internal/impl/pure/processor_cache_collector_test.go +++ b/internal/impl/pure/processor_cache_collector_test.go @@ -253,6 +253,236 @@ passthrough_mode: unprocessed assert.JSONEq(t, `{"type":"end","id":"8"}`, string(result)) } +func TestCacheCollectorProcessor_WithInitCheckCollisionsCheck(t *testing.T) { + spec := ` +resource: test_cache +key: test_key +init_check: this.type == "start" +init_map: root = [] +init_mode: check +append_check: this.type == "work" +append_map: | + root = this.cached.append(this.current) +flush_check: this.type == "end" +flush_keep_cache: false +passthrough_mode: unprocessed +` + processor, _, err := cacheCollectorProc(spec) + require.NoError(t, err) + + t.Cleanup(func() { require.NoError(t, processor.Close(t.Context())) }) + + batch := service.MessageBatch{ + service.NewMessage([]byte(`{"type":"work","id":"1"}`)), + + // start + service.NewMessage([]byte(`{"type":"start","id":"2"}`)), + service.NewMessage([]byte(`{"type":"work","id":"3"}`)), + service.NewMessage([]byte(`{"type":"start","id":"4"}`)), + service.NewMessage([]byte(`{"type":"work","id":"5"}`)), + service.NewMessage([]byte(`{"type":"ignore","id":"6"}`)), + service.NewMessage([]byte(`{"type":"end","id":"7"}`)), + // end + + service.NewMessage([]byte(`{"type":"work","id":"8"}`)), + service.NewMessage([]byte(`{"type":"end","id":"9"}`)), + } + + _, err = processor.ProcessBatch(t.Context(), batch) + + require.Error(t, err) +} + +func TestCacheCollectorProcessor_WithInitCheckCollisionsIgnore(t *testing.T) { + spec := ` +resource: test_cache +key: test_key +init_check: this.type == "start" +init_map: root = [] +init_mode: ignore +append_check: this.type == "work" +append_map: | + root = this.cached.append(this.current) +flush_check: this.type == "end" +flush_keep_cache: false +passthrough_mode: unprocessed +` + processor, _, err := cacheCollectorProc(spec) + require.NoError(t, err) + + t.Cleanup(func() { require.NoError(t, processor.Close(t.Context())) }) + + batch := service.MessageBatch{ + service.NewMessage([]byte(`{"type":"work","id":"1"}`)), + + // start + service.NewMessage([]byte(`{"type":"start","id":"2"}`)), + service.NewMessage([]byte(`{"type":"work","id":"3"}`)), + service.NewMessage([]byte(`{"type":"start","id":"4"}`)), + service.NewMessage([]byte(`{"type":"work","id":"5"}`)), + service.NewMessage([]byte(`{"type":"ignore","id":"6"}`)), + service.NewMessage([]byte(`{"type":"end","id":"7"}`)), + // end + + service.NewMessage([]byte(`{"type":"work","id":"8"}`)), + service.NewMessage([]byte(`{"type":"end","id":"9"}`)), + } + + results, err := processor.ProcessBatch(t.Context(), batch) + + require.NoError(t, err) + require.Len(t, results, 1) + require.Len(t, results[0], 5) + + result, err := results[0][0].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `{"type":"work","id":"1"}`, string(result)) + + result, err = results[0][1].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `{"type":"ignore","id":"6"}`, string(result)) + + result, err = results[0][2].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `[{"type":"work","id":"3"},{"type":"work","id":"5"}]`, string(result)) + + result, err = results[0][3].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `{"type":"work","id":"8"}`, string(result)) + + result, err = results[0][4].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `{"type":"end","id":"9"}`, string(result)) +} + +func TestCacheCollectorProcessor_WithInitCheckCollisionsReplace(t *testing.T) { + spec := ` +resource: test_cache +key: test_key +init_check: this.type == "start" +init_map: root = [] +init_mode: replace +append_check: this.type == "work" +append_map: | + root = this.cached.append(this.current) +flush_check: this.type == "end" +flush_keep_cache: false +passthrough_mode: unprocessed +` + processor, _, err := cacheCollectorProc(spec) + require.NoError(t, err) + + t.Cleanup(func() { require.NoError(t, processor.Close(t.Context())) }) + + batch := service.MessageBatch{ + service.NewMessage([]byte(`{"type":"work","id":"1"}`)), + + // start + service.NewMessage([]byte(`{"type":"start","id":"2"}`)), + service.NewMessage([]byte(`{"type":"work","id":"3"}`)), + service.NewMessage([]byte(`{"type":"start","id":"4"}`)), + service.NewMessage([]byte(`{"type":"work","id":"5"}`)), + service.NewMessage([]byte(`{"type":"ignore","id":"6"}`)), + service.NewMessage([]byte(`{"type":"end","id":"7"}`)), + // end + + service.NewMessage([]byte(`{"type":"work","id":"8"}`)), + service.NewMessage([]byte(`{"type":"end","id":"9"}`)), + } + + results, err := processor.ProcessBatch(t.Context(), batch) + + require.NoError(t, err) + require.Len(t, results, 1) + require.Len(t, results[0], 5) + + result, err := results[0][0].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `{"type":"work","id":"1"}`, string(result)) + + result, err = results[0][1].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `{"type":"ignore","id":"6"}`, string(result)) + + result, err = results[0][2].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `[{"type":"work","id":"5"}]`, string(result)) + + result, err = results[0][3].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `{"type":"work","id":"8"}`, string(result)) + + result, err = results[0][4].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `{"type":"end","id":"9"}`, string(result)) +} + +func TestCacheCollectorProcessor_WithInitCheckCollisionsFlush(t *testing.T) { + spec := ` +resource: test_cache +key: test_key +init_check: this.type == "start" +init_map: root = [] +init_mode: flush +append_check: this.type == "work" +append_map: | + root = this.cached.append(this.current) +flush_check: this.type == "end" +flush_keep_cache: false +passthrough_mode: unprocessed +` + processor, _, err := cacheCollectorProc(spec) + require.NoError(t, err) + + t.Cleanup(func() { require.NoError(t, processor.Close(t.Context())) }) + + batch := service.MessageBatch{ + service.NewMessage([]byte(`{"type":"work","id":"1"}`)), + + // start + service.NewMessage([]byte(`{"type":"start","id":"2"}`)), + service.NewMessage([]byte(`{"type":"work","id":"3"}`)), + service.NewMessage([]byte(`{"type":"start","id":"4"}`)), + service.NewMessage([]byte(`{"type":"work","id":"5"}`)), + service.NewMessage([]byte(`{"type":"ignore","id":"6"}`)), + service.NewMessage([]byte(`{"type":"end","id":"7"}`)), + // end + + service.NewMessage([]byte(`{"type":"work","id":"8"}`)), + service.NewMessage([]byte(`{"type":"end","id":"9"}`)), + } + + results, err := processor.ProcessBatch(t.Context(), batch) + + require.NoError(t, err) + require.Len(t, results, 1) + require.Len(t, results[0], 6) + + result, err := results[0][0].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `{"type":"work","id":"1"}`, string(result)) + + result, err = results[0][1].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `[{"type":"work","id":"3"}]`, string(result)) + + result, err = results[0][2].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `{"type":"ignore","id":"6"}`, string(result)) + + result, err = results[0][3].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `[{"type":"work","id":"5"}]`, string(result)) + + result, err = results[0][4].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `{"type":"work","id":"8"}`, string(result)) + + result, err = results[0][5].AsBytes() + require.NoError(t, err) + assert.JSONEq(t, `{"type":"end","id":"9"}`, string(result)) +} + func TestCacheCollectorProcessor_WithInitCheckAndFilter(t *testing.T) { spec := ` resource: test_cache From 4a3295cad3d468d0499303814a5b9b698424fd5c Mon Sep 17 00:00:00 2001 From: lublak <44057030+lublak@users.noreply.github.com> Date: Tue, 30 Jun 2026 14:33:02 +0200 Subject: [PATCH 23/24] fix nil pointer issue --- internal/impl/pure/processor_cache_collector.go | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/internal/impl/pure/processor_cache_collector.go b/internal/impl/pure/processor_cache_collector.go index 7d7c5c99d6..9e2deccc0a 100644 --- a/internal/impl/pure/processor_cache_collector.go +++ b/internal/impl/pure/processor_cache_collector.go @@ -363,7 +363,9 @@ func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch servi return nil, err } - newMsgs = append(newMsgs, fMsg) + if fMsg != nil { + newMsgs = append(newMsgs, fMsg) + } initMsg, err := initMap.Query(i) if err != nil { @@ -573,7 +575,9 @@ func (cc *cacheCollectorProcessor) ProcessBatch(ctx context.Context, batch servi return } - newMsgs = append(newMsgs, fMsg) + if fMsg != nil { + newMsgs = append(newMsgs, fMsg) + } err = cache.Set(ctx, key, initData, ttl) if err != nil { From af6b843b66204957bbd381f5f64db27b3267afa2 Mon Sep 17 00:00:00 2001 From: lublak Date: Tue, 30 Jun 2026 23:22:55 +0200 Subject: [PATCH 24/24] make docs --- website/docs/components/processors/cache_collector.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/website/docs/components/processors/cache_collector.md b/website/docs/components/processors/cache_collector.md index 99f0c30e0a..e7a8d7844d 100644 --- a/website/docs/components/processors/cache_collector.md +++ b/website/docs/components/processors/cache_collector.md @@ -125,13 +125,13 @@ init_map: 'root = {"count": 0, "total": 0}' ### `init_mode` -Option to change the behavior of the initialisation. `check` will check if the cache key already exists and returns to an error, `ignore` just ignores the current message and keeps the cached value and `replace` replaces the cached value. +Option to change the behavior of the initialisation. `check` will check if the cache key already exists and returns to an error, `ignore` just ignores the current message and keeps the cached value, `replace` replaces the cached value and `flush` flushes the cached value which gets replaced. This field supports [interpolation functions](/docs/configuration/interpolation#bloblang-queries). Type: `string` Default: `"check"` -Options: `check`, `ignore`, `replace`. +Options: `check`, `ignore`, `replace`, `flush`. ### `append_check`