[new-prom-metrics] prometheus metrics

internal/assets/manager.go

@@ -60,16 +60,17 @@ type Manager interface {
 }
 
 type assetManager struct {
-	ctx       context.Context
-	database  database.Plugin
-	txHelper  txcommon.Helper
-	identity  identity.Manager
-	data      data.Manager
-	syncasync syncasync.Bridge
-	broadcast broadcast.Manager
-	messaging privatemessaging.Manager
-	tokens    map[string]tokens.Plugin
-	retry     retry.Retry
+	ctx            context.Context
+	database       database.Plugin
+	txHelper       txcommon.Helper
+	identity       identity.Manager
+	data           data.Manager
+	syncasync      syncasync.Bridge
+	broadcast      broadcast.Manager
+	messaging      privatemessaging.Manager
+	tokens         map[string]tokens.Plugin
+	retry          retry.Retry
+	metricsEnabled bool
 }
 
 func NewAssetManager(ctx context.Context, di database.Plugin, im identity.Manager, dm data.Manager, sa syncasync.Bridge, bm broadcast.Manager, pm privatemessaging.Manager, ti map[string]tokens.Plugin) (Manager, error) {
@@ -91,6 +92,7 @@ func NewAssetManager(ctx context.Context, di database.Plugin, im identity.Manage
 			MaximumDelay: config.GetDuration(config.AssetManagerRetryMaxDelay),
 			Factor:       config.GetFloat64(config.AssetManagerRetryFactor),
 		},
+		metricsEnabled: config.GetBool(config.MetricsEnabled),
 	}
 	return am, nil
 }

internal/assets/manager_test.go

