Use exp package, don't use generics, split into separate file

Signed-off-by: Saswata Mukherjee <saswataminsta@yahoo.com>

Author: saswatamcode

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/nelkinda/health-go v0.0.1
 	github.com/oklog/run v1.1.0
 	github.com/prometheus/client_golang v1.20.6-0.20250117112434-e0800f53b498
+	github.com/prometheus/client_golang/exp v0.0.0-20250131124507-433de6814d4b
 	github.com/prometheus/client_model v0.6.1
 	github.com/prometheus/common v0.61.0
 	github.com/prometheus/prometheus v0.53.1
@@ -32,5 +33,3 @@ require (
 	google.golang.org/protobuf v1.35.2 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
-
-replace github.com/prometheus/client_golang => ../client_golang

go.sum

@@ -98,6 +98,10 @@ github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINE
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/prometheus/client_golang v1.20.6-0.20250117112434-e0800f53b498 h1:pvxJNIN8Tr8FleTHltMExyfFF2GOFBcYbgggNDKvWDw=
+github.com/prometheus/client_golang v1.20.6-0.20250117112434-e0800f53b498/go.mod h1:SJYWANZZtDbfu6CSYeDvRDkzhGX71yepDKePKDTya3M=
+github.com/prometheus/client_golang/exp v0.0.0-20250131124507-433de6814d4b h1:SeZR11qcDl6QeY27jlrC3dNggRyImBlKw3g2g4GJ7LA=
+github.com/prometheus/client_golang/exp v0.0.0-20250131124507-433de6814d4b/go.mod h1:0HIB0F0XkTu3oZ6jBaF7CMLznKdaOuFCTOq8lxgsN9Q=
 github.com/prometheus/client_model v0.6.1 h1:ZKSh/rekM+n3CeS952MLRAdFwIKqeY8b62p8ais2e9E=
 github.com/prometheus/client_model v0.6.1/go.mod h1:OrxVMOVHjw3lKMa8+x6HeMGkHMQyHDk9E3jmP2AmGiY=
 github.com/prometheus/common v0.61.0 h1:3gv/GThfX0cV2lpO7gkTUwZru38mxevy90Bj8YFSRQQ=

metrics/write.go

@@ -27,7 +27,6 @@ import (
 
 	"github.com/prometheus-community/avalanche/pkg/errors"
 	"github.com/prometheus/client_golang/exp/api/remote"
-	writev2 "github.com/prometheus/client_golang/exp/api/remote/genproto/v2"
 
 	"github.com/prometheus/client_golang/prometheus"
 	dto "github.com/prometheus/client_model/go"
@@ -180,113 +179,6 @@ func cloneRequest(r *http.Request) *http.Request {
 	return r2
 }
 
-func (c *Client) writeV2(ctx context.Context) error {
-	select {
-	// Wait for update first as write and collector.Run runs simultaneously.
-	case <-c.config.UpdateNotify:
-	case <-ctx.Done():
-		return ctx.Err()
-	}
-
-	tss, st, err := collectMetricsV2(c.gatherer, c.config.OutOfOrder)
-	if err != nil {
-		return err
-	}
-
-	var (
-		totalTime       time.Duration
-		totalSamplesExp = len(tss) * c.config.RequestCount
-		totalSamplesAct int
-		mtx             sync.Mutex
-		wgMetrics       sync.WaitGroup
-		merr            = &errors.MultiError{}
-	)
-
-	shouldRunForever := c.config.RequestCount == -1
-	if shouldRunForever {
-		log.Printf("Sending: %v timeseries infinitely, %v timeseries per request, %v delay between requests\n",
-			len(tss), c.config.BatchSize, c.config.RequestInterval)
-	} else {
-		log.Printf("Sending: %v timeseries, %v times, %v timeseries per request, %v delay between requests\n",
-			len(tss), c.config.RequestCount, c.config.BatchSize, c.config.RequestInterval)
-	}
-
-	ticker := time.NewTicker(c.config.RequestInterval)
-	defer ticker.Stop()
-
-	concurrencyLimitCh := make(chan struct{}, c.config.Concurrency)
-
-	for i := 0; ; {
-		if ctx.Err() != nil {
-			return ctx.Err()
-		}
-
-		if !shouldRunForever {
-			if i >= c.config.RequestCount {
-				break
-			}
-			i++
-		}
-
-		<-ticker.C
-		select {
-		case <-c.config.UpdateNotify:
-			log.Println("updating remote write metrics")
-			tss, st, err = collectMetricsV2(c.gatherer, c.config.OutOfOrder)
-			if err != nil {
-				merr.Add(err)
-			}
-		default:
-			tss = updateTimestampsV2(tss)
-		}
-
-		start := time.Now()
-		for i := 0; i < len(tss); i += c.config.BatchSize {
-			wgMetrics.Add(1)
-			concurrencyLimitCh <- struct{}{}
-			go func(i int) {
-				defer func() {
-					<-concurrencyLimitCh
-				}()
-				defer wgMetrics.Done()
-				end := i + c.config.BatchSize
-				if end > len(tss) {
-					end = len(tss)
-				}
-				req := &writev2.Request{
-					Timeseries: tss[i:end],
-					Symbols:    st.Symbols(), // We pass full symbols table to each request for now
-				}
-
-				if _, err := c.remoteAPI.Write(ctx, req); err != nil {
-					merr.Add(err)
-					c.logger.Error("error writing metrics", "error", err)
-					return
-				}
-
-				mtx.Lock()
-				totalSamplesAct += len(tss[i:end])
-				mtx.Unlock()
-			}(i)
-		}
-		wgMetrics.Wait()
-		totalTime += time.Since(start)
-		if merr.Count() > 20 {
-			merr.Add(fmt.Errorf("too many errors"))
-			return merr.Err()
-		}
-	}
-	if c.config.RequestCount*len(tss) != totalSamplesAct {
-		merr.Add(fmt.Errorf("total samples mismatch, exp:%v , act:%v", totalSamplesExp, totalSamplesAct))
-	}
-	c.logger.Info("metrics summary",
-		"total_time", totalTime.Round(time.Second),
-		"total_samples", totalSamplesAct,
-		"samples_per_sec", int(float64(totalSamplesAct)/totalTime.Seconds()),
-		"errors", merr.Count())
-	return merr.Err()
-}
-
 func (c *Client) write(ctx context.Context) error {
 	select {
 	// Wait for update first as write and collector.Run runs simultaneously.
@@ -393,14 +285,6 @@ func (c *Client) write(ctx context.Context) error {
 	return merr.Err()
 }
 
-func updateTimestampsV2(tss []*writev2.TimeSeries) []*writev2.TimeSeries {
-	now := time.Now().UnixMilli()
-	for i := range tss {
-		tss[i].Samples[0].Timestamp = now
-	}
-	return tss
-}
-
 func updateTimetamps(tss []prompb.TimeSeries) []prompb.TimeSeries {
 	t := int64(model.Now())
 	for i := range tss {
@@ -409,18 +293,6 @@ func updateTimetamps(tss []prompb.TimeSeries) []prompb.TimeSeries {
 	return tss
 }
 
-func collectMetricsV2(gatherer prometheus.Gatherer, outOfOrder bool) ([]*writev2.TimeSeries, writev2.SymbolsTable, error) {
-	metricFamilies, err := gatherer.Gather()
-	if err != nil {
-		return nil, writev2.SymbolsTable{}, err
-	}
-	tss, st := ToTimeSeriesSliceV2(metricFamilies)
-	if outOfOrder {
-		tss = shuffleTimestampsV2(tss)
-	}
-	return tss, st, nil
-}
-
 func collectMetrics(gatherer prometheus.Gatherer, outOfOrder bool) ([]prompb.TimeSeries, error) {
 	metricFamilies, err := gatherer.Gather()
 	if err != nil {
@@ -433,16 +305,6 @@ func collectMetrics(gatherer prometheus.Gatherer, outOfOrder bool) ([]prompb.Tim
 	return tss, nil
 }
 
-func shuffleTimestampsV2(tss []*writev2.TimeSeries) []*writev2.TimeSeries {
-	now := time.Now().UnixMilli()
-	offsets := []int64{0, -60 * 1000, -5 * 60 * 1000}
-	for i := range tss {
-		offset := offsets[i%len(offsets)]
-		tss[i].Samples[0].Timestamp = now + offset
-	}
-	return tss
-}
-
 func shuffleTimestamps(tss []prompb.TimeSeries) []prompb.TimeSeries {
 	now := time.Now().UnixMilli()
 	offsets := []int64{0, -60 * 1000, -5 * 60 * 1000}
@@ -490,47 +352,6 @@ func ToTimeSeriesSlice(metricFamilies []*dto.MetricFamily) []prompb.TimeSeries {
 	return tss
 }
 
-func ToTimeSeriesSliceV2(metricFamilies []*dto.MetricFamily) ([]*writev2.TimeSeries, writev2.SymbolsTable) {
-	st := writev2.NewSymbolTable()
-	timestamp := int64(model.Now())
-	tss := make([]*writev2.TimeSeries, 0, len(metricFamilies)*10)
-
-	skippedSamples := 0
-	for _, metricFamily := range metricFamilies {
-		for _, metric := range metricFamily.Metric {
-			labels := prompbLabels(*metricFamily.Name, metric.Label)
-			labelRefs := make([]uint32, 0, len(labels))
-			for _, label := range labels {
-				labelRefs = append(labelRefs, st.Symbolize(label.Name))
-				labelRefs = append(labelRefs, st.Symbolize(label.Value))
-			}
-			ts := &writev2.TimeSeries{
-				LabelsRefs: labelRefs,
-			}
-			switch *metricFamily.Type {
-			case dto.MetricType_COUNTER:
-				ts.Samples = []*writev2.Sample{{
-					Value:     *metric.Counter.Value,
-					Timestamp: timestamp,
-				}}
-				tss = append(tss, ts)
-			case dto.MetricType_GAUGE:
-				ts.Samples = []*writev2.Sample{{
-					Value:     *metric.Gauge.Value,
-					Timestamp: timestamp,
-				}}
-				tss = append(tss, ts)
-			default:
-				skippedSamples++
-			}
-		}
-	}
-	if skippedSamples > 0 {
-		log.Printf("WARN: Skipping %v samples; sending only %v samples, given only gauge and counters are currently implemented\n", skippedSamples, len(tss))
-	}
-	return tss, st
-}
-
 func prompbLabels(name string, label []*dto.LabelPair) []prompb.Label {
 	ret := make([]prompb.Label, 0, len(label)+1)
 	ret = append(ret, prompb.Label{