[FAB-10655]Problematic client conn stops eventhub evts

If a client connection hangs, the eventhub’s flow of events
completely stops until the problematic client connection decides
to return (likely after a grpc timeout). This CR adds a timeout
to the Send calls on the GRPC stream to ensure the eventhub is
not at the mercy of the client connection releasing the connection
before it can continue sending events. It also cleans up some code
and removes a large chunk of dead code, including the ability to
send chaincode events, which has not been used by Fabric at least
beginning with v1.0 if not earlier.

Change-Id: Idd86c1cec7d307becbd1a47a15c19f65eb0c64f1
Signed-off-by: Will Lahti <wtlahti@us.ibm.com>

Author: wlahti

events/consumer/consumer.go

@@ -1,17 +1,7 @@
 /*
-Copyright IBM Corp. 2016 All Rights Reserved.
+Copyright IBM Corp. All Rights Reserved.
 
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-		 http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
+SPDX-License-Identifier: Apache-2.0
 */
 
 package consumer
@@ -161,7 +151,7 @@ func (ec *EventsClient) UnregisterAsync(ies []*ehpb.Interest) error {
 	emsg := &ehpb.Event{Event: &ehpb.Event_Unregister{Unregister: &ehpb.Unregister{Events: ies}}, Creator: creator}
 
 	if err = ec.send(emsg); err != nil {
-		err = fmt.Errorf("error on unregister send %s\n", err)
+		err = fmt.Errorf("error on unregister send: %s", err)
 	}
 
 	return err
@@ -185,6 +175,7 @@ func (ec *EventsClient) Recv() (*ehpb.Event, error) {
 	}
 	return in, nil
 }
