Migrate fuzzer to use shiny new metrics2 package

fuzzer/go/aggregator/aggregator.go

 package aggregator
 
 import (
         "bytes"
         "crypto/sha1"
         "fmt"
         "io"
         "io/ioutil"
         "os"
         "path/filepath"
         "sort"
         "strings"
         "sync"
         "sync/atomic"
         "time"
 
-        go_metrics "github.com/rcrowley/go-metrics"
         "github.com/skia-dev/glog"
         "go.skia.org/infra/fuzzer/go/common"
         "go.skia.org/infra/fuzzer/go/config"
         "go.skia.org/infra/fuzzer/go/frontend/data"
         "go.skia.org/infra/go/exec"
         "go.skia.org/infra/go/fileutil"
+        "go.skia.org/infra/go/metrics2"
         "go.skia.org/infra/go/util"
         "golang.org/x/net/context"
         "google.golang.org/cloud/storage"
 )
 
 // Aggregator is a key part of the fuzzing operation
 // (see https://skia.googlesource.com/buildbot/+/master/fuzzer/DESIGN.md).
 // It will find new bad fuzzes generated by afl-fuzz and create the metadata required for them. It
 // does this by searching in the specified AflOutputPath for new crashes and moves them to a
 // temporary holding folder (specified by FuzzPath) for parsing, before sending them through the
@@ skipping 32 matching lines @@
         aggregationWaitGroup *sync.WaitGroup
         // These three counts are used to determine if there is any pending work.
         // There is no pending work if all three of these values are equal and the
         // work queues are empty.
         analysisCount  int64
         uploadCount    int64
         bugReportCount int64
 }
 
 const (
-        BAD_FUZZ  = "bad"
-        GREY_FUZZ = "grey"
+        BAD_FUZZ       = "bad"
+        GREY_FUZZ      = "grey"
+        HANG_THRESHOLD = 100
 )
 
 var (
         CLANG_DEBUG   = common.TEST_HARNESS_NAME + "_clang_debug"
         CLANG_RELEASE = common.TEST_HARNESS_NAME + "_clang_release"
         ASAN_DEBUG    = common.TEST_HARNESS_NAME + "_asan_debug"
         ASAN_RELEASE  = common.TEST_HARNESS_NAME + "_asan_release"
 )
 
 // uploadPackage is a struct containing all the pieces of a fuzz that need to be uploaded to GCS
@@ skipping 150 matching lines @@
 // scanHelper runs findBadFuzzPaths, logs the output and keeps alreadyFoundFuzzes up to date.
 func (agg *Aggregator) scanHelper(alreadyFoundFuzzes *SortedStringSlice) error {
         newlyFound, err := agg.findBadFuzzPaths(alreadyFoundFuzzes)
         if err != nil {
                 return err
         }
         // AFL-fuzz does not write crashes or hangs atomically, so this workaround waits for a bit after
         // we have references to where the crashes will be.
         // TODO(kjlubick), switch to using flock once afl-fuzz implements that upstream.
         time.Sleep(time.Second)
-        go_metrics.GetOrRegisterGauge("binary_newly_found_fuzzes", go_metrics.DefaultRegistry).Update(int64(len(newlyFound)))
-        glog.Infof("[%s] %d newly found bad binary fuzzes", agg.Category, len(newlyFound))
+        metrics2.GetInt64Metric("newly_found_fuzzes", map[string]string{"category": agg.Category}).Update(int64(len(newlyFound)))

[borenet, 2016/02/10 15:32:32]

Naming conventions: measurements and tag keys should be
words-separated-by-dashes, and measurement names which aren't shared (eg.
"counter", "liveness") should be dot-separated following roughly this format:
domain.category.metric, eg. "fuzzer.fuzzes.newly-found"

[kjlubick, 2016/02/10 16:14:29]

Done.

