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 204 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("fuzzer.fuzzes.newly-found", map[string]string{"category": agg.Category}).Update(int64(len(newlyFound)))
+        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))
         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("fuzzer.queue-size.analysis", map[string]string{"category": agg.Category}).Update(int64(len(agg.forAnalysis)))
+                        metrics2.GetInt64Metric("fuzzer.queue-size.upload", map[string]string{"category": agg.Category}).Update(int64(len(agg.forUpload)))
+                        metrics2.GetInt64Metric("fuzzer.queue-size.bug_report", map[string]string{"category": agg.Category}).Update(int64(len(agg.forBugReporting)))

[borenet, 2016/02/11 12:23:28]

Nit: bug_report -> bug-report

[kjlubick, 2016/02/11 17:54:41]

Done.

                 }
         }
 }
 
 // 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 30 matching lines @@
         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
                 }
                 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
 }