cmd/swarm/swarm-smoke: timeout flag

cmd/swarm/swarm-smoke/feed_upload_and_sync.go

@@ -2,11 +2,13 @@ package main
 
 import (
 	"bytes"
+	"context"
 	"crypto/md5"
 	"fmt"
 	"io"
 	"io/ioutil"
 	"net/http"
+	"net/http/httptrace"
 	"os"
 	"os/exec"
 	"strings"
@@ -16,8 +18,11 @@ import (
 	"github.com/ethereum/go-ethereum/common/hexutil"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/ethereum/go-ethereum/metrics"
+	"github.com/ethereum/go-ethereum/swarm/spancontext"
 	"github.com/ethereum/go-ethereum/swarm/storage/feed"
 	colorable "github.com/mattn/go-colorable"
+	opentracing "github.com/opentracing/opentracing-go"
 	"github.com/pborman/uuid"
 	cli "gopkg.in/urfave/cli.v1"
 )
@@ -26,11 +31,29 @@ const (
 	feedRandomDataLength = 8
 )
 
-// TODO: retrieve with manifest + extract repeating code
 func cliFeedUploadAndSync(c *cli.Context) error {
-
+	metrics.GetOrRegisterCounter("feed-and-sync", nil).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 {
+			metrics.GetOrRegisterCounter("feed-and-sync.fail", nil).Inc(1)
+		}
+		return err
+	case <-time.After(time.Duration(timeout) * time.Second):
+		metrics.GetOrRegisterCounter("feed-and-sync.timeout", nil).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)
@@ -284,14 +307,87 @@ func cliFeedUploadAndSync(c *cli.Context) error {
 }
 
 func fetchFeed(topic string, user string, endpoint string, original []byte, ruid string) error {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	ctx, sp := spancontext.StartSpan(ctx, "feed-and-sync.fetch")
+	defer sp.Finish()
+
 	log.Trace("sleeping", "ruid", ruid)
 	time.Sleep(3 * time.Second)
 
 	log.Trace("http get request (feed)", "ruid", ruid, "api", endpoint, "topic", topic, "user", user)
-	res, err := http.Get(endpoint + "/bzz-feed:/?topic=" + topic + "&user=" + user)
+
+	tn := time.Now()
+	reqUri := endpoint + "/bzz-feed:/?topic=" + topic + "&user=" + user
+	req, _ := http.NewRequest("GET", reqUri, nil)
+
+	opentracing.GlobalTracer().Inject(
+		sp.Context(),
+		opentracing.HTTPHeaders,
+		opentracing.HTTPHeadersCarrier(req.Header))
+
+	trace := &httptrace.ClientTrace{
+		GetConn: func(_ string) {
+			log.Trace("http get request (feed) - GetConn")
+			metrics.GetOrRegisterResettingTimer("feed-and-sync.fetch.clienttrace.getconn", nil).Update(time.Since(tn))
+		},
+		GotConn: func(_ httptrace.GotConnInfo) {
+			log.Trace("http get request (feed) - GotConn")
+			metrics.GetOrRegisterResettingTimer("feed-and-sync.fetch.clienttrace.gotconn", nil).Update(time.Since(tn))
+		},
+		PutIdleConn: func(err error) {
+			log.Trace("http get request (feed) - PutIdleConn", "err", err)
+			metrics.GetOrRegisterResettingTimer("feed-and-sync.fetch.clienttrace.putidle", nil).Update(time.Since(tn))
+		},
+		GotFirstResponseByte: func() {
+			log.Trace("http get request (feed) - GotFirstResponseByte")
+			metrics.GetOrRegisterResettingTimer("feed-and-sync.fetch.clienttrace.firstbyte", nil).Update(time.Since(tn))
+		},
+		Got100Continue: func() {
+			log.Trace("http get request (feed) - Got100Continue")
+			metrics.GetOrRegisterResettingTimer("feed-and-sync.fetch.clienttrace.got100continue", nil).Update(time.Since(tn))
+		},
+		DNSStart: func(_ httptrace.DNSStartInfo) {
+			log.Trace("http get request (feed) - DNSStart")
+			metrics.GetOrRegisterResettingTimer("feed-and-sync.fetch.clienttrace.dnsstart", nil).Update(time.Since(tn))
+		},
+		DNSDone: func(_ httptrace.DNSDoneInfo) {
+			log.Trace("http get request (feed) - DNSDone")
+			metrics.GetOrRegisterResettingTimer("feed-and-sync.fetch.clienttrace.dnsdone", nil).Update(time.Since(tn))
+		},
+		ConnectStart: func(network, addr string) {
+			log.Trace("http get request (feed) - ConnectStart", "network", network, "addr", addr)
+			metrics.GetOrRegisterResettingTimer("feed-and-sync.fetch.clienttrace.connectstart", nil).Update(time.Since(tn))
+		},
+		ConnectDone: func(network, addr string, err error) {
+			log.Trace("http get request (feed) - ConnectDone", "network", network, "addr", addr, "err", err)
+			metrics.GetOrRegisterResettingTimer("feed-and-sync.fetch.clienttrace.connectdone", nil).Update(time.Since(tn))
+		},
+		WroteHeaders: func() {
+			log.Trace("http get request (feed) - WroteHeaders(request)")
+			metrics.GetOrRegisterResettingTimer("feed-and-sync.fetch.clienttrace.wroteheaders", nil).Update(time.Since(tn))
+		},
+		Wait100Continue: func() {
+			log.Trace("http get request (feed) - Wait100Continue")
+			metrics.GetOrRegisterResettingTimer("feed-and-sync.fetch.clienttrace.wait100continue", nil).Update(time.Since(tn))
+		},
+
+		WroteRequest: func(_ httptrace.WroteRequestInfo) {
+			log.Trace("http get request (feed) - WroteRequest")
+			metrics.GetOrRegisterResettingTimer("feed-and-sync.fetch.clienttrace.wroterequest", nil).Update(time.Since(tn))
+		},
+	}
+	req = req.WithContext(httptrace.WithClientTrace(req.Context(), trace))
+	transport := http.DefaultTransport
+
+	//transport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
+
+	res, err := transport.RoundTrip(req)
 	if err != nil {
+		log.Error(err.Error(), "ruid", ruid)
 		return err
 	}
+
 	log.Trace("http get response (feed)", "ruid", ruid, "api", endpoint, "topic", topic, "user", user, "code", res.StatusCode, "len", res.ContentLength)
 
 	if res.StatusCode != 200 {

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,7 @@ var (
 	from             int
 	to               int
 	verbosity        int
+	timeout          int
 )
 
 func main() {
@@ -91,8 +102,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 +136,33 @@ func main() {
 
 	sort.Sort(cli.FlagsByName(app.Flags))
 	sort.Sort(cli.CommandsByName(app.Commands))
+	app.After = func(ctx *cli.Context) error {
+		return emitMetrics(ctx)
+	}
 
 	err := app.Run(os.Args)
 	if err != nil {
 		log.Error(err.Error())
+
 		os.Exit(1)
 	}
 }
+
+func emitMetrics(ctx *cli.Context) error {
+	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)
+		)
+		return influxdb.InfluxDBWithTagsOnce(gethmetrics.DefaultRegistry, endpoint, database, username, password, "swarm-smoke.", map[string]string{
+			"host":     hosttag,
+			"version":  gitCommit,
+			"filesize": fmt.Sprintf("%v", filesize),
+		})
+	}
+
+	return nil
+}