cmd/relui: publish task start message to pubsub

This change adds a new handler for starting a task by sending a message
to pubsub. A test pubsub server is used in unit testing. The exact
message used to start a task, as well as updating a task to mark it as
started, is still under design and likely to change dramatically. This
change is intended to unblock worker development.

For golang/go#40279

Co-authored-by: Carlos Amedee <[email protected]>
Change-Id: I59bb5c5261a9a5d17e52597c1835a2a980cf91f8
Reviewed-on: https://go-review.googlesource.com/c/build/+/257239
Trust: Alexander Rakoczy <[email protected]>
Run-TryBot: Alexander Rakoczy <[email protected]>
TryBot-Result: Go Bot <[email protected]>
Reviewed-by: Dmitri Shuralyov <[email protected]>

cmd/relui/main.go

@@ -41,6 +41,7 @@ func main() {
  }
  http.Handle("/workflows/create", http.HandlerFunc(s.createWorkflowHandler))
  http.Handle("/workflows/new", http.HandlerFunc(s.newWorkflowHandler))
+ http.Handle("/tasks/start", http.HandlerFunc(s.startTaskHandler))
  http.Handle("/", fileServerHandler(relativeFile("./static"), http.HandlerFunc(s.homeHandler)))
  port := os.Getenv("PORT")
  if port == "" {

cmd/relui/protos/relui.proto

@@ -47,6 +47,18 @@ message BuildableTask {
  string id = 7;
 }
 
+// StartBuildableTaskRequest is a message sent to workers to start working on a BuildableTask for a Workflow.
+message StartBuildableTaskRequest {
+ // workflow_id is the workflow to which the BuildableTask belongs.
+ string workflow_id = 1;
+
+ // buildable_task_id is the id of the BuildableTask to be started.
+ string buildable_task_id = 2;
+
+ // buildable_task_type is the type of the BuildableTask to be started.
+ string buildable_task_type = 3;
+}
+
 // LocalStorage is the persisted data of relui. It is used in development mode for saving application state.
 message LocalStorage {
  // workflows are a list of user-created workflows.

cmd/relui/store.go

@@ -17,8 +17,10 @@ import (
 
 // store is a persistence adapter for saving data.
 type store interface {
- Workflows() []*reluipb.Workflow
  AddWorkflow(workflow *reluipb.Workflow) error
+ BuildableTask(workflowId, id string) *reluipb.BuildableTask
+ Workflow(id string) *reluipb.Workflow
+ Workflows() []*reluipb.Workflow
 }
 
 var _ store = (*fileStore)(nil)
@@ -57,11 +59,33 @@ func (f *fileStore) AddWorkflow(w *reluipb.Workflow) error {
  return nil
 }
 
-// Workflows returns all workflows stored.
+// Workflows returns all reluipb.Workflows stored.
 func (f *fileStore) Workflows() []*reluipb.Workflow {
  return f.localStorage().GetWorkflows()
 }
 
+// Workflow returns a single reluipb.Workflow found by its id. If it is not found, it returns nil.
+func (f *fileStore) Workflow(id string) *reluipb.Workflow {
+ for _, w := range f.Workflows() {
+ if w.GetId() == id {
+ return w
+ }
+ }
+ return nil
+}
+
+// BuildableTask returns a single reluipb.BuildableTask found by the reluipb.Workflow id and its id.
+// If it is not found, it returns nil.
+func (f *fileStore) BuildableTask(workflowId, id string) *reluipb.BuildableTask {
+ wf := f.Workflow(workflowId)
+ for _, t := range wf.GetBuildableTasks() {
+ if t.GetId() == id {
+ return t
+ }
+ }
+ return nil
+}
+
 // localStorage returns a deep copy of data stored in fileStore.
 func (f *fileStore) localStorage() *reluipb.LocalStorage {
  f.mu.Lock()

cmd/relui/web.go

@@ -122,6 +122,28 @@ func (s *server) createWorkflowHandler(w http.ResponseWriter, r *http.Request) {
  http.Redirect(w, r, "/", http.StatusSeeOther)
 }
 
+func (s *server) startTaskHandler(w http.ResponseWriter, r *http.Request) {
+ wf := s.store.Workflow(r.PostFormValue("workflow.id"))
+ bt := s.store.BuildableTask(r.PostFormValue("workflow.id"), r.PostFormValue("task.id"))
+ if bt == nil {
+ http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound)
+ return
+ }
+ res := s.topic.Publish(r.Context(), &pubsub.Message{
+ Data: []byte((&reluipb.StartBuildableTaskRequest{
+ WorkflowId: wf.GetId(),
+ BuildableTaskId: bt.GetId(),
+ BuildableTaskType: bt.GetTaskType(),
+ }).String()),
+ })
+ if _, err := res.Get(r.Context()); err != nil {
+ log.Printf("Error publishing task start: %v", err)
+ http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
+ return
+ }
+ http.Redirect(w, r, "/", http.StatusSeeOther)
+}
+
 // relativeFile returns the path to the provided file or directory,
 // conditionally prepending a relative path depending on the environment.
 //