cmd/swarm/swarm-smoke: timeout flag

cmd/swarm/swarm-smoke/feed_upload_and_sync.go

@@ -26,11 +26,29 @@ const (
 	feedRandomDataLength = 8
 )
 
-// TODO: retrieve with manifest + extract repeating code
 func cliFeedUploadAndSync(c *cli.Context) error {
-
+	feedUploadAndSyncCount.Inc(1)
 	log.Root().SetHandler(log.CallerFileHandler(log.LvlFilterHandler(log.Lvl(verbosity), log.StreamHandler(colorable.NewColorableStderr(), log.TerminalFormat(true)))))
 
+	errc := make(chan error)
+	go func() {
+		errc <- feedUploadAndSync(c)
+	}()
+
+	select {
+	case err := <-errc:
+		if err != nil {
+			feedUploadAndSyncFailCount.Inc(1)
+		}
+		return err
+	case <-time.After(time.Duration(timeout) * time.Second):
+		feedUploadAndSyncTimeout.Inc(1)
+		return fmt.Errorf("timeout after %v sec", timeout)
+	}
+}
+
+// TODO: retrieve with manifest + extract repeating code
+func feedUploadAndSync(c *cli.Context) error {
 	defer func(now time.Time) { log.Info("total time", "time", time.Since(now), "size (kb)", filesize) }(time.Now())
 
 	generateEndpoints(scheme, cluster, appName, from, to)

cmd/swarm/swarm-smoke/main.go

@@ -17,14 +17,24 @@
 package main
 
 import (
+	"fmt"
 	"os"
 	"sort"
 
+	"github.com/ethereum/go-ethereum/cmd/utils"
+	gethmetrics "github.com/ethereum/go-ethereum/metrics"
+	"github.com/ethereum/go-ethereum/metrics/influxdb"
+	swarmmetrics "github.com/ethereum/go-ethereum/swarm/metrics"
+
 	"github.com/ethereum/go-ethereum/log"
 
 	cli "gopkg.in/urfave/cli.v1"
 )
 
+var (
+	gitCommit string // Git SHA1 commit hash of the release (set via linker flags)
+)
+
 var (
 	endpoints        []string
 	includeLocalhost bool
@@ -35,6 +45,12 @@ var (
 	from             int
 	to               int
 	verbosity        int
+	timeout          int
+)
+var (
+	feedUploadAndSyncCount     = gethmetrics.NewRegisteredCounter("feed-and-sync", nil)
+	feedUploadAndSyncFailCount = gethmetrics.NewRegisteredCounter("feed-and-sync.fail", nil)
+	feedUploadAndSyncTimeout   = gethmetrics.NewRegisteredCounter("feed-and-sync.timeout", nil)
 )
 
 func main() {
@@ -91,8 +107,23 @@ func main() {
 			Usage:       "verbosity",
 			Destination: &verbosity,
 		},
+		cli.IntFlag{
+			Name:        "timeout",
+			Value:       120,
+			Usage:       "timeout in seconds after which kill the process",
+			Destination: &timeout,
+		},
 	}
 
+	app.Flags = append(app.Flags, []cli.Flag{
+		utils.MetricsEnabledFlag,
+		swarmmetrics.MetricsInfluxDBEndpointFlag,
+		swarmmetrics.MetricsInfluxDBDatabaseFlag,
+		swarmmetrics.MetricsInfluxDBUsernameFlag,
+		swarmmetrics.MetricsInfluxDBPasswordFlag,
+		swarmmetrics.MetricsInfluxDBHostTagFlag,
+	}...)
+
 	app.Commands = []cli.Command{
 		{
 			Name:    "upload_and_sync",
@@ -110,10 +141,36 @@ func main() {
 
 	sort.Sort(cli.FlagsByName(app.Flags))
 	sort.Sort(cli.CommandsByName(app.Commands))
+	app.Before = func(ctx *cli.Context) error {
+		return nil
+	}
+	app.After = func(ctx *cli.Context) error {
+		emitMetrics(ctx)
+		return nil
+	}
 
 	err := app.Run(os.Args)
 	if err != nil {
 		log.Error(err.Error())
+
 		os.Exit(1)
 	}
 }
+
+func emitMetrics(ctx *cli.Context) {
+	if gethmetrics.Enabled {
+		var (
+			endpoint = ctx.GlobalString(swarmmetrics.MetricsInfluxDBEndpointFlag.Name)
+			database = ctx.GlobalString(swarmmetrics.MetricsInfluxDBDatabaseFlag.Name)
+			username = ctx.GlobalString(swarmmetrics.MetricsInfluxDBUsernameFlag.Name)
+			password = ctx.GlobalString(swarmmetrics.MetricsInfluxDBPasswordFlag.Name)
+			hosttag  = ctx.GlobalString(swarmmetrics.MetricsInfluxDBHostTagFlag.Name)
+		)
+
+		influxdb.InfluxDBWithTagsOnce(gethmetrics.DefaultRegistry, endpoint, database, username, password, "swarm-smoke.", map[string]string{
+			"host":     hosttag,
+			"version":  gitCommit,
+			"filesize": fmt.Sprintf("%v", filesize),
+		})
+	}
+}

cmd/swarm/swarm-smoke/upload_and_sync.go

@@ -33,18 +33,19 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/metrics"
 	"github.com/pborman/uuid"
 
 	cli "gopkg.in/urfave/cli.v1"
 )
 
 func generateEndpoints(scheme string, cluster string, app string, from int, to int) {
 	if cluster == "prod" {
-		for port := from; port <= to; port++ {
+		for port := from; port < to; port++ {
 			endpoints = append(endpoints, fmt.Sprintf("%s://%v.swarm-gateways.net", scheme, port))
 		}
 	} else {
-		for port := from; port <= to; port++ {
+		for port := from; port < to; port++ {
 			endpoints = append(endpoints, fmt.Sprintf("%s://%s-%v-%s.stg.swarm-gateways.net", scheme, app, port, cluster))
 		}
 	}
@@ -58,7 +59,32 @@ func cliUploadAndSync(c *cli.Context) error {
 	log.PrintOrigins(true)
 	log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(verbosity), log.StreamHandler(os.Stdout, log.TerminalFormat(true))))
 
-	defer func(now time.Time) { log.Info("total time", "time", time.Since(now), "kb", filesize) }(time.Now())
+	metrics.GetOrRegisterCounter("upload-and-sync", nil).Inc(1)
+
+	errc := make(chan error)
+	go func() {
+		errc <- uploadAndSync(c)
+	}()
+
+	select {
+	case err := <-errc:
+		if err != nil {
+			metrics.GetOrRegisterCounter("upload-and-sync.fail", nil).Inc(1)
+		}
+		return err
+	case <-time.After(time.Duration(timeout) * time.Second):
+		metrics.GetOrRegisterCounter("upload-and-sync.timeout", nil).Inc(1)
+		return fmt.Errorf("timeout after %v sec", timeout)
+	}
+}
+
+func uploadAndSync(c *cli.Context) error {
+	defer func(now time.Time) {
+		totalTime := time.Since(now)
+
+		log.Info("total time", "time", totalTime, "kb", filesize)
+		metrics.GetOrRegisterCounter("upload-and-sync.total-time", nil).Inc(int64(totalTime))
+	}(time.Now())
 
 	generateEndpoints(scheme, cluster, appName, from, to)
 
@@ -67,11 +93,13 @@ func cliUploadAndSync(c *cli.Context) error {
 	f, cleanup := generateRandomFile(filesize * 1000)
 	defer cleanup()
 
+	t1 := time.Now()
 	hash, err := upload(f, endpoints[0])
 	if err != nil {
 		log.Error(err.Error())
 		return err
 	}
+	metrics.GetOrRegisterCounter("upload-and-sync.upload-time", nil).Inc(int64(time.Since(t1)))
 
 	fhash, err := digest(f)
 	if err != nil {

metrics/influxdb/influxdb.go

@@ -58,6 +58,34 @@ func InfluxDBWithTags(r metrics.Registry, d time.Duration, url, database, userna
 	rep.run()
 }
 
+// InfluxDBWithTagsOnce runs once an InfluxDB reporter and post the given metrics.Registry with the specified tags
+func InfluxDBWithTagsOnce(r metrics.Registry, url, database, username, password, namespace string, tags map[string]string) {
+	u, err := uurl.Parse(url)
+	if err != nil {
+		log.Warn("Unable to parse InfluxDB", "url", url, "err", err)
+		return
+	}
+
+	rep := &reporter{
+		reg:       r,
+		url:       *u,
+		database:  database,
+		username:  username,
+		password:  password,
+		namespace: namespace,
+		tags:      tags,
+		cache:     make(map[string]int64),
+	}
+	if err := rep.makeClient(); err != nil {
+		log.Warn("Unable to make InfluxDB client", "err", err)
+		return
+	}
+
+	if err := rep.send(); err != nil {
+		log.Warn("Unable to send to InfluxDB", "err", err)
+	}
+}
+
 func (r *reporter) makeClient() (err error) {
 	r.client, err = client.NewClient(client.Config{
 		URL:      r.url,

swarm/metrics/flags.go

@@ -27,26 +27,26 @@ import (
 )
 
 var (
-	metricsEnableInfluxDBExportFlag = cli.BoolFlag{
+	MetricsEnableInfluxDBExportFlag = cli.BoolFlag{
 		Name:  "metrics.influxdb.export",
 		Usage: "Enable metrics export/push to an external InfluxDB database",
 	}
-	metricsInfluxDBEndpointFlag = cli.StringFlag{
+	MetricsInfluxDBEndpointFlag = cli.StringFlag{
 		Name:  "metrics.influxdb.endpoint",
 		Usage: "Metrics InfluxDB endpoint",
 		Value: "http://127.0.0.1:8086",
 	}
-	metricsInfluxDBDatabaseFlag = cli.StringFlag{
+	MetricsInfluxDBDatabaseFlag = cli.StringFlag{
 		Name:  "metrics.influxdb.database",
 		Usage: "Metrics InfluxDB database",
 		Value: "metrics",
 	}
-	metricsInfluxDBUsernameFlag = cli.StringFlag{
+	MetricsInfluxDBUsernameFlag = cli.StringFlag{
 		Name:  "metrics.influxdb.username",
 		Usage: "Metrics InfluxDB username",
 		Value: "",
 	}
-	metricsInfluxDBPasswordFlag = cli.StringFlag{
+	MetricsInfluxDBPasswordFlag = cli.StringFlag{
 		Name:  "metrics.influxdb.password",
 		Usage: "Metrics InfluxDB password",
 		Value: "",
@@ -55,7 +55,7 @@ var (
 	// It is used so that we can group all nodes and average a measurement across all of them, but also so
 	// that we can select a specific node and inspect its measurements.
 	// https://docs.influxdata.com/influxdb/v1.4/concepts/key_concepts/#tag-key
-	metricsInfluxDBHostTagFlag = cli.StringFlag{
+	MetricsInfluxDBHostTagFlag = cli.StringFlag{
 		Name:  "metrics.influxdb.host.tag",
 		Usage: "Metrics InfluxDB `host` tag attached to all measurements",
 		Value: "localhost",
@@ -65,20 +65,24 @@ var (
 // Flags holds all command-line flags required for metrics collection.
 var Flags = []cli.Flag{
 	utils.MetricsEnabledFlag,
-	metricsEnableInfluxDBExportFlag,
-	metricsInfluxDBEndpointFlag, metricsInfluxDBDatabaseFlag, metricsInfluxDBUsernameFlag, metricsInfluxDBPasswordFlag, metricsInfluxDBHostTagFlag,
+	MetricsEnableInfluxDBExportFlag,
+	MetricsInfluxDBEndpointFlag,
+	MetricsInfluxDBDatabaseFlag,
+	MetricsInfluxDBUsernameFlag,
+	MetricsInfluxDBPasswordFlag,
+	MetricsInfluxDBHostTagFlag,
 }
 
 func Setup(ctx *cli.Context) {
 	if gethmetrics.Enabled {
 		log.Info("Enabling swarm metrics collection")
 		var (
-			enableExport = ctx.GlobalBool(metricsEnableInfluxDBExportFlag.Name)
-			endpoint     = ctx.GlobalString(metricsInfluxDBEndpointFlag.Name)
-			database     = ctx.GlobalString(metricsInfluxDBDatabaseFlag.Name)
-			username     = ctx.GlobalString(metricsInfluxDBUsernameFlag.Name)
-			password     = ctx.GlobalString(metricsInfluxDBPasswordFlag.Name)
-			hosttag      = ctx.GlobalString(metricsInfluxDBHostTagFlag.Name)
+			enableExport = ctx.GlobalBool(MetricsEnableInfluxDBExportFlag.Name)
+			endpoint     = ctx.GlobalString(MetricsInfluxDBEndpointFlag.Name)
+			database     = ctx.GlobalString(MetricsInfluxDBDatabaseFlag.Name)
+			username     = ctx.GlobalString(MetricsInfluxDBUsernameFlag.Name)
+			password     = ctx.GlobalString(MetricsInfluxDBPasswordFlag.Name)
+			hosttag      = ctx.GlobalString(MetricsInfluxDBHostTagFlag.Name)
 		)
 
 		// Start system runtime metrics collection