Added optional timestamps to chaincode events by retrieving from Fabric

Signed-off-by: Jim Zhang <[email protected]>

Author: jimthematrix

go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/google/certificate-transparency-go v1.1.1 // indirect
 	github.com/google/uuid v1.3.0 // indirect
 	github.com/gorilla/websocket v1.4.2
+	github.com/hashicorp/golang-lru v0.5.4
 	github.com/hyperledger/fabric-config v0.0.7 // indirect
 	github.com/hyperledger/fabric-protos-go v0.0.0-20201028172056-a3136dde2354
 	github.com/hyperledger/fabric-sdk-go v1.0.1-0.20210729165856-3be4ed253dcf

go.sum

@@ -334,6 +334,8 @@ github.com/hashicorp/go-version v1.2.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09
 github.com/hashicorp/go.net v0.0.1/go.mod h1:hjKkEWcCURg++eb33jQU7oqQcI9XDCnUzHA0oac0k90=
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
+github.com/hashicorp/golang-lru v0.5.4 h1:YDjusn29QI/Das2iO9M0BHnIbxPeyuCHsjMW+lJfyTc=
+github.com/hashicorp/golang-lru v0.5.4/go.mod h1:iADmTwqILo4mZ8BN3D2Q6+9jd8WM5uGBxy+E8yxSoD4=
 github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=
 github.com/hashicorp/hcl v1.0.0/go.mod h1:E5yfLk+7swimpb2L/Alb/PJmXilQ/rhwaUYs4T20WEQ=
 github.com/hashicorp/logutils v1.0.0/go.mod h1:QIAnNjmIWmVIIkWDTG1z5v++HQmx9WQRO+LraFDTW64=

internal/events/eventstream.go

