Skip to content

feat(buffer_memory): spillover and active metrics #575

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Dec 1, 2025
Merged

feat(buffer_memory): spillover and active metrics #575

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
194a738
feat(buffer_memory): spillover and active metrics
alecmerdler Nov 26, 2025
e7a7eb4
update docs
jem-davies Dec 1, 2025
3e03963
Merge pull request #1 from jem-davies/add-docs-to-spillover-and-activ…
alecmerdler Dec 1, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 19 additions & 3 deletions internal/impl/pure/buffer_memory.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,13 +44,21 @@ This buffer intentionally weakens the delivery guarantees of the pipeline and th

## Batching

It is possible to batch up messages sent from this buffer using a [batch policy](/docs/configuration/batching#batch-policy).`).
It is possible to batch up messages sent from this buffer using a [batch policy](/docs/configuration/batching#batch-policy).

## Metrics

- ` + "`buffer_active`" + ` Gauge metric tracking the current number of bytes in the buffer.
- ` + "`buffer_spillover`" + ` Counter metric tracking the total number of bytes dropped because of spillover.

`).
Field(service.NewIntField("limit").
Description(`The maximum buffer size (in bytes) to allow before applying backpressure upstream.`).
Default(524288000)).
Field(service.NewBoolField("spillover").
Description("Whether to drop incoming messages that will exceed the buffer limit.").
Advanced().
Version("1.13.0").
Default(false)).
Field(service.NewInternalField(bs))
}
Expand Down Expand Up @@ -97,7 +105,7 @@ func newMemoryBufferFromConfig(conf *service.ParsedConfig, res *service.Resource
}
}

return newMemoryBuffer(limit, spilloverEnabled, batcher), nil
return newMemoryBuffer(limit, spilloverEnabled, batcher, res), nil
}

//------------------------------------------------------------------------------
Expand All @@ -118,14 +126,19 @@ type memoryBuffer struct {
closed bool

batcher *service.Batcher

activeBytes *service.MetricGauge
spilloverBytes *service.MetricCounter
}

func newMemoryBuffer(capacity int, spilloverEnabled bool, batcher *service.Batcher) *memoryBuffer {
func newMemoryBuffer(capacity int, spilloverEnabled bool, batcher *service.Batcher, res *service.Resources) *memoryBuffer {
return &memoryBuffer{
cap: capacity,
spilloverEnabled: spilloverEnabled,
cond: sync.NewCond(&sync.Mutex{}),
batcher: batcher,
activeBytes: res.Metrics().NewGauge("buffer_active"),
spilloverBytes: res.Metrics().NewCounter("buffer_spillover"),
Comment on lines +140 to +141
Copy link
Collaborator

@gregfurman gregfurman Nov 27, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you add a note in the docs that these metrics will now be tracked?

Copy link
Collaborator

@jem-davies jem-davies Dec 1, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@alecmerdler - was going to add but we don't have access so raised a PR: alecmerdler#1

Copy link
Contributor Author

@alecmerdler alecmerdler Dec 1, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks! I merged it and pulled in the changes to this branch now.

}
}

Expand Down Expand Up @@ -224,6 +237,7 @@ func (m *memoryBuffer) ReadBatch(ctx context.Context) (service.MessageBatch, ser
defer m.cond.L.Unlock()
if err == nil {
m.bytes -= outSize
m.activeBytes.Set(int64(m.bytes))
} else {
m.batches = append(batchSources, m.batches...)
}
Expand Down Expand Up @@ -262,6 +276,7 @@ func (m *memoryBuffer) WriteBatch(ctx context.Context, msgBatch service.MessageB

for (m.bytes + extraBytes) > m.cap {
if m.spilloverEnabled {
m.spilloverBytes.Incr(int64(extraBytes))
Copy link
Collaborator

@gregfurman gregfurman Nov 27, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: This Incr definitely looks like it'd benefit from generics given this unnecessary conversion 🤔

return component.ErrLimitReached
}

Expand All @@ -276,6 +291,7 @@ func (m *memoryBuffer) WriteBatch(ctx context.Context, msgBatch service.MessageB
size: extraBytes,
})
m.bytes += extraBytes
m.activeBytes.Set(int64(m.bytes))

m.cond.Broadcast()
return nil
Expand Down
8 changes: 8 additions & 0 deletions website/docs/components/buffers/memory.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,13 @@ This buffer intentionally weakens the delivery guarantees of the pipeline and th

It is possible to batch up messages sent from this buffer using a [batch policy](/docs/configuration/batching#batch-policy).

## Metrics

- `buffer_active` Gauge metric tracking the current number of bytes in the buffer.
- `buffer_spillover` Counter metric tracking the total number of bytes dropped because of spillover.



## Fields

### `limit`
Expand All @@ -90,6 +97,7 @@ Whether to drop incoming messages that will exceed the buffer limit.

Type: `bool`
Default: `false`
Requires version 1.13.0 or newer

### `batch_policy`

Expand Down