// 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)
				}
			}
		}
	}()
}

func MakeRules(cfgFile string, dbClient *influxdb.Client, tickInterval time.Duration, am *alerting.AlertManager, testing bool) ([]*Rule, error) {
	parsedRules, err := parseAlertRules(cfgFile)
	if err != nil {
		return nil, err
	}
	rules := map[string]*Rule{}
	for _, r := range parsedRules {
		r, err := newRule(r, dbClient, testing, tickInterval)
		if err != nil {
			return nil, err
		}
		if _, ok := rules[r.Name]; ok {
			return nil, fmt.Errorf("Found multiple rules with the same name: %s", r.Name)
		}
		rules[r.Name] = r
	}

	// Start the goroutines.
	rv := make([]*Rule, 0, len(rules))
	for _, r := range rules {
		rv = append(rv, r)
		go func(rule *Rule) {
			if err := rule.tick(am); err != nil {
				glog.Error(err)
			}
			for _ = range time.Tick(tickInterval) {
				if err := rule.tick(am); err != nil {
					glog.Error(err)
				}
			}
		}(r)
	}

	return rv, nil
}

func pollAndExecOnce() {
	pending, err := frontend.GetOldestPendingTaskV2()
	if err != nil {
		glog.Error(err)
		return
	}
	task := asPollerTask(pending)
	if task == nil {
		return
	}
	taskName, id := task.GetTaskName(), task.GetCommonCols().Id
	glog.Infof("Preparing to execute task %s %d", taskName, id)
	if err = updateAndBuild(); err != nil {
		glog.Error(err)
		return
	}
	glog.Infof("Executing task %s %d", taskName, id)
	if err = task.Execute(); err == nil {
		glog.Infof("Completed task %s %d", taskName, id)
	} else {
		glog.Errorf("Task %s %d failed: %v", taskName, id, err)
		if !*dryRun {
			if err := updateWebappTaskSetFailed(task); err != nil {
				glog.Error(err)
			}
		}
	}
}

func doWorkerHealthCheck() {
	if err := updateAndBuild(); err != nil {
		glog.Error(err)
		return
	}
	if err := checkWorkerHealth(); err != nil {
		glog.Error(err)
		return
	}
}

// Search returns a slice of Issues which fit the given criteria.
func (r Rietveld) Search(limit int, terms ...*SearchTerm) ([]*Issue, error) {
	searchUrl := fmt.Sprintf("/search?format=json&limit=%d", limit)
	for _, term := range terms {
		searchUrl += fmt.Sprintf("&%s=%s", term.Key, term.Value)
	}

	var issues issueListSortable
	cursor := ""
	for {
		var data rietveldResults
		err := r.get(searchUrl+cursor, &data)
		if err != nil {
			return nil, fmt.Errorf("Rietveld search failed: %v", err)
		}
		if len(data.Results) == 0 {
			break
		}
		for _, issue := range data.Results {
			fullIssue, err := r.getIssueProperties(issue.Issue, true)
			if err != nil {
				glog.Error(err)
			} else {
				fullIssue.Created = parseTime(fullIssue.CreatedString)
				fullIssue.Modified = parseTime(fullIssue.ModifiedString)
				for _, msg := range fullIssue.Messages {
					committed := false
					for _, r := range committedIssueRegexp {
						committed, err = regexp.MatchString(r, msg.Text)
						if committed {
							break
						}
					}
					msg.Date = parseTime(msg.DateString)
					if err != nil {
						glog.Error(err)
						continue
					}
					if committed {
						fullIssue.Committed = true
					}
				}
				issues = append(issues, &fullIssue)
			}
		}
		if len(issues) >= limit {
			break
		}
		cursor = "&cursor=" + data.Cursor
	}
	sort.Sort(issues)
	return issues, nil
}

// loop runs the AlertManager's main loop.
func (am *AlertManager) loop() {
	if err := am.tick(); err != nil {
		glog.Error(err)
	}
	for _ = range time.Tick(am.tickInterval) {
		select {
		case <-am.interrupt:
			return
		default:
		}
		if err := am.tick(); err != nil {
			glog.Error(err)
		}
	}
}

