Repository: google/exposure-notifications-server
Pull request #260 (2020-05-14 20:02:38)

add federation services to terraform config
I haven't actually run this yet, but the `terraform plan` output looks right.

I noticed there were a lot of env variables that seemed like they would apply across all our services, so I extracted those out into a local var. PTAL if any seem like they should not apply across all services.

I only added the noauth IAM for `federationout` since that's how it's configured in the test project.

I also didn't add the extra `gcloud alpha run` to deploy.yaml for these services since I wasn't sure why we would need them if we're using terraform to run the services.

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/260#" title="Author self-approved">danawillow</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **crwilcox**
You can assign the PR to them by writing `/assign @crwilcox` 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":["crwilcox"]} -->
mikehelmick: /lgtm
/assign @squee1945 
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/260#" title="Author self-approved">danawillow</a>*, *<a href="https://github.com/google/exposure-notifications-server/pull/260#pullrequestreview-412254498" title="Approved">squee1945</a>*
To complete the [pull request process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process), please assign **mikehelmick**
You can assign the PR to them by writing `/assign @mikehelmick` 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":["mikehelmick"]} -->
google-prow-robot: New changes are detected. LGTM label has been removed.
danawillow: /assign @crwilcox 
(since helmick is out today)

Patch:
diff --git a/builders/build.yaml b/builders/build.yaml
index 43cdd80..b3c353b 100644
--- a/builders/build.yaml
+++ b/builders/build.yaml
@@ -23,12 +23,12 @@ steps:
   args:
   - publish
   - -P
-  - ./cmd/federation-pull
+  - ./cmd/federationin
 - name: 'gcr.io/$PROJECT_ID/ko'
   args:
   - publish
   - -P
-  - ./cmd/federation
+  - ./cmd/federationout
 - name: 'gcr.io/$PROJECT_ID/ko'
   args:
   - publish
diff --git a/builders/deploy.yaml b/builders/deploy.yaml
index 10495b4..559688e 100644
--- a/builders/deploy.yaml
+++ b/builders/deploy.yaml
@@ -24,12 +24,12 @@ steps:
   args:
   - publish
   - -P
-  - ./cmd/federation-pull
+  - ./cmd/federationin
 - name: 'gcr.io/$PROJECT_ID/ko'
   args:
   - publish
   - -P
-  - ./cmd/federation
+  - ./cmd/federationout
 - name: 'gcr.io/$PROJECT_ID/ko'
   args:
   - publish
