Merge pull request #21 from interchainio/feature/prometheus

Add support for Prometheus metrics in master/slave mode

cmd/tm-load-test/main.go

@@ -4,12 +4,11 @@ import (
 	"github.com/interchainio/tm-load-test/pkg/loadtest"
 )
 
-const appLongDesc = `Load testing application for Tendermint kvstore with optional master/slave mode.
+const appLongDesc = `
+Load testing application for Tendermint with optional master/slave mode.
 Generates large quantities of arbitrary transactions and submits those 
-transactions to one or more Tendermint endpoints. Assumes the kvstore ABCI app
-to have been deployed on the target Tendermint network. For testing other kinds
-of ABCI apps, you will need to build your own load testing client. See
-https://github.com/interchainio/tm-load-test for details.
+transactions to one or more Tendermint endpoints. By default, it assumes that
+you are running the kvstore ABCI application on your Tendermint network.
 
 To run the application in a similar fashion to tm-bench (STANDALONE mode):
     tm-load-test -c 1 -T 10 -r 1000 -s 250 \

go.mod

@@ -27,6 +27,7 @@ require (
 	github.com/onsi/gomega v1.5.0 // indirect
 	github.com/pelletier/go-toml v1.4.0 // indirect
 	github.com/pkg/errors v0.8.1 // indirect
+	github.com/prometheus/client_golang v1.0.0
 	github.com/prometheus/common v0.6.0 // indirect
 	github.com/prometheus/procfs v0.0.3 // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20190706150252-9beb055b7962 // indirect

pkg/loadtest/cli.go

@@ -88,12 +88,17 @@ func buildCLI(cli *CLIConfig, logger logging.Logger) *cobra.Command {
 				logger.Error(err.Error())
 				os.Exit(1)
 			}
-			slave := NewSlave(&slaveCfg)
+			slave, err := NewSlave(&slaveCfg)
+			if err != nil {
+				logger.Error("Failed to create new slave", "err", err)
+				os.Exit(1)
+			}
 			if err := slave.Run(); err != nil {
 				os.Exit(1)
 			}
 		},
 	}
+	slaveCmd.PersistentFlags().StringVar(&slaveCfg.ID, "id", "", "An optional unique ID for this slave. Will show up in metrics and logs. If not specified, a UUID will be generated.")
 	slaveCmd.PersistentFlags().StringVar(&slaveCfg.MasterAddr, "master", "ws://localhost:26670", "The WebSockets URL on which to find the master node")
 	slaveCmd.PersistentFlags().IntVar(&slaveCfg.MasterConnectTimeout, "connect-timeout", 180, "The maximum number of seconds to keep trying to connect to the master")
 

pkg/loadtest/config.go

@@ -24,10 +24,12 @@ type MasterConfig struct {
 	BindAddr            string `json:"bind_addr"`       // The "host:port" to which to bind the master node to listen for incoming slaves.
 	ExpectSlaves        int    `json:"expect_slaves"`   // The number of slaves to expect before starting the load test.
 	SlaveConnectTimeout int    `json:"connect_timeout"` // The number of seconds to wait for all slaves to connect.
+	ShutdownWait        int    `json:"shutdown_wait"`   // The number of seconds to wait at shutdown (while keeping the HTTP server running - primarily to allow Prometheus to keep polling).
 }
 
 // SlaveConfig is the configuration options specific to a slave node.
 type SlaveConfig struct {
+	ID                   string `json:"id"`              // A unique ID for this slave instance. Will show up in the metrics reported by the master for this slave.
 	MasterAddr           string `json:"master_addr"`     // The address at which to find the master node.
 	MasterConnectTimeout int    `json:"connect_timeout"` // The maximum amount of time, in seconds, to allow for the master to become available.
 }