+        glog.Infof("[%s] %d newly found bad fuzzes", agg.Category, len(newlyFound))
         for _, f := range newlyFound {
                 agg.forAnalysis <- f
         }
         alreadyFoundFuzzes.Append(newlyFound)
         return nil
 }
 
 // findBadFuzzPaths looks through all the afl-fuzz directories contained in the passed in path and
 // returns the path to all files that are in a crash* folder that are not already in
 // 'alreadyFoundFuzzes'.  It also sends them to the forAnalysis channel when it finds them.
@@ skipping 65 matching lines @@
         }
         return badFuzzPaths, nil
 }
 
 // waitForAnalysis waits for files that need to be analyzed (from forAnalysis) and makes a copy of
 // them in agg.fuzzPath with their hash as a file name. It then analyzes it using the supplied
 // AnalysisPackage and then signals the results should be uploaded. If any unrecoverable errors
 // happen, this method terminates.
 func (agg *Aggregator) waitForAnalysis(identifier int) {
         defer agg.aggregationWaitGroup.Done()
-        defer go_metrics.GetOrRegisterCounter("analysis_process_count", go_metrics.DefaultRegistry).Dec(int64(1))
+        defer metrics2.NewCounter("analysis_process_count", map[string]string{"category": agg.Category}).Dec(int64(1))

[borenet, 2016/02/10 15:32:32]

Use dashes, ie. "analysis-process-count", here and below.

[kjlubick, 2016/02/10 16:14:29]

Done.

         glog.Infof("[%s] Spawning analyzer %d", agg.Category, identifier)
 
         // our own unique working folder
         executableDir := filepath.Join(agg.executablePath, fmt.Sprintf("analyzer%d", identifier))
         if err := agg.setupAnalysis(executableDir); err != nil {
                 glog.Errorf("[%s] Analyzer %d terminated due to error: %s", agg.Category, identifier, err)
                 return
         }
         for {
                 select {
@@ skipping 155 matching lines @@
 // Append adds all of the strings to the underlying slice and sorts it
 func (s *SortedStringSlice) Append(strs []string) {
         s.strings = append(s.strings, strs...)
         s.strings.Sort()
 }
 
 // waitForUploads waits for uploadPackages to be sent through the forUpload channel and then uploads
 // them.  If any unrecoverable errors happen, this method terminates.
 func (agg *Aggregator) waitForUploads(identifier int) {
         defer agg.aggregationWaitGroup.Done()
-        defer go_metrics.GetOrRegisterCounter("upload_process_count", go_metrics.DefaultRegistry).Dec(int64(1))
+        defer metrics2.NewCounter("upload_process_count", map[string]string{"category": agg.Category}).Dec(int64(1))
         glog.Infof("[%s] Spawning uploader %d", agg.Category, identifier)
         for {
                 select {
                 case p := <-agg.forUpload:
                         atomic.AddInt64(&agg.uploadCount, int64(1))
                         if !agg.UploadGreyFuzzes && p.FuzzType == GREY_FUZZ {
                                 glog.Infof("[%s] Skipping upload of grey fuzz %s", agg.Category, p.Data.Name)
                                 continue
                         }
                         if err := agg.upload(p); err != nil {
@@ skipping 98 matching lines @@
         if agg.MakeBugOnBadFuzz && p.IsBadFuzz {
                 glog.Warningf("[%s] Should create bug for %s", agg.Category, p.FuzzName)
         }
         return nil
 }
 
 // monitorStatus sets up the monitoring routine, which reports how big the work queues are and how
 // many processes are up.
 func (agg *Aggregator) monitorStatus(numAnalysisProcesses, numUploadProcesses int) {
         defer agg.monitoringWaitGroup.Done()
-        analysisProcessCount := go_metrics.GetOrRegisterCounter("analysis_process_count", go_metrics.DefaultRegistry)
-        analysisProcessCount.Clear()
+        analysisProcessCount := metrics2.NewCounter("analysis_process_count", map[string]string{"category": agg.Category})
+        analysisProcessCount.Reset()
         analysisProcessCount.Inc(int64(numAnalysisProcesses))
-        uploadProcessCount := go_metrics.GetOrRegisterCounter("upload_process_count", go_metrics.DefaultRegistry)
-        uploadProcessCount.Clear()
+        uploadProcessCount := metrics2.NewCounter("upload_process_count", map[string]string{"category": agg.Category})
+        uploadProcessCount.Reset()
         uploadProcessCount.Inc(int64(numUploadProcesses))
 
         t := time.Tick(config.Aggregator.StatusPeriod)
         for {
                 select {
                 case <-agg.monitoringShutdown:
                         glog.Infof("[%s] aggregator monitor got signal to shut down", agg.Category)
                         return
                 case <-t:
-                        go_metrics.GetOrRegisterGauge("binary_analysis_queue_size", go_metrics.DefaultRegistry).Update(int64(len(agg.forAnalysis)))
-                        go_metrics.GetOrRegisterGauge("binary_upload_queue_size", go_metrics.DefaultRegistry).Update(int64(len(agg.forUpload)))
-                        go_metrics.GetOrRegisterGauge("binary_bug_report_queue_size", go_metrics.DefaultRegistry).Update(int64(len(agg.forBugReporting)))
+                        metrics2.GetInt64Metric("analysis_queue_size", map[string]string{"category": agg.Category}).Update(int64(len(agg.forAnalysis)))
+                        metrics2.GetInt64Metric("upload_queue_size", map[string]string{"category": agg.Category}).Update(int64(len(agg.forUpload)))
+                        metrics2.GetInt64Metric("bug_report_queue_size", map[string]string{"category": agg.Category}).Update(int64(len(agg.forBugReporting)))
                 }
         }
 }
 
 // Shutdown gracefully shuts down the aggregator. Anything that was being processed will finish
 // prior to the shutdown.
 func (agg *Aggregator) ShutDown() {
         // once for the monitoring and once for the scanning routines
         agg.monitoringShutdown <- true
         agg.monitoringShutdown <- true
@@ skipping 20 matching lines @@
 func (agg *Aggregator) WaitForEmptyQueues() {
         a := len(agg.forAnalysis)
         u := len(agg.forUpload)
         b := len(agg.forBugReporting)
         if a == 0 && u == 0 && b == 0 && agg.analysisCount == agg.uploadCount && agg.uploadCount == agg.bugReportCount {
                 glog.Infof("[%s] Queues were already empty", agg.Category)
                 return
         }
         t := time.Tick(config.Aggregator.StatusPeriod)
         glog.Infof("[%s] Waiting %s for the aggregator's queues to be empty", agg.Category, config.Aggregator.StatusPeriod)
+        hangCount := 0
         for _ = range t {
                 a = len(agg.forAnalysis)
                 u = len(agg.forUpload)
                 b = len(agg.forBugReporting)
                 glog.Infof("[%s] AnalysisQueue: %d, UploadQueue: %d, BugReportingQueue: %d", agg.Category, a, u, b)
                 glog.Infof("[%s] AnalysisTotal: %d, UploadTotal: %d, BugReportingTotal: %d", agg.Category, agg.analysisCount, agg.uploadCount, agg.bugReportCount)
                 if a == 0 && u == 0 && b == 0 && agg.analysisCount == agg.uploadCount && agg.uploadCount == agg.bugReportCount {
                         break
                 }
+                // This prevents waiting forever if an upload crashes, aborts or otherwise hangs.
+                hangCount++
+                if hangCount >= HANG_THRESHOLD {
+                        glog.Warningf("Was waiting for %d rounds and still wasn't done.  Quitting anyway.", hangCount)
+                }

[borenet, 2016/02/10 15:32:32]

Does this logic belong in this CL?

[kjlubick, 2016/02/10 16:14:29]

Removed.

                 glog.Infof("[%s] Waiting %s for the aggregator's queues to be empty", agg.Category, config.Aggregator.StatusPeriod)
+
         }
 }
 
 // ForceAnalysis directly adds the given path to the analysis queue, where it will be analyzed,
 // uploaded and possibly bug reported.
 func (agg *Aggregator) ForceAnalysis(path string) {
         agg.forAnalysis <- path
 }