Merge pull request #53 from filecoin-project/feat/drand-setup

initial drand setup

lotus-soup/baseline.go

@@ -46,6 +46,7 @@ var baselineRoles = map[string]func(*TestEnvironment) error{
 	"bootstrapper": runBootstrapper,
 	"miner":        runMiner,
 	"client":       runBaselineClient,
+	"drand":        runDrandNode,
 }
 
 func runBaselineClient(t *TestEnvironment) error {
@@ -236,3 +237,4 @@ func extractCarData(ctx context.Context, rdata []byte, rpath string) []byte {
 	}
 	return rdata
 }
+

lotus-soup/common_roles.go

@@ -66,3 +66,17 @@ func runMiner(t *TestEnvironment) error {
 	t.SyncClient.MustSignalAndWait(ctx, stateDone, t.TestInstanceCount)
 	return nil
 }
+
+func runDrandNode(t *TestEnvironment) error {
+	t.RecordMessage("running drand node")
+	dr, err := prepareDrandNode(t)
+	if err != nil {
+		return err
+	}
+	defer dr.Cleanup()
+
+	// TODO add ability to halt / recover on demand
+	ctx := context.Background()
+	t.SyncClient.MustSignalAndWait(ctx, stateDone, t.TestInstanceCount)
+	return nil
+}

lotus-soup/compositions/composition-local-drand.toml

@@ -0,0 +1,72 @@
+[metadata]
+  name = "lotus-soup"
+  author = ""
+
+[global]
+  plan = "lotus-soup"
+  case = "lotus-baseline"
+  total_instances = 6
+  builder = "docker:go"
+  runner = "local:docker"
+
+[[groups]]
+  id = "bootstrapper"
+  [groups.resources]
+    memory = "120Mi"
+    cpu = "10m"
+  [groups.instances]
+    count = 1
+    percentage = 0.0
+  [groups.run]
+    [groups.run.test_params]
+      role = "bootstrapper"
+      clients = "1"
+      miners = "1"
+      balance = "2000"
+      sectors = "10"
+      random_beacon_type = "local-drand"
+
+[[groups]]
+  id = "miners"
+  [groups.resources]
+    memory = "120Mi"
+    cpu = "10m"
+  [groups.instances]
+    count = 1
+    percentage = 0.0
+  [groups.run]
+    [groups.run.test_params]
+      role = "miner"
+      clients = "1"
+      miners = "1"
+      balance = "2000"
+      sectors = "10"
+      random_beacon_type = "local-drand"
+
+
+[[groups]]
+  id = "clients"
+  [groups.resources]
+    memory = "120Mi"
+    cpu = "10m"
+  [groups.instances]
+    count = 1
+    percentage = 0.0
+  [groups.run]
+    [groups.run.test_params]
+      role = "client"
+      clients = "1"
+      miners = "1"
+      balance = "2000"
+      sectors = "10"
+      random_beacon_type = "local-drand"
+
+
+[[groups]]
+  id = "drand"
+  [groups.instances]
+    count = 3
+    percentage = 0.0
+  [groups.run]
+    [groups.run.test_params]
+      role = "drand"

lotus-soup/compositions/composition.toml

@@ -24,6 +24,7 @@
       miners = "2"
       balance = "2000000000"
       sectors = "10"
+      random_beacon_type = "mock"
 
 [[groups]]
   id = "miners"
@@ -37,6 +38,8 @@
       miners = "2"
       balance = "2000000000"
       sectors = "10"
+      random_beacon_type = "mock"
+
 
 [[groups]]
   id = "clients"
@@ -50,3 +53,5 @@
       miners = "2"
       balance = "2000000000"
       sectors = "10"
+      random_beacon_type = "mock"
+

lotus-soup/drand.go

