From bb724080cac9fa36ec6b638cfd5cf0e54bc23362 Mon Sep 17 00:00:00 2001 From: Elad Date: Tue, 11 Dec 2018 13:51:58 +0530 Subject: cmd/swarm, metrics, swarm/api/client, swarm/storage, swarm/metrics, swarm/api/http: add instrumentation (#18274) --- cmd/swarm/swarm-smoke/feed_upload_and_sync.go | 64 +++++++++-- cmd/swarm/swarm-smoke/main.go | 71 ++++++++++++ cmd/swarm/swarm-smoke/upload_and_sync.go | 160 +++++++++++++++++--------- 3 files changed, 233 insertions(+), 62 deletions(-) (limited to 'cmd/swarm') diff --git a/cmd/swarm/swarm-smoke/feed_upload_and_sync.go b/cmd/swarm/swarm-smoke/feed_upload_and_sync.go index 7ec152826..2c5e3fd23 100644 --- a/cmd/swarm/swarm-smoke/feed_upload_and_sync.go +++ b/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,13 @@ 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/api/client" + "github.com/ethereum/go-ethereum/swarm/spancontext" "github.com/ethereum/go-ethereum/swarm/storage/feed" + "github.com/ethereum/go-ethereum/swarm/testutil" colorable "github.com/mattn/go-colorable" + opentracing "github.com/opentracing/opentracing-go" "github.com/pborman/uuid" cli "gopkg.in/urfave/cli.v1" ) @@ -26,11 +33,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) @@ -204,12 +229,12 @@ func cliFeedUploadAndSync(c *cli.Context) error { log.Info("all endpoints synced random data successfully") // upload test file - log.Info("uploading to " + endpoints[0] + " and syncing") + seed := int(time.Now().UnixNano() / 1e6) + log.Info("feed uploading to "+endpoints[0]+" and syncing", "seed", seed) - f, cleanup := generateRandomFile(filesize * 1000) - defer cleanup() + randomBytes := testutil.RandomBytes(seed, filesize*1000) - hash, err := upload(f, endpoints[0]) + hash, err := upload(&randomBytes, endpoints[0]) if err != nil { return err } @@ -218,7 +243,7 @@ func cliFeedUploadAndSync(c *cli.Context) error { return err } multihashHex := hexutil.Encode(hashBytes) - fileHash, err := digest(f) + fileHash, err := digest(bytes.NewReader(randomBytes)) if err != nil { return err } @@ -284,14 +309,37 @@ func cliFeedUploadAndSync(c *cli.Context) error { } func fetchFeed(topic string, user string, endpoint string, original []byte, ruid string) error { + ctx, sp := spancontext.StartSpan(context.Background(), "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) + + var tn time.Time + 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 := client.GetClientTrace("feed-and-sync - http get", "feed-and-sync", ruid, &tn) + + req = req.WithContext(httptrace.WithClientTrace(ctx, trace)) + transport := http.DefaultTransport + + //transport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} + + tn = time.Now() + 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 { diff --git a/cmd/swarm/swarm-smoke/main.go b/cmd/swarm/swarm-smoke/main.go index 845998dc1..66cecdc5c 100644 --- a/cmd/swarm/swarm-smoke/main.go +++ b/cmd/swarm/swarm-smoke/main.go @@ -17,14 +17,25 @@ 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/swarm/tracing" + "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 @@ -32,9 +43,12 @@ var ( appName string scheme string filesize int + syncDelay int from int to int verbosity int + timeout int + single bool ) func main() { @@ -85,14 +99,42 @@ func main() { Usage: "file size for generated random file in KB", Destination: &filesize, }, + cli.IntFlag{ + Name: "sync-delay", + Value: 5, + Usage: "duration of delay in seconds to wait for content to be synced", + Destination: &syncDelay, + }, cli.IntFlag{ Name: "verbosity", Value: 1, Usage: "verbosity", Destination: &verbosity, }, + cli.IntFlag{ + Name: "timeout", + Value: 120, + Usage: "timeout in seconds after which kill the process", + Destination: &timeout, + }, + cli.BoolFlag{ + Name: "single", + Usage: "whether to fetch content from a single node or from all nodes", + Destination: &single, + }, } + app.Flags = append(app.Flags, []cli.Flag{ + utils.MetricsEnabledFlag, + swarmmetrics.MetricsInfluxDBEndpointFlag, + swarmmetrics.MetricsInfluxDBDatabaseFlag, + swarmmetrics.MetricsInfluxDBUsernameFlag, + swarmmetrics.MetricsInfluxDBPasswordFlag, + swarmmetrics.MetricsInfluxDBHostTagFlag, + }...) + + app.Flags = append(app.Flags, tracing.Flags...) + app.Commands = []cli.Command{ { Name: "upload_and_sync", @@ -111,9 +153,38 @@ func main() { sort.Sort(cli.FlagsByName(app.Flags)) sort.Sort(cli.CommandsByName(app.Commands)) + app.Before = func(ctx *cli.Context) error { + tracing.Setup(ctx) + return nil + } + + 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 +} diff --git a/cmd/swarm/swarm-smoke/upload_and_sync.go b/cmd/swarm/swarm-smoke/upload_and_sync.go index 3843457dc..d605f79a3 100644 --- a/cmd/swarm/swarm-smoke/upload_and_sync.go +++ b/cmd/swarm/swarm-smoke/upload_and_sync.go @@ -18,21 +18,27 @@ package main import ( "bytes" + "context" "crypto/md5" crand "crypto/rand" - "crypto/tls" "errors" "fmt" "io" "io/ioutil" + "math/rand" "net/http" + "net/http/httptrace" "os" - "os/exec" - "strings" "sync" "time" "github.com/ethereum/go-ethereum/log" + "github.com/ethereum/go-ethereum/metrics" + "github.com/ethereum/go-ethereum/swarm/api" + "github.com/ethereum/go-ethereum/swarm/api/client" + "github.com/ethereum/go-ethereum/swarm/spancontext" + "github.com/ethereum/go-ethereum/swarm/testutil" + opentracing "github.com/opentracing/opentracing-go" "github.com/pborman/uuid" cli "gopkg.in/urfave/cli.v1" @@ -40,11 +46,11 @@ import ( 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,22 +64,48 @@ 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) - generateEndpoints(scheme, cluster, appName, from, to) + errc := make(chan error) + go func() { + errc <- uploadAndSync(c) + }() - log.Info("uploading to " + endpoints[0] + " and syncing") + 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) + seed := int(time.Now().UnixNano() / 1e6) + log.Info("uploading to "+endpoints[0]+" and syncing", "seed", seed) - f, cleanup := generateRandomFile(filesize * 1000) - defer cleanup() + randomBytes := testutil.RandomBytes(seed, filesize*1000) - hash, err := upload(f, endpoints[0]) + t1 := time.Now() + hash, err := upload(&randomBytes, 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) + fhash, err := digest(bytes.NewReader(randomBytes)) if err != nil { log.Error(err.Error()) return err @@ -81,23 +113,47 @@ func cliUploadAndSync(c *cli.Context) error { log.Info("uploaded successfully", "hash", hash, "digest", fmt.Sprintf("%x", fhash)) - time.Sleep(3 * time.Second) + time.Sleep(time.Duration(syncDelay) * time.Second) wg := sync.WaitGroup{} - for _, endpoint := range endpoints { + if single { + rand.Seed(time.Now().UTC().UnixNano()) + randIndex := 1 + rand.Intn(len(endpoints)-1) ruid := uuid.New()[:8] wg.Add(1) go func(endpoint string, ruid string) { for { + start := time.Now() err := fetch(hash, endpoint, fhash, ruid) + fetchTime := time.Since(start) if err != nil { continue } + metrics.GetOrRegisterMeter("upload-and-sync.single.fetch-time", nil).Mark(int64(fetchTime)) wg.Done() return } - }(endpoint, ruid) + }(endpoints[randIndex], ruid) + } else { + for _, endpoint := range endpoints { + ruid := uuid.New()[:8] + wg.Add(1) + go func(endpoint string, ruid string) { + for { + start := time.Now() + err := fetch(hash, endpoint, fhash, ruid) + fetchTime := time.Since(start) + if err != nil { + continue + } + + metrics.GetOrRegisterMeter("upload-and-sync.each.fetch-time", nil).Mark(int64(fetchTime)) + wg.Done() + return + } + }(endpoint, ruid) + } } wg.Wait() log.Info("all endpoints synced random file successfully") @@ -107,16 +163,33 @@ func cliUploadAndSync(c *cli.Context) error { // fetch is getting the requested `hash` from the `endpoint` and compares it with the `original` file func fetch(hash string, endpoint string, original []byte, ruid string) error { + ctx, sp := spancontext.StartSpan(context.Background(), "upload-and-sync.fetch") + defer sp.Finish() + log.Trace("sleeping", "ruid", ruid) time.Sleep(3 * time.Second) - log.Trace("http get request", "ruid", ruid, "api", endpoint, "hash", hash) - client := &http.Client{Transport: &http.Transport{ - TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, - }} - res, err := client.Get(endpoint + "/bzz:/" + hash + "/") + + var tn time.Time + reqUri := endpoint + "/bzz:/" + hash + "/" + req, _ := http.NewRequest("GET", reqUri, nil) + + opentracing.GlobalTracer().Inject( + sp.Context(), + opentracing.HTTPHeaders, + opentracing.HTTPHeadersCarrier(req.Header)) + + trace := client.GetClientTrace("upload-and-sync - http get", "upload-and-sync", ruid, &tn) + + req = req.WithContext(httptrace.WithClientTrace(ctx, trace)) + transport := http.DefaultTransport + + //transport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} + + tn = time.Now() + res, err := transport.RoundTrip(req) if err != nil { - log.Warn(err.Error(), "ruid", ruid) + log.Error(err.Error(), "ruid", ruid) return err } log.Trace("http get response", "ruid", ruid, "api", endpoint, "hash", hash, "code", res.StatusCode, "len", res.ContentLength) @@ -147,16 +220,19 @@ func fetch(hash string, endpoint string, original []byte, ruid string) error { } // upload is uploading a file `f` to `endpoint` via the `swarm up` cmd -func upload(f *os.File, endpoint string) (string, error) { - var out bytes.Buffer - cmd := exec.Command("swarm", "--bzzapi", endpoint, "up", f.Name()) - cmd.Stdout = &out - err := cmd.Run() - if err != nil { - return "", err +func upload(dataBytes *[]byte, endpoint string) (string, error) { + swarm := client.NewClient(endpoint) + f := &client.File{ + ReadCloser: ioutil.NopCloser(bytes.NewReader(*dataBytes)), + ManifestEntry: api.ManifestEntry{ + ContentType: "text/plain", + Mode: 0660, + Size: int64(len(*dataBytes)), + }, } - hash := strings.TrimRight(out.String(), "\r\n") - return hash, nil + + // upload data to bzz:// and retrieve the content-addressed manifest hash, hex-encoded. + return swarm.Upload(f, "", false) } func digest(r io.Reader) ([]byte, error) { @@ -179,27 +255,3 @@ func generateRandomData(datasize int) ([]byte, error) { } return b, nil } - -// generateRandomFile is creating a temporary file with the requested byte size -func generateRandomFile(size int) (f *os.File, teardown func()) { - // create a tmp file - tmp, err := ioutil.TempFile("", "swarm-test") - if err != nil { - panic(err) - } - - // callback for tmp file cleanup - teardown = func() { - tmp.Close() - os.Remove(tmp.Name()) - } - - buf := make([]byte, size) - _, err = crand.Read(buf) - if err != nil { - panic(err) - } - ioutil.WriteFile(tmp.Name(), buf, 0755) - - return tmp, teardown -} -- cgit v1.2.3