// NewIngestionProcess creates a Process for ingesting data.
func NewIngestionProcess(git *gitinfo.GitInfo, tileDir, datasetName string, ri ingester.ResultIngester, config map[string]string, every time.Duration, nCommits int, minDuration time.Duration, statusDir, metricName string) ProcessStarter {
	return func() {
		i, err := ingester.NewIngester(git, tileDir, datasetName, ri, nCommits, minDuration, config, statusDir, metricName)
		if err != nil {
			glog.Fatalf("Failed to create Ingester: %s", err)
		}

		glog.Infof("Starting %s ingester. Run every %s.", datasetName, every.String())

		// oneStep is a single round of ingestion.
		oneStep := func() {
			glog.Infof("Running ingester: %s", datasetName)
			err := i.Update()
			if err != nil {
				glog.Error(err)
			}
			glog.Infof("Finished running ingester: %s", datasetName)
		}

		// Start the ingester.
		go func() {
			oneStep()
			for _ = range time.Tick(every) {
				oneStep()
			}
		}()
	}
}

func NewPollingStatus(value interface{}, poll func(interface{}) error, frequency time.Duration) (*PollingStatus, error) {
	s := PollingStatus{
		sync.RWMutex{},
		value,
		poll,
		make(chan bool),
	}
	if err := s.poll(); err != nil {
		return nil, err
	}
	go func(s *PollingStatus) {
		ticker := time.Tick(frequency)
		for {
			select {
			case <-s.stop:
				return
			case <-ticker:
				if err := s.poll(); err != nil {
					glog.Error(err)
				}
			}
		}
	}(&s)
	return &s, nil
}

func main() {
	defer common.LogPanic()
	common.Init()
	frontend.MustInit()

	// Send start email.
	emailsArr := util.ParseEmails(*emails)
	emailsArr = append(emailsArr, util.CtAdmins...)
	if len(emailsArr) == 0 {
		glog.Error("At least one email address must be specified")
		return
	}
	skutil.LogErr(frontend.UpdateWebappTaskSetStarted(&admin_tasks.RecreateWebpageArchivesUpdateVars{}, *gaeTaskID))
	skutil.LogErr(util.SendTaskStartEmail(emailsArr, "Capture archives", util.GetMasterLogLink(*runID), ""))
	// Ensure webapp is updated and completion email is sent even if task fails.
	defer updateWebappTask()
	defer sendEmail(emailsArr)
	// Cleanup tmp files after the run.
	defer util.CleanTmpDir()
	// Finish with glog flush and how long the task took.
	defer util.TimeTrack(time.Now(), "Capture archives on Workers")
	defer glog.Flush()

	if *pagesetType == "" {
		glog.Error("Must specify --pageset_type")
		return
	}
	if *chromiumBuild == "" {
		glog.Error("Must specify --chromium_build")
		return
	}

	cmd := []string{
		fmt.Sprintf("cd %s;", util.CtTreeDir),
		"git pull;",
		"make all;",
		// The main command that runs capture_archives on all workers.
		fmt.Sprintf("DISPLAY=:0 capture_archives --worker_num=%s --log_dir=%s --log_id=%s --pageset_type=%s --chromium_build=%s;", util.WORKER_NUM_KEYWORD, util.GLogDir, *runID, *pagesetType, *chromiumBuild),
	}

	_, err := util.SSH(strings.Join(cmd, " "), util.Slaves, util.CAPTURE_ARCHIVES_TIMEOUT)
	if err != nil {
		glog.Errorf("Error while running cmd %s: %s", cmd, err)
		return
	}
	*taskCompletedSuccessfully = true
}

func statusJsonHandler(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "application/json")
	// Obtain the status info. Only display error messages if the user
	// is a logged-in Googler.
	status := arb.GetStatus(login.IsAGoogler(r))
	if err := json.NewEncoder(w).Encode(&status); err != nil {
		glog.Error(err)
	}
}

