[FABG-762] Make event service config channel-specific

Added an event service policy to the channel policies.

Change-Id: Idb4285899708986de4c30c77259dfe0701d782fe
Signed-off-by: Bob Stasyszyn <Bob.Stasyszyn@securekey.com>

pkg/common/providers/fab/network.go

@@ -8,6 +8,7 @@ package fab
 
 import (
 	"crypto/x509"
+	"time"
 
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/errors/retry"
 )
@@ -37,6 +38,7 @@ type ChannelPolicies struct {
 	QueryChannelConfig QueryChannelConfigPolicy
 	Discovery          DiscoveryPolicy
 	Selection          SelectionPolicy
+	EventService       EventServicePolicy
 }
 
 //QueryChannelConfigPolicy defines policy for channelConfigBlock
@@ -140,3 +142,32 @@ type CertKeyPair struct {
 	Cert []byte
 	Key  []byte
 }
+
+// EventServicePolicy specifies the policy for the event service
+type EventServicePolicy struct {
+	// ResolverStrategy returns the peer resolver strategy to use when connecting to a peer
+	// Default: MinBlockHeightPeerResolver
+	ResolverStrategy ResolverStrategy
+
+	// Balancer is the balancer to use when choosing a peer to connect to
+	Balancer BalancerType
+
+	// BlockHeightLagThreshold returns the block height lag threshold. This value is used for choosing a peer
+	// to connect to. If a peer is lagging behind the most up-to-date peer by more than the given number of
+	// blocks then it will be excluded from selection.
+	// If set to 0 then only the most up-to-date peers are considered.
+	// If set to -1 then all peers (regardless of block height) are considered for selection.
+	BlockHeightLagThreshold int
+
+	// ReconnectBlockHeightLagThreshold - if >0 then the event client will disconnect from the peer if the peer's
+	// block height falls behind the specified number of blocks and will reconnect to a better performing peer.
+	// If set to 0 (default) then the peer will not disconnect based on block height.
+	// NOTE: Setting this value too low may cause the event client to disconnect/reconnect too frequently, thereby
+	// affecting performance.
+	ReconnectBlockHeightLagThreshold int
+
+	// PeerMonitorPeriod is the period in which the connected peer is monitored to see if
+	// the event client should disconnect from it and reconnect to another peer.
+	// If set to 0 then the peer will not be monitored and will not be disconnected.
+	PeerMonitorPeriod time.Duration
+}

pkg/common/providers/fab/provider.go

@@ -103,7 +103,6 @@ type EndpointConfig interface {
 	ChannelPeers(name string) []ChannelPeer
 	ChannelOrderers(name string) []OrdererConfig
 	TLSCACertPool() CertPool
-	EventServiceConfig() EventServiceConfig
 	TLSClientCerts() []tls.Certificate
 	CryptoConfigPath() string
 }
@@ -126,35 +125,6 @@ const (
 	PreferOrgStrategy ResolverStrategy = "PreferOrg"
 )
 
-// EventServiceConfig specifies configuration options for the event service
-type EventServiceConfig interface {
-	// ResolverStrategy returns the peer resolver strategy to use when connecting to a peer
-	// Default: MinBlockHeightPeerResolver
-	ResolverStrategy() ResolverStrategy
-
-	// Balancer is the balancer to use when choosing a peer to connect to
-	Balancer() BalancerType
-
-	// BlockHeightLagThreshold returns the block height lag threshold. This value is used for choosing a peer
-	// to connect to. If a peer is lagging behind the most up-to-date peer by more than the given number of
-	// blocks then it will be excluded from selection.
-	// If set to 0 then only the most up-to-date peers are considered.
-	// If set to -1 then all peers (regardless of block height) are considered for selection.
-	BlockHeightLagThreshold() int
-
-	// ReconnectBlockHeightLagThreshold - if >0 then the event client will disconnect from the peer if the peer's
-	// block height falls behind the specified number of blocks and will reconnect to a better performing peer.
-	// If set to 0 (default) then the peer will not disconnect based on block height.
-	// NOTE: Setting this value too low may cause the event client to disconnect/reconnect too frequently, thereby
-	// affecting performance.
-	ReconnectBlockHeightLagThreshold() int
-
-	// PeerMonitorPeriod is the period in which the connected peer is monitored to see if
-	// the event client should disconnect from it and reconnect to another peer.
-	// If set to 0 then the peer will not be monitored and will not be disconnected.
-	PeerMonitorPeriod() time.Duration
-}
-
 // TimeoutType enumerates the different types of outgoing connections
 type TimeoutType int
 

