multi: Remove high card keys for payment metrics

CHANGELOG_PENDING.md

@@ -2,12 +2,17 @@
 
 ## vX.X
 
+### Breaking Changes 🚨🚨
+
+- Payment/ticket metrics are no longer recorded with high cardinality keys (i.e. recipient, manifestID) which means those labels will no longer be available when using a monitoring system such as Prometheus
+
 ### Features ⚒
 
 #### General
 
 - \#1848 Use fee cut instead of fee share for user facing language in the CLI (@kyriediculous)
 - \#1854 Allow to pass region in the custom s3 storage URL (@darkdarkdragon)
+- \#1893 Remove high cardinality keys from payment metrics (@yondonfu)
 
 #### Broadcaster
 

core/orchestrator.go

@@ -183,7 +183,7 @@ func (orch *orchestrator) ProcessPayment(payment net.Payment, manifestID Manifes
 			glog.Errorf("Error receiving ticket sessionID=%v recipientRandHash=%x senderNonce=%v: %v", manifestID, ticket.RecipientRandHash, ticket.SenderNonce, err)
 
 			if monitor.Enabled {
-				monitor.PaymentRecvError(sender.String(), string(manifestID), err.Error())
+				monitor.PaymentRecvError(err.Error())
 			}
 			if _, ok := err.(*pm.FatalReceiveErr); ok {
 				return err
@@ -213,12 +213,9 @@ func (orch *orchestrator) ProcessPayment(payment net.Payment, manifestID Manifes
 	}
 
 	if monitor.Enabled {
-		senderStr := sender.String()
-		mid := string(manifestID)
-
-		monitor.TicketValueRecv(senderStr, mid, totalEV)
-		monitor.TicketsRecv(senderStr, mid, totalTickets)
-		monitor.WinningTicketsRecv(senderStr, totalWinningTickets)
+		monitor.TicketValueRecv(totalEV)
+		monitor.TicketsRecv(totalTickets)
+		monitor.WinningTicketsRecv(totalWinningTickets)
 	}
 
 	if receiveErr != nil {

monitor/census.go

@@ -606,21 +606,21 @@ func InitCensus(nodeType NodeType, version string) {
 			Name:        "ticket_value_sent",
 			Measure:     census.mTicketValueSent,
 			Description: "Ticket value sent",
-			TagKeys:     append([]tag.Key{census.kRecipient, census.kManifestID}, baseTags...),
+			TagKeys:     baseTags,
 			Aggregation: view.Sum(),
 		},
 		{
 			Name:        "tickets_sent",
 			Measure:     census.mTicketsSent,
 			Description: "Tickets sent",
-			TagKeys:     append([]tag.Key{census.kRecipient, census.kManifestID}, baseTags...),
+			TagKeys:     baseTags,
 			Aggregation: view.Sum(),
 		},
 		{
 			Name:        "payment_create_errors",
 			Measure:     census.mPaymentCreateError,
 			Description: "Errors when creating payments",
-			TagKeys:     append([]tag.Key{census.kRecipient, census.kManifestID}, baseTags...),
+			TagKeys:     baseTags,
 			Aggregation: view.Sum(),
 		},
 		{
@@ -643,21 +643,21 @@ func InitCensus(nodeType NodeType, version string) {
 			Name:        "ticket_value_recv",
 			Measure:     census.mTicketValueRecv,
 			Description: "Ticket value received",
-			TagKeys:     append([]tag.Key{census.kSender, census.kManifestID}, baseTags...),
+			TagKeys:     baseTags,
 			Aggregation: view.Sum(),
 		},
 		{
 			Name:        "tickets_recv",
 			Measure:     census.mTicketsRecv,
 			Description: "Tickets received",
-			TagKeys:     append([]tag.Key{census.kSender, census.kManifestID}, baseTags...),
+			TagKeys:     baseTags,
 			Aggregation: view.Sum(),
 		},
 		{
 			Name:        "payment_recv_errors",
 			Measure:     census.mPaymentRecvErr,
 			Description: "Errors when receiving payments",
-			TagKeys:     append([]tag.Key{census.kSender, census.kManifestID, census.kErrorCode}, baseTags...),
+			TagKeys:     append([]tag.Key{census.kErrorCode}, baseTags...),
 			Aggregation: view.Sum(),
 		},
 		{
@@ -678,7 +678,7 @@ func InitCensus(nodeType NodeType, version string) {
 			Name:        "ticket_redemption_errors",
 			Measure:     census.mTicketRedemptionError,
 			Description: "Errors when redeeming tickets",
-			TagKeys:     append([]tag.Key{census.kSender}, baseTags...),
+			TagKeys:     baseTags,
 			Aggregation: view.Sum(),
 		},
 		{
@@ -1281,51 +1281,36 @@ func (cen *censusMetricsCounter) streamEnded(nonce uint64) {
 	census.sendSuccess()
 }
 
-// TicketValueSent records the ticket value sent to a recipient for a manifestID
-func TicketValueSent(recipient string, manifestID string, value *big.Rat) {
+// TicketValueSent records the ticket value sent
+func TicketValueSent(value *big.Rat) {
 	census.lock.Lock()
 	defer census.lock.Unlock()
 
 	if value.Cmp(big.NewRat(0, 1)) <= 0 {
 		return
 	}
 
-	ctx, err := tag.New(census.ctx, tag.Insert(census.kRecipient, recipient), tag.Insert(census.kManifestID, manifestID))
-	if err != nil {
-		glog.Fatal(err)
-	}
-
-	stats.Record(ctx, census.mTicketValueSent.M(fracwei2gwei(value)))
+	stats.Record(census.ctx, census.mTicketValueSent.M(fracwei2gwei(value)))
 }
 
-// TicketsSent records the number of tickets sent to a recipient for a manifestID
-func TicketsSent(recipient string, manifestID string, numTickets int) {
+// TicketsSent records the number of tickets sent
+func TicketsSent(numTickets int) {
 	census.lock.Lock()
 	defer census.lock.Unlock()
 
 	if numTickets <= 0 {
 		return
 	}
 
-	ctx, err := tag.New(census.ctx, tag.Insert(census.kRecipient, recipient), tag.Insert(census.kManifestID, manifestID))
-	if err != nil {
-		glog.Fatal(err)
-	}
-
-	stats.Record(ctx, census.mTicketsSent.M(int64(numTickets)))
+	stats.Record(census.ctx, census.mTicketsSent.M(int64(numTickets)))
 }
 
 // PaymentCreateError records a error from payment creation
-func PaymentCreateError(recipient string, manifestID string) {
+func PaymentCreateError() {
 	census.lock.Lock()
 	defer census.lock.Unlock()
 
-	ctx, err := tag.New(census.ctx, tag.Insert(census.kRecipient, recipient), tag.Insert(census.kManifestID, manifestID))
-	if err != nil {
-		glog.Fatal(err)
-	}
-
-	stats.Record(ctx, census.mPaymentCreateError.M(1))
+	stats.Record(census.ctx, census.mPaymentCreateError.M(1))
 }
 
 // Deposit records the current deposit for the broadcaster
@@ -1338,41 +1323,31 @@ func Reserve(sender string, reserve *big.Int) {
 }
 
 // TicketValueRecv records the ticket value received from a sender for a manifestID
-func TicketValueRecv(sender string, manifestID string, value *big.Rat) {
+func TicketValueRecv(value *big.Rat) {
 	census.lock.Lock()
 	defer census.lock.Unlock()
 
 	if value.Cmp(big.NewRat(0, 1)) <= 0 {
 		return
 	}
 
-	ctx, err := tag.New(census.ctx, tag.Insert(census.kSender, sender), tag.Insert(census.kManifestID, manifestID))
-	if err != nil {
-		glog.Fatal(err)
-	}
-
-	stats.Record(ctx, census.mTicketValueRecv.M(fracwei2gwei(value)))
+	stats.Record(census.ctx, census.mTicketValueRecv.M(fracwei2gwei(value)))
 }
 
 // TicketsRecv records the number of tickets received from a sender for a manifestID
-func TicketsRecv(sender string, manifestID string, numTickets int) {
+func TicketsRecv(numTickets int) {
 	census.lock.Lock()
 	defer census.lock.Unlock()
 
 	if numTickets <= 0 {
 		return
 	}
 
-	ctx, err := tag.New(census.ctx, tag.Insert(census.kSender, sender), tag.Insert(census.kManifestID, manifestID))
-	if err != nil {
-		glog.Fatal(err)
-	}
-
-	stats.Record(ctx, census.mTicketsRecv.M(int64(numTickets)))
+	stats.Record(census.ctx, census.mTicketsRecv.M(int64(numTickets)))
 }
 
 // PaymentRecvError records an error from receiving a payment
-func PaymentRecvError(sender string, manifestID string, errStr string) {
+func PaymentRecvError(errStr string) {
 	census.lock.Lock()
 	defer census.lock.Unlock()
 
@@ -1391,8 +1366,6 @@ func PaymentRecvError(sender string, manifestID string, errStr string) {
 
 	ctx, err := tag.New(
 		census.ctx,
-		tag.Insert(census.kSender, sender),
-		tag.Insert(census.kManifestID, manifestID),
 		tag.Insert(census.kErrorCode, errCode),
 	)
 	if err != nil {
@@ -1402,51 +1375,36 @@ func PaymentRecvError(sender string, manifestID string, errStr string) {
 	stats.Record(ctx, census.mPaymentRecvErr.M(1))
 }
 
-// WinningTicketsRecv records the number of winning tickets received from a sender
-func WinningTicketsRecv(sender string, numTickets int) {
+// WinningTicketsRecv records the number of winning tickets received
+func WinningTicketsRecv(numTickets int) {
 	census.lock.Lock()
 	defer census.lock.Unlock()
 
 	if numTickets <= 0 {
 		return
 	}
 
-	ctx, err := tag.New(census.ctx, tag.Insert(census.kSender, sender))
-	if err != nil {
-		glog.Fatal(err)
-	}
-
-	stats.Record(ctx, census.mWinningTicketsRecv.M(int64(numTickets)))
+	stats.Record(census.ctx, census.mWinningTicketsRecv.M(int64(numTickets)))
 }
 
-// ValueRedeemed records the value from redeeming winning tickets from a sender
-func ValueRedeemed(sender string, value *big.Int) {
+// ValueRedeemed records the value from redeeming winning tickets
+func ValueRedeemed(value *big.Int) {
 	census.lock.Lock()
 	defer census.lock.Unlock()
 
 	if value.Cmp(big.NewInt(0)) <= 0 {
 		return
 	}
 
-	ctx, err := tag.New(census.ctx, tag.Insert(census.kSender, sender))
-	if err != nil {
-		glog.Fatal(err)
-	}
-
-	stats.Record(ctx, census.mValueRedeemed.M(wei2gwei(value)))
+	stats.Record(census.ctx, census.mValueRedeemed.M(wei2gwei(value)))
 }
 
 // TicketRedemptionError records an error from redeeming a ticket
-func TicketRedemptionError(sender string) {
+func TicketRedemptionError() {
 	census.lock.Lock()
 	defer census.lock.Unlock()
 
-	ctx, err := tag.New(census.ctx, tag.Insert(census.kSender, sender))
-	if err != nil {
-		glog.Fatal(err)
-	}
-
-	stats.Record(ctx, census.mTicketRedemptionError.M(1))
+	stats.Record(census.ctx, census.mTicketRedemptionError.M(1))
 }
 
 // SuggestedGasPrice records the last suggested gas price

pm/sendermonitor.go

@@ -358,13 +358,13 @@ func (sm *LocalSenderMonitor) redeemWinningTicket(ticket *SignedTicket) (*types.
 	used, err := sm.broker.IsUsedTicket(ticket.Ticket)
 	if err != nil {
 		if monitor.Enabled {
-			monitor.TicketRedemptionError(ticket.Ticket.Sender.String())
+			monitor.TicketRedemptionError()
 		}
 		return nil, err
 	}
 	if used {
 		if monitor.Enabled {
-			monitor.TicketRedemptionError(ticket.Ticket.Sender.String())
+			monitor.TicketRedemptionError()
 		}
 		return nil, errIsUsedTicket
 	}
@@ -412,23 +412,23 @@ func (sm *LocalSenderMonitor) redeemWinningTicket(ticket *SignedTicket) (*types.
 	tx, err := sm.broker.RedeemWinningTicket(ticket.Ticket, ticket.Sig, ticket.RecipientRand)
 	if err != nil {
 		if monitor.Enabled {
-			monitor.TicketRedemptionError(ticket.Ticket.Sender.String())
+			monitor.TicketRedemptionError()
 		}
 		return nil, err
 	}
 
 	// Wait for transaction to confirm
 	if err := sm.broker.CheckTx(tx); err != nil {
 		if monitor.Enabled {
-			monitor.TicketRedemptionError(ticket.Ticket.Sender.String())
+			monitor.TicketRedemptionError()
 		}
 		return nil, err
 	}
 
 	if monitor.Enabled {
 		// TODO(yondonfu): Handle case where < ticket.FaceValue is actually
 		// redeemed i.e. if sender reserve cannot cover the full ticket.FaceValue
-		monitor.ValueRedeemed(ticket.Ticket.Sender.String(), ticket.Ticket.FaceValue)
+		monitor.ValueRedeemed(ticket.Ticket.FaceValue)
 	}
 
 	return tx, nil

server/segment_rpc.go

@@ -387,9 +387,8 @@ func SubmitSegment(sess *BroadcastSession, seg *stream.HLSSegment, nonce uint64)
 	if err != nil {
 		glog.Errorf("Could not create payment nonce=%d manifestID=%s sessionID=%s seqNo=%d bytes=%v err=%v", nonce, sess.Params.ManifestID, sess.OrchestratorInfo.AuthToken.SessionId, seg.SeqNo, len(data), err)
 
-		if monitor.Enabled && sess.OrchestratorInfo.TicketParams != nil {
-			recipient := ethcommon.BytesToAddress(sess.OrchestratorInfo.TicketParams.Recipient).String()
-			monitor.PaymentCreateError(recipient, string(params.ManifestID))
+		if monitor.Enabled {
+			monitor.PaymentCreateError()
 		}
 
 		return nil, err
@@ -440,12 +439,9 @@ func SubmitSegment(sess *BroadcastSession, seg *stream.HLSSegment, nonce uint64)
 	// If the segment was submitted then we assume that any payment included was
 	// submitted as well so we consider the update's credit as spent
 	balUpdate.Status = CreditSpent
-	if monitor.Enabled && sess.OrchestratorInfo.TicketParams != nil {
-		recipient := ethcommon.BytesToAddress(sess.OrchestratorInfo.TicketParams.Recipient).String()
-		mid := string(params.ManifestID)
-
-		monitor.TicketValueSent(recipient, mid, balUpdate.NewCredit)
-		monitor.TicketsSent(recipient, mid, balUpdate.NumTickets)
+	if monitor.Enabled {
+		monitor.TicketValueSent(balUpdate.NewCredit)
+		monitor.TicketsSent(balUpdate.NumTickets)
 	}
 
 	if resp.StatusCode != 200 {