Use fuzzy pixel matching in local verification

CHANGELOG_PENDING.md

@@ -23,7 +23,10 @@
 #### Broadcaster
 - \#2709 Add logging for high keyframe interval, reduce log level for discovery loop
 - \#2684 Fix transcode success rate metric
+- \#2649 Fix for broadcaster local verification fails for vertical videos
+- \#2740 Fix incorrect processing of VerificationFreq parameter, which caused fast verification to run too often
 
 #### Orchestrator
 
 #### Transcoder
+- \#2732 Fix transcoder crash on session teardown, fix maxing out available sessions, because fast verification session teardown is not being called

server/broadcast_test.go

@@ -1414,9 +1414,9 @@ func TestVerifier_HLSInsertion(t *testing.T) {
 		retries: 2,
 		err:     verification.ErrTampered,
 		results: []verification.Results{
-			{Score: 5, Pixels: []int64{100}},
-			{Score: 9, Pixels: []int64{100}},
-			{Score: 1, Pixels: []int64{100}},
+			{Score: 5, Stats: []ffmpeg.MediaInfo{{Pixels: 100}}},
+			{Score: 9, Stats: []ffmpeg.MediaInfo{{Pixels: 100}}},
+			{Score: 1, Stats: []ffmpeg.MediaInfo{{Pixels: 100}}},
 		},
 	})
 
@@ -1464,9 +1464,9 @@ func TestDownloadSegError_SuspendAndRemove(t *testing.T) {
 		retries: 2,
 		err:     verification.ErrTampered,
 		results: []verification.Results{
-			{Score: 5, Pixels: []int64{100}},
-			{Score: 9, Pixels: []int64{100}},
-			{Score: 1, Pixels: []int64{100}},
+			{Score: 5, Stats: []ffmpeg.MediaInfo{{Pixels: 100}}},
+			{Score: 9, Stats: []ffmpeg.MediaInfo{{Pixels: 100}}},
+			{Score: 1, Stats: []ffmpeg.MediaInfo{{Pixels: 100}}},
 		},
 	})
 	oldDownloadSeg := downloadSeg

verification/epic.go