pkg/core/config/lookup/lookup.go

@@ -23,7 +23,7 @@ func New(coreBackends ...core.ConfigBackend) *ConfigLookup {
 
 //unmarshalOpts opts for unmarshal key function
 type unmarshalOpts struct {
-	hookFunc mapstructure.DecodeHookFunc
+	hooks []mapstructure.DecodeHookFunc
 }
 
 // UnmarshalOption describes a functional parameter unmarshaling
@@ -33,7 +33,7 @@ type UnmarshalOption func(o *unmarshalOpts)
 // for unmarshaling
 func WithUnmarshalHookFunction(hookFunction mapstructure.DecodeHookFunc) UnmarshalOption {
 	return func(o *unmarshalOpts) {
-		o.hookFunc = hookFunction
+		o.hooks = append(o.hooks, hookFunction)
 	}
 }
 
@@ -110,7 +110,8 @@ func (c *ConfigLookup) UnmarshalKey(key string, rawVal interface{}, opts ...Unma
 	}
 
 	//mandatory hook func
-	hookFn := mapstructure.StringToTimeDurationHookFunc()
+	var unmarshalHooks []mapstructure.DecodeHookFunc
+	unmarshalHooks = append(unmarshalHooks, mapstructure.StringToTimeDurationHookFunc())
 
 	//check for opts
 	unmarshalOpts := unmarshalOpts{}
@@ -119,9 +120,7 @@ func (c *ConfigLookup) UnmarshalKey(key string, rawVal interface{}, opts ...Unma
 	}
 
 	//compose multiple hook funcs to one if found in opts
-	if unmarshalOpts.hookFunc != nil {
-		hookFn = mapstructure.ComposeDecodeHookFunc(hookFn, unmarshalOpts.hookFunc)
-	}
+	hookFn := mapstructure.ComposeDecodeHookFunc(append(unmarshalHooks, unmarshalOpts.hooks...)...)
 
 	//build decoder
 	decoder, err := mapstructure.NewDecoder(&mapstructure.DecoderConfig{

pkg/core/config/testdata/config_test.yaml

@@ -137,6 +137,45 @@ channels:
         #to a lower priority list of peers which will be sorted according to block height.
         #Note: This property only applies to BlockHeightPriority sorter.
         BlockHeightLagThreshold: 5
+      #[Optional] options for event service
+      eventService:
+        # [Optional] resolverStrategy specifies the peer resolver strategy to use when connecting to a peer
+        # Possible values: [PreferOrg (default), MinBlockHeight, Balanced]
+        #
+        # PreferOrg:
+        #   Determines which peers are suitable based on block height lag threshold, although will prefer the peers in the
+        #   current org (as long as their block height is above a configured threshold). If none of the peers from the current org
+        #   are suitable then a peer from another org is chosen.
+        # MinBlockHeight:
+        #   Chooses the best peer according to a block height lag threshold. The maximum block height of all peers is
+        #   determined and the peers whose block heights are under the maximum height but above a provided "lag" threshold are load
+        #   balanced. The other peers are not considered.
+        # Balanced:
+        #   Chooses peers using the configured balancer.
+        resolverStrategy: MinBlockHeight
+        # [Optional] balancer is the balancer to use when choosing a peer to connect to
+        # Possible values: [Random (default), RoundRobin]
+        balancer: RoundRobin
+        # [Optional] blockHeightLagThreshold sets the block height lag threshold. This value is used for choosing a peer
+        # to connect to. If a peer is lagging behind the most up-to-date peer by more than the given number of
+        # blocks then it will be excluded from selection.
+        # If set to 0 then only the most up-to-date peers are considered.
+        # If set to -1 then all peers (regardless of block height) are considered for selection.
+        # Default: 5
+        blockHeightLagThreshold: 4
+        # [Optional] reconnectBlockHeightLagThreshold - if >0 then the event client will disconnect from the peer if the peer's
+        # block height falls behind the specified number of blocks and will reconnect to a better performing peer.
+        # If set to 0 then this feature is disabled.
+        # Default: 10
+        # NOTES:
+        #   - peerMonitorPeriod must be >0 to enable this feature
+        #   - Setting this value too low may cause the event client to disconnect/reconnect too frequently, thereby
+        #     affecting performance.
+        reconnectBlockHeightLagThreshold: 8
+        # [Optional] peerMonitorPeriod is the period in which the connected peer is monitored to see if
+        # the event client should disconnect from it and reconnect to another peer.
+        # Default: 0 (disabled)
+        peerMonitorPeriod: 6s
 
   # multi-org test channel
   orgchannel:

pkg/core/config/testdata/config_test_embedded_pems.yaml

@@ -125,6 +125,45 @@ channels:
           maxBackoff: 9s
           #[Optional] he factor by which the initial back off period is exponentially incremented
           backoffFactor: 3.0
+      #[Optional] options for event service
+      eventService:
+        # [Optional] resolverStrategy specifies the peer resolver strategy to use when connecting to a peer
+        # Possible values: [PreferOrg (default), MinBlockHeight, Balanced]
+        #
+        # PreferOrg:
+        #   Determines which peers are suitable based on block height lag threshold, although will prefer the peers in the
+        #   current org (as long as their block height is above a configured threshold). If none of the peers from the current org
+        #   are suitable then a peer from another org is chosen.
+        # MinBlockHeight:
+        #   Chooses the best peer according to a block height lag threshold. The maximum block height of all peers is
+        #   determined and the peers whose block heights are under the maximum height but above a provided "lag" threshold are load
+        #   balanced. The other peers are not considered.
+        # Balanced:
+        #   Chooses peers using the configured balancer.
+        resolverStrategy: Balanced
+        # [Optional] balancer is the balancer to use when choosing a peer to connect to
+        # Possible values: [Random (default), RoundRobin]
+        balancer: RoundRobin
+        # [Optional] blockHeightLagThreshold sets the block height lag threshold. This value is used for choosing a peer
+        # to connect to. If a peer is lagging behind the most up-to-date peer by more than the given number of
+        # blocks then it will be excluded from selection.
+        # If set to 0 then only the most up-to-date peers are considered.
+        # If set to -1 then all peers (regardless of block height) are considered for selection.
+        # Default: 5
+        blockHeightLagThreshold: 3
+        # [Optional] reconnectBlockHeightLagThreshold - if >0 then the event client will disconnect from the peer if the peer's
+        # block height falls behind the specified number of blocks and will reconnect to a better performing peer.
+        # If set to 0 then this feature is disabled.
+        # Default: 10
+        # NOTES:
+        #   - peerMonitorPeriod must be >0 to enable this feature
+        #   - Setting this value too low may cause the event client to disconnect/reconnect too frequently, thereby
+        #     affecting performance.
+        reconnectBlockHeightLagThreshold: 7
+        # [Optional] peerMonitorPeriod is the period in which the connected peer is monitored to see if
+        # the event client should disconnect from it and reconnect to another peer.
+        # Default: 0 (disabled)
+        peerMonitorPeriod: 8s
 
   # name of the channel
   mychannel:

pkg/core/config/testdata/template/config.yaml

@@ -46,48 +46,6 @@ client:
 #        # This interval will define how long a peer is greylisted
 #        greylistExpiry: 10s
 #
-#  eventService:
-#    # resolverStrategy specifies the peer resolver strategy to use when connecting to a peer.
-#    # Possible values: [PreferOrg (default), MinBlockHeight, Balanced]
-#    #
-#    # PreferOrg:
-#    #   Determines which peers are suitable based on block height lag threshold, although will prefer the peers in the
-#    #   current org (as long as their block height is above a configured threshold). If none of the peers from the current org
-#    #   are suitable then a peer from another org is chosen.
-#    # MinBlockHeight:
-#    #   Chooses the best peer according to a block height lag threshold. The maximum block height of all peers is
-#    #   determined and the peers whose block heights are under the maximum height but above a provided "lag" threshold are load
-#    #   balanced. The other peers are not considered.
-#    # Balanced:
-#    #   Chooses peers using the configured balancer.
-#    resolverStrategy: PreferOrg
-#
-#    # balancer is the balancer to use when choosing a peer to connect to
-#    # Possible values: [Random (default), RoundRobin]
-#    balancer: Random
-#
-#    # blockHeightLagThreshold sets the block height lag threshold. This value is used for choosing a peer
-#    # to connect to. If a peer is lagging behind the most up-to-date peer by more than the given number of
-#    # blocks then it will be excluded from selection.
-#    # If set to 0 then only the most up-to-date peers are considered.
-#    # If set to -1 then all peers (regardless of block height) are considered for selection.
-#    # Default: 5
-#    blockHeightLagThreshold: 5
-#
-#    # reconnectBlockHeightLagThreshold - if >0 then the event client will disconnect from the peer if the peer's
-#    # block height falls behind the specified number of blocks and will reconnect to a better performing peer.
-#    # If set to 0 then this feature is disabled.
-#    # Default: 0
-#    # NOTES:
-#    #   - peerMonitorPeriod must be >0 to enable this feature
-#    #   - Setting this value too low may cause the event client to disconnect/reconnect too frequently, thereby
-#    #     affecting performance.
-#    reconnectBlockHeightLagThreshold: 0
-#
-#    # peerMonitorPeriod is the period in which the connected peer is monitored to see if
-#    # the event client should disconnect from it and reconnect to another peer.
-#    # Default: 0 (disabled)
-#    peerMonitorPeriod: 5s
 
     # the below timeouts are commented out to use the default values that are found in
     # "pkg/fab/endpointconfig.go"
@@ -221,6 +179,45 @@ channels:
 #        #to a lower priority list of peers which will be sorted according to block height.
 #        #Note: This property only applies to BlockHeightPriority sorter.
 #        BlockHeightLagThreshold: 5
+#      #[Optional] options for event service
+#      eventService:
+#        # [Optional] resolverStrategy specifies the peer resolver strategy to use when connecting to a peer
+#        # Possible values: [PreferOrg (default), MinBlockHeight, Balanced]
+#        #
+#        # PreferOrg:
+#        #   Determines which peers are suitable based on block height lag threshold, although will prefer the peers in the
+#        #   current org (as long as their block height is above a configured threshold). If none of the peers from the current org
+#        #   are suitable then a peer from another org is chosen.
+#        # MinBlockHeight:
+#        #   Chooses the best peer according to a block height lag threshold. The maximum block height of all peers is
+#        #   determined and the peers whose block heights are under the maximum height but above a provided "lag" threshold are load
+#        #   balanced. The other peers are not considered.
+#        # Balanced:
+#        #   Chooses peers using the configured balancer.
+#        resolverStrategy: PreferOrg
+#        # [Optional] balancer is the balancer to use when choosing a peer to connect to
+#        # Possible values: [Random (default), RoundRobin]
+#        balancer: Random
+#        # [Optional] blockHeightLagThreshold sets the block height lag threshold. This value is used for choosing a peer
+#        # to connect to. If a peer is lagging behind the most up-to-date peer by more than the given number of
+#        # blocks then it will be excluded from selection.
+#        # If set to 0 then only the most up-to-date peers are considered.
+#        # If set to -1 then all peers (regardless of block height) are considered for selection.
+#        # Default: 5
+#        blockHeightLagThreshold: 5
+#        # [Optional] reconnectBlockHeightLagThreshold - if >0 then the event client will disconnect from the peer if the peer's
+#        # block height falls behind the specified number of blocks and will reconnect to a better performing peer.
+#        # If set to 0 then this feature is disabled.
+#        # Default: 10
+#        # NOTES:
+#        #   - peerMonitorPeriod must be >0 to enable this feature
+#        #   - Setting this value too low may cause the event client to disconnect/reconnect too frequently, thereby
+#        #     affecting performance.
+#        reconnectBlockHeightLagThreshold: 10
+#        # [Optional] peerMonitorPeriod is the period in which the connected peer is monitored to see if
+#        # the event client should disconnect from it and reconnect to another peer.
+#        # Default: 0 (disabled)
+#        peerMonitorPeriod: 5s
 
   # sample channel with channel matcher (sample*channel will return ch1 config where * can be any word or '')
 #  ch1:

pkg/fab/api.go

@@ -7,6 +7,8 @@ SPDX-License-Identifier: Apache-2.0
 package fab
 
 import (
+	"time"
+
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/errors/retry"
 	"github.com/hyperledger/fabric-sdk-go/pkg/core/config/endpoint"
 )
@@ -65,6 +67,8 @@ type ChannelPolicies struct {
 	Discovery DiscoveryPolicy
 	//Policy for endorser selection
 	Selection SelectionPolicy
+	//Policy for event service
+	EventService EventServicePolicy
 }
 
 //QueryChannelConfigPolicy defines opts for channelConfigBlock
@@ -120,6 +124,15 @@ type SelectionPolicy struct {
 	BlockHeightLagThreshold int
 }
 
+// EventServicePolicy specifies the policy for the event service
+type EventServicePolicy struct {
+	ResolverStrategy                 string
+	Balancer                         BalancerType
+	BlockHeightLagThreshold          int
+	ReconnectBlockHeightLagThreshold int
+	PeerMonitorPeriod                time.Duration
+}
+
 // PeerChannelConfig defines the peer capabilities
 type PeerChannelConfig struct {
 	EndorsingPeer  bool

pkg/fab/default_channel_test.go

@@ -9,7 +9,9 @@ package fab
 import (
 	"strings"
 	"testing"
+	"time"
 
+	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/fab"
 	"github.com/hyperledger/fabric-sdk-go/pkg/core/config"
 	"github.com/stretchr/testify/assert"
 )
@@ -32,6 +34,13 @@ func TestDefaultChannelWithDefaultChannelConfiguredAndNoMatchers(t *testing.T) {
 	assert.Equal(t, 1, chConfig.Policies.Discovery.MinResponses)
 	assert.Equal(t, 3, chConfig.Policies.Discovery.MaxTargets)
 
+	eventPolicies := chConfig.Policies.EventService
+	assert.Equalf(t, fab.BalancedStrategy, eventPolicies.ResolverStrategy, "Unexpected value for ResolverStrategy")
+	assert.Equal(t, fab.RoundRobin, eventPolicies.Balancer, "Unexpected value for Balancer")
+	assert.Equal(t, 3, eventPolicies.BlockHeightLagThreshold, "Unexpected value for BlockHeightLagThreshold")
+	assert.Equal(t, 7, eventPolicies.ReconnectBlockHeightLagThreshold, "Unexpected value for ReconnectBlockHeightLagThreshold")
+	assert.Equal(t, 8*time.Second, eventPolicies.PeerMonitorPeriod, "Unexpected value for PeerMonitorPeriod")
+
 	//When channel is not defined it should take channel peers from "_default"
 	chPeers := endpointConfig.ChannelPeers("test")
 	assert.NotNil(t, chPeers)