[FAB-9452] Enhance chaincode event listers

   This CR:
 - Allows registeration of multiple listeners for
   chaincode life-cycle events
 - Introduces a call back to the listeners when
   the chaincode deploy transaction commits or
   chaincode install completes

Change-Id: Ic9e26b430d8b0c844219edb454e5d410477c9516
Signed-off-by: manish <manish.sethi@gmail.com>

Author: manish-sethi

core/cclifecycle/subscription.go

@@ -24,7 +24,8 @@ type Subscription struct {
 
 type depCCsRetriever func(Query, ChaincodePredicate, ...string) (chaincode.MetadataSet, error)
 
-// HandleChaincodeDeploy is expected to be invoked when a chaincode is deployed via a deploy transaction
+// HandleChaincodeDeploy is expected to be invoked when a chaincode is deployed via a deploy transaction and the chaicndoe was already
+// installed on the peer. This also gets invoked when an already deployed chaincode is installed on the peer
 func (sub *Subscription) HandleChaincodeDeploy(chaincodeDefinition *cceventmgmt.ChaincodeDefinition, dbArtifactsTar []byte) error {
 	logger.Debug("Channel", sub.channel, "got a new deployment:", chaincodeDefinition)
 	query, err := sub.newQuery()
@@ -48,6 +49,12 @@ func (sub *Subscription) HandleChaincodeDeploy(chaincodeDefinition *cceventmgmt.
 	return nil
 }
 
+// ChaincodeDeployDone gets invoked when the chaincode deploy transaction or chaincode install
+// (the context in which the above function was invoked)
+func (sub *Subscription) ChaincodeDeployDone(succeeded bool) {
+	// Noop
+}
+
 func queryChaincodeDefinitions(query Query, ccs []chaincode.InstalledChaincode, deployedCCs depCCsRetriever) (chaincode.MetadataSet, error) {
 	// map from string and version to chaincode ID
 	installedCCsToIDs := make(map[nameVersion][]byte)

core/ledger/cceventmgmt/defs.go

@@ -27,8 +27,14 @@ func (cdef *ChaincodeDefinition) String() string {
 // to be able to listen to chaincode lifecycle events. 'dbArtifactsTar' represents db specific artifacts
 // (such as index specs) packaged in a tar
 type ChaincodeLifecycleEventListener interface {
-	// HandleChaincodeDeploy is expected to creates all the necessary statedb structures (such as indexes)
+	// HandleChaincodeDeploy is invoked when chaincode installed + defined becomes true.
+	// The expected usage are to creates all the necessary statedb structures (such as indexes) and update
+	// service discovery info. This function is invoked immediately before the committing the state changes
+	// that contain chaincode definition or when a chaincode install happens
 	HandleChaincodeDeploy(chaincodeDefinition *ChaincodeDefinition, dbArtifactsTar []byte) error
+	// ChaincodeDeployDone is invoked after the chaincode deployment is finished - `succeeded` indicates
+	// whether the deploy finished successfully
+	ChaincodeDeployDone(succeeded bool)
 }
 
 // ChaincodeInfoProvider interface enables event mgr to retrieve chaincode info for a given chaincode

core/ledger/cceventmgmt/lsccstate_listener.go

@@ -60,5 +60,5 @@ func (listener *KVLedgerLSCCStateListener) InterestedInNamespaces() []string {
 
 // StateCommitDone implements function from interface `ledger.StateListener` as a NOOP
 func (listener *KVLedgerLSCCStateListener) StateCommitDone(channelName string) {
-	// NOOP
+	GetMgr().ChaincodeDeployDone(channelName)
 }

core/ledger/cceventmgmt/mgmt_test.go

@@ -41,28 +41,45 @@ func TestCCEventMgmt(t *testing.T) {
 	setEventMgrForTest(newMgr(mockProvider))
 	defer clearEventMgrForTest()
 
-	handler1, handler2 := &mockHandler{}, &mockHandler{}
+	handler1, handler2, handler3 := &mockHandler{}, &mockHandler{}, &mockHandler{}
 	eventMgr := GetMgr()
 	assert.NotNil(t, eventMgr)
 	eventMgr.Register("channel1", handler1)
 	eventMgr.Register("channel2", handler2)
+	eventMgr.Register("channel1", handler3)
+	eventMgr.Register("channel2", handler3)
 
 	cc2ExpectedEvent := &mockEvent{cc2Def, cc2DBArtifactsTar}
+	_ = cc2ExpectedEvent
 	cc3ExpectedEvent := &mockEvent{cc3Def, cc3DBArtifactsTar}
 
-	// Deploy cc3 on chain1 - only handler1 should recieve event because cc3 is being deployed only on chain1
+	// Deploy cc3 on chain1 - handler1 and handler3 should recieve event because cc3 is being deployed only on chain1
 	eventMgr.HandleChaincodeDeploy("channel1", []*ChaincodeDefinition{cc3Def})
+	eventMgr.ChaincodeDeployDone("channel1")
 	assert.Contains(t, handler1.eventsRecieved, cc3ExpectedEvent)
 	assert.NotContains(t, handler2.eventsRecieved, cc3ExpectedEvent)
+	assert.Contains(t, handler3.eventsRecieved, cc3ExpectedEvent)
+	assert.Equal(t, 1, handler1.doneRecievedCount)
+	assert.Equal(t, 0, handler2.doneRecievedCount)
+	assert.Equal(t, 1, handler3.doneRecievedCount)
 
 	// Deploy cc3 on chain2 as well and this time handler2 should also recieve event
 	eventMgr.HandleChaincodeDeploy("channel2", []*ChaincodeDefinition{cc3Def})
+	eventMgr.ChaincodeDeployDone("channel2")
 	assert.Contains(t, handler2.eventsRecieved, cc3ExpectedEvent)
+	assert.Equal(t, 1, handler1.doneRecievedCount)
+	assert.Equal(t, 1, handler2.doneRecievedCount)
+	assert.Equal(t, 2, handler3.doneRecievedCount)
 
-	// Install CC2 - only handler1 should receive event because cc2 is deployed only on chain1 and not on chain2
+	// Install CC2 - handler1 and handler 3 should receive event because cc2 is deployed only on chain1 and not on chain2
 	eventMgr.HandleChaincodeInstall(cc2Def, cc2DBArtifactsTar)
+	eventMgr.ChaincodeInstallDone(true)
 	assert.Contains(t, handler1.eventsRecieved, cc2ExpectedEvent)
 	assert.NotContains(t, handler2.eventsRecieved, cc2ExpectedEvent)
+	assert.Contains(t, handler3.eventsRecieved, cc2ExpectedEvent)
+	assert.Equal(t, 2, handler1.doneRecievedCount)
+	assert.Equal(t, 1, handler2.doneRecievedCount)
+	assert.Equal(t, 3, handler3.doneRecievedCount)
 }
 
 func TestLSCCListener(t *testing.T) {
@@ -132,7 +149,8 @@ type mockProvider struct {
 }
 
 type mockHandler struct {
-	eventsRecieved []*mockEvent
+	eventsRecieved    []*mockEvent
+	doneRecievedCount int
 }
 
 type mockEvent struct {
@@ -145,6 +163,10 @@ func (l *mockHandler) HandleChaincodeDeploy(chaincodeDefinition *ChaincodeDefini
 	return nil
 }
 
+func (l *mockHandler) ChaincodeDeployDone(succeeded bool) {
+	l.doneRecievedCount++
+}
+
 func newMockProvider() *mockProvider {
 	return &mockProvider{
 		make(map[[3]string]bool),

core/ledger/cceventmgmt/mgr.go

@@ -7,7 +7,6 @@ SPDX-License-Identifier: Apache-2.0
 package cceventmgmt
 
 import (
-	"bytes"
 	"sync"
 
 	"github.com/hyperledger/fabric/common/flogging"
@@ -39,19 +38,15 @@ type Mgr struct {
 	// we use this lock for contextual use
 	rwlock               sync.RWMutex
 	infoProvider         ChaincodeInfoProvider
-	ccLifecycleListeners map[string]ChaincodeLifecycleEventListener
-	// latestChaincodeDeploys maintains last chaincode deployed for a ledger. As stated in the above comment,
-	// since it is not easy to synchronize across block commit and install activity, this leaves a small window
-	// where we could miss 'deployed AND installed' state. So, we explicitly maintain the chaincodes deplyed
-	// in the last block
-	latestChaincodeDeploys map[string][]*ChaincodeDefinition
+	ccLifecycleListeners map[string][]ChaincodeLifecycleEventListener
+	callbackStatus       *callbackStatus
 }
 
 func newMgr(chaincodeInfoProvider ChaincodeInfoProvider) *Mgr {
 	return &Mgr{
-		infoProvider:           chaincodeInfoProvider,
-		ccLifecycleListeners:   make(map[string]ChaincodeLifecycleEventListener),
-		latestChaincodeDeploys: make(map[string][]*ChaincodeDefinition)}
+		infoProvider:         chaincodeInfoProvider,
+		ccLifecycleListeners: make(map[string][]ChaincodeLifecycleEventListener),
+		callbackStatus:       newCallbackStatus()}
 }
 
 // Register registers a ChaincodeLifecycleEventListener for given ledgerid
@@ -60,7 +55,7 @@ func (m *Mgr) Register(ledgerid string, l ChaincodeLifecycleEventListener) {
 	// write lock to synchronize concurrent 'chaincode install' operations with ledger creation/open
 	m.rwlock.Lock()
 	defer m.rwlock.Unlock()
-	m.ccLifecycleListeners[ledgerid] = l
+	m.ccLifecycleListeners[ledgerid] = append(m.ccLifecycleListeners[ledgerid], l)
 }
 
 // HandleChaincodeDeploy is expected to be invoked when a chaincode is deployed via a deploy transaction
@@ -73,11 +68,8 @@ func (m *Mgr) Register(ledgerid string, l ChaincodeLifecycleEventListener) {
 // in this stored `chaincodeDefinitions`
 func (m *Mgr) HandleChaincodeDeploy(chainid string, chaincodeDefinitions []*ChaincodeDefinition) error {
 	logger.Debugf("Channel [%s]: Handling chaincode deploy event for chaincode [%s]", chainid, chaincodeDefinitions)
-	// Read lock to allow concurrent deploy on multiple channels but to synchronize concurrent `chaincode insall` operation
+	// Read lock to allow concurrent deploy on multiple channels but to synchronize concurrent `chaincode install` operation
 	m.rwlock.RLock()
-	defer m.rwlock.RUnlock()
-	// TODO, device a mechanism to cleanup entries in this map
-	m.latestChaincodeDeploys[chainid] = chaincodeDefinitions
 	for _, chaincodeDefinition := range chaincodeDefinitions {
 		installed, dbArtifacts, err := m.infoProvider.RetrieveChaincodeArtifacts(chaincodeDefinition)
 		if err != nil {
@@ -88,6 +80,7 @@ func (m *Mgr) HandleChaincodeDeploy(chainid string, chaincodeDefinitions []*Chai
 				chainid, chaincodeDefinition)
 			continue
 		}
+		m.callbackStatus.setDeployPending(chainid)
 		if err := m.invokeHandler(chainid, chaincodeDefinition, dbArtifacts); err != nil {
 			logger.Warningf("Channel [%s]: Error while invoking a listener for handling chaincode install event: %s", chainid, err)
 			return err
@@ -97,28 +90,35 @@ func (m *Mgr) HandleChaincodeDeploy(chainid string, chaincodeDefinitions []*Chai
 	return nil
 }
 
-// HandleChaincodeInstall is expected to gets invoked when a during installation of a chaincode package
+// ChaincodeDeployDone is expected to be called when the deploy transaction state is committed
+func (m *Mgr) ChaincodeDeployDone(chainid string) {
+	// release the lock aquired in function `HandleChaincodeDeploy`
+	defer m.rwlock.RUnlock()
+	if m.callbackStatus.isDeployPending(chainid) {
+		m.invokeDoneOnHandlers(chainid, true)
+		m.callbackStatus.unsetDeployPending(chainid)
+	}
+}
+
+// HandleChaincodeInstall is expected to get invoked during installation of a chaincode package
 func (m *Mgr) HandleChaincodeInstall(chaincodeDefinition *ChaincodeDefinition, dbArtifacts []byte) error {
 	logger.Debugf("HandleChaincodeInstall() - chaincodeDefinition=%#v", chaincodeDefinition)
 	// Write lock prevents concurrent deploy operations
 	m.rwlock.Lock()
-	defer m.rwlock.Unlock()
 	for chainid := range m.ccLifecycleListeners {
 		logger.Debugf("Channel [%s]: Handling chaincode install event for chaincode [%s]", chainid, chaincodeDefinition)
 		var deployed bool
 		var err error
-		deployed = m.isChaincodePresentInLatestDeploys(chainid, chaincodeDefinition)
-		if !deployed {
-			if deployed, err = m.infoProvider.IsChaincodeDeployed(chainid, chaincodeDefinition); err != nil {
-				logger.Warningf("Channel [%s]: Error while getting the deployment status of chaincode: %s", chainid, err)
-				return err
-			}
+		if deployed, err = m.infoProvider.IsChaincodeDeployed(chainid, chaincodeDefinition); err != nil {
+			logger.Warningf("Channel [%s]: Error while getting the deployment status of chaincode: %s", chainid, err)
+			return err
 		}
 		if !deployed {
 			logger.Debugf("Channel [%s]: Chaincode [%s] is not deployed on channel hence not creating chaincode artifacts.",
 				chainid, chaincodeDefinition)
 			continue
 		}
+		m.callbackStatus.setInstallPending(chainid)
 		if err := m.invokeHandler(chainid, chaincodeDefinition, dbArtifacts); err != nil {
 			logger.Warningf("Channel [%s]: Error while invoking a listener for handling chaincode install event: %s", chainid, err)
 			return err
@@ -128,23 +128,77 @@ func (m *Mgr) HandleChaincodeInstall(chaincodeDefinition *ChaincodeDefinition, d
 	return nil
 }
 
-func (m *Mgr) isChaincodePresentInLatestDeploys(chainid string, chaincodeDefinition *ChaincodeDefinition) bool {
-	ccDefs, ok := m.latestChaincodeDeploys[chainid]
-	if !ok {
-		return false
+// ChaincodeInstallDone is expected to get invoked when chaincode install finishes
+func (m *Mgr) ChaincodeInstallDone(succeeded bool) {
+	// release the lock acquired in function `HandleChaincodeInstall`
+	defer m.rwlock.Unlock()
+	for chainid := range m.callbackStatus.installPending {
+		m.invokeDoneOnHandlers(chainid, succeeded)
+		m.callbackStatus.unsetInstallPending(chainid)
 	}
-	for _, ccDef := range ccDefs {
-		if ccDef.Name == chaincodeDefinition.Name && ccDef.Version == chaincodeDefinition.Version && bytes.Equal(ccDef.Hash, chaincodeDefinition.Hash) {
-			return true
+}
+
+func (m *Mgr) invokeHandler(chainid string, chaincodeDefinition *ChaincodeDefinition, dbArtifactsTar []byte) error {
+	listeners := m.ccLifecycleListeners[chainid]
+	for _, listener := range listeners {
+		if err := listener.HandleChaincodeDeploy(chaincodeDefinition, dbArtifactsTar); err != nil {
+			return err
 		}
 	}
-	return false
+	return nil
 }
 
-func (m *Mgr) invokeHandler(chainid string, chaincodeDefinition *ChaincodeDefinition, dbArtifactsTar []byte) error {
-	listener := m.ccLifecycleListeners[chainid]
-	if listener == nil {
-		return nil
+func (m *Mgr) invokeDoneOnHandlers(chainid string, succeeded bool) {
+	listeners := m.ccLifecycleListeners[chainid]
+	for _, listener := range listeners {
+		listener.ChaincodeDeployDone(succeeded)
 	}
-	return listener.HandleChaincodeDeploy(chaincodeDefinition, dbArtifactsTar)
+}
+
+type callbackStatus struct {
+	l              sync.Mutex
+	deployPending  map[string]bool
+	installPending map[string]bool
+}
+
+func newCallbackStatus() *callbackStatus {
+	return &callbackStatus{
+		deployPending:  make(map[string]bool),
+		installPending: make(map[string]bool)}
+}
+
+func (s *callbackStatus) setDeployPending(channelID string) {
+	s.l.Lock()
+	defer s.l.Unlock()
+	s.deployPending[channelID] = true
+}
+
+func (s *callbackStatus) unsetDeployPending(channelID string) {
+	s.l.Lock()
+	defer s.l.Unlock()
+	delete(s.deployPending, channelID)
+}
+
+func (s *callbackStatus) isDeployPending(channelID string) bool {
+	s.l.Lock()
+	defer s.l.Unlock()
+	return s.deployPending[channelID]
+}
+
+func (s *callbackStatus) setInstallPending(channelID string) {
+	s.l.Lock()
+	defer s.l.Unlock()
+	s.installPending[channelID] = true
+}
+
+func (s *callbackStatus) unsetInstallPending(channelID string) {
+	s.l.Lock()
+	defer s.l.Unlock()
+	delete(s.installPending, channelID)
+}
+
+func (s *callbackStatus) isInstallPending(channelID string) bool {
+	s.l.Lock()
+	defer s.l.Unlock()
+	return s.installPending[channelID]
 }

core/ledger/kvledger/txmgmt/statedb/statecouchdb/statecouchdb.go

@@ -119,6 +119,11 @@ func (vdb *VersionedDB) HandleChaincodeDeploy(chaincodeDefinition *cceventmgmt.C
 
 }
 
+// ChaincodeDeployDone is a noop for couchdb state impl
+func (vdb *VersionedDB) ChaincodeDeployDone(succeeded bool) {
+	// NOOP
+}
+
 // GetDBHandle gets the handle to a named database
 func (provider *VersionedDBProvider) GetDBHandle(dbName string) (statedb.VersionedDB, error) {
 

core/ledger/kvledger/txmgmt/statedb/statecouchdb/statecouchdb_test.go

@@ -347,7 +347,10 @@ func TestHandleChaincodeDeploy(t *testing.T) {
 	_, err = db.ExecuteQuery("ns1", queryString)
 	testutil.AssertError(t, err, "Error should have been thrown for a missing index")
 
-	handleDefinition, _ := db.(cceventmgmt.ChaincodeLifecycleEventListener)
+	handleDefinition, ok := db.(cceventmgmt.ChaincodeLifecycleEventListener)
+	if !ok {
+		t.Fatalf("Couchdb state impl is expected to implement interface `cceventmgmt.ChaincodeLifecycleEventListener`")
+	}
 
 	chaincodeDef := &cceventmgmt.ChaincodeDefinition{Name: "ns1", Hash: nil, Version: ""}
 

core/scc/lscc/lscc.go

@@ -432,6 +432,9 @@ func (lscc *lifeCycleSysCC) executeInstall(stub shim.ChaincodeStubInterface, ccb
 	// Note - this step is done prior to PutChaincodeToLocalStorage() since this step is idempotent and harmless until endorsements start,
 	// that is, if there are errors deploying the indexes the chaincode install can safely be re-attempted later.
 	err = cceventmgmt.GetMgr().HandleChaincodeInstall(chaincodeDefinition, statedbArtifactsTar)
+	defer func() {
+		cceventmgmt.GetMgr().ChaincodeInstallDone(err == nil)
+	}()
 	if err != nil {
 		return err
 	}