Migrate to go/cloud/gcs (#936)

* use gcsfake

* migrate to go/cloud/gcs

Author: gfr10598

active/active.go

@@ -17,12 +17,11 @@ import (
 	"sync"
 	"time"
 
-	"github.com/googleapis/google-cloud-go-testing/storage/stiface"
 	"google.golang.org/api/iterator"
 
 	"cloud.google.com/go/storage"
-	"github.com/m-lab/etl/cloud/gcs"
 	"github.com/m-lab/etl/metrics"
+	"github.com/m-lab/go/cloud/gcs"
 	"github.com/m-lab/go/logx"
 )
 
@@ -46,9 +45,9 @@ type FileLister func(ctx context.Context) ([]*storage.ObjectAttrs, int64, error)
 // FileListerFunc creates a function that returns a slice of *storage.ObjectAttrs.
 // On certain GCS errors, it may return partial result and an error.
 // TODO - consider moving this to GardenerAPI.
-func FileListerFunc(sc stiface.Client, prefix string, filter *regexp.Regexp) FileLister {
+func FileListerFunc(bh *gcs.BucketHandle, prefix string, filter *regexp.Regexp) FileLister {
 	return func(ctx context.Context) ([]*storage.ObjectAttrs, int64, error) {
-		return gcs.GetFilesSince(ctx, sc, prefix, filter, time.Time{})
+		return bh.GetFilesSince(ctx, prefix, filter, time.Time{})
 	}
 }
 

active/active_test.go

@@ -10,14 +10,17 @@ import (
 	"testing"
 	"time"
 
+	"cloud.google.com/go/storage"
 	"github.com/googleapis/google-cloud-go-testing/storage/stiface"
-	"github.com/m-lab/go/logx"
 	"golang.org/x/sync/errgroup"
 	"google.golang.org/api/iterator"
 
-	"cloud.google.com/go/storage"
 	"github.com/m-lab/etl/active"
-	"github.com/m-lab/go/cloudtest"
+	"github.com/m-lab/go/cloud/gcs"
+	"github.com/m-lab/go/logx"
+	"github.com/m-lab/go/rtx"
+
+	"github.com/m-lab/go/cloudtest/gcsfake"
 )
 
 func init() {
@@ -77,23 +80,25 @@ func newCounter(t *testing.T) *counter {
 }
 
 func testClient() stiface.Client {
-	client := cloudtest.GCSClient{}
+	client := gcsfake.GCSClient{}
 	client.AddTestBucket("foobar",
-		cloudtest.BucketHandle{
+		gcsfake.BucketHandle{
 			ObjAttrs: []*storage.ObjectAttrs{
-				&storage.ObjectAttrs{Bucket: "foobar", Name: "ndt/ndt5/2019/01/01/obj1", Updated: time.Now()},
-				&storage.ObjectAttrs{Bucket: "foobar", Name: "ndt/ndt5/2019/01/01/obj2", Updated: time.Now()},
-				&storage.ObjectAttrs{Bucket: "foobar", Name: "ndt/ndt5/2019/01/01/obj3", Updated: time.Date(2000, 01, 01, 02, 03, 04, 0, time.UTC)},
-				&storage.ObjectAttrs{Bucket: "foobar", Name: "ndt/ndt5/2019/01/01/subdir/obj4", Updated: time.Now()},
-				&storage.ObjectAttrs{Bucket: "foobar", Name: "ndt/ndt5/2019/01/01/subdir/obj5", Updated: time.Now()},
-				&storage.ObjectAttrs{Bucket: "foobar", Name: "ndt/tcpinfo/2019/01/01/obj3", Updated: time.Date(2000, 01, 01, 02, 03, 04, 0, time.UTC)},
-				&storage.ObjectAttrs{Bucket: "foobar", Name: "obj6", Updated: time.Now()},
+				{Bucket: "foobar", Name: "ndt/ndt5/2019/01/01/obj1", Updated: time.Now()},
+				{Bucket: "foobar", Name: "ndt/ndt5/2019/01/01/obj2", Updated: time.Now()},
+				{Bucket: "foobar", Name: "ndt/ndt5/2019/01/01/obj3", Updated: time.Date(2000, 01, 01, 02, 03, 04, 0, time.UTC)},
+				{Bucket: "foobar", Name: "ndt/ndt5/2019/01/01/subdir/obj4", Updated: time.Now()},
+				{Bucket: "foobar", Name: "ndt/ndt5/2019/01/01/subdir/obj5", Updated: time.Now()},
+				{Bucket: "foobar", Name: "ndt/tcpinfo/2019/01/01/obj3", Updated: time.Date(2000, 01, 01, 02, 03, 04, 0, time.UTC)},
+				{Bucket: "foobar", Name: "obj6", Updated: time.Now()},
 			}})
 	return client
 }
 
 func standardLister() active.FileLister {
-	return active.FileListerFunc(testClient(), "gs://foobar/ndt/ndt5/2019/01/01/", nil)
+	bh, err := gcs.GetBucket(context.Background(), testClient(), "foobar")
+	rtx.Must(err, "GetBucket failed")
+	return active.FileListerFunc(bh, "ndt/ndt5/2019/01/01/", nil)
 }
 
 func runAll(ctx context.Context, rSrc active.RunnableSource) (*errgroup.Group, error) {

active/poller.go

@@ -20,6 +20,7 @@ import (
 
 	job "github.com/m-lab/etl-gardener/client"
 	"github.com/m-lab/etl-gardener/tracker"
+	"github.com/m-lab/go/cloud/gcs"
 	"github.com/m-lab/go/rtx"
 
 	"github.com/m-lab/etl/metrics"
@@ -128,21 +129,34 @@ func (g *GardenerAPI) RunAll(ctx context.Context, rSrc RunnableSource, job track
 	}
 }
 
+func failMetric(j tracker.Job, label string) {
+	JobFailures.WithLabelValues(
+		j.Experiment+"/"+j.Datatype, j.Date.Format("2006"), label).Inc()
+}
+
 // JobFileSource creates a gcsSource for the job.
 func (g *GardenerAPI) JobFileSource(ctx context.Context, job tracker.Job,
 	toRunnable func(*storage.ObjectAttrs) Runnable) (*GCSSource, error) {
 
 	filter, err := regexp.Compile(job.Filter)
 	if err != nil {
-		JobFailures.WithLabelValues(
-			job.Experiment+"/"+job.Datatype, job.Date.Format("2006"), "filter compile").Inc()
+		failMetric(job, "filter compile")
+		return nil, err
+	}
+	bh, err := gcs.GetBucket(ctx, g.gcs, job.Bucket)
+	if err != nil {
+		failMetric(job, "filesource")
 		return nil, err
 	}
-	lister := FileListerFunc(g.gcs, job.Path(), filter)
+	prefix, err := job.Prefix()
+	if err != nil {
+		failMetric(job, "prefix")
+		return nil, err
+	}
+	lister := FileListerFunc(bh, prefix, filter)
 	gcsSource, err := NewGCSSource(ctx, job.Path(), lister, toRunnable)
 	if err != nil {
-		JobFailures.WithLabelValues(
-			job.Experiment+"/"+job.Datatype, job.Date.Format("2006"), "filesource").Inc()
+		failMetric(job, "GCSSource")
 		return nil, err
 	}
 	return gcsSource, nil
@@ -217,5 +231,4 @@ func (g *GardenerAPI) Poll(ctx context.Context,
 // Status adds a small amount of status info to w.
 func (g *GardenerAPI) Status(w http.ResponseWriter) {
 	fmt.Fprintf(w, "Gardener API: %s\n", g.trackerBase.String())
-
 }