cmd,internal: move taskIDChangeInterval to config At the moment, taskIDChangeInterval is a hardcoded value in internal/queue. However, we will soon have two task queues running, which require different change intervals, so this value is now set in internal/config. Additionally, the taskIDChangeInterval for the worker is changed to 3 hours. Change-Id: I498abefce6543005463be7da99a5a778f3a6e973 Reviewed-on: https://team-review.git.corp.google.com/c/golang/discovery/+/758919 CI-Result: Cloud Build <devtools-proctor-result-processor@system.gserviceaccount.com> Reviewed-by: Jonathan Amsterdam <jba@google.com>
diff --git a/cmd/frontend/main.go b/cmd/frontend/main.go index 3906234..61b7bad 100644 --- a/cmd/frontend/main.go +++ b/cmd/frontend/main.go
@@ -92,7 +92,7 @@ Addr: cfg.RedisHAHost + ":" + cfg.RedisHAPort, }) } - server, err := frontend.NewServer(ds, fetchQueue, haClient, *staticPath, *thirdPartyPath, *devMode) + server, err := frontend.NewServer(ds, fetchQueue, haClient, config.TaskIDChangeIntervalWorker, *staticPath, *thirdPartyPath, *devMode) if err != nil { log.Fatalf(ctx, "frontend.NewServer: %v", err) }
diff --git a/cmd/worker/main.go b/cmd/worker/main.go index fc2a490..bf1138c 100644 --- a/cmd/worker/main.go +++ b/cmd/worker/main.go
@@ -89,7 +89,7 @@ fetchQueue := newQueue(ctx, cfg, proxyClient, sourceClient, db) reportingClient := reportingClient(ctx, cfg) redisClient := getRedis(ctx, cfg) - server, err := worker.NewServer(cfg, db, indexClient, proxyClient, sourceClient, redisClient, fetchQueue, reportingClient, *staticPath) + server, err := worker.NewServer(cfg, db, indexClient, proxyClient, sourceClient, redisClient, fetchQueue, reportingClient, config.TaskIDChangeIntervalWorker, *staticPath) if err != nil { log.Fatal(ctx, err) }
diff --git a/internal/config/config.go b/internal/config/config.go index 4c57b4b..eb5b48b 100644 --- a/internal/config/config.go +++ b/internal/config/config.go
@@ -135,6 +135,14 @@ // to fetch source code from third party URLs. const SourceTimeout = 1 * time.Minute +// TaskIDChangeIntervalWorker is the time period during which a given module +// version can be re-enqueued to fetch tasks. +const TaskIDChangeIntervalWorker = 3 * time.Hour + +// TaskIDChangeIntervalFrontend is the time period during which a given module +// version can be re-enqueued to frontend tasks. +const TaskIDChangeIntervalFrontend = 30 * time.Minute + // DBConnInfo returns a PostgreSQL connection string constructed from // environment variables, using the primary database host. func (c *Config) DBConnInfo() string {
diff --git a/internal/frontend/server.go b/internal/frontend/server.go index 2a0bdd8..f47857a 100644 --- a/internal/frontend/server.go +++ b/internal/frontend/server.go
@@ -34,19 +34,20 @@ queue queue.Queue // cmplClient is a redis client that has access to the "completions" sorted // set. - cmplClient *redis.Client - staticPath string - thirdPartyPath string - templateDir string - devMode bool - errorPage []byte + cmplClient *redis.Client + taskIDChangeInterval time.Duration + staticPath string + thirdPartyPath string + templateDir string + devMode bool + errorPage []byte mu sync.Mutex // Protects all fields below templates map[string]*template.Template } // NewServer creates a new Server for the given database and template directory. -func NewServer(ds internal.DataSource, q queue.Queue, cmplClient *redis.Client, staticPath string, thirdPartyPath string, devMode bool) (_ *Server, err error) { +func NewServer(ds internal.DataSource, q queue.Queue, cmplClient *redis.Client, taskIDChangeInterval time.Duration, staticPath string, thirdPartyPath string, devMode bool) (_ *Server, err error) { defer derrors.Wrap(&err, "NewServer(...)") templateDir := filepath.Join(staticPath, "html") ts, err := parsePageTemplates(templateDir) @@ -54,14 +55,15 @@ return nil, fmt.Errorf("error parsing templates: %v", err) } s := &Server{ - ds: ds, - queue: q, - cmplClient: cmplClient, - staticPath: staticPath, - thirdPartyPath: thirdPartyPath, - templateDir: templateDir, - devMode: devMode, - templates: ts, + ds: ds, + queue: q, + cmplClient: cmplClient, + staticPath: staticPath, + thirdPartyPath: thirdPartyPath, + templateDir: templateDir, + devMode: devMode, + templates: ts, + taskIDChangeInterval: taskIDChangeInterval, } errorPageBytes, err := s.renderErrorPage(context.Background(), http.StatusInternalServerError, nil) if err != nil {
diff --git a/internal/frontend/server_test.go b/internal/frontend/server_test.go index cf10e37..bfe95e8 100644 --- a/internal/frontend/server_test.go +++ b/internal/frontend/server_test.go
@@ -928,7 +928,7 @@ } func newTestServer(t *testing.T, experimentNames ...string) (*Server, http.Handler) { - s, err := NewServer(testDB, nil, nil, "../../content/static", "../../third_party", false) + s, err := NewServer(testDB, nil, nil, 10*time.Minute, "../../content/static", "../../third_party", false) if err != nil { t.Fatal(err) }
diff --git a/internal/queue/queue.go b/internal/queue/queue.go index b0b1fb9..6e567aa 100644 --- a/internal/queue/queue.go +++ b/internal/queue/queue.go
@@ -28,7 +28,7 @@ // A Queue provides an interface for asynchronous scheduling of fetch actions. type Queue interface { - ScheduleFetch(ctx context.Context, modulePath, version, suffix string) error + ScheduleFetch(ctx context.Context, modulePath, version, suffix string, taskIDChangeInterval time.Duration) error } // GCP provides a Queue implementation backed by the Google Cloud Tasks @@ -53,15 +53,15 @@ // ScheduleFetch enqueues a task on GCP to fetch the given modulePath and // version. It returns an error if there was an error hashing the task name, or // an error pushing the task to GCP. -func (q *GCP) ScheduleFetch(ctx context.Context, modulePath, version, suffix string) (err error) { +func (q *GCP) ScheduleFetch(ctx context.Context, modulePath, version, suffix string, taskIDChangeInterval time.Duration) (err error) { // the new taskqueue API requires a deadline of <= 30s ctx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() - defer derrors.Wrap(&err, "queue.ScheduleFetch(%q, %q, %q)", modulePath, version, suffix) + defer derrors.Wrap(&err, "queue.ScheduleFetch(%q, %q, %q, %d)", modulePath, version, suffix, taskIDChangeInterval) queueName := fmt.Sprintf("projects/%s/locations/%s/queues/%s", q.cfg.ProjectID, q.cfg.LocationID, q.queueID) mod := fmt.Sprintf("%s/@v/%s", modulePath, version) u := fmt.Sprintf("/fetch/" + mod) - taskID := newTaskID(modulePath, version, time.Now()) + taskID := newTaskID(modulePath, version, time.Now(), taskIDChangeInterval) req := &taskspb.CreateTaskRequest{ Parent: queueName, Task: &taskspb.Task{ @@ -93,9 +93,6 @@ return nil } -// How often the task ID for a given module and version will change. -const taskIDChangeInterval = time.Hour - // Create a task ID for the given module path and version. // Task IDs can contain only letters ([A-Za-z]), numbers ([0-9]), hyphens (-), or underscores (_). // Also include a truncated time in the hash, so it changes periodically. @@ -104,7 +101,7 @@ // for two identical tasks to appear within that time period (for example, one at 2:59 // and the other at 3:01) -- each is part of a different taskIDChangeInterval-sized chunk // of time. But there will never be a third identical task in that interval. -func newTaskID(modulePath, version string, now time.Time) string { +func newTaskID(modulePath, version string, now time.Time, taskIDChangeInterval time.Duration) string { t := now.Truncate(taskIDChangeInterval) return fmt.Sprintf("%x", sha256.Sum256([]byte(modulePath+"@"+version+"-"+t.String()))) } @@ -174,7 +171,7 @@ // ScheduleFetch pushes a fetch task into the local queue to be processed // asynchronously. -func (q *InMemory) ScheduleFetch(ctx context.Context, modulePath, version, suffix string) error { +func (q *InMemory) ScheduleFetch(ctx context.Context, modulePath, version, suffix string, taskIDChangeInterval time.Duration) error { q.queue <- moduleVersion{modulePath, version} return nil }
diff --git a/internal/queue/queue_test.go b/internal/queue/queue_test.go index 3933325..429e801 100644 --- a/internal/queue/queue_test.go +++ b/internal/queue/queue_test.go
@@ -12,18 +12,18 @@ func TestNewTaskID(t *testing.T) { // Verify that the task ID is the same within taskIDChangeInterval and changes // afterwards. - const ( - module = "mod" - version = "ver" + var ( + module = "mod" + version = "ver" + taskIDChangeInterval = 3 * time.Hour ) - tm := time.Now().Truncate(taskIDChangeInterval) - id1 := newTaskID(module, version, tm) - id2 := newTaskID(module, version, tm.Add(taskIDChangeInterval/2)) + id1 := newTaskID(module, version, tm, taskIDChangeInterval) + id2 := newTaskID(module, version, tm.Add(taskIDChangeInterval/2), taskIDChangeInterval) if id1 != id2 { t.Error("wanted same task ID, got different") } - id3 := newTaskID(module, version, tm.Add(taskIDChangeInterval+1)) + id3 := newTaskID(module, version, tm.Add(taskIDChangeInterval+1), taskIDChangeInterval) if id1 == id3 { t.Error("wanted different task ID, got same") }
diff --git a/internal/testing/integration/frontend_test.go b/internal/testing/integration/frontend_test.go index 0827351..631f322 100644 --- a/internal/testing/integration/frontend_test.go +++ b/internal/testing/integration/frontend_test.go
@@ -152,7 +152,7 @@ in(".DetailsContent", hasText("I'm a package"))), }, } - s, err := frontend.NewServer(testDB, nil, nil, "../../../content/static", "../../../third_party", false) + s, err := frontend.NewServer(testDB, nil, nil, 10*time.Minute, "../../../content/static", "../../../third_party", false) if err != nil { t.Fatal(err) }
diff --git a/internal/testing/integration/integration_test.go b/internal/testing/integration/integration_test.go index cfd45e3..ae81fa5 100644 --- a/internal/testing/integration/integration_test.go +++ b/internal/testing/integration/integration_test.go
@@ -78,7 +78,7 @@ // back to worker, rather than calling fetch itself. queue := queue.NewInMemory(ctx, proxyClient, source.NewClient(1*time.Second), testDB, 10, worker.FetchAndUpdateState) - workerServer, err := worker.NewServer(&config.Config{}, testDB, indexClient, proxyClient, source.NewClient(1*time.Second), redisHAClient, queue, nil, "../../../content/static") + workerServer, err := worker.NewServer(&config.Config{}, testDB, indexClient, proxyClient, source.NewClient(1*time.Second), redisHAClient, queue, nil, 10*time.Minute, "../../../content/static") if err != nil { t.Fatal(err) } @@ -86,7 +86,7 @@ workerServer.Install(workerMux.Handle) workerHTTP := httptest.NewServer(workerMux) - frontendServer, err := frontend.NewServer(testDB, queue, redisHAClient, "../../../content/static", "../../../third_party", false) + frontendServer, err := frontend.NewServer(testDB, queue, redisHAClient, 10*time.Minute, "../../../content/static", "../../../third_party", false) if err != nil { t.Fatal(err) }
diff --git a/internal/worker/server.go b/internal/worker/server.go index f014595..aba6b32 100644 --- a/internal/worker/server.go +++ b/internal/worker/server.go
@@ -37,14 +37,15 @@ // Server can be installed to serve the go discovery worker. type Server struct { - cfg *config.Config - indexClient *index.Client - proxyClient *proxy.Client - sourceClient *source.Client - redisClient *redis.Client - db *postgres.DB - queue queue.Queue - reportingClient *errorreporting.Client + cfg *config.Config + indexClient *index.Client + proxyClient *proxy.Client + sourceClient *source.Client + redisClient *redis.Client + db *postgres.DB + queue queue.Queue + reportingClient *errorreporting.Client + taskIDChangeInterval time.Duration indexTemplate *template.Template } @@ -58,6 +59,7 @@ redisClient *redis.Client, queue queue.Queue, reportingClient *errorreporting.Client, + taskIDChangeInterval time.Duration, staticPath string, ) (_ *Server, err error) { defer derrors.Wrap(&err, "NewServer(db, ic, pc, q, %q)", staticPath) @@ -68,15 +70,16 @@ } return &Server{ - cfg: cfg, - db: db, - indexClient: indexClient, - proxyClient: proxyClient, - sourceClient: sourceClient, - redisClient: redisClient, - queue: queue, - reportingClient: reportingClient, - indexTemplate: indexTemplate, + cfg: cfg, + db: db, + indexClient: indexClient, + proxyClient: proxyClient, + sourceClient: sourceClient, + redisClient: redisClient, + queue: queue, + reportingClient: reportingClient, + indexTemplate: indexTemplate, + taskIDChangeInterval: taskIDChangeInterval, }, nil } @@ -276,7 +279,7 @@ } log.Infof(ctx, "Scheduling modules to be fetched: %d new modules from index.golang.org", len(versions)) for _, version := range versions { - if err := s.queue.ScheduleFetch(ctx, version.Path, version.Version, suffixParam); err != nil { + if err := s.queue.ScheduleFetch(ctx, version.Path, version.Version, suffixParam, s.taskIDChangeInterval); err != nil { return err } } @@ -308,7 +311,7 @@ w.Header().Set("Content-Type", "text/plain") log.Infof(ctx, "Scheduling modules to be fetched: requeuing %d modules", len(versions)) for _, v := range versions { - if err := s.queue.ScheduleFetch(ctx, v.ModulePath, v.Version, suffixParam); err != nil { + if err := s.queue.ScheduleFetch(ctx, v.ModulePath, v.Version, suffixParam, s.taskIDChangeInterval); err != nil { return err } } @@ -456,7 +459,7 @@ return "", err } for _, v := range versions { - if err := s.queue.ScheduleFetch(ctx, stdlib.ModulePath, v, suffix); err != nil { + if err := s.queue.ScheduleFetch(ctx, stdlib.ModulePath, v, suffix, s.taskIDChangeInterval); err != nil { return "", fmt.Errorf("error scheduling fetch for %s: %w", v, err) } }
diff --git a/internal/worker/server_test.go b/internal/worker/server_test.go index 450e276..66f6686 100644 --- a/internal/worker/server_test.go +++ b/internal/worker/server_test.go
@@ -164,7 +164,7 @@ // Use 10 workers to have parallelism consistent with the worker binary. q := queue.NewInMemory(ctx, proxyClient, sourceClient, testDB, 10, FetchAndUpdateState) - s, err := NewServer(&config.Config{}, testDB, indexClient, proxyClient, sourceClient, nil, q, nil, "") + s, err := NewServer(&config.Config{}, testDB, indexClient, proxyClient, sourceClient, nil, q, nil, 10*time.Minute, "") if err != nil { t.Fatal(err) }