aboutsummaryrefslogtreecommitdiffstats
path: root/cmd/swarm
diff options
context:
space:
mode:
Diffstat (limited to 'cmd/swarm')
-rw-r--r--cmd/swarm/swarm-smoke/feed_upload_and_sync.go64
-rw-r--r--cmd/swarm/swarm-smoke/main.go71
-rw-r--r--cmd/swarm/swarm-smoke/upload_and_sync.go160
3 files changed, 233 insertions, 62 deletions
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,24 +17,38 @@
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
cluster string
appName string
scheme string
filesize int
+ syncDelay int
from int
to int
verbosity int
+ timeout int
+ single bool
)
func main() {
@@ -86,13 +100,41 @@ func main() {
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
-}
generated by cgit (git 2.34.1) at 2024-11-14 10:25:15 +0800