Add alerting for CTFE V2.

ct/go/ctfe/main.go

 /*
         The Cluster Telemetry Frontend.
 */
 
 package main
 
 import (
         "flag"
         "fmt"
         "net/http"
         "path/filepath"
         "runtime"
         "strconv"
         "time"
 
         "github.com/gorilla/mux"
         "github.com/skia-dev/glog"
 
+        metrics "github.com/rcrowley/go-metrics"
         "go.skia.org/infra/ct/go/ctfe/admin_tasks"
         "go.skia.org/infra/ct/go/ctfe/capture_skps"
         "go.skia.org/infra/ct/go/ctfe/chromium_builds"
         "go.skia.org/infra/ct/go/ctfe/chromium_perf"
         "go.skia.org/infra/ct/go/ctfe/lua_scripts"
         "go.skia.org/infra/ct/go/ctfe/pending_tasks"
         "go.skia.org/infra/ct/go/ctfe/task_common"
+        "go.skia.org/infra/ct/go/ctfe/task_types"
         ctfeutil "go.skia.org/infra/ct/go/ctfe/util"
         "go.skia.org/infra/ct/go/db"
         ctutil "go.skia.org/infra/ct/go/util"
         "go.skia.org/infra/go/common"
         "go.skia.org/infra/go/influxdb"
         "go.skia.org/infra/go/login"
         "go.skia.org/infra/go/metadata"
         "go.skia.org/infra/go/skiaversion"
         skutil "go.skia.org/infra/go/util"
         "go.skia.org/infra/go/webhook"
 )
 
 var (
         dbClient *influxdb.Client = nil
-
-        // Slice of all tasks supported by CTFE.
-        supportedTasks = []task_common.Task{
-                &chromium_perf.DBTask{},
-                &capture_skps.DBTask{},
-                &lua_scripts.DBTask{},
-                &chromium_builds.DBTask{},
-                &admin_tasks.RecreatePageSetsDBTask{},
-                &admin_tasks.RecreateWebpageArchivesDBTask{},
-        }
 )
 
 // flags
 var (
         graphiteServer         = flag.String("graphite_server", "localhost:2003", "Where is Graphite metrics ingestion server running.")
         host                   = flag.String("host", "localhost", "HTTP service host")
         port                   = flag.String("port", ":8002", "HTTP service port (e.g., ':8002')")
         local                  = flag.Bool("local", false, "Running locally if true. As opposed to in production.")
         workdir                = flag.String("workdir", ".", "Directory to use for scratch work.")
         resourcesDir           = flag.String("resources_dir", "", "The directory to find templates, JS, and CSS files. If blank the current directory will be used.")
@@ skipping 53 matching lines @@
         r.HandleFunc("/json/version", skiaversion.JsonHandler)
         r.HandleFunc("/oauth2callback/", login.OAuth2CallbackHandler)
         r.HandleFunc("/login/", loginHandler)
         r.HandleFunc("/logout/", login.LogoutHandler)
         r.HandleFunc("/loginstatus/", login.StatusHandler)
         http.Handle("/", skutil.LoggingGzipRequestResponse(r))
         glog.Infof("Ready to serve on %s", serverURL)
         glog.Fatal(http.ListenAndServe(*port, nil))
 }
 
+// startCtfeMetrics registers gauges with the graphite server that indicate CT is running healthily
+// and starts a goroutine to update them periodically.
+func startCtfeMetrics() {
+        pendingTasksGauge := metrics.GetOrRegisterGauge("num-pending-tasks", metrics.DefaultRegistry)
+        oldestPendingTaskAgeGauge := metrics.GetOrRegisterGaugeFloat64("oldest-pending-task-age", metrics.DefaultRegistry)
+        // 0=no tasks pending; 1=started; 2=not started
+        oldestPendingTaskStatusGauge := metrics.GetOrRegisterGauge("oldest-pending-task-status", metrics.DefaultRegistry)
+        go func() {
+                for _ = range time.Tick(common.SAMPLE_PERIOD) {
+                        pendingTaskCount, err := pending_tasks.GetPendingTaskCount()
+                        if err != nil {
+                                glog.Error(err)
+                        } else {
+                                pendingTasksGauge.Update(pendingTaskCount)
+                        }
+
+                        oldestPendingTask, err := pending_tasks.GetOldestPendingTask()
+                        if err != nil {
+                                glog.Error(err)
+                        } else if oldestPendingTask == nil {
+                                oldestPendingTaskAgeGauge.Update(0)
+                                oldestPendingTaskStatusGauge.Update(0)
+                        } else {
+                                addedTime := ctutil.GetTimeFromTs(strconv.FormatInt(oldestPendingTask.GetCommonCols().TsAdded.Int64, 10))
+                                oldestPendingTaskAgeGauge.Update(time.Since(addedTime).Seconds())
+                                if oldestPendingTask.GetCommonCols().TsStarted.Valid {
+                                        oldestPendingTaskStatusGauge.Update(1)
+                                } else {
+                                        oldestPendingTaskStatusGauge.Update(2)
+                                }
+                        }
+                }
+        }()
+}
+
 // repeatedTasksScheduler looks for all tasks that contain repeat_after_days
 // set to > 0 and schedules them when the specified time comes.
 // The function does the following:
 // 1. Look for tasks that need to be scheduled in the next 5 minutes.
 // 2. Loop over these tasks.
 //   2.1 Schedule the task again and set repeat_after_days to what it
 //       originally was.
 //   2.2 Update the original task and set repeat_after_days to 0 since the
 //       newly created task will now replace it.
 func repeatedTasksScheduler() {
 
         for _ = range time.Tick(*tasksSchedulerWaitTime) {
                 // Loop over all tasks to find tasks which need to be scheduled.
-                for _, prototype := range supportedTasks {
+                for _, prototype := range task_types.Prototypes() {
 
                         query, args := task_common.DBTaskQuery(prototype,
                                 task_common.QueryParams{
                                         FutureRunsOnly: true,
                                         Offset:         0,
                                         Size:           task_common.MAX_PAGE_SIZE,
                                 })
                         glog.Infof("Running %s", query)
                         data, err := prototype.Select(query, args...)
                         if err != nil {
@@ skipping 81 matching lines @@
         // Initialize the ctfe database.
         if !*local {
                 if err := dbConf.GetPasswordFromMetadata(); err != nil {
                         glog.Fatal(err)
                 }
         }
         if err := dbConf.InitDB(); err != nil {
                 glog.Fatal(err)
         }
 
+        startCtfeMetrics()
+
         // Start the repeated tasks scheduler.
         go repeatedTasksScheduler()
 
         runServer(serverURL)
 }