@@ -0,0 +1,245 @@
+package main
+
+import (
+	"bytes"
+	"context"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"net"
+	"os"
+	"path"
+	"time"
+
+	"github.com/drand/drand/chain"
+	hclient "github.com/drand/drand/client/http"
+	"github.com/drand/drand/log"
+	"github.com/drand/drand/lp2p"
+	"github.com/filecoin-project/lotus/node/modules/dtypes"
+	"github.com/libp2p/go-libp2p-core/peer"
+	ma "github.com/multiformats/go-multiaddr"
+	"github.com/testground/sdk-go/sync"
+
+	"github.com/drand/drand/demo/node"
+)
+
+var (
+	PrepareDrandTimeout = time.Minute
+	drandConfigTopic    = sync.NewTopic("drand-config", &DrandRuntimeInfo{})
+)
+
+type DrandRuntimeInfo struct {
+	Config          dtypes.DrandConfig
+	GossipBootstrap dtypes.DrandBootstrap
+}
+
+type DrandInstance struct {
+	Node node.Node
+	GossipRelay *lp2p.GossipRelayNode
+
+	stateDir string
+}
+
+func (d *DrandInstance) Cleanup() error {
+	return os.RemoveAll(d.stateDir)
+}
+
+// waitForDrandConfig should be called by filecoin instances before constructing the lotus Node
+// you can use the returned dtypes.DrandConfig to override the default production config.
+func waitForDrandConfig(ctx context.Context, client sync.Client) (*DrandRuntimeInfo, error) {
+	ch := make(chan *DrandRuntimeInfo, 1)
+	sub := client.MustSubscribe(ctx, drandConfigTopic, ch)
+	select {
+	case cfg := <-ch:
+		return cfg, nil
+	case err := <-sub.Done():
+		return nil, err
+	}
+}
+
+// prepareDrandNode starts a drand instance and runs a DKG with the other members of the composition group.
+// Once the chain is running, the leader publishes the chain info needed by lotus nodes on
+// drandConfigTopic
+func prepareDrandNode(t *TestEnvironment) (*DrandInstance, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), PrepareDrandTimeout)
+	defer cancel()
+
+	startTime := time.Now()
+
+	seq := t.GroupSeq
+	isLeader := seq == 1
+	nNodes := t.TestGroupInstanceCount
+
+	myAddr := t.NetClient.MustGetDataNetworkIP()
+	period := t.DurationParam("drand_period")
+	threshold := t.IntParam("drand_threshold")
+	runGossipRelay := t.BooleanParam("drand_gossip_relay")
+
+	beaconOffset := 3
+
+	stateDir, err := ioutil.TempDir("", fmt.Sprintf("drand-%d", t.GroupSeq))
+	if err != nil {
+		return nil, err
+	}
+
+	// TODO(maybe): use TLS?
+	n := node.NewLocalNode(int(seq), period.String(), stateDir, false, myAddr.String())
+
+	// share the node addresses with other nodes
+	// TODO: if we implement TLS, this is where we'd share public TLS keys
+	type NodeAddr struct {
+		PrivateAddr string
+		PublicAddr  string
+		IsLeader    bool
+	}
+	addrTopic := sync.NewTopic("drand-addrs", &NodeAddr{})
+	var publicAddrs []string
+	var leaderAddr string
+	ch := make(chan *NodeAddr)
+	_, sub := t.SyncClient.MustPublishSubscribe(ctx, addrTopic, &NodeAddr{
+		PrivateAddr: n.PrivateAddr(),
+		PublicAddr:  n.PublicAddr(),
+		IsLeader:    isLeader,
+	}, ch)
+	for i := 0; i < nNodes; i++ {
+		select {
+		case msg := <-ch:
+			publicAddrs = append(publicAddrs, fmt.Sprintf("http://%s", msg.PublicAddr))
+			if msg.IsLeader {
+				leaderAddr = msg.PrivateAddr
+			}
+		case err := <-sub.Done():
+			return nil, fmt.Errorf("unable to read drand addrs from sync service: %w", err)
+		}
+	}
+	if leaderAddr == "" {
+		return nil, fmt.Errorf("got %d drand addrs, but no leader", len(publicAddrs))
+	}
+
+	t.SyncClient.MustSignalAndWait(ctx, "drand-start", nNodes)
+	t.RecordMessage("Starting drand sharing ceremony")
+	if err := n.Start(stateDir); err != nil {
+		return nil, err
+	}
+
+	alive := false
+	waitSecs := 10
+	for i := 0; i < waitSecs; i++ {
+		if !n.Ping() {
+			time.Sleep(time.Second)
+			continue
+		}
+		t.R().RecordPoint("drand_first_ping", time.Now().Sub(startTime).Seconds())
+		alive = true
+		break
+	}
+	if !alive {
+		return nil, fmt.Errorf("drand node %d failed to start after %d seconds", t.GroupSeq, waitSecs)
+	}
+
+	// run DKG
+	t.SyncClient.MustSignalAndWait(ctx, "drand-dkg-start", nNodes)
+	if !isLeader {
+		time.Sleep(time.Second)
+	}
+	grp := n.RunDKG(nNodes, threshold, period.String(), isLeader, leaderAddr, beaconOffset)
+	if grp == nil {
+		return nil, fmt.Errorf("drand dkg failed")
+	}
+	t.R().RecordPoint("drand_dkg_complete", time.Now().Sub(startTime).Seconds())
+
+	t.RecordMessage("drand dkg complete, waiting for chain start")
+	// wait for chain to begin
+	to := time.Until(time.Unix(grp.GenesisTime, 0).Add(3 * time.Second).Add(grp.Period))
+	time.Sleep(to)
+
+	t.RecordMessage("drand beacon chain started, fetching initial round via http")
+	// verify that we can get a round of randomness from the chain using an http client
+	info := chain.NewChainInfo(grp)
+	myPublicAddr := fmt.Sprintf("http://%s", n.PublicAddr())
+	client, err := hclient.NewWithInfo(myPublicAddr, info, nil)
+	if err != nil {
+		return nil, fmt.Errorf("unable to create drand http client: %w", err)
+	}
+
+	_, err = client.Get(ctx, 1)
+	if err != nil {
+		return nil, fmt.Errorf("unable to get initial drand round: %w", err)
+	}
+
+	// start gossip relay (unless disabled via testplan parameter)
+	var gossipRelay *lp2p.GossipRelayNode
+	var relayAddrs []peer.AddrInfo
+
+	if runGossipRelay {
+		gossipDir := path.Join(stateDir, "gossip-relay")
+		listenAddr := fmt.Sprintf("/ip4/%s/tcp/7777", myAddr.String())
+		relayCfg := lp2p.GossipRelayConfig{
+			ChainHash:    hex.EncodeToString(info.Hash()),
+			Addr:         listenAddr,
+			DataDir:      gossipDir,
+			IdentityPath: path.Join(gossipDir, "identity.key"),
+			Insecure:     true,
+			Client:       client,
+		}
+		t.RecordMessage("starting drand gossip relay")
+		gossipRelay, err = lp2p.NewGossipRelayNode(log.DefaultLogger, &relayCfg)
+		if err != nil {
+			return nil, fmt.Errorf("failed to construct drand gossip relay: %w", err)
+		}
+
+		t.RecordMessage("sharing gossip relay addrs")
+		// share the gossip relay addrs so we can publish them in DrandRuntimeInfo
+		relayInfo, err := relayAddrInfo(gossipRelay.Multiaddrs(), myAddr)
+		if err != nil {
+			return nil, err
+		}
+		infoCh := make(chan *peer.AddrInfo, nNodes)
+		infoTopic := sync.NewTopic("drand-gossip-addrs", &peer.AddrInfo{})
+
+		_, sub := t.SyncClient.MustPublishSubscribe(ctx, infoTopic, relayInfo, infoCh)
+		for i := 0; i < nNodes; i++ {
+			select {
+			case ai := <-infoCh:
+				relayAddrs = append(relayAddrs, *ai)
+			case err := <-sub.Done():
+				return nil, fmt.Errorf("unable to get drand relay addr from sync service: %w", err)
+			}
+		}
+	}
+
+	// if we're the leader, publish the config to the sync service
+	if isLeader {
+		buf := bytes.Buffer{}
+		if err := info.ToJSON(&buf); err != nil {
+			return nil, fmt.Errorf("error marshaling chain info: %w", err)
+		}
+		cfg := DrandRuntimeInfo{
+			Config: dtypes.DrandConfig{
+				Servers:       publicAddrs,
+				ChainInfoJSON: buf.String(),
+			},
+			GossipBootstrap: relayAddrs,
+		}
+		dump, _ := json.Marshal(cfg)
+		t.RecordMessage("publishing drand config on sync topic: %s", string(dump))
+		t.SyncClient.MustPublish(ctx, drandConfigTopic, &cfg)
+	}
+
+	return &DrandInstance{
+		Node: n,
+		GossipRelay: gossipRelay,
+		stateDir: stateDir,
+	}, nil
+}
+
+func relayAddrInfo(addrs []ma.Multiaddr, dataIP net.IP) (*peer.AddrInfo, error) {
+	for _, a := range addrs {
+		if ip, _ := a.ValueForProtocol(ma.P_IP4); ip != dataIP.String() {
+			continue
+		}
+		return peer.AddrInfoFromP2pAddr(a)
+	}
+	return nil, fmt.Errorf("no addr found with data ip %s in addrs: %v", dataIP, addrs)
+}

