only run registry logic once

observIQ · Sep 5, 2024 · a77062f · a77062f
1 parent 90691a6
commit a77062f
Show file tree

Hide file tree

Showing 3 changed files with 25 additions and 18 deletions.
diff --git a/internal/measurements/throughput_test.go b/internal/measurements/throughput_test.go
@@ -225,7 +225,7 @@ func TestResettableThroughputMeasurementsRegistry(t *testing.T) {
 		tmp.AddMetrics(context.Background(), metrics)
 		tmp.AddTraces(context.Background(), traces)
 
-		reg.RegisterThroughputMeasurements("throughputmeasurement/1", tmp)
+		require.NoError(t, reg.RegisterThroughputMeasurements("throughputmeasurement/1", tmp))
 
 		actualMetrics := reg.OTLPMeasurements(nil)
 
@@ -257,7 +257,7 @@ func TestResettableThroughputMeasurementsRegistry(t *testing.T) {
 		tmp.AddMetrics(context.Background(), metrics)
 		tmp.AddTraces(context.Background(), traces)
 
-		reg.RegisterThroughputMeasurements("throughputmeasurement/1", tmp)
+		require.NoError(t, reg.RegisterThroughputMeasurements("throughputmeasurement/1", tmp))
 
 		actualMetrics := reg.OTLPMeasurements(nil)
 
@@ -281,7 +281,7 @@ func TestResettableThroughputMeasurementsRegistry(t *testing.T) {
 
 		tmp.AddMetrics(context.Background(), metrics)
 
-		reg.RegisterThroughputMeasurements("throughputmeasurement/1", tmp)
+		require.NoError(t, reg.RegisterThroughputMeasurements("throughputmeasurement/1", tmp))
 
 		actualMetrics := reg.OTLPMeasurements(nil)
 
@@ -306,7 +306,7 @@ func TestResettableThroughputMeasurementsRegistry(t *testing.T) {
 
 		tmp.AddMetrics(context.Background(), metrics)
 
-		reg.RegisterThroughputMeasurements("throughputmeasurement/1", tmp)
+		require.NoError(t, reg.RegisterThroughputMeasurements("throughputmeasurement/1", tmp))
 
 		actualMetrics := reg.OTLPMeasurements(nil)
 
@@ -341,7 +341,7 @@ func TestResettableThroughputMeasurementsRegistry(t *testing.T) {
 		tmp.AddMetrics(context.Background(), metrics)
 		tmp.AddTraces(context.Background(), traces)
 
-		reg.RegisterThroughputMeasurements("throughputmeasurement/1", tmp)
+		require.NoError(t, reg.RegisterThroughputMeasurements("throughputmeasurement/1", tmp))
 
 		actualMetrics := reg.OTLPMeasurements(nil)
 
@@ -373,7 +373,7 @@ func TestResettableThroughputMeasurementsRegistry(t *testing.T) {
 		tmp.AddMetrics(context.Background(), metrics)
 		tmp.AddTraces(context.Background(), traces)
 
-		reg.RegisterThroughputMeasurements("throughputmeasurement/1", tmp)
+		require.NoError(t, reg.RegisterThroughputMeasurements("throughputmeasurement/1", tmp))
 
 		reg.Reset()
 

diff --git a/opamp/observiq/measurements_test.go b/opamp/observiq/measurements_test.go
@@ -60,7 +60,7 @@ func TestMeasurementsSender(t *testing.T) {
 		tm.AddMetrics(context.Background(), m)
 
 		reg := measurements.NewResettableThroughputMeasurementsRegistry(false)
-		reg.RegisterThroughputMeasurements(processorID, tm)
+		require.NoError(t, reg.RegisterThroughputMeasurements(processorID, tm))
 
 		ms := newMeasurementsSender(zap.NewNop(), reg, client, 1*time.Millisecond, nil)
 		ms.Start()

diff --git a/processor/throughputmeasurementprocessor/processor.go b/processor/throughputmeasurementprocessor/processor.go
@@ -18,6 +18,7 @@ import (
 	"context"
 	"fmt"
 	"math/rand"
+	"sync"
 
 	"github.com/observiq/bindplane-agent/internal/measurements"
 	"go.opentelemetry.io/collector/component"
@@ -35,6 +36,7 @@ type throughputMeasurementProcessor struct {
 	samplingCutOffRatio float64
 	processorID         component.ID
 	bindplane           component.ID
+	startOnce           sync.Once
 }
 
 func newThroughputMeasurementProcessor(logger *zap.Logger, mp metric.MeterProvider, cfg *Config, processorID component.ID) (*throughputMeasurementProcessor, error) {
@@ -50,24 +52,29 @@ func newThroughputMeasurementProcessor(logger *zap.Logger, mp metric.MeterProvid
 		samplingCutOffRatio: cfg.SamplingRatio,
 		processorID:         processorID,
 		bindplane:           cfg.BindplaneExtension,
+		startOnce:           sync.Once{},
 	}, nil
 }
 
 func (tmp *throughputMeasurementProcessor) start(_ context.Context, host component.Host) error {
+	var err error
+	tmp.startOnce.Do(func() {
+		registry, getRegErr := GetThroughputRegistry(host, tmp.bindplane)
+		if getRegErr != nil {
+			err = fmt.Errorf("get throughput registry: %w", err)
+			return
+		}
 
-	registry, err := GetThroughputRegistry(host, tmp.bindplane)
-	if err != nil {
-		return fmt.Errorf("get throughput registry: %w", err)
-	}
-
-	if registry != nil {
-		err := registry.RegisterThroughputMeasurements(tmp.processorID.String(), tmp.measurements)
-		if err != nil {
-			return fmt.Errorf("register throughput measurements: %w", err)
+		if registry != nil {
+			registerErr := registry.RegisterThroughputMeasurements(tmp.processorID.String(), tmp.measurements)
+			if registerErr != nil {
+				err = fmt.Errorf("register throughput measurements: %w", err)
+				return
+			}
 		}
-	}
+	})
 
-	return nil
+	return err
 }
 
 func (tmp *throughputMeasurementProcessor) processTraces(ctx context.Context, td ptrace.Traces) (ptrace.Traces, error) {