internal/relui: email on schedule failure

The user who scheduled a workflow may not be closely watching at the
time of failure. This adds functionality to send an email if a scheduled
workflow cannot progress.

The current implementation logs on stall, but it's trivial to
configure to email as some tasks do.

For golang/go#54476

Change-Id: Id0deefd3c1b07f569585600a583ba4e04f8f7be1
Reviewed-on: https://go-review.googlesource.com/c/build/+/444695
Run-TryBot: Jenny Rakoczy <[email protected]>
TryBot-Result: Gopher Robot <[email protected]>
Reviewed-by: Heschi Kreinick <[email protected]>

Author: toothrot

cmd/relui/main.go

@@ -73,6 +73,10 @@ func main() {
 	addressVarFlag(&annMail.From, "announce-mail-from", "The From address to use for the (pre-)announcement mail.")
 	addressVarFlag(&annMail.To, "announce-mail-to", "The To address to use for the (pre-)announcement mail.")
 	addressListVarFlag(&annMail.BCC, "announce-mail-bcc", "The BCC address list to use for the (pre-)announcement mail.")
+	var schedMail task.MailHeader
+	addressVarFlag(&schedMail.From, "schedule-mail-from", "The From address to use for the scheduled workflow failure mail.")
+	addressVarFlag(&schedMail.To, "schedule-mail-to", "The To address to use for the the scheduled workflow failure mail.")
+	addressListVarFlag(&schedMail.BCC, "schedule-mail-bcc", "The BCC address list to use for the scheduled workflow failure mail.")
 	var twitterAPI secret.TwitterCredentials
 	secret.JSONVarFlag(&twitterAPI, "twitter-api-secret", "Twitter API secret to use for workflows involving tweeting.")
 	masterKey := secret.Flag("builder-master-key", "Builder master key")
@@ -212,18 +216,24 @@ func main() {
 	}
 	dh.RegisterDefinition("Tag x/ repos", tagTasks.NewDefinition())
 
-	w := relui.NewWorker(dh, dbPool, relui.NewPGListener(dbPool))
-	go w.Run(ctx)
-	if err := w.ResumeAll(ctx); err != nil {
-		log.Printf("w.ResumeAll() = %v", err)
-	}
 	var base *url.URL
 	if *baseURL != "" {
 		base, err = url.Parse(*baseURL)
 		if err != nil {
 			log.Fatalf("url.Parse(%q) = %v, %v", *baseURL, base, err)
 		}
 	}
+	l := &relui.PGListener{
+		DB:                        dbPool,
+		BaseURL:                   base,
+		ScheduleFailureMailHeader: schedMail,
+		SendMail:                  relui.LogOnlyMailer,
+	}
+	w := relui.NewWorker(dh, dbPool, l)
+	go w.Run(ctx)
+	if err := w.ResumeAll(ctx); err != nil {
+		log.Printf("w.ResumeAll() = %v", err)
+	}
 	s := relui.NewServer(dbPool, w, base, siteHeader, ms)
 	log.Fatalln(https.ListenAndServe(ctx, &ochttp.Handler{Handler: GRPCHandler(grpcServer, s)}))
 }

internal/relui/buildrelease_test.go

@@ -544,6 +544,11 @@ type verboseListener struct {
 	outputListener func(string, interface{})
 }
 
