Repository: google/exposure-notifications-server
Pull request #565 (2020-06-06 19:42:26)

Add interfaces for getting objects via the blobstore
This is required for testing, but could also be useful as the system expands. Added sample use in publish_test.go

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

This pull-request has been approved by: *<a href="https://github.com/google/exposure-notifications-server/pull/565#" title="Author self-approved">sethvargo</a>*

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

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

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

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

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: /assign @mikehelmick 

Patch:
diff --git a/internal/monolith/config.go b/internal/monolith/config.go
deleted file mode 100644
index 1ade931..0000000
--- a/internal/monolith/config.go
+++ /dev/null
@@ -1,68 +0,0 @@
-// Copyright 2020 Google LLC
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package monolith
-
-import (
-	"github.com/google/exposure-notifications-server/internal/authorizedapp"
-	"github.com/google/exposure-notifications-server/internal/cleanup"
-	"github.com/google/exposure-notifications-server/internal/database"
-	"github.com/google/exposure-notifications-server/internal/export"
-	"github.com/google/exposure-notifications-server/internal/federationin"
-	"github.com/google/exposure-notifications-server/internal/publish"
-	"github.com/google/exposure-notifications-server/internal/secrets"
-	"github.com/google/exposure-notifications-server/internal/setup"
-	"github.com/google/exposure-notifications-server/internal/signing"
-	"github.com/google/exposure-notifications-server/internal/storage"
-)
-
-var _ setup.AuthorizedAppConfigProvider = (*Config)(nil)
-var _ setup.BlobstoreConfigProvider = (*Config)(nil)
-var _ setup.DatabaseConfigProvider = (*Config)(nil)
-var _ setup.KeyManagerConfigProvider = (*Config)(nil)
-var _ setup.SecretManagerConfigProvider = (*Config)(nil)
-
-type Config struct {
-	AuthorizedApp authorizedapp.Config
-	Storage       storage.Config
-	Cleanup       cleanup.Config
-	Database      database.Config
-	Export        export.Config
-	FederationIn  federationin.Config
-	KeyManager    signing.Config
-	Publish       publish.Config
-	SecretManager secrets.Config
-
-	Port string `env:"PORT, default=8080"`
-}
-
-func (c *Config) AuthorizedAppConfig() *authorizedapp.Config {
-	return &c.AuthorizedApp
-}
-
-func (c *Config) BlobstoreConfig() *storage.Config {
-	return &c.Storage
-}
-
-func (c *Config) DatabaseConfig() *database.Config {
-	return &c.Database
-}
-
-func (c *Config) KeyManagerConfig() *signing.Config {
-	return &c.KeyManager
-}
-
-func (c *Config) SecretManagerConfig() *secrets.Config {
-	return &c.SecretManager
-}
diff --git a/internal/monolith/monolith.go b/internal/monolith/monolith.go
deleted file mode 100644
index cb8caf5..0000000
--- a/internal/monolith/monolith.go
+++ /dev/null
@@ -1,90 +0,0 @@
-// Copyright 2020 Google LLC
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-// Package monolith runs all the server components at different URL paths.
-package monolith
-
-import (
-	"context"
-	"fmt"
-	"net/http"
-
-	"github.com/google/exposure-notifications-server/internal/cleanup"
-	"github.com/google/exposure-notifications-server/internal/export"
-	"github.com/google/exposure-notifications-server/internal/federationin"
-	"github.com/google/exposure-notifications-server/internal/handlers"
-	"github.com/google/exposure-notifications-server/internal/logging"
-	"github.com/google/exposure-notifications-server/internal/publish"
-	"github.com/google/exposure-notifications-server/internal/server"
-	"github.com/google/exposure-notifications-server/internal/setup"
-
-	// Enable observability with distributed tracing and metrics.
-	_ "github.com/google/exposure-notifications-server/internal/observability"
-)
-
-func RunServer(ctx context.Context) error {
-	logger := logging.FromContext(ctx)
-
-	var config Config
-	env, err := setup.Setup(ctx, &config)
-	if err != nil {
-		return fmt.Errorf("setup.Setup: %w", err)
-	}
-	defer env.Close(ctx)
-
-	mux := http.NewServeMux()
-
-	// Cleanup export
-	cleanupExport, err := cleanup.NewExportHandler(&config.Cleanup, env)
-	if err != nil {
-		return fmt.Errorf("cleanup.NewExportHandler: %w", err)
-	}
-	mux.Handle("/cleanup-export", cleanupExport)
-
-	// Cleanup exposure
-	cleanupExposure, err := cleanup.NewExposureHandler(&config.Cleanup, env)
-	if err != nil {
-		return fmt.Errorf("cleanup.NewExposureHandler: %w", err)
-	}
-	mux.Handle("/cleanup-exposure", cleanupExposure)
-
-	// Export
-	exportServer, err := export.NewServer(&config.Export, env)
-	if err != nil {
-		return fmt.Errorf("export.NewServer: %w", err)
-	}
-	mux.HandleFunc("/export/create-batches", exportServer.CreateBatchesHandler)
-	mux.HandleFunc("/export/do-work", exportServer.WorkerHandler)
-
-	// Federation in
-	mux.Handle("/federation-in", federationin.NewHandler(env, &config.FederationIn))
-
-	// Federation out
-	// TODO: this is a grpc listener and requires a lot of setup.
-
-	// Publish
-	publishServer, err := publish.NewHandler(ctx, &config.Publish, env)
-	if err != nil {
-		return fmt.Errorf("publish.NewHandler: %w", err)
-	}
-	mux.HandleFunc("/publish", handlers.WithMinimumLatency(config.Publish.MinRequestDuration, publishServer))
-
-	server, err := server.New(config.Port)
-	if err != nil {
-		return fmt.Errorf("server.New: %w", err)
-	}
-	logger.Infof("listening on :%s", config.Port)
-
-	return server.ServeHTTPHandler(ctx, mux)
-}
diff --git a/internal/storage/aws_s3.go b/internal/storage/aws_s3.go
index fed03da..ee5f4d2 100644
--- a/internal/storage/aws_s3.go
+++ b/internal/storage/aws_s3.go
@@ -18,8 +18,10 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"io/ioutil"
 
 	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/s3"
 )
