[HCP Observability] OTELSink

[Achooo]

Description

Overall Feature background

This PR is part of the HCP Observability feature to send Server metrics to HCP's new Telemetry Gateway. The idea is to create a new OpenTelemetry go-metrics sink that will be initialized automatically when a Consul server is registered with the HCP management plane.

• We will be merging these approved changes into feature/hcp-telemetry
• That branch will be merged at the end into main and back ported
  This allows us to iteratively improve on this feature over time before. Split up for easier PR review. 🚢

This PR: MetricsClient

This PR adds a new OTELSink to the new Telemetry package.
It must satisfy the go metrics MetricsSink interface to be used by the fanout sink (follow up PR).
The key responsibility of the OTELExpoter is to use the MetricsClient to export metrics. To do this:

• Receives an emitted Consul metric
• Registers a metric with an OTEL instrument
• Periodically aggregates and exports data using a PeriodicReader. This reader uses the Exporter implemented in [HCP Observability] OTELExporter #17128

The OTEL library pieces are pegged at this commit hash 7dea2225a218872e86d2f580e82c089b321617b0 version due to build failures on certain architectures. See this build failure.

See this GitHub issue:

Although the fixes were implemented, this has yet to be released.
As soon as the fix is released, this version will be updated.

Next steps

• The sink will be initialized when a consul server is registered with CCM
• The initialized sink will be registered in lib/telemetry.go's fanout sinks

Reference image

Testing & Reproduction steps

Links

PR Checklist

• updated test coverage

• external facing docs updated

• appropriate backport labels added (not yet, in feature branch)

• not a security concern

[Achooo]

Keeping this struct as we have one last follow up PR that adds default filters and labels to the sink (so we'll have 4 parameters)

[loshz]

This approach is definitely more robust when testing for race conditions!

Small suggestion, I would change the goroutines to not pass the WaitGroup, and also not use a random value in order to save compute:

Suggested change

	gaugeStore := NewGaugeStore()
	wg := &sync.WaitGroup{}
	samples := 100
	for i := 0; i < samples; i++ {
	k := fmt.Sprintf("consul.test.%d", i)
	v := rand.Float64()
	wg.Add(1)
	go storeAndRetrieve(t, k, v, gaugeStore, wg)
	}
	wg.Wait()
	}
	func storeAndRetrieve(t *testing.T, k string, v float64, gaugeStore *gaugeStore, wg *sync.WaitGroup) {
	gaugeStore.Set(k, v, nil)
	gv, ok := gaugeStore.LoadAndDelete(k)
	require.True(t, ok)
	require.Equal(t, v, gv.Value)
	wg.Done()
	}
	gaugeStore := NewGaugeStore()
	wg := &sync.WaitGroup{}
	for i := 0; i < 100; i++ {
	wg.Add(1)
	key := fmt.Sprintf("consul.test.%d", i)
	value := 12.34
	go func() {
	defer wg.Done()
	gaugeStore.Set(key, value, nil)
	gv, ok := gaugeStore.LoadAndDelete(k)
	require.True(t, ok)
	require.Equal(t, value, gv.Value)
	}()
	}
	wg.Wait()
	}

[loshz]

I'm also unsure how the require package would work in this situation. I think it fails the test immediately, so I'm not 100% what will happen to the wg.Wait() 🤔

[loshz]

Yep, it calls t.FailNow() under the hood which is not safe inside a goroutine.

I would create a buffered channel before the loop, send errors in the goroutines, and assert no errors after the call to wg.Wait(). Something like this should do it:

func TestGaugeStore_Race(t *testing.T) {
	t.Parallel()

	gaugeStore := NewGaugeStore()
	wg := &sync.WaitGroup{}
	samples := 100
	errCh := make(chan error, samples)

	for i := 0; i < samples; i++ {
		wg.Add(1)
		key := fmt.Sprintf("consul.test.%d", i)
		value := 12.34
		go func() {
			defer wg.Done()
			gaugeStore.Set(key, value, nil)
			gv, _ := gaugeStore.LoadAndDelete(k)
			if gv.Value != value {
				errCh <- fmt.Errorf("expected value: '%s', but got: '%s' for key: '%s'", value, gv.Value, key)
			}
		}()
	}
	wg.Wait()

	require.Empty(t, errCh) // Might be better to fail with the individual errors, but not required
}

[Achooo]

thanks for catching this, I've fixed it and for the OTELSink test too.

[Achooo]

Using this commit hash 7dea2225a218872e86d2f580e82c089b321617b0 version due to build failures on certain architectures. See this build failure.

See this GitHub issue:

Although the fixes were implemented, this has yet to be released. I will make a JIRA ticket to make sure we update this.

[loshz]

Really great work, @Achooo! Thanks for fixing up the concurrency tests. I've left a couple of nits, but I'll approve as to not require another review.

[loshz]

Nit:

Suggested change

	performSinkOperation(t, sink, k, v, errCh)
	wg.Done()
	defer wg.Done()
	performSinkOperation(t, sink, k, v, errCh)

[loshz]

I don't think we need t anymore?

Suggested change

	func performSinkOperation(t *testing.T, sink *OTELSink, k string, v metricdata.Metrics, errCh chan error) {
	func performSinkOperation(sink *OTELSink, k string, v metricdata.Metrics, errCh chan error) {