refactoring, need to add telemetrygen to k8s via code

.github/workflows/otel-integration-e2e-test.yml

@@ -34,9 +34,8 @@ jobs:
           -f ./values.yaml \
           -f ./e2e-test/testdata/values-e2e-test.yaml
           kubectl wait --all --for=condition=ready --timeout=300s pod -l component=agent-collector
-          kubectl apply -f ./e2e-test/testdata/telemetrygen-deployment.yaml
+          kubectl apply -f ./e2e-test/k8s/telemetrygen-deployment.yaml
           kubectl wait --all --for=condition=ready --timeout=300s pod -l app=telemetrygen
-          kubectl get pods
       - name: Run E2E test
         run: |
           cd ./otel-integration/k8s-helm/e2e-test/

otel-integration/k8s-helm/e2e-test/testdata/telemetrygen-deployment.yaml → otel-integration/k8s-helm/e2e-test/k8s/telemetrygen-deployment.yaml

@@ -1,4 +1,4 @@
-apiVersion: apps/v1
+apiVersion: app/v1
 kind: Deployment
 metadata:
   labels:

otel-integration/k8s-helm/e2e-test/main_test.go

@@ -4,62 +4,37 @@
 package e2e
 
 import (
-	"context"
 	"strings"
 	"testing"
-	"time"
 
 	"github.com/davecgh/go-spew/spew"
-	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"go.opentelemetry.io/collector/component/componenttest"
 	"go.opentelemetry.io/collector/consumer/consumertest"
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	"go.opentelemetry.io/collector/pdata/ptrace"
-	"go.opentelemetry.io/collector/receiver/otlpreceiver"
-	"go.opentelemetry.io/collector/receiver/receivertest"
 )
 
 func TestE2E_Agent(t *testing.T) {
 	metricsConsumer := new(consumertest.MetricsSink)
 	tracesConsumer := new(consumertest.TracesSink)
-	shutdownSink := startUpSink(t, metricsConsumer, tracesConsumer)
+	// shutdownSink := startUpSink(t, metricsConsumer, tracesConsumer)
+
+	shutdownSink := StartUpSinks(t, ReceiverSinks{
+		MetricsConsumer: metricsConsumer,
+		TracesConsumer:  tracesConsumer,
+	})
+
 	defer shutdownSink()
 
-	waitTime := 3 * time.Minute
-	waitForData(waitTime)
+	WaitForMetrics(t, 10, metricsConsumer)
+	WaitForTraces(t, 10, tracesConsumer)
 
 	checkResourceMetrics(t, metricsConsumer.AllMetrics())
 	checkGeneratedTraces(t, tracesConsumer.AllTraces())
 
 }
 
