Add metric correctness support to testbed

testbed/correctness/metrics/correctness_test_case.go

@@ -0,0 +1,89 @@
+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,50 @@
+// 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/internal/data"
+
+type metricReceived struct {
+ md data.MetricData
+ received bool
+}
+
+type metricsReceivedIndex struct {
+ m map[string]*metricReceived
+}
+
+func newMetricsReceivedIndex(mds []data.MetricData) *metricsReceivedIndex {
+ mi := &metricsReceivedIndex{m: map[string]*metricReceived{}}
+ for _, md := range mds {
+ metrics := md.ResourceMetrics().At(0).InstrumentationLibraryMetrics().At(0).Metrics()
+ name := metrics.At(0).Name()
+ mi.m[name] = &metricReceived{md: md}
+ }
+ return mi
+}
+
+func (mi *metricsReceivedIndex) lookup(name string) (*metricReceived, bool) {
+ md, ok := mi.m[name]
+ return md, 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/internal/data"
+)
+
+type metricSupplier struct {
+ mds []data.MetricData
+ currIdx int
+}
+
+func newMetricSupplier(mds []data.MetricData) *metricSupplier {
+ return &metricSupplier{mds: mds}
+}
+
+func (p *metricSupplier) nextMetricData() (md data.MetricData, done bool) {
+ if p.currIdx == len(p.mds) {
+ return data.MetricData{}, true
+ }
+ md = p.mds[p.currIdx]
+ p.currIdx++
+ return md, 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/internal/data"
+ "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) []data.MetricData {
+ 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)
+ }
+}