+func (l *verboseListener) WorkflowStalled(workflowID uuid.UUID) error {
+	l.t.Logf("workflow %q: stalled", workflowID.String())
+	return nil
+}
+
 func (l *verboseListener) TaskStateChanged(_ uuid.UUID, _ string, st *workflow.TaskState) error {
 	switch {
 	case !st.Finished:

internal/relui/content.go

@@ -4,7 +4,9 @@
 
 package relui
 
-import "embed"
+import (
+	"embed"
+)
 
 // static is our static web server content.
 //

internal/relui/listener.go

@@ -5,26 +5,51 @@
 package relui
 
 import (
+	"bytes"
 	"context"
 	"database/sql"
 	"encoding/json"
 	"fmt"
+	"html/template"
 	"log"
+	"net/url"
 	"time"
 
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v4"
 	"golang.org/x/build/internal/relui/db"
+	"golang.org/x/build/internal/task"
 	"golang.org/x/build/internal/workflow"
 )
 
-func NewPGListener(db db.PGDBTX) *PGListener {
-	return &PGListener{db}
-}
-
 // PGListener implements workflow.Listener for recording workflow state.
 type PGListener struct {
-	db db.PGDBTX
+	DB db.PGDBTX
+
+	BaseURL *url.URL
+
+	ScheduleFailureMailHeader task.MailHeader
+	SendMail                  func(task.MailHeader, task.MailContent) error
+
+	templ *template.Template
+}
+
+// WorkflowStalled is called when no tasks are runnable.
+func (l *PGListener) WorkflowStalled(workflowID uuid.UUID) error {
+	wf, err := db.New(l.DB).Workflow(context.Background(), workflowID)
+	if err != nil || wf.ScheduleID.Int32 == 0 {
+		return err
+	}
+	var buf bytes.Buffer
+	body := scheduledFailureEmailBody{Workflow: wf}
+	if err := l.template("scheduled_workflow_failure_email.txt").Execute(&buf, body); err != nil {
+		log.Printf("WorkflowFinished: Execute(_, %v) = %q", body, err)
+		return err
+	}
+	return l.SendMail(l.ScheduleFailureMailHeader, task.MailContent{
+		Subject:  fmt.Sprintf("[relui] Scheduled workflow %q failed", wf.Name.String),
+		BodyText: buf.String(),
+	})
 }
 
 // TaskStateChanged is called whenever a task is updated by the
@@ -38,7 +63,7 @@ func (l *PGListener) TaskStateChanged(workflowID uuid.UUID, taskName string, sta
 	if err != nil {
 		return err
 	}
-	err = l.db.BeginFunc(ctx, func(tx pgx.Tx) error {
+	err = l.DB.BeginFunc(ctx, func(tx pgx.Tx) error {
 		q := db.New(tx)
 		updated := time.Now()
 		_, err := q.UpsertTask(ctx, db.UpsertTaskParams{
@@ -62,7 +87,7 @@ func (l *PGListener) TaskStateChanged(workflowID uuid.UUID, taskName string, sta
 
 // WorkflowStarted persists a new workflow execution in the database.
 func (l *PGListener) WorkflowStarted(ctx context.Context, workflowID uuid.UUID, name string, params map[string]interface{}, scheduleID int) error {
-	q := db.New(l.db)
+	q := db.New(l.DB)
 	m, err := json.Marshal(params)
 	if err != nil {
 		return err
@@ -80,11 +105,16 @@ func (l *PGListener) WorkflowStarted(ctx context.Context, workflowID uuid.UUID,
 	return err
 }
 
+type scheduledFailureEmailBody struct {
+	Workflow db.Workflow
+	Err      error
+}
+
 // WorkflowFinished saves the final state of a workflow after its run
 // has completed.
 func (l *PGListener) WorkflowFinished(ctx context.Context, workflowID uuid.UUID, outputs map[string]interface{}, workflowErr error) error {
-	log.Printf("WorkflowCompleted(%q, %v, %q)", workflowID, outputs, workflowErr)
-	q := db.New(l.db)
+	log.Printf("WorkflowFinished(%q, %v, %q)", workflowID, outputs, workflowErr)
+	q := db.New(l.DB)
 	m, err := json.Marshal(outputs)
 	if err != nil {
 		return err
@@ -102,9 +132,21 @@ func (l *PGListener) WorkflowFinished(ctx context.Context, workflowID uuid.UUID,
 	return err
 }
 
+func (l *PGListener) template(name string) *template.Template {
+	if l.templ == nil {
+		helpers := map[string]any{"baseLink": l.baseLink}
+		l.templ = template.Must(template.New("").Funcs(helpers).ParseFS(templates, "templates/*.txt"))
+	}
+	return l.templ.Lookup(name)
+}
+
+func (l *PGListener) baseLink(target string, extras ...string) string {
+	return BaseLink(l.BaseURL)(target, extras...)
+}
+
 func (l *PGListener) Logger(workflowID uuid.UUID, taskName string) workflow.Logger {
 	return &postgresLogger{
-		db:         l.db,
+		db:         l.DB,
 		workflowID: workflowID,
 		taskName:   taskName,
 	}
@@ -136,3 +178,8 @@ func (l *postgresLogger) Printf(format string, v ...interface{}) {
 		log.Printf("l.Printf(%q, %v) = %v", format, v, err)
 	}
 }
+
+func LogOnlyMailer(header task.MailHeader, content task.MailContent) error {
+	log.Println("Logging but not sending mail:", header, content)
+	return nil
+}

internal/relui/listener_test.go

@@ -7,13 +7,17 @@ package relui
 import (
 	"context"
 	"database/sql"
+	"errors"
+	"fmt"
+	"net/mail"
 	"testing"
 	"time"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/google/go-cmp/cmp/cmpopts"
 	"github.com/google/uuid"
 	"golang.org/x/build/internal/relui/db"
+	"golang.org/x/build/internal/task"
 	"golang.org/x/build/internal/workflow"
 )
 
@@ -76,7 +80,7 @@ func TestListenerTaskStateChanged(t *testing.T) {
 				t.Fatalf("q.CreateWorkflow(%v, %v) = %v, wanted no error", ctx, wfp, err)
 			}
 
-			l := &PGListener{db: dbp}
+			l := &PGListener{DB: dbp}
 			err = l.TaskStateChanged(wf.ID, "TestTask", c.state)
 			if err != nil {
 				t.Fatalf("l.TaskStateChanged(%v, %q, %v) = %v, wanted no error", wf.ID, "TestTask", c.state, err)
@@ -110,7 +114,7 @@ func TestListenerLogger(t *testing.T) {
 		t.Fatalf("q.UpsertTask(%v, %v) = %v, wanted no error", ctx, params, err)
 	}
 
-	l := &PGListener{db: dbp}
+	l := &PGListener{DB: dbp}
 	l.Logger(wf.ID, "TestTask").Printf("A fancy log line says %q", "hello")
 
 	logs, err := q.TaskLogs(ctx)
@@ -128,3 +132,99 @@ func TestListenerLogger(t *testing.T) {
 		t.Errorf("q.TaskLogs(_, %q) mismatch (-want +got):\n%s", wf.ID, diff)
 	}
 }
+
+func TestPGListenerWorkflowStalledNotification(t *testing.T) {
+	cases := []struct {
+		desc     string
+		schedule bool
+		taskErr  bool
+	}{
+		{
+			desc:     "scheduled workflow failure sends notification",
+			schedule: true,
+			taskErr:  true,
+		},
+		{
+			desc:     "scheduled workflow success sends nothing",
+			schedule: true,
+		},
+		{
+			desc: "unscheduled workflow success sends nothing",
+		},
+		{
+			desc:    "unscheduled workflow failure sends nothing",
+			taskErr: true,
+		},
+	}
+	for _, c := range cases {
+		t.Run(c.desc, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			p := testDB(ctx, t)
+			var schedID int
+			if c.schedule {
+				sched, err := db.New(p).CreateSchedule(ctx, db.CreateScheduleParams{WorkflowName: c.desc})
+				if err != nil {
+					t.Fatalf("CreateSchedule() = %v, wanted no error", err)
+				}
+				schedID = int(sched.ID)
+			}
+			wd := workflow.New()
+			complete := workflow.Task0(wd, "complete", func(ctx context.Context) (string, error) {
+				if c.taskErr {
+					return "", fmt.Errorf("c.taskErr: %t", c.taskErr)
+				}
+				return "done", nil
+			})
+			workflow.Output(wd, "finished", complete)
+			dh := NewDefinitionHolder()
+			dh.RegisterDefinition(c.desc, wd)
+
+			header := task.MailHeader{
+				From: mail.Address{Address: "[email protected]"},
+				To:   mail.Address{Address: "[email protected]"},
+			}
+			var gotHeader task.MailHeader
+			var gotContent task.MailContent
+			pgl := &PGListener{
+				DB: p,
+				SendMail: func(h task.MailHeader, c task.MailContent) error {
+					gotHeader, gotContent = h, c
+					return nil
+				},
+				ScheduleFailureMailHeader: header,
+			}
+			listener := &testWorkflowListener{
+				Listener:   pgl,
+				onFinished: cancel,
+			}
+			w := NewWorker(dh, p, listener)
+
+			id, err := w.StartWorkflow(ctx, c.desc, nil, schedID)
+			if err != nil {
+				t.Fatalf("w.StartWorkflow(_, %q, %v, %d) = %v, %v, wanted no error", c.desc, nil, schedID, id, err)
+			}
+			listener.onStalled = func() {
+				w.cancelWorkflow(id)
+			}
+			if err := w.Run(ctx); !errors.Is(err, context.Canceled) {
+				t.Errorf("w.Run() = %v, wanted %v", err, context.Canceled)
+			}
+
+			wantSend := c.taskErr && c.schedule
+			if (gotContent.Subject == "") == wantSend {
+				t.Errorf("gotContent.Subject = %q, wanted empty: %t", gotContent.Subject, !c.taskErr)
+			}
+			if (gotContent.BodyText == "") == wantSend {
+				t.Errorf("gotContent.BodyText = %q, wanted empty: %t", gotContent.BodyText, !c.taskErr)
+			}
+			var wantHeader task.MailHeader
+			if wantSend {
+				wantHeader = header
+			}
+			if diff := cmp.Diff(wantHeader, gotHeader); diff != "" {
+				t.Errorf("WorkflowFinished(_, %q) mismatch (-want +got):\n%s", id, diff)
+			}
+		})
+	}
+}

internal/relui/schedule_test.go

@@ -120,7 +120,7 @@ func TestSchedulerCreate(t *testing.T) {
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
 			p := testDB(ctx, t)
-			s := NewScheduler(p, NewWorker(NewDefinitionHolder(), p, &PGListener{p}))
+			s := NewScheduler(p, NewWorker(NewDefinitionHolder(), p, &PGListener{DB: p}))
 			row, err := s.Create(ctx, c.sched, c.workflowName, c.params)
 			if (err != nil) != c.wantErr {
 				t.Fatalf("s.Create(_, %v, %q, %v) = %v, %v, wantErr: %t", c.sched, c.workflowName, c.params, row, err, c.wantErr)
@@ -243,7 +243,7 @@ func TestSchedulerResume(t *testing.T) {
 			defer cancel()
 			p := testDB(ctx, t)
 			q := db.New(p)
-			s := NewScheduler(p, NewWorker(NewDefinitionHolder(), p, &PGListener{p}))
+			s := NewScheduler(p, NewWorker(NewDefinitionHolder(), p, &PGListener{DB: p}))
 
 			for _, csp := range c.scheds {
 				if _, err := q.CreateSchedule(ctx, csp); err != nil {
@@ -294,7 +294,7 @@ func TestScheduleDelete(t *testing.T) {
 			defer cancel()
 			p := testDB(ctx, t)
 			q := db.New(p)
-			s := NewScheduler(p, NewWorker(NewDefinitionHolder(), p, &PGListener{p}))
+			s := NewScheduler(p, NewWorker(NewDefinitionHolder(), p, &PGListener{DB: p}))
 			row, err := s.Create(ctx, c.sched, c.workflowName, c.params)
 			if err != nil {
 				t.Fatalf("s.Create(_, %v, %q, %v) = %v, %v, wanted no error", c.sched, c.workflowName, c.params, row, err)

internal/relui/templates/scheduled_workflow_failure_email.txt

@@ -0,0 +1,6 @@
+{{- /*gotype: golang.org/x/build/internal/relui.scheduledFailureEmailBody*/ -}}
+The scheduled workflow "{{.Workflow.Name.String}}" ({{.Workflow.ID.String}}) failed at {{.Workflow.UpdatedAt.UTC.Format "2006/01/02 15:04 MST"}}.
+
+See the following page for details:
+
+{{baseLink "/workflows/" .Workflow.ID.String}}