Merge pull request #64 from status-im/feat/message-reliability

Add remaining metrics for message reliability

cmd/server/main.go

@@ -65,8 +65,11 @@ func main() {
 	server.RegisterMetric(types.SentEnvelopeMetric, &metrics.SentEnvelope{})
 	server.RegisterMetric(types.PeerCountByShardMetric, &metrics.PeerCountByShard{})
 	server.RegisterMetric(types.PeerCountByOriginMetric, &metrics.PeerCountByOrigin{})
-
+	server.RegisterMetric(types.MissingMessageMetric, &metrics.MissingMessage{})
+	server.RegisterMetric(types.MissingRelevantMessageMetric, &metrics.MissingRelevantMessage{})
+	server.RegisterMetric(types.MessageDeliveryConfirmedMetric, &metrics.MessageDeliveryConfirmed{})
 	server.RegisterMetric(types.MessageCheckSuccessMetric, &metrics.MessageCheckSuccess{})
 	server.RegisterMetric(types.MessageCheckFailureMetric, &metrics.MessageCheckFailure{})
+	server.RegisterMetric(types.DialFailureMetric, &metrics.DialFailure{})
 	server.Start(*port)
 }

lib/common/test_utils.go

@@ -42,6 +42,10 @@ func DropTables(db *sql.DB) {
 		"peerCountByOrigin",
 		"messageCheckSuccess",
 		"messageCheckFailure",
+		"dialFailure",
+		"missingmessages",
+		"missingrelevantmessages",
+		"messageDeliveryConfirmed",
 		"schema_migrations",
 	}
 
@@ -98,14 +102,34 @@ func DropTables(db *sql.DB) {
 		log.Fatalf("an error '%s' was not expected when dropping the index", err)
 	}
 
-	_, err = tx.Exec("DROP TABLE IF EXISTS messageCheckSuccess")
+	_, err = tx.Exec("DROP INDEX IF EXISTS missingMessages_unique")
 	if err != nil {
-		log.Fatalf("an error '%s' was not expected when dropping the table", err)
+		log.Fatalf("an error '%s' was not expected when dropping the index", err)
 	}
 
-	_, err = tx.Exec("DROP TABLE IF EXISTS messageCheckFailure")
+	_, err = tx.Exec("DROP INDEX IF EXISTS messageCheckSuccess_unique")
 	if err != nil {
-		log.Fatalf("an error '%s' was not expected when dropping the table", err)
+		log.Fatalf("an error '%s' was not expected when dropping the index", err)
+	}
+
+	_, err = tx.Exec("DROP INDEX IF EXISTS messageCheckFailure_unique")
+	if err != nil {
+		log.Fatalf("an error '%s' was not expected when dropping the index", err)
+	}
+
+	_, err = tx.Exec("DROP INDEX IF EXISTS missingRelevantMessages_unique")
+	if err != nil {
+		log.Fatalf("an error '%s' was not expected when dropping the index", err)
+	}
+
+	_, err = tx.Exec("DROP INDEX IF EXISTS messageDeliveryConfirmed_unique")
+	if err != nil {
+		log.Fatalf("an error '%s' was not expected when dropping the index", err)
+	}
+
+	_, err = tx.Exec("DROP INDEX IF EXISTS dialFailure_unique")
+	if err != nil {
+		log.Fatalf("an error '%s' was not expected when dropping the index", err)
 	}
 
 	_, err = tx.Exec("DROP TABLE IF EXISTS schema_migrations")

lib/database/sql/000020_dial_failure.up.sql