diff --git a/internal/database/database.go b/internal/database/database.go
index 3ba3f73..4bdde53 100644
--- a/internal/database/database.go
+++ b/internal/database/database.go
@@ -26,6 +26,9 @@ import (
 var (
 	// ErrNotFound indicates that the requested record was not found in the database.
 	ErrNotFound = errors.New("record not found")
+
+	// ErrKeyConflict indicates that there was a key conflict inserting a row.
+	ErrKeyConflict = errors.New("key conflict")
 )
 
 func toNullString(s string) sql.NullString {
diff --git a/internal/database/export.go b/internal/database/export.go
index a6d5e26..cd5cc03 100644
--- a/internal/database/export.go
+++ b/internal/database/export.go
@@ -39,6 +39,9 @@ func (db *DB) AddExportConfig(ctx context.Context, ec *model.ExportConfig) error
 	if ec.Period > oneDay {
 		return errors.New("maximum period is 24h")
 	}
+	if ec.Period == 0 {
+		return errors.New("period must be non-zero")
+	}
 	if int64(oneDay.Seconds())%int64(ec.Period.Seconds()) != 0 {
 		return errors.New("period must divide equally into 24 hours (e.g., 2h, 4h, 12h, 15m, 30m)")
 	}
@@ -333,7 +336,11 @@ func (db *DB) FinalizeBatch(ctx context.Context, eb *model.ExportBatch, files []
 				Status:     model.ExportBatchComplete,
 			}
 			if err := addExportFile(ctx, tx, &ef); err != nil {
-				return fmt.Errorf("adding export file entry: %w", err)
+				if err == ErrKeyConflict {
+					logging.FromContext(ctx).Infof("ExportFile %q already exists in database, skipping without overwriting. This can occur when reprocessing a failed batch.", file)
+				} else {
+					return fmt.Errorf("adding export file entry: %w", err)
+				}
 			}
 		}
 
@@ -512,8 +519,10 @@ func (db *DB) DeleteFilesBefore(ctx context.Context, before time.Time, blobstore
 	return count, nil
 }
 
+// addExportFile adds a row to ExportFile. If the row already exists (based on the primary key),
+// ErrKeyConflict is returned.
 func addExportFile(ctx context.Context, tx pgx.Tx, ef *model.ExportFile) error {
-	_, err := tx.Exec(ctx, `
+	tag, err := tx.Exec(ctx, `
 		INSERT INTO
 			ExportFile
 			(bucket_name, filename, batch_id, region, batch_num, batch_size, status)
@@ -524,6 +533,9 @@ func addExportFile(ctx context.Context, tx pgx.Tx, ef *model.ExportFile) error {
 	if err != nil {
 		return fmt.Errorf("inserting to ExportFile: %w", err)
 	}
+	if tag.RowsAffected() == 0 {
+		return ErrKeyConflict
+	}
 	return nil
 }
 
diff --git a/terraform/main.tf b/terraform/main.tf
index 80e4ad7..573aed8 100644
--- a/terraform/main.tf
+++ b/terraform/main.tf
@@ -256,6 +256,51 @@ resource "google_cloudbuild_trigger" "build-and-publish" {
   }
 }
 
+locals {
+  common_cloudrun_env_vars = [
+    {
+      name  = "DB_POOL_MIN_CONNS"
+      value = "2"
+    },
+    {
+      name  = "DB_POOL_MAX_CONNS"
+      value = "10"
+    },
+    {
+      name  = "DB_PASSWORD"
+      value = "secret://${google_secret_manager_secret_version.db-pwd-initial.name}"
+    },
+    {
+      name  = "DB_SSLKEY"
+      value = "secret://${google_secret_manager_secret_version.db-key.name}"
+    },
+    {
+      name  = "DB_SSLCERT"
+      value = "secret://{$google_secret_manager_secret_version.db-cert.name}"
+    },
+    {
+      name  = "DB_SSLROOTCERT"
+      value = "secret://{$google_secret_manager_secret_version.db-ca-cert.name}"
+    },
+    {
+      name  = "DB_SSLMODE"
+      value = "require"
+    },
+    {
+      name  = "DB_HOST"
+      value = "/cloudsql/${data.google_project.project.project_id}:${var.region}:${google_sql_database_instance.db-inst.name}"
+    },
+    {
+      name  = "DB_USER"
+      value = google_sql_user.user.name
+    },
+    {
+      name  = "DB_NAME"
+      value = google_sql_database.db.name
+    },
+  ]
+}
+
 resource "google_cloud_run_service" "exposure" {
   name     = "exposure"
   location = var.region
@@ -267,45 +312,12 @@ resource "google_cloud_run_service" "exposure" {
           name  = "CONFIG_REFRESH_DURATION"
           value = "5m"
         }
-        env {
-          name  = "DB_POOL_MIN_CONNS"
-          value = "2"
-        }
-        env {
-          name  = "DB_POOL_MAX_CONNS"
-          value = "10"
-        }
-        env {
-          name  = "DB_PASSWORD_SECRET"
-          value = google_secret_manager_secret_version.db-pwd-initial.name
-        }
-        env {
-          name  = "DB_SSLKEY_SECRET"
-          value = google_secret_manager_secret_version.db-key.name
-        }
-        env {
-          name  = "DB_SSLCERT_SECRET"
-          value = google_secret_manager_secret_version.db-cert.name
-        }
-        env {
-          name  = "DB_SSLROOTCERT_SECRET"
-          value = google_secret_manager_secret_version.db-ca-cert.name
-        }
-        env {
-          name  = "DB_SSLMODE"
-          value = "require"
-        }
-        env {
-          name  = "DB_HOST"
-          value = "/cloudsql/${data.google_project.project.project_id}:${var.region}:${google_sql_database_instance.db-inst.name}"
-        }
-        env {
-          name  = "DB_USER"
-          value = google_sql_user.user.name
-        }
-        env {
-          name  = "DB_NAME"
-          value = google_sql_database.db.name
+        dynamic "env" {
+          for_each = local.common_cloudrun_env_vars
+          content {
+            name = env.value["name"]
+            value = env.value["value"]
+          }
         }
       }
     }
@@ -334,45 +346,64 @@ resource "google_cloud_run_service" "export" {
           name  = "EXPORT_BUCKET"
           value = google_storage_bucket.export.name
         }
-        env {
-          name  = "DB_POOL_MIN_CONNS"
-          value = "2"
-        }
-        env {
-          name  = "DB_POOL_MAX_CONNS"
-          value = "10"
-        }
-        env {
-          name  = "DB_PASSWORD_SECRET"
-          value = google_secret_manager_secret_version.db-pwd-initial.name
-        }
-        env {
-          name  = "DB_SSLKEY_SECRET"
-          value = google_secret_manager_secret_version.db-key.name
-        }
-        env {
-          name  = "DB_SSLCERT_SECRET"
-          value = google_secret_manager_secret_version.db-cert.name
+        dynamic "env" {
+          for_each = local.common_cloudrun_env_vars
+          content {
+            name = env.value["name"]
+            value = env.value["value"]
+          }
         }
-        env {
-          name  = "DB_SSLROOTCERT_SECRET"
-          value = google_secret_manager_secret_version.db-ca-cert.name
-        }
-        env {
-          name  = "DB_SSLMODE"
-          value = "require"
-        }
-        env {
-          name  = "DB_HOST"
-          value = "/cloudsql/${data.google_project.project.project_id}:${var.region}:${google_sql_database_instance.db-inst.name}"
-        }
-        env {
-          name  = "DB_USER"
-          value = google_sql_user.user.name
+      }
+    }
+    metadata {
+      annotations = {
+        "run.googleapis.com/cloudsql-instances" : "${data.google_project.project.project_id}:${var.region}:${google_sql_database_instance.db-inst.name}"
+        "autoscaling.knative.dev/maxScale"      : "1000"
+      }
+    }
+  }
+	depends_on = [google_cloudbuild_trigger.build-and-publish]
+}
+
+resource "google_cloud_run_service" "federationin" {
+  name     = "federationin"
+  location = var.region
+  template {
+    spec {
+      containers {
+        image = "us.gcr.io/${data.google_project.project.project_id}/github.com/google/exposure-notifications-server/cmd/federationin:latest"
+        dynamic "env" {
+          for_each = local.common_cloudrun_env_vars
+          content {
+            name = env.value["name"]
+            value = env.value["value"]
+          }
         }
-        env {
-          name  = "DB_NAME"
-          value = google_sql_database.db.name
+      }
+    }
+    metadata {
+      annotations = {
+        "run.googleapis.com/cloudsql-instances" : "${data.google_project.project.project_id}:${var.region}:${google_sql_database_instance.db-inst.name}"
+        "autoscaling.knative.dev/maxScale"      : "1000"
+      }
+    }
+  }
+  depends_on = [google_cloudbuild_trigger.build-and-publish]
+}
+
+resource "google_cloud_run_service" "federationout" {
+  name     = "federationout"
+  location = var.region
+  template {
+    spec {
+      containers {
+        image = "us.gcr.io/${data.google_project.project.project_id}/github.com/google/exposure-notifications-server/cmd/federationout:latest"
+        dynamic "env" {
+          for_each = local.common_cloudrun_env_vars
+          content {
+            name = env.value["name"]
+            value = env.value["value"]
+          }
         }
       }
     }
@@ -398,9 +429,9 @@ data "google_iam_policy" "noauth" {
 }
 
 resource "google_cloud_run_service_iam_policy" "export-noauth" {
-  location = google_cloud_run_service.exposure.location
-  project  = google_cloud_run_service.exposure.project
-  service  = google_cloud_run_service.exposure.name
+  location = google_cloud_run_service.export.location
+  project  = google_cloud_run_service.export.project
+  service  = google_cloud_run_service.export.name
 
   policy_data = data.google_iam_policy.noauth.policy_data
 }
@@ -413,6 +444,14 @@ resource "google_cloud_run_service_iam_policy" "exposure-noauth" {
   policy_data = data.google_iam_policy.noauth.policy_data
 }
 
+resource "google_cloud_run_service_iam_policy" "federationout-noauth" {
+  location = google_cloud_run_service.federationout.location
+  project  = google_cloud_run_service.federationout.project
+  service  = google_cloud_run_service.federationout.name
+
+  policy_data = data.google_iam_policy.noauth.policy_data
+}
+
 resource "google_project_iam_member" "export-worker" {
   project = data.google_project.project.project_id
   role    = "roles/run.invoker"


Test patch:
diff --git a/internal/database/export_test.go b/internal/database/export_test.go
index 69b5f13..51cd2f9 100644
--- a/internal/database/export_test.go
+++ b/internal/database/export_test.go
@@ -16,6 +16,7 @@ package database
 
 import (
 	"context"
+	"errors"
 	"sort"
 	"testing"
 	"time"
@@ -428,3 +429,78 @@ func TestKeysInBatch(t *testing.T) {
 		t.Fatalf("Incorrect exposure key in batch, got %q, want %q", got[0].ExposureKey, want)
 	}
 }
+
+// TestAddExportFileSkipsDuplicates ensures that ExportFile records are not overwritten.
+func TestAddExportFileSkipsDuplicates(t *testing.T) {
+	if testDB == nil {
+		t.Skip("no test DB")
+	}
+	defer resetTestDB(t)
+	ctx := context.Background()
+
+	// Add foreign key records.
+	ec := &model.ExportConfig{Period: time.Hour}
+	if err := testDB.AddExportConfig(ctx, ec); err != nil {
+		t.Fatal(err)
+	}
+	eb := &model.ExportBatch{ConfigID: ec.ConfigID, Status: model.ExportBatchOpen}
+	if err := testDB.AddExportBatches(ctx, []*model.ExportBatch{eb}); err != nil {
+		t.Fatal(err)
+	}
+	// Lease the batch to get the ID.
+	eb, err := testDB.LeaseBatch(ctx, time.Hour, time.Now())
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	wantBucketName := "bucket-1"
+	ef := &model.ExportFile{
+		Filename:   "file",
+		BucketName: wantBucketName,
+		BatchID:    eb.BatchID,
+	}
+
+	// Add a record.
+	err = testDB.inTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
+		if err := addExportFile(ctx, tx, ef); err != nil {
+			return err
+		}
+		return nil
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Check that the row is present.
+	got, err := testDB.LookupExportFile(ctx, ef.Filename)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.BucketName != wantBucketName {
+		t.Fatalf("bucket name mismatch got %q, want %q", got.BucketName, wantBucketName)
+	}
+
+	// Add a second record with same filename, must return ErrKeyConflict, and not overwrite.
+	ef.BucketName = "bucket-2"
+	err = testDB.inTx(ctx, pgx.Serializable, func(tx pgx.Tx) error {
+		if err := addExportFile(ctx, tx, ef); err != nil {
+			if err == ErrKeyConflict {
+				return nil // Expected result.
+			}
+			return err
+		}
+		return errors.New("missing expected ErrKeyConflict")
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Row must not be updated.
+	got, err = testDB.LookupExportFile(ctx, ef.Filename)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got.BucketName != wantBucketName {
+		t.Fatalf("bucket name mismatch got %q, want %q", got.BucketName, wantBucketName)
+	}
+}