[FAB-9443] Enhance discovery client expressiveness

This change set enhances the discovery client reference API with
abilities that enable the user more control over how the endorsers
are selected, i.e:

 - Prefer endorsers with a higher ledger height
 - Exclude certain peers from being chosen, which is useful
   for cases where the SDK detected an unreachable peer and when
   it requests from the discovery client object an endorser set
   it would want the discovery service to skip selecting that peer.

Also, this change set:
- Adds input validation to StateInfo and AliveMessages returned from peers
- Fixes golint warnings

Change-Id: Ib8de21c91ecf7706ab42ec898e00d0e2f40af8be
Signed-off-by: yacovm <yacovm@il.ibm.com>

Author: yacovm

discovery/client/api.go

@@ -10,7 +10,6 @@ import (
 	"github.com/hyperledger/fabric/protos/discovery"
 	"github.com/hyperledger/fabric/protos/gossip"
 	"github.com/pkg/errors"
-	"golang.org/x/net/context"
 	"google.golang.org/grpc"
 )
 
@@ -26,12 +25,6 @@ type Signer func(msg []byte) ([]byte, error)
 // Dialer connects to the server
 type Dialer func() (*grpc.ClientConn, error)
 
-// Client defines the client-side API of the discovery service
-type Client interface {
-	// Send sends the Request and returns the response, or error on failure
-	Send(context.Context, *Request) (Response, error)
-}
-
 // Response aggregates several responses from the discovery service
 type Response interface {
 	// ForChannel returns a ChannelResponse in the context of a given channel
@@ -50,7 +43,8 @@ type ChannelResponse interface {
 	// chaincode in a given channel context, or error if something went wrong.
 	// The method returns a random set of endorsers, such that signatures from all of them
 	// combined, satisfy the endorsement policy.
-	Endorsers(string) (Endorsers, error)
+	// The selection is based on the given selection hint
+	Endorsers(cc string, hint Selector) (Endorsers, error)
 }
 
 // Endorsers defines a set of peers that are sufficient

discovery/client/client.go

@@ -13,8 +13,8 @@ import (
 	"time"
 
 	"github.com/golang/protobuf/proto"
-	"github.com/hyperledger/fabric/gossip/util"
 	"github.com/hyperledger/fabric/protos/discovery"
+	"github.com/hyperledger/fabric/protos/gossip"
 	"github.com/hyperledger/fabric/protos/msp"
 	"github.com/pkg/errors"
 )
@@ -23,7 +23,8 @@ var (
 	configTypes = []discovery.QueryType{discovery.ConfigQueryType, discovery.PeerMembershipQueryType, discovery.ChaincodeQueryType}
 )
 
-type client struct {
+// Client interacts with the discovery server
+type Client struct {
 	lastRequest      []byte
 	lastSignature    []byte
 	createConnection Dialer
@@ -109,7 +110,7 @@ func (req *Request) addQueryMapping(queryType discovery.QueryType, key string) {
 }
 
 // Send sends the request and returns the response, or error on failure
-func (c *client) Send(ctx context.Context, req *Request) (Response, error) {
+func (c *Client) Send(ctx context.Context, req *Request) (Response, error) {
 	req.Authentication = c.authInfo
 	payload, err := proto.Marshal(req.Request)
 	if err != nil {
@@ -119,7 +120,7 @@ func (c *client) Send(ctx context.Context, req *Request) (Response, error) {
 	sig := c.lastSignature
 	// Only sign the Request if it is different than the previous Request sent.
 	// Otherwise, use the last signature from the previous send.
-	// This is not only to save CPU cycles in the client-side,
+	// This is not only to save CPU cycles in the Client-side,
 	// but also for the server side to be able to memoize the signature verification.
 	// We have the use the previous signature, because many signature schemes are not deterministic.
 	if !bytes.Equal(c.lastRequest, payload) {
@@ -201,7 +202,7 @@ func (cr *channelResponse) Peers() ([]*Peer, error) {
 	return nil, res.(error)
 }
 
-func (cr *channelResponse) Endorsers(cc string) (Endorsers, error) {
+func (cr *channelResponse) Endorsers(cc string, s Selector) (Endorsers, error) {
 	// If we have a key that has no chaincode field,
 	// it means it's an error returned from the service
 	if err, exists := cr.response[key{
@@ -224,18 +225,33 @@ func (cr *channelResponse) Endorsers(cc string) (Endorsers, error) {
 
 	desc := res.(*endorsementDescriptor)
 	rand.Seed(time.Now().Unix())
-	randomLayoutIndex := rand.Intn(len(desc.layouts))
-	layout := desc.layouts[randomLayoutIndex]
+	// We iterate over all layouts to find one that we have enough peers to select
+	for _, index := range rand.Perm(len(desc.layouts)) {
+		layout := desc.layouts[index]
+		endorsers, canLayoutBeSatisfied := selectPeersForLayout(desc.endorsersByGroups, layout, s)
+		if canLayoutBeSatisfied {
+			return endorsers, nil
+		}
+	}
+	return nil, errors.New("no endorsement combination can be satisfied")
+}
+
+func selectPeersForLayout(endorsersByGroups map[string][]*Peer, layout map[string]int, s Selector) (Endorsers, bool) {
 	var endorsers []*Peer
 	for grp, count := range layout {
-		endorsersOfGrp := randomEndorsers(count, desc.endorsersByGroups[grp])
+		shuffledEndorsers := Endorsers(endorsersByGroups[grp]).Shuffle()
+		endorsersOfGrp := s.Select(shuffledEndorsers)
+		// We couldn't select enough peers for this layout because the current group
+		// requires more peers than we have available to be selected
 		if len(endorsersOfGrp) < count {
-			return nil, errors.Errorf("layout has a group that requires at least %d peers, but only %d peers are known", count, len(endorsersOfGrp))
+			return nil, false
 		}
+		endorsersOfGrp = endorsersOfGrp[:count]
 		endorsers = append(endorsers, endorsersOfGrp...)
 	}
-
-	return endorsers, nil
+	// The current (randomly chosen) layout can be satisfied, so return it
+	// instead of checking the next one.
+	return endorsers, true
 }
 
 func (resp response) ForChannel(ch string) ChannelResponse {
@@ -330,6 +346,12 @@ func peersForChannel(membersRes *discovery.PeerMembershipResult) ([]*Peer, error
 			if err != nil {
 				return nil, errors.Wrap(err, "failed unmarshaling stateInfo message")
 			}
+			if err := validateAliveMessage(aliveMsg); err != nil {
+				return nil, errors.Wrap(err, "failed validating alive message")
+			}
+			if err := validateStateInfoMessage(stateInfoMsg); err != nil {
+				return nil, errors.Wrap(err, "failed validating stateInfo message")
+			}
 			peers = append(peers, &Peer{
 				MSPID:            org,
 				Identity:         peer.Identity,
@@ -426,35 +448,63 @@ func endorser(peer *discovery.Peer, chaincode, channel string) (*Peer, error) {
 	if err != nil {
 		return nil, errors.Wrap(err, "failed unmarshaling gossip envelope to state info message")
 	}
-	sId := &msp.SerializedIdentity{}
-	if err := proto.Unmarshal(peer.Identity, sId); err != nil {
+	if err := validateAliveMessage(aliveMsg); err != nil {
+		return nil, errors.Wrap(err, "failed validating alive message")
+	}
+	if err := validateStateInfoMessage(stateInfMsg); err != nil {
+		return nil, errors.Wrap(err, "failed validating stateInfo message")
+	}
+	sID := &msp.SerializedIdentity{}
+	if err := proto.Unmarshal(peer.Identity, sID); err != nil {
 		return nil, errors.Wrap(err, "failed unmarshaling peer's identity")
 	}
 	return &Peer{
 		Identity:         peer.Identity,
 		StateInfoMessage: stateInfMsg,
 		AliveMessage:     aliveMsg,
-		MSPID:            sId.Mspid,
+		MSPID:            sID.Mspid,
 	}, nil
 }
 
-func randomEndorsers(count int, totalPeers []*Peer) Endorsers {
-	var endorsers []*Peer
-	for _, index := range util.GetRandomIndices(count, len(totalPeers)-1) {
-		endorsers = append(endorsers, totalPeers[index])
-	}
-	return endorsers
-}
-
 type endorsementDescriptor struct {
 	endorsersByGroups map[string][]*Peer
 	layouts           []map[string]int
 }
 
-func NewClient(createConnection Dialer, authInfo *discovery.AuthInfo, s Signer) *client {
-	return &client{
+// NewClient creates a new Client instance
+func NewClient(createConnection Dialer, authInfo *discovery.AuthInfo, s Signer) *Client {
+	return &Client{
 		createConnection: createConnection,
 		authInfo:         authInfo,
 		signRequest:      s,
 	}
 }
+
+func validateAliveMessage(message *gossip.SignedGossipMessage) error {
+	am := message.GetAliveMsg()
+	if am == nil {
+		return errors.New("message isn't an alive message")
+	}
+	m := am.Membership
+	if m == nil {
+		return errors.New("membership is empty")
+	}
+	if am.Timestamp == nil {
+		return errors.New("timestamp is nil")
+	}
+	return nil
+}
+
+func validateStateInfoMessage(message *gossip.SignedGossipMessage) error {
+	si := message.GetStateInfo()
+	if si == nil {
+		return errors.New("message isn't a stateInfo message")
+	}
+	if si.Timestamp == nil {
+		return errors.New("timestamp is nil")
+	}
+	if si.Properties == nil {
+		return errors.New("properties is nil")
+	}
+	return nil
+}

discovery/client/client_test.go

@@ -15,6 +15,7 @@ import (
 	"path/filepath"
 	"strconv"
 	"testing"
+	"time"
 
 	"github.com/golang/protobuf/proto"
 	"github.com/hyperledger/fabric/common/cauthdsl"
@@ -307,7 +308,7 @@ func TestClient(t *testing.T) {
 	assert.Equal(t, ErrNotFound, err)
 	assert.Nil(t, peers)
 
-	endorsers, err := fakeChannel.Endorsers("mycc")
+	endorsers, err := fakeChannel.Endorsers("mycc", Selector{})
 	assert.Equal(t, ErrNotFound, err)
 	assert.Nil(t, endorsers)
 
@@ -327,7 +328,7 @@ func TestClient(t *testing.T) {
 	// We should see all peers as provided above
 	assert.Len(t, peers, 8)
 
-	endorsers, err = mychannel.Endorsers("mycc")
+	endorsers, err = mychannel.Endorsers("mycc", Selector{})
 	// However, since we didn't provide any chaincodes to these peers - the server shouldn't
 	// be able to construct the descriptor.
 	// Just check that the appropriate error is returned, and nothing crashes.
@@ -347,7 +348,7 @@ func TestClient(t *testing.T) {
 	assert.Len(t, peers, 8)
 
 	// We should get a valid endorsement descriptor from the service
-	endorsers, err = mychannel.Endorsers("mycc")
+	endorsers, err = mychannel.Endorsers("mycc", Selector{})
 	assert.NoError(t, err)
 	// The combinations of endorsers should be in the expected combinations
 	assert.Contains(t, expectedOrgCombinations, getMSPs(endorsers))
@@ -440,7 +441,7 @@ func TestBadResponses(t *testing.T) {
 	r, err = cl.Send(ctx, req)
 	assert.NoError(t, err)
 	mychannel := r.ForChannel("mychannel")
-	endorsers, err := mychannel.Endorsers("mycc")
+	endorsers, err := mychannel.Endorsers("mycc", Selector{})
 	assert.Nil(t, endorsers)
 	assert.Equal(t, ErrNotFound, err)
 
@@ -472,9 +473,9 @@ func TestBadResponses(t *testing.T) {
 	r, err = cl.Send(ctx, req)
 	assert.NoError(t, err)
 	mychannel = r.ForChannel("mychannel")
-	endorsers, err = mychannel.Endorsers("mycc")
+	endorsers, err = mychannel.Endorsers("mycc", Selector{})
 	assert.Nil(t, endorsers)
-	assert.Contains(t, err.Error(), "layout has a group that requires at least 2 peers, but only 0 peers are known")
+	assert.Contains(t, err.Error(), "no endorsement combination can be satisfied")
 
 	// Scenario VI: discovery service sends back a layout that has a group that doesn't have a matching peer set
 	svc.On("Discover").Return(&discovery.Response{
@@ -491,6 +492,52 @@ func TestBadResponses(t *testing.T) {
 	assert.Empty(t, r)
 }
 
+func TestValidateAliveMessage(t *testing.T) {
+	am := aliveMessage(1)
+	msg, _ := am.ToGossipMessage()
+
+	// Scenario I: Valid alive message
+	assert.NoError(t, validateAliveMessage(msg))
+
+	// Scenario II: Nullify timestamp
+	msg.GetAliveMsg().Timestamp = nil
+	err := validateAliveMessage(msg)
+	assert.Equal(t, "timestamp is nil", err.Error())
+
+	// Scenario III: Nullify membership
+	msg.GetAliveMsg().Membership = nil
+	err = validateAliveMessage(msg)
+	assert.Equal(t, "membership is empty", err.Error())
+
+	// Scenario IV: Nullify the entire alive message part
+	msg.Content = nil
+	err = validateAliveMessage(msg)
+	assert.Equal(t, "message isn't an alive message", err.Error())
+
+}
+
+func TestValidateStateInfoMessage(t *testing.T) {
+	si := stateInfoWithHeight(100)
+
+	// Scenario I: Valid state info message
+	assert.NoError(t, validateStateInfoMessage(si))
+
+	// Scenario II: Nullify properties
+	si.GetStateInfo().Properties = nil
+	err := validateStateInfoMessage(si)
+	assert.Equal(t, "properties is nil", err.Error())
+
+	// Scenario III: Nullify timestamp
+	si.GetStateInfo().Timestamp = nil
+	err = validateStateInfoMessage(si)
+	assert.Equal(t, "timestamp is nil", err.Error())
+
+	// Scenario IV: Nullify the state info message part
+	si.Content = nil
+	err = validateStateInfoMessage(si)
+	assert.Equal(t, "message isn't a stateInfo message", err.Error())
+}
+
 func getMSP(peer *Peer) string {
 	endpoint := peer.AliveMessage.GetAliveMsg().Membership.Endpoint
 	id, _ := strconv.ParseInt(endpoint[1:], 10, 64)
@@ -573,11 +620,11 @@ func (ip *inquireablePolicy) SatisfiedBy() []policies.PrincipalSet {
 
 func peerIdentity(mspID string, i int) api.PeerIdentityInfo {
 	p := []byte(fmt.Sprintf("p%d", i))
-	sId := &msp.SerializedIdentity{
+	sID := &msp.SerializedIdentity{
 		Mspid:   mspID,
 		IdBytes: p,
 	}
-	b, _ := proto.Marshal(sId)
+	b, _ := proto.Marshal(sID)
 	return api.PeerIdentityInfo{
 		Identity:     api.PeerIdentityType(b),
 		PKIId:        gossipcommon.PKIidType(p),
@@ -595,6 +642,10 @@ func aliveMessage(id int) *gossip.Envelope {
 	g := &gossip.GossipMessage{
 		Content: &gossip.GossipMessage_AliveMsg{
 			AliveMsg: &gossip.AliveMessage{
+				Timestamp: &gossip.PeerTime{
+					SeqNum: uint64(id),
+					IncNum: uint64(time.Now().UnixNano()),
+				},
 				Membership: &gossip.Member{
 					Endpoint: fmt.Sprintf("p%d", id),
 				},
@@ -609,6 +660,10 @@ func stateInfoMessage(chaincodes ...*gossip.Chaincode) *gossip.Envelope {
 	g := &gossip.GossipMessage{
 		Content: &gossip.GossipMessage_StateInfo{
 			StateInfo: &gossip.StateInfo{
+				Timestamp: &gossip.PeerTime{
+					SeqNum: 5,
+					IncNum: uint64(time.Now().UnixNano()),
+				},
 				Properties: &gossip.Properties{
 					Chaincodes: chaincodes,
 				},