test: soak test and migrate ilogtail to loongcollector (#1971)

* test: add soak test engine

* rename logtail to loongcollector

* fix

* soak test

* fix

* fix

test/engine/cleanup/chaos.go

@@ -33,7 +33,7 @@ func DestoryAllChaos(ctx context.Context) (context.Context, error) {
 	switch setup.Env.GetType() {
 	case "host":
 		command := "/opt/chaosblade/blade status --type create --status Success"
-		response, err := setup.Env.ExecOnLogtail(command)
+		response, err := setup.Env.ExecOnLoongCollector(command)
 		if err != nil {
 			return ctx, err
 		}
@@ -43,7 +43,7 @@ func DestoryAllChaos(ctx context.Context) (context.Context, error) {
 		}
 		for _, result := range status.Result {
 			command = "/opt/chaosblade/blade destroy " + result["Uid"]
-			if _, err := setup.Env.ExecOnLogtail(command); err != nil {
+			if _, err := setup.Env.ExecOnLoongCollector(command); err != nil {
 				fmt.Println("Destroy chaos failed: ", err)
 			}
 		}

test/engine/control/agent.go

@@ -23,7 +23,7 @@ import (
 )
 
 func RestartAgent(ctx context.Context) (context.Context, error) {
-	if _, err := setup.Env.ExecOnLogtail("/etc/init.d/loongcollectord restart"); err != nil {
+	if _, err := setup.Env.ExecOnLoongCollector("/etc/init.d/loongcollectord restart"); err != nil {
 		return ctx, err
 	}
 	return setup.SetAgentPID(ctx)
@@ -34,14 +34,14 @@ func ForceRestartAgent(ctx context.Context) (context.Context, error) {
 	if currentPID != nil {
 		currentPIDs := strings.Split(strings.TrimSpace(currentPID.(string)), "\n")
 		for _, pid := range currentPIDs {
-			if _, err := setup.Env.ExecOnLogtail("kill -9 " + pid); err != nil {
+			if _, err := setup.Env.ExecOnLoongCollector("kill -9 " + pid); err != nil {
 				fmt.Println("Force kill agent pid failed: ", err)
 			}
 		}
 	} else {
 		fmt.Println("No agent pid found, skip force restart")
 	}
-	if _, err := setup.Env.ExecOnLogtail("/etc/init.d/loongcollectord restart"); err != nil {
+	if _, err := setup.Env.ExecOnLoongCollector("/etc/init.d/loongcollectord restart"); err != nil {
 		return ctx, err
 	}
 	return setup.SetAgentPID(ctx)

test/engine/control/config.go

@@ -55,7 +55,7 @@ func AddLocalConfig(ctx context.Context, configName, c string) (context.Context,
 	} else {
 		command := fmt.Sprintf(`cd %s && cat << 'EOF' > %s.yaml
 %s`, config.TestConfig.LocalConfigDir, configName, c)
-		if _, err := setup.Env.ExecOnLogtail(command); err != nil {
+		if _, err := setup.Env.ExecOnLoongCollector(command); err != nil {
 			return ctx, err
 		}
 		time.Sleep(5 * time.Second)
@@ -65,7 +65,7 @@ func AddLocalConfig(ctx context.Context, configName, c string) (context.Context,
 
 func RemoveAllLocalConfig(ctx context.Context) (context.Context, error) {
 	command := fmt.Sprintf("cd %s && rm -rf *.yaml", config.TestConfig.LocalConfigDir)
-	if _, err := setup.Env.ExecOnLogtail(command); err != nil {
+	if _, err := setup.Env.ExecOnLoongCollector(command); err != nil {
 		return ctx, err
 	}
 	return ctx, nil

test/engine/setup/chaos/network.go

@@ -81,7 +81,7 @@ func NetworkDelay(ctx context.Context, time int, ip string) (context.Context, er
 	switch setup.Env.GetType() {
 	case "host":
 		command := "/opt/chaosblade/blade create network delay --time " + strconv.FormatInt(int64(time), 10) + " --exclude-port 22 --interface eth0 --destination-ip " + ip
-		_, err := setup.Env.ExecOnLogtail(command)
+		_, err := setup.Env.ExecOnLoongCollector(command)
 		if err != nil {
 			return ctx, err
 		}
@@ -115,7 +115,7 @@ func NetworkLoss(ctx context.Context, percentage int, ip string) (context.Contex
 	switch setup.Env.GetType() {
 	case "host":
 		command := "/opt/chaosblade/blade create network loss --percent " + strconv.FormatInt(int64(percentage), 10) + " --exclude-port 22 --interface eth0 --destination-ip " + ip
-		_, err := setup.Env.ExecOnLogtail(command)
+		_, err := setup.Env.ExecOnLoongCollector(command)
 		if err != nil {
 			return ctx, err
 		}

test/engine/setup/controller/kubernetes.go

@@ -65,10 +65,7 @@ func (c *DeploymentController) GetDeploymentPods(deploymentName, deploymentNames
 	if err != nil {
 		return nil, err
 	}
-	labels := map[string]string{
-		"app": deployment.Spec.Template.Labels["app"],
-	}
-	selector := metav1.FormatLabelSelector(&metav1.LabelSelector{MatchLabels: labels})
+	selector := metav1.FormatLabelSelector(deployment.Spec.Selector)
 	listOptions := metav1.ListOptions{LabelSelector: selector}
 
 	pods, err := c.k8sClient.CoreV1().Pods(deploymentNamespace).List(context.TODO(), listOptions)
@@ -230,11 +227,13 @@ func (c *DynamicController) Apply(filePath string) error {
 	}
 
 	// Apply the object to the Kubernetes cluster
-	resourceInterface := c.dynamicClient.Resource(mapping.Resource)
+	var resourceInterface dynamic.ResourceInterface
+	if obj.GetNamespace() == "" {
+		resourceInterface = c.dynamicClient.Resource(mapping.Resource)
+	} else {
+		resourceInterface = c.dynamicClient.Resource(mapping.Resource).Namespace(obj.GetNamespace())
+	}
 	if oldObj, err := resourceInterface.Get(context.TODO(), obj.GetName(), metav1.GetOptions{}); err != nil {
-		if !meta.IsNoMatchError(err) {
-			return err
-		}
 		// Object does not exist, create it
 		if _, err := resourceInterface.Create(context.TODO(), obj, metav1.CreateOptions{}); err != nil {
 			return err
@@ -257,7 +256,12 @@ func (c *DynamicController) Delete(filePath string) error {
 	}
 
 	// Delete the object from the Kubernetes cluster
-	resourceInterface := c.dynamicClient.Resource(mapping.Resource)
+	var resourceInterface dynamic.ResourceInterface
+	if obj.GetNamespace() == "" {
+		resourceInterface = c.dynamicClient.Resource(mapping.Resource)
+	} else {
+		resourceInterface = c.dynamicClient.Resource(mapping.Resource).Namespace(obj.GetNamespace())
+	}
 	if err := resourceInterface.Delete(context.TODO(), obj.GetName(), metav1.DeleteOptions{}); err != nil {
 		return err
 	}

test/engine/setup/docker_compose.go

@@ -124,7 +124,7 @@ func (d *DockerComposeEnv) Clean() error {
 	return nil
 }
 
-func (d *DockerComposeEnv) ExecOnLogtail(command string) (string, error) {
+func (d *DockerComposeEnv) ExecOnLoongCollector(command string) (string, error) {
 	// exec on host of docker compose
 	fmt.Println(command)
 	cmd := exec.Command("sh", "-c", command)

test/engine/setup/env.go

@@ -23,7 +23,7 @@ var Env TestEnv
 
 type TestEnv interface {
 	GetType() string
-	ExecOnLogtail(command string) (string, error)
+	ExecOnLoongCollector(command string) (string, error)
 	ExecOnSource(ctx context.Context, command string) (string, error)
 }
 
@@ -49,7 +49,7 @@ func Mkdir(ctx context.Context, dir string) (context.Context, error) {
 
 func SetAgentPID(ctx context.Context) (context.Context, error) {
 	command := "ps -e | grep loongcollector | grep -v grep | awk '{print $1}'"
-	result, err := Env.ExecOnLogtail(command)
+	result, err := Env.ExecOnLoongCollector(command)
 	if err != nil {
 		if err.Error() == "not implemented" {
 			return ctx, nil
@@ -58,3 +58,13 @@ func SetAgentPID(ctx context.Context) (context.Context, error) {
 	}
 	return context.WithValue(ctx, config.AgentPIDKey, result), nil
 }
+
+func RunCommandOnSource(ctx context.Context, command string) (context.Context, error) {
+	_, err := Env.ExecOnSource(ctx, command)
+	return ctx, err
+}
+
+func RunCommandOnLoongCollector(ctx context.Context, command string) (context.Context, error) {
+	_, err := Env.ExecOnLoongCollector(command)
+	return ctx, err
+}

test/engine/setup/host.go

@@ -38,7 +38,7 @@ func (h *HostEnv) GetType() string {
 	return "host"
 }
 
-func (h *HostEnv) ExecOnLogtail(command string) (string, error) {
+func (h *HostEnv) ExecOnLoongCollector(command string) (string, error) {
 	return h.exec(command)
 }
 

test/engine/setup/k8s.go

@@ -63,19 +63,19 @@ func (k *K8sEnv) GetType() string {
 	return k.deployType
 }
 
-func (k *K8sEnv) ExecOnLogtail(command string) (string, error) {
+func (k *K8sEnv) ExecOnLoongCollector(command string) (string, error) {
 	if k.k8sClient == nil {
 		return "", fmt.Errorf("k8s client init failed")
 	}
 	var pods *corev1.PodList
 	var err error
 	if k.deployType == "daemonset" {
-		pods, err = k.daemonsetController.GetDaemonSetPods("logtail-ds", "kube-system")
+		pods, err = k.daemonsetController.GetDaemonSetPods("loongcollector-ds", "kube-system")
 		if err != nil {
 			return "", err
 		}
 	} else if k.deployType == "deployment" {
-		pods, err = k.deploymentController.GetRunningDeploymentPods("cluster-agent", "loong-collector")
+		pods, err = k.deploymentController.GetRunningDeploymentPods("loongcollector-cluster", "kube-system")
 		if err != nil {
 			return "", err
 		}

test/engine/setup/subscriber/sls.go

@@ -32,6 +32,7 @@ type SLSSubscriber struct {
 	Aliuid        string
 	Region        string
 	Endpoint      string
+	QueryEndpoint string
 	Project       string
 	Logstore      string
 }
@@ -112,7 +113,7 @@ func (s *SLSSubscriber) getLogFromSLS(sql string, from int32) (*sls.GetLogsRespo
 		From:  tea.Int32(from),
 		To:    tea.Int32(now),
 	}
-	resp, err := s.client.GetLogs(tea.String(config.TestConfig.Project), tea.String(config.TestConfig.GetLogstore(s.TelemetryType)), req)
+	resp, err := s.client.GetLogs(tea.String(s.Project), tea.String(s.Logstore), req)
 	if err != nil {
 		return nil, err
 	}
@@ -134,15 +135,7 @@ func createSLSClient(accessKeyID, accessKeySecret, endpoint string) *sls.Client
 
 func init() {
 	RegisterCreator(slsName, func(spec map[string]interface{}) (Subscriber, error) {
-		telemetryType := "logs"
-		if v, ok := spec["telemetry_type"]; ok {
-			telemetryType = v.(string)
-		}
-		fmt.Println("create sls subscriber with telemetry type", telemetryType)
-		l := &SLSSubscriber{
-			client:        createSLSClient(config.TestConfig.AccessKeyID, config.TestConfig.AccessKeySecret, config.GetQueryEndpoint()),
-			TelemetryType: telemetryType,
-		}
+		l := &SLSSubscriber{}
 		if v, ok := spec["aliuid"]; ok {
 			l.Aliuid = v.(string)
 		} else {
@@ -166,8 +159,19 @@ func init() {
 		if v, ok := spec["logstore"]; ok {
 			l.Logstore = v.(string)
 		} else {
-			l.Logstore = config.TestConfig.GetLogstore(telemetryType)
+			l.Logstore = config.TestConfig.Logstore
+		}
+		if v, ok := spec["query_endpoint"]; ok {
+			l.QueryEndpoint = v.(string)
+		} else {
+			l.QueryEndpoint = config.TestConfig.Endpoint
+		}
+		if v, ok := spec["telemetry_type"]; ok {
+			l.TelemetryType = v.(string)
+		} else {
+			l.TelemetryType = "logs"
 		}
+		l.client = createSLSClient(config.TestConfig.AccessKeyID, config.TestConfig.AccessKeySecret, l.QueryEndpoint)
 		return l, nil
 	})
 	doc.Register("subscriber", slsName, new(SLSSubscriber))

test/engine/steps.go

@@ -32,6 +32,8 @@ func ScenarioInitializer(ctx *godog.ScenarioContext) {
 	ctx.Given(`^subcribe data from \{(\S+)\} with config`, subscriber.InitSubscriber)
 	ctx.Given(`^mkdir \{(.*)\}`, setup.Mkdir)
 	ctx.Given(`^docker-compose boot type \{(\S+)\}$`, setup.SetDockerComposeBootType)
+	ctx.Given(`^run command on datasource \{(.*)\}$`, setup.RunCommandOnSource)
+	ctx.Given(`^run command on loongcollector \{(.*)\}$`, setup.RunCommandOnLoongCollector)
 
 	// chaos
 	ctx.Given(`^network delay package \{(\d+)\}ms for ip \{(.*)\}`, chaos.NetworkDelay)

test/engine/verify/agent.go

@@ -27,7 +27,7 @@ const (
 
 func AgentNotCrash(ctx context.Context) (context.Context, error) {
 	// verify agent crash
-	result, err := setup.Env.ExecOnLogtail(queryPIDCommand)
+	result, err := setup.Env.ExecOnLoongCollector(queryPIDCommand)
 	if err != nil {
 		if err.Error() == "not implemented" {
 			return ctx, nil

test/main.go

@@ -0,0 +1,68 @@
+package main
+
+import (
+	"crypto/rand"
+	"flag"
+	"fmt"
+	"math/big"
+	"os"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/cucumber/godog"
+
+	"github.com/alibaba/ilogtail/test/engine"
+)
+
+func main() {
+	name := flag.String("test_case", "soak", "test case type")
+	flag.Parse()
+
+	for {
+		triggerOneRound(*name)
+		time.Sleep(5 * time.Minute)
+	}
+}
+
+func triggerOneRound(name string) {
+	fmt.Println("=====================================")
+	fmt.Printf("Trigger one round of %s\n", name)
+	// walk all files name endwith .feature in name
+	files := make([]string, 0)
+	err := filepath.Walk(name, func(path string, info os.FileInfo, err error) error {
+		if err != nil {
+			return err
+		}
+		if strings.HasSuffix(path, ".feature") {
+			files = append(files, path)
+		}
+		return nil
+	})
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
+	fmt.Printf("found %d files\n", len(files))
+	randomIndex, err := rand.Int(rand.Reader, big.NewInt(int64(len(files))))
+	if err != nil {
+		fmt.Println(err)
+		return
+	}
+	fmt.Printf("run %s\n", files[randomIndex.Int64()])
+	fmt.Println("=====================================")
+
+	// run godog
+	suite := godog.TestSuite{
+		Name:                "Soak",
+		ScenarioInitializer: engine.ScenarioInitializer,
+		Options: &godog.Options{
+			Format: "pretty",
+			Paths:  []string{files[randomIndex.Int64()]},
+		},
+	}
+	if suite.Run() != 0 {
+		fmt.Printf("run %s failed\n", files[randomIndex.Int64()])
+		return
+	}
+}