refactor!: sidecars to inetrface

.github/workflows/knuu_testing.yml

@@ -19,7 +19,7 @@ jobs:
             timeout: 15m
           - pkg: "./e2e/system"
             timeout: 15m
-          - pkg: "./e2e/bittwister"
+          - pkg: "./e2e/netshaper"
             timeout: 60m
     runs-on: ubuntu-latest
 

e2e/basic/observability_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/observability"
+)
+
+const (
+	prometheusPort   = observability.DefaultOtelMetricsPort
+	prometheusImage  = "prom/prometheus:latest"
+	prometheusConfig = "/etc/prometheus/prometheus.yml"
+	prometheusArgs   = "--config.file=/etc/prometheus/prometheus.yml"
+
+	targetImage = "curlimages/curl:latest"
+	otlpPort    = observability.DefaultOtelOtlpPort
+)
+
+// TestObservabilityCollector is a test function that verifies the functionality of the otel collector setup
+func TestObservabilityCollector(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 observabilitySidecar collector
+	observabilitySidecar := observability.New()
+
+	require.NoError(t, observabilitySidecar.SetOtelEndpoint(4318))
+
+	err = observabilitySidecar.SetPrometheusEndpoint(otlpPort, fmt.Sprintf("knuu-%s", knuu.Scope()), "10s")
+	require.NoError(t, err)
+
+	require.NoError(t, observabilitySidecar.SetJaegerEndpoint(14250, 6831, 14268))
+
+	require.NoError(t, observabilitySidecar.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(), observabilitySidecar))
+
+	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,13 +10,14 @@ import (
 	"time"
 
 	"github.com/celestiaorg/knuu/pkg/knuu"
+	"github.com/celestiaorg/knuu/pkg/sidecars/netshaper"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
 // TestReverseProxy is a test function that verifies the functionality of a reverse proxy setup.
-// It mainly tests the ability to reach to a service running in a sidecar like BitTwister.
+// It mainly tests the ability to reach to a service running in a sidecar like netshaper (BitTwister).
 // It calls an endpoint of the service and checks if the response is as expected.
 func TestReverseProxy(t *testing.T) {
 	t.Parallel()
@@ -33,6 +34,9 @@ func TestReverseProxy(t *testing.T) {
 
 	require.NoError(t, main.Commit(), "Error committing instance")
 
+	btSidecar := netshaper.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 → e2e/netshaper/netshaper_test.go

@@ -1,4 +1,4 @@
-package bittwister
+package netshaper
 
 import (
 	"context"
@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/celestiaorg/knuu/pkg/instance"
+	"github.com/celestiaorg/knuu/pkg/sidecars/netshaper"
 )
 
 const (
@@ -18,7 +19,7 @@ const (
 	gopingImage        = "ghcr.io/celestiaorg/goping:4803195"
 )
 
-func (s *Suite) TestBittwisterBandwidth() {
+func (s *Suite) TestNetShaperBandwidth() {
 	s.T().Parallel()
 	// Setup
 
@@ -43,8 +44,11 @@ func (s *Suite) TestBittwisterBandwidth() {
 	iperfClient, err := iperfMother.CloneWithName("iperf-client")
 	s.Require().NoError(err)
 
+	btSidecar := netshaper.New()
+	s.Require().NoError(iperfServer.AddSidecar(ctx, btSidecar))
+
 	s.T().Cleanup(func() {
-		s.T().Log("Tearing down TestBittwisterBandwidth test...")
+		s.T().Log("Tearing down TestNetShaperBandwidth test...")
 		err := instance.BatchDestroy(ctx, iperfServer, iperfClient)
 		if err != nil {
 			s.T().Logf("error destroying instances: %v", err)
@@ -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()
@@ -121,7 +124,7 @@ func (s *Suite) TestBittwisterBandwidth() {
 	}
 }
 
-func (s *Suite) TestBittwisterPacketloss() {
+func (s *Suite) TestNetShaperPacketloss() {
 	s.T().Parallel()
 	// Setup
 
@@ -147,11 +150,14 @@ func (s *Suite) TestBittwisterPacketloss() {
 	target, err := mother.CloneWithName("target")
 	s.Require().NoError(err)
 
+	btSidecar := netshaper.New()
+	s.Require().NoError(target.AddSidecar(ctx, btSidecar))
+
 	executor, err := mother.CloneWithName("executor")
 	s.Require().NoError(err)
 
 	s.T().Cleanup(func() {
-		s.T().Log("Tearing down TestBittwisterPacketloss test...")
+		s.T().Log("Tearing down TestNetShaperPacketloss test...")
 		err := instance.BatchDestroy(ctx, executor, target)
 		if err != nil {
 			s.T().Logf("error destroying instances: %v", 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()
@@ -224,7 +227,7 @@ func (s *Suite) TestBittwisterPacketloss() {
 	}
 }
 
-func (s *Suite) TestBittwisterLatency() {
+func (s *Suite) TestNetShaperLatency() {
 	s.T().Parallel()
 	// Setup
 
@@ -250,11 +253,14 @@ func (s *Suite) TestBittwisterLatency() {
 	target, err := mother.CloneWithName("target")
 	s.Require().NoError(err)
 
+	btSidecar := netshaper.New()
+	s.Require().NoError(target.AddSidecar(ctx, btSidecar))
+
 	executor, err := mother.CloneWithName("executor")
 	s.Require().NoError(err)
 
 	s.T().Cleanup(func() {
-		s.T().Log("Tearing down TestBittwisterLatency test...")
+		s.T().Log("Tearing down TestNetShaperLatency test...")
 		err := instance.BatchDestroy(ctx, executor, target)
 		if err != nil {
 			s.T().Logf("error destroying instances: %v", 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.")
@@ -332,7 +337,7 @@ func (s *Suite) TestBittwisterLatency() {
 		})
 	}
 }
-func (s *Suite) TestBittwisterJitter() {
+func (s *Suite) TestNetShaperJitter() {
 	s.T().Parallel()
 	// Setup
 
@@ -358,11 +363,14 @@ func (s *Suite) TestBittwisterJitter() {
 	target, err := mother.CloneWithName("target")
 	s.Require().NoError(err)
 
+	btSidecar := netshaper.New()
+	s.Require().NoError(target.AddSidecar(ctx, btSidecar))
+
 	executor, err := mother.CloneWithName("executor")
 	s.Require().NoError(err)
 
 	s.T().Cleanup(func() {
-		s.T().Log("Tearing down TestBittwisterJitter test...")
+		s.T().Log("Tearing down TestNetShaperJitter test...")
 		err := instance.BatchDestroy(ctx, executor, target)
 		if err != nil {
 			s.T().Logf("error destroying instances: %v", 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/bittwister/suite_setup_test.go → e2e/netshaper/suite_setup_test.go

@@ -1,4 +1,4 @@
-package bittwister
+package netshaper
 
 import (
 	"context"