Add connector usage to the testbed

.chloggen/connector-testbed.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: testbed
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add the use of connectors to the testbed
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [30165]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [api]

exporter/elasticsearchexporter/integrationtest/go.mod

@@ -28,6 +28,7 @@ require (
 )
 
 require (
+	github.com/alecthomas/participle/v2 v2.1.1 // indirect
 	github.com/apache/thrift v0.20.0 // indirect
 	github.com/armon/go-radix v1.0.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
@@ -48,6 +49,7 @@ require (
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-ole/go-ole v1.3.0 // indirect
 	github.com/go-viper/mapstructure/v2 v2.0.0-alpha.1 // indirect
+	github.com/gobwas/glob v0.2.3 // indirect
 	github.com/gogo/googleapis v1.4.1 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
@@ -56,6 +58,8 @@ require (
 	github.com/google/uuid v1.6.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.19.1 // indirect
 	github.com/hashicorp/go-version v1.6.0 // indirect
+	github.com/hashicorp/golang-lru v1.0.2 // indirect
+	github.com/iancoleman/strcase v0.3.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/influxdata/go-syslog/v3 v3.0.1-0.20230911200830-875f5bc594a4 // indirect
 	github.com/jaegertracing/jaeger v1.57.0 // indirect
@@ -67,18 +71,22 @@ require (
 	github.com/knadh/koanf/v2 v2.1.1 // indirect
 	github.com/leodido/ragel-machinery v0.0.0-20181214104525-299bdde78165 // indirect
 	github.com/lestrrat-go/strftime v1.0.6 // indirect
+	github.com/lightstep/go-expohisto v1.0.0 // indirect
 	github.com/lufia/plan9stats v0.0.0-20240226150601-1dcf7310316a // indirect
 	github.com/mitchellh/copystructure v1.2.0 // indirect
 	github.com/mitchellh/reflectwalk v1.0.2 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/mostynb/go-grpc-compression v1.2.2 // indirect
+	github.com/open-telemetry/opentelemetry-collector-contrib/connector/routingconnector v0.100.0 // indirect
+	github.com/open-telemetry/opentelemetry-collector-contrib/connector/spanmetricsconnector v0.100.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/opencensusexporter v0.100.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/syslogexporter v0.100.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/exporter/zipkinexporter v0.100.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/coreinternal v0.100.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/internal/sharedcomponent v0.100.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/golden v0.100.0 // indirect
+	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl v0.100.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/pdatautil v0.100.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza v0.100.0 // indirect
 	github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/jaeger v0.100.0 // indirect
@@ -102,6 +110,7 @@ require (
 	github.com/soheilhy/cmux v0.1.5 // indirect
 	github.com/spf13/cobra v1.8.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
+	github.com/tilinna/clock v1.1.0 // indirect
 	github.com/tklauser/go-sysconf v0.3.13 // indirect
 	github.com/tklauser/numcpus v0.7.0 // indirect
 	github.com/valyala/fastjson v1.6.4 // indirect
@@ -257,3 +266,9 @@ replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/resourceto
 replace github.com/open-telemetry/opentelemetry-collector-contrib/exporter/prometheusremotewriteexporter => ../../prometheusremotewriteexporter
 
 replace github.com/open-telemetry/opentelemetry-collector-contrib/extension/ackextension => ../../../extension/ackextension
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/connector/spanmetricsconnector => ../../../connector/spanmetricsconnector
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl => ../../../pkg/ottl
+
+replace github.com/open-telemetry/opentelemetry-collector-contrib/connector/routingconnector => ../../../connector/routingconnector

testbed/Makefile

@@ -35,3 +35,7 @@ list-correctness-traces-tests:
 .PHONY: run-correctness-traces-tests
 run-correctness-traces-tests: $(GOJUNIT)
 	TESTS_DIR=correctnesstests/traces GOJUNIT=$(GOJUNIT) ./runtests.sh
+
+.PHONY: run-correctness-connectors-tests
+run-correctness-connectors-tests: $(GOJUNIT)
+	TESTS_DIR=correctnesstests/connectors GOJUNIT=$(GOJUNIT) ./runtests.sh

testbed/correctnesstests/connectors/correctness_test.go

@@ -0,0 +1,97 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package connectors
+
+import (
+	"log"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/testbed/correctnesstests"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/testbed/testbed"
+)
+
+var correctnessResults testbed.TestResultsSummary = &testbed.CorrectnessResults{}
+
+func TestMain(m *testing.M) {
+	testbed.DoTestMain(m, correctnessResults)
+}
+
+func TestGoldenData(t *testing.T) {
+	processors := map[string]string{
+		"batch": `
+  batch:
+    send_batch_size: 1024
+`,
+	}
+	sampleTest := correctnesstests.PipelineDef{
+		TestName:  "test routing",
+		Receiver:  "otlp",
+		Exporter:  "otlp",
+		Connector: "routing",
+	}
+
+	sampleTest.DataSender = correctnesstests.ConstructTraceSender(t, sampleTest.Receiver)
+	sampleTest.DataReceiver = correctnesstests.ConstructReceiver(t, sampleTest.Exporter)
+	sampleTest.DataConnector = correctnesstests.ConstructConnector(t, sampleTest.Connector, "traces")
+	t.Run(sampleTest.TestName, func(t *testing.T) {
+		testWithGoldenDataset(t, sampleTest.DataSender, sampleTest.DataReceiver, sampleTest.ResourceSpec, sampleTest.DataConnector, processors)
+	})
+
+}
+
+func testWithGoldenDataset(
+	t *testing.T,
+	sender testbed.DataSender,
+	receiver testbed.DataReceiver,
+	resourceSpec testbed.ResourceSpec,
+	connector testbed.DataConnector,
+	processors map[string]string,
+) {
+	dataProvider := testbed.NewGoldenDataProvider(
+		"../../../internal/coreinternal/goldendataset/testdata/generated_pict_pairs_traces.txt",
+		"../../../internal/coreinternal/goldendataset/testdata/generated_pict_pairs_spans.txt",
+		"")
+	factories, err := testbed.Components()
+	require.NoError(t, err, "default components resulted in: %v", err)
+	runner := testbed.NewInProcessCollector(factories)
+	validator := testbed.NewCorrectTestValidator(sender.ProtocolName(), receiver.ProtocolName(), dataProvider)
+	config := correctnesstests.CreateConfigYaml(t, sender, receiver, connector, processors)
+	log.Println(config)
+	configCleanup, cfgErr := runner.PrepareConfig(config)
+	require.NoError(t, cfgErr, "collector configuration resulted in: %v", cfgErr)
+	defer configCleanup()
+	tc := testbed.NewTestCase(
+		t,
+		dataProvider,
+		sender,
+		receiver,
+		runner,
+		validator,
+		correctnessResults,
+		testbed.WithResourceLimits(resourceSpec),
+	)
+	defer tc.Stop()
+
+	tc.EnableRecording()
+	tc.StartBackend()
+	tc.StartAgent()
+
+	tc.StartLoad(testbed.LoadOptions{
+		DataItemsPerSecond: 1024,
+		ItemsPerBatch:      1,
+	})
+
+	tc.Sleep(2 * time.Second)
+
+	tc.StopLoad()
+
+	tc.WaitForN(func() bool { return tc.LoadGenerator.DataItemsSent() == tc.MockBackend.DataItemsReceived() },
+		3*time.Second, "all data items received")
+
+	tc.StopAgent()
+
+}

testbed/correctnesstests/metrics/correctness_test_case.go

@@ -34,7 +34,7 @@ func newCorrectnessTestCase(
 
 func (tc *correctnessTestCase) startCollector() {
 	tc.collector = testbed.NewInProcessCollector(componentFactories(tc.t))
-	_, err := tc.collector.PrepareConfig(correctnesstests.CreateConfigYaml(tc.t, tc.sender, tc.receiver, nil, "metrics"))
+	_, err := tc.collector.PrepareConfig(correctnesstests.CreateConfigYaml(tc.t, tc.sender, tc.receiver, nil, nil))
 	require.NoError(tc.t, err)
 	rd, err := newResultsDir(tc.t.Name())
 	require.NoError(tc.t, err)

testbed/correctnesstests/traces/correctness_test.go

@@ -56,7 +56,7 @@ func testWithTracingGoldenDataset(
 	require.NoError(t, err, "default components resulted in: %v", err)
 	runner := testbed.NewInProcessCollector(factories)
 	validator := testbed.NewCorrectTestValidator(sender.ProtocolName(), receiver.ProtocolName(), dataProvider)
-	config := correctnesstests.CreateConfigYaml(t, sender, receiver, processors, "traces")
+	config := correctnesstests.CreateConfigYaml(t, sender, receiver, nil, processors)
 	log.Println(config)
 	configCleanup, cfgErr := runner.PrepareConfig(config)
 	require.NoError(t, cfgErr, "collector configuration resulted in: %v", cfgErr)
@@ -124,7 +124,7 @@ func TestSporadicGoldenDataset(t *testing.T) {
     sending_queue:
       enabled: false
 `)
-		_, err = runner.PrepareConfig(correctnesstests.CreateConfigYaml(t, sender, receiver, nil, "traces"))
+		_, err = runner.PrepareConfig(correctnesstests.CreateConfigYaml(t, sender, receiver, nil, nil))
 		require.NoError(t, err, "collector configuration resulted in: %v", err)
 		validator := testbed.NewCorrectTestValidator(sender.ProtocolName(), receiver.ProtocolName(), dataProvider)
 		tc := testbed.NewTestCase(