@@ -77,3 +79,27 @@ func (s *AWSS3) DeleteObject(ctx context.Context, bucket, key string) error {
 	}
 	return nil
 }
+
+// GetObject returns the contents for the given object. If the object does not
+// exist, it returns ErrNotFound.
+func (s *AWSS3) GetObject(ctx context.Context, bucket, key string) ([]byte, error) {
+	o, err := s.svc.GetObjectWithContext(ctx, &s3.GetObjectInput{
+		Bucket: aws.String(bucket),
+		Key:    aws.String(key),
+	})
+	if err != nil {
+		if aerr, ok := err.(awserr.Error); ok &&
+			aerr.Code() == s3.ErrCodeNoSuchBucket || aerr.Code() == s3.ErrCodeNoSuchKey {
+			return nil, ErrNotFound
+		}
+		return nil, fmt.Errorf("failed to get object: %w", err)
+	}
+	defer o.Body.Close()
+
+	b, err := ioutil.ReadAll(o.Body)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read object: %w", err)
+	}
+
+	return b, nil
+}
diff --git a/internal/storage/azure_blobstore.go b/internal/storage/azure_blobstore.go
index ae3ffa5..02fa613 100644
--- a/internal/storage/azure_blobstore.go
+++ b/internal/storage/azure_blobstore.go
@@ -15,8 +15,10 @@
 package storage
 
 import (
+	"bytes"
 	"context"
 	"fmt"
+	"io"
 	"net/url"
 	"os"
 
@@ -95,3 +97,23 @@ func (s *AzureBlobstore) DeleteObject(ctx context.Context, container, name strin
 	}
 	return nil
 }
