upload_and_sync.go

// Copyright 2018 The go-ethereum Authors
// This file is part of go-ethereum.
//
// go-ethereum is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// go-ethereum is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with go-ethereum. If not, see <http://www.gnu.org/licenses/>.

package main

import (
	"bytes"
	"context"
	"encoding/hex"
	"fmt"
	"io/ioutil"
	"math/rand"
	"os"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/ethereum/go-ethereum/log"
	"github.com/ethereum/go-ethereum/metrics"
	"github.com/ethereum/go-ethereum/rpc"
	"github.com/ethereum/go-ethereum/swarm/chunk"
	"github.com/ethereum/go-ethereum/swarm/storage"
	"github.com/ethereum/go-ethereum/swarm/testutil"

	cli "gopkg.in/urfave/cli.v1"
)

func uploadAndSyncCmd(ctx *cli.Context) error {
	// use input seed if it has been set
	if inputSeed != 0 {
		seed = inputSeed
	}

	randomBytes := testutil.RandomBytes(seed, filesize*1000)

	errc := make(chan error)

	go func() {
		errc <- uploadAndSync(ctx, randomBytes)
	}()

	var err error
	select {
	case err = <-errc:
		if err != nil {
			metrics.GetOrRegisterCounter(fmt.Sprintf("%s.fail", commandName), nil).Inc(1)
		}
	case <-time.After(time.Duration(timeout) * time.Second):
		metrics.GetOrRegisterCounter(fmt.Sprintf("%s.timeout", commandName), nil).Inc(1)

		err = fmt.Errorf("timeout after %v sec", timeout)
	}

	// trigger debug functionality on randomBytes
	e := trackChunks(randomBytes[:], true)
	if e != nil {
		log.Error(e.Error())
	}

	return err
}

func trackChunks(testData []byte, submitMetrics bool) error {
	addrs, err := getAllRefs(testData)
	if err != nil {
		return err
	}

	for i, ref := range addrs {
		log.Debug(fmt.Sprintf("ref %d", i), "ref", ref)
	}

	var globalYes, globalNo int
	var globalMu sync.Mutex
	var hasErr bool

	var wg sync.WaitGroup
	wg.Add(len(hosts))

	var mu sync.Mutex                    // mutex protecting the allHostsChunks and bzzAddrs maps
	allHostChunks := map[string]string{} // host->bitvector of presence for chunks
	bzzAddrs := map[string]string{}      // host->bzzAddr

	for _, host := range hosts {
		host := host
		go func() {
			defer wg.Done()
			httpHost := fmt.Sprintf("ws://%s:%d", host, 8546)

			ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
			defer cancel()

			rpcClient, err := rpc.DialContext(ctx, httpHost)
			if rpcClient != nil {
				defer rpcClient.Close()
			}
			if err != nil {
				log.Error("error dialing host", "err", err, "host", httpHost)
				hasErr = true
				return
			}

			hostChunks, err := getChunksBitVectorFromHost(rpcClient, addrs)
			if err != nil {
				log.Error("error getting chunks bit vector from host", "err", err, "host", httpHost)
				hasErr = true
				return
			}

			bzzAddr, err := getBzzAddrFromHost(rpcClient)
			if err != nil {
				log.Error("error getting bzz addrs from host", "err", err, "host", httpHost)
				hasErr = true
				return
			}

			mu.Lock()
			allHostChunks[host] = hostChunks
			bzzAddrs[host] = bzzAddr
			mu.Unlock()

			yes, no := 0, 0
			for _, val := range hostChunks {
				if val == '1' {
					yes++
				} else {
					no++
				}
			}

			if no == 0 {
				log.Info("host reported to have all chunks", "host", host)
			}

			log.Debug("chunks", "chunks", hostChunks, "yes", yes, "no", no, "host", host)

			if submitMetrics {
				globalMu.Lock()
				globalYes += yes
				globalNo += no
				globalMu.Unlock()
			}
		}()
	}

	wg.Wait()

	checkChunksVsMostProxHosts(addrs, allHostChunks, bzzAddrs)

	if !hasErr && submitMetrics {
		// remove the chunks stored on the uploader node
		globalYes -= len(addrs)

		metrics.GetOrRegisterCounter("deployment.chunks.yes", nil).Inc(int64(globalYes))
		metrics.GetOrRegisterCounter("deployment.chunks.no", nil).Inc(int64(globalNo))
		metrics.GetOrRegisterCounter("deployment.chunks.refs", nil).Inc(int64(len(addrs)))
	}

	return nil
}

// getChunksBitVectorFromHost returns a bit vector of presence for a given slice of chunks from a given host
func getChunksBitVectorFromHost(client *rpc.Client, addrs []storage.Address) (string, error) {
	var hostChunks string

	err := client.Call(&hostChunks, "bzz_has", addrs)
	if err != nil {
		return "", err
	}

	return hostChunks, nil
}

// getBzzAddrFromHost returns the bzzAddr for a given host
func getBzzAddrFromHost(client *rpc.Client) (string, error) {
	var hive string

	err := client.Call(&hive, "bzz_hive")
	if err != nil {
		return "", err
	}

	// we make an ugly assumption about the output format of the hive.String() method
	// ideally we should replace this with an API call that returns the bzz addr for a given host,
	// but this also works for now (provided we don't change the hive.String() method, which we haven't in some time
	ss := strings.Split(strings.Split(hive, "\n")[3], " ")
	return ss[len(ss)-1], nil
}