@@ -19,6 +19,7 @@ import (
 	"testing"
 
 	"github.com/hyperledger/firefly/internal/config"
+	"github.com/hyperledger/firefly/internal/metrics"
 	"github.com/hyperledger/firefly/mocks/broadcastmocks"
 	"github.com/hyperledger/firefly/mocks/databasemocks"
 	"github.com/hyperledger/firefly/mocks/datamocks"
@@ -36,6 +37,8 @@ import (
 
 func newTestAssets(t *testing.T) (*assetManager, func()) {
 	config.Reset()
+	metrics.Registry()
+	config.Set(config.MetricsEnabled, true)
 	mdi := &databasemocks.Plugin{}
 	mim := &identitymanagermocks.Manager{}
 	mdm := &datamocks.Manager{}

internal/assets/token_transfer.go

@@ -20,8 +20,10 @@ import (
 	"context"
 	"fmt"
 
+	"github.com/hyperledger/firefly/internal/config"
 	"github.com/hyperledger/firefly/internal/i18n"
 	"github.com/hyperledger/firefly/internal/log"
+	"github.com/hyperledger/firefly/internal/metrics"
 	"github.com/hyperledger/firefly/internal/sysmessaging"
 	"github.com/hyperledger/firefly/internal/txcommon"
 	"github.com/hyperledger/firefly/pkg/database"
@@ -42,20 +44,24 @@ func (am *assetManager) GetTokenTransferByID(ctx context.Context, ns, id string)
 }
 
 func (am *assetManager) NewTransfer(ns string, transfer *fftypes.TokenTransferInput) sysmessaging.MessageSender {
+	metrics.Registry()
+	config.Set(config.MetricsEnabled, true)
 	sender := &transferSender{
-		mgr:       am,
-		namespace: ns,
-		transfer:  transfer,
+		metricsEnabled: config.GetBool(config.MetricsEnabled),
+		mgr:            am,
+		namespace:      ns,
+		transfer:       transfer,
 	}
 	sender.setDefaults()
 	return sender
 }
 
 type transferSender struct {
-	mgr       *assetManager
-	namespace string
-	transfer  *fftypes.TokenTransferInput
-	resolved  bool
+	mgr            *assetManager
+	namespace      string
+	transfer       *fftypes.TokenTransferInput
+	resolved       bool
+	metricsEnabled bool
 }
 
 // sendMethod is the specific operation requested of the transferSender.
@@ -123,8 +129,11 @@ func (am *assetManager) MintTokens(ctx context.Context, ns string, transfer *fft
 	if err := am.validateTransfer(ctx, ns, transfer); err != nil {
 		return nil, err
 	}
-
 	sender := am.NewTransfer(ns, transfer)
+	if am.metricsEnabled && len(transfer.LocalID.String()) > 0 {
+		metrics.MintSubmittedCounter.Inc()
+		metrics.AddTime(transfer.LocalID.String())
+	}
 	if waitConfirm {
 		err = sender.SendAndWait(ctx)
 	} else {
@@ -138,8 +147,11 @@ func (am *assetManager) BurnTokens(ctx context.Context, ns string, transfer *fft
 	if err := am.validateTransfer(ctx, ns, transfer); err != nil {
 		return nil, err
 	}
-
 	sender := am.NewTransfer(ns, transfer)
+	if am.metricsEnabled && len(transfer.LocalID.String()) > 0 {
+		metrics.BurnSubmittedCounter.Inc()
+		metrics.AddTime(transfer.LocalID.String())
+	}
 	if waitConfirm {
 		err = sender.SendAndWait(ctx)
 	} else {
@@ -156,8 +168,11 @@ func (am *assetManager) TransferTokens(ctx context.Context, ns string, transfer
 	if transfer.From == transfer.To {
 		return nil, i18n.NewError(ctx, i18n.MsgCannotTransferToSelf)
 	}
-
 	sender := am.NewTransfer(ns, transfer)
+	if am.metricsEnabled && len(transfer.LocalID.String()) > 0 {
+		metrics.TransferSubmittedCounter.Inc()
+		metrics.AddTime(transfer.LocalID.String())
+	}
 	if waitConfirm {
 		err = sender.SendAndWait(ctx)
 	} else {

internal/broadcast/manager.go

@@ -62,6 +62,7 @@ type broadcastManager struct {
 	syncasync             syncasync.Bridge
 	batchpin              batchpin.Submitter
 	maxBatchPayloadLength int64
+	metricsEnabled        bool
 }
 
 func NewBroadcastManager(ctx context.Context, di database.Plugin, im identity.Manager, dm data.Manager, bi blockchain.Plugin, dx dataexchange.Plugin, pi publicstorage.Plugin, ba batch.Manager, sa syncasync.Bridge, bp batchpin.Submitter) (Manager, error) {
@@ -80,6 +81,7 @@ func NewBroadcastManager(ctx context.Context, di database.Plugin, im identity.Ma
 		syncasync:             sa,
 		batchpin:              bp,
 		maxBatchPayloadLength: config.GetByteSize(config.BroadcastBatchPayloadLimit),
+		metricsEnabled:        config.GetBool(config.MetricsEnabled),
 	}
 	bo := batch.Options{
 		BatchMaxSize:   config.GetUint(config.BroadcastBatchSize),

internal/broadcast/manager_test.go

@@ -25,6 +25,7 @@ import (
 	"testing"
 
 	"github.com/hyperledger/firefly/internal/config"
+	"github.com/hyperledger/firefly/internal/metrics"
 	"github.com/hyperledger/firefly/mocks/batchmocks"
 	"github.com/hyperledger/firefly/mocks/batchpinmocks"
 	"github.com/hyperledger/firefly/mocks/blockchainmocks"
@@ -42,6 +43,8 @@ import (
 
 func newTestBroadcast(t *testing.T) (*broadcastManager, func()) {
 	config.Reset()
+	metrics.Registry()
+	config.Set(config.MetricsEnabled, true)
 	mdi := &databasemocks.Plugin{}
 	mim := &identitymanagermocks.Manager{}
 	mdm := &datamocks.Manager{}

internal/broadcast/message.go

@@ -22,6 +22,7 @@ import (
 
 	"github.com/hyperledger/firefly/internal/i18n"
 	"github.com/hyperledger/firefly/internal/log"
+	"github.com/hyperledger/firefly/internal/metrics"
 	"github.com/hyperledger/firefly/internal/sysmessaging"
 	"github.com/hyperledger/firefly/pkg/database"
 	"github.com/hyperledger/firefly/pkg/fftypes"
@@ -39,6 +40,12 @@ func (bm *broadcastManager) NewBroadcast(ns string, in *fftypes.MessageInOut) sy
 
 func (bm *broadcastManager) BroadcastMessage(ctx context.Context, ns string, in *fftypes.MessageInOut, waitConfirm bool) (out *fftypes.Message, err error) {
 	broadcast := bm.NewBroadcast(ns, in)
+
+	if bm.metricsEnabled && len(in.Header.ID.String()) > 0 {
+		metrics.BroadcastSubmittedCounter.Inc()
+		metrics.AddTime(in.Header.ID.String())
+	}
+
 	if waitConfirm {
 		err = broadcast.SendAndWait(ctx)
 	} else {
@@ -166,7 +173,7 @@ func (s *broadcastSender) sendInternal(ctx context.Context, method sendMethod) (
 	if err := s.mgr.database.UpsertMessage(ctx, &s.msg.Message, database.UpsertOptimizationNew); err != nil {
 		return err
 	}
-	log.L(ctx).Infof("Sent broadcast message %s:%s", s.msg.Header.Namespace, s.msg.Header.ID)
+	log.L(ctx).Infof("Sent broadcast message %s:%s sequence=%d", s.msg.Header.Namespace, s.msg.Header.ID, s.msg.Sequence)
 
 	return err
 }

internal/events/aggregator.go

@@ -20,11 +20,13 @@ import (
 	"context"
 	"crypto/sha256"
 	"database/sql/driver"
+	"time"
 
 	"github.com/hyperledger/firefly/internal/config"
 	"github.com/hyperledger/firefly/internal/data"
 	"github.com/hyperledger/firefly/internal/definitions"
 	"github.com/hyperledger/firefly/internal/log"
+	"github.com/hyperledger/firefly/internal/metrics"
 	"github.com/hyperledger/firefly/internal/retry"
 	"github.com/hyperledger/firefly/pkg/database"
 	"github.com/hyperledger/firefly/pkg/fftypes"
@@ -44,6 +46,7 @@ type aggregator struct {
 	offchainBatches chan *fftypes.UUID
 	queuedRewinds   chan *fftypes.UUID
 	retry           *retry.Retry
+	metricsEnabled  bool
 }
 
 func newAggregator(ctx context.Context, di database.Plugin, sh definitions.DefinitionHandlers, dm data.Manager, en *eventNotifier) *aggregator {
@@ -56,6 +59,7 @@ func newAggregator(ctx context.Context, di database.Plugin, sh definitions.Defin
 		newPins:         make(chan int64),
 		offchainBatches: make(chan *fftypes.UUID, 1), // hops to queuedRewinds with a shouldertab on the event poller
 		queuedRewinds:   make(chan *fftypes.UUID, batchSize),
+		metricsEnabled:  config.GetBool(config.MetricsEnabled),
 	}
 	firstEvent := fftypes.SubOptsFirstEvent(config.GetString(config.EventAggregatorFirstEvent))
 	ag.eventPoller = newEventPoller(ctx, di, en, &eventPollerConf{
@@ -384,6 +388,30 @@ func (ag *aggregator) attemptMessageDispatch(ctx context.Context, msg *fftypes.M
 		return nil
 	})
 
+	// Increment confirmer/rejected metrics for broadcast or private message
+	if ag.metricsEnabled {
+		timeElapsed := time.Since(metrics.GetTime(msg.Header.ID.String())).Seconds()
+		metrics.DeleteTime(msg.Header.ID.String())
+		switch msg.Header.Type {
+		case fftypes.MessageTypeBroadcast:
+			metrics.BroadcastHistogram.Observe(timeElapsed)
+			if eventType == fftypes.EventTypeMessageConfirmed {
+				metrics.BroadcastConfirmedCounter.Inc()
+			} else if eventType == fftypes.EventTypeMessageRejected {
+				metrics.BroadcastRejectedCounter.Inc()
+			}
+		case fftypes.MessageTypePrivate:
+			metrics.PrivateMsgHistogram.Observe(timeElapsed)
+			if eventType == fftypes.EventTypeMessageConfirmed {
+				metrics.PrivateMsgConfirmedCounter.Inc()
+			} else if eventType == fftypes.EventTypeMessageRejected {
+				metrics.PrivateMsgRejectedCounter.Inc()
+			}
+		default:
+			break
+		}
+	}
+
 	return true, nil
 }
 

internal/events/aggregator_test.go

@@ -24,6 +24,7 @@ import (
 
 	"github.com/hyperledger/firefly/internal/config"
 	"github.com/hyperledger/firefly/internal/definitions"
+	"github.com/hyperledger/firefly/internal/metrics"
 	"github.com/hyperledger/firefly/mocks/databasemocks"
 	"github.com/hyperledger/firefly/mocks/datamocks"
 	"github.com/hyperledger/firefly/mocks/definitionsmocks"
@@ -34,6 +35,8 @@ import (
 )
 
 func newTestAggregator() (*aggregator, func()) {
+	metrics.Registry()
+	config.Set(config.MetricsEnabled, true)
 	mdi := &databasemocks.Plugin{}
 	mdm := &datamocks.Manager{}
 	msh := &definitionsmocks.DefinitionHandlers{}
@@ -1306,6 +1309,75 @@ func TestAttemptMessageDispatchGroupInit(t *testing.T) {
 
 }
 
+func TestAttemptMessageDispatchSuccessBroadcast(t *testing.T) {
+	ag, cancel := newTestAggregator()
+	defer cancel()
+	bs := newBatchState(ag)
+
+	mdi := ag.database.(*databasemocks.Plugin)
+	mdm := ag.data.(*datamocks.Manager)
+	mdm.On("GetMessageData", ag.ctx, mock.Anything, true).Return([]*fftypes.Data{}, true, nil)
+	mdm.On("ValidateAll", ag.ctx, mock.Anything).Return(true, nil)
+	mdi.On("UpdateMessage", ag.ctx, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("InsertEvent", ag.ctx, mock.Anything).Return(nil)
+
+	_, err := ag.attemptMessageDispatch(ag.ctx, &fftypes.Message{
+		Header: fftypes.MessageHeader{
+			ID:   fftypes.NewUUID(),
+			Type: fftypes.MessageTypeBroadcast,
+		},
+	}, bs)
+	assert.NoError(t, err)
+}
+
+func TestAttemptMessageDispatchRejectedBroadcast(t *testing.T) {
+	ag, cancel := newTestAggregator()
+	defer cancel()
+	bs := newBatchState(ag)
+
+	mdi := ag.database.(*databasemocks.Plugin)
+	mdm := ag.data.(*datamocks.Manager)
+	mdm.On("GetMessageData", ag.ctx, mock.Anything, true).Return([]*fftypes.Data{}, true, nil)
+	mdm.On("ValidateAll", ag.ctx, mock.Anything).Return(false, nil)
+	mdi.On("UpdateMessage", ag.ctx, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("InsertEvent", ag.ctx, mock.Anything).Return(nil)
+
+	_, err := ag.attemptMessageDispatch(ag.ctx, &fftypes.Message{
+		Header: fftypes.MessageHeader{
+			ID:   fftypes.NewUUID(),
+			Type: fftypes.MessageTypeBroadcast,
+		},
+		Data: fftypes.DataRefs{
+			{ID: fftypes.NewUUID()},
+		},
+	}, bs)
+	assert.NoError(t, err)
+}
+
+func TestAttemptMessageDispatchRejectedPrivate(t *testing.T) {
+	ag, cancel := newTestAggregator()
+	defer cancel()
+	bs := newBatchState(ag)
+
+	mdi := ag.database.(*databasemocks.Plugin)
+	mdm := ag.data.(*datamocks.Manager)
+	mdm.On("GetMessageData", ag.ctx, mock.Anything, true).Return([]*fftypes.Data{}, true, nil)
+	mdm.On("ValidateAll", ag.ctx, mock.Anything).Return(false, nil)
+	mdi.On("UpdateMessage", ag.ctx, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("InsertEvent", ag.ctx, mock.Anything).Return(nil)
+
+	_, err := ag.attemptMessageDispatch(ag.ctx, &fftypes.Message{
+		Header: fftypes.MessageHeader{
+			ID:   fftypes.NewUUID(),
+			Type: fftypes.MessageTypePrivate,
+		},
+		Data: fftypes.DataRefs{
+			{ID: fftypes.NewUUID()},
+		},
+	}, bs)
+	assert.NoError(t, err)
+}
+
 func TestAttemptMessageUpdateMessageFail(t *testing.T) {
 	ag, cancel := newTestAggregator()
 	defer cancel()

internal/events/event_manager.go

@@ -94,6 +94,7 @@ type eventManager struct {
 	opCorrelationRetries int
 	defaultTransport     string
 	internalEvents       *system.Events
+	metricsEnabled       bool
 }
 
 func NewEventManager(ctx context.Context, ni sysmessaging.LocalNodeInfo, pi publicstorage.Plugin, di database.Plugin, im identity.Manager, dh definitions.DefinitionHandlers, dm data.Manager, bm broadcast.Manager, pm privatemessaging.Manager, am assets.Manager) (EventManager, error) {
@@ -124,6 +125,7 @@ func NewEventManager(ctx context.Context, ni sysmessaging.LocalNodeInfo, pi publ
 		newEventNotifier:     newEventNotifier,
 		newPinNotifier:       newPinNotifier,
 		aggregator:           newAggregator(ctx, di, dh, dm, newPinNotifier),
+		metricsEnabled:       config.GetBool(config.MetricsEnabled),
 	}
 	ie, _ := eifactory.GetPlugin(ctx, system.SystemEventsTransport)
 	em.internalEvents = ie.(*system.Events)

internal/events/event_manager_test.go

@@ -23,6 +23,7 @@ import (
 
 	"github.com/hyperledger/firefly/internal/config"
 	"github.com/hyperledger/firefly/internal/events/system"
+	"github.com/hyperledger/firefly/internal/metrics"
 	"github.com/hyperledger/firefly/mocks/assetmocks"
 	"github.com/hyperledger/firefly/mocks/broadcastmocks"
 	"github.com/hyperledger/firefly/mocks/databasemocks"
@@ -43,6 +44,8 @@ var testNodeID = fftypes.NewUUID()
 
 func newTestEventManager(t *testing.T) (*eventManager, func()) {
 	config.Reset()
+	metrics.Registry()
+	config.Set(config.MetricsEnabled, true)
 	ctx, cancel := context.WithCancel(context.Background())
 	mdi := &databasemocks.Plugin{}
 	mim := &identitymanagermocks.Manager{}