internal/frontend: add queue

The frontend server is not initiated with a queue. The frontend task
queue will be used to support frontend fetches.

frontend.FetchAndUpdateState is added, which is a copy of
worker.FetchAndUpdateState for use in testing and locally.

Updates golang/go#36811
Updates golang/go#37002
Updates golang/go#37106

Change-Id: I41922d30462d2623a061aa1f207bb2b39f7b54e2
Reviewed-on: https://team-review.git.corp.google.com/c/golang/discovery/+/743102
Reviewed-by: Jonathan Amsterdam <[email protected]>

Author: julieqiu

cmd/frontend/main.go

@@ -1,7 +1,6 @@
 // Copyright 2019 The Go Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style
 // license that can be found in the LICENSE file.
-
 package main
 
 import (
@@ -14,6 +13,7 @@ import (
 	"strings"
 	"time"
 
+	cloudtasks "cloud.google.com/go/cloudtasks/apiv2"
 	"cloud.google.com/go/profiler"
 	"contrib.go.opencensus.io/integrations/ocsql"
 	"github.com/go-redis/redis/v7"
@@ -28,42 +28,44 @@ import (
 	"golang.org/x/pkgsite/internal/postgres"
 	"golang.org/x/pkgsite/internal/proxy"
 	"golang.org/x/pkgsite/internal/proxydatasource"
+	"golang.org/x/pkgsite/internal/queue"
+	"golang.org/x/pkgsite/internal/source"
 )
 
 var (
+	queueName       = config.GetEnv("GO_DISCOVERY_FRONTEND_TASK_QUEUE", "")
 	staticPath      = flag.String("static", "content/static", "path to folder containing static files served")
 	thirdPartyPath  = flag.String("third_party", "third_party", "path to folder containing third-party libraries")
 	reloadTemplates = flag.Bool("reload_templates", false, "reload templates on each page load (to be used during development)")
-	directProxy     = flag.String("direct_proxy", "", "if set to a valid URL, uses the module proxy referred to by this URL "+
+	proxyURL        = flag.String("proxy_url", "https://proxy.golang.org", "Uses the module proxy referred to by this URL "+
+		"for direct proxy mode and frontend fetches")
+	directProxy = flag.Bool("direct_proxy", false, "if set to true, uses the module proxy referred to by this URL "+
 		"as a direct backend, bypassing the database")
 )
 
 func main() {
 	flag.Parse()
-
 	ctx := context.Background()
-
 	cfg, err := config.Init(ctx)
 	if err != nil {
 		log.Fatal(ctx, err)
 	}
 	cfg.Dump(os.Stderr)
-
 	if cfg.UseProfiler {
 		if err := profiler.Start(profiler.Config{}); err != nil {
 			log.Fatalf(ctx, "profiler.Start: %v", err)
 		}
 	}
-
 	var (
-		ds  internal.DataSource
-		exp internal.ExperimentSource
+		ds         internal.DataSource
+		exp        internal.ExperimentSource
+		fetchQueue queue.Queue
 	)
-	if *directProxy != "" {
-		proxyClient, err := proxy.New(*directProxy)
-		if err != nil {
-			log.Fatal(ctx, err)
-		}
+	proxyClient, err := proxy.New(*proxyURL)
+	if err != nil {
+		log.Fatal(ctx, err)
+	}
+	if *directProxy {
 		ds = proxydatasource.New(proxyClient)
 		exp = internal.NewLocalExperimentSource(readLocalExperiments(ctx))
 	} else {
@@ -80,14 +82,16 @@ func main() {
 		defer db.Close()
 		ds = db
 		exp = db
+		sourceClient := source.NewClient(config.SourceTimeout)
+		fetchQueue = newQueue(ctx, cfg, proxyClient, sourceClient, db)
 	}
 	var haClient *redis.Client
 	if cfg.RedisHAHost != "" {
 		haClient = redis.NewClient(&redis.Options{
 			Addr: cfg.RedisHAHost + ":" + cfg.RedisHAPort,
 		})
 	}
-	server, err := frontend.NewServer(ds, haClient, *staticPath, *thirdPartyPath, *reloadTemplates)
+	server, err := frontend.NewServer(ds, fetchQueue, haClient, *staticPath, *thirdPartyPath, *reloadTemplates)
 	if err != nil {
 		log.Fatalf(ctx, "frontend.NewServer: %v", err)
 	}
@@ -99,7 +103,6 @@ func main() {
 		})
 	}
 	server.Install(router.Handle, cacheClient)
-
 	views := append(dcensus.ServerViews,
 		postgres.SearchLatencyDistribution,
 		postgres.SearchResponseCount,
@@ -119,7 +122,6 @@ func main() {
 		}
 		go http.ListenAndServe(cfg.DebugAddr("localhost:8081"), dcensusServer)
 	}
-
 	panicHandler, err := server.PanicHandler()
 	if err != nil {
 		log.Fatal(ctx, err)
@@ -129,7 +131,6 @@ func main() {
 	if err != nil {
 		log.Fatal(ctx, err)
 	}
-
 	mw := middleware.Chain(
 		middleware.RequestLog(requestLogger),
 		middleware.Quota(cfg.Quota),
@@ -140,19 +141,31 @@ func main() {
 		middleware.Timeout(54*time.Second),
 		middleware.Experiment(experimenter),
 	)
-
 	addr := cfg.HostAddr("localhost:8080")
 	log.Infof(ctx, "Listening on addr %s", addr)
 	log.Fatal(ctx, http.ListenAndServe(addr, mw(router)))
 }
 
+func newQueue(ctx context.Context, cfg *config.Config, proxyClient *proxy.Client, sourceClient *source.Client, db *postgres.DB) queue.Queue {
+	if !cfg.OnAppEngine() {
+		return queue.NewInMemory(ctx, proxyClient, sourceClient, db, 10, frontend.FetchAndUpdateState)
+	}
+	client, err := cloudtasks.NewClient(ctx)
+	if err != nil {
+		log.Fatal(ctx, err)
+	}
+	if queueName == "" {
+		log.Fatalf(ctx, "queueName cannot be empty")
+	}
+	return queue.NewGCP(cfg, client, queueName)
+}
+
 // openDB opens a connection to a database with the given driver, using connection info from
 // the given config.
 // It first tries the main connection info (DBConnInfo), and if that fails, it uses backup
 // connection info it if exists (DBSecondaryConnInfo).
 func openDB(ctx context.Context, cfg *config.Config, driver string) (_ *database.DB, err error) {
 	derrors.Wrap(&err, "openDB(ctx, cfg, %q)", driver)
-
 	log.Infof(ctx, "opening database on host %s", cfg.DBHost)
 	ddb, err := database.Open(driver, cfg.DBConnInfo())
 	if err == nil {
@@ -167,7 +180,6 @@ func openDB(ctx context.Context, cfg *config.Config, driver string) (_ *database
 		cfg.DBHost, err, cfg.DBSecondaryHost)
 	return database.Open(driver, ci)
 }
-
 func getLogger(ctx context.Context, cfg *config.Config) middleware.Logger {
 	if cfg.OnAppEngine() {
 		logger, err := log.UseStackdriver(ctx, cfg, "frontend-log")

internal/frontend/fetch.go

@@ -0,0 +1,61 @@
+// Copyright 2020 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package frontend
+
+import (
+	"context"
+	"net/http"
+
+	"golang.org/x/pkgsite/internal"
+	"golang.org/x/pkgsite/internal/derrors"
+	"golang.org/x/pkgsite/internal/fetch"
+	"golang.org/x/pkgsite/internal/log"
+	"golang.org/x/pkgsite/internal/postgres"
+	"golang.org/x/pkgsite/internal/proxy"
+	"golang.org/x/pkgsite/internal/source"
+)
+
+// FetchAndUpdateState is used by the InMemory queue for testing in
+// internal/frontend and running cmd/frontend locally. It is a copy of
+// worker.FetchAndUpdateState that does not update module_version_states, so that
+// we don't have to import internal/worker here. It is not meant to be used
+// when running on AppEngine.
+func FetchAndUpdateState(ctx context.Context, modulePath, requestedVersion string, proxyClient *proxy.Client, sourceClient *source.Client, db *postgres.DB) (_ int, err error) {
+	defer func() {
+		if err != nil {
+			log.Infof(ctx, "FetchAndUpdateState(%q, %q) completed with err: %v. ", modulePath, requestedVersion, err)
+		} else {
+			log.Infof(ctx, "FetchAndUpdateState(%q, %q) succeeded", modulePath, requestedVersion)
+		}
+		derrors.Wrap(&err, "FetchAndUpdateState(%q, %q)", modulePath, requestedVersion)
+	}()
+
+	fr := fetch.FetchModule(ctx, modulePath, requestedVersion, proxyClient, sourceClient)
+	if fr.Error == nil {
+		// Only attempt to insert the module into module_version_states if the
+		// fetch process was successful.
+		if err := db.InsertModule(ctx, fr.Module); err != nil {
+			return http.StatusInternalServerError, err
+		}
+	}
+	var errMsg string
+	if fr.Error != nil {
+		errMsg = fr.Error.Error()
+	}
+	vm := &internal.VersionMap{
+		ModulePath:       fr.ModulePath,
+		RequestedVersion: fr.RequestedVersion,
+		ResolvedVersion:  fr.ResolvedVersion,
+		Status:           fr.Status,
+		Error:            errMsg,
+	}
+	if err := db.UpsertVersionMap(ctx, vm); err != nil {
+		return http.StatusInternalServerError, err
+	}
+	if fr.Error != nil {
+		return fr.Status, fr.Error
+	}
+	return http.StatusOK, nil
+}

internal/frontend/server.go

@@ -24,11 +24,13 @@ import (
 	"golang.org/x/pkgsite/internal/licenses"
 	"golang.org/x/pkgsite/internal/log"
 	"golang.org/x/pkgsite/internal/middleware"
+	"golang.org/x/pkgsite/internal/queue"
 )
 
 // Server can be installed to serve the go discovery frontend.
 type Server struct {
-	ds internal.DataSource
+	ds    internal.DataSource
+	queue queue.Queue
 	// cmplClient is a redis client that has access to the "completions" sorted
 	// set.
 	cmplClient      *redis.Client
@@ -45,14 +47,15 @@ type Server struct {
 // NewServer creates a new Server for the given database and template directory.
 // reloadTemplates should be used during development when it can be helpful to
 // reload templates from disk each time a page is loaded.
-func NewServer(ds internal.DataSource, cmplClient *redis.Client, staticPath string, thirdPartyPath string, reloadTemplates bool) (*Server, error) {
+func NewServer(ds internal.DataSource, q queue.Queue, cmplClient *redis.Client, staticPath string, thirdPartyPath string, reloadTemplates bool) (*Server, error) {
 	templateDir := filepath.Join(staticPath, "html")
 	ts, err := parsePageTemplates(templateDir)
 	if err != nil {
 		return nil, fmt.Errorf("error parsing templates: %v", err)
 	}
 	s := &Server{
 		ds:              ds,
+		queue:           q,
 		cmplClient:      cmplClient,
 		staticPath:      staticPath,
 		thirdPartyPath:  thirdPartyPath,

internal/frontend/server_test.go

@@ -35,7 +35,7 @@ func TestMain(m *testing.M) {
 }
 
 func TestHTMLInjection(t *testing.T) {
-	s, err := NewServer(testDB, nil, "../../content/static", "../../third_party", false)
+	s, err := NewServer(testDB, nil, nil, "../../content/static", "../../third_party", false)
 	if err != nil {
 		t.Fatalf("NewServer: %v", err)
 	}
@@ -201,7 +201,7 @@ func testServer(t *testing.T, experimentNames ...string) {
 	ctx = experiment.NewContext(ctx, experiment.NewSet(expmap))
 	insertTestModules(ctx, t, testModules)
 
-	s, err := NewServer(testDB, nil, "../../content/static", "../../third_party", false)
+	s, err := NewServer(testDB, nil, nil, "../../content/static", "../../third_party", false)
 	if err != nil {
 		t.Fatalf("NewServer: %v", err)
 	}
@@ -860,7 +860,7 @@ func TestServerErrors(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	s, err := NewServer(testDB, nil, "../../content/static", "../../third_party", false)
+	s, err := NewServer(testDB, nil, nil, "../../content/static", "../../third_party", false)
 	if err != nil {
 		t.Fatalf("NewServer: %v", err)
 	}

internal/testing/integration/frontend_test.go

@@ -152,7 +152,7 @@ func TestModulePackageDirectoryResolution(t *testing.T) {
 				in(".DetailsContent", hasText("I'm a package"))),
 		},
 	}
-	s, err := frontend.NewServer(testDB, nil, "../../../content/static", "../../../third_party", false)
+	s, err := frontend.NewServer(testDB, nil, nil, "../../../content/static", "../../../third_party", false)
 	if err != nil {
 		t.Fatal(err)
 	}

internal/testing/integration/integration_test.go

@@ -86,7 +86,7 @@ func TestEndToEndProcessing(t *testing.T) {
 	workerServer.Install(workerMux.Handle)
 	workerHTTP := httptest.NewServer(workerMux)
 
-	frontendServer, err := frontend.NewServer(testDB, redisHAClient, "../../../content/static", "../../../third_party", false)
+	frontendServer, err := frontend.NewServer(testDB, queue, redisHAClient, "../../../content/static", "../../../third_party", false)
 	if err != nil {
 		t.Fatal(err)
 	}