upload: s3: add FailOnError config by tommyblue · Pull Request #27 · AdRoll/baker

tommyblue · 2020-08-12T10:23:18Z

❓ What

Add a FailOnError config value (default to false) that makes baker fail when an error happens instead of only log it.
The uploaders now return an error and it's managed in the topology

✅ Checklists

This section contains a list of checklists for common uses, please delete the checklists that are useless for your current use case (or add another checklist if your use case isn't covered yet).

Is there unit/integration test coverage for all new and/or changed functionality added in this PR?
Have the changes in this PR been functionally tested?
Has make gofmt-write been run on the code?
Has make govet been run on the code? Has the code been fixed accordingly to the output?
Have the changes been added to the CHANGELOG.md file?
Have the steps in CONTRIBUTING.md been followed to update a Go module?

arl

We just have changed the prototype of baker.Output interface so that Run returns an error.
We did this in order to avoid having the components themselves call os.Exit.

IMHO rather than adding a bunch of log.Fatal in the S3 uploader we should make baker.Upload.Run returns an error, and for now just handle that error in the Topology by calling os.Exit there, exactly as we did for errors returned by baker.Output.Run.

arl

LGTM

Only problem I see is with the test which do not remove the created folders and should use the machine temporary directory rather than the current directory (./upload) in this case.

arl · 2020-09-07T13:45:42Z

+func prepareUploadS3TestFolder(t *testing.T, numFiles int) (string, []string) {
+	t.Helper()
+
+	// Create a folder to store files to be uploaded
+	srcDir, err := ioutil.TempDir(".", "upload_s3_test")
+	if err != nil {
+		t.Fatalf("Can't setup test: %v", err)
+	}
+	defer os.Remove(srcDir)
+
+	// Write a bunch of files
+	var fnames []string
+	for i := 0; i < numFiles; i++ {
+		fname := filepath.Join(srcDir, fmt.Sprintf("test_file_%d", i))
+
+		if err := ioutil.WriteFile(fname, []byte("abc"), 0644); err != nil {
+			t.Fatalf("can't create temp file: %v", err)
+		}
+
+		fnames = append(fnames, fname)
+	}
+
+	return srcDir, fnames
+}


It seems this function has been extracted from Test_uploadDirectory. In that case Test_uploadDirectory could also use that test function.

However, in Test_uploadDirectory the defer was called at the end of the test, effectively removing the created temp folder. os.Remove only removes a directory if it's empty.

But there are 2 problems with the actual code:

in prepareUploadS3TestFolder the defer is called when exiting the function, but does nothing since the directory is not empty

worse, the directories remain after the tests and pollute the $REPO/upload directory. Those directories should be cleaned up

I think those directories should be (as it was the case in Test_uploadDirectory in the machine temp folder, so even in case of panic or anything, the test directories won't pollute the working tree.

To do so prepareUploadS3TestFolder should also return a callback (rmdir or something) called with a defer in the test function. This callback should call os.RemoveAll to ensure the directory gets removed even if the directory is not empty.

arl · 2020-09-07T13:54:43Z

+	srcDir, _ := prepareUploadS3TestFolder(t, numFiles)
+
+	mockUploadFn := func(uploader *s3manager.Uploader, bucket, prefix, localPath, fpath string) error {
+		time.Sleep(100 * time.Millisecond)


Why is this sleep here needed?

arl · 2020-09-07T13:56:21Z

+		// time.Sleep(100 * time.Millisecond)
+		// return errors.New("Fake error")


arl · 2020-09-07T13:56:45Z

+	tmpDir, fnames := prepareUploadS3TestFolder(t, 1)
+	fname := fnames[0]
+
+	stagingDir, err := ioutil.TempDir(".", "upload_s3_test_staging")


same as before: this directory is not cleanup after the test

arl · 2020-09-07T13:58:11Z

+	tmpDir, fnames := prepareUploadS3TestFolder(t, 1)
+	fname := fnames[0]
+
+	stagingDir, err := ioutil.TempDir(".", "upload_s3_test_staging")


It can be useful to use t.Name() for temporary folders related to a specific test

arl · 2020-09-07T14:05:43Z

-	err := filepath.Walk(u.Cfg.StagingPath, func(fpath string, info os.FileInfo, err error) error {
-		if err != nil {
-			return err
+	globFatalErr := atomic.Value{}


nit: calling this global may be a bit misleading and be confused with a global variable.
name proposition: mainErr or exitErr

arl · 2020-09-07T14:07:46Z

+		if globFatalErr.Load() != nil {
+			return globFatalErr.Load().(error)


Instead of calling Load twice we would store the result of Load in a variable and reuse it

arl · 2020-09-07T14:10:55Z

+		t.Fatalf("Can't setup test: %v", err)
+	}
+	mockUploadFn := func(uploader *s3manager.Uploader, bucket, prefix, localPath, fpath string) error {
+		time.Sleep(100 * time.Millisecond)


same, why is this needed?

tommyblue · 2020-09-08T09:36:41Z

All comments should have been addressed

arl

LGTM

Only thing is the addition, just for test, of a function pointer in the production code that could/should be avoided IMHO
and a nit :D

arl · 2020-09-08T09:52:23Z

 	totalerr int64
 	queuedn  int64
+
+	uploadFn func(uploader *s3manager.Uploader, bucket, prefix, localPath, fpath string) error


Is this have been added only for tests?
In general it's better to avoid modifying production code just to make it more testable, unless absolutely necessary.

However in this case there is already a mocked S3 service, instead of modifying the Upload itself, you could modify the mock to satisfy your needs, which are:

report the number of uploaded files (as required by Test_uploadDirectory). To do so just increment an atomic counter in the both CompleteMultipartUploadOutput and PutObjectOutput. Those are both the handlers for correct termination of a file upload.

optionally return an error (required by Test_uploadDirectoryError). For this it's probably enough to play with the HTTP response code

arl · 2020-09-08T09:53:07Z

+				if err := u.uploadFn(u.uploader, u.Cfg.Bucket, u.Cfg.Prefix, u.Cfg.StagingPath, fpath); err == nil {
 					atomic.AddInt64(&u.queuedn, int64(-1))
 					break
 				} else {


nit: since the previous if breaks, the else and curly braces can be removed

I did this way so that the err variable exists in both branches, but I can refactor the code assigning err before the if

as you wish, the code was already like that. it's just a nit so it's not important

1) #27 introduced an issue: the upload returned too soon and the last file in the logs path wasn't uploaded 2) we found an existing issue that caused the last file in the staging path not to be uploaded. Now the last upload is performed when the upload channel is closed

tommyblue force-pushed the upload_s3_fail_on_err branch from e7dcc99 to f8bf618 Compare August 12, 2020 10:24

tommyblue requested a review from arl August 12, 2020 10:24

tommyblue force-pushed the upload_s3_fail_on_err branch 2 times, most recently from 6c7ad7d to e2283d5 Compare August 12, 2020 10:26

arl reviewed Aug 12, 2020

View reviewed changes

Comment thread CHANGELOG.md Outdated

Comment thread upload/s3.go Outdated

arl reviewed Sep 7, 2020

View reviewed changes

Tommaso Visconti added 8 commits September 8, 2020 10:11

upload: s3: add FailOnError config

ae921f7

upload.Run() now returns an error

6f08d92

Add some tests

648e6ea

upload: s3: uploadDirectory exits at 1st error if ExitOnError

b906cca

s3 upload returns error and stops if ExitOnError

aae9b99

upload changelog

0e2acce

try to fix error on CI

b4e9414

fix behaviour on channel close

cdaf041

tommyblue force-pushed the upload_s3_fail_on_err branch from da2ca55 to cdaf041 Compare September 8, 2020 08:12

address comments

34894e7

arl approved these changes Sep 8, 2020

View reviewed changes

Tommaso Visconti added 3 commits September 8, 2020 15:23

tests refactoring to use S3Mock

633a156

nitting

240dd8f

fix TestS3Upload test

9b30a7d

tommyblue merged commit cc35bb7 into master Sep 9, 2020

tommyblue deleted the upload_s3_fail_on_err branch September 9, 2020 08:33

tommyblue mentioned this pull request Sep 17, 2020

Fix 2 issues with the upload.s3 component #40

Merged

6 tasks

		// time.Sleep(100 * time.Millisecond)
		// return errors.New("Fake error")

		if globFatalErr.Load() != nil {
		return globFatalErr.Load().(error)

Conversation

tommyblue commented Aug 12, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

❓ What

✅ Checklists

Uh oh!

arl left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

arl left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tommyblue commented Sep 8, 2020

Uh oh!

arl left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

tommyblue commented Aug 12, 2020 •

edited

Loading