func main() {
	defer common.LogPanic()
	common.Init()

	if *emailTokenPath == "" {
		glog.Error("Must specify --email_token_path")
		return
	}

	defer glog.Flush()

	for _, shiftType := range allShiftTypes {

		res, err := http.Get(shiftType.nextSheriffEndpoint)
		if err != nil {
			glog.Fatalf("Could not HTTP Get: %s", err)
		}
		defer util.Close(res.Body)

		var jsonType map[string]interface{}
		if err := json.NewDecoder(res.Body).Decode(&jsonType); err != nil {
			glog.Fatalf("Could not unmarshal JSON: %s", err)
		}
		sheriffEmail, _ := jsonType["username"].(string)
		if sheriffEmail == NO_SHERIFF {
			glog.Infof("Skipping emailing %s because %s was specified", shiftType.shiftName, NO_SHERIFF)
			continue
		}
		sheriffUsername := strings.Split(string(sheriffEmail), "@")[0]

		emailTemplateParsed := template.Must(template.New("sheriff_email").Parse(EMAIL_TEMPLATE))
		emailBytes := new(bytes.Buffer)
		if err := emailTemplateParsed.Execute(emailBytes, struct {
			SheriffName      string
			SheriffType      string
			SheriffSchedules string
			SheriffDoc       string
			ScheduleStart    string
			ScheduleEnd      string
		}{
			SheriffName:      sheriffUsername,
			SheriffType:      shiftType.shiftName,
			SheriffSchedules: shiftType.schedulesLink,
			SheriffDoc:       shiftType.documentationLink,
			ScheduleStart:    jsonType["schedule_start"].(string),
			ScheduleEnd:      jsonType["schedule_end"].(string),
		}); err != nil {
			glog.Errorf("Failed to execute template: %s", err)
			return
		}

		emailSubject := fmt.Sprintf("%s is the next %s", sheriffUsername, shiftType.shiftName)
		if err := sendEmail([]string{sheriffEmail, EXTRA_RECIPIENT}, emailSubject, emailBytes.String()); err != nil {
			glog.Fatalf("Error sending email to sheriff: %s", err)
		}
	}
}

// FileExists returns true if the given path exists and false otherwise.
// If there is an error it will return false and log the error message.
func FileExists(path string) bool {
	if _, err := os.Stat(path); os.IsNotExist(err) {
		return false
	} else if err != nil {
		glog.Error(err)
		return false
	}
	return true
}

func main() {
	common.Init()
	webhook.MustInitRequestSaltFromFile(util.WebhookRequestSaltPath)

	// Send start email.
	emailsArr := util.ParseEmails(*emails)
	emailsArr = append(emailsArr, util.CtAdmins...)
	if len(emailsArr) == 0 {
		glog.Error("At least one email address must be specified")
		return
	}
	skutil.LogErr(frontend.UpdateWebappTaskSetStarted(&frontend.RecreatePageSetsUpdateVars{}, *gaeTaskID))
	skutil.LogErr(util.SendTaskStartEmail(emailsArr, "Creating pagesets"))
	// Ensure webapp is updated and completion email is sent even if task fails.
	defer updateWebappTask()
	defer sendEmail(emailsArr)
	// Cleanup tmp files after the run.
	defer util.CleanTmpDir()
	// Finish with glog flush and how long the task took.
	defer util.TimeTrack(time.Now(), "Creating Pagesets on Workers")
	defer glog.Flush()

	if *pagesetType == "" {
		glog.Error("Must specify --pageset_type")
		return
	}

	cmd := []string{
		fmt.Sprintf("cd %s;", util.CtTreeDir),
		"git pull;",
		"make all;",
		// The main command that runs create_pagesets on all workers.
		fmt.Sprintf("create_pagesets --worker_num=%s --log_dir=%s --pageset_type=%s;", util.WORKER_NUM_KEYWORD, util.GLogDir, *pagesetType),
	}

	// Setting a 4 hour timeout since it may take a while to upload page sets to
	// Google Storage when doing 10k page sets per worker.
	if _, err := util.SSH(strings.Join(cmd, " "), util.Slaves, 4*time.Hour); err != nil {
		glog.Errorf("Error while running cmd %s: %s", cmd, err)
		return
	}
	*taskCompletedSuccessfully = true
}

func main() {
	defer common.LogPanic()
	master_common.Init()

	// Send start email.
	emailsArr := util.ParseEmails(*emails)
	emailsArr = append(emailsArr, util.CtAdmins...)
	if len(emailsArr) == 0 {
		glog.Error("At least one email address must be specified")
		return
	}
	skutil.LogErr(frontend.UpdateWebappTaskSetStarted(&admin_tasks.RecreatePageSetsUpdateVars{}, *gaeTaskID))
	skutil.LogErr(util.SendTaskStartEmail(emailsArr, "Creating pagesets", util.GetMasterLogLink(*runID), ""))
	// Ensure webapp is updated and completion email is sent even if task fails.
	defer updateWebappTask()
	defer sendEmail(emailsArr)
	if !*master_common.Local {
		// Cleanup tmp files after the run.
		defer util.CleanTmpDir()
	}
	// Finish with glog flush and how long the task took.
	defer util.TimeTrack(time.Now(), "Creating Pagesets on Workers")
	defer glog.Flush()

	if *pagesetType == "" {
		glog.Error("Must specify --pageset_type")
		return
	}

	cmd := append(master_common.WorkerSetupCmds(),
		// The main command that runs create_pagesets on all workers.
		fmt.Sprintf(
			"create_pagesets --worker_num=%s --log_dir=%s --log_id=%s --pageset_type=%s --local=%t;",
			util.WORKER_NUM_KEYWORD, util.GLogDir, *runID, *pagesetType, *master_common.Local))

	_, err := util.SSH(strings.Join(cmd, " "), util.Slaves, util.CREATE_PAGESETS_TIMEOUT)
	if err != nil {
		glog.Errorf("Error while running cmd %s: %s", cmd, err)
		return
	}
	*taskCompletedSuccessfully = true
}

