Add metric correctness support to testbed

testbed/correctness/metrics/correctness_test_case.go

@@ -0,0 +1,103 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package metrics
+
+import (
+	"log"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/service/defaultcomponents"
+	"go.opentelemetry.io/collector/testbed/correctness"
+	"go.opentelemetry.io/collector/testbed/testbed"
+)
+
+type correctnessTestCase struct {
+	t         *testing.T
+	sender    testbed.DataSender
+	receiver  testbed.DataReceiver
+	harness   *testHarness
+	collector *testbed.InProcessCollector
+}
+
+func newCorrectnessTestCase(
+	t *testing.T,
+	sender testbed.DataSender,
+	receiver testbed.DataReceiver,
+	harness *testHarness,
+) *correctnessTestCase {
+	return &correctnessTestCase{t: t, sender: sender, receiver: receiver, harness: harness}
+}
+
+func (tc *correctnessTestCase) startCollector() {
+	tc.collector = testbed.NewInProcessCollector(componentFactories(tc.t), tc.sender.GetCollectorPort())
+	_, err := tc.collector.PrepareConfig(correctness.CreateConfigYaml(tc.sender, tc.receiver, nil, "metrics"))
+	require.NoError(tc.t, err)
+	rd, err := newResultsDir(tc.t.Name())
+	require.NoError(tc.t, err)
+	err = rd.mkDir()
+	require.NoError(tc.t, err)
+	fname, err := rd.fullPath("agent.log")
+	require.NoError(tc.t, err)
+	log.Println("starting collector")
+	_, err = tc.collector.Start(testbed.StartParams{
+		Name:        "Agent",
+		LogFilePath: fname,
+		Cmd:         "foo",
+		CmdArgs:     []string{"--metrics-level=NONE"},
+	})
+	require.NoError(tc.t, err)
+}
+
+func (tc *correctnessTestCase) stopCollector() {
+	_, err := tc.collector.Stop()
+	require.NoError(tc.t, err)
+}
+
+func (tc *correctnessTestCase) startTestbedSender() {
+	log.Println("starting testbed sender")
+	err := tc.sender.Start()
+	require.NoError(tc.t, err)
+}
+
+func (tc *correctnessTestCase) startTestbedReceiver() {
+	log.Println("starting testbed receiver")
+	err := tc.receiver.Start(&testbed.MockTraceConsumer{}, tc.harness, &testbed.MockLogConsumer{})
+	require.NoError(tc.t, err)
+}
+
+func (tc *correctnessTestCase) stopTestbedReceiver() {
+	log.Println("stopping testbed receiver")
+	err := tc.receiver.Stop()
+	require.NoError(tc.t, err)
+}
+
+func (tc *correctnessTestCase) sendFirstMetric() {
+	tc.harness.sendNextMetric()
+}
+
+func (tc *correctnessTestCase) waitForAllMetrics() {
+	log.Println("waiting for allMetricsReceived")
+	<-tc.harness.allMetricsReceived
+	log.Println("all metrics received")
+}
+
+func componentFactories(t *testing.T) component.Factories {
+	factories, err := defaultcomponents.Components()
+	require.NoError(t, err)
+	return factories
+}

testbed/correctness/metrics/doc.go

@@ -0,0 +1,36 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// Package metrics contains functionality for testing an otelcol pipeline end to end for metric correctness.
+// Partly because of how Prometheus works (being pull-based) metrics correctness works differently than
+// the performance testbed in the parent directory. Whereas performance testing sends a relatively large
+// number of datapoints into the collector, this package sends metrics in one at a time, and only sends the
+// next datapoint when the previous datapoint has been processed and compared to the original.
+//
+// Mostly simlar to the performance testing pipeline, this pipeline looks like the following:
+
+// [testbed exporter] -> [otelcol receiver] -> [otelcol exporter] -> [testbed receiver] -> [test harness]
+//
+// the difference being the testHarness, which is connected to [testbed receiver] as its metrics
+// consumer, listening for datapoints. To start the process, one datapoint is sent into the testbed
+// exporter, it goes through the pipeline, and arrives at the testbed receiver, which passes it along to the
+// test harness. The test harness compares the received datapoint to the original datapoint it sent, and saves
+// any diffs it found in a diffAccumulator instance. Then it sends the next datapoint. This continues until
+// there are no more datapoints. The simple diagram above should have a loop, where [test harness] connects
+// back to [testbed exporter].
+//
+// Datapoints are supplied to the testHarness by a metricSupplier, which receives all of the metrics it needs
+// upfront. Those metrics are in turn generated by a metricGenerator, which receives its config from a PICT
+// generated file, as the trace correctness funcionality does.
+package metrics

testbed/correctness/metrics/metric_index.go

