Method executeActionsParallel

workflow/executor.go:151–211

executeActionsParallel executes actions in parallel

(ctx context.Context, client *csclient.Client, beaconID string, actions []Action)

// executeActionsParallel executes actions in parallel
func (e *Executor) executeActionsParallel(ctx context.Context, client *csclient.Client, beaconID string, actions []Action) error {
	var wg sync.WaitGroup
	errCh := make(chan error, len(actions))

	for i, action := range actions {
		// Check conditions before starting goroutine
		if !e.evaluateActionConditions(action) {
			e.logInfo("[%d] Conditions not met, skipping action: %s", i+1, action.Name)
			continue
		}

		wg.Add(1)
		go func(idx int, act Action) {
			defer wg.Done()

			e.logInfo("[%d] Executing action: %s (type: %s)", idx+1, act.Name, act.Type)

			output, err := e.executeAction(ctx, client, beaconID, act)
			if err != nil {
				e.logError("[%d] Action failed: %v", idx+1, err)

				// Execute on_failure actions
				if len(act.OnFailure) > 0 {
					e.logInfo("[%d] Executing on_failure actions", idx+1)
					if failErr := e.executeActions(ctx, client, beaconID, act.OnFailure); failErr != nil {
						errCh <- failErr
						return
					}
				}

				errCh <- fmt.Errorf("action %s failed: %w", act.Name, err)
				return
			}

			// Store output (thread-safe)
			e.outputMu.Lock()
			e.outputs[act.Name] = output
			e.outputMu.Unlock()

			e.logInfo("[%d] Action completed successfully", idx+1)

			// Execute on_success actions
			if len(act.OnSuccess) > 0 {
				e.logInfo("[%d] Executing on_success actions", idx+1)
				if succErr := e.executeActions(ctx, client, beaconID, act.OnSuccess); succErr != nil {
					errCh <- succErr
				}
			}
		}(i, action)
	}

	wg.Wait()
	close(errCh)

	// Check for errors
	for err := range errCh {
		return err
	}