Add multi beacon client support

chain/info.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"crypto/sha256"
 	"encoding/binary"
+	"encoding/hex"
 	"time"
 
 	"github.com/drand/drand/common/scheme"
@@ -60,6 +61,11 @@ func (c *Info) Hash() []byte {
 	return h.Sum(nil)
 }
 
+// HashString returns the value of Hash in string format
+func (c *Info) HashString() string {
+	return hex.EncodeToString(c.Hash())
+}
+
 // Equal indicates if two Chain Info objects are equivalent
 func (c *Info) Equal(c2 *Info) bool {
 	return c.GenesisTime == c2.GenesisTime &&

client/client_test.go

@@ -48,10 +48,16 @@ func TestClientMultiple(t *testing.T) {
 	addr2, _, cancel2, _ := httpmock.NewMockHTTPPublicServer(t, false, sch)
 	defer cancel2()
 
+	httpClients := http.ForURLs([]string{"http://" + addr1, "http://" + addr2}, chainInfo.Hash())
+	if len(httpClients) == 0 {
+		t.Error("http clients is empty")
+		return
+	}
+
 	var c client.Client
 	var e error
 	c, e = client.New(
-		client.From(http.ForURLs([]string{"http://" + addr1, "http://" + addr2}, chainInfo.Hash())...),
+		client.From(httpClients...),
 		client.WithChainHash(chainInfo.Hash()))
 
 	if e != nil {
@@ -93,9 +99,15 @@ func TestClientCache(t *testing.T) {
 	addr1, chainInfo, cancel, _ := httpmock.NewMockHTTPPublicServer(t, false, sch)
 	defer cancel()
 
+	httpClients := http.ForURLs([]string{"http://" + addr1}, chainInfo.Hash())
+	if len(httpClients) == 0 {
+		t.Error("http clients is empty")
+		return
+	}
+
 	var c client.Client
 	var e error
-	c, e = client.New(client.From(http.ForURLs([]string{"http://" + addr1}, chainInfo.Hash())...),
+	c, e = client.New(client.From(httpClients...),
 		client.WithChainHash(chainInfo.Hash()), client.WithCacheSize(1))
 
 	if e != nil {
@@ -123,10 +135,16 @@ func TestClientWithoutCache(t *testing.T) {
 	addr1, chainInfo, cancel, _ := httpmock.NewMockHTTPPublicServer(t, false, sch)
 	defer cancel()
 
+	httpClients := http.ForURLs([]string{"http://" + addr1}, chainInfo.Hash())
+	if len(httpClients) == 0 {
+		t.Error("http clients is empty")
+		return
+	}
+
 	var c client.Client
 	var e error
 	c, e = client.New(
-		client.From(http.ForURLs([]string{"http://" + addr1}, chainInfo.Hash())...),
+		client.From(httpClients...),
 		client.WithChainHash(chainInfo.Hash()),
 		client.WithCacheSize(0))
 
@@ -230,6 +248,11 @@ func TestClientAutoWatch(t *testing.T) {
 	defer cancel()
 
 	httpClient := http.ForURLs([]string{"http://" + addr1}, chainInfo.Hash())
+	if len(httpClient) == 0 {
+		t.Error("http clients is empty")
+		return
+	}
+
 	r1, _ := httpClient[0].Get(context.Background(), 1)
 	r2, _ := httpClient[0].Get(context.Background(), 2)
 	results := []client.Result{r1, r2}

client/grpc/client.go

@@ -7,6 +7,8 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/drand/drand/protobuf/common"
+
 	"github.com/drand/drand/chain"
 	"github.com/drand/drand/client"
 	"github.com/drand/drand/log"
@@ -20,14 +22,15 @@ import (
 const grpcDefaultTimeout = 5 * time.Second
 
 type grpcClient struct {
-	address string
-	client  drand.PublicClient
-	conn    *grpc.ClientConn
-	l       log.Logger
+	address   string
+	chainHash []byte
+	client    drand.PublicClient
+	conn      *grpc.ClientConn
+	l         log.Logger
 }
 
 // New creates a drand client backed by a GRPC connection.
-func New(address, certPath string, insecure bool) (client.Client, error) {
+func New(address, certPath string, insecure bool, chainHash []byte) (client.Client, error) {
 	opts := []grpc.DialOption{}
 	if certPath != "" {
 		creds, err := credentials.NewClientTLSFromFile(certPath, "")
@@ -48,7 +51,8 @@ func New(address, certPath string, insecure bool) (client.Client, error) {
 	if err != nil {
 		return nil, err
 	}
-	return &grpcClient{address, drand.NewPublicClient(conn), conn, log.DefaultLogger()}, nil
+
+	return &grpcClient{address, chainHash, drand.NewPublicClient(conn), conn, log.DefaultLogger()}, nil
 }
 
 func asRD(r *drand.PublicRandResponse) *client.RandomData {
@@ -67,7 +71,9 @@ func (g *grpcClient) String() string {
 
 // Get returns a the randomness at `round` or an error.
 func (g *grpcClient) Get(ctx context.Context, round uint64) (client.Result, error) {
-	curr, err := g.client.PublicRand(ctx, &drand.PublicRandRequest{Round: round})
+	metadata := common.Metadata{ChainHash: g.chainHash}
+
+	curr, err := g.client.PublicRand(ctx, &drand.PublicRandRequest{Round: round, Metadata: &metadata})
 	if err != nil {
 		return nil, err
 	}
@@ -80,7 +86,9 @@ func (g *grpcClient) Get(ctx context.Context, round uint64) (client.Result, erro
 
 // Watch returns new randomness as it becomes available.
 func (g *grpcClient) Watch(ctx context.Context) <-chan client.Result {
-	stream, err := g.client.PublicRandStream(ctx, &drand.PublicRandRequest{Round: 0})
+	metadata := common.Metadata{ChainHash: g.chainHash}
+
+	stream, err := g.client.PublicRandStream(ctx, &drand.PublicRandRequest{Round: 0, Metadata: &metadata})
 	ch := make(chan client.Result, 1)
 	if err != nil {
 		close(ch)
@@ -92,7 +100,9 @@ func (g *grpcClient) Watch(ctx context.Context) <-chan client.Result {
 
 // Info returns information about the chain.
 func (g *grpcClient) Info(ctx context.Context) (*chain.Info, error) {
-	proto, err := g.client.ChainInfo(ctx, &drand.ChainInfoRequest{})
+	metadata := common.Metadata{ChainHash: g.chainHash}
+
+	proto, err := g.client.ChainInfo(ctx, &drand.ChainInfoRequest{Metadata: &metadata})
 	if err != nil {
 		return nil, err
 	}
@@ -119,7 +129,10 @@ func (g *grpcClient) translate(stream drand.Public_PublicRandStreamClient, out c
 func (g *grpcClient) RoundAt(t time.Time) uint64 {
 	ctx, cancel := context.WithTimeout(context.Background(), grpcDefaultTimeout)
 	defer cancel()
-	info, err := g.client.ChainInfo(ctx, &drand.ChainInfoRequest{})
+
+	metadata := common.Metadata{ChainHash: g.chainHash}
+
+	info, err := g.client.ChainInfo(ctx, &drand.ChainInfoRequest{Metadata: &metadata})
 	if err != nil {
 		return 0
 	}

client/grpc/client_test.go

@@ -22,7 +22,7 @@ func TestClient(t *testing.T) {
 	go l.Start()
 	defer l.Stop(context.Background())
 
-	c, err := New(addr, "", true)
+	c, err := New(addr, "", true, []byte(""))
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -71,7 +71,7 @@ func TestClientClose(t *testing.T) {
 	go l.Start()
 	defer l.Stop(context.Background())
 
-	c, err := New(addr, "", true)
+	c, err := New(addr, "", true, []byte(""))
 	if err != nil {
 		t.Fatal(err)
 	}

client/http/http.go

@@ -230,7 +230,7 @@ type httpInfoResponse struct {
 	err       error
 }
 
-// FetchGroupInfo attempts to initialize an httpClient when
+// FetchChainInfo attempts to initialize an httpClient when
 // it does not know the full group parameters for a drand group. The chain hash
 // is the hash of the chain info.
 func (h *httpClient) FetchChainInfo(ctx context.Context, chainHash []byte) (*chain.Info, error) {
@@ -243,7 +243,14 @@ func (h *httpClient) FetchChainInfo(ctx context.Context, chainHash []byte) (*cha
 	defer cancel()
 
 	go func() {
-		req, err := nhttp.NewRequestWithContext(ctx, "GET", fmt.Sprintf("%sinfo", h.root), nhttp.NoBody)
+		url := ""
+		if len(chainHash) > 0 {
+			url = fmt.Sprintf("%s%x/info", h.root, chainHash)
+		} else {
+			url = fmt.Sprintf("%sinfo", h.root)
+		}
+
+		req, err := nhttp.NewRequestWithContext(ctx, "GET", url, nhttp.NoBody)
 		if err != nil {
 			resC <- httpInfoResponse{nil, fmt.Errorf("creating request: %w", err)}
 			return
@@ -295,13 +302,13 @@ type httpGetResponse struct {
 	err    error
 }
 
-// Get returns a the randomness at `round` or an error.
+// Get returns the randomness at `round` or an error.
 func (h *httpClient) Get(ctx context.Context, round uint64) (client.Result, error) {
 	var url string
 	if round == 0 {
-		url = fmt.Sprintf("%spublic/latest", h.root)
+		url = fmt.Sprintf("%s%x/public/latest", h.root, h.chainInfo.Hash())
 	} else {
-		url = fmt.Sprintf("%spublic/%d", h.root, round)
+		url = fmt.Sprintf("%s%x/public/%d", h.root, h.chainInfo.Hash(), round)
 	}
 
 	resC := make(chan httpGetResponse, 1)

client/test/http/mock/httpserver.go

@@ -47,12 +47,14 @@ func NewMockHTTPPublicServer(t *testing.T, badSecondRound bool, sch scheme.Schem
 		t.Fatal("could not use server after 3 attempts.")
 	}
 
+	handler.HandlerDrand.CreateBeaconHandler(client, chainInfo.HashString())
+
 	listener, err := net.Listen("tcp", ":0")
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	httpServer := http.Server{Handler: handler}
+	httpServer := http.Server{Handler: handler.HandlerHTTP}
 	go httpServer.Serve(listener)
 
 	return listener.Addr().String(), chainInfo, func() {

cmd/client/lib/cli.go

@@ -54,6 +54,11 @@ var (
 		Usage:   "The hash (in hex) for the chain to follow",
 		Aliases: []string{"chain-hash"}, // DEPRECATED
 	}
+	// HashListFlag is the CLI flag for the hashes list (in hex) for the relay to follow.
+	HashListFlag = &cli.StringSliceFlag{
+		Name:  "hash-list",
+		Usage: "The hash list (in hex) for the relay to follow",
+	}
 	// GroupConfFlag is the CLI flag for specifying the path to the drand group configuration (TOML encoded) or chain info (JSON encoded).
 	GroupConfFlag = &cli.PathFlag{
 		Name: "group-conf",
@@ -155,7 +160,22 @@ func Create(c *cli.Context, withInstrumentation bool, opts ...client.Option) (cl
 
 func buildGrpcClient(c *cli.Context, info **chain.Info) ([]client.Client, error) {
 	if c.IsSet(GRPCConnectFlag.Name) {
-		gc, err := grpc.New(c.String(GRPCConnectFlag.Name), c.String(CertFlag.Name), c.Bool(InsecureFlag.Name))
+		hash := make([]byte, 0)
+
+		if c.IsSet(HashFlag.Name) {
+			var err error
+
+			hash, err = hex.DecodeString(c.String(HashFlag.Name))
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		if *info != nil && len(hash) == 0 {
+			hash = (*info).Hash()
+		}
+
+		gc, err := grpc.New(c.String(GRPCConnectFlag.Name), c.String(CertFlag.Name), c.Bool(InsecureFlag.Name), hash)
 		if err != nil {
 			return nil, err
 		}

cmd/drand-cli/cli_test.go

@@ -775,7 +775,7 @@ func launchDrandInstances(t *testing.T, n int) ([]*drandInstance, string) {
 		pubPath := path.Join(tmpPath, "pub.key")
 
 		freePort := test.FreePort()
-		addr := "127.0.0." + strconv.Itoa(i) + ":" + freePort
+		addr := "127.0.0.1:" + freePort
 		ctrlPort := test.FreePort()
 		metricsPort := test.FreePort()
 

cmd/drand-cli/public.go

@@ -75,7 +75,7 @@ func getPublicRandomness(c *cli.Context) error {
 	var resp client.Result
 	var foundCorrect bool
 	for _, id := range ids {
-		grpcClient, err := grpc.New(id.Addr, certPath, !id.TLS)
+		grpcClient, err := grpc.New(id.Addr, certPath, !id.TLS, group.Hash())
 		if err != nil {
 			fmt.Fprintf(os.Stderr, "drand: could not connect to %s: %s", id.Addr, err)
 			break

cmd/relay/main.go

@@ -1,12 +1,14 @@
 package main
 
 import (
+	"encoding/hex"
 	"fmt"
 	"net"
 	"net/http"
 	"net/http/httptest"
 	"os"
 
+	dclient "github.com/drand/drand/client"
 	"github.com/drand/drand/cmd/client/lib"
 	"github.com/drand/drand/common"
 	dhttp "github.com/drand/drand/http"
@@ -44,6 +46,7 @@ var metricsFlag = &cli.StringFlag{
 }
 
 // Relay a GRPC connection to an HTTP server.
+// nolint:gocyclo
 func Relay(c *cli.Context) error {
 	version := common.GetAppVersion()
 
@@ -66,15 +69,42 @@ func Relay(c *cli.Context) error {
 		return fmt.Errorf("failed to create rest handler: %w", err)
 	}
 
+	if c.IsSet(lib.HashFlag.Name) {
+		hash, err := hex.DecodeString(c.String(lib.HashFlag.Name))
+		if err != nil {
+			return fmt.Errorf("failed to decode hash flag: %w", err)
+		}
+		handler.HandlerDrand.CreateBeaconHandler(client, string(hash))
+	} else {
+		if c.IsSet(lib.HashListFlag.Name) {
+			hashList := c.StringSlice(lib.HashListFlag.Name)
+			for _, hashHex := range hashList {
+				hash, err := hex.DecodeString(hashHex)
+				if err != nil {
+					return fmt.Errorf("failed to decode hash flag: %w", err)
+				}
+
+				c, err := lib.Create(c, c.IsSet(metricsFlag.Name), dclient.WithChainHash(hash))
+				if err != nil {
+					return err
+				}
+
+				handler.HandlerDrand.CreateBeaconHandler(c, fmt.Sprintf("%x", hash))
+			}
+		} else {
+			return fmt.Errorf("must specify flag %s or %s", lib.HashFlag.Name, lib.HashListFlag.Name)
+		}
+	}
+
 	if c.IsSet(accessLogFlag.Name) {
 		logFile, err := os.OpenFile(c.String(accessLogFlag.Name), os.O_CREATE|os.O_APPEND|os.O_WRONLY, accessLogPermFolder)
 		if err != nil {
 			return fmt.Errorf("failed to open access log: %w", err)
 		}
 		defer logFile.Close()
-		handler = handlers.CombinedLoggingHandler(logFile, handler)
+		handler.HandlerHTTP = handlers.CombinedLoggingHandler(logFile, handler.HandlerHTTP)
 	} else {
-		handler = handlers.CombinedLoggingHandler(os.Stdout, handler)
+		handler.HandlerHTTP = handlers.CombinedLoggingHandler(os.Stdout, handler.HandlerHTTP)
 	}
 
 	bind := "localhost:0"
@@ -89,13 +119,13 @@ func Relay(c *cli.Context) error {
 	// jumpstart bootup
 	req, _ := http.NewRequest("GET", "/public/0", http.NoBody)
 	rr := httptest.NewRecorder()
-	handler.ServeHTTP(rr, req)
+	handler.HandlerHTTP.ServeHTTP(rr, req)
 	if rr.Code != http.StatusOK {
 		log.DefaultLogger().Warnw("", "binary", "relay", "startup failed", rr.Code)
 	}
 
 	fmt.Printf("Listening at %s\n", listener.Addr())
-	return http.Serve(listener, handler)
+	return http.Serve(listener, handler.HandlerHTTP)
 }
 
 func main() {
@@ -105,7 +135,7 @@ func main() {
 		Name:    "relay",
 		Version: version.String(),
 		Usage:   "Relay a Drand group to a public HTTP Rest API",
-		Flags:   append(lib.ClientFlags, listenFlag, accessLogFlag, metricsFlag),
+		Flags:   append(lib.ClientFlags, lib.HashListFlag, listenFlag, accessLogFlag, metricsFlag),
 		Action:  Relay,
 	}
 	cli.VersionPrinter = func(c *cli.Context) {