Add sink.close (#898)

* add json annotations

* improve error logging

* add Sink.Close

* travis log each outcome

* improve comments on errors

* return Put errors

* disable manual GCS test

* improve ProcessAll error handling comments

* Add closer to task instead of parser

* move row.NullCloser

* update bindata

Author: gfr10598

.travis.yml

@@ -105,6 +105,7 @@ script:
 - for module in $MODULES; do
     go test -v -coverpkg=$COVER_PKGS -coverprofile=${module//\//_}.cov github.com/m-lab/etl/$module ;
     EC=$[ $EC || $? ] ;
+    echo "status $EC" ;
   done
 - echo "summary status $EC" ;
 - if [[ $EC != 0 ]]; then false; fi ;
@@ -119,6 +120,7 @@ script:
   for module in metrics ; do
     go test -v -coverpkg=$COVER_PKGS -coverprofile=${module//\//_}.cov github.com/m-lab/etl/$module -tags=integration ;
     EC=$[ $EC || $? ] ;
+    echo "summary $EC" ;
   done ;
   echo "summary status $EC" ;
   if [[ $EC != 0 ]]; then false; fi ;

bq/insert.go

@@ -409,6 +409,9 @@ func (in *sink) Commit(rows []interface{}, label string) (int, error) {
 	return in.flushSlice(rows, label, label)
 }
 
+// Close synchronizes on the tokens, and closes the backing file.
+func (in *sink) Close() error { return nil }
+
 // flushSlice flushes a slice of rows to BigQuery.
 // It returns the number of rows successfully committed.
 // It is NOT threadsafe.
@@ -571,7 +574,7 @@ func (sf *bqSinkFactory) Get(
 
 	client, err := GetClient(pdt.Project)
 	if err != nil {
-		return nil, factory.NewError(path.DataType, "StorageClient",
+		return nil, factory.NewError(path.DataType, "BQClient",
 			http.StatusInternalServerError, err)
 	}
 

parser/annotation.go

@@ -109,7 +109,9 @@ func (ap *AnnotationParser) ParseAndInsert(meta map[string]bigquery.Value, testN
 	metrics.RowSizeHistogram.WithLabelValues(ap.TableName()).Observe(float64(len(test)))
 
 	// Insert the row.
-	ap.Base.Put(&row)
+	if err = ap.Base.Put(&row); err != nil {
+		return err
+	}
 
 	// Count successful inserts.
 	metrics.TestCount.WithLabelValues(ap.TableName(), "annotation", "ok").Inc()

parser/ndt5_result.go

@@ -109,7 +109,9 @@ func (dp *NDT5ResultParser) ParseAndInsert(meta map[string]bigquery.Value, testN
 		metrics.RowSizeHistogram.WithLabelValues(
 			dp.TableName()).Observe(float64(len(test)))
 
-		dp.Base.Put(&stats)
+		if err = dp.Base.Put(&stats); err != nil {
+			return err
+		}
 		// Count successful inserts.
 		metrics.TestCount.WithLabelValues(dp.TableName(), "ndt5_result", "ok").Inc()
 	}

parser/ndt7_result.go

@@ -109,8 +109,10 @@ func (dp *NDT7ResultParser) ParseAndInsert(meta map[string]bigquery.Value, testN
 		dp.TableName()).Observe(float64(len(test)))
 
 	// Insert the row.
-	dp.Base.Put(&row)
-
+	err = dp.Base.Put(&row)
+	if err != nil {
+		return err
+	}
 	// Count successful inserts.
 	metrics.TestCount.WithLabelValues(dp.TableName(), "ndt7_result", "ok").Inc()
 	return nil

parser/tcpinfo.go

@@ -190,7 +190,12 @@ func (p *TCPInfoParser) ParseAndInsert(fileMetadata map[string]bigquery.Value, t
 		}
 	}
 
-	p.Put(&row)
+	// For GCS, errors here are generally fatal.
+	// For BigQuery, errors here could be fatal, or could be due to quota exceeded.
+	err = p.Put(&row)
+	if err != nil {
+		return err
+	}
 	metrics.TestCount.WithLabelValues(p.TableName(), "tcpinfo", "ok").Inc()
 	return nil
 }

parser/tcpinfo_test.go

@@ -12,6 +12,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/googleapis/google-cloud-go-testing/storage/stiface"
 	v2 "github.com/m-lab/annotation-service/api/v2"
 
 	"cloud.google.com/go/bigquery"
@@ -94,6 +95,10 @@ func (in *inMemorySink) Commit(data []interface{}, label string) (int, error) {
 	return len(data), nil
 }
 
+func (in *inMemorySink) Close() error {
+	return nil
+}
+
 func (in *inMemorySink) Flush() error {
 	in.committed = len(in.data)
 	return nil
@@ -102,6 +107,10 @@ func (in *inMemorySink) Committed() int {
 	return in.committed
 }
 
+type nullCloser struct{}
+
+func (nc nullCloser) Close() error { return nil }
+
 // NOTE: This uses a fake annotator which returns no annotations.
 // TODO: This test seems to be flakey in travis - sometimes only 357 tests instead of 362
 func TestTCPParser(t *testing.T) {
@@ -118,7 +127,7 @@ func TestTCPParser(t *testing.T) {
 	// Inject fake inserter and annotator
 	ins := newInMemorySink()
 	p := parser.NewTCPInfoParser(ins, "test", "_suffix", &fakeAnnotator{})
-	task := task.NewTask(filename, src, p)
+	task := task.NewTask(filename, src, p, nullCloser{})
 
 	startDecode := time.Now()
 	n, err := task.ProcessAllTests()
@@ -228,7 +237,7 @@ func TestTCPTask(t *testing.T) {
 		t.Fatal("Failed reading testdata from", filename)
 	}
 
-	task := task.NewTask(filename, src, p)
+	task := task.NewTask(filename, src, p, &nullCloser{})
 
 	n, err := task.ProcessAllTests()
 	if err != nil {
@@ -253,7 +262,7 @@ func TestBQSaver(t *testing.T) {
 		t.Fatal("Failed reading testdata from", filename)
 	}
 
-	task := task.NewTask(filename, src, p)
+	task := task.NewTask(filename, src, p, &nullCloser{})
 
 	_, err = task.ProcessAllTests()
 	if err != nil {
@@ -272,6 +281,51 @@ func TestBQSaver(t *testing.T) {
 	}
 }
 
+// This test writes 364 rows to a json file in GCS.
+// The rows can then be loaded into a BQ table, using the schema in testdata, like:
+// bq load --source_format=NEWLINE_DELIMITED_JSON \
+//    mlab-sandbox:gfr.small_tcpinfo gs://archive-mlab-testing/gfr/tcpinfo.json ./schema.json
+// Recommend commenting out snapshots in tcpinfo.go.
+func TestTaskToGCS(t *testing.T) {
+	t.Skip("Skipping test intended for manual experimentation")
+
+	os.Setenv("RELEASE_TAG", "foobar")
+	parser.InitParserVersionForTest()
+
+	c, err := storage.GetStorageClient(true)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	rw, err := storage.NewRowWriter(context.Background(), stiface.AdaptClient(c), "archive-mlab-testing", "gfr/tcpinfo.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	// Inject fake inserter and annotator
+	p := parser.NewTCPInfoParser(rw, "test", "suffix", &fakeAnnotator{})
+
+	filename := "testdata/20190516T013026.744845Z-tcpinfo-mlab4-arn02-ndt.tgz"
+	src, err := fileSource(filename)
+	if err != nil {
+		t.Fatal("Failed reading testdata from", filename)
+	}
+
+	task := task.NewTask(filename, src, p, &nullCloser{})
+
+	n, err := task.ProcessAllTests()
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = rw.Close()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if n != 364 {
+		t.Errorf("Expected ProcessAllTests to handle %d files, but it handled %d.\n", 364, n)
+	}
+}
+
 func BenchmarkTCPParser(b *testing.B) {
 	os.Setenv("RELEASE_TAG", "foobar")
 	parser.InitParserVersionForTest()
@@ -288,7 +342,7 @@ func BenchmarkTCPParser(b *testing.B) {
 			b.Fatalf("cannot read testdata.")
 		}
 
-		task := task.NewTask(filename, src, p)
+		task := task.NewTask(filename, src, p, &nullCloser{})
 
 		n, err = task.ProcessAllTests()
 		if err != nil {

row/row.go

@@ -6,6 +6,7 @@ package row
 import (
 	"context"
 	"errors"
+	"io"
 	"log"
 	"sync"
 	"time"
@@ -109,6 +110,7 @@ type HasStats interface {
 // Implementations should be threadsafe.
 type Sink interface {
 	Commit(rows []interface{}, label string) (int, error)
+	io.Closer
 }
 
 // Buffer provides all basic functionality generally needed for buffering, annotating, and inserting
@@ -313,9 +315,14 @@ func (pb *Base) TaskError() error {
 	return nil
 }
 
+var logAnnError = logx.NewLogEvery(nil, 60*time.Second)
+
 func (pb *Base) commit(rows []interface{}) error {
-	// TODO - care about error?
-	_ = pb.ann.Annotate(rows, pb.label)
+	err := pb.ann.Annotate(rows, pb.label)
+	if err != nil {
+		logAnnError.Println("annotation: ", err)
+	}
+
 	// TODO do we need these to be done in order.
 	// This is synchronous, blocking, and thread safe.
 	done, err := pb.sink.Commit(rows, pb.label)
@@ -342,7 +349,7 @@ func (pb *Base) Flush() error {
 // sequential calls to Put.  However, once a block of rows is submitted
 // to pb.commit, it should be written in the same order to the Sink.
 // TODO improve Annotatable architecture.
-func (pb *Base) Put(row Annotatable) {
+func (pb *Base) Put(row Annotatable) error {
 	rows := pb.buf.Append(row)
 	pb.stats.Inc()
 
@@ -351,9 +358,11 @@ func (pb *Base) Put(row Annotatable) {
 		// TODO consider making this asynchronous.
 		err := pb.commit(rows)
 		if err != nil {
-			log.Println(err)
+			log.Println(pb.label, err)
+			return err
 		}
 	}
+	return nil
 }
 
 // NullAnnotator satisfies the Annotatable interface without actually doing

row/row_test.go

@@ -72,6 +72,8 @@ func (in *inMemorySink) Commit(data []interface{}, label string) (int, error) {
 	return len(data), nil
 }
 
+func (in *inMemorySink) Close() error { return nil }
+
 func TestBase(t *testing.T) {
 	ins := &inMemorySink{}
 

schema/annotation.go

@@ -13,11 +13,11 @@ import (
 // AnnotationRow defines the BQ schema using 'Standard Columns' conventions for
 // the annotation datatype produced by the uuid-annotator.
 type AnnotationRow struct {
-	UUID   string                      `bigquery:"id"` // NOTE: there is no 'a' record for AnnotationRows.
-	Server annotator.ServerAnnotations `bigquery:"server"`
-	Client annotator.ClientAnnotations `bigquery:"client"`
-	Parser ParseInfo                   `bigquery:"parser"`
-	Date   civil.Date                  `bigquery:"date"`
+	UUID   string                      `bigquery:"id" json:"id"` // NOTE: there is no 'a' record for AnnotationRows.
+	Server annotator.ServerAnnotations `bigquery:"server" json:"server"`
+	Client annotator.ClientAnnotations `bigquery:"client" json:"client"`
+	Parser ParseInfo                   `bigquery:"parser" json:"parser"`
+	Date   civil.Date                  `bigquery:"date" json:"date"`
 
 	// NOTE: there is no 'Raw' field for annotation datatypes because the
 	// uuid-annotator output schema was designed to be used directly by the parser.