[FAB-10963] Discovery: filtering in membership queries

This change set adds support to the discovery service to filter
channel membership queries according to endorsement capabilities.

Now the client can pass invocation chains to the query, and the
discovery service would only return peers that satisfy the invocation
chain:
1) Peers that have chaincodes installed on them
2) Peers that are authorized by a given collection configuration

This change set also enhances the golang client implementation with
an ability to pass in the Peers() method - an invocation chain.

To preserve backward compatibility as much as possible,
the invocation chain is passed as a variadic parameter.

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

Author: yacovm

discovery/api.go

@@ -11,7 +11,7 @@ import (
 	"github.com/hyperledger/fabric/gossip/common"
 	"github.com/hyperledger/fabric/gossip/discovery"
 	common2 "github.com/hyperledger/fabric/protos/common"
-	discovery2 "github.com/hyperledger/fabric/protos/discovery"
+	discprotos "github.com/hyperledger/fabric/protos/discovery"
 )
 
 // AccessControlSupport checks if clients are eligible of being serviced
@@ -50,13 +50,19 @@ type GossipSupport interface {
 // for chaincodes
 type EndorsementSupport interface {
 	// PeersForEndorsement returns an EndorsementDescriptor for a given set of peers, channel, and chaincode
-	PeersForEndorsement(channel common.ChainID, interest *discovery2.ChaincodeInterest) (*discovery2.EndorsementDescriptor, error)
+	PeersForEndorsement(channel common.ChainID, interest *discprotos.ChaincodeInterest) (*discprotos.EndorsementDescriptor, error)
+
+	// PeersAuthorizedByCriteria returns the peers of the channel that are authorized by the given chaincode interest
+	// That is - taking in account if the chaincode(s) in the interest are installed on the peers, and also
+	// taking in account whether the peers are part of the collections of the chaincodes.
+	// If a nil interest, or an empty interest is passed - no filtering is done.
+	PeersAuthorizedByCriteria(chainID common.ChainID, interest *discprotos.ChaincodeInterest) (discovery.Members, error)
 }
 
 // ConfigSupport provides access to channel configuration
 type ConfigSupport interface {
 	// Config returns the channel's configuration
-	Config(channel string) (*discovery2.ConfigResult, error)
+	Config(channel string) (*discprotos.ConfigResult, error)
 }
 
 // Support defines an interface that allows the discovery service

discovery/client/api.go

@@ -40,7 +40,7 @@ type ChannelResponse interface {
 	Config() (*discovery.ConfigResult, error)
 
 	// Peers returns a response for a peer membership query, or error if something went wrong
-	Peers() ([]*Peer, error)
+	Peers(invocationChain ...*discovery.ChaincodeCall) ([]*Peer, error)
 
 	// Endorsers returns the response for an endorser query for a given
 	// chaincode in a given channel context, or error if something went wrong.

discovery/client/client.go

@@ -10,6 +10,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"fmt"
 	"math/rand"
 	"time"
 
@@ -105,24 +106,38 @@ func (req *Request) AddLocalPeersQuery() *Request {
 	req.Queries = append(req.Queries, &discovery.Query{
 		Query: q,
 	})
-	req.addQueryMapping(discovery.LocalMembershipQueryType, "")
+	var ic InvocationChain
+	req.addQueryMapping(discovery.LocalMembershipQueryType, channnelAndInvocationChain("", ic))
 	return req
 }
 
 // AddPeersQuery adds to the request a peer query
-func (req *Request) AddPeersQuery() *Request {
+func (req *Request) AddPeersQuery(invocationChain ...*discovery.ChaincodeCall) *Request {
 	ch := req.lastChannel
 	q := &discovery.Query_PeerQuery{
-		PeerQuery: &discovery.PeerMembershipQuery{},
+		PeerQuery: &discovery.PeerMembershipQuery{
+			Filter: &discovery.ChaincodeInterest{
+				Chaincodes: invocationChain,
+			},
+		},
 	}
 	req.Queries = append(req.Queries, &discovery.Query{
 		Channel: ch,
 		Query:   q,
 	})
-	req.addQueryMapping(discovery.PeerMembershipQueryType, ch)
+	var ic InvocationChain
+	if len(invocationChain) > 0 {
+		ic = InvocationChain(invocationChain)
+	}
+	req.addChaincodeQueryMapping([]InvocationChain{ic})
+	req.addQueryMapping(discovery.PeerMembershipQueryType, channnelAndInvocationChain(ch, ic))
 	return req
 }
 
+func channnelAndInvocationChain(ch string, ic InvocationChain) string {
+	return fmt.Sprintf("%s %s", ch, ic.String())
+}
+
 // OfChannel sets the next queries added to be in the given channel's context
 func (req *Request) OfChannel(ch string) *Request {
 	req.lastChannel = ch
@@ -204,7 +219,7 @@ type channelResponse struct {
 func (cr *channelResponse) Config() (*discovery.ConfigResult, error) {
 	res, exists := cr.response[key{
 		queryType: discovery.ConfigQueryType,
-		channel:   cr.channel,
+		k:         cr.channel,
 	}]
 
 	if !exists {
@@ -218,11 +233,12 @@ func (cr *channelResponse) Config() (*discovery.ConfigResult, error) {
 	return nil, res.(error)
 }
 
-func parsePeers(queryType discovery.QueryType, r response, channel string) ([]*Peer, error) {
-	res, exists := r[key{
+func parsePeers(queryType discovery.QueryType, r response, channel string, invocationChain ...*discovery.ChaincodeCall) ([]*Peer, error) {
+	peerKeys := key{
 		queryType: queryType,
-		channel:   channel,
-	}]
+		k:         fmt.Sprintf("%s %s", channel, InvocationChain(invocationChain).String()),
+	}
+	res, exists := r[peerKeys]
 
 	if !exists {
 		return nil, ErrNotFound
@@ -235,24 +251,24 @@ func parsePeers(queryType discovery.QueryType, r response, channel string) ([]*P
 	return nil, res.(error)
 }
 
-func (cr *channelResponse) Peers() ([]*Peer, error) {
-	return parsePeers(discovery.PeerMembershipQueryType, cr.response, cr.channel)
+func (cr *channelResponse) Peers(invocationChain ...*discovery.ChaincodeCall) ([]*Peer, error) {
+	return parsePeers(discovery.PeerMembershipQueryType, cr.response, cr.channel, invocationChain...)
 }
 
 func (cr *channelResponse) Endorsers(invocationChain InvocationChain, ps PrioritySelector, ef ExclusionFilter) (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{
 		queryType: discovery.ChaincodeQueryType,
-		channel:   cr.channel,
+		k:         cr.channel,
 	}]; exists {
 		return nil, err.(error)
 	}
 
 	// Else, the service returned a response that isn't an error
 	res, exists := cr.response[key{
 		queryType:       discovery.ChaincodeQueryType,
-		channel:         cr.channel,
+		k:               cr.channel,
 		invocationChain: invocationChain.String(),
 	}]
 
@@ -306,7 +322,7 @@ func (resp response) ForChannel(ch string) ChannelResponse {
 
 type key struct {
 	queryType       discovery.QueryType
-	channel         string
+	k               string
 	invocationChain string
 }
 
@@ -318,7 +334,7 @@ func (req *Request) computeResponse(r *discovery.Response) (response, error) {
 		case discovery.ConfigQueryType:
 			err = resp.mapConfig(channel2index, r)
 		case discovery.ChaincodeQueryType:
-			err = resp.mapEndorsers(channel2index, r, req.queryMapping, req.invocationChainMapping)
+			err = resp.mapEndorsers(channel2index, r, req.invocationChainMapping)
 		case discovery.PeerMembershipQueryType:
 			err = resp.mapPeerMembership(channel2index, r, discovery.PeerMembershipQueryType)
 		case discovery.LocalMembershipQueryType:
@@ -340,7 +356,7 @@ func (resp response) mapConfig(channel2index map[string]int, r *discovery.Respon
 		}
 		key := key{
 			queryType: discovery.ConfigQueryType,
-			channel:   ch,
+			k:         ch,
 		}
 
 		if err != nil {
@@ -353,15 +369,16 @@ func (resp response) mapConfig(channel2index map[string]int, r *discovery.Respon
 	return nil
 }
 
-func (resp response) mapPeerMembership(channel2index map[string]int, r *discovery.Response, qt discovery.QueryType) error {
-	for ch, index := range channel2index {
+func (resp response) mapPeerMembership(key2Index map[string]int, r *discovery.Response, qt discovery.QueryType) error {
+	for k, index := range key2Index {
 		membersRes, err := r.MembershipAt(index)
 		if membersRes == nil && err == nil {
 			return errors.Errorf("expected QueryResult of either PeerMembershipResult or Error but got %v instead", r.Results[index])
 		}
+
 		key := key{
 			queryType: qt,
-			channel:   ch,
+			k:         k,
 		}
 
 		if err != nil {
@@ -418,7 +435,6 @@ func isStateInfoExpected(qt discovery.QueryType) bool {
 func (resp response) mapEndorsers(
 	channel2index map[string]int,
 	r *discovery.Response,
-	queryMapping map[discovery.QueryType]map[string]int,
 	chaincodeQueryMapping map[int][]InvocationChain) error {
 	for ch, index := range channel2index {
 		ccQueryRes, err := r.EndorsersAt(index)
@@ -429,7 +445,7 @@ func (resp response) mapEndorsers(
 		if err != nil {
 			key := key{
 				queryType: discovery.ChaincodeQueryType,
-				channel:   ch,
+				k:         ch,
 			}
 			resp[key] = errors.New(err.Content)
 			continue
@@ -453,7 +469,7 @@ func (resp response) mapEndorsersOfChannel(ccRs *discovery.ChaincodeQueryResult,
 		}
 		key := key{
 			queryType:       discovery.ChaincodeQueryType,
-			channel:         channel,
+			k:               channel,
 			invocationChain: invocationChain[i].String(),
 		}
 

discovery/client/client_test.go

@@ -27,7 +27,7 @@ import (
 	"github.com/hyperledger/fabric/discovery/endorsement"
 	"github.com/hyperledger/fabric/gossip/api"
 	gossipcommon "github.com/hyperledger/fabric/gossip/common"
-	discovery3 "github.com/hyperledger/fabric/gossip/discovery"
+	gdisc "github.com/hyperledger/fabric/gossip/discovery"
 	"github.com/hyperledger/fabric/protos/common"
 	"github.com/hyperledger/fabric/protos/discovery"
 	"github.com/hyperledger/fabric/protos/gossip"
@@ -101,7 +101,7 @@ var (
 		},
 	}
 
-	channelPeersWithChaincodes = discovery3.Members{
+	channelPeersWithChaincodes = gdisc.Members{
 		newPeer(0, stateInfoMessage(cc, cc2), propertiesWithChaincodes).NetworkMember,
 		newPeer(1, stateInfoMessage(cc, cc2), propertiesWithChaincodes).NetworkMember,
 		newPeer(2, stateInfoMessage(cc, cc2), propertiesWithChaincodes).NetworkMember,
@@ -112,7 +112,7 @@ var (
 		newPeer(7, stateInfoMessage(cc, cc2), propertiesWithChaincodes).NetworkMember,
 	}
 
-	channelPeersWithoutChaincodes = discovery3.Members{
+	channelPeersWithoutChaincodes = gdisc.Members{
 		newPeer(0, stateInfoMessage(), nil).NetworkMember,
 		newPeer(1, stateInfoMessage(), nil).NetworkMember,
 		newPeer(2, stateInfoMessage(), nil).NetworkMember,
@@ -123,7 +123,7 @@ var (
 		newPeer(7, stateInfoMessage(), nil).NetworkMember,
 	}
 
-	membershipPeers = discovery3.Members{
+	membershipPeers = gdisc.Members{
 		newPeer(0, aliveMessage(0), nil).NetworkMember,
 		newPeer(1, aliveMessage(1), nil).NetworkMember,
 		newPeer(2, aliveMessage(2), nil).NetworkMember,
@@ -422,6 +422,22 @@ func TestClient(t *testing.T) {
 		assert.Contains(t, expectedOrgCombinations2, getMSPs(endorsers))
 	})
 
+	t.Run("Peer membership query with collections and chaincodes", func(t *testing.T) {
+		sup.On("PeersOfChannel").Return(channelPeersWithChaincodes).Once()
+		interest := ccCall("mycc2")
+		interest[0].CollectionNames = append(interest[0].CollectionNames, "col")
+		req = NewRequest().OfChannel("mychannel").AddPeersQuery(interest...)
+		r, err = cl.Send(ctx, req, authInfo)
+		assert.NoError(t, err)
+		mychannel := r.ForChannel("mychannel")
+		peers, err := mychannel.Peers(interest...)
+		assert.NoError(t, err)
+		// We should see all peers that aren't in ORG A since it's not part of the collection
+		for _, p := range peers {
+			assert.NotEqual(t, "A", p.MSPID)
+		}
+		assert.Len(t, peers, 6)
+	})
 }
 
 func TestUnableToSign(t *testing.T) {
@@ -692,6 +708,8 @@ func (pf *policyFetcher) PolicyByChaincode(channel string, cc string) policies.I
 
 type endorsementAnalyzer interface {
 	PeersForEndorsement(chainID gossipcommon.ChainID, interest *discovery.ChaincodeInterest) (*discovery.EndorsementDescriptor, error)
+
+	PeersAuthorizedByCriteria(chainID gossipcommon.ChainID, interest *discovery.ChaincodeInterest) (gdisc.Members, error)
 }
 
 type inquireablePolicy struct {
@@ -734,7 +752,7 @@ func peerIdentity(mspID string, i int) api.PeerIdentityInfo {
 type peerInfo struct {
 	identity api.PeerIdentityType
 	pkiID    gossipcommon.PKIidType
-	discovery3.NetworkMember
+	gdisc.NetworkMember
 }
 
 func aliveMessage(id int) *gossip.Envelope {
@@ -778,7 +796,7 @@ func newPeer(i int, env *gossip.Envelope, properties *gossip.Properties) *peerIn
 	return &peerInfo{
 		pkiID:    gossipcommon.PKIidType(p),
 		identity: api.PeerIdentityType(p),
-		NetworkMember: discovery3.NetworkMember{
+		NetworkMember: gdisc.NetworkMember{
 			PKIid:            gossipcommon.PKIidType(p),
 			Endpoint:         p,
 			InternalEndpoint: p,
@@ -808,18 +826,22 @@ func (*mockSupport) ChannelExists(channel string) bool {
 	return true
 }
 
-func (ms *mockSupport) PeersOfChannel(gossipcommon.ChainID) discovery3.Members {
-	return ms.Called().Get(0).(discovery3.Members)
+func (ms *mockSupport) PeersOfChannel(gossipcommon.ChainID) gdisc.Members {
+	return ms.Called().Get(0).(gdisc.Members)
 }
 
-func (ms *mockSupport) Peers() discovery3.Members {
-	return ms.Called().Get(0).(discovery3.Members)
+func (ms *mockSupport) Peers() gdisc.Members {
+	return ms.Called().Get(0).(gdisc.Members)
 }
 
 func (ms *mockSupport) PeersForEndorsement(channel gossipcommon.ChainID, interest *discovery.ChaincodeInterest) (*discovery.EndorsementDescriptor, error) {
 	return ms.endorsementAnalyzer.PeersForEndorsement(channel, interest)
 }
 
+func (ms *mockSupport) PeersAuthorizedByCriteria(channel gossipcommon.ChainID, interest *discovery.ChaincodeInterest) (gdisc.Members, error) {
+	return ms.endorsementAnalyzer.PeersAuthorizedByCriteria(channel, interest)
+}
+
 func (*mockSupport) EligibleForService(channel string, data common.SignedData) error {
 	return nil
 }