@@ -0,0 +1,52 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package metrics
+
+import (
+	"go.opentelemetry.io/collector/consumer/pdata"
+)
+
+type metricReceived struct {
+	pdm      pdata.Metrics
+	received bool
+}
+
+type metricsReceivedIndex struct {
+	m map[string]*metricReceived
+}
+
+func newMetricsReceivedIndex(pdms []pdata.Metrics) *metricsReceivedIndex {
+	mi := &metricsReceivedIndex{m: map[string]*metricReceived{}}
+	for _, pdm := range pdms {
+		metrics := pdm.ResourceMetrics().At(0).InstrumentationLibraryMetrics().At(0).Metrics()
+		name := metrics.At(0).Name()
+		mi.m[name] = &metricReceived{pdm: pdm}
+	}
+	return mi
+}
+
+func (mi *metricsReceivedIndex) lookup(name string) (*metricReceived, bool) {
+	mr, ok := mi.m[name]
+	return mr, ok
+}
+
+func (mi *metricsReceivedIndex) allReceived() bool {
+	for _, m := range mi.m {
+		if !m.received {
+			return false
+		}
+	}
+	return true
+}

testbed/correctness/metrics/metric_supplier.go

@@ -0,0 +1,37 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package metrics
+
+import (
+	"go.opentelemetry.io/collector/consumer/pdata"
+)
+
+type metricSupplier struct {
+	pdms    []pdata.Metrics
+	currIdx int
+}
+
+func newMetricSupplier(pdms []pdata.Metrics) *metricSupplier {
+	return &metricSupplier{pdms: pdms}
+}
+
+func (p *metricSupplier) nextMetrics() (pdm pdata.Metrics, done bool) {
+	if p.currIdx == len(p.pdms) {
+		return pdata.Metrics{}, true
+	}
+	pdm = p.pdms[p.currIdx]
+	p.currIdx++
+	return pdm, false
+}

testbed/correctness/metrics/metrics_correctness_test.go

@@ -0,0 +1,92 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package metrics
+
+import (
+	"fmt"
+	"log"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"go.opentelemetry.io/collector/consumer/pdata"
+	"go.opentelemetry.io/collector/internal/goldendataset"
+	"go.opentelemetry.io/collector/testbed/correctness"
+	"go.opentelemetry.io/collector/testbed/testbed"
+)
+
+func TestMetricsGoldenData(t *testing.T) {
+	tests, err := correctness.LoadPictOutputPipelineDefs("../testdata/generated_pict_pairs_metrics_pipeline.txt")
+	require.NoError(t, err)
+	for _, test := range tests {
+		test.TestName = fmt.Sprintf("%s-%s", test.Receiver, test.Exporter)
+		test.DataSender = correctness.ConstructMetricsSender(t, test.Receiver)
+		test.DataReceiver = correctness.ConstructReceiver(t, test.Exporter)
+		t.Run(test.TestName, func(t *testing.T) {
+			testWithMetricsGoldenDataset(t, test.DataSender.(testbed.MetricDataSender), test.DataReceiver)
+		})
+	}
+}
+
+func testWithMetricsGoldenDataset(t *testing.T, sender testbed.MetricDataSender, receiver testbed.DataReceiver) {
+	mds := getTestMetrics(t)
+	accumulator := newDiffAccumulator()
+	h := newTestHarness(
+		t,
+		newMetricSupplier(mds),
+		newMetricsReceivedIndex(mds),
+		sender,
+		accumulator,
+	)
+	tc := newCorrectnessTestCase(t, sender, receiver, h)
+
+	tc.startTestbedReceiver()
+	tc.startCollector()
+	tc.startTestbedSender()
+
+	tc.sendFirstMetric()
+	tc.waitForAllMetrics()
+
+	tc.stopTestbedReceiver()
+	tc.stopCollector()
+
+	if accumulator.foundDiffs {
+		t.Fail()
+	}
+}
+
+func getTestMetrics(t *testing.T) []pdata.Metrics {
+	const file = "../../../internal/goldendataset/testdata/generated_pict_pairs_metrics.txt"
+	mds, err := goldendataset.GenerateMetricDatas(file)
+	require.NoError(t, err)
+	return mds
+}
+
+type diffAccumulator struct {
+	foundDiffs bool
+}
+
+var _ diffConsumer = (*diffAccumulator)(nil)
+
+func newDiffAccumulator() *diffAccumulator {
+	return &diffAccumulator{}
+}
+
+func (d *diffAccumulator) accept(metricName string, diffs []*MetricDiff) {
+	if len(diffs) > 0 {
+		d.foundDiffs = true
+		log.Printf("Found diffs for [%v]\n%v", metricName, diffs)
+	}
+}