Use GCS archive path date for row.Date (#896)

* Use gcs archive date for row.date
* Parse PathDate in GCSSource
* Add Date to etl.TestSource interface
* Update to bionic

Author: stephen-soltesz

.travis.yml

@@ -15,6 +15,7 @@
 # TODO(soltesz): Add deployment automation when fine-grained permissions are
 # possible.
 
+dist: bionic
 language: go
 go:
  - "1.13.8"

etl/etl.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"cloud.google.com/go/bigquery"
+	"cloud.google.com/go/civil"
 )
 
 // ProcessingError extends error to provide dataType and detail for metrics,
@@ -143,8 +144,9 @@ type TestSource interface {
 	NextTest(maxSize int64) (string, []byte, error)
 	Close() error
 
-	Detail() string // Detail for logs.
-	Type() string   // Data type for logs and metrics
+	Detail() string   // Detail for logs.
+	Type() string     // Data type for logs and metrics
+	Date() civil.Date // Date associated with test source
 }
 
 //========================================================================

parser/annotation.go

@@ -103,7 +103,7 @@ func (ap *AnnotationParser) ParseAndInsert(meta map[string]bigquery.Value, testN
 	// the given timestamp, regardless of the timestamp's timezone. Since we
 	// run our systems in UTC, all timestamps will be relative to UTC and as
 	// will these dates.
-	row.Date = civil.DateOf(raw.Timestamp)
+	row.Date = meta["date"].(civil.Date)
 
 	// Estimate the row size based on the input JSON size.
 	metrics.RowSizeHistogram.WithLabelValues(ap.TableName()).Observe(float64(len(test)))

parser/annotation_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 
 	"cloud.google.com/go/bigquery"
+	"cloud.google.com/go/civil"
 	"github.com/m-lab/etl/parser"
 	"github.com/m-lab/go/rtx"
 )
