Add metrics to heartbeat event.

cmd/algod/main.go

@@ -36,6 +36,7 @@ import (
 	"github.com/algorand/go-algorand/logging"
 	"github.com/algorand/go-algorand/logging/telemetryspec"
 	"github.com/algorand/go-algorand/protocol"
+	"github.com/algorand/go-algorand/util/metrics"
 	"github.com/algorand/go-algorand/util/tokens"
 )
 
@@ -71,13 +72,20 @@ func main() {
 		rand.Seed(time.Now().UnixNano())
 	}
 
+	version := config.GetCurrentVersion()
 	if *versionCheck {
-		version := config.GetCurrentVersion()
 		fmt.Printf("%d\n%s.%s [%s] (commit #%s)\n%s\n", version.AsUInt64(), version.String(),
 			version.Channel, version.Branch, version.GetCommitHash(), config.GetLicenseInfo())
 		return
 	}
 
+	heartbeatGauge := metrics.MakeStringGauge()
+	heartbeatGauge.Set("version", version.String())
+	heartbeatGauge.Set("version-num", strconv.FormatUint(version.AsUInt64(), 10))
+	heartbeatGauge.Set("channel", version.Channel)
+	heartbeatGauge.Set("branch", version.Branch)
+	heartbeatGauge.Set("commit-hash", version.GetCommitHash())
+
 	if *branchCheck {
 		fmt.Println(config.Branch)
 		return
@@ -183,10 +191,18 @@ func main() {
 
 				log.EventWithDetails(telemetryspec.ApplicationState, telemetryspec.StartupEvent, startupDetails)
 				// Send a heartbeat event every 10 minutes as a sign of life
+				ticker := time.NewTicker(10 * time.Minute)
 				go func() {
+					values := make(map[string]string)
 					for {
-						log.Event(telemetryspec.ApplicationState, telemetryspec.HeartbeatEvent)
-						<-time.After(10 * time.Minute)
+						metrics.DefaultRegistry().AddMetrics(values)
+
+						heartbeatDetails := telemetryspec.HeartbeatEventDetails{
+							Metrics: values,
+						}
+
+						log.EventWithDetails(telemetryspec.ApplicationState, telemetryspec.HeartbeatEvent, heartbeatDetails)
+						<-ticker.C
 					}
 				}()
 			}

logging/telemetryspec/event.go

@@ -41,6 +41,11 @@ type StartupEventDetails struct {
 // HeartbeatEvent is sent periodically to indicate node is running
 const HeartbeatEvent Event = "Heartbeat"
 
+// HeartbeatEventDetails contains details for the StartupEvent
+type HeartbeatEventDetails struct {
+	Metrics map[string]string
+}
+
 // CatchupStartEvent event
 const CatchupStartEvent Event = "CatchupStart"
 

network/wsNetwork.go

@@ -109,6 +109,10 @@ var meanPing = metrics.MakeGauge(metrics.MetricName{Name: "algod_network_peer_me
 var medianPing = metrics.MakeGauge(metrics.MetricName{Name: "algod_network_peer_median_ping_seconds", Description: "Network round trip time to median peer in seconds."})
 var maxPing = metrics.MakeGauge(metrics.MetricName{Name: "algod_network_peer_max_ping_seconds", Description: "Network round trip time to slowest peer in seconds."})
 
+var peers = metrics.MakeGauge(metrics.MetricName{Name: "algod_network_peers", Description: "Number of active peers."})
+var incomingPeers = metrics.MakeGauge(metrics.MetricName{Name: "algod_network_incoming_peers", Description: "Number of active incoming peers."})
+var outgoingPeers = metrics.MakeGauge(metrics.MetricName{Name: "algod_network_outgoing_peers", Description: "Number of active outgoing peers."})
+
 // Peer opaque interface for referring to a neighbor in the network
 type Peer interface{}
 
@@ -849,6 +853,9 @@ func (wn *WebsocketNetwork) ServeHTTP(response http.ResponseWriter, request *htt
 			Incoming:     true,
 			InstanceName: otherInstanceName,
 		})
+
+	peers.Set(float64(wn.NumPeers()), nil)
+	incomingPeers.Set(float64(wn.numIncomingPeers()), nil)
 }
 
 func (wn *WebsocketNetwork) messageHandlerThread() {
@@ -1446,6 +1453,9 @@ func (wn *WebsocketNetwork) tryConnect(addr, gossipAddr string) {
 			InstanceName: myInstanceName,
 		})
 
+	peers.Set(float64(wn.NumPeers()), nil)
+	outgoingPeers.Set(float64(wn.numOutgoingPeers()), nil)
+
 	if wn.prioScheme != nil {
 		challenge := response.Header.Get(PriorityChallengeHeader)
 		if challenge != "" {
@@ -1521,6 +1531,10 @@ func (wn *WebsocketNetwork) removePeer(peer *wsPeer, reason disconnectReason) {
 			Reason: string(reason),
 		})
 
+	peers.Set(float64(wn.NumPeers()), nil)
+	incomingPeers.Set(float64(wn.numIncomingPeers()), nil)
+	outgoingPeers.Set(float64(wn.numOutgoingPeers()), nil)
+
 	wn.peersLock.Lock()
 	defer wn.peersLock.Unlock()
 	if peer.peerIndex < len(wn.peers) && wn.peers[peer.peerIndex] == peer {

util/metrics/counter.go

@@ -172,3 +172,23 @@ func (counter *Counter) WriteMetric(buf *strings.Builder, parentLabels string) {
 		buf.WriteString("\n")
 	}
 }
+
+// AddMetric adds the metric into the map
+func (counter *Counter) AddMetric(values map[string]string) {
+	counter.Lock()
+	defer counter.Unlock()
+
+	if len(counter.values) < 1 {
+		return
+	}
+
+	for _, l := range counter.values {
+		sum := l.counter
+		if len(l.labels) == 0 {
+			sum += float64(atomic.LoadUint64(&counter.intValue))
+		}
+
+		values[counter.name] = strconv.FormatFloat(sum, 'f', -1, 32)
+	}
+
+}

util/metrics/gauge.go

@@ -184,3 +184,19 @@ func (gauge *Gauge) WriteMetric(buf *strings.Builder, parentLabels string) {
 		buf.WriteString("\n")
 	}
 }
+
+// AddMetric adds the metric into the map
+func (gauge *Gauge) AddMetric(values map[string]string) {
+	gauge.Lock()
+	defer gauge.Unlock()
+
+	gauge.filterExpiredMetrics()
+
+	if len(gauge.valuesIndices) < 1 {
+		return
+	}
+
+	for _, l := range gauge.valuesIndices {
+		values[gauge.name] = strconv.FormatFloat(l.gauge, 'f', -1, 32)
+	}
+}

util/metrics/registry.go

@@ -66,3 +66,12 @@ func (r *Registry) WriteMetrics(buf *strings.Builder, parentLabels string) {
 		m.WriteMetric(buf, parentLabels)
 	}
 }
+
+// AddMetrics will add all the metrics that were registered to this registry
+func (r *Registry) AddMetrics(values map[string]string) {
+	r.metricsMu.Lock()
+	defer r.metricsMu.Unlock()
+	for _, m := range r.metrics {
+		m.AddMetric(values)
+	}
+}

util/metrics/registryCommon.go

@@ -25,6 +25,7 @@ import (
 // Metric represent any collectable metric
 type Metric interface {
 	WriteMetric(buf *strings.Builder, parentLabels string)
+	AddMetric(values map[string]string)
 }
 
 // Registry represents a single set of metrics registry

util/metrics/registry_test.go

@@ -0,0 +1,46 @@
+// +build telemetry
+
+package metrics
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestWriteAdd(t *testing.T) {
+	// Test AddMetrics and WriteMetrics with a counter
+	counter := MakeCounter(MetricName{Name: "gauge-name", Description: "gauge description"})
+	counter.Add(12.34, nil)
+
+	results := make(map[string]string)
+	DefaultRegistry().AddMetrics(results)
+
+	require.Equal(t, 1, len(results))
+	require.True(t, hasKey(results, "gauge-name"))
+	require.Equal(t, "12.34", results["gauge-name"])
+
+	bufBefore := strings.Builder{}
+	DefaultRegistry().WriteMetrics(&bufBefore, "label")
+	require.True(t, bufBefore.Len() > 0)
+
+	// Test that WriteMetrics does not change after adding a StringGauge
+	stringGauge := MakeStringGauge()
+	stringGauge.Set("string-key", "value")
+
+	DefaultRegistry().AddMetrics(results)
+
+	require.True(t, hasKey(results, "string-key"))
+	require.Equal(t, "value", results["string-key"])
+	require.True(t, hasKey(results, "gauge-name"))
+	require.Equal(t, "12.34", results["gauge-name"])
+
+	// not included in string builder
+	bufAfter := strings.Builder{}
+	DefaultRegistry().WriteMetrics(&bufAfter, "label")
+	require.Equal(t, bufBefore.String(), bufAfter.String())
+
+	stringGauge.Deregister(nil)
+	counter.Deregister(nil)
+}

util/metrics/stringGauge.go

@@ -0,0 +1,48 @@
+package metrics
+
+import (
+	"strings"
+)
+
+// MakeStringGauge create a new StringGauge.
+func MakeStringGauge() *StringGauge {
+	c := &StringGauge{
+		values: make(map[string]string),
+	}
+	c.Register(nil)
+	return c
+}
+
+// Register registers the StringGauge with the default/specific registry
+func (stringGauge *StringGauge) Register(reg *Registry) {
+	if reg == nil {
+		DefaultRegistry().Register(stringGauge)
+	} else {
+		reg.Register(stringGauge)
+	}
+}
+
+// Deregister deregisters the StringGauge with the default/specific registry
+func (stringGauge *StringGauge) Deregister(reg *Registry) {
+	if reg == nil {
+		DefaultRegistry().Deregister(stringGauge)
+	} else {
+		reg.Deregister(stringGauge)
+	}
+}
+
+// Set updates a key with a value.
+func (stringGauge *StringGauge) Set(key string, value string) {
+	stringGauge.values[key] = value
+}
+
+// WriteMetric omit string gauges from the metrics report, not sure how they act with prometheus
+func (stringGauge *StringGauge) WriteMetric(buf *strings.Builder, parentLabels string) {
+}
+
+// AddMetric sets all the key value pairs in the provided map.
+func (stringGauge *StringGauge) AddMetric(values map[string]string) {
+	for k, v := range stringGauge.values {
+		values[k] = v
+	}
+}

util/metrics/stringGaugeCommon.go

@@ -0,0 +1,11 @@
+package metrics
+
+import (
+	"github.com/algorand/go-deadlock"
+)
+
+// StringGauge represents a map of key value pairs available to be written with the AddMetric
+type StringGauge struct {
+	deadlock.Mutex
+	values map[string]string
+}

util/metrics/stringGauge_test.go

@@ -0,0 +1,36 @@
+package metrics
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func hasKey(data map[string]string, key string) bool {
+	_, ok := data[key]
+	return ok
+}
+
+func TestMetricStringGauge(t *testing.T) {
+	stringGauge := MakeStringGauge()
+	stringGauge.Set("number-key", "1")
+	stringGauge.Set("string-key", "value")
+
+	results := make(map[string]string)
+	DefaultRegistry().AddMetrics(results)
+
+	// values are populated
+	require.Equal(t, 2, len(results))
+	require.True(t, hasKey(results, "number-key"))
+	require.Equal(t, "1", results["number-key"])
+	require.True(t, hasKey(results, "string-key"))
+	require.Equal(t, "value", results["string-key"])
+
+	// not included in string builder
+	buf := strings.Builder{}
+	DefaultRegistry().WriteMetrics(&buf, "not used")
+	require.Equal(t, "", buf.String())
+
+	stringGauge.Deregister(nil)
+}