[metrics] [dataexchange] [networkmap] DXConnectEvent Callbacks for Node Identity Check Metrics

Signed-off-by: hfuss <hayden.fuss@kaleido.io>

Author: onelapahead

.gitignore

@@ -12,5 +12,6 @@ __debug*
 !deploy/charts/firefly
 containerlogs
 .vscode/*.log
-.idea
-doc-site/site
+.idea/
+doc-site/site
+*.iml

internal/coremsgs/en_error_messages.go

@@ -318,4 +318,5 @@ var (
 	MsgDuplicateContractListenerFilterLocation = ffe("FF10477", "Duplicate filter provided for contract listener for location", 400)
 	MsgInvalidNamespaceForOperationUpdate      = ffe("FF10478", "Received different namespace for operation update '%s' than expected for manager '%s'")
 	MsgEmptyPluginForOperationUpdate           = ffe("FF10479", "Received empty plugin for operation update '%s'")
+	MsgInvalidIdentityPatch                    = ffe("FF10480", "A profile must be provided when updating an identity", 400)
 )

internal/dataexchange/ffdx/ffdx.go

@@ -295,6 +295,14 @@ func (h *FFDX) beforeConnect(ctx context.Context, w wsclient.WSClient) error {
 			return fmt.Errorf("DX returned non-ready status: %s", status.Status)
 		}
 	}
+
+	for _, cb := range h.callbacks.handlers {
+		err := cb.DXConnectEvent(h)
+		if err != nil {
+			log.L(ctx).Errorf("error handling DX connect event: %v", err)
+		}
+	}
+
 	h.initialized = true
 	return nil
 }

internal/dataexchange/ffdx/ffdx_test.go

@@ -488,6 +488,8 @@ func TestMessageEventsBackgroundStart(t *testing.T) {
 	h.SetOperationHandler("ns1", ocb)
 	h.AddNode(context.Background(), "ns1", "node1", fftypes.JSONObject{"id": "peer1"})
 
+	mcb.On("DXConnectEvent", h).Return(nil)
+
 	err := h.Start()
 	assert.NoError(t, err)
 
@@ -547,6 +549,7 @@ func TestMessageEvents(t *testing.T) {
 	ocb := &coremocks.OperationCallbacks{}
 	h.SetOperationHandler("ns1", ocb)
 	h.AddNode(context.Background(), "ns1", "node1", fftypes.JSONObject{"id": "peer1"})
+	mcb.On("DXConnectEvent", h).Return(nil)
 
 	err := h.Start()
 	assert.NoError(t, err)
@@ -612,6 +615,7 @@ func TestBlobEvents(t *testing.T) {
 	ocb := &coremocks.OperationCallbacks{}
 	h.SetOperationHandler("ns1", ocb)
 	h.AddNode(context.Background(), "ns1", "node1", fftypes.JSONObject{"id": "peer1"})
+	mcb.On("DXConnectEvent", h).Return(nil)
 
 	err := h.Start()
 	assert.NoError(t, err)

internal/metrics/batch_pin.go

@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2025 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -20,16 +20,16 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 )
 
-var BatchPinCounter prometheus.Counter
+var BatchPinCounter *prometheus.CounterVec
 
 // MetricsBatchPin is the prometheus metric for total number of batch pins submitted
 var MetricsBatchPin = "ff_batchpin_total"
 
 func InitBatchPinMetrics() {
-	BatchPinCounter = prometheus.NewCounter(prometheus.CounterOpts{
+	BatchPinCounter = prometheus.NewCounterVec(prometheus.CounterOpts{
 		Name: MetricsBatchPin,
 		Help: "Number of batch pins submitted",
-	})
+	}, namespaceLabels)
 }
 
 func RegisterBatchPinMetrics() {

internal/metrics/broadcast.go

@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2025 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -20,10 +20,10 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 )
 
-var BroadcastSubmittedCounter prometheus.Counter
-var BroadcastConfirmedCounter prometheus.Counter
-var BroadcastRejectedCounter prometheus.Counter
-var BroadcastHistogram prometheus.Histogram
+var BroadcastSubmittedCounter *prometheus.CounterVec
+var BroadcastConfirmedCounter *prometheus.CounterVec
+var BroadcastRejectedCounter *prometheus.CounterVec
+var BroadcastHistogram *prometheus.HistogramVec
 
 // BroadcastSubmittedCounterName is the prometheus metric for tracking the total number of broadcasts submitted
 var BroadcastSubmittedCounterName = "ff_broadcast_submitted_total"
@@ -38,22 +38,22 @@ var BroadcastRejectedCounterName = "ff_broadcast_rejected_total"
 var BroadcastHistogramName = "ff_broadcast_histogram"
 
 func InitBroadcastMetrics() {
-	BroadcastSubmittedCounter = prometheus.NewCounter(prometheus.CounterOpts{
+	BroadcastSubmittedCounter = prometheus.NewCounterVec(prometheus.CounterOpts{
 		Name: BroadcastSubmittedCounterName,
 		Help: "Number of submitted broadcasts",
-	})
-	BroadcastConfirmedCounter = prometheus.NewCounter(prometheus.CounterOpts{
+	}, namespaceLabels)
+	BroadcastConfirmedCounter = prometheus.NewCounterVec(prometheus.CounterOpts{
 		Name: BroadcastConfirmedCounterName,
 		Help: "Number of confirmed broadcasts",
-	})
-	BroadcastRejectedCounter = prometheus.NewCounter(prometheus.CounterOpts{
+	}, namespaceLabels)
+	BroadcastRejectedCounter = prometheus.NewCounterVec(prometheus.CounterOpts{
 		Name: BroadcastRejectedCounterName,
 		Help: "Number of rejected broadcasts",
-	})
-	BroadcastHistogram = prometheus.NewHistogram(prometheus.HistogramOpts{
+	}, namespaceLabels)
+	BroadcastHistogram = prometheus.NewHistogramVec(prometheus.HistogramOpts{
 		Name: BroadcastHistogramName,
 		Help: "Histogram of broadcasts, bucketed by time to finished",
-	})
+	}, namespaceLabels)
 }
 
 func RegisterBroadcastMetrics() {

internal/metrics/contracts.go

@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2025 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //

internal/metrics/identity.go

@@ -0,0 +1,76 @@
+// Copyright © 2025 Kaleido, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+// 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.
+
+package metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+	"time"
+)
+
+var NodeIdentityDXCertMismatchGauge *prometheus.GaugeVec
+var NodeIdentityDXCertExpiryGauge *prometheus.GaugeVec
+
+const (
+	NodeIdentityDXCertMismatch = "ff_multiparty_node_identity_dx_mismatch"
+	NodeIdentityDXCertExpiry   = "ff_multiparty_node_identity_dx_expiry_epoch"
+)
+
+func InitIdentityMetrics() {
+	NodeIdentityDXCertMismatchGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: NodeIdentityDXCertMismatch,
+		Help: "Status of node identity DX cert mismatch",
+	}, namespaceLabels)
+
+	NodeIdentityDXCertExpiryGauge = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: NodeIdentityDXCertExpiry,
+		Help: "Timestamp, in Unix epoch format, of node identity DX cert's expiry",
+	}, namespaceLabels)
+}
+
+func RegisterIdentityMetrics() {
+	registry.MustRegister(NodeIdentityDXCertMismatchGauge)
+	registry.MustRegister(NodeIdentityDXCertExpiryGauge)
+}
+
+// TODO should this type live elsewhere ??
+type NodeIdentityDXCertMismatchStatus string
+
+const (
+	NodeIdentityDXCertMismatchStatusMismatched NodeIdentityDXCertMismatchStatus = "mismatched"
+	NodeIdentityDXCertMismatchStatusHealthy    NodeIdentityDXCertMismatchStatus = "healthy"
+	NodeIdentityDXCertMismatchStatusUnknown    NodeIdentityDXCertMismatchStatus = "unknown"
+)
+
+func (mm *metricsManager) NodeIdentityDXCertMismatch(namespace string, state NodeIdentityDXCertMismatchStatus) {
+	var gaugeState float64
+	switch state {
+	case NodeIdentityDXCertMismatchStatusMismatched:
+		gaugeState = 1.0
+	case NodeIdentityDXCertMismatchStatusHealthy:
+		gaugeState = 0.0
+	case NodeIdentityDXCertMismatchStatusUnknown:
+		fallthrough
+	default:
+		gaugeState = -1.0
+	}
+
+	NodeIdentityDXCertMismatchGauge.WithLabelValues(namespace).Set(gaugeState)
+}
+
+func (mm *metricsManager) NodeIdentityDXCertExpiry(namespace string, expiry time.Time) {
+	NodeIdentityDXCertExpiryGauge.WithLabelValues(namespace).Set(float64(expiry.UTC().Unix()))
+}

internal/metrics/metrics.go

@@ -29,8 +29,10 @@ import (
 
 var mutex = &sync.Mutex{}
 
+var namespaceLabels = []string{"ns"}
+
 type Manager interface {
-	CountBatchPin()
+	CountBatchPin(namespace string)
 	MessageSubmitted(msg *core.Message)
 	MessageConfirmed(msg *core.Message, eventType fftypes.FFEnum)
 	TransferSubmitted(transfer *core.TokenTransfer)
@@ -39,6 +41,8 @@ type Manager interface {
 	BlockchainTransaction(location, methodName string)
 	BlockchainQuery(location, methodName string)
 	BlockchainEvent(location, signature string)
+	NodeIdentityDXCertMismatch(namespace string, mismatch NodeIdentityDXCertMismatchStatus)
+	NodeIdentityDXCertExpiry(namespace string, expiry time.Time)
 	AddTime(id string)
 	GetTime(id string) time.Time
 	DeleteTime(id string)
@@ -61,17 +65,17 @@ func NewMetricsManager(ctx context.Context) Manager {
 	return mm
 }
 
-func (mm *metricsManager) CountBatchPin() {
-	BatchPinCounter.Inc()
+func (mm *metricsManager) CountBatchPin(namespace string) {
+	BatchPinCounter.WithLabelValues(namespace).Inc()
 }
 
 func (mm *metricsManager) MessageSubmitted(msg *core.Message) {
 	if len(msg.Header.ID.String()) > 0 {
 		switch msg.Header.Type {
 		case core.MessageTypeBroadcast:
-			BroadcastSubmittedCounter.Inc()
+			BroadcastSubmittedCounter.WithLabelValues(msg.LocalNamespace).Inc()
 		case core.MessageTypePrivate:
-			PrivateMsgSubmittedCounter.Inc()
+			PrivateMsgSubmittedCounter.WithLabelValues(msg.LocalNamespace).Inc()
 		}
 		mm.AddTime(msg.Header.ID.String())
 	}
@@ -87,23 +91,23 @@ func (mm *metricsManager) MessageConfirmed(msg *core.Message, eventType fftypes.
 		if !eventTime.IsZero() {
 			// Check that we recorded the submission
 			// as we might not be the party submitting
-			BroadcastHistogram.Observe(timeElapsed)
+			BroadcastHistogram.WithLabelValues(msg.LocalNamespace).Observe(timeElapsed)
 		}
 		if eventType == core.EventTypeMessageConfirmed { // Broadcast Confirmed
-			BroadcastConfirmedCounter.Inc()
+			BroadcastConfirmedCounter.WithLabelValues(msg.LocalNamespace).Inc()
 		} else if eventType == core.EventTypeMessageRejected { // Broadcast Rejected
-			BroadcastRejectedCounter.Inc()
+			BroadcastRejectedCounter.WithLabelValues(msg.LocalNamespace).Inc()
 		}
 	case core.MessageTypePrivate:
 		if !eventTime.IsZero() {
 			// Check that we recorded the submission
 			// as we might not be the party submitting
-			PrivateMsgHistogram.Observe(timeElapsed)
+			PrivateMsgHistogram.WithLabelValues(msg.LocalNamespace).Observe(timeElapsed)
 		}
 		if eventType == core.EventTypeMessageConfirmed { // Private Msg Confirmed
-			PrivateMsgConfirmedCounter.Inc()
+			PrivateMsgConfirmedCounter.WithLabelValues(msg.LocalNamespace).Inc()
 		} else if eventType == core.EventTypeMessageRejected { // Private Msg Rejected
-			PrivateMsgRejectedCounter.Inc()
+			PrivateMsgRejectedCounter.WithLabelValues(msg.LocalNamespace).Inc()
 		}
 	}
 }
@@ -112,11 +116,11 @@ func (mm *metricsManager) TransferSubmitted(transfer *core.TokenTransfer) {
 	if len(transfer.LocalID.String()) > 0 {
 		switch transfer.Type {
 		case core.TokenTransferTypeMint: // Mint submitted
-			MintSubmittedCounter.Inc()
+			MintSubmittedCounter.WithLabelValues(transfer.Namespace).Inc()
 		case core.TokenTransferTypeTransfer: // Transfer submitted
-			TransferSubmittedCounter.Inc()
+			TransferSubmittedCounter.WithLabelValues(transfer.Namespace).Inc()
 		case core.TokenTransferTypeBurn: // Burn submitted
-			BurnSubmittedCounter.Inc()
+			BurnSubmittedCounter.WithLabelValues(transfer.Namespace).Inc()
 		}
 		mm.AddTime(transfer.LocalID.String())
 	}
@@ -130,19 +134,19 @@ func (mm *metricsManager) TransferConfirmed(transfer *core.TokenTransfer) {
 	switch transfer.Type {
 	case core.TokenTransferTypeMint: // Mint confirmed
 		if !transferEvent.IsZero() {
-			MintHistogram.Observe(timeElapsed)
+			MintHistogram.WithLabelValues(transfer.Namespace).Observe(timeElapsed)
 		}
-		MintConfirmedCounter.Inc()
+		MintConfirmedCounter.WithLabelValues(transfer.Namespace).Inc()
 	case core.TokenTransferTypeTransfer: // Transfer confirmed
 		if !transferEvent.IsZero() {
-			TransferHistogram.Observe(timeElapsed)
+			TransferHistogram.WithLabelValues(transfer.Namespace).Observe(timeElapsed)
 		}
-		TransferConfirmedCounter.Inc()
+		TransferConfirmedCounter.WithLabelValues(transfer.Namespace).Inc()
 	case core.TokenTransferTypeBurn: // Burn confirmed
 		if !transferEvent.IsZero() {
-			BurnHistogram.Observe(timeElapsed)
+			BurnHistogram.WithLabelValues(transfer.Namespace).Observe(timeElapsed)
 		}
-		BurnConfirmedCounter.Inc()
+		BurnConfirmedCounter.WithLabelValues(transfer.Namespace).Inc()
 	}
 }
 

internal/metrics/metrics_test.go

@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2025 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -62,7 +62,7 @@ func newTestMetricsManager(t *testing.T) (*metricsManager, func()) {
 func TestCountBatchPin(t *testing.T) {
 	mm, cancel := newTestMetricsManager(t)
 	defer cancel()
-	mm.CountBatchPin()
+	mm.CountBatchPin("a-ns")
 }
 
 func TestMessageSubmittedBroadcast(t *testing.T) {
@@ -228,3 +228,35 @@ func TestIsMetricsEnabledFalse(t *testing.T) {
 	mm.metricsEnabled = false
 	assert.Equal(t, mm.IsMetricsEnabled(), false)
 }
+
+func TestNodeIdentityDXCertMismatchSetsMismatchedState(t *testing.T) {
+	mm, cancel := newTestMetricsManager(t)
+	defer cancel()
+	mm.NodeIdentityDXCertMismatch("test-namespace", NodeIdentityDXCertMismatchStatusMismatched)
+	gaugeValue := testutil.ToFloat64(NodeIdentityDXCertMismatchGauge.WithLabelValues("test-namespace"))
+	assert.Equal(t, 1.0, gaugeValue)
+}
+
+func TestNodeIdentityDXCertMismatchSetsHealthyState(t *testing.T) {
+	mm, cancel := newTestMetricsManager(t)
+	defer cancel()
+	mm.NodeIdentityDXCertMismatch("test-namespace", NodeIdentityDXCertMismatchStatusHealthy)
+	gaugeValue := testutil.ToFloat64(NodeIdentityDXCertMismatchGauge.WithLabelValues("test-namespace"))
+	assert.Equal(t, 0.0, gaugeValue)
+}
+
+func TestNodeIdentityDXCertMismatchSetsUnknownState(t *testing.T) {
+	mm, cancel := newTestMetricsManager(t)
+	defer cancel()
+	mm.NodeIdentityDXCertMismatch("test-namespace", NodeIdentityDXCertMismatchStatusUnknown)
+	gaugeValue := testutil.ToFloat64(NodeIdentityDXCertMismatchGauge.WithLabelValues("test-namespace"))
+	assert.Equal(t, -1.0, gaugeValue)
+}
+
+func TestNodeIdentityDXCertMismatchSetsDefaultState(t *testing.T) {
+	mm, cancel := newTestMetricsManager(t)
+	defer cancel()
+	mm.NodeIdentityDXCertMismatch("test-namespace", "invalid-state")
+	gaugeValue := testutil.ToFloat64(NodeIdentityDXCertMismatchGauge.WithLabelValues("test-namespace"))
+	assert.Equal(t, -1.0, gaugeValue)
+}