[FAB-10031] Add option to disable discovery auth cache

This change set adds an option to disable the authentication cache
of the discovery service.

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

Author: yacovm

discovery/authcache.go

@@ -36,6 +36,7 @@ type acSupport interface {
 }
 
 type authCacheConfig struct {
+	enabled bool
 	// maxCacheSize is the maximum size of the cache, after which
 	// a purge takes place
 	maxCacheSize int
@@ -64,6 +65,9 @@ func newAuthCache(s acSupport, conf authCacheConfig) *authCache {
 // Eligible returns whether the given peer is eligible for receiving
 // service from the discovery service for a given channel
 func (ac *authCache) EligibleForService(channel string, data common.SignedData) error {
+	if !ac.conf.enabled {
+		return ac.acSupport.EligibleForService(channel, data)
+	}
 	// Check whether we already have a cache for this channel
 	ac.RLock()
 	cache := ac.credentialCache[channel]

discovery/authcache_test.go

@@ -45,6 +45,24 @@ func (as *mockAcSupport) ConfigSequence(channel string) uint64 {
 	return as.Called(channel).Get(0).(uint64)
 }
 
+func TestCacheDisabled(t *testing.T) {
+	sd := common.SignedData{
+		Data:      []byte{1, 2, 3},
+		Identity:  []byte("authorizedIdentity"),
+		Signature: []byte{1, 2, 3},
+	}
+
+	as := &mockAcSupport{}
+	as.On("ConfigSequence", "foo").Return(uint64(0))
+	as.On("EligibleForService", "foo", sd).Return(nil)
+	cache := newAuthCache(as, authCacheConfig{maxCacheSize: 100, purgeRetentionRatio: 0.5})
+
+	// Call the cache twice with the same argument and ensure the call isn't cached
+	cache.EligibleForService("foo", sd)
+	cache.EligibleForService("foo", sd)
+	as.AssertNumberOfCalls(t, "EligibleForService", 2)
+}
+
 func TestCacheUsage(t *testing.T) {
 	as := &mockAcSupport{}
 	as.On("ConfigSequence", "foo").Return(uint64(0))
@@ -166,7 +184,7 @@ func TestCacheConfigChange(t *testing.T) {
 
 func TestCachePurgeCache(t *testing.T) {
 	as := &mockAcSupport{}
-	cache := newAuthCache(as, authCacheConfig{maxCacheSize: 4, purgeRetentionRatio: 0.75})
+	cache := newAuthCache(as, authCacheConfig{maxCacheSize: 4, purgeRetentionRatio: 0.75, enabled: true})
 	as.On("ConfigSequence", "mychannel").Return(uint64(0))
 
 	// Warm up the cache - attempt to place 4 identities to fill it up
@@ -258,5 +276,5 @@ func TestCacheConcurrentConfigUpdate(t *testing.T) {
 }
 
 func defaultConfig() authCacheConfig {
-	return authCacheConfig{maxCacheSize: defaultMaxCacheSize, purgeRetentionRatio: defaultRetentionRatio}
+	return authCacheConfig{maxCacheSize: defaultMaxCacheSize, purgeRetentionRatio: defaultRetentionRatio, enabled: true}
 }

discovery/service.go

@@ -9,6 +9,7 @@ package discovery
 import (
 	"bytes"
 	"encoding/hex"
+	"fmt"
 
 	"github.com/hyperledger/fabric/common/flogging"
 	"github.com/hyperledger/fabric/common/util"
@@ -45,17 +46,27 @@ type service struct {
 // Config defines the configuration of the discovery service
 type Config struct {
 	TLS                          bool
+	AuthCacheEnabled             bool
 	AuthCacheMaxSize             int
 	AuthCachePurgeRetentionRatio float64
 }
 
+// String returns a string representation of this Config
+func (c Config) String() string {
+	if c.AuthCacheEnabled {
+		return fmt.Sprintf("TLS: %t, authCacheMaxSize: %d, authCachePurgeRatio: %f", c.TLS, c.AuthCacheMaxSize, c.AuthCachePurgeRetentionRatio)
+	}
+	return fmt.Sprintf("TLS: %t, auth cache disabled", c.TLS)
+}
+
 // peerMapping maps PKI-IDs to Peers
 type peerMapping map[string]*discovery.Peer
 
 // NewService creates a new discovery service instance
 func NewService(config Config, sup Support) *service {
 	s := &service{
 		auth: newAuthCache(sup, authCacheConfig{
+			enabled:             config.AuthCacheEnabled,
 			maxCacheSize:        config.AuthCacheMaxSize,
 			purgeRetentionRatio: config.AuthCachePurgeRetentionRatio,
 		}),
@@ -69,6 +80,7 @@ func NewService(config Config, sup Support) *service {
 	s.localDispatchers = map[discovery.QueryType]dispatcher{
 		discovery.LocalMembershipQueryType: s.localMembershipResponse,
 	}
+	logger.Info("Created with config", config)
 	return s
 }
 

discovery/service_test.go

@@ -24,8 +24,24 @@ import (
 	"golang.org/x/net/context"
 )
 
+func TestConfig(t *testing.T) {
+	for _, trueOfFalse := range []bool{true, false} {
+		conf := Config{
+			AuthCacheEnabled:             trueOfFalse,
+			AuthCachePurgeRetentionRatio: 0.5,
+			AuthCacheMaxSize:             42,
+		}
+		service := NewService(conf, &mockSupport{})
+		assert.Equal(t, trueOfFalse, service.auth.conf.enabled)
+		assert.Equal(t, 42, service.auth.conf.maxCacheSize)
+		assert.Equal(t, 0.5, service.auth.conf.purgeRetentionRatio)
+	}
+}
+
 func TestService(t *testing.T) {
-	conf := Config{}
+	conf := Config{
+		AuthCacheEnabled: true,
+	}
 	ctx := context.Background()
 	req := &discovery.Request{
 		Authentication: &discovery.AuthInfo{

peer/node/start.go

@@ -433,6 +433,7 @@ func registerDiscoveryService(peerServer *comm.GRPCServer, mcs api.MessageCrypto
 	support := discsupport.NewDiscoverySupport(acl, gSup, ea, confSup, acl)
 	svc := discovery.NewService(discovery.Config{
 		TLS:                          peerServer.TLSEnabled(),
+		AuthCacheEnabled:             viper.GetBool("peer.discovery.authCacheEnabled"),
 		AuthCacheMaxSize:             viper.GetInt("peer.discovery.authCacheMaxSize"),
 		AuthCachePurgeRetentionRatio: viper.GetFloat64("peer.discovery.authCachePurgeRetentionRatio"),
 	}, support)

sampleconfig/core.yaml

@@ -422,6 +422,8 @@ peer:
     # what possible sets of peers satisfy the endorsement policy.
     discovery:
         enabled: true
+        # Whether the authentication cache is enabled or not.
+        authCacheEnabled: true
         # The maximum size of the cache, after which a purge takes place
         authCacheMaxSize: 1000
         # The proportion (0 to 1) of entries that remain in the cache after the cache is purged due to overpopulation