test: refactor E2E trigger (#1861)

* test: refactor E2E trigger

* fix

* fix

* fix

* fix

* fix

* fix

* fix

* fix

.github/workflows/benchmark.yaml

@@ -37,6 +37,7 @@ jobs:
     strategy:
       matrix:
         go-version: [ 1.19.10 ]
+        python-version: [ 3.8 ]
         runner: [ ubuntu-latest ]
       fail-fast: true
     permissions:
@@ -62,6 +63,11 @@ jobs:
         with:
           go-version: ${{ matrix.go-version }}
 
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v5
+        with:
+          python-version: ${{ matrix.python-version }}
+
       - name: Check out code
         uses: actions/checkout@v2
         with:
@@ -83,6 +89,7 @@ jobs:
           BUILD_LOGTAIL_UT: OFF
           WITHOUTGDB: ON
         run: |
+          pip3 install -r test/requirements.txt
           make benchmark
           git stash
 

core/unittest/pipeline/PipelineUnittest.cpp

@@ -2916,6 +2916,7 @@ void PipelineUnittest::TestWaitAllItemsInProcessFinished() const {
     APSARA_TEST_NOT_EQUAL(std::future_status::ready, future.wait_for(std::chrono::seconds(0)));
     pipeline->mInProcessCnt.store(0);
     // recover
+    usleep(3000);
     APSARA_TEST_EQUAL(std::future_status::ready, future.wait_for(std::chrono::seconds(0)));
 }
 

docker/Dockerfile.e2e-test

@@ -1,12 +1,11 @@
-FROM golang:1.19
+FROM python:3.8
 
-RUN go env -w GOPROXY="https://goproxy.cn,direct"
 RUN mkdir -p /tmp/loongcollector
 
 WORKDIR /root
 COPY . ./loongcollector
 
 WORKDIR /root/loongcollector/test
-RUN go mod download
+RUN pip3 install -r requirements.txt
 
 CMD ["sh", "-c", "while true; do sleep 3600; done"]

test/benchmark/test_cases/performance_file_to_blackhole_filebeat/case.feature

@@ -8,5 +8,5 @@ Feature: performance file to blackhole filebeat
     Given docker-compose boot type {benchmark}
     When start docker-compose {performance_file_to_blackhole_filebeat}
     When start monitor {filebeat}
-    When generate random nginx logs to file, speed {10}MB/s, total {3}min, to file {./a.log}
+    When generate random nginx logs to file, speed {10}MB/s, total {3}min, to file {./test_cases/performance_file_to_blackhole_filebeat/a.log}
     When wait monitor until log processing finished

test/benchmark/test_cases/performance_file_to_blackhole_fluentbit/case.feature

@@ -8,5 +8,5 @@ Feature: performance file to blackhole fluentbit
     Given docker-compose boot type {benchmark}
     When start docker-compose {performance_file_to_blackhole_fluentbit}
     When start monitor {fluent-bit}
-    When generate random nginx logs to file, speed {10}MB/s, total {3}min, to file {./a.log}
+    When generate random nginx logs to file, speed {10}MB/s, total {3}min, to file {./test_cases/performance_file_to_blackhole_fluentbit/a.log}
     When wait monitor until log processing finished

test/benchmark/test_cases/performance_file_to_blackhole_ilogtail/case.feature

@@ -8,5 +8,5 @@ Feature: performance file to blackhole iLogtail
     Given docker-compose boot type {benchmark}
     When start docker-compose {performance_file_to_blackhole_ilogtail}
     When start monitor {ilogtailC}
-    When generate random nginx logs to file, speed {10}MB/s, total {3}min, to file {./a.log}
+    When generate random nginx logs to file, speed {10}MB/s, total {3}min, to file {./test_cases/performance_file_to_blackhole_ilogtail/a.log}
     When wait monitor until log processing finished

test/benchmark/test_cases/performance_file_to_blackhole_ilogtailspl/case.feature

@@ -8,5 +8,5 @@ Feature: performance file to blackhole iLogtail
     Given docker-compose boot type {benchmark}
     When start docker-compose {performance_file_to_blackhole_ilogtailspl}
     When start monitor {ilogtailC}
-    When generate random nginx logs to file, speed {10}MB/s, total {3}min, to file {./a.log}
+    When generate random nginx logs to file, speed {10}MB/s, total {3}min, to file {./test_cases/performance_file_to_blackhole_ilogtailspl/a.log}
     When wait monitor until log processing finished

test/benchmark/test_cases/performance_file_to_blackhole_vector/case.feature

@@ -8,5 +8,5 @@ Feature: performance file to blackhole vector
     Given docker-compose boot type {benchmark}
     When start docker-compose {performance_file_to_blackhole_vector}
     When start monitor {vector}
-    When generate random nginx logs to file, speed {10}MB/s, total {3}min, to file {./a.log}
+    When generate random nginx logs to file, speed {10}MB/s, total {3}min, to file {./test_cases/performance_file_to_blackhole_vector/a.log}
     When wait monitor until log processing finished

test/config/config.go

@@ -15,6 +15,7 @@ package config
 
 import (
 	"os"
+	"path/filepath"
 	"strconv"
 	"time"
 
@@ -72,6 +73,10 @@ func ParseConfig() {
 		TestConfig.GeneratedLogDir = "/tmp/loongcollector"
 	}
 	TestConfig.WorkDir = os.Getenv("WORK_DIR")
+	if len(TestConfig.WorkDir) == 0 {
+		testFileDir, _ := os.Getwd()
+		TestConfig.WorkDir = filepath.Dir(testFileDir)
+	}
 
 	// SSH
 	TestConfig.SSHUsername = os.Getenv("SSH_USERNAME")

test/engine/setup/docker_compose.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"os/exec"
 	"path/filepath"
 	"time"
 
@@ -128,5 +129,10 @@ func (d *DockerComposeEnv) ExecOnLogtail(command string) (string, error) {
 }
 
 func (d *DockerComposeEnv) ExecOnSource(ctx context.Context, command string) (string, error) {
-	return "", fmt.Errorf("not implemented")
+	// exec on host of docker compose
+	fmt.Println(command)
+	cmd := exec.Command("sh", "-c", command)
+	output, err := cmd.CombinedOutput()
+	fmt.Println(string(output))
+	return string(output), err
 }

test/engine/steps.go

@@ -10,8 +10,11 @@ import (
 	"github.com/alibaba/ilogtail/test/engine/cleanup"
 	"github.com/alibaba/ilogtail/test/engine/control"
 	"github.com/alibaba/ilogtail/test/engine/setup"
+	"github.com/alibaba/ilogtail/test/engine/setup/monitor"
 	"github.com/alibaba/ilogtail/test/engine/setup/subscriber"
 	"github.com/alibaba/ilogtail/test/engine/trigger"
+	"github.com/alibaba/ilogtail/test/engine/trigger/ebpf"
+	"github.com/alibaba/ilogtail/test/engine/trigger/log"
 	"github.com/alibaba/ilogtail/test/engine/verify"
 )
 
@@ -27,6 +30,7 @@ func ScenarioInitializer(ctx *godog.ScenarioContext) {
 	ctx.Given(`^remove http config \{(.*)\}`, control.RemoveHTTPConfig)
 	ctx.Given(`^subcribe data from \{(\S+)\} with config`, subscriber.InitSubscriber)
 	ctx.Given(`^mkdir \{(.*)\}`, setup.Mkdir)
+	ctx.Given(`^docker-compose boot type \{(\S+)\}$`, setup.SetDockerComposeBootType)
 	// ------------------------------------------
 
 	// When
@@ -41,19 +45,25 @@ func ScenarioInitializer(ctx *godog.ScenarioContext) {
 
 	// generate
 	ctx.When(`^begin trigger`, trigger.BeginTrigger)
-	ctx.When(`^generate \{(\d+)\} regex logs to file \{(.*)\}, with interval \{(\d+)\}ms$`, trigger.RegexSingle)
-	ctx.When(`^generate \{(\d+)\} multiline regex logs to file \{(.*)\}, with interval \{(\d+)\}ms$`, trigger.RegexMultiline)
-	ctx.When(`^generate \{(\d+)\} regex gbk logs to file \{(.*)\}, with interval \{(\d+)\}ms$`, trigger.RegexSingleGBK)
-	ctx.When(`^generate \{(\d+)\} http logs, with interval \{(\d+)\}ms, url: \{(.*)\}, method: \{(.*)\}, body:`, trigger.HTTP)
-	ctx.When(`^generate \{(\d+)\} apsara logs to file \{(.*)\}, with interval \{(\d+)\}ms$`, trigger.Apsara)
-	ctx.When(`^generate \{(\d+)\} delimiter logs to file \{(.*)\}, with interval \{(\d+)\}ms, with delimiter \{(.*)\} and quote \{(.*)\}$`, trigger.DelimiterSingle)
-	ctx.When(`^generate \{(\d+)\} multiline delimiter logs to file \{(.*)\}, with interval \{(\d+)\}ms, with delimiter \{(.*)\} and quote \{(.*)\}$`, trigger.DelimiterMultiline)
-	ctx.When(`^generate \{(\d+)\} json logs to file \{(.*)\}, with interval \{(\d+)\}ms$`, trigger.JSONSingle)
-	ctx.When(`^generate \{(\d+)\} multiline json logs to file \{(.*)\}, with interval \{(\d+)\}ms$`, trigger.JSONMultiline)
-	ctx.When(`^execute \{(\d+)\} commands to generate process security events`, trigger.TrigerProcessSecurityEvents)
-	ctx.When(`^execute \{(\d+)\} commands to generate network security events on url \{(.*)\}$`, trigger.TrigerNetworksSecurityEvents)
-	ctx.When(`^execute \{(\d+)\} commands to generate file security events on files \{(.*)\}$`, trigger.TrigerFileSecurityEvents)
-	ctx.When(`^generate \{(\d+)\} HTTP requests, with interval \{(\d+)\}ms, url: \{(.*)\}`, trigger.TrigerHTTP)
+	// log
+	ctx.When(`^generate \{(\d+)\} regex logs to file \{(.*)\}, with interval \{(\d+)\}ms$`, log.RegexSingle)
+	ctx.When(`^generate \{(\d+)\} multiline regex logs to file \{(.*)\}, with interval \{(\d+)\}ms$`, log.RegexMultiline)
+	ctx.When(`^generate \{(\d+)\} regex gbk logs to file \{(.*)\}, with interval \{(\d+)\}ms$`, log.RegexSingleGBK)
+	ctx.When(`^generate \{(\d+)\} http logs, with interval \{(\d+)\}ms, url: \{(.*)\}, method: \{(.*)\}, body:`, log.HTTP)
+	ctx.When(`^generate \{(\d+)\} apsara logs to file \{(.*)\}, with interval \{(\d+)\}ms$`, log.Apsara)
+	ctx.When(`^generate \{(\d+)\} delimiter logs to file \{(.*)\}, with interval \{(\d+)\}ms, with delimiter \{(.*)\} and quote \{(.*)\}$`, log.DelimiterSingle)
+	ctx.When(`^generate \{(\d+)\} multiline delimiter logs to file \{(.*)\}, with interval \{(\d+)\}ms, with delimiter \{(.*)\} and quote \{(.*)\}$`, log.DelimiterMultiline)
+	ctx.When(`^generate \{(\d+)\} json logs to file \{(.*)\}, with interval \{(\d+)\}ms$`, log.JSONSingle)
+	ctx.When(`^generate \{(\d+)\} multiline json logs to file \{(.*)\}, with interval \{(\d+)\}ms$`, log.JSONMultiline)
+	ctx.When(`^generate random nginx logs to file, speed \{(\d+)\}MB/s, total \{(\d+)\}min, to file \{(.*)\}`, log.Nginx)
+	ctx.When(`^start monitor \{(\S+)\}`, monitor.StartMonitor)
+	ctx.When(`^wait monitor until log processing finished$`, monitor.WaitMonitorUntilProcessingFinished)
+
+	// ebpf
+	ctx.When(`^execute \{(\d+)\} commands to generate process security events`, ebpf.ProcessSecurityEvents)
+	ctx.When(`^execute \{(\d+)\} commands to generate network security events on url \{(.*)\}$`, ebpf.NetworksSecurityEvents)
+	ctx.When(`^execute \{(\d+)\} commands to generate file security events on files \{(.*)\}$`, ebpf.FileSecurityEvents)
+	ctx.When(`^generate \{(\d+)\} HTTP requests, with interval \{(\d+)\}ms, url: \{(.*)\}`, ebpf.HTTP)
 	// ------------------------------------------
 
 	// Then

test/engine/trigger/protocol.go → test/engine/trigger/ebpf/http.go

@@ -11,7 +11,7 @@
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
-package trigger
+package ebpf
 
 import (
 	"context"
@@ -22,7 +22,7 @@ import (
 	"github.com/alibaba/ilogtail/test/engine/setup"
 )
 
-func TrigerHTTP(ctx context.Context, count int, interval int, url string) (context.Context, error) {
+func HTTP(ctx context.Context, count int, interval int, url string) (context.Context, error) {
 	logger.Debugf(context.Background(), "count:%d interval:%d url:%s", count, interval, url)
 	cmd := fmt.Sprintf("curl -vL %s", url)
 	time.Sleep(time.Second * 5)

test/engine/trigger/ebpf/remote_mmap.py

@@ -0,0 +1,22 @@
+import argparse
+import mmap
+import os
+
+def main():
+    parser = argparse.ArgumentParser(description='mmap')
+    parser.add_argument('--commandCnt', type=int, default=10, help='command count')
+    parser.add_argument('--filename', type=str, default='/tmp/loongcollector/ebpfFileSecurityHook3.log', help='filename')
+
+    args = parser.parse_args()
+
+    with open(args.filename, 'w') as f:
+        fd = f.fileno()
+        for i in range(args.commandCnt):
+            mm = mmap.mmap(fd, 20, prot=mmap.PROT_READ | mmap.PROT_WRITE, flags=mmap.MAP_SHARED)
+            mm.close()
+
+    os.remove(args.filename)
+
+
+if __name__ == '__main__':
+    main()

test/engine/trigger/ebpf_trigger.go → test/engine/trigger/ebpf/security.go

@@ -11,24 +11,24 @@
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
-package trigger
+package ebpf
 
 import (
 	"context"
-	"html/template"
+	"strconv"
 	"strings"
 	"time"
 
-	"github.com/alibaba/ilogtail/test/config"
 	"github.com/alibaba/ilogtail/test/engine/setup"
+	"github.com/alibaba/ilogtail/test/engine/trigger"
 )
 
 /*
 ********************
 input_process_security
 ********************
 */
-func TrigerProcessSecurityEvents(ctx context.Context, commandCnt int) (context.Context, error) {
+func ProcessSecurityEvents(ctx context.Context, commandCnt int) (context.Context, error) {
 	time.Sleep(5 * time.Second)
 	if err := execveCommands(ctx, commandCnt); err != nil {
 		return ctx, err
@@ -51,7 +51,7 @@ func execveCommands(ctx context.Context, commandCnt int) error {
 input_network_security
 ********************
 */
-func TrigerNetworksSecurityEvents(ctx context.Context, commandCnt int, url string) (context.Context, error) {
+func NetworksSecurityEvents(ctx context.Context, commandCnt int, url string) (context.Context, error) {
 	time.Sleep(5 * time.Second)
 	if err := curlURL(ctx, commandCnt, url); err != nil {
 		return ctx, err
@@ -74,9 +74,8 @@ func curlURL(ctx context.Context, commandCnt int, url string) error {
 input_file_security
 ********************
 */
-const triggerFileSecurityTemplate = "cd {{.WorkDir}} && COMMAND_CNT={{.CommandCnt}} FILE_NAME={{.FileName}} {{.Command}}"
 
-func TrigerFileSecurityEvents(ctx context.Context, commandCnt int, filenames string) (context.Context, error) {
+func FileSecurityEvents(ctx context.Context, commandCnt int, filenames string) (context.Context, error) {
 	time.Sleep(5 * time.Second)
 	if err := rwFile(ctx, commandCnt, filenames); err != nil {
 		return ctx, err
@@ -112,20 +111,10 @@ func rwFile(ctx context.Context, commandCnt int, filenames string) error {
 }
 
 func mmapFile(ctx context.Context, commandCnt int, filenames string) error {
-	mmapFileCommand := getRunTriggerCommand("TestGenerateMmapCommand")
 	files := strings.Split(filenames, ",")
 	for _, file := range files {
-		var triggerEBPFCommand strings.Builder
-		template := template.Must(template.New("trigger").Parse(triggerFileSecurityTemplate))
-		if err := template.Execute(&triggerEBPFCommand, map[string]interface{}{
-			"WorkDir":    config.TestConfig.WorkDir,
-			"CommandCnt": commandCnt,
-			"FileName":   file,
-			"Command":    mmapFileCommand,
-		}); err != nil {
-			return err
-		}
-		if _, err := setup.Env.ExecOnSource(ctx, triggerEBPFCommand.String()); err != nil {
+		mmapFileCommand := trigger.GetRunTriggerCommand("ebpf", "mmap", "commandCnt", strconv.FormatInt(int64(commandCnt), 10), "filename", file)
+		if _, err := setup.Env.ExecOnSource(ctx, mmapFileCommand); err != nil {
 			return err
 		}
 	}