// checkChunksVsMostProxHosts is checking:
// 1. whether a chunk has been found at less than 2 hosts. Considering our NN size, this should not happen.
// 2. if a chunk is not found at its closest node. This should also not happen.
// Together with the --only-upload flag, we could run this smoke test and make sure that our syncing
// functionality is correct (without even trying to retrieve the content).
//
// addrs - a slice with all uploaded chunk refs
// allHostChunks - host->bit vector, showing what chunks are present on what hosts
// bzzAddrs - host->bzz address, used when determining the most proximate host for a given chunk
func checkChunksVsMostProxHosts(addrs []storage.Address, allHostChunks map[string]string, bzzAddrs map[string]string) {
	for k, v := range bzzAddrs {
		log.Trace("bzzAddr", "bzz", v, "host", k)
	}

	for i := range addrs {
		var foundAt int
		maxProx := -1
		var maxProxHost string
		for host := range allHostChunks {
			if allHostChunks[host][i] == '1' {
				foundAt++
			}

			ba, err := hex.DecodeString(bzzAddrs[host])
			if err != nil {
				panic(err)
			}

			// calculate the host closest to any chunk
			prox := chunk.Proximity(addrs[i], ba)
			if prox > maxProx {
				maxProx = prox
				maxProxHost = host
			}
		}

		if allHostChunks[maxProxHost][i] == '0' {
			log.Error("chunk not found at max prox host", "ref", addrs[i], "host", maxProxHost, "bzzAddr", bzzAddrs[maxProxHost])
		} else {
			log.Trace("chunk present at max prox host", "ref", addrs[i], "host", maxProxHost, "bzzAddr", bzzAddrs[maxProxHost])
		}

		// if chunk found at less than 2 hosts
		if foundAt < 2 {
			log.Error("chunk found at less than two hosts", "foundAt", foundAt, "ref", addrs[i])
		}
	}
}

func getAllRefs(testData []byte) (storage.AddressCollection, error) {
	datadir, err := ioutil.TempDir("", "chunk-debug")
	if err != nil {
		return nil, fmt.Errorf("unable to create temp dir: %v", err)
	}
	defer os.RemoveAll(datadir)
	fileStore, err := storage.NewLocalFileStore(datadir, make([]byte, 32), chunk.NewTags())
	if err != nil {
		return nil, err
	}

	reader := bytes.NewReader(testData)
	return fileStore.GetAllReferences(context.Background(), reader, false)
}

func uploadAndSync(c *cli.Context, randomBytes []byte) error {
	log.Info("uploading to "+httpEndpoint(hosts[0])+" and syncing", "seed", seed)

	t1 := time.Now()
	hash, err := upload(randomBytes, httpEndpoint(hosts[0]))
	if err != nil {
		log.Error(err.Error())
		return err
	}
	t2 := time.Since(t1)
	metrics.GetOrRegisterResettingTimer("upload-and-sync.upload-time", nil).Update(t2)

	fhash, err := digest(bytes.NewReader(randomBytes))
	if err != nil {
		log.Error(err.Error())
		return err
	}

	log.Info("uploaded successfully", "hash", hash, "took", t2, "digest", fmt.Sprintf("%x", fhash))

	// wait to sync and log chunks before fetch attempt, only if syncDelay is set to true
	if syncDelay {
		waitToSync()

		log.Debug("chunks before fetch attempt", "hash", hash)

		err = trackChunks(randomBytes, false)
		if err != nil {
			log.Error(err.Error())
		}
	}

	if onlyUpload {
		log.Debug("only-upload is true, stoppping test", "hash", hash)
		return nil
	}

	randIndex := 1 + rand.Intn(len(hosts)-1)

	for {
		start := time.Now()
		err := fetch(hash, httpEndpoint(hosts[randIndex]), fhash, "")
		if err != nil {
			time.Sleep(2 * time.Second)
			continue
		}
		ended := time.Since(start)

		metrics.GetOrRegisterResettingTimer("upload-and-sync.single.fetch-time", nil).Update(ended)
		log.Info("fetch successful", "took", ended, "endpoint", httpEndpoint(hosts[randIndex]))
		break
	}

	return nil
}

func isSyncing(wsHost string) (bool, error) {
	rpcClient, err := rpc.Dial(wsHost)
	if rpcClient != nil {
		defer rpcClient.Close()
	}

	if err != nil {
		log.Error("error dialing host", "err", err)
		return false, err
	}

	var isSyncing bool
	err = rpcClient.Call(&isSyncing, "bzz_isSyncing")
	if err != nil {
		log.Error("error calling host for isSyncing", "err", err)
		return false, err
	}

	log.Debug("isSyncing result", "host", wsHost, "isSyncing", isSyncing)

	return isSyncing, nil
}

func waitToSync() {
	t1 := time.Now()

	ns := uint64(1)

	for ns > 0 {
		time.Sleep(3 * time.Second)

		notSynced := uint64(0)
		var wg sync.WaitGroup
		wg.Add(len(hosts))
		for i := 0; i < len(hosts); i++ {
			i := i
			go func(idx int) {
				stillSyncing, err := isSyncing(wsEndpoint(hosts[idx]))

				if stillSyncing || err != nil {
					atomic.AddUint64(&notSynced, 1)
				}
				wg.Done()
			}(i)
		}
		wg.Wait()

		ns = atomic.LoadUint64(&notSynced)
	}

	t2 := time.Since(t1)
	metrics.GetOrRegisterResettingTimer("upload-and-sync.single.wait-for-sync.deployment", nil).Update(t2)
}