feat(telemetry)_: replace telemetry with prometheus metrics

api/geth_backend.go

@@ -38,6 +38,7 @@ import (
 	"github.com/status-im/status-go/internal/sentry"
 	"github.com/status-im/status-go/internal/version"
 	"github.com/status-im/status-go/logutils"
+	"github.com/status-im/status-go/metrics"
 	"github.com/status-im/status-go/multiaccounts"
 	"github.com/status-im/status-go/multiaccounts/accounts"
 	multiacccommon "github.com/status-im/status-go/multiaccounts/common"
@@ -103,6 +104,7 @@ type GethStatusBackend struct {
 	allowAllRPC              bool // used only for tests, disables api method restrictions
 	LocalPairingStateManager *statecontrol.ProcessStateManager
 	centralizedMetrics       *centralizedmetrics.MetricService
+	prometheusMetrics        *metrics.Server
 	sentryDSN                string
 
 	logger *zap.Logger
@@ -267,6 +269,15 @@ func (b *GethStatusBackend) AcceptTerms() error {
 	return b.multiaccountsDB.UpdateHasAcceptedTerms(accounts[0].KeyUID, true)
 }
 
+func (b *GethStatusBackend) StartPrometheusMetricsServer(address string) error {
+	if b.prometheusMetrics != nil {
+		return nil
+	}
+	b.prometheusMetrics = metrics.NewMetricsServer(address, nil)
+	go b.prometheusMetrics.Listen()
+	return nil
+}
+
 func (b *GethStatusBackend) getAccountByKeyUID(keyUID string) (*multiaccounts.Account, error) {
 	b.mu.Lock()
 	defer b.mu.Unlock()

cmd/status-cli/util.go

@@ -1,7 +1,6 @@
 package main
 
 import (
-	"context"
 	"errors"
 	"fmt"
 	"os"
@@ -12,11 +11,11 @@ import (
 
 	"github.com/status-im/status-go/api"
 	"github.com/status-im/status-go/logutils"
+	"github.com/status-im/status-go/metrics/wakumetrics"
 	"github.com/status-im/status-go/multiaccounts"
 	"github.com/status-im/status-go/params"
 	"github.com/status-im/status-go/protocol/requests"
 	"github.com/status-im/status-go/services/wakuv2ext"
-	"github.com/status-im/status-go/telemetry"
 
 	"github.com/urfave/cli/v2"
 )
@@ -74,14 +73,12 @@ func start(p StartParams, logger *zap.SugaredLogger) (*StatusCLI, error) {
 	}
 
 	if p.TelemetryURL != "" {
-		telemetryLogger, err := getLogger(true)
+		waku := backend.StatusNode().WakuV2Service()
+		telemetryClient, err := wakumetrics.NewClient(wakumetrics.WithPeerID(waku.PeerID().String()))
 		if err != nil {
 			return nil, err
 		}
-		waku := backend.StatusNode().WakuV2Service()
-		telemetryClient := telemetry.NewClient(telemetryLogger, p.TelemetryURL, backend.SelectedAccountKeyID(), p.Name, "cli", telemetry.WithPeerID(waku.PeerID().String()))
-		telemetryClient.Start(context.Background())
-		backend.StatusNode().WakuV2Service().SetStatusTelemetryClient(telemetryClient)
+		backend.StatusNode().WakuV2Service().SetMetricsHandler(telemetryClient)
 	}
 	wakuAPI := wakuv2ext.NewPublicAPI(wakuService)
 
@@ -152,7 +149,8 @@ func createAccountAndLogin(b *api.GethStatusBackend, rootDataDir, password strin
 			HTTPHost:    "127.0.0.1",
 			HTTPPort:    p.Port,
 		},
-		TelemetryServerURL: p.TelemetryURL,
+		TelemetryServerURL:                     p.TelemetryURL,
+		WakuV2EnableMissingMessageVerification: true,
 	}
 	return b.CreateAccountAndLogin(req,
 		params.WithFleet(p.Fleet),

cmd/statusd/main.go

@@ -11,6 +11,7 @@ import (
 	"os/signal"
 	"path/filepath"
 	"runtime"
+	"strconv"
 	"strings"
 	"time"
 
@@ -313,7 +314,7 @@ func main() {
 		if *metricsEnabled || gethmetrics.Enabled {
 			go startCollectingNodeMetrics(interruptCh, backend.StatusNode())
 			go gethmetrics.CollectProcessMetrics(3 * time.Second)
-			go metrics.NewMetricsServer(*metricsPort, gethmetrics.DefaultRegistry).Listen()
+			go metrics.NewMetricsServer("localhost:"+strconv.Itoa(*metricsPort), gethmetrics.DefaultRegistry).Listen()
 		}
 
 		// Check if profiling shall be enabled.

metrics/metrics.go

@@ -1,7 +1,6 @@
 package metrics
 
 import (
-	"fmt"
 	"net/http"
 	"time"
 
@@ -22,13 +21,13 @@ type Server struct {
 	server *http.Server
 }
 
-func NewMetricsServer(port int, r metrics.Registry) *Server {
+func NewMetricsServer(address string, r metrics.Registry) *Server {
 	mux := http.NewServeMux()
 	mux.Handle("/health", healthHandler())
 	mux.Handle("/metrics", Handler(r))
 	p := Server{
 		server: &http.Server{
-			Addr:              fmt.Sprintf(":%d", port),
+			Addr:              address,
 			ReadHeaderTimeout: 5 * time.Second,
 			Handler:           mux,
 		},
@@ -48,9 +47,13 @@ func healthHandler() http.Handler {
 func Handler(reg metrics.Registry) http.Handler {
 	// we disable compression because geth doesn't support it
 	opts := promhttp.HandlerOpts{DisableCompression: true}
-	// we are combining handlers to avoid having 2 endpoints
-	statusMetrics := promhttp.HandlerFor(prom.DefaultGatherer, opts) // our metrics
-	gethMetrics := gethprom.Handler(reg)                             // geth metrics
+	// we are using only our own metrics
+	statusMetrics := promhttp.HandlerFor(prom.DefaultGatherer, opts)
+	if reg == nil {
+		return statusMetrics
+	}
+	// if registry is provided, combine handlers
+	gethMetrics := gethprom.Handler(reg)
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		statusMetrics.ServeHTTP(w, r)
 		gethMetrics.ServeHTTP(w, r)
@@ -62,3 +65,11 @@ func (p *Server) Listen() {
 	defer common.LogOnPanic()
 	logutils.ZapLogger().Info("metrics server stopped", zap.Error(p.server.ListenAndServe()))
 }
+
+// Stop gracefully shuts down the metrics server
+func (p *Server) Stop() error {
+	if p.server != nil {
+		return p.server.Close()
+	}
+	return nil
+}

metrics/wakumetrics/client.go

@@ -0,0 +1,183 @@
+package wakumetrics
+
+import (
+	"fmt"
+	"strconv"
+
+	"github.com/status-im/status-go/protocol/transport"
+	wakutypes "github.com/status-im/status-go/waku/types"
+	"github.com/status-im/status-go/wakuv2"
+
+	v1protocol "github.com/status-im/status-go/protocol/v1"
+	v2common "github.com/status-im/status-go/wakuv2/common"
+	wps "github.com/waku-org/go-waku/waku/v2/peerstore"
+	v2protocol "github.com/waku-org/go-waku/waku/v2/protocol"
+)
+
+type ReceivedMessages struct {
+	Filter     transport.Filter
+	SSHMessage *wakutypes.Message
+	Messages   []*v1protocol.StatusMessage
+}
+
+type Client struct {
+	peerId               string
+	deviceType           string
+	version              string
+	lastPeerConnFailures map[string]int
+}
+
+type TelemetryClientOption func(*Client)
+
+func WithPeerID(peerId string) TelemetryClientOption {
+	return func(c *Client) {
+		c.peerId = peerId
+		nodePeerId.WithLabelValues(peerId).Set(1)
+	}
+}
+
+func WithDeviceType(deviceType string) TelemetryClientOption {
+	return func(c *Client) {
+		c.deviceType = deviceType
+	}
+}
+
+func WithVersion(version string) TelemetryClientOption {
+	return func(c *Client) {
+		c.version = version
+	}
+}
+
+func NewClient(opts ...TelemetryClientOption) (*Client, error) {
+	client := &Client{
+		lastPeerConnFailures: make(map[string]int),
+	}
+
+	for _, opt := range opts {
+		opt(client)
+	}
+
+	return client, nil
+}
+
+// RegisterWithRegistry registers all metrics with the provided registry
+func (c *Client) RegisterWithRegistry() error {
+	if err := RegisterMetrics(); err != nil {
+		return fmt.Errorf("failed to register metrics: %v", err)
+	}
+	return nil
+}
+
+func (c *Client) SetDeviceType(deviceType string) {
+	c.deviceType = deviceType
+}
+
+func (c *Client) PushReceivedMessages(receivedMessages ReceivedMessages) {
+	messagesReceivedTotal.WithLabelValues(
+		receivedMessages.Filter.PubsubTopic,
+		receivedMessages.Filter.ContentTopic.String(),
+		receivedMessages.Filter.ChatID,
+	).Add(float64(len(receivedMessages.Messages)))
+}
+
+func (c *Client) PushSentEnvelope(sentEnvelope wakuv2.SentEnvelope) {
+	EnvelopeSentTotal.WithLabelValues(
+		sentEnvelope.Envelope.PubsubTopic(),
+		sentEnvelope.Envelope.Message().ContentTopic,
+		sentEnvelope.PublishMethod.String(),
+	).Inc()
+}
+
+func (c *Client) PushErrorSendingEnvelope(errorSendingEnvelope wakuv2.ErrorSendingEnvelope) {
+	envelopeSentErrors.WithLabelValues(
+		errorSendingEnvelope.SentEnvelope.Envelope.PubsubTopic(),
+		errorSendingEnvelope.SentEnvelope.Envelope.Message().ContentTopic,
+	).Inc()
+}
+
+func (c *Client) PushPeerCount(peerCount int) {
+	connectedPeers.Set(float64(peerCount))
+}
+
+func (c *Client) PushPeerConnFailures(peerConnFailures map[string]int) {
+	for peerID, failures := range peerConnFailures {
+		if lastFailures, exists := c.lastPeerConnFailures[peerID]; exists {
+			if failures == lastFailures {
+				continue
+			}
+		}
+		c.lastPeerConnFailures[peerID] = failures
+		peerConnectionFailures.Add(float64(failures))
+	}
+}
+
+func (c *Client) PushMessageCheckSuccess() {
+	storeQuerySuccesses.Inc()
+}
+
+func (c *Client) PushMessageCheckFailure() {
+	storeQueryFailures.Inc()
+}
+
+func (c *Client) PushPeerCountByShard(peerCountByShard map[uint16]uint) {
+	for shard, count := range peerCountByShard {
+		peersByShard.WithLabelValues(strconv.FormatUint(uint64(shard), 10)).Set(float64(count))
+	}
+}
+
+func (c *Client) PushPeerCountByOrigin(peerCountByOrigin map[wps.Origin]uint) {
+	for origin, count := range peerCountByOrigin {
+		peersByOrigin.WithLabelValues(getOriginString(origin)).Set(float64(count))
+	}
+}
+
+func (c *Client) PushDialFailure(dialFailure v2common.DialError) {
+	peerDialFailures.WithLabelValues(
+		dialFailure.ErrType.String(),
+		dialFailure.Protocols,
+	).Inc()
+}
+
+func (c *Client) PushMissedMessage(envelope *v2protocol.Envelope) {
+	missedMessages.WithLabelValues(
+		envelope.PubsubTopic(),
+		envelope.Message().ContentTopic,
+	).Inc()
+}
+
+func (c *Client) PushMissedRelevantMessage(receivedMessage *v2common.ReceivedMessage) {
+	missedMessages.WithLabelValues(
+		receivedMessage.PubsubTopic,
+		receivedMessage.ContentTopic.String(),
+	).Inc()
+}
+
+func (c *Client) PushMessageDeliveryConfirmed() {
+	messageDeliveryConfirmations.Inc()
+}
+
+func (c *Client) PushSentMessageTotal(messageSize uint32, publishMethod string) {
+	wakuMessagesSizeBytes.WithLabelValues(publishMethod).Add(float64(messageSize))
+	messagesSentTotal.WithLabelValues(publishMethod).Inc()
+}
+
+func getOriginString(origin wps.Origin) string {
+	switch origin {
+	case wps.Unknown:
+		return "unknown"
+	case wps.Discv5:
+		return "discv5"
+	case wps.Static:
+		return "static"
+	case wps.PeerExchange:
+		return "peer_exchange"
+	case wps.DNSDiscovery:
+		return "dns_discovery"
+	case wps.Rendezvous:
+		return "rendezvous"
+	case wps.PeerManager:
+		return "peer_manager"
+	default:
+		return "unknown"
+	}
+}