+
+// GetObject returns the contents for the given object. If the object does not
+// exist, it returns ErrNotFound.
+func (s *AzureBlobstore) GetObject(ctx context.Context, container, name string) ([]byte, error) {
+	blobURL := s.serviceURL.NewContainerURL(container).NewBlockBlobURL(name)
+	dr, err := blobURL.Download(ctx, 0, azblob.CountToEnd, azblob.BlobAccessConditions{}, false)
+	if err != nil {
+		return nil, fmt.Errorf("failed to download object: %w", err)
+	}
+
+	body := dr.Body(azblob.RetryReaderOptions{MaxRetryRequests: 5})
+	defer body.Close()
+
+	var b bytes.Buffer
+	if _, err := io.Copy(&b, body); err != nil {
+		return nil, fmt.Errorf("failed to read object: %w", err)
+	}
+
+	return b.Bytes(), nil
+}
diff --git a/internal/storage/filesystem_storage.go b/internal/storage/filesystem_storage.go
index 9088bab..e4f9721 100644
--- a/internal/storage/filesystem_storage.go
+++ b/internal/storage/filesystem_storage.go
@@ -55,3 +55,17 @@ func (s *FilesystemStorage) DeleteObject(ctx context.Context, folder, filename s
 	}
 	return nil
 }
+
+// GetObject returns the contents for the given object. If the object does not
+// exist, it returns ErrNotFound.
+func (s *FilesystemStorage) GetObject(ctx context.Context, folder, filename string) ([]byte, error) {
+	pth := filepath.Join(folder, filename)
+	b, err := ioutil.ReadFile(pth)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil, ErrNotFound
+		}
+		return nil, fmt.Errorf("failed to read file: %w", err)
+	}
+	return b, nil
+}
diff --git a/internal/storage/google_cloud_storage.go b/internal/storage/google_cloud_storage.go
index 009c055..20e16e6 100644
--- a/internal/storage/google_cloud_storage.go
+++ b/internal/storage/google_cloud_storage.go
@@ -15,9 +15,11 @@
 package storage
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
+	"io"
 
 	"cloud.google.com/go/storage"
 )
@@ -42,13 +44,13 @@ func NewGoogleCloudStorage(ctx context.Context) (Blobstore, error) {
 }
 
 // CreateObject creates a new cloud storage object or overwrites an existing one.
