refactor!: sidecars to inetrface

e2e/basic/obsy_test.go

@@ -0,0 +1,120 @@
+package basic
+
+import (
+	"context"
+	"fmt"
+	"io/ioutil"
+	"net/http"
+	"os"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/celestiaorg/knuu/pkg/knuu"
+	"github.com/celestiaorg/knuu/pkg/sidecars/obsy"
+)
+
+const (
+	prometheusPort   = obsy.DefaultOtelMetricsPort
+	prometheusImage  = "prom/prometheus:latest"
+	prometheusConfig = "/etc/prometheus/prometheus.yml"
+	prometheusArgs   = "--config.file=/etc/prometheus/prometheus.yml"
+
+	targetImage = "curlimages/curl:latest"
+	otlpPort    = obsy.DefaultOtelOtlpPort
+)
+
+// TestObsyCollector is a test function that verifies the functionality of the otel collector setup
+func TestObsyCollector(t *testing.T) {
+	t.Parallel()
+
+	// Setup Prometheus
+	prometheus, err := knuu.NewInstance("prometheus")
+	require.NoError(t, err)
+
+	require.NoError(t, prometheus.SetImage(prometheusImage))
+	require.NoError(t, prometheus.AddPortTCP(prometheusPort))
+
+	// enable proxy for this port
+	err, prometheusEndpoint := prometheus.AddHost(prometheusPort)
+	require.NoError(t, err)
+
+	require.NoError(t, prometheus.Commit())
+
+	// Add Prometheus config file
+	prometheusConfigContent := fmt.Sprintf(`
+global:
+  scrape_interval: '10s'
+scrape_configs:
+  - job_name: 'otel-collector'
+    static_configs:
+      - targets: ['otel-collector:%d']
+`, otlpPort)
+	require.NoError(t, prometheus.AddFileBytes([]byte(prometheusConfigContent), prometheusConfig, "0:0"))
+
+	require.NoError(t, prometheus.SetArgs(prometheusArgs))
+	require.NoError(t, prometheus.Start())
+
+	// Setup obsySidecar collector
+	obsySidecar := obsy.New()
+
+	require.NoError(t, obsySidecar.SetOtelEndpoint(4318))
+
+	err = obsySidecar.SetPrometheusEndpoint(otlpPort, fmt.Sprintf("knuu-%s", knuu.Scope()), "10s")
+	require.NoError(t, err)
+
+	require.NoError(t, obsySidecar.SetJaegerEndpoint(14250, 6831, 14268))
+
+	require.NoError(t, obsySidecar.SetOtlpExporter("prometheus:9090", "", ""))
+
+	// Create and start a target pod and configure it to use the obsySidecar to push metrics
+	target, err := knuu.NewInstance("target")
+	require.NoError(t, err, "Error creating target instance")
+
+	err = target.SetImage(targetImage)
+	require.NoError(t, err, "Error setting target image")
+
+	err = target.SetCommand("sh", "-c", "while true; do curl -X POST http://localhost:8888/v1/traces; sleep 5; done")
+	require.NoError(t, err, "Error setting target command")
+
+	require.NoError(t, target.AddSidecar(context.Background(), obsySidecar))
+
+	require.NoError(t, target.Commit(), "Error committing target instance")
+
+	require.NoError(t, target.Start(), "Error starting target instance")
+
+	t.Cleanup(func() {
+		if os.Getenv("KNUU_SKIP_CLEANUP") == "true" {
+			t.Log("Skipping cleanup")
+			return
+		}
+		err := knuu.BatchDestroy(prometheus, target)
+		if err != nil {
+			t.Log("Error destroying instances: ", err)
+		}
+	})
+
+	// Wait for the target pod to push data to the otel collector
+	time.Sleep(1 * time.Minute)
+
+	// Verify that data has been pushed to Prometheus
+
+	prometheusURL := fmt.Sprintf("%s/api/v1/query?query=up", prometheusEndpoint)
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	req, err := http.NewRequestWithContext(ctx, "GET", prometheusURL, nil)
+	require.NoError(t, err)
+
+	resp, err := http.DefaultClient.Do(req)
+	require.NoError(t, err)
+	require.Equal(t, 200, resp.StatusCode, "Prometheus API is not accessible")
+
+	defer resp.Body.Close()
+	body, err := ioutil.ReadAll(resp.Body)
+	require.NoError(t, err)
+	require.Contains(t, string(body), "otel-collector", "otel-collector data source not found in Prometheus")
+
+	t.Log("otel-collector data source is available in Prometheus")
+}

