Merge changes I16b47247,Ibbd47540

* changes:
  FAB-16544 Override orderer endpoints if specified
  FAB-16544 Add orderer overrides to config

Author: Jason Yellick

core/deliverservice/config.go

@@ -7,11 +7,13 @@ SPDX-License-Identifier: Apache-2.0
 package deliverservice
 
 import (
+	"crypto/x509"
 	"io/ioutil"
 	"time"
 
 	"github.com/hyperledger/fabric/core/comm"
 	"github.com/hyperledger/fabric/core/config"
+	"github.com/hyperledger/fabric/internal/pkg/peer/orderers"
 
 	"github.com/pkg/errors"
 	"github.com/spf13/viper"
@@ -38,6 +40,16 @@ type DeliverServiceConfig struct {
 	KeepaliveOptions comm.KeepaliveOptions
 	// SecOpts provides the TLS info for connections
 	SecOpts comm.SecureOptions
+
+	// OrdererEndpointOverrides is a map of orderer addresses which should be
+	// re-mapped to a different orderer endpoint.
+	OrdererEndpointOverrides map[string]*orderers.Endpoint
+}
+
+type AddressOverride struct {
+	From        string `mapstructure:"from"`
+	To          string `mapstructure:"to"`
+	CACertsFile string `mapstructure:"caCertsFile"`
 }
 
 // GlobalConfig obtains a set of configuration from viper, build and returns the config struct.
@@ -47,6 +59,41 @@ func GlobalConfig() *DeliverServiceConfig {
 	return c
 }
 
+func LoadOverridesMap() (map[string]*orderers.Endpoint, error) {
+	var overrides []AddressOverride
+	err := viper.UnmarshalKey("peer.deliveryclient.addressOverrides", &overrides)
+	if err != nil {
+		return nil, errors.WithMessage(err, "could not unmarshal peer.deliveryclient.addressOverrides")
+	}
+
+	if len(overrides) == 0 {
+		return nil, nil
+	}
+
+	overrideMap := map[string]*orderers.Endpoint{}
+	for _, override := range overrides {
+		certPool := x509.NewCertPool()
+		if override.CACertsFile != "" {
+			pem, err := ioutil.ReadFile(override.CACertsFile)
+			if err != nil {
+				logger.Warningf("could not read file '%s' specified for caCertsFile of orderer endpoint override from '%s' to '%s': %s", override.CACertsFile, override.From, override.To, err)
+				continue
+			}
+			success := certPool.AppendCertsFromPEM(pem)
+			if !success {
+				logger.Warningf("Attempted to create a cert pool for override of orderer address '%s' to '%s' but did not find any valid certs in '%s'", override.From, override.To, override.CACertsFile)
+				continue
+			}
+		}
+		overrideMap[override.From] = &orderers.Endpoint{
+			Address:  override.To,
+			CertPool: certPool,
+		}
+	}
+
+	return overrideMap, nil
+}
+
 func (c *DeliverServiceConfig) loadDeliverServiceConfig() {
 	c.PeerTLSEnabled = viper.GetBool("peer.tls.enabled")
 
@@ -100,4 +147,11 @@ func (c *DeliverServiceConfig) loadDeliverServiceConfig() {
 		}
 		c.SecOpts.Certificate = certPEM
 	}
+
+	overridesMap, err := LoadOverridesMap()
+	if err != nil {
+		panic(err)
+	}
+
+	c.OrdererEndpointOverrides = overridesMap
 }

core/deliverservice/config_test.go

