cmd/swarm/swarm-smoke: refactor endpoints, remove endpoints generation

This commit is contained in:
Anton Evangelatov 2019-02-04 14:44:09 +01:00
parent 245f3146c2
commit 1e55c043e4
6 changed files with 67 additions and 109 deletions

View file

@ -29,9 +29,7 @@ const (
func feedUploadAndSync(c *cli.Context) error { func feedUploadAndSync(c *cli.Context) error {
defer func(now time.Time) { log.Info("total time", "time", time.Since(now), "size (kb)", filesize) }(time.Now()) defer func(now time.Time) { log.Info("total time", "time", time.Since(now), "size (kb)", filesize) }(time.Now())
generateEndpoints(scheme, cluster, appName, from, to) log.Info("generating and uploading feeds to " + httpEndpoint(hosts[0]) + " and syncing")
log.Info("generating and uploading feeds to " + endpoints[0] + " and syncing")
// create a random private key to sign updates with and derive the address // create a random private key to sign updates with and derive the address
pkFile, err := ioutil.TempFile("", "swarm-feed-smoke-test") pkFile, err := ioutil.TempFile("", "swarm-feed-smoke-test")
@ -85,7 +83,7 @@ func feedUploadAndSync(c *cli.Context) error {
// create feed manifest, topic only // create feed manifest, topic only
var out bytes.Buffer var out bytes.Buffer
cmd := exec.Command("swarm", "--bzzapi", endpoints[0], "feed", "create", "--topic", topicHex, "--user", userHex) cmd := exec.Command("swarm", "--bzzapi", httpEndpoint(hosts[0]), "feed", "create", "--topic", topicHex, "--user", userHex)
cmd.Stdout = &out cmd.Stdout = &out
log.Debug("create feed manifest topic cmd", "cmd", cmd) log.Debug("create feed manifest topic cmd", "cmd", cmd)
err = cmd.Run() err = cmd.Run()
@ -100,7 +98,7 @@ func feedUploadAndSync(c *cli.Context) error {
out.Reset() out.Reset()
// create feed manifest, subtopic only // create feed manifest, subtopic only
cmd = exec.Command("swarm", "--bzzapi", endpoints[0], "feed", "create", "--name", subTopicHex, "--user", userHex) cmd = exec.Command("swarm", "--bzzapi", httpEndpoint(hosts[0]), "feed", "create", "--name", subTopicHex, "--user", userHex)
cmd.Stdout = &out cmd.Stdout = &out
log.Debug("create feed manifest subtopic cmd", "cmd", cmd) log.Debug("create feed manifest subtopic cmd", "cmd", cmd)
err = cmd.Run() err = cmd.Run()
@ -115,7 +113,7 @@ func feedUploadAndSync(c *cli.Context) error {
out.Reset() out.Reset()
// create feed manifest, merged topic // create feed manifest, merged topic
cmd = exec.Command("swarm", "--bzzapi", endpoints[0], "feed", "create", "--topic", topicHex, "--name", subTopicHex, "--user", userHex) cmd = exec.Command("swarm", "--bzzapi", httpEndpoint(hosts[0]), "feed", "create", "--topic", topicHex, "--name", subTopicHex, "--user", userHex)
cmd.Stdout = &out cmd.Stdout = &out
log.Debug("create feed manifest mergetopic cmd", "cmd", cmd) log.Debug("create feed manifest mergetopic cmd", "cmd", cmd)
err = cmd.Run() err = cmd.Run()
@ -141,7 +139,7 @@ func feedUploadAndSync(c *cli.Context) error {
dataHex := hexutil.Encode(data) dataHex := hexutil.Encode(data)
// update with topic // update with topic
cmd = exec.Command("swarm", "--bzzaccount", pkFile.Name(), "--bzzapi", endpoints[0], "feed", "update", "--topic", topicHex, dataHex) cmd = exec.Command("swarm", "--bzzaccount", pkFile.Name(), "--bzzapi", httpEndpoint(hosts[0]), "feed", "update", "--topic", topicHex, dataHex)
cmd.Stdout = &out cmd.Stdout = &out
log.Debug("update feed manifest topic cmd", "cmd", cmd) log.Debug("update feed manifest topic cmd", "cmd", cmd)
err = cmd.Run() err = cmd.Run()
@ -152,7 +150,7 @@ func feedUploadAndSync(c *cli.Context) error {
out.Reset() out.Reset()
// update with subtopic // update with subtopic
cmd = exec.Command("swarm", "--bzzaccount", pkFile.Name(), "--bzzapi", endpoints[0], "feed", "update", "--name", subTopicHex, dataHex) cmd = exec.Command("swarm", "--bzzaccount", pkFile.Name(), "--bzzapi", httpEndpoint(hosts[0]), "feed", "update", "--name", subTopicHex, dataHex)
cmd.Stdout = &out cmd.Stdout = &out
log.Debug("update feed manifest subtopic cmd", "cmd", cmd) log.Debug("update feed manifest subtopic cmd", "cmd", cmd)
err = cmd.Run() err = cmd.Run()
@ -163,7 +161,7 @@ func feedUploadAndSync(c *cli.Context) error {
out.Reset() out.Reset()
// update with merged topic // update with merged topic
cmd = exec.Command("swarm", "--bzzaccount", pkFile.Name(), "--bzzapi", endpoints[0], "feed", "update", "--topic", topicHex, "--name", subTopicHex, dataHex) cmd = exec.Command("swarm", "--bzzaccount", pkFile.Name(), "--bzzapi", httpEndpoint(hosts[0]), "feed", "update", "--topic", topicHex, "--name", subTopicHex, dataHex)
cmd.Stdout = &out cmd.Stdout = &out
log.Debug("update feed manifest merged topic cmd", "cmd", cmd) log.Debug("update feed manifest merged topic cmd", "cmd", cmd)
err = cmd.Run() err = cmd.Run()
@ -177,14 +175,14 @@ func feedUploadAndSync(c *cli.Context) error {
// retrieve the data // retrieve the data
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
for _, endpoint := range endpoints { for _, host := range hosts {
// raw retrieve, topic only // raw retrieve, topic only
for _, hex := range []string{topicHex, subTopicOnlyHex, mergedSubTopicHex} { for _, hex := range []string{topicHex, subTopicOnlyHex, mergedSubTopicHex} {
wg.Add(1) wg.Add(1)
ruid := uuid.New()[:8] ruid := uuid.New()[:8]
go func(hex string, endpoint string, ruid string) { go func(hex string, endpoint string, ruid string) {
for { for {
err := fetchFeed(hex, userHex, endpoint, dataHash, ruid) err := fetchFeed(hex, userHex, httpEndpoint(host), dataHash, ruid)
if err != nil { if err != nil {
continue continue
} }
@ -192,8 +190,7 @@ func feedUploadAndSync(c *cli.Context) error {
wg.Done() wg.Done()
return return
} }
}(hex, endpoint, ruid) }(hex, httpEndpoint(host), ruid)
} }
} }
wg.Wait() wg.Wait()
@ -201,12 +198,12 @@ func feedUploadAndSync(c *cli.Context) error {
// upload test file // upload test file
seed := int(time.Now().UnixNano() / 1e6) seed := int(time.Now().UnixNano() / 1e6)
log.Info("feed uploading to "+endpoints[0]+" and syncing", "seed", seed) log.Info("feed uploading to "+httpEndpoint(hosts[0])+" and syncing", "seed", seed)
h = md5.New() h = md5.New()
r := io.TeeReader(io.LimitReader(crand.Reader, int64(filesize*1000)), h) r := io.TeeReader(io.LimitReader(crand.Reader, int64(filesize*1000)), h)
hash, err := upload(r, filesize*1000, endpoints[0]) hash, err := upload(r, filesize*1000, httpEndpoint(hosts[0]))
if err != nil { if err != nil {
return err return err
} }
@ -220,7 +217,7 @@ func feedUploadAndSync(c *cli.Context) error {
log.Info("uploaded successfully", "hash", hash, "digest", fmt.Sprintf("%x", fileHash)) log.Info("uploaded successfully", "hash", hash, "digest", fmt.Sprintf("%x", fileHash))
// update file with topic // update file with topic
cmd = exec.Command("swarm", "--bzzaccount", pkFile.Name(), "--bzzapi", endpoints[0], "feed", "update", "--topic", topicHex, multihashHex) cmd = exec.Command("swarm", "--bzzaccount", pkFile.Name(), "--bzzapi", httpEndpoint(hosts[0]), "feed", "update", "--topic", topicHex, multihashHex)
cmd.Stdout = &out cmd.Stdout = &out
err = cmd.Run() err = cmd.Run()
if err != nil { if err != nil {
@ -230,7 +227,7 @@ func feedUploadAndSync(c *cli.Context) error {
out.Reset() out.Reset()
// update file with subtopic // update file with subtopic
cmd = exec.Command("swarm", "--bzzaccount", pkFile.Name(), "--bzzapi", endpoints[0], "feed", "update", "--name", subTopicHex, multihashHex) cmd = exec.Command("swarm", "--bzzaccount", pkFile.Name(), "--bzzapi", httpEndpoint(hosts[0]), "feed", "update", "--name", subTopicHex, multihashHex)
cmd.Stdout = &out cmd.Stdout = &out
err = cmd.Run() err = cmd.Run()
if err != nil { if err != nil {
@ -240,7 +237,7 @@ func feedUploadAndSync(c *cli.Context) error {
out.Reset() out.Reset()
// update file with merged topic // update file with merged topic
cmd = exec.Command("swarm", "--bzzaccount", pkFile.Name(), "--bzzapi", endpoints[0], "feed", "update", "--topic", topicHex, "--name", subTopicHex, multihashHex) cmd = exec.Command("swarm", "--bzzaccount", pkFile.Name(), "--bzzapi", httpEndpoint(hosts[0]), "feed", "update", "--topic", topicHex, "--name", subTopicHex, multihashHex)
cmd.Stdout = &out cmd.Stdout = &out
err = cmd.Run() err = cmd.Run()
if err != nil { if err != nil {
@ -251,7 +248,7 @@ func feedUploadAndSync(c *cli.Context) error {
time.Sleep(3 * time.Second) time.Sleep(3 * time.Second)
for _, endpoint := range endpoints { for _, host := range hosts {
// manifest retrieve, topic only // manifest retrieve, topic only
for _, url := range []string{manifestWithTopic, manifestWithSubTopic, manifestWithMergedTopic} { for _, url := range []string{manifestWithTopic, manifestWithSubTopic, manifestWithMergedTopic} {
@ -267,7 +264,7 @@ func feedUploadAndSync(c *cli.Context) error {
wg.Done() wg.Done()
return return
} }
}(url, endpoint, ruid) }(url, httpEndpoint(host), ruid)
} }
} }

View file

@ -37,15 +37,12 @@ var (
) )
var ( var (
endpoints []string allhosts string
includeLocalhost bool hosts []string
cluster string
appName string
scheme string
filesize int filesize int
syncDelay int syncDelay int
from int httpPort int
to int wsPort int
verbosity int verbosity int
timeout int timeout int
single bool single bool
@ -59,39 +56,22 @@ func main() {
app.Flags = []cli.Flag{ app.Flags = []cli.Flag{
cli.StringFlag{ cli.StringFlag{
Name: "cluster-endpoint", Name: "hosts",
Value: "prod", Value: "",
Usage: "cluster to point to (prod or a given namespace)", Usage: "comma-separated list of swarm hosts",
Destination: &cluster, Destination: &allhosts,
},
cli.StringFlag{
Name: "app",
Value: "swarm",
Usage: "application to point to (swarm or swarm-private)",
Destination: &appName,
}, },
cli.IntFlag{ cli.IntFlag{
Name: "cluster-from", Name: "http-port",
Value: 8501, Value: 80,
Usage: "swarm node (from)", Usage: "http port",
Destination: &from, Destination: &httpPort,
}, },
cli.IntFlag{ cli.IntFlag{
Name: "cluster-to", Name: "ws-port",
Value: 8512, Value: 8546,
Usage: "swarm node (to)", Usage: "ws port",
Destination: &to, Destination: &wsPort,
},
cli.StringFlag{
Name: "cluster-scheme",
Value: "http",
Usage: "http or https",
Destination: &scheme,
},
cli.BoolFlag{
Name: "include-localhost",
Usage: "whether to include localhost:8500 as an endpoint",
Destination: &includeLocalhost,
}, },
cli.IntFlag{ cli.IntFlag{
Name: "filesize", Name: "filesize",

View file

@ -43,9 +43,8 @@ type uploadResult struct {
} }
func slidingWindow(c *cli.Context) error { func slidingWindow(c *cli.Context) error {
generateEndpoints(scheme, cluster, appName, from, to)
hashes := []uploadResult{} //swarm hashes of the uploads hashes := []uploadResult{} //swarm hashes of the uploads
nodes := to - from nodes := len(hosts)
const iterationTimeout = 30 * time.Second const iterationTimeout = 30 * time.Second
log.Info("sliding window test started", "nodes", nodes, "filesize(kb)", filesize, "timeout", timeout) log.Info("sliding window test started", "nodes", nodes, "filesize(kb)", filesize, "timeout", timeout)
uploadedBytes := 0 uploadedBytes := 0
@ -54,13 +53,13 @@ func slidingWindow(c *cli.Context) error {
outer: outer:
for { for {
log.Info("uploading to "+endpoints[0]+" and syncing", "seed", seed) log.Info("uploading to "+httpEndpoint(hosts[0])+" and syncing", "seed", seed)
h := md5.New() h := md5.New()
r := io.TeeReader(io.LimitReader(crand.Reader, int64(filesize*1000)), h) r := io.TeeReader(io.LimitReader(crand.Reader, int64(filesize*1000)), h)
t1 := time.Now() t1 := time.Now()
hash, err := upload(r, filesize*1000, endpoints[0]) hash, err := upload(r, filesize*1000, httpEndpoint(hosts[0]))
if err != nil { if err != nil {
log.Error(err.Error()) log.Error(err.Error())
return err return err
@ -88,10 +87,10 @@ outer:
metrics.GetOrRegisterCounter("sliding-window.single.error", nil).Inc(1) metrics.GetOrRegisterCounter("sliding-window.single.error", nil).Inc(1)
break inner break inner
default: default:
randIndex := 1 + rand.Intn(len(endpoints)-1) idx := 1 + rand.Intn(len(hosts)-1)
ruid := uuid.New()[:8] ruid := uuid.New()[:8]
start := time.Now() start := time.Now()
err := fetch(v.hash, endpoints[randIndex], v.digest, ruid) err := fetch(v.hash, httpEndpoint(hosts[idx]), v.digest, ruid)
if err != nil { if err != nil {
continue inner continue inner
} }

View file

@ -33,16 +33,15 @@ import (
) )
func uploadAndSync(c *cli.Context) error { func uploadAndSync(c *cli.Context) error {
generateEndpoints(scheme, cluster, appName, from, to)
seed := int(time.Now().UnixNano() / 1e6) seed := int(time.Now().UnixNano() / 1e6)
log.Info("uploading to "+endpoints[0]+" and syncing", "seed", seed) log.Info("uploading to "+httpEndpoint(hosts[0])+" and syncing", "seed", seed)
h := md5.New() h := md5.New()
r := io.TeeReader(io.LimitReader(crand.Reader, int64(filesize*1000)), h) r := io.TeeReader(io.LimitReader(crand.Reader, int64(filesize*1000)), h)
t1 := time.Now() t1 := time.Now()
hash, err := upload(r, filesize*1000, endpoints[0]) hash, err := upload(r, filesize*1000, httpEndpoint(hosts[0]))
if err != nil { if err != nil {
log.Error(err.Error()) log.Error(err.Error())
return err return err
@ -58,7 +57,7 @@ func uploadAndSync(c *cli.Context) error {
wg := sync.WaitGroup{} wg := sync.WaitGroup{}
if single { if single {
rand.Seed(time.Now().UTC().UnixNano()) rand.Seed(time.Now().UTC().UnixNano())
randIndex := 1 + rand.Intn(len(endpoints)-1) randIndex := 1 + rand.Intn(len(hosts)-1)
ruid := uuid.New()[:8] ruid := uuid.New()[:8]
wg.Add(1) wg.Add(1)
go func(endpoint string, ruid string) { go func(endpoint string, ruid string) {
@ -73,9 +72,9 @@ func uploadAndSync(c *cli.Context) error {
wg.Done() wg.Done()
return return
} }
}(endpoints[randIndex], ruid) }(httpEndpoint(hosts[randIndex]), ruid)
} else { } else {
for _, endpoint := range endpoints[1:] { for _, endpoint := range hosts[1:] {
ruid := uuid.New()[:8] ruid := uuid.New()[:8]
wg.Add(1) wg.Add(1)
go func(endpoint string, ruid string) { go func(endpoint string, ruid string) {
@ -90,11 +89,11 @@ func uploadAndSync(c *cli.Context) error {
wg.Done() wg.Done()
return return
} }
}(endpoint, ruid) }(httpEndpoint(endpoint), ruid)
} }
} }
wg.Wait() wg.Wait()
log.Info("all endpoints synced random file successfully") log.Info("all hosts synced random file successfully")
return nil return nil
} }

View file

@ -30,15 +30,18 @@ import (
) )
func uploadSpeed(c *cli.Context) error { func uploadSpeed(c *cli.Context) error {
endpoint := generateEndpoint(scheme, cluster, appName, from) if len(hosts) < 2 {
log.Crit("less than 2 hosts")
}
seed := int(time.Now().UnixNano() / 1e6) seed := int(time.Now().UnixNano() / 1e6)
log.Info("uploading to "+endpoint, "seed", seed) log.Info("uploading to "+hosts[0], "seed", seed)
h := md5.New() h := md5.New()
r := io.TeeReader(io.LimitReader(crand.Reader, int64(filesize*1000)), h) r := io.TeeReader(io.LimitReader(crand.Reader, int64(filesize*1000)), h)
t1 := time.Now() t1 := time.Now()
hash, err := upload(r, filesize*1000, endpoint) hash, err := upload(r, filesize*1000, hosts[0])
if err != nil { if err != nil {
log.Error(err.Error()) log.Error(err.Error())
return err return err

View file

@ -28,6 +28,7 @@ import (
"net/http" "net/http"
"net/http/httptrace" "net/http/httptrace"
"os" "os"
"strings"
"time" "time"
"github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/log"
@ -43,11 +44,21 @@ var (
commandName = "" commandName = ""
) )
func httpEndpoint(host string) string {
return fmt.Sprintf("http://%s:%d", host, httpPort)
}
func wsEndpoint(host string) string {
return fmt.Sprintf("ws://%s:%d", host, wsPort)
}
func wrapCliCommand(name string, killOnTimeout bool, command func(*cli.Context) error) func(*cli.Context) error { func wrapCliCommand(name string, killOnTimeout bool, command func(*cli.Context) error) func(*cli.Context) error {
return func(ctx *cli.Context) error { return func(ctx *cli.Context) error {
log.PrintOrigins(true) log.PrintOrigins(true)
log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(verbosity), log.StreamHandler(os.Stdout, log.TerminalFormat(false)))) log.Root().SetHandler(log.LvlFilterHandler(log.Lvl(verbosity), log.StreamHandler(os.Stdout, log.TerminalFormat(false))))
hosts = strings.Split(allhosts, ",")
defer func(now time.Time) { defer func(now time.Time) {
totalTime := time.Since(now) totalTime := time.Since(now)
log.Info("total time", "time", totalTime, "kb", filesize) log.Info("total time", "time", totalTime, "kb", filesize)
@ -85,37 +96,6 @@ func wrapCliCommand(name string, killOnTimeout bool, command func(*cli.Context)
} }
} }
func generateEndpoints(scheme string, cluster string, app string, from int, to int) {
if cluster == "prod" {
for port := from; port < to; port++ {
endpoints = append(endpoints, fmt.Sprintf("%s://%v.swarm-gateways.net", scheme, port))
}
} else if cluster == "private-internal" {
for port := from; port < to; port++ {
endpoints = append(endpoints, fmt.Sprintf("%s://swarm-private-internal-%v:8500", scheme, port))
}
} else {
for port := from; port < to; port++ {
endpoints = append(endpoints, fmt.Sprintf("%s://%s-%v-%s.stg.swarm-gateways.net", scheme, app, port, cluster))
}
}
if includeLocalhost {
endpoints = append(endpoints, "http://localhost:8500")
}
}
//just use the first endpoint
func generateEndpoint(scheme string, cluster string, app string, from int) string {
if cluster == "prod" {
return fmt.Sprintf("%s://%v.swarm-gateways.net", scheme, from)
} else if cluster == "private-internal" {
return fmt.Sprintf("%s://swarm-private-internal-%v:8500", scheme, from)
} else {
return fmt.Sprintf("%s://%s-%v-%s.stg.swarm-gateways.net", scheme, app, from, cluster)
}
}
func fetchFeed(topic string, user string, endpoint string, original []byte, ruid string) error { func fetchFeed(topic string, user string, endpoint string, original []byte, ruid string) error {
ctx, sp := spancontext.StartSpan(context.Background(), "feed-and-sync.fetch") ctx, sp := spancontext.StartSpan(context.Background(), "feed-and-sync.fetch")
defer sp.Finish() defer sp.Finish()