+
 func (ec *EventsClient) processEvents() error {
 	defer ec.stream.CloseSend()
 	for {

events/consumer/consumer_test.go

@@ -21,8 +21,7 @@ import (
 	coreutil "github.com/hyperledger/fabric/core/testutil"
 	"github.com/hyperledger/fabric/events/producer"
 	"github.com/hyperledger/fabric/msp/mgmt/testtools"
-	ehpb "github.com/hyperledger/fabric/protos/peer"
-	"github.com/spf13/viper"
+	pb "github.com/hyperledger/fabric/protos/peer"
 	"github.com/stretchr/testify/assert"
 	"google.golang.org/grpc"
 )
@@ -43,41 +42,39 @@ type BadAdapter struct {
 }
 
 var peerAddress = "0.0.0.0:7303"
-var ies = []*ehpb.Interest{{EventType: ehpb.EventType_CHAINCODE, RegInfo: &ehpb.Interest_ChaincodeRegInfo{ChaincodeRegInfo: &ehpb.ChaincodeReg{ChaincodeId: "0xffffffff", EventName: "event1"}}}}
+var ies = []*pb.Interest{{EventType: pb.EventType_BLOCK}}
 var testCert = &x509.Certificate{
 	Raw: []byte("test"),
 }
 
 var adapter *MockAdapter
-var obcEHClient *EventsClient
+var ehClient *EventsClient
 
-func (a *ZeroAdapter) GetInterestedEvents() ([]*ehpb.Interest, error) {
-	return []*ehpb.Interest{}, nil
+func (a *ZeroAdapter) GetInterestedEvents() ([]*pb.Interest, error) {
+	return []*pb.Interest{}, nil
 }
-func (a *ZeroAdapter) Recv(msg *ehpb.Event) (bool, error) {
+func (a *ZeroAdapter) Recv(msg *pb.Event) (bool, error) {
 	panic("not implemented")
 }
 func (a *ZeroAdapter) Disconnected(err error) {
 	panic("not implemented")
 }
 
-func (a *BadAdapter) GetInterestedEvents() ([]*ehpb.Interest, error) {
-	return []*ehpb.Interest{}, fmt.Errorf("Error")
+func (a *BadAdapter) GetInterestedEvents() ([]*pb.Interest, error) {
+	return []*pb.Interest{}, fmt.Errorf("Error")
 }
-func (a *BadAdapter) Recv(msg *ehpb.Event) (bool, error) {
+func (a *BadAdapter) Recv(msg *pb.Event) (bool, error) {
 	panic("not implemented")
 }
 func (a *BadAdapter) Disconnected(err error) {
 	panic("not implemented")
 }
 
-func (a *MockAdapter) GetInterestedEvents() ([]*ehpb.Interest, error) {
-	return []*ehpb.Interest{
-		{EventType: ehpb.EventType_BLOCK},
-	}, nil
+func (a *MockAdapter) GetInterestedEvents() ([]*pb.Interest, error) {
+	return ies, nil
 }
 
-func (a *MockAdapter) Recv(msg *ehpb.Event) (bool, error) {
+func (a *MockAdapter) Recv(msg *pb.Event) (bool, error) {
 	return true, nil
 }
 
@@ -159,25 +156,28 @@ func TestUnregisterAsync(t *testing.T) {
 	done := make(chan struct{})
 	adapter := &MockAdapter{notify: done}
 
-	obcEHClient, _ = NewEventsClient(peerAddress, 5, adapter)
+	ehClient, _ = NewEventsClient(peerAddress, 5, adapter)
 
-	if err = obcEHClient.Start(); err != nil {
-		obcEHClient.Stop()
+	if err = ehClient.Start(); err != nil {
+		ehClient.Stop()
 		t.Fail()
 	}
 
-	regConfig := &RegistrationConfig{InterestedEvents: ies, Timestamp: util.CreateUtcTimestamp(), TlsCert: testCert}
-	obcEHClient.RegisterAsync(regConfig)
-	err = obcEHClient.UnregisterAsync(ies)
+	regConfig := &RegistrationConfig{
+		InterestedEvents: ies,
+		Timestamp:        util.CreateUtcTimestamp(),
+		TlsCert:          testCert,
+	}
+	ehClient.RegisterAsync(regConfig)
+	err = ehClient.UnregisterAsync(ies)
 	assert.NoError(t, err)
 
-	obcEHClient.Stop()
-
+	ehClient.Stop()
 }
 
 func TestStart(t *testing.T) {
 	var err error
-	var regTimeout = 5 * time.Second
+	var regTimeout = 1 * time.Second
 	done := make(chan struct{})
 
 	var cases = []struct {
@@ -215,33 +215,32 @@ func TestStart(t *testing.T) {
 	for _, test := range cases {
 		t.Run(test.name, func(t *testing.T) {
 			t.Logf("Running test: %s", test.name)
-			obcEHClient, _ = NewEventsClient(test.address, regTimeout, test.adapter)
-			err = obcEHClient.Start()
+			ehClient, _ = NewEventsClient(test.address, regTimeout, test.adapter)
+			err = ehClient.Start()
 			if test.expected {
 				assert.NoError(t, err)
 			} else {
 				assert.Error(t, err)
 			}
-			obcEHClient.Stop()
+			ehClient.Stop()
 		})
 	}
 }
 
 func TestStop(t *testing.T) {
 	var err error
-	var regTimeout = 5 * time.Second
+	var regTimeout = 1 * time.Second
 	done := make(chan struct{})
 	adapter := &MockAdapter{notify: done}
 
-	obcEHClient, _ = NewEventsClient(peerAddress, regTimeout, adapter)
+	ehClient, _ = NewEventsClient(peerAddress, regTimeout, adapter)
 
-	if err = obcEHClient.Start(); err != nil {
+	if err = ehClient.Start(); err != nil {
 		t.Fail()
-		t.Logf("Error client start %s", err)
+		t.Logf("Error starting client: %s", err)
 	}
-	err = obcEHClient.Stop()
+	err = ehClient.Stop()
 	assert.NoError(t, err)
-
 }
 
 func TestMain(m *testing.M) {
@@ -263,24 +262,25 @@ func TestMain(m *testing.M) {
 	}
 
 	extract := func(msg proto.Message) []byte {
-		evt, isEvent := msg.(*ehpb.Event)
+		evt, isEvent := msg.(*pb.Event)
 		if !isEvent || evt == nil {
 			return nil
 		}
 		return evt.TlsCertHash
 	}
 
+	timeout := 10 * time.Millisecond
 	ehConfig := &producer.EventsServerConfig{
-		BufferSize:       uint(viper.GetInt("peer.events.buffersize")),
-		Timeout:          viper.GetDuration("peer.events.timeout"),
-		TimeWindow:       viper.GetDuration("peer.events.timewindow"),
-		BindingInspector: comm.NewBindingInspector(false, extract)}
+		BufferSize:       100,
+		Timeout:          timeout,
+		SendTimeout:      timeout,
+		TimeWindow:       time.Minute,
+		BindingInspector: comm.NewBindingInspector(false, extract),
+	}
 	ehServer := producer.NewEventsServer(ehConfig)
-
-	ehpb.RegisterEventsServer(grpcServer, ehServer)
+	pb.RegisterEventsServer(grpcServer, ehServer)
 
 	go grpcServer.Serve(lis)
 
-	time.Sleep(2 * time.Second)
 	os.Exit(m.Run())
 }

events/producer/eventhelper.go

@@ -138,22 +138,12 @@ func CreateBlockEvents(block *common.Block) (bevent *pb.Event, fbevent *pb.Event
 	return CreateBlockEvent(blockForEvent), CreateFilteredBlockEvent(filteredBlockForEvent), channelID, nil
 }
 
-//CreateBlockEvent creates a Event from a Block
+// CreateBlockEvent creates a Event from a Block
 func CreateBlockEvent(te *common.Block) *pb.Event {
 	return &pb.Event{Event: &pb.Event_Block{Block: te}}
 }
 
-//CreateFilteredBlockEvent creates a Event from a FilteredBlock
+// CreateFilteredBlockEvent creates a Event from a FilteredBlock
 func CreateFilteredBlockEvent(te *pb.FilteredBlock) *pb.Event {
 	return &pb.Event{Event: &pb.Event_FilteredBlock{FilteredBlock: te}}
 }
-
-//CreateChaincodeEvent creates a Event from a ChaincodeEvent
-func CreateChaincodeEvent(te *pb.ChaincodeEvent) *pb.Event {
-	return &pb.Event{Event: &pb.Event_ChaincodeEvent{ChaincodeEvent: te}}
-}
-
-//CreateRejectionEvent creates an Event from TxResults
-func CreateRejectionEvent(tx *pb.Transaction, errorMsg string) *pb.Event {
-	return &pb.Event{Event: &pb.Event_Rejection{Rejection: &pb.Rejection{Tx: tx, ErrorMsg: errorMsg}}}
-}