feat: Introduce SeriesRefMappingStore

docs/sources/reference/cli/run.md

@@ -66,19 +66,28 @@ The following flags are supported:
 * `--stability.level`: The minimum permitted stability level of functionality. Supported values: `experimental`, `public-preview`, and `generally-available` (default `"generally-available"`).
 * `--feature.community-components.enabled`: Enable community components (default `false`).
 * `--feature.component-shutdown-deadline`: Maximum duration to wait for a component to shut down before giving up and logging an error (default `"10m"`).
+* `--feature.prometheus.direct-fanout.enabled`: Enable experimental direct fanout for metric forwarding without a global label store.
 * `--windows.priority`: The priority to set for the {{< param "PRODUCT_NAME" >}} process when running on Windows. This is only available on Windows. Supported values: `above_normal`, `below_normal`, `normal`, `high`, `idle`, or `realtime` (default `"normal"`).
 
 {{< admonition type="note" >}}
-The `--windows.priority` flag is in [Public preview][] and is not covered by {{< param "FULL_PRODUCT_NAME" >}} [backward compatibility][] guarantees.
+The `--feature.prometheus.direct-fanout.enabled` flag is an [experimental][] feature.
+Experimental features are subject to frequent breaking changes, and may be removed with no equivalent replacement.
+To enable and use an experimental feature, you must set the `stability.level` [flag](#permitted-stability-levels) to `experimental`.
 
-### Deprecated flags
+[experimental]: https://grafana.com/docs/release-life-cycle/
+{{< /admonition >}}
 
-* `--feature.prometheus.metric-validation-scheme`: This flag is deprecated and has no effect. You can configure the metric validation scheme individually for each `prometheus.scrape` component in your {{< param "PRODUCT_NAME" >}} configuration file.
+{{< admonition type="note" >}}
+The `--windows.priority` flag is in [Public preview][] and is not covered by {{< param "FULL_PRODUCT_NAME" >}} [backward compatibility][] guarantees.
 
 [Public preview]: https://grafana.com/docs/release-life-cycle/
 [backward compatibility]: ../../../introduction/backward-compatibility/
 {{< /admonition >}}
 
+### Deprecated flags
+
+* `--feature.prometheus.metric-validation-scheme`: This flag is deprecated and has no effect. You can configure the metric validation scheme individually for each `prometheus.scrape` component in your {{< param "PRODUCT_NAME" >}} configuration file.
+
 ## Update the configuration file
 
 The configuration file can be reloaded from disk by either:

internal/alloycli/cmd_run.go

@@ -19,7 +19,6 @@ import (
 
 	"github.com/fatih/color"
 	"github.com/go-kit/log"
-	"github.com/grafana/alloy/internal/util"
 	"github.com/grafana/ckit/advertise"
 	"github.com/grafana/ckit/peer"
 	"github.com/prometheus/client_golang/prometheus"
@@ -28,6 +27,8 @@ import (
 	"go.opentelemetry.io/otel"
 	"golang.org/x/exp/maps"
 
+	"github.com/grafana/alloy/internal/util"
+
 	"github.com/grafana/alloy/internal/alloyseed"
 	"github.com/grafana/alloy/internal/boringcrypto"
 	"github.com/grafana/alloy/internal/component"
@@ -170,6 +171,7 @@ depending on the nature of the reload error.
 		cmd.Flags().StringVar(&r.windowsPriority, "windows.priority", r.windowsPriority, fmt.Sprintf("Process priority to use when running on windows. This flag is currently in public preview. Supported values: %s", strings.Join(slices.Collect(windowspriority.PriorityValues()), ", ")))
 	}
 	cmd.Flags().DurationVar(&r.taskShutdownDeadline, "feature.component-shutdown-deadline", r.taskShutdownDeadline, "Maximum duration to wait for a component to shut down before giving up and logging an error")
+	cmd.Flags().BoolVar(&r.enableDirectFanout, "feature.prometheus.direct-fanout.enabled", r.enableDirectFanout, "Enable experimental direct fanout for metric forwarding without a global label store")
 
 	addDeprecatedFlags(cmd)
 	return cmd
@@ -184,6 +186,7 @@ type alloyRun struct {
 	enablePprof                  bool
 	disableReporting             bool
 	clusterEnabled               bool
+	enableDirectFanout           bool
 	clusterNodeName              string
 	clusterAdvAddr               string
 	clusterJoinAddr              string
@@ -208,6 +211,18 @@ type alloyRun struct {
 	taskShutdownDeadline         time.Duration
 }
 
+func (fr *alloyRun) checkExperimentalFlags() error {
+	if fr.minStability.Permits(featuregate.StabilityExperimental) {
+		return nil
+	}
+
+	if fr.enableDirectFanout {
+		return fmt.Errorf("the '--feature.prometheus.direct-fanout.enabled' can be used only at experimental stability level")
+	}
+
+	return nil
+}
+
 func (fr *alloyRun) Run(cmd *cobra.Command, configPath string) error {
 	var wg sync.WaitGroup
 	defer wg.Wait()
@@ -219,6 +234,10 @@ func (fr *alloyRun) Run(cmd *cobra.Command, configPath string) error {
 		return fmt.Errorf("path argument not provided")
 	}
 
+	if err := fr.checkExperimentalFlags(); err != nil {
+		return err
+	}
+
 	// Buffer logs until log format has been determined
 	l, err := logging.NewDeferred(os.Stderr)
 	if err != nil {
@@ -370,7 +389,11 @@ func (fr *alloyRun) Run(cmd *cobra.Command, configPath string) error {
 		return fmt.Errorf("failed to create otel service")
 	}
 
-	labelService := labelstore.New(l, reg)
+	if fr.enableDirectFanout {
+		level.Info(l).Log("msg", "global label store is disabled")
+	}
+
+	labelService := labelstore.New(l, reg, !fr.enableDirectFanout)
 	alloyseed.Init(fr.storagePath, l)
 
 	f, err := alloy_runtime.New(alloy_runtime.Options{

internal/component/otelcol/exporter/prometheus/prometheus.go

@@ -8,14 +8,15 @@ import (
 	"time"
 
 	"github.com/go-kit/log"
+	"github.com/prometheus/prometheus/storage"
+
 	"github.com/grafana/alloy/internal/component"
 	"github.com/grafana/alloy/internal/component/otelcol"
 	"github.com/grafana/alloy/internal/component/otelcol/exporter/prometheus/internal/convert"
 	"github.com/grafana/alloy/internal/component/otelcol/internal/lazyconsumer"
 	"github.com/grafana/alloy/internal/component/prometheus"
 	"github.com/grafana/alloy/internal/featuregate"
 	"github.com/grafana/alloy/internal/service/labelstore"
-	"github.com/prometheus/prometheus/storage"
 )
 
 func init() {
@@ -116,6 +117,8 @@ func New(o component.Options, c Arguments) (*Component, error) {
 
 // Run implements Component.
 func (c *Component) Run(ctx context.Context) error {
+	defer c.fanout.Clear()
+
 	for {
 		select {
 		case <-ctx.Done():

internal/component/prometheus/appenders/new.go

@@ -0,0 +1,21 @@
+package appenders
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/prometheus/storage"
+)
+
+// New returns an appropriate appender based on the number of children.
+func New(children []storage.Appender, store *SeriesRefMappingStore, deadRefThreshold storage.SeriesRef, writeLatency prometheus.Histogram, samplesForwarded prometheus.Counter) storage.Appender {
+	// No destination, no work to do.
+	if len(children) == 0 {
+		return Noop{}
+	}
+
+	// Single destination, no need to fanout.
+	if len(children) == 1 {
+		return NewPassthrough(children[0], deadRefThreshold, writeLatency, samplesForwarded)
+	}
+
+	return NewSeriesRefMapping(children, store, writeLatency, samplesForwarded)
+}

internal/component/prometheus/appenders/new_test.go

@@ -0,0 +1,63 @@
+package appenders
+
+import (
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/storage"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNew_NoChildrenReturnsNoop(t *testing.T) {
+	app := New(nil, nil, 0, nil, nil)
+
+	_, ok := app.(Noop)
+	assert.True(t, ok, "expected Noop appender for zero children")
+}
+
+func TestNew_SingleChildReturnsPassthrough(t *testing.T) {
+	child := &mockAppender{}
+	app := New([]storage.Appender{child}, nil, 0, nil, nil)
+
+	_, ok := app.(*passthrough)
+	assert.True(t, ok, "expected passthrough appender for single child")
+}
+
+func TestNew_MultipleChildrenReturnsSeriesRefMapping(t *testing.T) {
+	store := NewSeriesRefMappingStore(nil)
+	t.Cleanup(func() { store.Clear() })
+
+	child1 := &mockAppender{}
+	child2 := &mockAppender{}
+	app := New([]storage.Appender{child1, child2}, store, 0, nil, nil)
+
+	_, ok := app.(*seriesRefMapping)
+	assert.True(t, ok, "expected seriesRefMapping appender for multiple children")
+}
+
+func TestNew_PassthroughReceivesDeadRefThreshold(t *testing.T) {
+	store := NewSeriesRefMappingStore(nil)
+
+	// Issue a mapping so nextUniqueRef advances past 1.
+	lbls := labels.FromStrings("job", "test")
+	store.CreateMapping([]storage.SeriesRef{100, 200}, lbls)
+
+	// Clear advances firstRefOfCurrentGeneration to the current nextUniqueRef.
+	threshold := store.Clear()
+	require.Greater(t, uint64(threshold), uint64(0))
+
+	sf := prometheus.NewCounter(prometheus.CounterOpts{Name: "test_forwarded", Help: "test"})
+	child := &mockAppender{appendFn: func(ref storage.SeriesRef, _ labels.Labels, _ int64, _ float64) (storage.SeriesRef, error) {
+		return ref, nil // echo back whatever ref we receive
+	}}
+	app := New([]storage.Appender{child}, store, threshold, nil, sf)
+
+	// A ref below the threshold must be zeroed by the passthrough.
+	staleRef := threshold - 1
+	_, err := app.Append(staleRef, lbls, 1, 1.0)
+	require.NoError(t, err)
+	require.Equal(t, storage.SeriesRef(0), child.appendRefs[0],
+		"passthrough must zero refs below the dead ref threshold")
+}

internal/component/prometheus/appenders/noop.go

@@ -0,0 +1,53 @@
+package appenders
+
+import (
+	"context"
+
+	"github.com/prometheus/prometheus/model/exemplar"
+	"github.com/prometheus/prometheus/model/histogram"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/model/metadata"
+	"github.com/prometheus/prometheus/storage"
+)
+
+type Noop struct {
+}
+
+func (n Noop) Appender(_ context.Context) storage.Appender {
+	return n
+}
+
+func (n Noop) Append(ref storage.SeriesRef, _ labels.Labels, _ int64, _ float64) (storage.SeriesRef, error) {
+	return ref, nil
+}
+
+func (n Noop) Commit() error {
+	return nil
+}
+
+func (n Noop) Rollback() error {
+	return nil
+}
+
+func (n Noop) SetOptions(_ *storage.AppendOptions) {
+}
+
+func (n Noop) AppendExemplar(ref storage.SeriesRef, _ labels.Labels, _ exemplar.Exemplar) (storage.SeriesRef, error) {
+	return ref, nil
+}
+
+func (n Noop) AppendHistogram(ref storage.SeriesRef, _ labels.Labels, _ int64, _ *histogram.Histogram, _ *histogram.FloatHistogram) (storage.SeriesRef, error) {
+	return ref, nil
+}
+
+func (n Noop) AppendHistogramSTZeroSample(ref storage.SeriesRef, _ labels.Labels, _, _ int64, _ *histogram.Histogram, _ *histogram.FloatHistogram) (storage.SeriesRef, error) {
+	return ref, nil
+}
+
+func (n Noop) UpdateMetadata(ref storage.SeriesRef, _ labels.Labels, _ metadata.Metadata) (storage.SeriesRef, error) {
+	return ref, nil
+}
+
+func (n Noop) AppendSTZeroSample(ref storage.SeriesRef, _ labels.Labels, _, _ int64) (storage.SeriesRef, error) {
+	return ref, nil
+}

internal/component/prometheus/appenders/passthrough.go

@@ -0,0 +1,111 @@
+package appenders
+
+import (
+	"time"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/prometheus/model/exemplar"
+	"github.com/prometheus/prometheus/model/histogram"
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/prometheus/prometheus/model/metadata"
+	"github.com/prometheus/prometheus/storage"
+)
+
+type passthrough struct {
+	wrapping         storage.Appender
+	start            time.Time
+	writeLatency     prometheus.Histogram
+	samplesForwarded prometheus.Counter
+	// deadRefThreshold marks the boundary of the current ref generation. Any incoming
+	// ref below this value is from a previous generation and meaningless to this child;
+	// it must be zeroed so the child allocates a fresh ref.
+	deadRefThreshold storage.SeriesRef
+}
+
+func NewPassthrough(wrapping storage.Appender, deadRefThreshold storage.SeriesRef, writeLatency prometheus.Histogram, samplesForwarded prometheus.Counter) storage.Appender {
+	return &passthrough{
+		wrapping:         wrapping,
+		deadRefThreshold: deadRefThreshold,
+		writeLatency:     writeLatency,
+		samplesForwarded: samplesForwarded,
+	}
+}
+
+// sanitizeRef zeros ref if it is from a previous generation.
+func (p *passthrough) sanitizeRef(ref storage.SeriesRef) storage.SeriesRef {
+	if ref != 0 && ref < p.deadRefThreshold {
+		return 0
+	}
+	return ref
+}
+
+func (p *passthrough) Commit() error {
+	defer p.recordLatency()
+	return p.wrapping.Commit()
+}
+
+func (p *passthrough) Rollback() error {
+	defer p.recordLatency()
+	return p.wrapping.Rollback()
+}
+
+func (p *passthrough) recordLatency() {
+	if p.start.IsZero() {
+		return
+	}
+	duration := time.Since(p.start)
+	p.writeLatency.Observe(duration.Seconds())
+}
+
+func (p *passthrough) SetOptions(opts *storage.AppendOptions) {
+	p.wrapping.SetOptions(opts)
+}
+
+func (p *passthrough) Append(ref storage.SeriesRef, l labels.Labels, t int64, v float64) (storage.SeriesRef, error) {
+	if p.start.IsZero() {
+		p.start = time.Now()
+	}
+
+	ref, err := p.wrapping.Append(p.sanitizeRef(ref), l, t, v)
+
+	if err == nil {
+		p.samplesForwarded.Inc()
+	}
+
+	return ref, err
+}
+
+func (p *passthrough) AppendExemplar(ref storage.SeriesRef, l labels.Labels, e exemplar.Exemplar) (storage.SeriesRef, error) {
+	if p.start.IsZero() {
+		p.start = time.Now()
+	}
+	return p.wrapping.AppendExemplar(p.sanitizeRef(ref), l, e)
+}
+
+func (p *passthrough) AppendHistogram(ref storage.SeriesRef, l labels.Labels, t int64, h *histogram.Histogram, fh *histogram.FloatHistogram) (storage.SeriesRef, error) {
+	if p.start.IsZero() {
+		p.start = time.Now()
+	}
+	return p.wrapping.AppendHistogram(p.sanitizeRef(ref), l, t, h, fh)
+}
+
+func (p *passthrough) AppendHistogramSTZeroSample(ref storage.SeriesRef, l labels.Labels, t, st int64, h *histogram.Histogram, fh *histogram.FloatHistogram) (storage.SeriesRef, error) {
+	if p.start.IsZero() {
+		p.start = time.Now()
+	}
+	return p.wrapping.AppendHistogramSTZeroSample(p.sanitizeRef(ref), l, t, st, h, fh)
+}
+
+func (p *passthrough) UpdateMetadata(ref storage.SeriesRef, l labels.Labels, m metadata.Metadata) (storage.SeriesRef, error) {
+	if p.start.IsZero() {
+		p.start = time.Now()
+	}
+	return p.wrapping.UpdateMetadata(p.sanitizeRef(ref), l, m)
+}
+
+func (p *passthrough) AppendSTZeroSample(ref storage.SeriesRef, l labels.Labels, t, st int64) (storage.SeriesRef, error) {
+	if p.start.IsZero() {
+		p.start = time.Now()
+	}
+	return p.wrapping.AppendSTZeroSample(p.sanitizeRef(ref), l, t, st)
+}