GithubBackup/httprunner
1
0
Fork 0
mirror of https://github.com/httprunner/httprunner.git synced 2026-09-04 23:16:57 +08:00
Code Issues Packages Projects Releases Wiki Activity

change: update docs

Browse Source
This commit is contained in:
徐聪
2022-07-25 21:29:19 +08:00
parent 60a6aebbcb
commit 1c5039697a
14 changed files with 93 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hrp/boomer.go
+4 -1
View File
@@ -208,6 +208,7 @@ func (b *HRPBoomer) runTestCases(testCases []*TCase, profile *boomer.Profile) {
testcases = append(testcases, tesecase)
}
b.SetProfile(profile)
b.InitBoomer()
log.Info().Interface("testcases", testcases).Interface("profile", profile).Msg("run tasks successful")
@@ -222,7 +223,7 @@ func (b *HRPBoomer) rebalanceBoomer(profile *boomer.Profile) {
log.Info().Interface("profile", profile).Msg("rebalance tasks successful")
}
func (b *HRPBoomer) PollTasks() {
func (b *HRPBoomer) PollTasks(ctx context.Context) {
for {
select {
case task := <-b.Boomer.GetTasksChan():
@@ -240,6 +241,8 @@ func (b *HRPBoomer) PollTasks() {
case <-b.Boomer.GetCloseChan():
return
case <-ctx.Done():
return
}
}
}
hrp/cmd/boom.go
+5 -4
View File
@@ -71,7 +71,7 @@ var boomCmd = &cobra.Command{
if boomArgs.autoStart {
hrpBoomer.InitBoomer()
} else {
go hrpBoomer.StartServer()
go hrpBoomer.StartServer(ctx, boomArgs.masterHttpAddress)
}
go hrpBoomer.PollTestCases(ctx)
hrpBoomer.RunMaster()
@@ -79,9 +79,8 @@ var boomCmd = &cobra.Command{
if boomArgs.ignoreQuit {
hrpBoomer.SetIgnoreQuit()
}
go hrpBoomer.PollTasks()
go hrpBoomer.PollTasks(ctx)
hrpBoomer.RunWorker()
time.Sleep(3 * time.Second)
case "standalone":
if venv != "" {
hrpBoomer.SetPython3Venv(venv)
@@ -102,6 +101,7 @@ type BoomArgs struct {
masterPort int
masterBindHost string
masterBindPort int
masterHttpAddress string
autoStart bool
expectWorkers int
expectWorkersMaxWait int
@@ -129,11 +129,12 @@ func init() {
boomCmd.Flags().BoolVar(&boomArgs.master, "master", false, "master of distributed testing")
boomCmd.Flags().StringVar(&boomArgs.masterBindHost, "master-bind-host", "127.0.0.1", "Interfaces (hostname, ip) that hrp master should bind to. Only used when running with --master. Defaults to * (all available interfaces).")
boomCmd.Flags().IntVar(&boomArgs.masterBindPort, "master-bind-port", 5557, "Port that hrp master should bind to. Only used when running with --master. Defaults to 5557.")
boomCmd.Flags().StringVar(&boomArgs.masterHttpAddress, "master-http-address", ":9771", "Interfaces (ip:port) that hrp master should control by user. Only used when running with --master. Defaults to *:9771.")
boomCmd.Flags().BoolVar(&boomArgs.worker, "worker", false, "worker of distributed testing")
boomCmd.Flags().BoolVar(&boomArgs.ignoreQuit, "ignore-quit", false, "ignores quit from master (only when --worker is used)")
boomCmd.Flags().StringVar(&boomArgs.masterHost, "master-host", "127.0.0.1", "Host or IP address of hrp master for distributed load testing.")
boomCmd.Flags().IntVar(&boomArgs.masterPort, "master-port", 5557, "The port to connect to that is used by the hrp master for distributed load testing.")
boomCmd.Flags().BoolVar(&boomArgs.autoStart, "autostart", false, "Starts the test immediately (without disabling the web UI). Use --spawn-count and --spawn-rate to control user count and run time")
boomCmd.Flags().BoolVar(&boomArgs.autoStart, "autostart", false, "Starts the test immediately (without disabling the web UI). Use --spawn-count and --spawn-rate to control user count and increase rate")
boomCmd.Flags().IntVar(&boomArgs.expectWorkers, "expect-workers", 1, "How many workers master should expect to connect before starting the test (only when --autostart is used)")
boomCmd.Flags().IntVar(&boomArgs.expectWorkersMaxWait, "expect-workers-max-wait", 0, "How many workers master should expect to connect before starting the test (only when --autostart is used")
}
hrp/internal/boomer/boomer.go
+3
View File
@@ -461,6 +461,9 @@ func (b *Boomer) Start(Args *Profile) error {
if b.masterRunner.isStarted() {
return errors.New("already started")
}
if b.masterRunner.getState() == StateStopping {
return errors.New("Please wait for all workers to finish")
}
b.SetSpawnCount(Args.SpawnCount)
b.SetSpawnRate(Args.SpawnRate)
b.SetProfile(Args)
hrp/internal/boomer/client_grpc.go
+1 -1
View File
@@ -216,7 +216,7 @@ func (c *grpcClient) newBiStreamClient() (err error) {
return err
}
c.config.setBiStreamClient(biStream)
println("successful to establish bidirectional stream with master, press Ctrl+c to quit.\n")
println("successful to establish bidirectional stream with master, press Ctrl+c to quit.")
return nil
}
hrp/internal/boomer/runner.go
+30 -38
View File
@@ -10,12 +10,10 @@ import (
"sync/atomic"
"time"
"github.com/go-errors/errors"
"github.com/httprunner/httprunner/v4/hrp/internal/boomer/grpc/messager"
"github.com/httprunner/httprunner/v4/hrp/internal/builtin"
"github.com/jinzhu/copier"
"github.com/go-errors/errors"
"github.com/olekukonko/tablewriter"
"github.com/rs/zerolog/log"
)
@@ -200,15 +198,15 @@ type runner struct {
controller *Controller
loop *Loop // specify loop count for testcase, count = loopCount * spawnCount
// rebalance spawn
// dynamically balance boomer running parameters
rebalance chan bool
// all running workers(goroutines) will select on this channel.
// close this channel will stop all running workers.
// stop signals the run goroutine should shutdown.
stopChan chan bool
// all running workers(goroutines) will select on this channel.
// stopping is closed by run goroutine on shutdown.
stoppingChan chan bool
// done is closed when all goroutines from start() complete.
doneChan chan bool
reportChan chan bool
@@ -216,10 +214,10 @@ type runner struct {
// close this channel will stop all goroutines used in runner.
closeChan chan bool
// wgMu blocks concurrent waitgroup mutation while server stopping
// wgMu blocks concurrent waitgroup mutation while boomer stopping
wgMu sync.RWMutex
// wg is used to wait for the goroutines that depends on the server state
// to exit when stopping the server.
// wg is used to wait for all running workers(goroutines) that depends on the boomer state
// to exit when stopping the boomer.
wg sync.WaitGroup
outputs []Output
@@ -544,15 +542,20 @@ func (r *runner) statsStart() {
func (r *runner) stop() {
// stop previous goroutines without blocking
// those goroutines will exit when r.safeRun returns
r.Stop()
r.gracefulStop()
if r.rateLimitEnabled {
r.rateLimiter.Stop()
}
r.updateState(StateStopped)
}
// HardStop stops the server without coordination with other members in the cluster.
func (r *runner) hardStop() {
// gracefulStop stops the boomer gracefully, and shuts down the running goroutine.
// gracefulStop should be called after a start(), otherwise it will block forever.
// When stopping leader, Stop transfers its leadership to one of its peers
// before stopping the boomer.
// gracefulStop terminates the boomer and performs any necessary finalization.
// Do and Process cannot be called after Stop has been invoked.
func (r *runner) gracefulStop() {
select {
case r.stopChan <- true:
case <-r.doneChan:
@@ -561,30 +564,16 @@ func (r *runner) hardStop() {
<-r.doneChan
}
// Stop stops the server gracefully, and shuts down the running goroutine.
// Stop should be called after a Start(s), otherwise it will block forever.
// When stopping leader, Stop transfers its leadership to one of its peers
// before stopping the server.
// Stop terminates the Server and performs any necessary finalization.
// Do and Process cannot be called after Stop has been invoked.
func (r *runner) Stop() {
r.hardStop()
}
// StopNotify returns a channel that receives a bool type value
// when the runner is stopped.
func (r *runner) StopNotify() <-chan bool { return r.doneChan }
// StopNotify returns a channel that receives a empty struct
// when the server is stopped.
func (r *runner) StopNotify() <-chan bool { return r.stopChan }
// DoneNotify returns a channel that receives a empty struct
// when the server is stopped.
func (r *runner) DoneNotify() <-chan bool { return r.doneChan }
// StoppingNotify returns a channel that receives a empty struct
// when the server is being stopped.
// StoppingNotify returns a channel that receives a bool type value
// when the runner is being stopped.
func (r *runner) StoppingNotify() <-chan bool { return r.stoppingChan }
// RebalanceNotify returns a channel that receives a empty struct
// when the server is being stopped.
// RebalanceNotify returns a channel that receives a bool type value
// when the runner is being rebalance.
func (r *runner) RebalanceNotify() <-chan bool { return r.rebalance }
func (r *runner) getState() int32 {
@@ -758,6 +747,7 @@ func (r *workerRunner) onMessage(msg *genericMessage) {
r.onSpawnMessage(msg)
case "quit":
if r.ignoreQuit {
log.Warn().Msg("master already quit, waiting to reconnect master.")
break
}
r.close()
@@ -777,6 +767,7 @@ func (r *workerRunner) onMessage(msg *genericMessage) {
case "quit":
r.stop()
if r.ignoreQuit {
log.Warn().Msg("master already quit, waiting to reconnect master.")
break
}
r.close()
@@ -788,6 +779,7 @@ func (r *workerRunner) onMessage(msg *genericMessage) {
r.onSpawnMessage(msg)
case "quit":
if r.ignoreQuit {
log.Warn().Msg("master already quit, waiting to reconnect master.")
break
}
r.close()
@@ -815,13 +807,13 @@ func (r *workerRunner) startListener() {
// run worker service
func (r *workerRunner) run() {
println("\n========================= HttpRunner Worker for Distributed Load Testing ========================= ")
println("==================== HttpRunner Worker for Distributed Load Testing ==================== ")
r.updateState(StateInit)
r.client = newClient(r.masterHost, r.masterPort, r.nodeID)
println(fmt.Sprintf("ready to connect master to %s:%d", r.masterHost, r.masterPort))
err := r.client.start()
if err != nil {
log.Error().Err(err).Msg(fmt.Sprintf("failed to connect to master(%s:%d) with error %v\n", r.masterHost, r.masterPort))
log.Error().Err(err).Msg(fmt.Sprintf("failed to connect to master(%s:%d)", r.masterHost, r.masterPort))
}
if err = r.client.register(r.client.config.ctx); err != nil {
@@ -904,7 +896,7 @@ func (r *workerRunner) start() {
r.rateLimiter.Start()
}
r.once.Do(r.outputOnStart)
r.outputOnStart()
go r.spawnWorkers(r.getSpawnCount(), r.getSpawnRate(), r.stoppingChan, r.spawnComplete)
hrp/internal/boomer/runner_test.go
+2 -2
View File
@@ -126,13 +126,13 @@ func TestStopNotify(t *testing.T) {
close(r.doneChan)
}()
notifier := r.DoneNotify()
notifier := r.StopNotify()
select {
case <-notifier:
t.Fatalf("received unexpected stop notification")
default:
}
r.Stop()
r.gracefulStop()
select {
case <-notifier:
default:
hrp/server.go
+8 -5
View File
@@ -163,7 +163,7 @@ func (api *apiHandler) Index(w http.ResponseWriter, r *http.Request) {
http.Error(w, "Not Found", http.StatusNotFound)
return
}
w.Header().Set("Content-Security-Policy", "default-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' camo.githubusercontent.com")
w.Header().Set("Content-Security-Policy", "default-src 'self'; style-src 'self' 'unsafe-inline'; img-src 'self' www.httprunner.com")
fmt.Fprintf(w, "Welcome to httprunner page!")
}
@@ -315,23 +315,26 @@ func (api *apiHandler) Handler() http.Handler {
func (apiHandler) ServeHTTP(http.ResponseWriter, *http.Request) {}
func (b *HRPBoomer) StartServer() {
func (b *HRPBoomer) StartServer(ctx context.Context, addr string) {
h := b.NewAPIHandler()
mux := h.Handler()
server := &http.Server{
Addr: ":9771",
Addr: addr,
Handler: mux,
}
go func() {
<-b.GetCloseChan()
select {
case <-ctx.Done():
case <-b.GetCloseChan():
}
if err := server.Shutdown(context.Background()); err != nil {
log.Fatal("shutdown server:", err)
}
}()
log.Println("Starting HTTP server...")
log.Println(fmt.Sprintf("starting HTTP server (%v), please use the API to control master", server.Addr))
err := server.ListenAndServe()
if err != nil {
if err == http.ErrServerClosed {