Merge pull request #111 from discentem/pr111

consolidate metadata writing

Author: discentem

WORKSPACE

@@ -96,6 +96,20 @@ go_repository(
     version = "v0.3.0",
 )
 
+go_repository(
+    name = "com_github_hashicorp_go_multierror",
+    importpath = "github.com/hashicorp/go-multierror",
+    sum = "h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo=",
+    version = "v1.1.1",
+)
+
+go_repository(
+    name = "com_github_hashicorp_errwrap",
+    importpath = "github.com/hashicorp/errwrap",
+    sum = "h1:OxrOeh75EUXMY8TBjag2fzXGZ40LB6IKw45YeGUDY2I=",
+    version = "v1.1.0",
+)
+
 bazel_skylib_workspace()
 
 load("//:repositories.bzl", "go_repositories")

go.mod

@@ -54,6 +54,8 @@ require (
 	github.com/googleapis/gax-go/v2 v2.8.0 // indirect
 	github.com/gorilla/handlers v1.5.1 // indirect
 	github.com/gorilla/mux v1.8.0 // indirect
+	github.com/hashicorp/errwrap v1.0.0 // indirect
+	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/hashicorp/hcl v1.0.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect

go.sum

@@ -230,6 +230,10 @@ github.com/gorilla/handlers v1.5.1/go.mod h1:t8XrUpc4KVXb7HGyJ4/cEnwQiaxrX/hz1Zv
 github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI=
 github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
 github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw=
+github.com/hashicorp/errwrap v1.0.0 h1:hLrqtEDnRye3+sgx6z4qVLNuviH3MR5aQ0ykNJa/UYA=
+github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
+github.com/hashicorp/go-multierror v1.1.1 h1:H5DkEtf6CXdFp0N0Em5UCwQpXMWke8IA0+lD48awMYo=
+github.com/hashicorp/go-multierror v1.1.1/go.mod h1:iw975J/qwKPdAO1clOe2L8331t/9/fmwbPZ6JB6eMoM=
 github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
 github.com/hashicorp/hcl v1.0.0 h1:0Anlzjpi4vEasTeNFn2mLJgTSwt0+6sfsiTG8qcWGx4=

internal/metadata/metadata.go

@@ -13,6 +13,8 @@ import (
 
 const MetadataFileExtension string = "cfile"
 
+var ErrFileExtensionEmpty = fmt.Errorf("options.MetadatafileExtension cannot be %q", "")
+
 type ObjectMetaData struct {
 	Name         string    `json:"name"`
 	Checksum     string    `json:"checksum"`

internal/stores/BUILD.bazel

@@ -17,6 +17,7 @@ go_library(
         "@com_github_aws_aws_sdk_go_v2_feature_s3_manager//:manager",
         "@com_github_aws_aws_sdk_go_v2_service_s3//:s3",
         "@com_github_google_logger//:logger",
+        "@com_github_hashicorp_go_multierror//:go-multierror",
         "@com_github_spf13_afero//:afero",
         "@com_google_cloud_go_storage//:storage",
         "@org_golang_google_api//option",

internal/stores/README.md

@@ -4,7 +4,7 @@ Stores are the interface by which metadata is managed and files are uploaded/ret
 ## Upload
 When uploading files, the following items should be handled in the implementation:
 
-* Metadata should be generated from the file and written alongside the binary on disk
+* Metadata should be generated from the file and written alongside the binary on the filesystem.
 * Upload the file to the bucket
 
 ## Retrieve

internal/stores/gcs.go

@@ -2,7 +2,6 @@ package stores
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
@@ -14,6 +13,7 @@ import (
 	gcsStorage "cloud.google.com/go/storage"
 	"github.com/discentem/cavorite/internal/metadata"
 	"github.com/google/logger"
+	"github.com/hashicorp/go-multierror"
 	"github.com/spf13/afero"
 	"google.golang.org/api/option"
 )
@@ -63,12 +63,13 @@ func (s *GCSStore) GetOptions() Options {
 	return s.Options
 }
 
+func (s *GCSStore) GetFsys() afero.Fs {
+	return s.fsys
+}
+
 // Upload generates the metadata, writes it s.fsys and uploads the file to the GCS bucket
 func (s *GCSStore) Upload(ctx context.Context, objects ...string) error {
-	if s.Options.MetadataFileExtension == "" {
-		return ErrMetadataFileExtensionEmpty
-	}
-
+	var multErr error
 	for _, o := range objects {
 		logger.V(2).Infof("Object: %s\n", o)
 		f, err := s.fsys.Open(o)
@@ -77,22 +78,13 @@ func (s *GCSStore) Upload(ctx context.Context, objects ...string) error {
 		}
 		defer f.Close()
 
-		// Generate cavorite metadata
-		m, err := metadata.GenerateFromFile(f)
-		if err != nil {
-			return err
-		}
-		logger.V(2).Infof("%s has a checksum of %q", o, m.Checksum)
-		// convert metadata to json
-		blob, err := json.MarshalIndent(m, "", " ")
+		// cleanupFn is function that can be called if
+		// uploading to s3 fails. cleanupFn deletes the cfile
+		// so that we don't retain a cfile without a corresponding binary
+		cleanupFn, err := WriteMetadataToFsys(s, o, f)
 		if err != nil {
 			return err
 		}
-		// Write metadata to fsys
-		metadataPath := fmt.Sprintf("%s.%s", o, s.Options.MetadataFileExtension)
-		if err := afero.WriteFile(s.fsys, metadataPath, blob, 0644); err != nil {
-			return err
-		}
 
 		// ToDo(natewalck) Expose this timeout as a setting
 		ctx, cancel := context.WithTimeout(ctx, time.Second*1800)
@@ -104,22 +96,36 @@ func (s *GCSStore) Upload(ctx context.Context, objects ...string) error {
 		wc := gcsObject.If(storage.Conditions{DoesNotExist: true}).NewWriter(ctx)
 
 		// Reset to the start of the file because metadata generation has already read it once
-		_, err = f.Seek(0, io.SeekStart)
+		_, seekErr := f.Seek(0, io.SeekStart)
 		if err != nil {
-			return err
+			// seek failed, add this failure to multErr
+			multErr = multierror.Append(multErr, fmt.Errorf("f.Seek() error: %w", seekErr))
+			if cleanupErr := cleanupFn(); err != nil {
+				// cleanup also failed, add to multErr
+				multErr = multierror.Append(multErr, fmt.Errorf("cleanupFn() error: %w", cleanupErr))
+			}
+			// return multiple errors
+			return multErr
 		}
 		_, err = io.Copy(wc, f)
 		if err != nil {
-			logger.V(2).Infof("Failed to upload %s", o)
-			return err
+			multErr = multierror.Append(multErr, fmt.Errorf("io.Copy() error: %w", err))
+			if cleanupErr := cleanupFn(); err != nil {
+				multErr = multierror.Append(multErr, fmt.Errorf("cleanupFn() error: %w", cleanupErr))
+			}
+			return multErr
 		}
 
 		if err := wc.Close(); err != nil {
 			// Error will contain this string if the DoesNotExist condition isn't met
 			if strings.Contains(err.Error(), "conditionNotMet") {
 				logger.Infof("%s already exists, skipping...", o)
 			} else {
-				return err
+				multErr = multierror.Append(multErr, fmt.Errorf("wc.Close() err: %w", err))
+				if cleanupErr := cleanupFn(); err != nil {
+					multErr = multierror.Append(multErr, fmt.Errorf("cleanupFn() error: %w", cleanupErr))
+				}
+				return multErr
 			}
 		}
 	}

internal/stores/s3.go

@@ -1,13 +1,10 @@
 package stores
 
 import (
-	"bytes"
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
-	"os"
 	"path"
 	"path/filepath"
 	"strings"
@@ -130,45 +127,29 @@ func (s *S3Store) GetOptions() Options {
 	return s.Options
 }
 
-// TODO(discentem): #34 largely copy-pasted from stores/local/local.go. Can be consolidated
+func (s *S3Store) GetFsys() afero.Fs {
+	return s.fsys
+}
+
 // Upload generates the metadata, writes it to disk and uploads the file to the S3 bucket
 func (s *S3Store) Upload(ctx context.Context, objects ...string) error {
-	if s.Options.MetadataFileExtension == "" {
-		return ErrMetadataFileExtensionEmpty
-	}
-
 	for _, o := range objects {
-		logger.V(2).Infof("object: %s", o)
 		f, err := s.fsys.Open(o)
 		if err != nil {
 			return err
 		}
-		defer f.Close()
-		// TODO(discentem): probably inefficient, reading entire file into memory
-		b, err := afero.ReadFile(s.fsys, o)
+		// cleanupFn is function that can be called if
+		// uploading to s3 fails. cleanupFn deletes the cfile
+		// so that we don't retain a cfile without a corresponding binary
+		cleanupFn, err := WriteMetadataToFsys(s, o, f)
 		if err != nil {
 			return err
 		}
-
-		// generate metadata
-		m, err := metadata.GenerateFromFile(f)
+		_, err = f.Seek(0, io.SeekStart)
 		if err != nil {
-			return err
-		}
-		logger.V(2).Infof("%s has a checksum of %q", o, m.Checksum)
-		// convert metadata to json
-		blob, err := json.MarshalIndent(m, "", " ")
-		if err != nil {
-			return err
-		}
-		// Create path for metadata if it doesn't already exist
-		if err := s.fsys.MkdirAll(filepath.Dir(filepath.Dir(o)), os.ModePerm); err != nil {
-			return err
-		}
-		// Write metadata to disk
-		metadataPath := fmt.Sprintf("%s.%s", o, s.Options.MetadataFileExtension)
-		logger.V(2).Infof("writing metadata to %s", metadataPath)
-		if err := afero.WriteFile(s.fsys, metadataPath, blob, 0644); err != nil {
+			if err := cleanupFn(); err != nil {
+				return err
+			}
 			return err
 		}
 
@@ -177,13 +158,19 @@ func (s *S3Store) Upload(ctx context.Context, objects ...string) error {
 		obj := s3.PutObjectInput{
 			Bucket: aws.String(buck),
 			Key:    &o,
-			Body:   bytes.NewReader(b),
+			Body:   f,
 		}
 		out, err := s.s3Uploader.Upload(ctx, &obj)
 		if err != nil {
+			if err := cleanupFn(); err != nil {
+				return fmt.Errorf("cleanup() failed after Upload failure: %w", err)
+			}
 			logger.Error(out)
 			return err
 		}
+		if err := f.Close(); err != nil {
+			return err
+		}
 	}
 	return nil
 }

internal/stores/stores.go

@@ -2,11 +2,15 @@ package stores
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
+	"fmt"
 	"os"
 	"path/filepath"
 	"strings"
 
+	"github.com/discentem/cavorite/internal/metadata"
+	"github.com/google/logger"
 	"github.com/spf13/afero"
 )
 
@@ -26,6 +30,7 @@ type Store interface {
 	Upload(ctx context.Context, objects ...string) error
 	Retrieve(ctx context.Context, objects ...string) error
 	GetOptions() Options
+	GetFsys() afero.Fs
 }
 
 func openOrCreateFile(fsys afero.Fs, filename string) (afero.File, error) {
@@ -39,3 +44,37 @@ func openOrCreateFile(fsys afero.Fs, filename string) (afero.File, error) {
 func inferObjPath(cfilePath string) string {
 	return strings.TrimSuffix(cfilePath, filepath.Ext(cfilePath))
 }
+
+// WriteMetadata generates Cavorite metadata for obj and writes it to s.Fsys
+func WriteMetadataToFsys(s Store, obj string, f afero.File) (cleanup func() error, err error) {
+	opts := s.GetOptions()
+	if opts.MetadataFileExtension == "" {
+		return nil, metadata.ErrFileExtensionEmpty
+	}
+	fsys := s.GetFsys()
+	logger.V(2).Infof("object: %s", obj)
+
+	// generate metadata
+	m, err := metadata.GenerateFromFile(f)
+	if err != nil {
+		return nil, err
+	}
+	logger.V(2).Infof("%s has a checksum of %q", obj, m.Checksum)
+	// convert metadata to json
+	blob, err := json.MarshalIndent(m, "", " ")
+	if err != nil {
+		return nil, err
+	}
+	// Write metadata to disk
+	metadataPath := fmt.Sprintf("%s.%s", obj, opts.MetadataFileExtension)
+	logger.V(2).Infof("writing metadata to %s", metadataPath)
+	if err := afero.WriteFile(fsys, metadataPath, blob, 0644); err != nil {
+		return nil, err
+	}
+
+	cleanup = func() error {
+		return fsys.Remove(metadataPath)
+	}
+
+	return cleanup, nil
+}