@@ -0,0 +1,34 @@
+CREATE TABLE IF NOT EXISTS dial_error_types (
+    id INTEGER PRIMARY KEY,
+    name TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS dialFailure (
+    id SERIAL PRIMARY KEY,
+    recordId INTEGER NOT NULL,
+    errorType INTEGER NOT NULL,
+    errorMsg TEXT,
+    protocols TEXT NOT NULL,
+    timestamp INTEGER NOT NULL,
+    CONSTRAINT dialFailure_unique UNIQUE (recordId, errorType, protocols, timestamp),
+    CONSTRAINT fk_dialFailure_errorType FOREIGN KEY (errorType) REFERENCES dial_error_types(id)
+);
+
+ALTER TABLE dialFailure ADD CONSTRAINT fk_dialFailure_telemetryRecord
+            FOREIGN KEY (recordId) REFERENCES telemetryRecord(id);
+
+INSERT INTO dial_error_types (id, name)
+SELECT v.id, v.name
+FROM (VALUES
+    (0, 'Unknown'),
+    (1, 'I/O Timeout'),
+    (2, 'Connection Refused'),
+    (3, 'Relay Circuit Failed'),
+    (4, 'Relay No Reservation'),
+    (5, 'Security Negotiation Failed'),
+    (6, 'Concurrent Dial Succeeded'),
+    (7, 'Concurrent Dial Failed')
+) AS v(id, name)
+WHERE NOT EXISTS (
+    SELECT 1 FROM dial_error_types WHERE id = v.id
+);

lib/database/sql/000021_missing_messages.up.sql

@@ -0,0 +1,39 @@
+CREATE TABLE IF NOT EXISTS missingMessages (
+    id SERIAL PRIMARY KEY,
+    messageHash TEXT NOT NULL,
+    sentAt INTEGER NOT NULL,
+    contentTopic TEXT NOT NULL,
+    pubsubTopic TEXT NOT NULL,  
+    recordId INTEGER NOT NULL
+);
+
+ALTER TABLE missingMessages ADD CONSTRAINT fk_missingMessages_telemetryRecord
+            FOREIGN KEY (recordId) REFERENCES telemetryRecord(id);
+
+ALTER TABLE missingMessages 
+ADD CONSTRAINT missingMessages_unique 
+UNIQUE (
+    recordId,
+    messageHash,
+    contentTopic
+);
+
+CREATE TABLE IF NOT EXISTS missingRelevantMessages (
+    id SERIAL PRIMARY KEY,
+    messageHash TEXT NOT NULL,
+    sentAt INTEGER NOT NULL,
+    contentTopic TEXT NOT NULL,
+    pubsubTopic TEXT NOT NULL,  
+    recordId INTEGER NOT NULL
+);
+
+ALTER TABLE missingRelevantMessages ADD CONSTRAINT fk_missingRelevantMessages_telemetryRecord
+            FOREIGN KEY (recordId) REFERENCES telemetryRecord(id);
+
+ALTER TABLE missingRelevantMessages 
+ADD CONSTRAINT missingRelevantMessages_unique 
+UNIQUE (
+    recordId,
+    messageHash,
+    contentTopic
+);

lib/database/sql/000022_message_delivery_confirmed.up.sql

@@ -0,0 +1,10 @@
+CREATE TABLE IF NOT EXISTS messageDeliveryConfirmed (
+    id SERIAL PRIMARY KEY,
+    recordId INTEGER NOT NULL,
+    messageHash TEXT NOT NULL,
+    timestamp INTEGER NOT NULL,
+    CONSTRAINT messageDeliveryConfirmed_unique UNIQUE (recordId, messageHash, timestamp)
+);
+
+ALTER TABLE messageDeliveryConfirmed ADD CONSTRAINT fk_messageDeliveryConfirmed_telemetryRecord
+            FOREIGN KEY (recordId) REFERENCES telemetryRecord(id);

lib/metrics/dial_failure.go

@@ -0,0 +1,13 @@
+package metrics
+
+import (
+	"github.com/status-im/telemetry/pkg/types"
+)
+
+type DialFailure struct {
+	GenericMetric[types.DialFailure]
+}
+
+var (
+	DialFailureProcessor = &DialFailure{}
+)

lib/metrics/missing_messages.go

@@ -0,0 +1,23 @@
+package metrics
+
+import (
+	"github.com/status-im/telemetry/pkg/types"
+)
+
+type MissingMessage struct {
+	GenericMetric[types.MissingMessage]
+}
+
+type MissingRelevantMessage struct {
+	GenericMetric[types.MissingRelevantMessage]
+}
+
+type MessageDeliveryConfirmed struct {
+	GenericMetric[types.MessageDeliveryConfirmed]
+}
+
+var (
+	MissingMessageProcessor           = &MissingMessage{}
+	MissingRelevantMessageProcessor   = &MissingRelevantMessage{}
+	MessageDeliveryConfirmedProcessor = &MessageDeliveryConfirmed{}
+)

pkg/types/types.go

@@ -5,18 +5,23 @@ import "encoding/json"
 type TelemetryType string
 
 const (
-	ProtocolStatsMetric        TelemetryType = "ProtocolStats"
-	ReceivedEnvelopeMetric     TelemetryType = "ReceivedEnvelope"
-	SentEnvelopeMetric         TelemetryType = "SentEnvelope"
-	UpdateEnvelopeMetric       TelemetryType = "UpdateEnvelope"
-	ReceivedMessagesMetric     TelemetryType = "ReceivedMessages"
-	ErrorSendingEnvelopeMetric TelemetryType = "ErrorSendingEnvelope"
-	PeerCountMetric            TelemetryType = "PeerCount"
-	PeerConnFailureMetric      TelemetryType = "PeerConnFailure"
-	PeerCountByShardMetric     TelemetryType = "PeerCountByShard"
-	PeerCountByOriginMetric    TelemetryType = "PeerCountByOrigin"
-	MessageCheckSuccessMetric  TelemetryType = "MessageCheckSuccess"
-	MessageCheckFailureMetric  TelemetryType = "MessageCheckFailure"
+	ProtocolStatsMetric            TelemetryType = "ProtocolStats"
+	ReceivedEnvelopeMetric         TelemetryType = "ReceivedEnvelope"
+	SentEnvelopeMetric             TelemetryType = "SentEnvelope"
+	UpdateEnvelopeMetric           TelemetryType = "UpdateEnvelope"
+	ReceivedMessagesMetric         TelemetryType = "ReceivedMessages"
+	ErrorSendingEnvelopeMetric     TelemetryType = "ErrorSendingEnvelope"
+	PeerCountMetric                TelemetryType = "PeerCount"
+	PeerConnFailureMetric          TelemetryType = "PeerConnFailure"
+	PeerCountByShardMetric         TelemetryType = "PeerCountByShard"
+	PeerCountByOriginMetric        TelemetryType = "PeerCountByOrigin"
+	MessageCheckSuccessMetric      TelemetryType = "MessageCheckSuccess"
+	MessageCheckFailureMetric      TelemetryType = "MessageCheckFailure"
+	DialFailureMetric              TelemetryType = "DialFailure"
+	StoreConfrimationErrorMetric   TelemetryType = "StoreConfrimationError"
+	MissingMessageMetric           TelemetryType = "MissedMessage"
+	MissingRelevantMessageMetric   TelemetryType = "MissedRelevantMessage"
+	MessageDeliveryConfirmedMetric TelemetryType = "MessageDeliveryConfirmed"
 )
 
 type Origin int64
@@ -31,6 +36,19 @@ const (
 	PeerManager
 )
 
+type DialErrorType int
+
+const (
+	ErrorUnknown DialErrorType = iota
+	ErrorIOTimeout
+	ErrorConnectionRefused
+	ErrorRelayCircuitFailed
+	ErrorRelayNoReservation
+	ErrorSecurityNegotiationFailed
+	ErrorConcurrentDialSucceeded
+	ErrorConcurrentDialFailed
+)
+
 type TelemetryRequest struct {
 	ID            int              `json:"id"`
 	TelemetryType TelemetryType    `json:"telemetry_type"`
@@ -147,3 +165,35 @@ type MessageCheckFailure struct {
 	MessageHash string `json:"messageHash"`
 	Timestamp   int64  `json:"timestamp"`
 }
+type DialFailure struct {
+	TelemetryRecord
+	ErrorType DialErrorType `json:"errorType"`
+	ErrorMsg  string        `json:"errorMsg"`
+	Protocols string        `json:"protocols"`
+	Timestamp int64         `json:"timestamp"`
+}
+
+type MissingMessage struct {
+	TelemetryRecord
+	ID           int    `json:"id"`
+	ContentTopic string `json:"contentTopic"`
+	MessageHash  string `json:"messageHash"`
+	SentAt       int64  `json:"sentAt"`
+	PubsubTopic  string `json:"pubsubTopic"`
+}
+
+type MissingRelevantMessage struct {
+	TelemetryRecord
+	ID           int    `json:"id"`
+	ContentTopic string `json:"contentTopic"`
+	MessageHash  string `json:"messageHash"`
+	SentAt       int64  `json:"sentAt"`
+	PubsubTopic  string `json:"pubsubTopic"`
+}
+
+type MessageDeliveryConfirmed struct {
+	TelemetryRecord
+	ID          int    `json:"id"`
+	MessageHash string `json:"messageHash"`
+	Timestamp   int64  `json:"timestamp"`
+}