@@ -7,6 +7,7 @@ SPDX-License-Identifier: Apache-2.0
 package deliverservice_test
 
 import (
+	"bytes"
 	"io/ioutil"
 	"os"
 	"path/filepath"
@@ -127,3 +128,102 @@ func TestGlobalConfigDefault(t *testing.T) {
 
 	assert.Equal(t, expectedConfig, coreConfig)
 }
+
+func TestLoadOverridesMap(t *testing.T) {
+	defer viper.Reset()
+
+	t.Run("GreenPath", func(t *testing.T) {
+		config := `
+                  peer:
+                    deliveryclient:
+                      addressOverrides:
+                        - from: addressFrom1
+                          to: addressTo1
+                          caCertsFile: testdata/cert.pem
+                        - from: addressFrom2
+                          to: addressTo2
+                          caCertsFile: testdata/cert.pem
+                `
+
+		viper.Reset()
+		viper.SetConfigType("yaml")
+		viper.ReadConfig(bytes.NewBuffer([]byte(config)))
+		res, err := deliverservice.LoadOverridesMap()
+		require.NoError(t, err)
+		require.Len(t, res, 2)
+		ep1, ok := res["addressFrom1"]
+		require.True(t, ok)
+		assert.Equal(t, "addressTo1", ep1.Address)
+		ep2, ok := res["addressFrom2"]
+		require.True(t, ok)
+		assert.Equal(t, "addressTo2", ep2.Address)
+	})
+
+	t.Run("MissingCAFiles", func(t *testing.T) {
+		config := `
+                  peer:
+                    deliveryclient:
+                      addressOverrides:
+                        - from: addressFrom1
+                          to: addressTo1
+                          caCertsFile: missing/cert.pem
+                        - from: addressFrom2
+                          to: addressTo2
+                          caCertsFile: testdata/cert.pem
+                `
+
+		viper.Reset()
+		viper.SetConfigType("yaml")
+		viper.ReadConfig(bytes.NewBuffer([]byte(config)))
+		res, err := deliverservice.LoadOverridesMap()
+		require.NoError(t, err)
+		assert.Len(t, res, 1)
+	})
+
+	t.Run("EmptyCAFiles", func(t *testing.T) {
+		config := `
+                  peer:
+                    deliveryclient:
+                      addressOverrides:
+                        - from: addressFrom1
+                          to: addressTo1
+                        - from: addressFrom2
+                          to: addressTo2
+                `
+
+		viper.Reset()
+		viper.SetConfigType("yaml")
+		viper.ReadConfig(bytes.NewBuffer([]byte(config)))
+		res, err := deliverservice.LoadOverridesMap()
+		require.NoError(t, err)
+		assert.Len(t, res, 2)
+	})
+
+	t.Run("BadYaml", func(t *testing.T) {
+		config := `
+                  peer:
+                    deliveryclient:
+                      addressOverrides: foo
+                `
+
+		viper.Reset()
+		viper.SetConfigType("yaml")
+		viper.ReadConfig(bytes.NewBuffer([]byte(config)))
+		_, err := deliverservice.LoadOverridesMap()
+		require.EqualError(t, err, "could not unmarshal peer.deliveryclient.addressOverrides: '': source data must be an array or slice, got string")
+	})
+
+	t.Run("EmptyYaml", func(t *testing.T) {
+		config := `
+                  peer:
+                    deliveryclient:
+                `
+
+		viper.Reset()
+		viper.SetConfigType("yaml")
+		viper.ReadConfig(bytes.NewBuffer([]byte(config)))
+		res, err := deliverservice.LoadOverridesMap()
+		require.NoError(t, err)
+		assert.Nil(t, res)
+	})
+}

core/peer/peer.go

@@ -174,13 +174,14 @@ func (c *configSupport) GetChannelConfig(cid string) cc.Config {
 
 // A Peer holds references to subsystems and channels associated with a Fabric peer.
 type Peer struct {
-	Server            *comm.GRPCServer
-	ServerConfig      comm.ServerConfig
-	CredentialSupport *comm.CredentialSupport
-	StoreProvider     transientstore.StoreProvider
-	GossipService     *gossipservice.GossipService
-	LedgerMgr         *ledgermgmt.LedgerMgr
-	CryptoProvider    bccsp.BCCSP
+	Server                   *comm.GRPCServer
+	ServerConfig             comm.ServerConfig
+	CredentialSupport        *comm.CredentialSupport
+	StoreProvider            transientstore.StoreProvider
+	GossipService            *gossipservice.GossipService
+	LedgerMgr                *ledgermgmt.LedgerMgr
+	OrdererEndpointOverrides map[string]*orderers.Endpoint
+	CryptoProvider           bccsp.BCCSP
 
 	// validationWorkersSemaphore is used to limit the number of concurrent validation
 	// go routines.
@@ -290,7 +291,7 @@ func (p *Peer) createChannel(
 
 	osLogger := flogging.MustGetLogger("peer.orderers")
 	namedOSLogger := osLogger.With("channel", cid)
-	ordererSource := orderers.NewConnectionSource(namedOSLogger)
+	ordererSource := orderers.NewConnectionSource(namedOSLogger, p.OrdererEndpointOverrides)
 
 	ordererSourceCallback := func(bundle *channelconfig.Bundle) {
 		globalAddresses := bundle.ChannelConfig().OrdererAddresses()

gossip/service/gossip_service_test.go

@@ -192,7 +192,7 @@ func TestLeaderElectionWithDeliverClient(t *testing.T) {
 		gossips[i].deliveryFactory = deliverServiceFactory
 		deliverServiceFactory.service.running[channelName] = false
 
-		gossips[i].InitializeChannel(channelName, orderers.NewConnectionSource(flogging.MustGetLogger("peer.orderers")), store.Store, Support{
+		gossips[i].InitializeChannel(channelName, orderers.NewConnectionSource(flogging.MustGetLogger("peer.orderers"), nil), store.Store, Support{
 			Committer: &mockLedgerInfo{1},
 		})
 		service, exist := gossips[i].leaderElection[channelName]
@@ -256,7 +256,7 @@ func TestWithStaticDeliverClientLeader(t *testing.T) {
 	for i := 0; i < n; i++ {
 		gossips[i].deliveryFactory = deliverServiceFactory
 		deliverServiceFactory.service.running[channelName] = false
-		gossips[i].InitializeChannel(channelName, orderers.NewConnectionSource(flogging.MustGetLogger("peer.orderers")), store.Store, Support{
+		gossips[i].InitializeChannel(channelName, orderers.NewConnectionSource(flogging.MustGetLogger("peer.orderers"), nil), store.Store, Support{
 			Committer: &mockLedgerInfo{1},
 		})
 	}
@@ -269,7 +269,7 @@ func TestWithStaticDeliverClientLeader(t *testing.T) {
 	channelName = "chanB"
 	for i := 0; i < n; i++ {
 		deliverServiceFactory.service.running[channelName] = false
-		gossips[i].InitializeChannel(channelName, orderers.NewConnectionSource(flogging.MustGetLogger("peer.orderers")), store.Store, Support{
+		gossips[i].InitializeChannel(channelName, orderers.NewConnectionSource(flogging.MustGetLogger("peer.orderers"), nil), store.Store, Support{
 			Committer: &mockLedgerInfo{1},
 		})
 	}
@@ -317,7 +317,7 @@ func TestWithStaticDeliverClientNotLeader(t *testing.T) {
 	for i := 0; i < n; i++ {
 		gossips[i].deliveryFactory = deliverServiceFactory
 		deliverServiceFactory.service.running[channelName] = false
-		gossips[i].InitializeChannel(channelName, orderers.NewConnectionSource(flogging.MustGetLogger("peer.orderers")), store.Store, Support{
+		gossips[i].InitializeChannel(channelName, orderers.NewConnectionSource(flogging.MustGetLogger("peer.orderers"), nil), store.Store, Support{
 			Committer: &mockLedgerInfo{1},
 		})
 	}
@@ -365,7 +365,7 @@ func TestWithStaticDeliverClientBothStaticAndLeaderElection(t *testing.T) {
 	for i := 0; i < n; i++ {
 		gossips[i].deliveryFactory = deliverServiceFactory
 		assert.Panics(t, func() {
-			gossips[i].InitializeChannel(channelName, orderers.NewConnectionSource(flogging.MustGetLogger("peer.orderers")), store.Store, Support{
+			gossips[i].InitializeChannel(channelName, orderers.NewConnectionSource(flogging.MustGetLogger("peer.orderers"), nil), store.Store, Support{
 				Committer: &mockLedgerInfo{1},
 			})
 		}, "Dynamic leader election based and static connection to ordering service can't exist simultaneously")
@@ -943,7 +943,7 @@ func TestInvalidInitialization(t *testing.T) {
 	go grpcServer.Serve(socket)
 	defer grpcServer.Stop()
 
-	dc := gService.deliveryFactory.Service(gService, orderers.NewConnectionSource(flogging.MustGetLogger("peer.orderers")), &naiveCryptoService{})
+	dc := gService.deliveryFactory.Service(gService, orderers.NewConnectionSource(flogging.MustGetLogger("peer.orderers"), nil), &naiveCryptoService{})
 	assert.NotNil(t, dc)
 }
 

gossip/service/integration_test.go

@@ -126,7 +126,7 @@ func TestLeaderYield(t *testing.T) {
 			},
 			deliverGRPCClient: grpcClient,
 		}}
-		gs.InitializeChannel(channelName, orderers.NewConnectionSource(flogging.MustGetLogger("peer.orderers")), store.Store, Support{
+		gs.InitializeChannel(channelName, orderers.NewConnectionSource(flogging.MustGetLogger("peer.orderers"), nil), store.Store, Support{
 			Committer: &mockLedgerInfo{1},
 		})
 		return gs

internal/peer/node/start.go

@@ -252,12 +252,16 @@ func serve(args []string) error {
 	if err != nil {
 		return errors.WithMessage(err, "failed to open transient store")
 	}
+
+	deliverServiceConfig := deliverservice.GlobalConfig()
+
 	peerInstance := &peer.Peer{
-		Server:            peerServer,
-		ServerConfig:      serverConfig,
-		CredentialSupport: cs,
-		StoreProvider:     transientStoreProvider,
-		CryptoProvider:    factory.GetDefault(),
+		Server:                   peerServer,
+		ServerConfig:             serverConfig,
+		CredentialSupport:        cs,
+		StoreProvider:            transientStoreProvider,
+		CryptoProvider:           factory.GetDefault(),
+		OrdererEndpointOverrides: deliverServiceConfig.OrdererEndpointOverrides,
 	}
 
 	localMSP := mgmt.GetLocalMSP(factory.GetDefault())
@@ -267,8 +271,6 @@ func serve(args []string) error {
 	}
 	policyMgr := policies.PolicyManagerGetterFunc(peerInstance.GetPolicyManager)
 
-	deliverServiceConfig := deliverservice.GlobalConfig()
-
 	deliverGRPCClient, err := comm.NewGRPCClient(comm.ClientConfig{
 		Timeout: deliverServiceConfig.ConnectionTimeout,
 		KaOpts:  deliverServiceConfig.KeepaliveOptions,

internal/pkg/peer/orderers/connection.go

@@ -24,6 +24,7 @@ type ConnectionSource struct {
 	allEndpoints       []*Endpoint
 	orgToEndpointsHash map[string][]byte
 	logger             *flogging.FabricLogger
+	overrides          map[string]*Endpoint
 }
 
 type Endpoint struct {
@@ -37,10 +38,11 @@ type OrdererOrg struct {
 	RootCerts [][]byte
 }
 
-func NewConnectionSource(logger *flogging.FabricLogger) *ConnectionSource {
+func NewConnectionSource(logger *flogging.FabricLogger, overrides map[string]*Endpoint) *ConnectionSource {
 	return &ConnectionSource{
 		orgToEndpointsHash: map[string][]byte{},
 		logger:             logger,
+		overrides:          overrides,
 	}
 }
 
@@ -132,7 +134,7 @@ func (cs *ConnectionSource) Update(globalAddrs []string, orgs map[string]Orderer
 		// Alert any existing consumers that have a reference to the old endpoints
 		// that their reference is now stale and they should get a new one.
 		// This is done even for endpoints which have the same TLS certs and address
-		// but this is deseriable to help load balance.  For instance if only
+		// but this is desirable to help load balance.  For instance if only
 		// one orderer were defined, and the config is updated to include 4 more, we
 		// want the peers to disconnect from that original orderer and reconnect
 		// evenly across the now five.
@@ -161,6 +163,16 @@ func (cs *ConnectionSource) Update(globalAddrs []string, orgs map[string]Orderer
 		// Note, if !hasOrgEndpoints, this for loop is a no-op, so
 		// certPool is never referenced.
 		for _, address := range org.Addresses {
+			overrideEndpoint, ok := cs.overrides[address]
+			if ok {
+				cs.allEndpoints = append(cs.allEndpoints, &Endpoint{
+					Address:   overrideEndpoint.Address,
+					CertPool:  overrideEndpoint.CertPool,
+					Refreshed: make(chan struct{}),
+				})
+				continue
+			}
+
 			cs.allEndpoints = append(cs.allEndpoints, &Endpoint{
 				Address:   address,
 				CertPool:  certPool,
@@ -177,6 +189,16 @@ func (cs *ConnectionSource) Update(globalAddrs []string, orgs map[string]Orderer
 	}
 
 	for _, address := range globalAddrs {
+		overrideEndpoint, ok := cs.overrides[address]
+		if ok {
+			cs.allEndpoints = append(cs.allEndpoints, &Endpoint{
+				Address:   overrideEndpoint.Address,
+				CertPool:  overrideEndpoint.CertPool,
+				Refreshed: make(chan struct{}),
+			})
+			continue
+		}
+
 		cs.allEndpoints = append(cs.allEndpoints, &Endpoint{
 			Address:   address,
 			CertPool:  globalCertPool,