Remove labelstore interactions from the interceptor

Author: kgeckhart

CHANGELOG.md

@@ -24,6 +24,8 @@ Main (unreleased)
 
 - Add `meta_cache_address` to `beyla.ebpf` component. (@skl)
 
+- Remove labelstore interactions from the prometheus interceptor simplifying prometheus pipelines. (@kgeckhart)
+
 ### Bugfixes
 
 - `loki.source.api` no longer drops request when relabel rules drops a specific stream. (@kalleep)
@@ -98,7 +100,6 @@ v1.12.0-rc.0
   - The `otelcol.processor.servicegraph` component now supports defining the maximum number of buckets for generated exponential histograms.
   - See the upstream [core][https://github.com/open-telemetry/opentelemetry-collector/blob/v0.139.0/CHANGELOG.md] and [contrib][https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/v0.139.0/CHANGELOG.md] changelogs for more details.
 
-
 ### Enhancements
 
 - Add per-application rate limiting with the `strategy` attribute in the `faro.receiver` component, to prevent one application from consuming the rate limit quota of others. (@hhertout)

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.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

@@ -5,15 +5,16 @@ import (
 	"testing"
 	"time"
 
+	prom "github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/storage"
+	"github.com/stretchr/testify/require"
+
 	"github.com/grafana/alloy/internal/component"
 	"github.com/grafana/alloy/internal/component/discovery"
 	"github.com/grafana/alloy/internal/component/prometheus"
 	"github.com/grafana/alloy/internal/service/labelstore"
 	"github.com/grafana/alloy/internal/util"
-	prom "github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/prometheus/model/labels"
-	"github.com/prometheus/prometheus/storage"
-	"github.com/stretchr/testify/require"
 )
 
 func TestEnricher(t *testing.T) {
@@ -108,9 +109,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/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
 	}

internal/component/prometheus/pipeline_test.go

@@ -221,7 +221,7 @@ func newDefaultPipeline(t testing.TB, logger log.Logger, handlerFunc http.Handle
 	ls := labelstore.New(logger, promclient.DefaultRegisterer)
 	rwAppendable := newRemoteWriteComponent(t, logger, srv.URL, ls)
 	pipelineAppendable := prometheus.NewFanout([]storage.Appendable{rwAppendable}, promclient.DefaultRegisterer, ls)
-	scrapeInterceptor := scrape.NewInterceptor("prometheus.scrape.test", ls, livedebugging.NewLiveDebugging(), pipelineAppendable)
+	scrapeInterceptor := scrape.NewInterceptor("prometheus.scrape.test", livedebugging.NewLiveDebugging(), pipelineAppendable)
 
 	return scrapeInterceptor, srv, ls
 }
@@ -233,7 +233,7 @@ func newRelabelPipeline(t testing.TB, logger log.Logger, handlerFunc http.Handle
 	rwAppendable := newRemoteWriteComponent(t, logger, srv.URL, ls)
 	relabelAppendable := newRelabelComponent(t, logger, []storage.Appendable{rwAppendable}, ls)
 	pipelineAppendable := prometheus.NewFanout([]storage.Appendable{relabelAppendable}, promclient.DefaultRegisterer, ls)
-	scrapeInterceptor := scrape.NewInterceptor("prometheus.scrape.test", ls, livedebugging.NewLiveDebugging(), pipelineAppendable)
+	scrapeInterceptor := scrape.NewInterceptor("prometheus.scrape.test", livedebugging.NewLiveDebugging(), pipelineAppendable)
 
 	return scrapeInterceptor, srv, ls
 }

internal/component/prometheus/receive_http/receive_http_test.go

@@ -17,12 +17,6 @@ import (
 	"time"
 
 	"github.com/golang/snappy"
-	"github.com/grafana/alloy/internal/component"
-	fnet "github.com/grafana/alloy/internal/component/common/net"
-	alloyprom "github.com/grafana/alloy/internal/component/prometheus"
-	"github.com/grafana/alloy/internal/service/labelstore"
-	"github.com/grafana/alloy/internal/util"
-	"github.com/grafana/alloy/syntax/alloytypes"
 	"github.com/phayes/freeport"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/config"
@@ -35,6 +29,13 @@ import (
 	"github.com/stretchr/testify/require"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/protoadapt"
+
+	"github.com/grafana/alloy/internal/component"
+	fnet "github.com/grafana/alloy/internal/component/common/net"
+	alloyprom "github.com/grafana/alloy/internal/component/prometheus"
+	"github.com/grafana/alloy/internal/service/labelstore"
+	"github.com/grafana/alloy/internal/util"
+	"github.com/grafana/alloy/syntax/alloytypes"
 )
 
 // generateTestCertAndKey generates a self-signed certificate and private key for testing
@@ -569,10 +570,8 @@ func testAppendable(actualSamples chan testSample) []storage.Appendable {
 		return ref, nil
 	}
 
-	ls := labelstore.New(nil, prometheus.DefaultRegisterer)
 	return []storage.Appendable{alloyprom.NewInterceptor(
 		nil,
-		ls,
 		alloyprom.WithAppendHook(
 			hookFn))}
 }

internal/component/prometheus/relabel/relabel.go

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

internal/component/prometheus/relabel/relabel_test.go

@@ -25,30 +25,6 @@ import (
 	"github.com/grafana/alloy/syntax"
 )
 
-func TestRelabelThroughAppend(t *testing.T) {
-	appendable, relabeller := generateRelabel(t)
-	lbls := labels.FromStrings("__address__", "localhost")
-
-	app := appendable.Appender(t.Context())
-	relabedRef, err := app.Append(storage.SeriesRef(0), lbls, time.Now().UnixMilli(), 0)
-	require.NoError(t, err)
-	require.NoError(t, app.Commit())
-
-	require.True(t, relabeller.cache.Len() == 1)
-	// Get the first entry since we only have one we can get oldest
-	ref, cachedLbls, _ := relabeller.cache.GetOldest()
-
-	// We shouldn't have allowed a zero ref to be cached
-	require.NotEqual(t, storage.SeriesRef(0), ref)
-	require.NotEqual(t, lbls, cachedLbls)
-
-	// We should have added a new ref after relabeling
-	require.NotEqual(t, storage.SeriesRef(0), relabedRef)
-
-	// That ref should not be the cached ref
-	require.NotEqual(t, relabedRef, ref)
-}
-
 func TestUpdateReset(t *testing.T) {
 	_, relabeller := generateRelabel(t)
 	lbls := labels.FromStrings("__address__", "localhost")
@@ -72,8 +48,7 @@ func TestValidator(t *testing.T) {
 }
 
 func TestNil(t *testing.T) {
-	ls := labelstore.New(nil, prom.DefaultRegisterer)
-	fanout := prometheus.NewInterceptor(nil, ls, prometheus.WithAppendHook(func(ref storage.SeriesRef, _ labels.Labels, _ int64, _ float64, _ storage.Appender) (storage.SeriesRef, error) {
+	fanout := prometheus.NewInterceptor(nil, prometheus.WithAppendHook(func(ref storage.SeriesRef, _ labels.Labels, _ int64, _ float64, _ storage.Appender) (storage.SeriesRef, error) {
 		require.True(t, false)
 		return ref, nil
 	}))
@@ -138,8 +113,7 @@ func TestMetrics(t *testing.T) {
 }
 
 func BenchmarkCache(b *testing.B) {
-	ls := labelstore.New(nil, prom.DefaultRegisterer)
-	fanout := prometheus.NewInterceptor(nil, ls, prometheus.WithAppendHook(func(ref storage.SeriesRef, l labels.Labels, _ int64, _ float64, _ storage.Appender) (storage.SeriesRef, error) {
+	fanout := prometheus.NewInterceptor(nil, prometheus.WithAppendHook(func(ref storage.SeriesRef, l labels.Labels, _ int64, _ float64, _ storage.Appender) (storage.SeriesRef, error) {
 		require.True(b, l.Has("new_label"))
 		return ref, nil
 	}))
@@ -174,8 +148,7 @@ func BenchmarkCache(b *testing.B) {
 }
 
 func generateRelabel(t *testing.T) (storage.Appendable, *Component) {
-	ls := labelstore.New(nil, prom.DefaultRegisterer)
-	fanout := prometheus.NewInterceptor(nil, ls)
+	fanout := prometheus.NewInterceptor(nil)
 	var appendable storage.Appendable
 	relabeller, err := New(component.Options{
 		ID:     "1",

internal/component/prometheus/remotewrite/remote_write.go

@@ -153,7 +153,6 @@ func New(o component.Options, args Arguments) (*Component, error) {
 
 	res.receiver = prometheus.NewInterceptor(
 		res.storage,
-		ls,
 		prometheus.WithComponentID(res.opts.ID),
 		// In the methods below, conversion is needed because remote_writes assume
 		// they are responsible for generating ref IDs. This means two