cmd/relui, cmd/relworker: remove pubsub and datastore

This will make room for a new workflow definition, and a new datastore.

For golang/go#47401

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

Author: toothrot

cmd/relui/main.go

@@ -9,40 +9,17 @@
 package main
 
 import (
-	"context"
-	"flag"
-	"io/ioutil"
 	"log"
 	"net/http"
 	"os"
-	"path/filepath"
 
-	"cloud.google.com/go/datastore"
-	"cloud.google.com/go/pubsub"
-	"github.com/golang/protobuf/proto"
-	"github.com/googleapis/google-cloud-go-testing/datastore/dsiface"
-	reluipb "golang.org/x/build/cmd/relui/protos"
-	"google.golang.org/grpc/codes"
-	"google.golang.org/grpc/status"
-)
-
-var (
-	projectID = flag.String("project-id", os.Getenv("PUBSUB_PROJECT_ID"), "Pubsub project ID for communicating with workers. Uses PUBSUB_PROJECT_ID if unset.")
-	topicID   = flag.String("topic-id", "relui-development", "Pubsub topic ID for communicating with workers.")
+	"golang.org/x/build/internal/datastore/fake"
 )
 
 func main() {
-	flag.Parse()
-	ctx := context.Background()
-	dsc, err := datastore.NewClient(ctx, *projectID)
-	if err != nil {
-		log.Fatalf("datastore.NewClient(_, %q) = _, %v, wanted no error", *projectID, err)
-	}
-	d := &dsStore{client: dsiface.AdaptClient(dsc)}
+	d := &dsStore{client: &fake.Client{}}
 	s := &server{
-		configs: loadWorkflowConfig("./workflows"),
-		store:   d,
-		topic:   getTopic(ctx),
+		store: d,
 	}
 	http.Handle("/workflows/create", http.HandlerFunc(s.createWorkflowHandler))
 	http.Handle("/workflows/new", http.HandlerFunc(s.newWorkflowHandler))
@@ -55,44 +32,3 @@ func main() {
 	log.Printf("Listening on :" + port)
 	log.Fatal(http.ListenAndServe(":"+port, http.DefaultServeMux))
 }
-
-// getTopic creates and returns a pubsub topic from the project specified in projectId, which is to be used for
-// communicating with relui workers.
-//
-// It is safe to call if a topic already exists. A reference to the topic will be returned.
-func getTopic(ctx context.Context) *pubsub.Topic {
-	client, err := pubsub.NewClient(ctx, *projectID)
-	if err != nil {
-		log.Fatalf("pubsub.NewClient(_, %q) = %v, wanted no error", *projectID, err)
-	}
-	_, err = client.CreateTopic(ctx, *topicID)
-	if err != nil && status.Code(err) != codes.AlreadyExists {
-		log.Fatalf("client.CreateTopic(_, %q) = %v, wanted no error", *topicID, err)
-	}
-	return client.Topic(*topicID)
-}
-
-// loadWorkflowConfig loads Workflow configuration files from dir. It expects all files to be in textproto format.
-func loadWorkflowConfig(dir string) []*reluipb.Workflow {
-	fs, err := filepath.Glob(filepath.Join(relativeFile(dir), "*.textpb"))
-	if err != nil {
-		log.Fatalf("Error perusing %q for configuration", filepath.Join(dir, "*.textpb"))
-	}
-	if len(fs) == 0 {
-		log.Println("No workflow configuration found.")
-	}
-	var ws []*reluipb.Workflow
-	for _, f := range fs {
-		b, err := ioutil.ReadFile(f)
-		if err != nil {
-			log.Printf("ioutil.ReadFile(%q) = _, %v, wanted no error", f, err)
-		}
-		w := new(reluipb.Workflow)
-		if err = proto.UnmarshalText(string(b), w); err != nil {
-			log.Printf("Error unmarshalling Workflow from %q: %v", f, err)
-			continue
-		}
-		ws = append(ws, w)
-	}
-	return ws
-}

cmd/relui/web.go

@@ -19,7 +19,6 @@ import (
 	"path"
 	"path/filepath"
 
-	"cloud.google.com/go/pubsub"
 	"github.com/golang/protobuf/proto"
 	"github.com/google/uuid"
 	reluipb "golang.org/x/build/cmd/relui/protos"
@@ -56,9 +55,6 @@ type server struct {
 
 	// store is for persisting application state.
 	store store
-
-	// topic is for communicating with relui workers.
-	topic *pubsub.Topic
 }
 
 type homeResponse struct {
@@ -119,24 +115,11 @@ func (s *server) createWorkflowHandler(w http.ResponseWriter, r *http.Request) {
 }
 
 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)
 }
 

cmd/relui/web_test.go

@@ -8,7 +8,6 @@
 package main
 
 import (
-	"context"
 	"embed"
 	"io/ioutil"
 	"net/http"
@@ -17,12 +16,8 @@ import (
 	"strings"
 	"testing"
 
-	"cloud.google.com/go/pubsub"
-	"cloud.google.com/go/pubsub/pstest"
 	reluipb "golang.org/x/build/cmd/relui/protos"
 	"golang.org/x/build/internal/datastore/fake"
-	"google.golang.org/api/option"
-	"google.golang.org/grpc"
 )
 
 // testStatic is our static web server content.
@@ -182,41 +177,8 @@ func TestServerCreateWorkflowHandler(t *testing.T) {
 	}
 }
 
-// newPSTest creates a new pstest.Server and returns a pubsub.Client connected to it and the server.
-//
-// cleanup will close the client, connection, and test server.
-func newPSTest(ctx context.Context, t *testing.T) (c *pubsub.Client, s *pstest.Server, cleanup func()) {
-	t.Helper()
-	s = pstest.NewServer()
-	conn, err := grpc.DialContext(ctx, s.Addr, grpc.WithInsecure())
-	if err != nil {
-		s.Close()
-		t.Fatalf("grpc.DialContext(_, %q, %v) = _, %v, wanted no error", s.Addr, grpc.WithInsecure(), err)
-	}
-	c, err = pubsub.NewClient(ctx, "relui-test", option.WithGRPCConn(conn))
-	if err != nil {
-		s.Close()
-		conn.Close()
-		t.Fatalf("pubsub.NewClient(_, %q, %v) = _, %v, wanted no error", "relui-test", option.WithGRPCConn(conn), err)
-	}
-	return c, s, func() {
-		c.Close()
-		conn.Close()
-		s.Close()
-	}
-}
-
 func TestServerStartTaskHandler(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-	client, pssrv, cleanup := newPSTest(ctx, t)
-	defer cleanup()
-	topic, err := client.CreateTopic(ctx, "relui-test-topic")
-	if err != nil {
-		t.Fatalf("client.CreateTopic(_, %q) = _, %v", "relui-test-topic", err)
-	}
-
-	s := server{store: &dsStore{client: &fake.Client{}}, topic: topic}
+	s := server{store: &dsStore{client: &fake.Client{}}}
 	wf := &reluipb.Workflow{
 		Id:   "someworkflow",
 		Name: "test_workflow",
@@ -226,14 +188,9 @@ func TestServerStartTaskHandler(t *testing.T) {
 			Id:       "sometask",
 		}},
 	}
-	if s.store.AddWorkflow(wf) != nil {
+	if err := s.store.AddWorkflow(wf); err != nil {
 		t.Fatalf("store.AddWorkflow(%v) = %v, wanted no error", wf, err)
 	}
-	want := &reluipb.StartBuildableTaskRequest{
-		WorkflowId:        "someworkflow",
-		BuildableTaskId:   "sometask",
-		BuildableTaskType: "TestTask",
-	}
 	params := url.Values{"workflow.id": []string{"someworkflow"}, "task.id": []string{"sometask"}}
 	req := httptest.NewRequest(http.MethodPost, "/tasks/start", strings.NewReader(params.Encode()))
 	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
@@ -248,13 +205,6 @@ func TestServerStartTaskHandler(t *testing.T) {
 	if resp.Header.Get("Location") != "/" {
 		t.Errorf("resp.Header.Get(%q) = %q, wanted %q", "Location", resp.Header.Get("Location"), "/")
 	}
-	if len(pssrv.Messages()) != 1 {
-		t.Fatalf("len(pssrv.Messages()) = %d, wanted %d", len(pssrv.Messages()), 1)
-	}
-	msg := pssrv.Messages()[0]
-	if string(msg.Data) != want.String() {
-		t.Errorf("msg.Data = %q, wanted %q", string(msg.Data), "hello world")
-	}
 }
 
 func TestStartTaskHandlerErrors(t *testing.T) {
@@ -278,27 +228,11 @@ func TestStartTaskHandlerErrors(t *testing.T) {
 			params:   url.Values{"workflow.id": []string{"someworkflow"}, "task.id": []string{"notexist"}},
 			wantCode: http.StatusNotFound,
 		},
-		{
-			desc:     "pubsub publish failure",
-			params:   url.Values{"workflow.id": []string{"someworkflow"}, "task.id": []string{"sometask"}},
-			wantCode: http.StatusInternalServerError,
-		},
 	}
 	for _, c := range cases {
 		t.Run(c.desc, func(t *testing.T) {
-			ctx, cancel := context.WithCancel(context.Background())
-			defer cancel()
-			client, pssrv, cleanup := newPSTest(ctx, t)
-			defer cleanup()
-			topic, err := client.CreateTopic(ctx, "relui-test-topic")
-			if err != nil {
-				t.Fatalf("client.CreateTopic(_, %q) = _, %v", "relui-test-topic", err)
-			}
-			// Simulate pubsub failure by stopping publishing.
-			topic.Stop()
-
-			s := server{store: &dsStore{client: &fake.Client{}}, topic: topic}
-			if s.store.AddWorkflow(wf) != nil {
+			s := server{store: &dsStore{client: &fake.Client{}}}
+			if err := s.store.AddWorkflow(wf); err != nil {
 				t.Fatalf("store.AddWorkflow(%v) = %v, wanted no error", wf, err)
 			}
 			req := httptest.NewRequest(http.MethodPost, "/tasks/start", strings.NewReader(c.params.Encode()))
@@ -311,9 +245,6 @@ func TestStartTaskHandlerErrors(t *testing.T) {
 			if resp.StatusCode != c.wantCode {
 				t.Errorf("resp.StatusCode = %d, wanted %d", resp.StatusCode, c.wantCode)
 			}
-			if len(pssrv.Messages()) != 0 {
-				t.Fatalf("len(pssrv.Messages()) = %d, wanted %d", len(pssrv.Messages()), 0)
-			}
 		})
 	}
 }

cmd/relui/workflows/local_go_release.textpb

@@ -6,7 +6,6 @@ require (
 	cloud.google.com/go v0.65.0
 	cloud.google.com/go/bigquery v1.8.0
 	cloud.google.com/go/datastore v1.1.0
-	cloud.google.com/go/pubsub v1.3.1
 	cloud.google.com/go/storage v1.10.0
 	contrib.go.opencensus.io/exporter/prometheus v0.3.0
 	contrib.go.opencensus.io/exporter/stackdriver v0.13.5

cmd/relworker/README.md

@@ -29,7 +29,6 @@ cloud.google.com/go/datastore v1.1.0/go.mod h1:umbIZjpQpHh4hmRpGhH4tLFup+FVzqBi1
 cloud.google.com/go/pubsub v1.0.1/go.mod h1:R0Gpsv3s54REJCy4fxDixWD93lHJMoZTyQ2kNxGRt3I=
 cloud.google.com/go/pubsub v1.1.0/go.mod h1:EwwdRX2sKPjnvnqCa270oGRyludottCI76h+R3AArQw=
 cloud.google.com/go/pubsub v1.2.0/go.mod h1:jhfEVHT8odbXTkndysNHCcx0awwzvfOlguIAii9o8iA=
-cloud.google.com/go/pubsub v1.3.1 h1:ukjixP1wl0LpnZ6LWtZJ0mX5tBmjp1f8Sqer8Z2OMUU=
 cloud.google.com/go/pubsub v1.3.1/go.mod h1:i+ucay31+CNRpDW4Lu78I4xXG+O1r/MAHgjpRVR+TSU=
 cloud.google.com/go/storage v1.0.0/go.mod h1:IhtSnM/ZTZV8YYJWCY8RULGVqBDmpoyjwiyrjsg+URw=
 cloud.google.com/go/storage v1.5.0/go.mod h1:tpKbwo567HUNpVclU5sGELwQWBDZ8gh0ZeosJ0Rtdos=