@@ -69,8 +69,8 @@ type EpicClassifier struct {
 func epicResultsToVerificationResults(er *epicResults) (*Results, error) {
 	// find average of scores and build list of pixels
 	var (
-		score  float64
-		pixels []int64
+		score float64
+		mis   []ffmpeg.MediaInfo
 	)
 	var err error
 	// If an error is gathered, continue to gather overall pixel counts
@@ -88,9 +88,9 @@ func epicResultsToVerificationResults(er *epicResults) (*Results, error) {
 			err = ErrVideoUnavailable
 		}
 		score += v.OCSVMDist / float64(len(er.Results))
-		pixels = append(pixels, v.Pixels)
+		mis = append(mis, ffmpeg.MediaInfo{Pixels: v.Pixels})
 	}
-	return &Results{Score: score, Pixels: pixels}, err
+	return &Results{Score: score, Stats: mis}, err
 }
 
 func (e *EpicClassifier) Verify(params *Params) (*Results, error) {

verification/epic_test.go

@@ -29,7 +29,7 @@ func TestEpic_EpicResultsToVerificationResults(t *testing.T) {
 	assert.Nil(err)
 	assert.NotNil(res)
 	assert.Equal(0., res.Score)
-	assert.Empty(res.Pixels)
+	assert.Empty(res.Stats)
 
 	// Success case
 	r = &epicResults{
@@ -40,7 +40,7 @@ func TestEpic_EpicResultsToVerificationResults(t *testing.T) {
 	}
 	res, err = fn(r)
 	assert.Nil(err)
-	assert.Equal([]int64{123, 456}, res.Pixels)
+	assert.Equal([]int64{123, 456}, []int64{res.Stats[0].Pixels, res.Stats[1].Pixels})
 	assert.Equal((2.0+3.0)/2., res.Score)
 
 	// Check various errors. Should always count pixels/score for each rendition
@@ -50,7 +50,7 @@ func TestEpic_EpicResultsToVerificationResults(t *testing.T) {
 	r.Results[0].AudioDistance = 1.0
 	res, err = fn(r)
 	assert.Equal(ErrAudioMismatch, err)
-	assert.Equal([]int64{123, 456}, res.Pixels)
+	assert.Equal([]int64{123, 456}, []int64{res.Stats[0].Pixels, res.Stats[1].Pixels})
 	assert.Equal((2.0+3.0)/2., res.Score)
 
 	// Check tampering. Tamper first result only.
@@ -59,44 +59,44 @@ func TestEpic_EpicResultsToVerificationResults(t *testing.T) {
 	r.Results[0].OCSVMDist = -2.0
 	res, err = fn(r)
 	assert.Equal(ErrAudioMismatch, err)
-	assert.Equal([]int64{123, 456}, res.Pixels)
+	assert.Equal([]int64{123, 456}, []int64{res.Stats[0].Pixels, res.Stats[1].Pixels})
 	assert.Equal((-2.0+3.0)/2., res.Score)
 	// Disable audio mismatch and we should get a tamper error
 	r.Results[0].AudioDistance = 0.0
 	res, err = fn(r)
 	assert.Equal(ErrTampered, err)
-	assert.Equal([]int64{123, 456}, res.Pixels)
+	assert.Equal([]int64{123, 456}, []int64{res.Stats[0].Pixels, res.Stats[1].Pixels})
 	assert.Equal((-2.0+3.0)/2., res.Score)
 
 	// Tamper both results
 	r.Results[1].Tamper = 1
 	r.Results[1].OCSVMDist = -3.0
 	res, err = fn(r)
 	assert.Equal(ErrTampered, err)
-	assert.Equal([]int64{123, 456}, res.Pixels)
+	assert.Equal([]int64{123, 456}, []int64{res.Stats[0].Pixels, res.Stats[1].Pixels})
 	assert.Equal((-2.0+-3.0)/2., res.Score)
 
 	// Tamper second result only
 	r.Results[0].Tamper = 0
 	r.Results[0].OCSVMDist = 2.0
 	res, err = fn(r)
 	assert.Equal(ErrTampered, err)
-	assert.Equal([]int64{123, 456}, res.Pixels)
+	assert.Equal([]int64{123, 456}, []int64{res.Stats[0].Pixels, res.Stats[1].Pixels})
 	assert.Equal((2.0-3.0)/2., res.Score)
 
 	// Unset tamper, sanity check success
 	r.Results[1].Tamper = 0
 	r.Results[1].OCSVMDist = 3.0
 	res, err = fn(r)
 	assert.Nil(err)
-	assert.Equal([]int64{123, 456}, res.Pixels)
+	assert.Equal([]int64{123, 456}, []int64{res.Stats[0].Pixels, res.Stats[1].Pixels})
 	assert.Equal((2.0+3.0)/2., res.Score)
 
 	// Unset video availability
 	r.Results[0].VideoAvailable = false
 	res, err = fn(r)
 	assert.Equal(ErrVideoUnavailable, err)
-	assert.Equal([]int64{123, 456}, res.Pixels) // Does this even make sense?
+	assert.Equal([]int64{123, 456}, []int64{res.Stats[0].Pixels, res.Stats[1].Pixels}) // Does this even make sense?
 	assert.Equal((2.0+3.0)/2., res.Score)
 }
 
@@ -247,7 +247,7 @@ func TestEpic_Verify(t *testing.T) {
 	ec.Addr = ts.URL + "/checkReq"
 	res, err := ec.Verify(params)
 	assert.Nil(err)
-	assert.Equal([]int64{123, 456}, res.Pixels)
+	assert.Equal([]int64{123, 456}, []int64{res.Stats[0].Pixels, res.Stats[1].Pixels})
 	assert.Equal((1.0+2.0)/2.0, res.Score)
 }
 

verification/verify.go

@@ -2,17 +2,13 @@ package verification
 
 import (
 	"errors"
-	"fmt"
-	"io/ioutil"
-	"os"
 	"sort"
 
 	"github.com/golang/glog"
 
 	ethcommon "github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/crypto"
 
-	"github.com/livepeer/go-livepeer/common"
 	"github.com/livepeer/go-livepeer/core"
 	lpcrypto "github.com/livepeer/go-livepeer/crypto"
 	"github.com/livepeer/go-livepeer/net"
@@ -67,8 +63,8 @@ type Results struct {
 	// Verifier specific score
 	Score float64
 
-	// Number of pixels decoded in this result
-	Pixels []int64
+	// Decoder stats
+	Stats []ffmpeg.MediaInfo
 }
 
 type Verifier interface {
@@ -134,14 +130,14 @@ func (sv *SegmentVerifier) Verify(params *Params) (*Params, error) {
 
 	// Check pixel counts
 	if (err == nil || (err != ErrAudioMismatch && IsRetryable(err))) && res != nil && params.Results != nil {
-		pxls := res.Pixels
-		if len(pxls) != len(params.Results.Segments) {
-			pxls, err = countPixelParams(params)
+		stats := res.Stats
+		if len(stats) != len(params.Results.Segments) {
+			stats, err = GetAllDecoderStats(params)
 		}
-		for i := 0; err == nil && i < len(params.Results.Segments) && i < len(pxls); i++ {
+		for i := 0; err == nil && i < len(params.Results.Segments) && i < len(stats); i++ {
 			reportedPixels := params.Results.Segments[i].Pixels
-			verifiedPixels := pxls[i]
-			if reportedPixels != verifiedPixels {
+			verifiedMediainfo := stats[i]
+			if !ffmpeg.FuzzyMatchMediaInfo(verifiedMediainfo, reportedPixels) {
 				err = ErrPixelMismatch
 			}
 		}
@@ -214,55 +210,20 @@ func (sv *SegmentVerifier) sigVerification(params *Params) error {
 	return nil
 }
 
-func countPixelParams(params *Params) ([]int64, error) {
+func GetAllDecoderStats(params *Params) ([]ffmpeg.MediaInfo, error) {
 
 	if len(params.Results.Segments) != len(params.Renditions) {
 		return nil, ErrPixelsAbsent
 	}
 
-	pxls := make([]int64, len(params.Results.Segments))
+	mis := make([]ffmpeg.MediaInfo, len(params.Results.Segments))
 
 	for i := 0; i < len(params.Results.Segments); i++ {
-		count, err := countPixels(params.Renditions[i])
+		mi, err := ffmpeg.GetDecoderStatsBytes(params.Renditions[i])
 		if err != nil {
 			return nil, err
 		}
-		pxls[i] = count
+		mis[i] = *mi
 	}
-	return pxls, nil
-}
-
-func countPixels(data []byte) (int64, error) {
-	// write the data to a temp file
-	tempfile, err := ioutil.TempFile("", common.RandName())
-	if err != nil {
-		return 0, fmt.Errorf("error creating temp file for pixels verification: %w", err)
-	}
-	defer os.Remove(tempfile.Name())
-
-	if _, err := tempfile.Write(data); err != nil {
-		tempfile.Close()
-		return 0, fmt.Errorf("error writing temp file for pixels verification: %w", err)
-	}
-
-	if err = tempfile.Close(); err != nil {
-		return 0, fmt.Errorf("error closing temp file for pixels verification: %w", err)
-	}
-
-	p, err := pixels(tempfile.Name())
-	if err != nil {
-		return 0, err
-	}
-
-	return p, nil
-}
-
-func pixels(fname string) (int64, error) {
-	in := &ffmpeg.TranscodeOptionsIn{Fname: fname}
-	res, err := ffmpeg.Transcode3(in, nil)
-	if err != nil {
-		return 0, err
-	}
-
-	return res.Decoded.Pixels, nil
+	return mis, nil
 }