@@ -30,6 +30,7 @@ import (
 	eventsapi "github.com/hyperledger/firefly-fabconnect/internal/events/api"
 	"github.com/hyperledger/firefly-fabconnect/internal/ws"
 
+	lru "github.com/hashicorp/golang-lru"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -73,6 +74,7 @@ type StreamInfo struct {
 	Webhook              *webhookActionInfo   `json:"webhook,omitempty"`
 	WebSocket            *webSocketActionInfo `json:"websocket,omitempty"`
 	Timestamps           bool                 `json:"timestamps,omitempty"` // Include block timestamps in the events generated
+	TimestampCacheSize   int                  `json:"timestampCacheSize,omitempty"`
 }
 
 type webhookActionInfo struct {
@@ -91,26 +93,27 @@ type webSocketActionInfo struct {
 type eventHandler func(*eventData)
 
 type eventStream struct {
-	sm                subscriptionManager
-	allowPrivateIPs   bool
-	spec              *StreamInfo
-	eventStream       chan *eventData
-	eventHandler      eventHandler
-	stopped           bool
-	processorDone     bool
-	pollingInterval   time.Duration
-	pollerDone        bool
-	inFlight          uint64
-	batchCond         *sync.Cond
-	batchQueue        *list.List
-	batchCount        uint64
-	initialRetryDelay time.Duration
-	backoffFactor     float64
-	updateInProgress  bool
-	updateInterrupt   chan struct{}   // a zero-sized struct used only for signaling (hand rolled alternative to context)
-	updateWG          *sync.WaitGroup // Wait group for the go routines to reply back after they have stopped
-	action            eventStreamAction
-	wsChannels        ws.WebSocketChannels
+	sm                  subscriptionManager
+	allowPrivateIPs     bool
+	spec                *StreamInfo
+	eventStream         chan *eventData
+	eventHandler        eventHandler
+	stopped             bool
+	processorDone       bool
+	pollingInterval     time.Duration
+	pollerDone          bool
+	inFlight            uint64
+	batchCond           *sync.Cond
+	batchQueue          *list.List
+	batchCount          uint64
+	initialRetryDelay   time.Duration
+	backoffFactor       float64
+	updateInProgress    bool
+	updateInterrupt     chan struct{}   // a zero-sized struct used only for signaling (hand rolled alternative to context)
+	updateWG            *sync.WaitGroup // Wait group for the go routines to reply back after they have stopped
+	action              eventStreamAction
+	wsChannels          ws.WebSocketChannels
+	blockTimestampCache *lru.Cache
 }
 
 type eventStreamAction interface {
@@ -149,6 +152,9 @@ func newEventStream(sm subscriptionManager, spec *StreamInfo, wsChannels ws.WebS
 	} else {
 		spec.ErrorHandling = ErrorHandlingSkip
 	}
+	if spec.TimestampCacheSize == 0 {
+		spec.TimestampCacheSize = DefaultTimestampCacheSize
+	}
 
 	a = &eventStream{
 		sm:                sm,
@@ -164,6 +170,10 @@ func newEventStream(sm subscriptionManager, spec *StreamInfo, wsChannels ws.WebS
 	}
 	a.eventHandler = a.handleEvent
 
+	if a.blockTimestampCache, err = lru.New(spec.TimestampCacheSize); err != nil {
+		return nil, errors.Errorf(errors.EventStreamsCreateStreamResourceErr, err)
+	}
+
 	if a.pollingInterval == 0 {
 		// Let's us do this from UTs, without exposing it
 		a.pollingInterval = 10 * time.Millisecond

internal/events/eventstream_test.go

@@ -380,7 +380,7 @@ func TestProcessEventsEnd2EndWebhook(t *testing.T) {
 		&StreamInfo{
 			BatchSize:  1,
 			Webhook:    &webhookActionInfo{},
-			Timestamps: false,
+			Timestamps: true,
 		}, db, 200)
 	defer svr.Close()
 
@@ -392,9 +392,15 @@ func TestProcessEventsEnd2EndWebhook(t *testing.T) {
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
 	go func() {
+		// the block event
 		e1s := <-eventStream
 		assert.Equal(1, len(e1s))
 		assert.Equal(uint64(11), e1s[0].BlockNumber)
+		// the chaincode event
+		e2s := <-eventStream
+		assert.Equal(1, len(e2s))
+		assert.Equal(uint64(10), e2s[0].BlockNumber)
+		assert.Equal(int64(1000000), e2s[0].Timestamp)
 		wg.Done()
 	}()
 	wg.Wait()
@@ -417,7 +423,7 @@ func TestProcessEventsEnd2EndCatchupWebhook(t *testing.T) {
 	_ = db.Init()
 	sm, stream, svr, eventStream := newTestStreamForBatching(
 		&StreamInfo{
-			BatchSize:  1,
+			BatchSize:  2,
 			Webhook:    &webhookActionInfo{},
 			Timestamps: false,
 		}, db, 200)
@@ -432,11 +438,9 @@ func TestProcessEventsEnd2EndCatchupWebhook(t *testing.T) {
 	wg.Add(1)
 	go func() {
 		e1s := <-eventStream
-		assert.Equal(1, len(e1s))
+		assert.Equal(2, len(e1s))
 		assert.Equal(uint64(1), e1s[0].BlockNumber)
-		e2s := <-eventStream
-		assert.Equal(1, len(e2s))
-		assert.Equal(uint64(11), e2s[0].BlockNumber)
+		assert.Equal(uint64(11), e1s[1].BlockNumber)
 		wg.Done()
 	}()
 	wg.Wait()
@@ -516,6 +520,10 @@ func TestProcessEventsEnd2EndWithReset(t *testing.T) {
 		e1s := <-eventStream
 		assert.Equal(1, len(e1s))
 		assert.Equal(uint64(11), e1s[0].BlockNumber)
+		// the chaincode event
+		e2s := <-eventStream
+		assert.Equal(1, len(e2s))
+		assert.Equal(uint64(10), e2s[0].BlockNumber)
 		wg.Done()
 	}()
 	wg.Wait()
@@ -625,7 +633,7 @@ func TestPauseResumeAfterCheckpoint(t *testing.T) {
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
 	go func() {
-		for i := 0; i < 1; i++ {
+		for i := 0; i < 2; i++ {
 			<-eventStream
 		}
 		wg.Done()
@@ -691,7 +699,7 @@ func TestPauseResumeBeforeCheckpoint(t *testing.T) {
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
 	go func() {
-		for i := 0; i < 1; i++ {
+		for i := 0; i < 2; i++ {
 			<-eventStream
 		}
 		wg.Done()
@@ -731,7 +739,7 @@ func TestMarkStaleOnError(t *testing.T) {
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
 	go func() {
-		for i := 0; i < 1; i++ {
+		for i := 0; i < 2; i++ {
 			<-eventStream
 		}
 		wg.Done()
@@ -809,7 +817,7 @@ func TestStoreCheckpointStoreError(t *testing.T) {
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
 	go func() {
-		for i := 0; i < 1; i++ {
+		for i := 0; i < 2; i++ {
 			<-eventStream
 		}
 		wg.Done()
@@ -1079,6 +1087,7 @@ func TestUpdateStreamMissingWebhookURL(t *testing.T) {
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
 	go func() {
+		<-eventStream
 		<-eventStream
 		wg.Done()
 	}()
@@ -1122,6 +1131,7 @@ func TestUpdateStreamInvalidWebhookURL(t *testing.T) {
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
 	go func() {
+		<-eventStream
 		<-eventStream
 		wg.Done()
 	}()

internal/events/subscription.go

@@ -140,13 +140,35 @@ func (s *subscription) processNewEvents() {
 				EventName:     ccEvent.EventName,
 				Payload:       ccEvent.Payload,
 			}
+			if s.ep.stream.spec.Timestamps {
+				s.getEventTimestamp(event)
+			}
 			if err := s.ep.processEventEntry(s.info, event); err != nil {
 				log.Errorf("Failed to process event: %s", err)
 			}
 		}
 	}
 }
 
+func (s *subscription) getEventTimestamp(evt *eventsapi.EventEntry) {
+	// the key in the cache is the block number represented as a string
+	blockNumber := strconv.FormatUint(evt.BlockNumber, 10)
+	if ts, ok := s.ep.stream.blockTimestampCache.Get(blockNumber); ok {
+		// we found the timestamp for the block in our local cache, assert it's type and return, no need to query the chain
+		evt.Timestamp = ts.(int64)
+		return
+	}
+	// we didn't find the timestamp in our cache, query the node for the block header where we can find the timestamp
+	_, block, err := s.client.QueryBlock(s.info.ChannelId, evt.BlockNumber, s.info.Signer)
+	if err != nil {
+		log.Errorf("Unable to retrieve block[%s] timestamp: %s", blockNumber, err)
+		evt.Timestamp = 0 // set to 0, we were not able to retrieve the timestamp.
+		return
+	}
+	evt.Timestamp = block.Timestamp
+	s.ep.stream.blockTimestampCache.Add(blockNumber, evt.Timestamp)
+}
+
 func (s *subscription) unsubscribe(deleting bool) {
 	log.Infof("%s: Unsubscribing existing filter (deleting=%t)", s.info.ID, deleting)
 	s.deleting = deleting

internal/events/test_helper.go

@@ -29,6 +29,7 @@ import (
 	"github.com/hyperledger/fabric-protos-go/peer"
 	"github.com/hyperledger/firefly-fabconnect/internal/conf"
 	eventsapi "github.com/hyperledger/firefly-fabconnect/internal/events/api"
+	"github.com/hyperledger/firefly-fabconnect/internal/fabric/utils"
 	"github.com/hyperledger/firefly-fabconnect/internal/kvstore"
 	mockfabric "github.com/hyperledger/firefly-fabconnect/mocks/fabric/client"
 	mockkvstore "github.com/hyperledger/firefly-fabconnect/mocks/kvstore"
@@ -190,8 +191,18 @@ func mockRPCClient(fromBlock string, withReset ...bool) *mockfabric.RPCClient {
 			Height: 10,
 		},
 	}
+	rawBlock := &utils.RawBlock{
+		Header: &common.BlockHeader{
+			Number: uint64(20),
+		},
+	}
+	block := &utils.Block{
+		Number:    uint64(20),
+		Timestamp: int64(1000000),
+	}
 	rpc.On("SubscribeEvent", mock.Anything, mock.Anything).Return(nil, roBlockEventChan, roCCEventChan, nil)
 	rpc.On("QueryChainInfo", mock.Anything, mock.Anything).Return(res, nil)
+	rpc.On("QueryBlock", mock.Anything, mock.Anything, mock.Anything).Return(rawBlock, block, nil)
 	rpc.On("Unregister", mock.Anything).Return()
 
 	go func() {
@@ -203,6 +214,11 @@ func mockRPCClient(fromBlock string, withReset ...bool) *mockfabric.RPCClient {
 		blockEventChan <- &fab.BlockEvent{
 			Block: constructBlock(11),
 		}
+		ccEventChan <- &fab.CCEvent{
+			BlockNumber: uint64(10),
+			TxID:        "3144a3ad43dcc11374832bbb71561320de81fd80d69cc8e26a9ea7d3240a5e84",
+			ChaincodeID: "asset_transfer",
+		}
 		if len(withReset) > 0 {
 			blockEventChan <- &fab.BlockEvent{
 				Block: constructBlock(11),

internal/fabric/client/api.go

@@ -22,6 +22,7 @@ import (
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/fab"
 	"github.com/hyperledger/fabric-sdk-go/pkg/common/providers/msp"
 	eventsapi "github.com/hyperledger/firefly-fabconnect/internal/events/api"
+	"github.com/hyperledger/firefly-fabconnect/internal/fabric/utils"
 )
 
 type ChaincodeSpec struct {
@@ -54,6 +55,7 @@ type RPCClient interface {
 	Invoke(channelId, signer, chaincodeName, method string, args []string, isInit bool) (*TxReceipt, error)
 	Query(channelId, signer, chaincodeName, method string, args []string) ([]byte, error)
 	QueryChainInfo(channelId, signer string) (*fab.BlockchainInfoResponse, error)
+	QueryBlock(channelId string, blockNumber uint64, signer string) (*utils.RawBlock, *utils.Block, error)
 	QueryTransaction(channelId, signer, txId string) (map[string]interface{}, error)
 	SubscribeEvent(subInfo *eventsapi.SubscriptionInfo, since uint64) (*RegistrationWrapper, <-chan *fab.BlockEvent, <-chan *fab.CCEvent, error)
 	Unregister(*RegistrationWrapper)

internal/fabric/client/client_ccp.go

@@ -29,6 +29,7 @@ import (
 	mspImpl "github.com/hyperledger/fabric-sdk-go/pkg/msp"
 	"github.com/hyperledger/firefly-fabconnect/internal/errors"
 	eventsapi "github.com/hyperledger/firefly-fabconnect/internal/events/api"
+	"github.com/hyperledger/firefly-fabconnect/internal/fabric/utils"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -148,6 +149,19 @@ func (w *ccpRPCWrapper) QueryChainInfo(channelId, signer string) (*fab.Blockchai
 	return result, nil
 }
 
+func (w *ccpRPCWrapper) QueryBlock(channelId string, blockNumber uint64, signer string) (*utils.RawBlock, *utils.Block, error) {
+	log.Tracef("RPC [%s] --> QueryBlock %v", channelId, blockNumber)
+
+	rawblock, block, err := w.ledgerClientWrapper.queryBlock(channelId, blockNumber, signer)
+	if err != nil {
+		log.Errorf("Failed to query block %v on channel %s. %s", blockNumber, channelId, err)
+		return nil, nil, err
+	}
+
+	log.Tracef("RPC [%s] <-- success", channelId)
+	return rawblock, block, nil
+}
+
 // The returned registration must be closed when done
 func (w *ccpRPCWrapper) SubscribeEvent(subInfo *eventsapi.SubscriptionInfo, since uint64) (*RegistrationWrapper, <-chan *fab.BlockEvent, <-chan *fab.CCEvent, error) {
 	reg, blockEventCh, ccEventCh, err := w.eventClientWrapper.subscribeEvent(subInfo, since)

internal/fabric/client/client_gateway_clientside.go

@@ -25,6 +25,7 @@ import (
 	"github.com/hyperledger/fabric-sdk-go/pkg/gateway"
 	"github.com/hyperledger/firefly-fabconnect/internal/errors"
 	eventsapi "github.com/hyperledger/firefly-fabconnect/internal/events/api"
+	"github.com/hyperledger/firefly-fabconnect/internal/fabric/utils"
 	log "github.com/sirupsen/logrus"
 )
 
@@ -110,6 +111,19 @@ func (w *gwRPCWrapper) QueryChainInfo(channelId, signer string) (*fab.Blockchain
 	return result, nil
 }
 
+func (w *gwRPCWrapper) QueryBlock(channelId string, blockNumber uint64, signer string) (*utils.RawBlock, *utils.Block, error) {
+	log.Tracef("RPC [%s] --> QueryBlock %v", channelId, blockNumber)
+
+	rawblock, block, err := w.ledgerClientWrapper.queryBlock(channelId, blockNumber, signer)
+	if err != nil {
+		log.Errorf("Failed to query block %v on channel %s. %s", blockNumber, channelId, err)
+		return nil, nil, err
+	}
+
+	log.Tracef("RPC [%s] <-- success", channelId)
+	return rawblock, block, nil
+}
+
 func (w *gwRPCWrapper) QueryTransaction(channelId, signer, txId string) (map[string]interface{}, error) {
 	log.Tracef("RPC [%s] --> QueryTransaction %s", channelId, txId)
 

internal/fabric/client/ledger.go

@@ -60,6 +60,19 @@ func (l *ledgerClientWrapper) queryChainInfo(channelId, signer string) (*fab.Blo
 	return result, nil
 }
 
+func (l *ledgerClientWrapper) queryBlock(channelId string, blockNumber uint64, signer string) (*utils.RawBlock, *utils.Block, error) {
+	client, err := l.getLedgerClient(channelId, signer)
+	if err != nil {
+		return nil, nil, errors.Errorf("Failed to get channel client. %s", err)
+	}
+	result, err := client.QueryBlock(blockNumber)
+	if err != nil {
+		return nil, nil, err
+	}
+	rawblock, block, err := utils.DecodeBlock(result)
+	return rawblock, block, err
+}
+
 func (l *ledgerClientWrapper) queryTransaction(channelId, signer, txId string) (map[string]interface{}, error) {
 	client, err := l.getLedgerClient(channelId, signer)
 	if err != nil {