-func (gcs *GoogleCloudStorage) CreateObject(ctx context.Context, bucket, objectName string, contents []byte, cacheable bool) error {
+func (s *GoogleCloudStorage) CreateObject(ctx context.Context, bucket, objectName string, contents []byte, cacheable bool) error {
 	cacheControl := "public, max-age=86400"
 	if !cacheable {
 		cacheControl = "no-cache, max-age=0"
 	}
 
-	wc := gcs.client.Bucket(bucket).Object(objectName).NewWriter(ctx)
+	wc := s.client.Bucket(bucket).Object(objectName).NewWriter(ctx)
 	wc.Metadata = map[string]string{
 		"Cache-Control": cacheControl,
 	}
@@ -63,8 +65,8 @@ func (gcs *GoogleCloudStorage) CreateObject(ctx context.Context, bucket, objectN
 
 // DeleteObject deletes a cloud storage object, returns nil if the object was
 // successfully deleted, or of the object doesn't exist.
-func (gcs *GoogleCloudStorage) DeleteObject(ctx context.Context, bucket, objectName string) error {
-	if err := gcs.client.Bucket(bucket).Object(objectName).Delete(ctx); err != nil {
+func (s *GoogleCloudStorage) DeleteObject(ctx context.Context, bucket, objectName string) error {
+	if err := s.client.Bucket(bucket).Object(objectName).Delete(ctx); err != nil {
 		if errors.Is(err, storage.ErrObjectNotExist) {
 			// Object doesn't exist; presumably already deleted.
 			return nil
@@ -73,3 +75,22 @@ func (gcs *GoogleCloudStorage) DeleteObject(ctx context.Context, bucket, objectN
 	}
 	return nil
 }
+
+// GetObject returns the contents for the given object. If the object does not
+// exist, it returns ErrNotFound.
+func (s *GoogleCloudStorage) GetObject(ctx context.Context, bucket, object string) ([]byte, error) {
+	r, err := s.client.Bucket(bucket).Object(object).NewReader(ctx)
+	if err != nil {
+		if errors.Is(err, storage.ErrObjectNotExist) {
+			return nil, ErrNotFound
+		}
+	}
+	defer r.Close()
+
+	var b bytes.Buffer
+	if _, err := io.Copy(&b, r); err != nil {
+		return nil, fmt.Errorf("failed to download bytes: %w", err)
+	}
+
+	return b.Bytes(), nil
+}
diff --git a/internal/storage/memory.go b/internal/storage/memory.go
index f98ce00..83ea547 100644
--- a/internal/storage/memory.go
+++ b/internal/storage/memory.go
@@ -57,3 +57,17 @@ func (s *Memory) DeleteObject(_ context.Context, folder, filename string) error
 	delete(s.data, pth)
 	return nil
 }
+
+// GetObject returns the contents for the given object. If the object does not
+// exist, it returns ErrNotFound.
+func (s *Memory) GetObject(_ context.Context, folder, filename string) ([]byte, error) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	pth := path.Join(folder, filename)
+	v, ok := s.data[pth]
+	if !ok {
+		return nil, ErrNotFound
+	}
+	return v, nil
+}
diff --git a/internal/storage/noop.go b/internal/storage/noop.go
index 4ca2e08..04c6e5e 100644
--- a/internal/storage/noop.go
+++ b/internal/storage/noop.go
@@ -26,10 +26,14 @@ func NewNoop(ctx context.Context) (Blobstore, error) {
 	return &Noop{}, nil
 }
 
-func (s *Noop) CreateObject(ctx context.Context, folder, filename string, contents []byte, cacheable bool) error {
+func (s *Noop) CreateObject(_ context.Context, _, _ string, _ []byte, _ bool) error {
 	return nil
 }
 
-func (s *Noop) DeleteObject(ctx context.Context, folder, filename string) error {
+func (s *Noop) DeleteObject(_ context.Context, _, _ string) error {
 	return nil
 }
+
+func (s *Noop) GetObject(_ context.Context, _, _ string) ([]byte, error) {
+	return nil, nil
+}
diff --git a/internal/storage/storage.go b/internal/storage/storage.go
index 7e8f9dc..b12c7c6 100644
--- a/internal/storage/storage.go
+++ b/internal/storage/storage.go
@@ -20,13 +20,18 @@ import (
 	"fmt"
 )
 
+var ErrNotFound = fmt.Errorf("storage object not found")
+
 // Blobstore defines the minimum interface for a blob storage system.
 type Blobstore interface {
 	// CreateObject creates or overwrites an object in the storage system.
-	CreateObject(ctx context.Context, bucket, objectName string, contents []byte, cacheable bool) error
+	CreateObject(ctx context.Context, parent, name string, contents []byte, cacheable bool) error
+
+	// DeleteObject deletes an object or does nothing if the object doesn't exist.
+	DeleteObject(ctx context.Context, parent, bame string) error
 
-	// DeleteObject deltes an object or does nothing if the object doesn't exist.
-	DeleteObject(ctx context.Context, bucket, objectName string) error
+	// GetObject fetches the object's contents.
+	GetObject(ctx context.Context, parent, name string) ([]byte, error)
 }
 
 // BlobstoreFor returns the blob store for the given type, or an error if one
diff --git a/tools/monolith/main.go b/tools/monolith/main.go
deleted file mode 100644
index 191e2b4..0000000
--- a/tools/monolith/main.go
+++ /dev/null
@@ -1,35 +0,0 @@
-// Copyright 2020 Google LLC
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-// Package main runs all the server components at different URL paths.
-package main
-
-import (
-	"github.com/google/exposure-notifications-server/internal/interrupt"
-	"github.com/google/exposure-notifications-server/internal/logging"
-	"github.com/google/exposure-notifications-server/internal/monolith"
-
-	// Enable observability with distributed tracing and metrics.
-	_ "github.com/google/exposure-notifications-server/internal/observability"
-)
-
-func main() {
-	ctx, done := interrupt.Context()
-	defer done()
-
-	if err := monolith.RunServer(ctx); err != nil {
-		logger := logging.FromContext(ctx)
-		logger.Fatal(err)
-	}
-}


Test patch:
diff --git a/internal/integration/integration_test.go b/internal/integration/integration_test.go
index 2411cf4..d54997f 100644
--- a/internal/integration/integration_test.go
+++ b/internal/integration/integration_test.go
@@ -25,7 +25,6 @@ import (
 	"github.com/google/exposure-notifications-server/internal/database"
 	"github.com/google/exposure-notifications-server/internal/export"
 	"github.com/google/exposure-notifications-server/internal/federationin"
-	"github.com/google/exposure-notifications-server/internal/handlers"
 	"github.com/google/exposure-notifications-server/internal/publish"
 	"github.com/google/exposure-notifications-server/internal/secrets"
 	"github.com/google/exposure-notifications-server/internal/server"
@@ -141,7 +140,7 @@ func testServer(tb testing.TB) (*serverenv.ServerEnv, *http.Client) {
 	if err != nil {
 		tb.Fatal(err)
 	}
-	mux.HandleFunc("/publish", handlers.WithMinimumLatency(publishConfig.MinRequestDuration, publishHandler))
+	mux.Handle("/publish", publishHandler)
 
 	srv, err := server.New("")
 	if err != nil {
diff --git a/internal/integration/publish_test.go b/internal/integration/publish_test.go
index 4aff6f2..147dcd3 100644
--- a/internal/integration/publish_test.go
+++ b/internal/integration/publish_test.go
@@ -134,6 +134,11 @@ func TestPublish(t *testing.T) {
 	defer resp.Body.Close()
 
 	// TODO: verify export has the correct file
+	b, err := env.Blobstore().GetObject(ctx, "my-bucket", "index.txt")
+	if err != nil {
+		t.Fatal(err)
+	}
+	_ = b
 
 	// TODO: verify signature
 }
diff --git a/internal/storage/filesystem_storage_test.go b/internal/storage/filesystem_storage_test.go
index 0a9c129..a094c2b 100644
--- a/internal/storage/filesystem_storage_test.go
+++ b/internal/storage/filesystem_storage_test.go
@@ -24,6 +24,8 @@ import (
 )
 
 func TestFilesystemStorage_CreateObject(t *testing.T) {
+	t.Parallel()
+
 	tmp, err := ioutil.TempDir("", "")
 	if err != nil {
 		t.Fatal(err)
@@ -56,6 +58,8 @@ func TestFilesystemStorage_CreateObject(t *testing.T) {
 		tc := tc
 
 		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
 			ctx := context.Background()
 
 			storage, err := NewFilesystemStorage(ctx)
@@ -83,6 +87,8 @@ func TestFilesystemStorage_CreateObject(t *testing.T) {
 }
 
 func TestFilesystemStorage_DeleteObject(t *testing.T) {
+	t.Parallel()
+
 	f, err := ioutil.TempFile("", "")
 	if err != nil {
 		t.Fatal(err)
@@ -109,6 +115,8 @@ func TestFilesystemStorage_DeleteObject(t *testing.T) {
 		tc := tc
 
 		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
 			ctx := context.Background()
 
 			storage, err := NewFilesystemStorage(ctx)
@@ -122,3 +130,60 @@ func TestFilesystemStorage_DeleteObject(t *testing.T) {
 		})
 	}
 }
+
+func TestFilesystemStorage_GetObject(t *testing.T) {
+	t.Parallel()
+
+	f, err := ioutil.TempFile("", "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := f.Write([]byte("hello")); err != nil {
+		t.Fatal(err)
+	}
+
+	cases := []struct {
+		name     string
+		folder   string
+		filepath string
+		contents []byte
+		err      bool
+	}{
+		{
+			name:     "default",
+			folder:   filepath.Dir(f.Name()),
+			filepath: filepath.Base(f.Name()),
+			contents: []byte("hello"),
+		},
+		{
+			name:     "not_exist",
+			folder:   filepath.Dir(f.Name()),
+			filepath: "not-exist",
+			err:      true,
+		},
+	}
+
+	for _, tc := range cases {
+		tc := tc
+
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			ctx := context.Background()
+
+			storage, err := NewFilesystemStorage(ctx)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			b, err := storage.GetObject(ctx, tc.folder, tc.filepath)
+			if (err != nil) != tc.err {
+				t.Fatal(err)
+			}
+
+			if got, want := b, tc.contents; !bytes.Equal(got, want) {
+				t.Errorf("expected %v to be %v", got, want)
+			}
+		})
+	}
+}