From 7e56de50d5ed265312d288c3d4a8112065d194e2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Geyslan=20Greg=C3=B3rio?= Date: Wed, 13 Nov 2024 13:29:54 -0300 Subject: [PATCH] wip --- cmd/evt/cmd/stress/stress.go | 94 ++++++++++++++++++++++++++-------- cmd/evt/cmd/trigger/trigger.go | 39 +++++++------- 2 files changed, 94 insertions(+), 39 deletions(-) diff --git a/cmd/evt/cmd/stress/stress.go b/cmd/evt/cmd/stress/stress.go index 333fb62f9b9c..640ad08309cb 100644 --- a/cmd/evt/cmd/stress/stress.go +++ b/cmd/evt/cmd/stress/stress.go @@ -13,13 +13,14 @@ import ( "syscall" "time" - "github.com/aquasecurity/tracee/pkg/cmd/flags" - "github.com/aquasecurity/tracee/pkg/logger" - "github.com/aquasecurity/tracee/pkg/policy/v1beta1" "github.com/dustin/go-humanize" "github.com/spf13/cobra" "golang.org/x/exp/slices" "gopkg.in/yaml.v2" + + "github.com/aquasecurity/tracee/pkg/cmd/flags" + "github.com/aquasecurity/tracee/pkg/logger" + "github.com/aquasecurity/tracee/pkg/policy/v1beta1" ) func init() { @@ -177,6 +178,24 @@ func getStressConfig(cmd *cobra.Command) (*StressConfig, error) { }, nil } +const coolDownTime = 10 * time.Second + +func coolDownCtx(ctx context.Context, msg string, coolDownTime time.Duration) { + fmt.Printf("%s: waiting %v for cool down...\n", msg, coolDownTime) + + select { + case <-time.After(coolDownTime): + return + case <-ctx.Done(): + return + } +} + +func coolDown(msg string, coolDownTime time.Duration) { + fmt.Printf("%s: waiting %v for cool down...\n", msg, coolDownTime) + time.Sleep(coolDownTime) +} + func stressRun(cmd *cobra.Command, args []string) error { logger.Init(logger.NewDefaultLoggingConfig()) @@ -228,9 +247,7 @@ func stressRun(cmd *cobra.Command, args []string) error { triggerPids = append(triggerPids, triggerPid) // limit the comm length to 15 printable characters comm := fmt.Sprintf("%s.sh", evt) - if len(comm) > 15 { - comm = comm[:15] - } + comm = comm[:min(len(comm), 15)] triggerComms = append(triggerComms, comm) } } @@ -243,24 +260,59 @@ func stressRun(cmd *cobra.Command, args []string) error { return err } - const coolDownTime = 10 * time.Second - fmt.Printf("tracee started: waiting %v for cool down...\n", coolDownTime) - select { - case <-time.After(coolDownTime): - case <-ctx.Done(): - goto cleanup - } + coolDownCtx(ctx, "tracee started", coolDownTime) // signal all triggers to start - err = syscall.Kill(-os.Getpid(), syscall.SIGUSR1) - if err != nil { - return fmt.Errorf("sending SIGUSR1 to all triggers: %w", err) + for _, pid := range triggerPids { + err = syscall.Kill(pid, syscall.SIGUSR1) + if err != nil { + return fmt.Errorf("sending SIGUSR1 to trigger %d: %w", pid, err) + } + } + // err = syscall.Kill(-os.Getpid(), syscall.SIGUSR1) + // if err != nil { + // return fmt.Errorf("sending SIGUSR1 to all triggers: %w", err) + // } + + if cfg.Mode.Selected == StressModeOps { + wg.Add(1) + go func(cancel context.CancelFunc) { + logger.Debugw("checkTriggersLiveness goroutine started") + defer logger.Debugw("checkTriggersLiveness goroutine finished") + + defer wg.Done() + + for { + select { + case <-time.After(1 * time.Second): + allDone := true + for _, pid := range triggerPids { + if syscall.Kill(pid, 0) == nil { + allDone = false + break + } + } + + if allDone { + coolDown("triggers finished", coolDownTime) + cancel() + return + } + } + } + }(cancel) } // block until tracee is finished or context is done for { select { case <-ctx.Done(): + for _, pid := range triggerPids { + err = syscall.Kill(pid, syscall.SIGTERM) + if err != nil { + return fmt.Errorf("sending SIGTERM to trigger %d: %w", pid, err) + } + } goto cleanup case err := <-traceeStatus: if err != nil { @@ -289,8 +341,8 @@ func triggerEvent(ctx context.Context, wg *sync.WaitGroup, event string, ops uin cmd := exec.CommandContext(ctx, "./dist/evt", "trigger", "-e", event, "-o", fmt.Sprintf("%d", ops)) cmd.SysProcAttr = &syscall.SysProcAttr{ - Setpgid: true, - Pgid: os.Getpid(), + // Setpgid: true, + // Pgid: os.Getpid(), } cmd.Stdin = nil cmd.Stdout = os.Stdout @@ -306,11 +358,14 @@ func triggerEvent(ctx context.Context, wg *sync.WaitGroup, event string, ops uin wg.Add(1) go func() { + logger.Debugw("waitForTrigger goroutine started") + defer logger.Debugw("waitForTrigger goroutine finished") + defer wg.Done() waitErr := cmd.Wait() if waitErr != nil { - err = fmt.Errorf("trigger finished: %w", waitErr) + fmt.Errorf("waiting for trigger: %w", waitErr) } }() @@ -411,7 +466,6 @@ func RunTracee( errCh <- fmt.Errorf("tracee finished: %w", err) case <-time.After(1 * time.Second): // give tracee some time to start - break } if err != nil { return diff --git a/cmd/evt/cmd/trigger/trigger.go b/cmd/evt/cmd/trigger/trigger.go index eb81a0eb73fd..7129523cec6d 100644 --- a/cmd/evt/cmd/trigger/trigger.go +++ b/cmd/evt/cmd/trigger/trigger.go @@ -104,26 +104,38 @@ func triggerRun(cmd *cobra.Command, args []string) error { startChan := make(chan os.Signal, 1) signal.Notify(startChan, syscall.SIGUSR1) fmt.Printf("[trigger:%d:%s] Waiting for start signal\n", os.Getpid(), cfg.Event) + select { case <-ctx.Done(): + fmt.Printf("[trigger:%d:%s] Stopping triggering: %v\n", os.Getpid(), cfg.Event, ctx.Err()) return ctx.Err() case <-startChan: fmt.Printf("[trigger:%d:%s] Starting triggering %d ops with %v sleep time\n", os.Getpid(), cfg.Event, cfg.Ops, cfg.Sleep) } for i := uint32(0); i < cfg.Ops; i++ { - cmd := exec.CommandContext(ctx, fmt.Sprintf("./cmd/evt/cmd/trigger/triggers/%s.sh", cfg.Event)) - err := cmd.Start() - if err != nil { - return fmt.Errorf("[trigger:%d:%s] failed to start command: %w", os.Getpid(), cfg.Event, err) + select { + case <-ctx.Done(): + fmt.Printf("[trigger:%d:%s] Stopping triggering: %v\n", os.Getpid(), cfg.Event, ctx.Err()) + return ctx.Err() + case <-time.After(cfg.Sleep): + // continue } - err = cmd.Wait() + cmd := exec.CommandContext(ctx, fmt.Sprintf("./cmd/evt/cmd/trigger/triggers/%s.sh", cfg.Event)) + err := cmd.Run() if err != nil { - return fmt.Errorf("[trigger:%d:%s] failed to wait for command: %w", os.Getpid(), cfg.Event, err) + return fmt.Errorf("[trigger:%d:%s] failed to run command: %w", os.Getpid(), cfg.Event, err) } - - time.Sleep(cfg.Sleep) + // err := cmd.Start() + // if err != nil { + // return fmt.Errorf("[trigger:%d:%s] failed to start command: %w", os.Getpid(), cfg.Event, err) + // } + + // err = cmd.Wait() + // if err != nil { + // return fmt.Errorf("[trigger:%d:%s] failed to wait for command: %w", os.Getpid(), cfg.Event, err) + // } } fmt.Printf("[trigger:%d:%s] Finished triggering %d ops\n", os.Getpid(), cfg.Event, cfg.Ops) @@ -131,17 +143,6 @@ func triggerRun(cmd *cobra.Command, args []string) error { return nil } -// func waitForCommand(cmd *exec.Cmd) <-chan error { -// done := make(chan error) - -// go func() { -// done <- cmd.Wait() -// close(done) -// }() - -// return done -// } - func Cmd() *cobra.Command { return triggerCmd }