Merge pull request #206 from m-lab/break

Fix break bug.

Author: gfr10598

storage/storage.go

@@ -34,8 +34,9 @@ type TarReader interface {
 }
 
 type ETLSource struct {
-	TarReader // TarReader interface provided by an embedded struct.
-	io.Closer // Closer interface to be provided by an embedded struct.
+	TarReader                   // TarReader interface provided by an embedded struct.
+	io.Closer                   // Closer interface to be provided by an embedded struct.
+	RetryBaseTime time.Duration // The base time for backoff and retry.
 }
 
 // Retrieve next file header.
@@ -121,10 +122,11 @@ func (rr *ETLSource) NextTest(maxSize int64) (string, []byte, error) {
 	var data []byte
 	var h *tar.Header
 
-	// Last trial will be after total delay of 16ms + 32ms + ... + 8192ms,
-	// or about 15 seconds.
+	// With default RetryBaseTime, the last trial will be after total delay of
+	// 16ms + 32ms + ... + 8192ms, or about 15 seconds.
+	// TODO - should add a random element to the backoff?
 	trial := 0
-	delay := 16 * time.Millisecond
+	delay := rr.RetryBaseTime
 	for {
 		trial++
 		var retry bool
@@ -150,7 +152,7 @@ func (rr *ETLSource) NextTest(maxSize int64) (string, []byte, error) {
 	}
 
 	trial = 0
-	delay = 16 * time.Millisecond
+	delay = rr.RetryBaseTime
 	for {
 		trial++
 		var retry bool
@@ -236,7 +238,8 @@ func NewETLSource(client *http.Client, uri string) (*ETLSource, error) {
 	}
 	tarReader := tar.NewReader(rdr)
 
-	return &ETLSource{tarReader, closer}, nil
+	timeout := 16 * time.Millisecond
+	return &ETLSource{tarReader, closer, timeout}, nil
 }
 
 // Create a storage reader client.

storage/storage_test.go

@@ -23,7 +23,7 @@ func TestNewTarReader(t *testing.T) {
 	defer src.Close()
 
 	count := 0
-	for _, _, err := src.NextTest(); err != io.EOF; _, _, err = src.NextTest() {
+	for _, _, err := src.NextTest(10000000); err != io.EOF; _, _, err = src.NextTest(10000000) {
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -42,7 +42,7 @@ func TestNewTarReaderGzip(t *testing.T) {
 	defer src.Close()
 
 	count := 0
-	for _, _, err := src.NextTest(); err != io.EOF; _, _, err = src.NextTest() {
+	for _, _, err := src.NextTest(10000000); err != io.EOF; _, _, err = src.NextTest(10000000) {
 		if err != nil {
 			t.Fatal(err)
 		}

task/task.go

@@ -30,7 +30,7 @@ type Task struct {
 	etl.Parser         // Parser to parse the tests.
 
 	meta        map[string]bigquery.Value // Metadata about this task.
-	maxFileSize int64  // Max file size to avoid OOM.
+	maxFileSize int64                     // Max file size to avoid OOM.
 }
 
 // NewTask constructs a task, injecting the source and the parser.
@@ -61,19 +61,20 @@ func (tt *Task) ProcessAllTests() (int, error) {
 	var err error
 	// Read each file from the tar
 
+OUTER:
 	for testname, data, err = tt.NextTest(tt.maxFileSize); err != io.EOF; testname, data, err = tt.NextTest(tt.maxFileSize) {
 		files++
 		if err != nil {
 			switch {
 			case err == io.EOF:
-				break
+				break OUTER
 			case err == storage.OVERSIZE_FILE:
 				log.Printf("filename:%s testname:%s files:%d, duration:%v err:%v",
 					tt.meta["filename"], testname, files,
 					time.Since(tt.meta["parse_time"].(time.Time)), err)
 				metrics.TestCount.WithLabelValues(
 					tt.Parser.TableName(), "unknown", "oversize file").Inc()
-				continue
+				continue OUTER
 			default:
 				// We are seeing several of these per hour, a little more than
 				// one in one thousand files.  duration varies from 10 seconds
@@ -93,7 +94,7 @@ func (tt *Task) ProcessAllTests() (int, error) {
 					tt.Parser.TableName(), "unknown", "unrecovered").Inc()
 				// Since we don't understand these errors, safest thing to do is
 				// stop processing the tar file (and task).
-				break
+				break OUTER
 			}
 		}
 		if data == nil {

task/task_test.go

@@ -5,12 +5,15 @@ package task_test
 import (
 	"archive/tar"
 	"bytes"
+	"errors"
 	"fmt"
 	"reflect"
 	"testing"
 
 	"cloud.google.com/go/bigquery"
 
+	"time"
+
 	"github.com/m-lab/etl/parser"
 	"github.com/m-lab/etl/storage" // TODO - would be better not to have this.
 	"github.com/m-lab/etl/task"
@@ -59,7 +62,7 @@ func MakeTestSource(t *testing.T) *storage.ETLSource {
 		t.Fatal(err)
 	}
 
-	return &storage.ETLSource{tar.NewReader(b), NullCloser{}}
+	return &storage.ETLSource{tar.NewReader(b), NullCloser{}, time.Millisecond}
 }
 
 type TestParser struct {
@@ -83,8 +86,37 @@ func (tp *TestParser) ParseAndInsert(meta map[string]bigquery.Value, testName st
 	return nil
 }
 
-// TODO(dev) - add unit tests for tgz and tar.gz files
-// TODO(dev) - add good comments
+type badSource struct{}
+
+func (bs *badSource) Next() (*tar.Header, error) {
+	return nil, errors.New("Random Error")
+}
+func (bs *badSource) Read(b []byte) (int, error) {
+	return 0, errors.New("Read error")
+}
+
+// TODO - this test is very slow, because it triggers the backoff and retry mechanism.
+func TestBadTarFileInput(t *testing.T) {
+	rdr := &storage.ETLSource{&badSource{}, NullCloser{}, time.Millisecond}
+
+	tp := &TestParser{}
+
+	// Among other things, this requires that tp implements etl.Parser.
+	tt := task.NewTask("filename", rdr, tp)
+	fc, err := tt.ProcessAllTests()
+	if err.Error() != "Random Error" {
+		t.Error("Expected Random Error, but got %v", err)
+	}
+	// Should see 1 files.
+	if fc != 1 {
+		t.Error("Expected 1 file: ", fc)
+	}
+	// ... but process none.
+	if len(tp.files) != 0 {
+		t.Error("Should have processed no files: ", len(tp.files))
+	}
+}
+
 func TestTarFileInput(t *testing.T) {
 	rdr := MakeTestSource(t)