@@ -39,6 +40,7 @@ func TestAnnotationParser_ParseAndInsert(t *testing.T) {
 
 			meta := map[string]bigquery.Value{
 				"filename": "gs://mlab-test-bucket/ndt/ndt7/2020/03/18/" + tt.file,
+				"date":     civil.Date{Year: 2020, Month: 3, Day: 18},
 			}
 
 			if err := n.ParseAndInsert(meta, tt.file, data); (err != nil) != tt.wantErr {

parser/ndt7_result.go

@@ -93,7 +93,7 @@ func (dp *NDT7ResultParser) ParseAndInsert(meta map[string]bigquery.Value, testN
 	// the given timestamp, regardless of the timestamp's timezone. Since we
 	// run our systems in UTC, all timestamps will be relative to UTC and as
 	// will these dates.
-	row.Date = civil.DateOf(row.Raw.StartTime)
+	row.Date = meta["date"].(civil.Date)
 	if row.Raw.Download != nil {
 		row.A = downSummary(row.Raw.Download)
 	} else if row.Raw.Upload != nil {

parser/ndt7_result_test.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 
 	"cloud.google.com/go/bigquery"
+	"cloud.google.com/go/civil"
 	"github.com/go-test/deep"
 
 	"github.com/m-lab/etl/parser"
@@ -39,6 +40,7 @@ func TestNDT7ResultParser_ParseAndInsert(t *testing.T) {
 			}
 			meta := map[string]bigquery.Value{
 				"filename": "gs://mlab-test-bucket/ndt/ndt7/2020/03/18/ndt_ndt7_2020_03_18_20200318T003853.425987Z-ndt7-mlab3-syd03-ndt.tgz",
+				"date":     civil.Date{Year: 2020, Month: 3, Day: 18},
 			}
 
 			if err := n.ParseAndInsert(meta, tt.testName, resultData); (err != nil) != tt.wantErr {

parser/tcpinfo_test.go

@@ -15,6 +15,7 @@ import (
 	v2 "github.com/m-lab/annotation-service/api/v2"
 
 	"cloud.google.com/go/bigquery"
+	"cloud.google.com/go/civil"
 	"github.com/m-lab/annotation-service/api"
 	"github.com/m-lab/etl/etl"
 	"github.com/m-lab/etl/parser"
@@ -52,7 +53,8 @@ func fileSource(fn string) (etl.TestSource, error) {
 	tarReader := tar.NewReader(rdr)
 
 	timeout := 16 * time.Millisecond
-	return &storage.GCSSource{TarReader: tarReader, Closer: raw, RetryBaseTime: timeout, TableBase: "test"}, nil
+	return &storage.GCSSource{TarReader: tarReader, Closer: raw,
+		RetryBaseTime: timeout, TableBase: "test", PathDate: civil.Date{Year: 2020, Month: 6, Day: 11}}, nil
 }
 
 type fakeAnnotator struct{}

storage/storage.go

@@ -20,6 +20,7 @@ import (
 	"strings"
 	"time"
 
+	"cloud.google.com/go/civil"
 	gcs "cloud.google.com/go/storage"
 	"google.golang.org/api/option"
 
@@ -44,6 +45,7 @@ type GCSSource struct {
 	io.Closer                   // Closer interface to be provided by an embedded struct.
 	RetryBaseTime time.Duration // The base time for backoff and retry.
 	TableBase     string        // TableBase is BQ table associated with this source, or "invalid".
+	PathDate      civil.Date    // Date associated with YYYY/MM/DD in FilePath.
 }
 
 // Retrieve next file header.
@@ -125,6 +127,11 @@ func (src *GCSSource) Detail() string {
 	return src.FilePath
 }
 
+// Date returns a civil.Date associated with the GCSSource archive path.
+func (src *GCSSource) Date() civil.Date {
+	return src.PathDate
+}
+
 // NextTest reads the next test object from the tar file.
 // Skips reading contents of any file larger than maxSize, returning empty data
 // and storage.ErrOversizeFile.
@@ -218,25 +225,30 @@ var errNoClient = errors.New("client should be non-null")
 //
 // uri should be of form gs://bucket/filename.tar or gs://bucket/filename.tgz
 // FYI Using a persistent client saves about 80 msec, and 220 allocs, totalling 70kB.
-func NewTestSource(client *gcs.Client, uri string, label string) (etl.TestSource, error) {
+func NewTestSource(client *gcs.Client, dp etl.DataPath, label string) (etl.TestSource, error) {
 	if client == nil {
 		return nil, errNoClient
 	}
 	// For now only handle gcs paths.
-	if !strings.HasPrefix(uri, "gs://") {
-		return nil, errors.New("invalid file path: " + uri)
+	if !strings.HasPrefix(dp.URI, "gs://") {
+		return nil, errors.New("invalid file path: " + dp.URI)
 	}
-	parts := strings.SplitN(uri, "/", 4)
+	parts := strings.SplitN(dp.URI, "/", 4)
 	if len(parts) != 4 {
-		return nil, errors.New("invalid file path: " + uri)
+		return nil, errors.New("invalid file path: " + dp.URI)
 	}
 	bucket := parts[2]
 	fn := parts[3]
 
+	archiveDate, err := time.Parse("2006/01/02", dp.DatePath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse archive date path: %w", err)
+	}
+
 	// TODO - consider just always testing for valid gzip file.
 	if !(strings.HasSuffix(fn, ".tgz") || strings.HasSuffix(fn, ".tar") ||
 		strings.HasSuffix(fn, ".tar.gz")) {
-		return nil, errors.New("not tar or tgz: " + uri)
+		return nil, errors.New("not tar or tgz: " + dp.URI)
 	}
 
 	// TODO(prod) Evaluate whether this is long enough.
@@ -268,7 +280,15 @@ func NewTestSource(client *gcs.Client, uri string, label string) (etl.TestSource
 	tarReader := tar.NewReader(rdr)
 
 	baseTimeout := 16 * time.Millisecond
-	return &GCSSource{uri, tarReader, closer, baseTimeout, label}, nil
+	gcs := &GCSSource{
+		FilePath:      dp.URI,
+		TarReader:     tarReader,
+		Closer:        closer,
+		RetryBaseTime: baseTimeout,
+		TableBase:     label,
+		PathDate:      civil.DateOf(archiveDate),
+	}
+	return gcs, nil
 }
 
 // GetStorageClient provides a storage reader client.
@@ -304,7 +324,7 @@ func (sf *gcsSourceFactory) Get(ctx context.Context, dp etl.DataPath) (etl.TestS
 			http.StatusInternalServerError, etl.ErrBadDataType)
 	}
 
-	tr, err := NewTestSource(sf.client, dp.URI, label)
+	tr, err := NewTestSource(sf.client, dp, label)
 	if err != nil {
 		log.Printf("Error opening gcs file: %v", err)
 		// TODO - anything better we could do here?

storage/storage_test.go

@@ -18,8 +18,8 @@ import (
 )
 
 var testBucket = "mlab-testing.appspot.com"
-var tarFile = "gs://" + testBucket + "/test.tar"
-var tgzFile = "gs://" + testBucket + "/test.tgz"
+var tarFile = "gs://" + testBucket + "/ndt/ndt5/2020/06/11/20200611T123456.12345Z-ndt5-mlab1-foo01-ndt.tar"
+var tgzFile = "gs://" + testBucket + "/ndt/ndt5/2020/06/11/20200611T123456.12345Z-ndt5-mlab1-foo01-ndt.tgz"
 
 func assertGCSourceIsTestSource(in etl.TestSource) {
 	func(in etl.TestSource) {}(&GCSSource{})
@@ -46,7 +46,11 @@ func TestNewTarReader(t *testing.T) {
 	if testing.Short() {
 		t.Skip("Skipping tests that access GCS")
 	}
-	src, err := NewTestSource(client, tarFile, "label")
+	dpf, err := etl.ValidateTestPath(tarFile)
+	if err != nil {
+		t.Fatal(err)
+	}
+	src, err := NewTestSource(client, dpf, "label")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -68,7 +72,11 @@ func TestNewTarReaderGzip(t *testing.T) {
 	if testing.Short() {
 		t.Skip("Skipping tests that access GCS")
 	}
-	src, err := NewTestSource(client, tgzFile, "label")
+	dpf, err := etl.ValidateTestPath(tgzFile)
+	if err != nil {
+		t.Fatal(err)
+	}
+	src, err := NewTestSource(client, dpf, "label")
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -98,17 +106,25 @@ func init() {
 }
 
 func BenchmarkNewTarReader(b *testing.B) {
+	dpf, err := etl.ValidateTestPath(tarFile)
+	if err != nil {
+		b.Fatal(err)
+	}
 	for i := 0; i < b.N; i++ {
-		src, err := NewTestSource(client, tarFile, "label")
+		src, err := NewTestSource(client, dpf, "label")
 		if err == nil {
 			src.Close()
 		}
 	}
 }
 
 func BenchmarkNewTarReaderGzip(b *testing.B) {
+	dpf, err := etl.ValidateTestPath(tgzFile)
+	if err != nil {
+		b.Fatal(err)
+	}
 	for i := 0; i < b.N; i++ {
-		src, err := NewTestSource(client, tgzFile, "label")
+		src, err := NewTestSource(client, dpf, "label")
 		if err == nil {
 			src.Close()
 		}

task/task.go

@@ -49,6 +49,7 @@ func NewTask(filename string, src etl.TestSource, prsr etl.Parser) *Task {
 	meta["filename"] = filename
 	meta["parse_time"] = time.Now()
 	meta["attempt"] = 1
+	meta["date"] = src.Date()
 	t := Task{src, prsr, meta, DefaultMaxFileSize}
 	return &t
 }