Repository: google/exposure-notifications-server
Pull request #308 (2020-05-15 23:37:26)

make truncate window configurable
Eventually there might be a better home for this sort of
cross-cutting global config, but for now, an env var works.

This will allow quicker end-to-end tests of the round trip
of upload a key, wait for it to get into an export batch,
test exposure on another device.

google-prow-robot: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/google/exposure-notifications-server/pull/308#" title="Author self-approved">gurayAlsac</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **sethvargo**
You can assign the PR to them by writing `/assign @sethvargo` in a comment when ready.

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=google%2Fexposure-notifications-server).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/google/exposure-notifications-server/blob/master/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["sethvargo"]} -->

Patch:
diff --git a/internal/api/export/batcher.go b/internal/api/export/batcher.go
index b60efb6..c620e85 100644
--- a/internal/api/export/batcher.go
+++ b/internal/api/export/batcher.go
@@ -98,7 +98,7 @@ func (s *Server) maybeCreateBatches(ctx context.Context, ec *model.ExportConfig,
 		return 0, fmt.Errorf("fetching most recent batch for config %d: %w", ec.ConfigID, err)
 	}
 
-	ranges := makeBatchRanges(ec.Period, latestEnd, now)
+	ranges := makeBatchRanges(ec.Period, latestEnd, now, s.config.TruncateWindow)
 	if len(ranges) == 0 {
 		metrics.WriteInt("export-batcher-no-work", true, 1)
 		logger.Debugf("Batch creation for config %d is not required, skipping", ec.ConfigID)
@@ -134,10 +134,10 @@ type batchRange struct {
 
 var sanityDate = time.Date(2019, 1, 1, 0, 0, 0, 0, time.UTC)
 
-func makeBatchRanges(period time.Duration, latestEnd, now time.Time) []batchRange {
+func makeBatchRanges(period time.Duration, latestEnd, now time.Time, truncateWindow time.Duration) []batchRange {
 
 	// Compute the end of the exposure publish window; we don't want any batches with an end date greater than this time.
-	publishEnd := model.TruncateWindow(now)
+	publishEnd := model.TruncateWindow(now, truncateWindow)
 
 	// Special case: if there have not been batches before, return only a single one.
 	// We use sanityDate here because the loop below will happily create batch ranges
diff --git a/internal/api/export/config.go b/internal/api/export/config.go
index b86bccb..0d3ee77 100644
--- a/internal/api/export/config.go
+++ b/internal/api/export/config.go
@@ -38,6 +38,7 @@ type Config struct {
 	MaxRecords        int           `envconfig:"EXPORT_FILE_MAX_RECORDS" default:"30000"`
 	DefaultKeyID      string        `envconfig:"EXPORT_FILE_DEFAULT_KEY_ID" default:"ExampleServer"`
 	DefaultKeyVersion string        `envconfig:"EXPORT_FILE_DEFAULT_KEY_VERSION" default:"1"`
+	TruncateWindow    time.Duration `envconfig:"TRUNCATE_WINDOW" default:"1h"`
 }
 
 // DB returns the database config.
diff --git a/internal/api/export/worker.go b/internal/api/export/worker.go
index 84454a1..865e853 100644
--- a/internal/api/export/worker.go
+++ b/internal/api/export/worker.go
@@ -284,7 +284,7 @@ func randomInt(min, max int) (int, error) {
 
 func ensureMinNumExposures(exposures []*model.Exposure, region string, minLength, jitter int) ([]*model.Exposure, error) {
 	if len(exposures) == 0 {
-		return nil, fmt.Errorf("cannot pad zero length exposures")
+		return exposures, nil
 	}
 
 	extra, _ := randomInt(0, jitter)
diff --git a/internal/api/federationin/config.go b/internal/api/federationin/config.go
index bbc2118..c0d4e85 100644
--- a/internal/api/federationin/config.go
+++ b/internal/api/federationin/config.go
@@ -39,9 +39,10 @@ var _ setup.DBConfigProvider = (*Config)(nil)
 
 // Config is the configuration for federation-pull components (data pulled from other servers).
 type Config struct {
-	Database *database.Config
-	Port     string        `envconfig:"PORT" default:"8080"`
-	Timeout  time.Duration `envconfig:"RPC_TIMEOUT" default:"10m"`
+	Database       *database.Config
+	Port           string        `envconfig:"PORT" default:"8080"`
+	Timeout        time.Duration `envconfig:"RPC_TIMEOUT" default:"10m"`
+	TruncateWindow time.Duration `envconfig:"TRUNCATE_WINDOW" default:"1h"`
 
 	// TLSSkipVerify, if set to true, causes the server certificate to not be verified.
 	// This is typically used when testing locally with self-signed certificates.
diff --git a/internal/api/federationin/federationin.go b/internal/api/federationin/federationin.go
index 8d777db..28fc4d5 100644
--- a/internal/api/federationin/federationin.go
+++ b/internal/api/federationin/federationin.go
@@ -173,7 +173,7 @@ func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		startFederationSync: h.db.StartFederationInSync,
 	}
 	batchStart := time.Now()
-	if err := pull(timeoutContext, metrics, deps, query, batchStart); err != nil {
+	if err := pull(timeoutContext, metrics, deps, query, batchStart, h.config.TruncateWindow); err != nil {
 		internalErrorf(ctx, w, "Federation query %q failed: %v", queryID, err)
 		return
 	}
@@ -183,7 +183,7 @@ func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-func pull(ctx context.Context, metrics metrics.Exporter, deps pullDependencies, q *model.FederationInQuery, batchStart time.Time) error {
+func pull(ctx context.Context, metrics metrics.Exporter, deps pullDependencies, q *model.FederationInQuery, batchStart time.Time, truncateWindow time.Duration) error {
 	logger := logging.FromContext(ctx)
 	logger.Infof("Processing query %q", q.QueryID)
 
@@ -204,7 +204,7 @@ func pull(ctx context.Context, metrics metrics.Exporter, deps pullDependencies,
 		logger.Infof("Inserted %d keys", total)
 	}()
 
-	createdAt := model.TruncateWindow(batchStart)
+	createdAt := model.TruncateWindow(batchStart, truncateWindow)
 	partial := true
 	for partial {
 
diff --git a/internal/api/federationout/config.go b/internal/api/federationout/config.go
index bf02b8a..98b9b7e 100644
--- a/internal/api/federationout/config.go
+++ b/internal/api/federationout/config.go
@@ -26,9 +26,10 @@ var _ setup.DBConfigProvider = (*Config)(nil)
 
 // Config is the configuration for the federation components (data sent to other servers).
 type Config struct {
-	Port     string        `envconfig:"PORT" default:"8080"`
-	Timeout  time.Duration `envconfig:"RPC_TIMEOUT" default:"5m"`
-	Database *database.Config
+	Database       *database.Config
+	Port           string        `envconfig:"PORT" default:"8080"`
+	Timeout        time.Duration `envconfig:"RPC_TIMEOUT" default:"5m"`
+	TruncateWindow time.Duration `envconfig:"TRUNCATE_WINDOW" default:"1h"`
 
 	// AllowAnyClient, if true, removes authentication requirements on the federation endpoint.
 	// In practise, this is only useful in local testing.
diff --git a/internal/api/federationout/federationout.go b/internal/api/federationout/federationout.go
index 33f756f..abcf848 100644
--- a/internal/api/federationout/federationout.go
+++ b/internal/api/federationout/federationout.go
@@ -67,7 +67,7 @@ func (s Server) Fetch(ctx context.Context, req *pb.FederationFetchRequest) (*pb.
 	ctx, cancel := context.WithTimeout(ctx, s.config.Timeout)
 	defer cancel()
 	logger := logging.FromContext(ctx)
-	response, err := s.fetch(ctx, req, s.db.IterateExposures, model.TruncateWindow(time.Now())) // Don't fetch the current window, which isn't complete yet. TODO(squee1945): should I double this for safety?
+	response, err := s.fetch(ctx, req, s.db.IterateExposures, model.TruncateWindow(time.Now(), s.config.TruncateWindow)) // Don't fetch the current window, which isn't complete yet. TODO(squee1945): should I double this for safety?
 	if err != nil {
 		s.env.MetricsExporter(ctx).WriteInt("federation-fetch-failed", true, 1)
 		logger.Errorf("Fetch error: %v", err)
diff --git a/internal/api/publish/config.go b/internal/api/publish/config.go
index 3d5ba68..ef0643f 100644
--- a/internal/api/publish/config.go
+++ b/internal/api/publish/config.go
@@ -35,6 +35,7 @@ type Config struct {
 	MaxKeysOnPublish         int           `envconfig:"MAX_KEYS_ON_PUBLISH" default:"14"`
 	MaxIntervalAge           time.Duration `envconfig:"MAX_INTERVAL_AGE_ON_PUBLISH" default:"360h"`
 	APIConfigRefreshDuration time.Duration `envconfig:"CONFIG_REFRESH_DURATION" default:"5m"`
+	TruncateWindow           time.Duration `envconfig:"TRUNCATE_WINDOW" default:"1h"`
 
 	APIConfigOpts *dbapiconfig.ConfigOpts
 
diff --git a/internal/api/publish/publish.go b/internal/api/publish/publish.go
index 5f98b24..579a075 100644
--- a/internal/api/publish/publish.go
+++ b/internal/api/publish/publish.go
@@ -41,12 +41,13 @@ func NewHandler(ctx context.Context, config *Config, env *serverenv.ServerEnv) (
 		return nil, fmt.Errorf("missing apiconfig provider in server environment")
 	}
 
-	transformer, err := model.NewTransformer(config.MaxKeysOnPublish, config.MaxIntervalAge)
+	transformer, err := model.NewTransformer(config.MaxKeysOnPublish, config.MaxIntervalAge, config.TruncateWindow)
 	if err != nil {
 		return nil, fmt.Errorf("model.NewTransformer: %w", err)
 	}
 	logger.Infof("max keys per upload: %v", config.MaxKeysOnPublish)
 	logger.Infof("max interval start age: %v", config.MaxIntervalAge)
+	logger.Infof("truncate window: %v", config.TruncateWindow)
 
 	return &publishHandler{
 		serverenv:   env,
diff --git a/internal/model/exposure.go b/internal/model/exposure.go
index b6fe1c7..028ba73 100644
--- a/internal/model/exposure.go
+++ b/internal/model/exposure.go
@@ -41,9 +41,6 @@ const (
 
 	// Self explanitory.
 	oneDay = time.Hour * 24
-	// Alignment window for requests. The created_at time for an exposure
-	// record is rounded down to the beginning of the createWindow.
-	createWindow = time.Hour
 
 	// interval length
 	intervalLength = 10 * time.Minute
@@ -119,26 +116,28 @@ func IntervalNumber(t time.Time) int32 {
 }
 
 // TruncateWindow truncates a time based on the size of the creation window.
-func TruncateWindow(t time.Time) time.Time {
-	return t.Truncate(createWindow)
+func TruncateWindow(t time.Time, d time.Duration) time.Time {
+	return t.Truncate(d)
 }
 
 // Transformer represents a configured Publish -> Exposure[] transformer.
 type Transformer struct {
 	maxExposureKeys     int
 	maxIntervalStartAge time.Duration // How many intervals old does this server accept?
+	truncateWindow      time.Duration
 }
 
 // NewTransformer creates a transformer for turning publish API requests into
 // records for insertion into the database. On the call to TransofmrPublish
 // all data is validated according to the transformer that is used.
-func NewTransformer(maxExposureKeys int, maxIntervalStartAge time.Duration) (*Transformer, error) {
+func NewTransformer(maxExposureKeys int, maxIntervalStartAge time.Duration, truncateWindow time.Duration) (*Transformer, error) {
 	if maxExposureKeys < 0 || maxExposureKeys > maxKeysPerPublish {
 		return nil, fmt.Errorf("maxExposureKeys must be > 0 and <= %v, got %v", maxKeysPerPublish, maxExposureKeys)
 	}
 	return &Transformer{
 		maxExposureKeys:     maxExposureKeys,
 		maxIntervalStartAge: maxIntervalStartAge,
+		truncateWindow:      truncateWindow,
 	}, nil
 }
 
@@ -158,7 +157,7 @@ func (t *Transformer) TransformPublish(inData *Publish, batchTime time.Time) ([]
 		return nil, fmt.Errorf("too many exposure keys in publish: %v, max of %v is allowed", len(inData.Keys), t.maxExposureKeys)
 	}
 
-	createdAt := TruncateWindow(batchTime)
+	createdAt := TruncateWindow(batchTime, t.truncateWindow)
 	entities := make([]*Exposure, 0, len(inData.Keys))
 
 	// An exposure key must have an interval >= minInteravl (max configured age)
diff --git a/terraform/README.md b/terraform/README.md
index 36e96d5..e1bf326 100644
--- a/terraform/README.md
+++ b/terraform/README.md
@@ -30,7 +30,7 @@ the exposure server code will trigger a new deployment. To enable this:
 
     1. Remember which repo you used. You will need to set the repo owner (e.g. 'google') and name (e.g. 'exposure-notifications-server') as variables in the `terraform apply`
 
-1. Log in to gcloud using `gcloud auth application-default login`.
+1. Log in to gcloud using `gcloud auth login && gcloud auth application-default login` (this will open two authentication windows in your web browser).
 
 1. Change to this directory and run `terraform init`.  Terraform will
 automatically download the plugins required to execute this code.


Test patch:
diff --git a/internal/api/export/batcher_test.go b/internal/api/export/batcher_test.go
index 04cdbdc..45c961d 100644
--- a/internal/api/export/batcher_test.go
+++ b/internal/api/export/batcher_test.go
@@ -108,7 +108,7 @@ func TestMakeBatchRanges(t *testing.T) {
 			nowT := fromSimpleTime(t, now)
 			latestEndT := fromSimpleTime(t, tc.latestEnd)
 
-			got := makeBatchRanges(tc.period, latestEndT, nowT)
+			got := makeBatchRanges(tc.period, latestEndT, nowT, time.Hour)
 
 			if len(got) != len(tc.want) {
 				t.Errorf("incorrect number of batches got %v, want %v", toSimpleBatchRange(t, got), tc.want)
diff --git a/internal/api/export/worker_test.go b/internal/api/export/worker_test.go
index 092ed33..650f0c5 100644
--- a/internal/api/export/worker_test.go
+++ b/internal/api/export/worker_test.go
@@ -47,9 +47,12 @@ func TestRandomInt(t *testing.T) {
 
 func TestDoNotPadZeroLength(t *testing.T) {
 	exposures := make([]*model.Exposure, 0)
-	_, err := ensureMinNumExposures(exposures, "US", 1000, 100)
-	if err == nil {
-		t.Errorf("expected error, got nil")
+	exposures, err := ensureMinNumExposures(exposures, "US", 1000, 100)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if len(exposures) != 0 {
+		t.Errorf("zero length slice shouldn't have been padded, but was")
 	}
 }
 
diff --git a/internal/api/federationin/federationin_test.go b/internal/api/federationin/federationin_test.go
index f2fc1aa..7667361 100644
--- a/internal/api/federationin/federationin_test.go
+++ b/internal/api/federationin/federationin_test.go
@@ -249,7 +249,7 @@ func TestFederationPull(t *testing.T) {
 				startFederationSync: sdb.startFederationSync,
 			}
 
-			err := pull(ctx, metrics.NewLogsBasedFromContext(ctx), deps, query, batchStart)
+			err := pull(ctx, metrics.NewLogsBasedFromContext(ctx), deps, query, batchStart, time.Hour)
 			if err != nil {
 				t.Fatalf("pull returned err=%v, want err=nil", err)
 			}
diff --git a/internal/model/exposure_test.go b/internal/model/exposure_test.go
index 1d08f3f..c069a1b 100644
--- a/internal/model/exposure_test.go
+++ b/internal/model/exposure_test.go
@@ -41,7 +41,7 @@ func TestInvalidNew(t *testing.T) {
 	}
 
 	for i, c := range cases {
-		_, err := NewTransformer(c.maxKeys, time.Hour)
+		_, err := NewTransformer(c.maxKeys, time.Hour, time.Hour)
 		if err != nil && errMsg == "" {
 			t.Errorf("%v unexpected error: %v", i, err)
 		} else if err != nil && !strings.Contains(err.Error(), c.message) {
@@ -51,7 +51,7 @@ func TestInvalidNew(t *testing.T) {
 }
 
 func TestInvalidBase64(t *testing.T) {
-	transformer, err := NewTransformer(1, time.Hour*24)
+	transformer, err := NewTransformer(1, time.Hour*24, time.Hour)
 	if err != nil {
 		t.Fatalf("error creating transformer: %v", err)
 	}
@@ -108,7 +108,7 @@ func TestPublishValidation(t *testing.T) {
 	currentInterval := IntervalNumber(captureStartTime)
 	minInterval := IntervalNumber(captureStartTime.Add(-1 * maxAge))
 
-	tf, err := NewTransformer(2, maxAge)
+	tf, err := NewTransformer(2, maxAge, time.Hour)
 	if err != nil {
 		t.Fatalf("unepected error: %v", err)
 	}
@@ -319,7 +319,7 @@ func TestTransform(t *testing.T) {
 		},
 	}
 	batchTime := captureStartTime.Add(time.Hour * 24 * 7)
-	batchTimeRounded := TruncateWindow(batchTime)
+	batchTimeRounded := TruncateWindow(batchTime, time.Hour)
 	for i, v := range want {
 		want[i] = &Exposure{
 			ExposureKey:      v.ExposureKey,
@@ -334,7 +334,7 @@ func TestTransform(t *testing.T) {
 	}
 
 	allowedAge := 14 * 24 * time.Hour
-	transformer, err := NewTransformer(10, allowedAge)
+	transformer, err := NewTransformer(10, allowedAge, time.Hour)
 	if err != nil {
 		t.Fatalf("NewTransformer returned unexpected error: %v", err)
 	}
@@ -404,7 +404,7 @@ func TestTransformOverlapping(t *testing.T) {
 		t.Run(c.name, func(t *testing.T) {
 			batchTime := captureStartTime.Add(time.Hour * 24 * 7)
 			allowedAge := 14 * 24 * time.Hour
-			transformer, err := NewTransformer(10, allowedAge)
+			transformer, err := NewTransformer(10, allowedAge, time.Hour)
 			if err != nil {
 				t.Fatalf("NewTransformer returned unexpected error: %v", err)
 			}