@@ -102,6 +104,9 @@ func (c Config) ToJSON() string {
 }
 
 func (c SlaveConfig) Validate() error {
+	if len(c.ID) > 0 && !isValidSlaveID(c.ID) {
+		return fmt.Errorf("Invalid slave ID \"%s\": slave IDs can only be lowercase alphanumeric characters", c.ID)
+	}
 	if len(c.MasterAddr) == 0 {
 		return fmt.Errorf("master address must be specified")
 	}

pkg/loadtest/integration_test.go

@@ -2,15 +2,22 @@ package loadtest_test
 
 import (
 	"fmt"
+	"io/ioutil"
 	"net"
+	"net/http"
 	"net/url"
+	"strconv"
+	"strings"
 	"testing"
+	"time"
 
 	"github.com/interchainio/tm-load-test/pkg/loadtest"
 	"github.com/tendermint/tendermint/abci/example/kvstore"
 	rpctest "github.com/tendermint/tendermint/rpc/test"
 )
 
+const totalTxsPerSlave = 50
+
 func TestMasterSlaveHappyPath(t *testing.T) {
 	app := kvstore.NewKVStoreApplication()
 	node := rpctest.StartTendermint(app, rpctest.SuppressStdout, rpctest.RecreateConfig)
@@ -26,6 +33,7 @@ func TestMasterSlaveHappyPath(t *testing.T) {
 		BindAddr:            fmt.Sprintf("localhost:%d", freePort),
 		ExpectSlaves:        2,
 		SlaveConnectTimeout: 10,
+		ShutdownWait:        1,
 	}
 	master := loadtest.NewMaster(&cfg, &masterCfg)
 	masterErr := make(chan error, 1)
@@ -37,18 +45,28 @@ func TestMasterSlaveHappyPath(t *testing.T) {
 		MasterAddr:           fmt.Sprintf("ws://localhost:%d", freePort),
 		MasterConnectTimeout: 10,
 	}
-	slave1 := loadtest.NewSlave(&slaveCfg)
+	slave1, err := loadtest.NewSlave(&slaveCfg)
+	if err != nil {
+		t.Fatal(err)
+	}
 	slave1Err := make(chan error, 1)
 	go func() {
 		slave1Err <- slave1.Run()
 	}()
 
-	slave2 := loadtest.NewSlave(&slaveCfg)
+	slave2, err := loadtest.NewSlave(&slaveCfg)
+	if err != nil {
+		t.Fatal(err)
+	}
 	slave2Err := make(chan error, 1)
 	go func() {
 		slave2Err <- slave2.Run()
 	}()
 
+	slave1Stopped := false
+	slave2Stopped := false
+	metricsTested := false
+
 	for i := 0; i < 3; i++ {
 		select {
 		case err := <-masterErr:
@@ -57,16 +75,58 @@ func TestMasterSlaveHappyPath(t *testing.T) {
 			}
 
 		case err := <-slave1Err:
+			slave1Stopped = true
 			if err != nil {
 				t.Fatal(err)
 			}
 
 		case err := <-slave2Err:
+			slave2Stopped = true
+			if err != nil {
+				t.Fatal(err)
+			}
+
+		case <-time.After(time.Duration(cfg.Time * 2) * time.Second):
+			t.Fatal("Timed out waiting for test to complete")
+		}
+
+		// at this point the master should be waiting a little
+		if slave1Stopped && slave2Stopped && !metricsTested {
+			metricsTested = true
+			// grab the prometheus metrics from the master
+			resp, err := http.Get(fmt.Sprintf("http://localhost:%d/metrics", freePort))
 			if err != nil {
 				t.Fatal(err)
 			}
+			defer resp.Body.Close()
+			if resp.StatusCode != 200 {
+				t.Fatalf("Expected status code 200 from Prometheus endpoint, but got %d", resp.StatusCode)
+			}
+			body, err := ioutil.ReadAll(resp.Body)
+			if err != nil {
+				t.Fatal("Failed to read response body from Prometheus endpoint:", err)
+			}
+			for _, line := range strings.Split(string(body), "\n") {
+				if strings.HasPrefix(line, "tmloadtest_master_total_txs") {
+					parts := strings.Split(line, " ")
+					if len(parts) < 2 {
+						t.Fatal("Invalid Prometheus metrics format")
+					}
+					txCount, err := strconv.Atoi(parts[1])
+					if err != nil {
+						t.Fatal(err)
+					}
+					if txCount != (totalTxsPerSlave * 2) {
+						t.Fatalf("Expected %d transactions to have been recorded by the master, but got %d", totalTxsPerSlave, txCount)
+					}
+				}
+			}
 		}
 	}
+
+	if !metricsTested {
+		t.Fatal("Expected to have tested Prometheus metrics, but did not")
+	}
 }
 
 func getRPCAddress() string {
@@ -85,7 +145,7 @@ func testConfig() loadtest.Config {
 		SendPeriod:        1,
 		Rate:              100,
 		Size:              100,
-		Count:             -1,
+		Count:             totalTxsPerSlave,
 		BroadcastTxMethod: "async",
 		Endpoints:         []string{getRPCAddress()},
 	}

pkg/loadtest/main_test.go

@@ -0,0 +1,13 @@
+package loadtest_test
+
+import (
+	"os"
+	"testing"
+
+	"github.com/sirupsen/logrus"
+)
+
+func TestMain(m *testing.M) {
+	logrus.SetLevel(logrus.DebugLevel)
+	os.Exit(m.Run())
+}