refactor: session runner

hrp/boomer.go

@@ -17,11 +17,17 @@ func NewBoomer(spawnCount int, spawnRate float64) *HRPBoomer {
 		Boomer:       boomer.NewStandaloneBoomer(spawnCount, spawnRate),
 		pluginsMutex: new(sync.RWMutex),
 	}
+
+	b.hrpRunner = NewRunner(nil)
+	// set client transport for high concurrency load testing
+	b.hrpRunner.SetClientTransport(b.GetSpawnCount(), b.GetDisableKeepAlive(), b.GetDisableCompression())
+
 	return b
 }
 
 type HRPBoomer struct {
 	*boomer.Boomer
+	hrpRunner    *HRPRunner
 	plugins      []funplugin.IPlugin // each task has its own plugin process
 	pluginsMutex *sync.RWMutex       // avoid data race
 }
@@ -72,16 +78,17 @@ func (b *HRPBoomer) Quit() {
 }
 
 func (b *HRPBoomer) convertBoomerTask(testcase *TestCase, rendezvousList []*Rendezvous) *boomer.Task {
-	hrpRunner := NewRunner(nil)
-	// set client transport for high concurrency load testing
-	hrpRunner.SetClientTransport(b.GetSpawnCount(), b.GetDisableKeepAlive(), b.GetDisableCompression())
 	config := testcase.Config
 
-	// each testcase has its own plugin process
-	plugin, _ := initPlugin(config.Path, false)
-	if plugin != nil {
+	// each testcase has its own session runner
+	sessionRunner, err := b.hrpRunner.NewSessionRunner(testcase)
+	if err != nil {
+		log.Error().Err(err).Msg("failed to create session runner")
+		os.Exit(1)
+	}
+	if sessionRunner.parser.plugin != nil {
 		b.pluginsMutex.Lock()
-		b.plugins = append(b.plugins, plugin)
+		b.plugins = append(b.plugins, sessionRunner.parser.plugin)
 		b.pluginsMutex.Unlock()
 	}
 
@@ -97,9 +104,6 @@ func (b *HRPBoomer) convertBoomerTask(testcase *TestCase, rendezvousList []*Rend
 		Name:   config.Name,
 		Weight: config.Weight,
 		Fn: func() {
-			sessionRunner := hrpRunner.NewSessionRunner(testcase)
-			sessionRunner.parser.plugin = plugin
-
 			testcaseSuccess := true       // flag whole testcase result
 			var transactionSuccess = true // flag current transaction result
 
@@ -131,7 +135,7 @@ func (b *HRPBoomer) convertBoomerTask(testcase *TestCase, rendezvousList []*Rend
 					testcaseSuccess = false
 					transactionSuccess = false
 
-					if hrpRunner.failfast {
+					if b.hrpRunner.failfast {
 						log.Error().Msg("abort running due to failfast setting")
 						break
 					}

hrp/runner.go

@@ -10,6 +10,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pkg/errors"
 	"github.com/rs/zerolog/log"
 	"golang.org/x/net/http2"
 
@@ -151,22 +152,31 @@ func (r *HRPRunner) Run(testcases ...ITestCase) error {
 
 	// run testcase one by one
 	for _, testcase := range testCases {
+		// each testcase has its own session runner
+		sessionRunner, err := r.NewSessionRunner(testcase)
+		if err != nil {
+			log.Error().Err(err).Msg("[Run] init session runner failed")
+			return err
+		}
+		defer sessionRunner.parser.plugin.Quit()
+
 		cfg := testcase.Config
 		// parse config parameters
-		err := initParameterIterator(cfg, "runner")
+		err = initParameterIterator(cfg, "runner")
 		if err != nil {
 			log.Error().Interface("parameters", cfg.Parameters).Err(err).Msg("parse config parameters failed")
 			return err
 		}
 		// 在runner模式下，指定整体策略，cfg.ParametersSetting.Iterators仅包含一个CartesianProduct的迭代器
 		for it := cfg.ParametersSetting.Iterators[0]; it.HasNext(); {
+			var parameterVariables map[string]interface{}
 			// iterate through all parameter iterators and update case variables
 			for _, it := range cfg.ParametersSetting.Iterators {
 				if it.HasNext() {
-					cfg.Variables = mergeVariables(it.Next(), cfg.Variables)
+					parameterVariables = it.Next()
 				}
 			}
-			sessionRunner := r.NewSessionRunner(testcase)
+			sessionRunner.parseConfig(parameterVariables)
 			if err = sessionRunner.Start(); err != nil {
 				log.Error().Err(err).Msg("[Run] run testcase failed")
 				return err
@@ -200,13 +210,25 @@ func (r *HRPRunner) Run(testcases ...ITestCase) error {
 	return nil
 }
 
-func (r *HRPRunner) NewSessionRunner(testcase *TestCase) *SessionRunner {
+func (r *HRPRunner) NewSessionRunner(testcase *TestCase) (*SessionRunner, error) {
 	sessionRunner := &SessionRunner{
 		testCase:  testcase,
 		hrpRunner: r,
 		parser:    newParser(),
 		summary:   newSummary(),
 	}
-	sessionRunner.init()
-	return sessionRunner
+
+	// init parser plugin
+	plugin, err := initPlugin(testcase.Config.Path, r.pluginLogOn)
+	if err != nil {
+		return nil, errors.Wrap(err, "init plugin failed")
+	}
+	sessionRunner.parser.plugin = plugin
+
+	// parse testcase config
+	if err := sessionRunner.parseConfig(nil); err != nil {
+		return nil, errors.Wrap(err, "parse testcase config failed")
+	}
+
+	return sessionRunner, nil
 }

hrp/session.go

@@ -24,9 +24,8 @@ type SessionRunner struct {
 	summary      *TestCaseSummary // record test case summary
 }
 
-func (r *SessionRunner) init() {
-	log.Info().Msg("init session runner")
-	r.parsedConfig = &TConfig{}
+func (r *SessionRunner) resetSession() {
+	log.Info().Msg("clear session runner")
 	r.sessionVariables = make(map[string]interface{})
 	r.transactions = make(map[string]map[transactionType]time.Time)
 	r.startTime = time.Now()
@@ -49,26 +48,9 @@ func (r *SessionRunner) Start() error {
 	config := r.testCase.Config
 	log.Info().Str("testcase", config.Name).Msg("run testcase start")
 
-	// init session runner
-	r.init()
+	// reset session runner
+	r.resetSession()
 
-	// init plugin
-	var err error
-	if r.parser.plugin, err = initPlugin(config.Path, r.hrpRunner.pluginLogOn); err != nil {
-		return err
-	}
-	defer func() {
-		if r.parser.plugin != nil {
-			r.parser.plugin.Quit()
-		}
-	}()
-
-	// parse config
-	if err := r.parseConfig(nil); err != nil {
-		return err
-	}
-
-	r.startTime = time.Now()
 	// run step in sequential order
 	for _, step := range r.testCase.TestSteps {
 		log.Info().Str("step", step.Name()).
@@ -123,6 +105,7 @@ func (r *SessionRunner) MergeStepVariables(vars map[string]interface{}) (map[str
 func (r *SessionRunner) parseConfig(variables map[string]interface{}) error {
 	cfg := r.testCase.Config
 
+	r.parsedConfig = &TConfig{}
 	// deep copy config to avoid data racing
 	if err := copier.Copy(r.parsedConfig, cfg); err != nil {
 		log.Error().Err(err).Msg("copy testcase config failed")
@@ -133,20 +116,20 @@ func (r *SessionRunner) parseConfig(variables map[string]interface{}) error {
 	mergedVars := mergeVariables(variables, cfg.Variables)
 	parsedVariables, err := r.parser.ParseVariables(mergedVars)
 	if err != nil {
-		log.Error().Interface("variables", cfg.Variables).Err(err).Msg("parse config variables failed")
+		log.Error().Interface("variables", mergedVars).Err(err).Msg("parse config variables failed")
 		return err
 	}
 	r.parsedConfig.Variables = parsedVariables
 
 	// parse config name
-	parsedName, err := r.parser.ParseString(cfg.Name, cfg.Variables)
+	parsedName, err := r.parser.ParseString(cfg.Name, parsedVariables)
 	if err != nil {
 		return err
 	}
 	r.parsedConfig.Name = convertString(parsedName)
 
 	// parse config base url
-	parsedBaseURL, err := r.parser.ParseString(cfg.BaseURL, cfg.Variables)
+	parsedBaseURL, err := r.parser.ParseString(cfg.BaseURL, parsedVariables)
 	if err != nil {
 		return err
 	}

hrp/step_request_test.go

@@ -80,8 +80,8 @@ func TestRunRequestRun(t *testing.T) {
 		TestSteps: []IStep{stepGET, stepPOSTData},
 	}
 	runner := NewRunner(t).SetRequestsLogOn()
-	sessionRunner := runner.NewSessionRunner(testcase)
-	sessionRunner.parseConfig(nil)
+	sessionRunner, _ := runner.NewSessionRunner(testcase)
+
 	if _, err := stepGET.Run(sessionRunner); err != nil {
 		t.Fatalf("stepGET.Run() error: %v", err)
 	}

hrp/step_testcase.go

@@ -74,7 +74,11 @@ func (s *StepTestCaseWithOptionalArgs) Run(r *SessionRunner) (*StepResult, error
 	// merge & override extractors
 	copiedTestCase.Config.Export = mergeSlices(s.step.Export, copiedTestCase.Config.Export)
 
-	sessionRunner := r.hrpRunner.NewSessionRunner(copiedTestCase)
+	sessionRunner, err := r.hrpRunner.NewSessionRunner(copiedTestCase)
+	if err != nil {
+		log.Error().Err(err).Msg("create session runner failed")
+		return stepResult, err
+	}
 
 	start := time.Now()
 	err = sessionRunner.Start()