feat: add support to upload tshark to s3

e2e/tshark/tshark_test.go

@@ -0,0 +1,113 @@
+package basic
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/celestiaorg/knuu/pkg/instance"
+	"github.com/celestiaorg/knuu/pkg/knuu"
+)
+
+const (
+	s3BucketName = "tshark-test-bucket"
+	s3Location   = "eu-east-1"
+)
+
+func TestTshark(t *testing.T) {
+	t.Parallel()
+	// Setup
+
+	ctx := context.Background()
+
+	kn, err := knuu.New(ctx, knuu.Options{})
+	require.NoError(t, err, "error creating knuu")
+
+	defer func() {
+		if err := kn.CleanUp(ctx); err != nil {
+			t.Logf("error cleaning up knuu: %v", err)
+		}
+	}()
+
+	scope := kn.Scope()
+	t.Logf("Test scope: %s", scope)
+
+	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")
+
+	t.Log("deploying minio as s3 backend")
+	err = kn.MinioClient.DeployMinio(ctx)
+	require.NoError(t, err, "error deploying minio")
+
+	t.Log("getting minio configs")
+	minioConf, err := kn.MinioClient.GetConfigs(ctx)
+	require.NoError(t, err, "error getting S3 (minio) configs")
+
+	var (
+		filename  = target.K8sName() + instance.TsharkCaptureFileExtension
+		keyPrefix = "tshark/" + scope
+		fileKey   = filepath.Join(keyPrefix, filename)
+	)
+
+	err = target.EnableTsharkCollector(
+		instance.TsharkCollectorConfig{
+			VolumeSize:     "10Gi",
+			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, err, "error enabling tshark collector")
+
+	err = target.Commit()
+	require.NoError(t, err, "error committing instance")
+
+	t.Cleanup(func() {
+		if err := kn.CleanUp(ctx); err != nil {
+			t.Logf("error cleaning up knuu: %v", err)
+		}
+	})
+
+	// 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")
+
+	// Perform a ping to do generate network traffic to allow tshark to capture it
+	_, err = target.ExecuteCommand(ctx, "ping", "-c", "4", "google.com")
+	require.NoError(t, err, "error executing command")
+
+	url, err := kn.MinioClient.GetMinioURL(ctx, fileKey, s3BucketName)
+	require.NoError(t, err, "error getting minio url")
+
+	resp, err := http.Get(url)
+	require.NoError(t, err, "error downloading from minio URL")
+	defer resp.Body.Close()
+
+	assert.Equal(t, http.StatusOK, resp.StatusCode, "URL does not exist or is not accessible")
+
+	bodyBytes, err := io.ReadAll(resp.Body)
+	require.NoError(t, err, "error reading response body")
+	assert.NotEmpty(t, bodyBytes, "downloaded log file is empty")
+}

pkg/instance/errors.go

@@ -198,4 +198,13 @@ var (
 	ErrAddingToProxy                             = errors.New("AddingToProxy", "error adding '%s' to traefik proxy for service '%s'")
 	ErrGettingProxyURL                           = errors.New("GettingProxyURL", "error getting proxy URL for service '%s'")
 	ErrProxyNotInitialized                       = errors.New("ProxyNotInitialized", "proxy not initialized")
+	ErrTsharkCollectorAlreadyEnabled             = errors.New("TsharkCollectorAlreadyEnabled", "tshark collector already enabled for instance '%s'")
+	ErrCreatingTsharkCollectorInstance           = errors.New("CreatingTsharkCollectorInstance", "error creating tshark collector instance")
+	ErrAddingTsharkCollectorSidecar              = errors.New("AddingTsharkCollectorSidecar", "error adding tshark collector sidecar for instance '%s'")
+	ErrTsharkCollectorConfigNotSet               = errors.New("TsharkCollectorConfigNotSet", "tshark collector config not set for instance '%s'. volumeSize: %s, s3AccessKey: %s, s3SecretKey: %s, s3Region: %s, s3BucketName: %s, s3KeyPrefix: %s")
+	ErrTsharkCollectorInvalidVolumeSize          = errors.New("TsharkCollectorInvalidVolumeSize", "invalid volume size format for tshark collector: %s")
+	ErrTsharkCollectorInvalidS3AccessKey         = errors.New("TsharkCollectorInvalidS3AccessKey", "invalid S3 access key format for tshark collector: %s")
+	ErrTsharkCollectorInvalidS3SecretKey         = errors.New("TsharkCollectorInvalidS3SecretKey", "invalid S3 secret key format for tshark collector: %s")
+	ErrTsharkCollectorS3RegionOrBucketEmpty      = errors.New("TsharkCollectorS3RegionOrBucketEmpty", "S3 region or bucket cannot be empty for tshark collector: %s, %s")
+	ErrRegexpCompile                             = errors.New("RegexpCompile", "error compiling regexp for %s")
 )

pkg/instance/helper.go

@@ -361,34 +361,35 @@ func (i *Instance) cloneWithSuffix(suffix string) *Instance {
 	clonedBitTwister.SetClient(nil) // reset client to avoid reusing the same client
 
 	return &Instance{
-		name:                 i.name + suffix,
-		k8sName:              i.k8sName + suffix,
-		imageName:            i.imageName,
-		state:                i.state,
-		instanceType:         i.instanceType,
-		kubernetesService:    i.kubernetesService,
-		builderFactory:       i.builderFactory,
-		kubernetesReplicaSet: i.kubernetesReplicaSet,
-		portsTCP:             i.portsTCP,
-		portsUDP:             i.portsUDP,
-		command:              i.command,
-		args:                 i.args,
-		env:                  i.env,
-		volumes:              i.volumes,
-		memoryRequest:        i.memoryRequest,
-		memoryLimit:          i.memoryLimit,
-		cpuRequest:           i.cpuRequest,
-		policyRules:          i.policyRules,
-		livenessProbe:        i.livenessProbe,
-		readinessProbe:       i.readinessProbe,
-		startupProbe:         i.startupProbe,
-		isSidecar:            false,
-		parentInstance:       nil,
-		sidecars:             clonedSidecars,
-		obsyConfig:           i.obsyConfig,
-		securityContext:      &clonedSecurityContext,
-		BitTwister:           &clonedBitTwister,
-		SystemDependencies:   i.SystemDependencies,
+		name:                  i.name + suffix,
+		k8sName:               i.k8sName + suffix,
+		imageName:             i.imageName,
+		state:                 i.state,
+		instanceType:          i.instanceType,
+		kubernetesService:     i.kubernetesService,
+		builderFactory:        i.builderFactory,
+		kubernetesReplicaSet:  i.kubernetesReplicaSet,
+		portsTCP:              i.portsTCP,
+		portsUDP:              i.portsUDP,
+		command:               i.command,
+		args:                  i.args,
+		env:                   i.env,
+		volumes:               i.volumes,
+		memoryRequest:         i.memoryRequest,
+		memoryLimit:           i.memoryLimit,
+		cpuRequest:            i.cpuRequest,
+		policyRules:           i.policyRules,
+		livenessProbe:         i.livenessProbe,
+		readinessProbe:        i.readinessProbe,
+		startupProbe:          i.startupProbe,
+		isSidecar:             false,
+		parentInstance:        nil,
+		sidecars:              clonedSidecars,
+		obsyConfig:            i.obsyConfig,
+		tsharkCollectorConfig: i.tsharkCollectorConfig,
+		securityContext:       &clonedSecurityContext,
+		BitTwister:            &clonedBitTwister,
+		SystemDependencies:    i.SystemDependencies,
 	}
 }
 
@@ -594,6 +595,17 @@ func (i *Instance) addOtelCollectorSidecar(ctx context.Context) error {
 	return nil
 }
 
+func (i *Instance) addTsharkCollectorSidecar(ctx context.Context) error {
+	tsharkSidecar, err := i.createTsharkCollectorInstance(ctx)
+	if err != nil {
+		return ErrCreatingTsharkCollectorInstance.WithParams(i.k8sName).Wrap(err)
+	}
+	if err := i.AddSidecar(tsharkSidecar); err != nil {
+		return ErrAddingTsharkCollectorSidecar.WithParams(i.k8sName).Wrap(err)
+	}
+	return nil
+}
+
 func (i *Instance) createBitTwisterInstance(ctx context.Context) (*Instance, error) {
 	bt, err := New("bit-twister", i.SystemDependencies)
 	if err != nil {