get matrics for fast verification per orchestrator (livepeer#2397)

We should be able to determine the fast verification success/failure rate for each O
ad-astra-video · May 25, 2022 · 48122e5 · 48122e5
1 parent 0580610
commit 48122e5
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 12 deletions.
diff --git a/monitor/census.go b/monitor/census.go
@@ -755,14 +755,14 @@ func InitCensus(nodeType NodeType, version string) {
 			Name:        "fast_verification_done",
 			Measure:     census.mFastVerificationDone,
 			Description: "Number of fast verifications done",
-			TagKeys:     baseTagsWithManifestID,
+			TagKeys:     append([]tag.Key{census.kOrchestratorURI}, baseTagsWithManifestID...),
 			Aggregation: view.Count(),
 		},
 		{
 			Name:        "fast_verification_failed",
 			Measure:     census.mFastVerificationFailed,
 			Description: "Number of fast verifications failed",
-			TagKeys:     baseTagsWithManifestID,
+			TagKeys:     append([]tag.Key{census.kOrchestratorURI}, baseTagsWithManifestID...),
 			Aggregation: view.Count(),
 		},
 		{
@@ -1617,16 +1617,18 @@ func fracwei2gwei(wei *big.Rat) float64 {
 	return floatWei / gweiConversionFactor
 }
 
-func FastVerificationDone(ctx context.Context) {
+func FastVerificationDone(ctx context.Context, uri string) {
 	if err := stats.RecordWithTags(census.ctx,
-		manifestIDTag(ctx), census.mFastVerificationDone.M(1)); err != nil {
+		manifestIDTag(ctx, tag.Insert(census.kOrchestratorURI, uri)),
+		census.mFastVerificationDone.M(1)); err != nil {
 		clog.Errorf(ctx, "Error recording metrics err=%q", err)
 	}
 }
 
-func FastVerificationFailed(ctx context.Context) {
+func FastVerificationFailed(ctx context.Context, uri string) {
 	if err := stats.RecordWithTags(census.ctx,
-		manifestIDTag(ctx), census.mFastVerificationFailed.M(1)); err != nil {
+		manifestIDTag(ctx, tag.Insert(census.kOrchestratorURI, uri)),
+		census.mFastVerificationFailed.M(1)); err != nil {
 		clog.Errorf(ctx, "Error recording metrics err=%q", err)
 	}
 }
diff --git a/server/broadcast.go b/server/broadcast.go
@@ -517,18 +517,19 @@ func (bsm *BroadcastSessionsManager) chooseResults(ctx context.Context, submitRe
 	// verify untrusted hashes
 	var sessionsToSuspend []*BroadcastSession
 	for _, untrustedResult := range untrustedResults {
+		ouri := untrustedResult.Session.Transcoder()
 		untrustedHash, err := drivers.GetSegmentData(ctx, untrustedResult.TranscodeResult.Segments[segmToCheckIndex].PerceptualHashUrl)
 		if err != nil {
-			err = fmt.Errorf("error downloading perceptual hash from url=%s err=%w",
+			err = fmt.Errorf("error uri=%s downloading perceptual hash from url=%s err=%w", ouri,
 				untrustedResult.TranscodeResult.Segments[segmToCheckIndex].PerceptualHashUrl, err)
 			return nil, nil, err
 		}
 		equal, err := ffmpeg.CompareSignatureByBuffer(trustedHash, untrustedHash)
 		if monitor.Enabled {
-			monitor.FastVerificationDone(ctx)
+			monitor.FastVerificationDone(ctx, ouri)
 		}
 		if err != nil {
-			clog.Errorf(ctx, "error comparing perceptual hashes from url=%s err=%q",
+			clog.Errorf(ctx, "error uri=%s comparing perceptual hashes from url=%s err=%q", ouri,
 				untrustedResult.TranscodeResult.Segments[segmToCheckIndex].PerceptualHashUrl, err)
 		}
 		clog.Infof(ctx, "Hashes from url=%s and url=%s are equal=%v",
@@ -539,13 +540,13 @@ func (bsm *BroadcastSessionsManager) chooseResults(ctx context.Context, submitRe
 			// download untrusted video segment
 			untrustedSegm, err := drivers.GetSegmentData(ctx, untrustedResult.TranscodeResult.Segments[segmToCheckIndex].Url)
 			if err != nil {
-				err = fmt.Errorf("error downloading segment from url=%s err=%w",
+				err = fmt.Errorf("error uri=%s downloading segment from url=%s err=%w", ouri,
 					untrustedResult.TranscodeResult.Segments[segmToCheckIndex].Url, err)
 				return nil, nil, err
 			}
 			vequal, err = ffmpeg.CompareVideoByBuffer(trustedSegm, untrustedSegm)
 			if err != nil {
-				clog.Errorf(ctx, "error comparing video from url=%s err=%q",
+				clog.Errorf(ctx, "error uri=%s comparing video from url=%s err=%q", ouri,
 					untrustedResult.TranscodeResult.Segments[segmToCheckIndex].Url, err)
 				return nil, nil, err
 			}
@@ -563,7 +564,7 @@ func (bsm *BroadcastSessionsManager) chooseResults(ctx context.Context, submitRe
 		} else {
 			sessionsToSuspend = append(sessionsToSuspend, untrustedResult.Session)
 			if monitor.Enabled {
-				monitor.FastVerificationFailed(ctx)
+				monitor.FastVerificationFailed(ctx, ouri)
 			}
 		}
 	}