[som] add tsmon to alerts-dispatcher

go/src/infra/monitoring/dispatcher/dispatcher.go

 // Copyright 2015 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 // Dispatcher usage:
 // go run infra/monitoring/dispatcher
 // Expects gatekeeper.json to be in the current directory.
 
 package main
 
 import (
         "encoding/json"
         "expvar"
         "flag"
         "fmt"
         "io/ioutil"
         "log"
         "net"
         "net/http"
         "net/url"
         "os"
         "sort"
         "strings"
         "time"
 
         "github.com/luci/luci-go/common/auth"
         "github.com/luci/luci-go/common/clock"
         "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/logging/gologger"
+        "github.com/luci/luci-go/common/tsmon"
+        "github.com/luci/luci-go/common/tsmon/field"
+        "github.com/luci/luci-go/common/tsmon/metric"
+        "github.com/luci/luci-go/common/tsmon/types"
 
         "golang.org/x/net/context"
 
         "infra/monitoring/analyzer"
         "infra/monitoring/client"
         "infra/monitoring/looper"
         "infra/monitoring/messages"
 )
 
 type stringSlice []string
@@ skipping 29 matching lines @@
         duration, cycle time.Duration
 
         // gk is the gatekeeper config.
         gks = []*messages.GatekeeperConfig{}
         // gkt is the gatekeeper trees config.
         gkts             = map[string][]messages.TreeMasterConfig{}
         filteredFailures = uint64(0)
         expvars          = expvar.NewMap("dispatcher")
         errLog           = log.New(os.Stderr, "", log.Lshortfile|log.Ltime)
         infoLog          = log.New(os.Stdout, "", log.Lshortfile|log.Ltime)
+
+        // tsmon metrics
+        iterations = metric.NewCounter("alerts_dispatcher/iterations",
+                "Number if iterations of the main polling loop.", types.MetricMetadata{}, field.String("status"))
+        postErrors = metric.NewCounter("alerts_dispatcher/post_errors",
+                "Number of posting errors.", types.MetricMetadata{})
+        alertCount = metric.NewInt("alerts_dispatcher/alert_count",
+                "Number of alerts generated.", types.MetricMetadata{},
+                field.String("tree"))
 )
 
 func init() {
         flag.Usage = func() {
                 fmt.Printf("By default runs a single check, saves any alerts to ./alerts.json and exits.")
                 flag.PrintDefaults()
         }
 }
 
 func analyzeBuildExtract(ctx context.Context, a *analyzer.Analyzer, tree string, masterURL *messages.MasterLocation, b *messages.BuildExtract) []messages.Alert {
@@ skipping 102 matching lines @@
                                                         errLog.Printf("Couldn't get revision summaries: %v", err)
                                                         continue
                                                 }
                                                 for _, rev := range revs {
                                                         alerts.RevisionSummaries[rev.GitHash] = rev
                                                 }
                                         }
                                 }
                         }
                         alerts.Timestamp = messages.TimeToEpochTime(time.Now())
+                        alertCount.Set(ctx, int64(len(alerts.Alerts)), tree)
 
                         if *alertsBaseURL == "" {
                                 infoLog.Printf("No data_url provided. Writing to %s-alerts.json", tree)
 
                                 abytes, err := json.MarshalIndent(alerts, "", "\t")
                                 if err != nil {
                                         errLog.Printf("Couldn't marshal alerts json: %v", err)
                                         errs <- err
                                         return
                                 }
 
                                 if err := ioutil.WriteFile(fmt.Sprintf("%s-alerts.json", tree), abytes, 0644); err != nil {
                                         errLog.Printf("Couldn't write to alerts.json: %v", err)
                                         errs <- err
                                         return
                                 }
                         } else {
                                 alertsURL := fmt.Sprintf("%s/%s", *alertsBaseURL, tree)
                                 w := client.NewWriter(alertsURL, transport)
                                 infoLog.Printf("Posting alerts to %s", alertsURL)
                                 err := w.PostAlerts(alerts)
                                 if err != nil {
                                         errLog.Printf("Couldn't post alerts: %v", err)
+                                        postErrors.Add(ctx, 1)
                                         errs <- err
                                         return
                                 }
                         }
 
                         infoLog.Printf("Filtered failures: %v", filteredFailures)
                         done <- nil
                 }()
         }
 
@@ skipping 10 matching lines @@
 
 func main() {
         flag.Var(&gatekeeperJSON, "gatekeeper", "Location of gatekeeper json file. Can have multiple comma separated values.")
         flag.Var(&gatekeeperTreesJSON, "gatekeeper-trees", "Location of gatekeeper tree json file. Can have multiple comma separated values.")
 
         flag.Parse()
 
         ctx := context.Background()
         ctx = gologger.StdConfig.Use(ctx)
         logging.SetLevel(ctx, logging.Debug)
-
         authOptions := auth.Options{
                 ServiceAccountJSONPath: *serviceAccountJSON,
                 Scopes: []string{
                         auth.OAuthScopeEmail,
                         "https://www.googleapis.com/auth/projecthosting",
                 },
                 Method: auth.ServiceAccountMethod,
         }
 
         mode := auth.SilentLogin
         if *login {
                 mode = auth.InteractiveLogin
         }
 
         transport, err := auth.NewAuthenticator(ctx, mode, authOptions).Transport()
         if err != nil {
                 errLog.Printf("AuthenticatedTransport: %v", err)
                 if !*login {
                         errLog.Printf("Consider re-running with -login")
                 }
                 os.Exit(1)
         }
         ctx = context.Background()
 
+        tsFlags := tsmon.NewFlags()
+        tsFlags.Target.TargetType = "task"
+        tsFlags.Target.TaskServiceName = "alerts-dispatcher"
+        tsFlags.Flush = "auto"
+        tsmon.InitializeFromFlags(ctx, &tsFlags)
+
         // Start serving expvars.
         go func() {
                 listener, err := net.Listen("tcp", "127.0.0.1:0")
                 if err != nil {
                         errLog.Printf("Listen: %s", err)
                         os.Exit(1)
                 }
 
                 infoLog.Printf("expvars listening on %v", listener.Addr())
 
@@ skipping 108 matching lines @@
         for tree := range trees {
                 if _, ok := gkts[tree]; !ok {
                         errLog.Printf("Unrecognized tree name: %s", tree)
                         os.Exit(1)
                 }
         }
 
         // This is the polling/analysis/alert posting function, which will run in a loop until
         // a timeout or max errors is reached.
         f := func(ctx context.Context) error {
-                return mainLoop(ctx, a, trees, transport)
+                err := mainLoop(ctx, a, trees, transport)
+                if err == nil {
+                        iterations.Add(ctx, 1, "success")

[dsansome, 2016/08/26 08:44:52]

Why not do this in looper instead, and use the same metric definitions as
infra/libs/service_utils/outer_loop.py?

[seanmccullough1, 2016/08/26 19:48:22]

+1 for doing it in looper, Not sure about using the same names since those imply
where the metric is coming from (i.e. outer_loop is in the metric name)

+                } else {
+                        iterations.Add(ctx, 1, "failure")
+                }
+                return err
         }
 
         ctx, cancel := context.WithTimeout(ctx, duration)
         defer cancel()
 
         loopResults := looper.Run(ctx, f, cycle, *maxErrs, clock.GetSystemClock())
 
+        tsmon.Shutdown(ctx)
+
         if !loopResults.Success {
                 errLog.Printf("Failed to run loop, %v errors", loopResults.Errs)
                 os.Exit(1)
         }
 }