lotus-soup/go.mod

@@ -4,10 +4,11 @@ go 1.14
 
 require (
 	github.com/davecgh/go-spew v1.1.1
+	github.com/drand/drand v0.9.2-0.20200616080806-a94e9c1636a4
 	github.com/filecoin-project/go-address v0.0.2-0.20200504173055-8b6f2fb2b3ef
 	github.com/filecoin-project/go-fil-markets v0.3.0
 	github.com/filecoin-project/go-storedcounter v0.0.0-20200421200003-1c99c62e8a5b
-	github.com/filecoin-project/lotus v0.4.1-0.20200623104442-68d38eff33e4
+	github.com/filecoin-project/lotus v0.4.1-0.20200623211458-e8642442267b
 	github.com/filecoin-project/specs-actors v0.6.2-0.20200617175406-de392ca14121
 	github.com/ipfs/go-cid v0.0.6
 	github.com/ipfs/go-datastore v0.4.4

lotus-soup/go.sum

@@ -246,8 +246,8 @@ github.com/filecoin-project/go-statestore v0.1.0 h1:t56reH59843TwXHkMcwyuayStBIi
 github.com/filecoin-project/go-statestore v0.1.0/go.mod h1:LFc9hD+fRxPqiHiaqUEZOinUJB4WARkRfNl10O7kTnI=
 github.com/filecoin-project/go-storedcounter v0.0.0-20200421200003-1c99c62e8a5b h1:fkRZSPrYpk42PV3/lIXiL0LHetxde7vyYYvSsttQtfg=
 github.com/filecoin-project/go-storedcounter v0.0.0-20200421200003-1c99c62e8a5b/go.mod h1:Q0GQOBtKf1oE10eSXSlhN45kDBdGvEcVOqMiffqX+N8=
-github.com/filecoin-project/lotus v0.4.1-0.20200623104442-68d38eff33e4 h1:h0iwqUzi4+E+xs1vLtmM2uxTUoTSNk4P/hFG7Dk9jnw=
+github.com/filecoin-project/lotus v0.4.1-0.20200623211458-e8642442267b h1:Oq1ABSZVYNFtvWqnj3bHINcw34T0if/0/zlsRq8rGgc=
-github.com/filecoin-project/lotus v0.4.1-0.20200623104442-68d38eff33e4/go.mod h1:uxvEKQiyuXisy/7MB6pFwb4WxxcA3UY3hyJnhL+k7M4=
+github.com/filecoin-project/lotus v0.4.1-0.20200623211458-e8642442267b/go.mod h1:uxvEKQiyuXisy/7MB6pFwb4WxxcA3UY3hyJnhL+k7M4=
 github.com/filecoin-project/sector-storage v0.0.0-20200615154852-728a47ab99d6/go.mod h1:M59QnAeA/oV+Z8oHFLoNpGMv0LZ8Rll+vHVXX7GirPM=
 github.com/filecoin-project/sector-storage v0.0.0-20200618073200-d9de9b7cb4b4 h1:lQC8Fbyn31/H4QxYAYwVV3PYZ9vS61EmjktZc5CaiYs=
 github.com/filecoin-project/sector-storage v0.0.0-20200618073200-d9de9b7cb4b4/go.mod h1:M59QnAeA/oV+Z8oHFLoNpGMv0LZ8Rll+vHVXX7GirPM=

lotus-soup/manifest.toml

@@ -26,3 +26,13 @@ instances = { min = 1, max = 100, default = 5 }
   balance = { type = "int", default = 1 }
   sectors = { type = "int", default = 1 }
   role = { type = "string" }
+
+  random_beacon_type = { type = "enum", default = "mock", options = ["mock", "local-drand", "external-drand"] }
+
+  # Params relevant to drand nodes. drand nodes should have role="drand", and must all be
+  # in the same composition group. There must be at least threshold drand nodes.
+  # To get lotus nodes to actually use the drand nodes, you must set random_beacon_type="local-drand"
+  # for the lotus node groups.
+  drand_period = { type = "duration", default="10s" }
+  drand_threshold = { type = "int", default = 2 }
+  drand_gossip_relay = { type = "bool", default = true }

lotus-soup/node.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"crypto/rand"
 	"os"
+	"strings"
 
 	//"encoding/json"
 	"fmt"
@@ -16,6 +17,7 @@ import (
 	"github.com/filecoin-project/lotus/api"
 	"github.com/filecoin-project/lotus/build"
 	"github.com/filecoin-project/lotus/chain/actors"
+	"github.com/filecoin-project/lotus/chain/beacon"
 	genesis_chain "github.com/filecoin-project/lotus/chain/gen/genesis"
 	"github.com/filecoin-project/lotus/chain/types"
 	"github.com/filecoin-project/lotus/chain/wallet"
@@ -82,6 +84,19 @@ type TestEnvironment struct {
 	*run.InitContext
 }
 
+// workaround for default params being wrapped in quote chars
+func (t *TestEnvironment) StringParam(name string) string {
+	return strings.Trim(t.RunEnv.StringParam(name), "\"")
+}
+
+func (t *TestEnvironment) DurationParam(name string) time.Duration {
+	d, err := time.ParseDuration(t.StringParam(name))
+	if err != nil {
+		panic(fmt.Errorf("invalid duration value for param '%s': %w", name, err))
+	}
+	return d
+}
+
 type Node struct {
 	fullApi  api.FullNode
 	minerApi api.StorageMiner
@@ -116,6 +131,11 @@ func prepareBootstrapper(t *TestEnvironment) (*Node, error) {
 	miners := t.IntParam("miners")
 	nodes := clients + miners
 
+	drandOpt, err := getDrandConfig(ctx, t)
+	if err != nil {
+		return nil, err
+	}
+
 	// the first duty of the boostrapper is to construct the genesis block
 	// first collect all client and miner balances to assign initial funds
 	balances, err := waitForBalances(t, ctx, nodes)
@@ -177,6 +197,7 @@ func prepareBootstrapper(t *TestEnvironment) (*Node, error) {
 		withListenAddress(bootstrapperIP),
 		withBootstrapper(nil),
 		withPubsubConfig(true),
+		drandOpt,
 	)
 	if err != nil {
 		return nil, err
@@ -229,6 +250,11 @@ func prepareMiner(t *TestEnvironment) (*Node, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), PrepareNodeTimeout)
 	defer cancel()
 
+	drandOpt, err := getDrandConfig(ctx, t)
+	if err != nil {
+		return nil, err
+	}
+
 	// first create a wallet
 	walletKey, err := wallet.GenerateKey(crypto.SigTypeBLS)
 	if err != nil {
@@ -342,6 +368,7 @@ func prepareMiner(t *TestEnvironment) (*Node, error) {
 		withListenAddress(minerIP),
 		withBootstrapper(genesisMsg.Bootstrapper),
 		withPubsubConfig(false),
+		drandOpt,
 	)
 	if err != nil {
 		return nil, err
@@ -446,6 +473,11 @@ func prepareClient(t *TestEnvironment) (*Node, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), PrepareNodeTimeout)
 	defer cancel()
 
+	drandOpt, err := getDrandConfig(ctx, t)
+	if err != nil {
+		return nil, err
+	}
+
 	// first create a wallet
 	walletKey, err := wallet.GenerateKey(crypto.SigTypeBLS)
 	if err != nil {
@@ -475,6 +507,7 @@ func prepareClient(t *TestEnvironment) (*Node, error) {
 		withListenAddress(clientIP),
 		withBootstrapper(genesisMsg.Bootstrapper),
 		withPubsubConfig(false),
+		drandOpt,
 	)
 	if err != nil {
 		return nil, err
@@ -636,3 +669,39 @@ func collectClientAddrs(t *TestEnvironment, ctx context.Context, clients int) ([
 
 	return addrs, nil
 }
+
+func getDrandConfig(ctx context.Context, t *TestEnvironment) (node.Option, error) {
+	beaconType := t.StringParam("random_beacon_type")
+	switch beaconType {
+	case "external-drand":
+		noop := func(settings *node.Settings) error {
+			return nil
+		}
+		return noop, nil
+
+	case "local-drand":
+		cfg, err := waitForDrandConfig(ctx, t.SyncClient)
+		if err != nil {
+			t.RecordMessage("error getting drand config: %w", err)
+			return nil, err
+			
+		}
+		t.RecordMessage("setting drand config: %v", cfg)
+		return node.Options(
+			node.Override(new(dtypes.DrandConfig), cfg.Config),
+			node.Override(new(dtypes.DrandBootstrap), cfg.GossipBootstrap),
+		), nil
+
+	case "mock":
+		return node.Options(
+			node.Override(new(beacon.RandomBeacon), modtest.RandomBeacon),
+			node.Override(new(dtypes.DrandConfig), dtypes.DrandConfig{
+				ChainInfoJSON: "{\"Hash\":\"wtf\"}",
+			}),
+			node.Override(new(dtypes.DrandBootstrap), dtypes.DrandBootstrap{}),
+		), nil
+
+	default:
+		return nil, fmt.Errorf("unknown random_beacon_type: %s", beaconType)
+	}
+}