remove labelstore from prometheus interceptor (#4890)

* Add some high level pipeline tests for prometheus

* Remove labelstore interactions from the interceptor

* Move back to uber atomic and fix flaky remote write test

* One more thing to move back to uber atomic

* Actually fix the flaky test and lints

* Fix flaky test for real and add a comment to Fanout about its role in global labels

* Put componentID back on fanout for now

Author: kgeckhart

CHANGELOG.md

@@ -40,6 +40,8 @@ Main (unreleased)
   - `explain_plans`
     - always send an explain plan log message for each query, even skipped or errored queries. (@rgeyer)
 
+- Reduced resource overhead of `prometheus.scrape`, `prometheus.relabel`, `prometheus.enrich`, and `prometheus.remote_write` by removing unnecessary usage of labelstore.LabelStore. (@kgeckhart)
+
 ### Bugfixes
 
 - (_Public Preview_) Additions to `database_observability.postgres` component:

internal/component/prometheus/enrich/enrich.go

@@ -100,7 +100,6 @@ func New(opts component.Options, args Arguments) (*Component, error) {
 	c.fanout = prometheus.NewFanout(args.ForwardTo, opts.ID, opts.Registerer, ls)
 	c.receiver = prometheus.NewInterceptor(
 		c.fanout,
-		ls,
 		prometheus.WithComponentID(c.opts.ID),
 		prometheus.WithAppendHook(func(_ storage.SeriesRef, l labels.Labels, t int64, v float64, next storage.Appender) (storage.SeriesRef, error) {
 			if c.exited.Load() {

internal/component/prometheus/enrich/enrich_test.go

@@ -108,9 +108,8 @@ func TestEnricher(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			ls := labelstore.New(nil, prom.DefaultRegisterer)
 			fanout := prometheus.NewInterceptor(
-				nil, ls,
+				nil,
 				prometheus.WithAppendHook(func(ref storage.SeriesRef, l labels.Labels, _ int64, _ float64, _ storage.Appender) (storage.SeriesRef, error) {
 					for name, value := range tt.expectedLabels {
 						require.Equal(t, l.Get(name), value)

internal/component/prometheus/fanout.go

@@ -23,6 +23,7 @@ import (
 var _ storage.Appendable = (*Fanout)(nil)
 
 // Fanout supports the default Alloy style of appendables since it can go to multiple outputs. It also allows the intercepting of appends.
+// It also maintains the responsibility of assigning global ref IDs to a series via the label store.
 type Fanout struct {
 	mut sync.RWMutex
 	// children is where to fan out.

internal/component/prometheus/fanout_test.go

@@ -3,9 +3,10 @@ package prometheus
 import (
 	"testing"
 
-	"github.com/grafana/alloy/internal/service/labelstore"
 	"github.com/prometheus/client_golang/prometheus"
 
+	"github.com/grafana/alloy/internal/service/labelstore"
+
 	"github.com/prometheus/prometheus/storage"
 
 	"github.com/stretchr/testify/require"

internal/component/prometheus/interceptor.go

@@ -8,9 +8,6 @@ import (
 	"github.com/prometheus/prometheus/model/labels"
 	"github.com/prometheus/prometheus/model/metadata"
 	"github.com/prometheus/prometheus/storage"
-	"go.uber.org/atomic"
-
-	"github.com/grafana/alloy/internal/service/labelstore"
 )
 
 // Interceptor is a storage.Appendable which invokes callback functions upon
@@ -26,23 +23,16 @@ type Interceptor struct {
 	// next is the next appendable to pass in the chain.
 	next storage.Appendable
 
-	ls labelstore.LabelStore
-
-	// lastSeriesCount stores the number of series that were sent through the last interceptappender. It helps to estimate how
-	// much memory to allocate for the staleness trackers.
-	lastSeriesCount atomic.Int64
-
 	componentID string
 }
 
 var _ storage.Appendable = (*Interceptor)(nil)
 
 // NewInterceptor creates a new Interceptor storage.Appendable. Options can be
 // provided to NewInterceptor to install custom hooks for different methods.
-func NewInterceptor(next storage.Appendable, ls labelstore.LabelStore, opts ...InterceptorOption) *Interceptor {
+func NewInterceptor(next storage.Appendable, opts ...InterceptorOption) *Interceptor {
 	i := &Interceptor{
 		next: next,
-		ls:   ls,
 	}
 	for _, opt := range opts {
 		opt(i)
@@ -102,27 +92,23 @@ func WithComponentID(id string) InterceptorOption {
 }
 
 // Appender satisfies the Appendable interface.
-func (f *Interceptor) Appender(ctx context.Context) storage.Appender {
+func (i *Interceptor) Appender(ctx context.Context) storage.Appender {
 	app := &interceptappender{
-		interceptor:       f,
-		ls:                f.ls,
-		stalenessTrackers: make([]labelstore.StalenessTracker, 0, f.lastSeriesCount.Load()),
+		interceptor: i,
 	}
-	if f.next != nil {
-		app.child = f.next.Appender(ctx)
+	if i.next != nil {
+		app.child = i.next.Appender(ctx)
 	}
 	return app
 }
 
-func (f *Interceptor) String() string {
-	return f.componentID + ".receiver"
+func (i *Interceptor) String() string {
+	return i.componentID + ".receiver"
 }
 
 type interceptappender struct {
-	interceptor       *Interceptor
-	child             storage.Appender
-	ls                labelstore.LabelStore
-	stalenessTrackers []labelstore.StalenessTracker
+	interceptor *Interceptor
+	child       storage.Appender
 }
 
 func (a *interceptappender) SetOptions(opts *storage.AppendOptions) {
@@ -135,15 +121,6 @@ var _ storage.Appender = (*interceptappender)(nil)
 
 // Append satisfies the Appender interface.
 func (a *interceptappender) Append(ref storage.SeriesRef, l labels.Labels, t int64, v float64) (storage.SeriesRef, error) {
-	if ref == 0 {
-		ref = storage.SeriesRef(a.ls.GetOrAddGlobalRefID(l))
-	}
-	a.stalenessTrackers = append(a.stalenessTrackers, labelstore.StalenessTracker{
-		GlobalRefID: uint64(ref),
-		Labels:      l,
-		Value:       v,
-	})
-
 	if a.interceptor.onAppend != nil {
 		return a.interceptor.onAppend(ref, l, t, v, a.child)
 	}
@@ -155,8 +132,6 @@ func (a *interceptappender) Append(ref storage.SeriesRef, l labels.Labels, t int
 
 // Commit satisfies the Appender interface.
 func (a *interceptappender) Commit() error {
-	a.interceptor.lastSeriesCount.Store(int64(len(a.stalenessTrackers)))
-	a.ls.TrackStaleness(a.stalenessTrackers)
 	if a.child == nil {
 		return nil
 	}
@@ -165,8 +140,6 @@ func (a *interceptappender) Commit() error {
 
 // Rollback satisfies the Appender interface.
 func (a *interceptappender) Rollback() error {
-	a.interceptor.lastSeriesCount.Store(int64(len(a.stalenessTrackers)))
-	a.ls.TrackStaleness(a.stalenessTrackers)
 	if a.child == nil {
 		return nil
 	}
@@ -180,10 +153,6 @@ func (a *interceptappender) AppendExemplar(
 	e exemplar.Exemplar,
 ) (storage.SeriesRef, error) {
 
-	if ref == 0 {
-		ref = storage.SeriesRef(a.ls.GetOrAddGlobalRefID(l))
-	}
-
 	if a.interceptor.onAppendExemplar != nil {
 		return a.interceptor.onAppendExemplar(ref, l, e, a.child)
 	}
@@ -200,10 +169,6 @@ func (a *interceptappender) UpdateMetadata(
 	m metadata.Metadata,
 ) (storage.SeriesRef, error) {
 
-	if ref == 0 {
-		ref = storage.SeriesRef(a.ls.GetOrAddGlobalRefID(l))
-	}
-
 	if a.interceptor.onUpdateMetadata != nil {
 		return a.interceptor.onUpdateMetadata(ref, l, m, a.child)
 	}
@@ -221,10 +186,6 @@ func (a *interceptappender) AppendHistogram(
 	fh *histogram.FloatHistogram,
 ) (storage.SeriesRef, error) {
 
-	if ref == 0 {
-		ref = storage.SeriesRef(a.ls.GetOrAddGlobalRefID(l))
-	}
-	// TODO histograms are not currently tracked for staleness causing them to be held forever
 	if a.interceptor.onAppendHistogram != nil {
 		return a.interceptor.onAppendHistogram(ref, l, t, h, fh, a.child)
 	}
@@ -240,10 +201,6 @@ func (a *interceptappender) AppendCTZeroSample(
 	t, ct int64,
 ) (storage.SeriesRef, error) {
 
-	if ref == 0 {
-		ref = storage.SeriesRef(a.ls.GetOrAddGlobalRefID(l))
-	}
-
 	if a.interceptor.onAppendCTZeroSample != nil {
 		return a.interceptor.onAppendCTZeroSample(ref, l, t, ct, a.child)
 	}
@@ -261,10 +218,6 @@ func (a *interceptappender) AppendHistogramCTZeroSample(
 	fh *histogram.FloatHistogram,
 ) (storage.SeriesRef, error) {
 
-	if ref == 0 {
-		ref = storage.SeriesRef(a.ls.GetOrAddGlobalRefID(l))
-	}
-
 	if a.child == nil {
 		return 0, nil
 	}