func mainHandler(w http.ResponseWriter, r *http.Request) {
	if _, err := w.Write([]byte(`
<html>
<body>
Hello World
</body>
</html>
`)); err != nil {
		glog.Error(err)
	}
}

// LastSkpCommit returns the time of the last change to the SKP_VERSION file.
func (g *GitInfo) LastSkpCommit() (time.Time, error) {
	// Executes a git log command that looks like:
	//
	// git log --format=format:%ct -n 1 SKP_VERSION
	//
	// The output should be a single unix timestamp.
	cmd := exec.Command("git", "log", "--format=format:%ct", "-n", "1", "SKP_VERSION")
	cmd.Dir = g.dir
	b, err := cmd.Output()
	if err != nil {
		glog.Error("Failed to read git log: ", err)
		return time.Time{}, err
	}
	ts, err := strconv.ParseInt(string(b), 10, 64)
	if err != nil {
		glog.Error("Failed to parse timestamp: ", string(b), err)
		return time.Time{}, err
	}
	return time.Unix(ts, 0), nil
}

func rulesJsonHandler(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "application/json")
	rules := struct {
		Rules []*rules.Rule `json:"rules"`
	}{
		Rules: rulesList,
	}
	if err := json.NewEncoder(w).Encode(&rules); err != nil {
		glog.Error(err)
	}
}

// ReadLinesFromPipe reads from the given pipe and logs its output.
func ReadLinesFromPipe(pipe io.Reader, lines chan string) {
	buf := []byte{}

	// writeLine recovers from a panic when writing to the channel.
	writeLine := func(s string) {
		defer func() {
			if r := recover(); r != nil {
				glog.Warningf("Panic writing to channel... are we exiting?")
			}
		}()
		lines <- s
	}

	// readLines reads output lines from the buffer and pushes them into the channel.
	readlines := func() {
		readIdx := 0
		// Read lines from buf.
		for i, b := range buf {
			if b == '\n' {
				s := string(buf[readIdx:i])
				writeLine(s)
				readIdx = i + 1
			}
		}
		// Remove the lines we read.
		buf = buf[readIdx:]
	}

	// Loop forever, reading bytes from the pipe.
	readbuf := make([]byte, 4096)
	for {
		read, err := pipe.Read(readbuf)
		if read > 0 {
			buf = append(buf, readbuf[:read]...)

			// Read lines from the buffers.
			readlines()
		}
		if err != nil {
			if err == io.EOF {
				break
			} else {
				glog.Error(err)
			}
		} else if read == 0 {
			time.Sleep(20 * time.Millisecond)
		}
	}
	// Read any remaining lines from the buffers.
	readlines()
	// "Flush" any incomplete lines from the buffers.
	writeLine(string(buf))
}

// JsonHandler is a pre-built handler for HTTP requests which returns version
// information in JSON format.
func JsonHandler(w http.ResponseWriter, r *http.Request) {
	w.Header().Set("Content-Type", "application/json")
	v, err := GetVersion()
	if err != nil {
		glog.Error(err)
		v = &Version{
			Commit: "(unknown)",
			Date:   time.Time{},
		}
	}
	if err := json.NewEncoder(w).Encode(v); err != nil {
		glog.Errorf("Failed to write or encode output: %s", err)
		return
	}
}

// Runs commonly-used initialization metrics.
func Init() {
	flag.Parse()
	defer glog.Flush()
	flag.VisitAll(func(f *flag.Flag) {
		glog.Infof("Flags: --%s=%v", f.Name, f.Value)
	})

	// See skbug.com/4386 for details on why the below section exists.
	glog.Info("Initializing logserver for log level INFO.")
	glog.Warning("Initializing logserver for log level WARNING.")
	glog.Error("Initializing logserver for log level ERROR.")

	// Use all cores.
	runtime.GOMAXPROCS(runtime.NumCPU())
}