-func startUpSink(t *testing.T, mc *consumertest.MetricsSink, tc *consumertest.TracesSink) func() {
-	f := otlpreceiver.NewFactory()
-	cfg := f.CreateDefaultConfig().(*otlpreceiver.Config)
-
-	metricsRcvr, err := f.CreateMetricsReceiver(context.Background(), receivertest.NewNopCreateSettings(), cfg, mc)
-	require.NoError(t, metricsRcvr.Start(context.Background(), componenttest.NewNopHost()))
-	require.NoError(t, err, "failed creating metrics receiver")
-
-	spansRcvr, err := f.CreateTracesReceiver(context.Background(), receivertest.NewNopCreateSettings(), cfg, tc)
-	require.NoError(t, spansRcvr.Start(context.Background(), componenttest.NewNopHost()))
-	require.NoError(t, err, "failed creating traces receiver")
-	return func() {
-		assert.NoError(t, metricsRcvr.Shutdown(context.Background()))
-	}
-}
-
-func waitForData(wait time.Duration) {
-	deadline := time.Now().Add(wait)
-	for {
-		if time.Now().After(deadline) {
-			break
-		}
-	}
-}
-
 func checkResourceMetrics(t *testing.T, actual []pmetric.Metrics) error {
 	if len(actual) == 0 {
 		t.Fatal("No resource metrics received")

otel-integration/k8s-helm/e2e-test/sinks.go

@@ -0,0 +1,69 @@
+package e2e
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/component/componenttest"
+	"go.opentelemetry.io/collector/consumer/consumertest"
+	"go.opentelemetry.io/collector/receiver/otlpreceiver"
+	"go.opentelemetry.io/collector/receiver/receivertest"
+)
+
+type ReceiverSinks struct {
+	MetricsConsumer *consumertest.MetricsSink
+	TracesConsumer  *consumertest.TracesSink
+}
+
+func StartUpSinks(t *testing.T, sinks ReceiverSinks) func() {
+	shutDownFuncs := []func(){}
+
+	if sinks.MetricsConsumer != nil {
+		fMetric := otlpreceiver.NewFactory()
+		cfg := fMetric.CreateDefaultConfig().(*otlpreceiver.Config)
+		metricsRcvr, err := fMetric.CreateMetricsReceiver(context.Background(), receivertest.NewNopCreateSettings(), cfg, sinks.MetricsConsumer)
+		require.NoError(t, metricsRcvr.Start(context.Background(), componenttest.NewNopHost()))
+		require.NoError(t, err, "failed creating metrics receiver")
+		shutDownFuncs = append(shutDownFuncs, func() {
+			assert.NoError(t, metricsRcvr.Shutdown(context.Background()))
+		})
+	}
+
+	if sinks.TracesConsumer != nil {
+		fTrace := otlpreceiver.NewFactory()
+		cfg := fTrace.CreateDefaultConfig().(*otlpreceiver.Config)
+		traceRcvr, err := fTrace.CreateTracesReceiver(context.Background(), receivertest.NewNopCreateSettings(), cfg, sinks.TracesConsumer)
+		require.NoError(t, traceRcvr.Start(context.Background(), componenttest.NewNopHost()))
+		require.NoError(t, err, "failed creating traces receiver")
+		shutDownFuncs = append(shutDownFuncs, func() {
+			assert.NoError(t, traceRcvr.Shutdown(context.Background()))
+		})
+	}
+
+	return func() {
+		for _, f := range shutDownFuncs {
+			f()
+		}
+	}
+}
+
+func WaitForMetrics(t *testing.T, entriesNum int, mc *consumertest.MetricsSink) {
+	timeoutMinutes := 5
+	require.Eventually(t, func() bool {
+		return len(mc.AllMetrics()) >= entriesNum
+	}, time.Duration(timeoutMinutes)*time.Minute, time.Second,
+		"failed to receive %d entries,  received %d metrics in %d minutes",
+		entriesNum, len(mc.AllMetrics()), timeoutMinutes)
+}
+
+func WaitForTraces(t *testing.T, entriesNum int, tc *consumertest.TracesSink) {
+	timeoutMinutes := 5
+	require.Eventually(t, func() bool {
+		return len(tc.AllTraces()) >= entriesNum
+	}, time.Duration(timeoutMinutes)*time.Minute, time.Second,
+		"failed to receive %d entries,  received %d traces in %d minutes",
+		entriesNum, len(tc.AllTraces()), timeoutMinutes)
+}

otel-integration/k8s-helm/e2e-test/telemetrygen.go

@@ -0,0 +1,21 @@
+package e2e
+
+// import (
+// 	"testing"
+// 	"strings"
+// 	"time"
+// }
+
+// type TelemetrygenObjInfo struct {
+// 	Namespace         string
+// 	PodLabelSelectors map[string]any
+// 	DataType          string
+// 	Workload          string
+// }
+
+// type TelemetrygenCreateOpts struct {
+// 	TestID       string
+// 	ManifestsDir string
+// 	OtlpEndpoint string
+// 	DataTypes    []string
+// }