e2e/basic/reverse_proxy_test.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/celestiaorg/knuu/pkg/knuu"
+	"github.com/celestiaorg/knuu/pkg/sidecars/bittwister"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -33,6 +34,9 @@ func TestReverseProxy(t *testing.T) {
 
 	require.NoError(t, main.Commit(), "Error committing instance")
 
+	btSidecar := bittwister.New()
+	require.NoError(t, main.AddSidecar(context.Background(), btSidecar))
+
 	t.Cleanup(func() {
 		if os.Getenv("KNUU_SKIP_CLEANUP") == "true" {
 			t.Log("Skipping cleanup")
@@ -42,20 +46,19 @@ func TestReverseProxy(t *testing.T) {
 		require.NoError(t, main.Destroy(), "Error destroying instance")
 	})
 
-	require.NoError(t, main.EnableBitTwister(), "Error enabling BitTwister")
 	require.NoError(t, main.Start(), "Error starting main instance")
 
 	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Minute)
 	defer cancel()
 
-	require.NoError(t, main.BitTwister.WaitForStart(ctx), "Error waiting for BitTwister to start")
+	require.NoError(t, btSidecar.WaitForStart(ctx), "Error waiting for BitTwister to start")
 
 	// test if BitTwister running in a sidecar is accessible
-	err = main.SetBandwidthLimit(1000)
+	err = btSidecar.SetBandwidthLimit(1000)
 	assert.NoError(t, err, "Error setting bandwidth limit")
 
 	// Check if the BitTwister service is set
-	out, err := main.BitTwister.Client().AllServicesStatus()
+	out, err := btSidecar.AllServicesStatus()
 	assert.NoError(t, err, "Error getting all services status")
 	assert.GreaterOrEqual(t, len(out), 1, "No services found")
 	assert.NotEmpty(t, out[0].Name, "Service name is empty")

e2e/bittwister/bittwister_test.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/celestiaorg/knuu/pkg/instance"
+	"github.com/celestiaorg/knuu/pkg/sidecars/bittwister"
 )
 
 const (
@@ -43,6 +44,9 @@ func (s *Suite) TestBittwisterBandwidth() {
 	iperfClient, err := iperfMother.CloneWithName("iperf-client")
 	s.Require().NoError(err)
 
+	btSidecar := bittwister.New()
+	s.Require().NoError(iperfServer.AddSidecar(ctx, btSidecar))
+
 	s.T().Cleanup(func() {
 		s.T().Log("Tearing down TestBittwisterBandwidth test...")
 		err := instance.BatchDestroy(ctx, iperfServer, iperfClient)
@@ -53,9 +57,9 @@ func (s *Suite) TestBittwisterBandwidth() {
 
 	// Prepare iperf client & server
 
-	s.Require().NoError(iperfServer.EnableBitTwister())
 	s.Require().NoError(iperfServer.Start(ctx))
-	s.Require().NoError(iperfServer.BitTwister.WaitForStart(ctx))
+	s.Require().NoError(btSidecar.WaitForStart(ctx))
+
 	s.Require().NoError(iperfClient.Start(ctx))
 
 	iperfServerIP, err := iperfServer.GetIP(ctx)
@@ -86,8 +90,7 @@ func (s *Suite) TestBittwisterBandwidth() {
 		tc := tc
 		s.Run(tc.name, func() {
 			s.T().Logf("Max bandwidth: %v \t tolerance: %v%%", formatBandwidth(float64(tc.targetBandwidth)), tc.tolerancePercent)
-
-			s.Require().NoError(iperfServer.SetBandwidthLimit(tc.targetBandwidth))
+			s.Require().NoError(btSidecar.SetBandwidthLimit(tc.targetBandwidth))
 
 			s.T().Log("Starting bandwidth test. It takes a while.")
 			startTime := time.Now()
@@ -147,6 +150,9 @@ func (s *Suite) TestBittwisterPacketloss() {
 	target, err := mother.CloneWithName("target")
 	s.Require().NoError(err)
 
+	btSidecar := bittwister.New()
+	s.Require().NoError(target.AddSidecar(ctx, btSidecar))
+
 	executor, err := mother.CloneWithName("executor")
 	s.Require().NoError(err)
 
@@ -160,10 +166,9 @@ func (s *Suite) TestBittwisterPacketloss() {
 
 	// Prepare ping executor & target
 
-	s.Require().NoError(target.EnableBitTwister())
 	s.Require().NoError(target.Start(ctx))
+	s.Require().NoError(btSidecar.WaitForStart(ctx))
 
-	s.Require().NoError(target.BitTwister.WaitForStart(ctx))
 	s.Require().NoError(executor.Start(ctx))
 
 	// Perform the test
@@ -191,9 +196,7 @@ func (s *Suite) TestBittwisterPacketloss() {
 		tc := tc
 		s.Run(tc.name, func() {
 			s.T().Logf("Target packetloss: %v%% \t tolerance: %v%%", tc.targetPacketlossRate, tc.tolerancePercent)
-
-			err = target.SetPacketLoss(tc.targetPacketlossRate)
-			s.Require().NoError(err)
+			s.Require().NoError(btSidecar.SetPacketLoss(tc.targetPacketlossRate))
 
 			s.T().Log("Starting packetloss test. It takes a while.")
 			startTime := time.Now()
@@ -250,6 +253,9 @@ func (s *Suite) TestBittwisterLatency() {
 	target, err := mother.CloneWithName("target")
 	s.Require().NoError(err)
 
+	btSidecar := bittwister.New()
+	s.Require().NoError(target.AddSidecar(ctx, btSidecar))
+
 	executor, err := mother.CloneWithName("executor")
 	s.Require().NoError(err)
 
@@ -263,10 +269,9 @@ func (s *Suite) TestBittwisterLatency() {
 
 	// Prepare ping executor & target
 
-	s.Require().NoError(target.EnableBitTwister())
 	s.Require().NoError(target.Start(ctx))
+	s.Require().NoError(btSidecar.WaitForStart(ctx))
 
-	s.Require().NoError(target.BitTwister.WaitForStart(ctx))
 	s.Require().NoError(executor.Start(ctx))
 
 	// Perform the test
@@ -299,7 +304,7 @@ func (s *Suite) TestBittwisterLatency() {
 		s.Run(tc.name, func() {
 			s.T().Logf("Max latency: %v ms \t tolerance: %v%%", tc.targetLatency.Milliseconds(), tc.tolerancePercent)
 
-			err = target.SetLatencyAndJitter(tc.targetLatency.Milliseconds(), 0)
+			err = btSidecar.SetLatencyAndJitter(tc.targetLatency.Milliseconds(), 0)
 			s.Require().NoError(err)
 
 			s.T().Log("Starting latency test. It takes a while.")
@@ -358,6 +363,9 @@ func (s *Suite) TestBittwisterJitter() {
 	target, err := mother.CloneWithName("target")
 	s.Require().NoError(err)
 
+	btSidecar := bittwister.New()
+	s.Require().NoError(target.AddSidecar(ctx, btSidecar))
+
 	executor, err := mother.CloneWithName("executor")
 	s.Require().NoError(err)
 
@@ -371,9 +379,9 @@ func (s *Suite) TestBittwisterJitter() {
 
 	// Prepare ping executor & target
 
-	s.Require().NoError(target.EnableBitTwister())
 	s.Require().NoError(target.Start(ctx))
-	s.Require().NoError(target.BitTwister.WaitForStart(ctx))
+	s.Require().NoError(btSidecar.WaitForStart(ctx))
+
 	s.Require().NoError(executor.Start(ctx))
 
 	// Perform the test
@@ -400,7 +408,7 @@ func (s *Suite) TestBittwisterJitter() {
 		s.Run(tc.name, func() {
 			s.T().Logf("Max jitter: %v", tc.maxTargetJitter.Milliseconds())
 
-			err = target.SetLatencyAndJitter(0, tc.maxTargetJitter.Milliseconds())
+			err = btSidecar.SetLatencyAndJitter(0, tc.maxTargetJitter.Milliseconds())
 			s.Require().NoError(err)
 
 			s.T().Log("Starting jitter test. It takes a while.")

e2e/tshark/tshark_test.go

@@ -13,10 +13,10 @@ import (
 	"github.com/stretchr/testify/require"
 	"k8s.io/apimachinery/pkg/api/resource"
 
-	"github.com/celestiaorg/knuu/pkg/instance"
 	"github.com/celestiaorg/knuu/pkg/k8s"
 	"github.com/celestiaorg/knuu/pkg/knuu"
 	"github.com/celestiaorg/knuu/pkg/minio"
+	"github.com/celestiaorg/knuu/pkg/sidecars/tshark"
 )
 
 const (
@@ -43,7 +43,6 @@ func TestTshark(t *testing.T) {
 
 	kn, err := knuu.New(ctx, knuu.Options{MinioClient: minioClient, K8sClient: k8sClient})
 	require.NoError(t, err, "error creating knuu")
-
 	defer func() {
 		if err := kn.CleanUp(ctx); err != nil {
 			t.Logf("error cleaning up knuu: %v", err)
@@ -56,49 +55,39 @@ func TestTshark(t *testing.T) {
 	target, err := kn.NewInstance("busybox")
 	require.NoError(t, err, "error creating instance")
 
-	err = target.SetImage(ctx, "busybox")
-	require.NoError(t, err, "error setting image")
-
-	err = target.SetCommand("sleep", "infinity")
-	require.NoError(t, err, "error setting command")
+	require.NoError(t, target.SetImage(ctx, "busybox"))
+	require.NoError(t, target.SetCommand("sleep", "infinity"))
 
 	t.Log("getting minio configs")
 	minioConf, err := kn.MinioClient.GetConfigs(ctx)
 	require.NoError(t, err, "error getting S3 (minio) configs")
 
+	keyPrefix := "tshark/" + scope
+
+	tsc := &tshark.Tshark{
+		VolumeSize:     tsharkVolumeSize,
+		S3AccessKey:    minioConf.AccessKeyID,
+		S3SecretKey:    minioConf.SecretAccessKey,
+		S3Region:       s3Location,
+		S3Bucket:       s3BucketName,
+		CreateBucket:   true, // Since we fire up a fresh minio server, we need to create the bucket
+		S3KeyPrefix:    keyPrefix,
+		S3Endpoint:     minioConf.Endpoint,
+		UploadInterval: 1 * time.Second, // for sake of the test we keep this short
+	}
+
+	require.NoError(t, target.AddSidecar(ctx, tsc))
 	var (
-		filename  = target.K8sName() + instance.TsharkCaptureFileExtension + ".tar.gz" // compressed file extension
-		keyPrefix = "tshark/" + scope
-		fileKey   = filepath.Join(keyPrefix, filename)
+		filename = tsc.Instance().K8sName() + tshark.TsharkCaptureFileExtension
+		fileKey  = filepath.Join(keyPrefix, filename)
 	)
 
-	err = target.EnableTsharkCollector(
-		instance.TsharkCollectorConfig{
-			VolumeSize:     tsharkVolumeSize,
-			S3AccessKey:    minioConf.AccessKeyID,
-			S3SecretKey:    minioConf.SecretAccessKey,
-			S3Region:       s3Location,
-			S3Bucket:       s3BucketName,
-			CreateBucket:   true, // Since we fire up a fresh minio server, we need to create the bucket
-			S3KeyPrefix:    keyPrefix,
-			S3Endpoint:     minioConf.Endpoint,
-			UploadInterval: 1 * time.Second, // for sake of the test we keep this short
-			CompressFiles:  true,
-		},
-	)
-	require.NoError(t, err, "error enabling tshark collector")
-
-	err = target.Commit()
-	require.NoError(t, err, "error committing instance")
+	require.NoError(t, target.Commit())
 
 	// Test logic
 
 	t.Log("starting target instance")
-	err = target.Start(ctx)
-	require.NoError(t, err, "error starting instance")
-
-	err = target.WaitInstanceIsRunning(ctx)
-	require.NoError(t, err, "error waiting for instance to be running")
+	require.NoError(t, target.Start(ctx))
 
 	// Perform a ping to do generate network traffic to allow tshark to capture it
 	_, err = target.ExecuteCommand(ctx, "ping", "-c", "4", "google.com")