inject task.Factory into ProcessGKETask (#870)

* inject task.Factory into ProcessGKETask

* PR feedback tweaks

* unsaved files

Author: gfr10598

cmd/etl_worker/etl_worker.go

@@ -14,17 +14,19 @@ import (
 	"sync/atomic"
 	"time"
 
-	"cloud.google.com/go/storage"
+	gcs "cloud.google.com/go/storage"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 
-	"github.com/m-lab/go/bqx"
 	"github.com/m-lab/go/prometheusx"
 	"github.com/m-lab/go/rtx"
 
 	"github.com/m-lab/etl/active"
 	"github.com/m-lab/etl/bq"
 	"github.com/m-lab/etl/etl"
+	"github.com/m-lab/etl/factory"
 	"github.com/m-lab/etl/metrics"
+	"github.com/m-lab/etl/storage"
+	"github.com/m-lab/etl/task"
 	"github.com/m-lab/etl/worker"
 
 	// Enable profiling. For more background and usage information, see:
@@ -121,7 +123,7 @@ func handleRequest(rwr http.ResponseWriter, rq *http.Request) {
 
 	// Throttle by grabbing a semaphore from channel.
 	if shouldThrottle() {
-		metrics.TaskCount.WithLabelValues("unknown", "worker", "TooManyRequests").Inc()
+		metrics.TaskCount.WithLabelValues("unknown", "TooManyRequests").Inc()
 		rwr.WriteHeader(http.StatusTooManyRequests)
 		fmt.Fprintf(rwr, `{"message": "Too many tasks."}`)
 		return
@@ -177,7 +179,7 @@ func subworker(rawFileName string, executionCount, retryCount int, age time.Dura
 	// This handles base64 encoding, and requires a gs:// prefix.
 	fn, err := etl.GetFilename(rawFileName)
 	if err != nil {
-		metrics.TaskCount.WithLabelValues("unknown", "worker", "BadRequest").Inc()
+		metrics.TaskCount.WithLabelValues("unknown", "BadRequest").Inc()
 		log.Printf("Invalid filename: %s\n", fn)
 		return http.StatusBadRequest, `{"message": "Invalid filename."}`
 	}
@@ -217,36 +219,28 @@ func setMaxInFlight() {
 }
 
 type runnable struct {
-	storage.ObjectAttrs
+	tf task.Factory
+	gcs.ObjectAttrs
 }
 
 func (r *runnable) Run() error {
 	path := fmt.Sprintf("gs://%s/%s", r.Bucket, r.Name)
-	data, err := etl.ValidateTestPath(path)
+	dp, err := etl.ValidateTestPath(path)
 	if err != nil {
 		log.Printf("Invalid filename: %v\n", err)
 		return err
 	}
 
-	// TODO This is short term hack to fix the injection bug.
-	// It will be removed in the third PR that introduces Factories
-	dataType := data.GetDataType()
-	pdt := bqx.PDT{Project: dataType.BigqueryProject(), Dataset: dataType.Dataset(), Table: dataType.Table()}
-	client, err := bq.GetClient(pdt.Project)
-	if err != nil {
-		return err
-	}
-	up := client.Dataset(pdt.Dataset).Table(pdt.Table).Uploader()
-	// This avoids problems when a single row of the insert has invalid
-	// data.  We then have to carefully parse the returned error object.
-	up.SkipInvalidRows = true
-
 	start := time.Now()
 	log.Println("Processing", path)
-	// TODO pass in storage client, or pass in TestSource.
-	statusCode, err := worker.ProcessGKETask(path, up, nil) // Use default uploader and annotator
+
+	statusCode := http.StatusOK
+	pErr := worker.ProcessGKETask(dp, r.tf)
+	if pErr != nil {
+		statusCode = pErr.Code()
+	}
 	metrics.DurationHistogram.WithLabelValues(
-		data.DataType, http.StatusText(statusCode)).Observe(
+		dp.DataType, http.StatusText(statusCode)).Observe(
 		time.Since(start).Seconds())
 	return err
 }
@@ -256,8 +250,17 @@ func (r *runnable) Info() string {
 	return r.Name
 }
 
-func toRunnable(obj *storage.ObjectAttrs) active.Runnable {
-	return &runnable{*obj}
+func toRunnable(obj *gcs.ObjectAttrs) active.Runnable {
+	c, err := storage.GetStorageClient(false)
+	if err != nil {
+		return nil // TODO add an error?
+	}
+	taskFactory := worker.StandardTaskFactory{
+		Annotator: factory.DefaultAnnotatorFactory(),
+		Sink:      bq.NewSinkFactory(),
+		Source:    storage.GCSSourceFactory(c),
+	}
+	return &runnable{&taskFactory, *obj}
 }
 
 func mustGardenerAPI(ctx context.Context, jobServer string) *active.GardenerAPI {

etl/globals.go

@@ -129,7 +129,7 @@ func ValidateTestPath(path string) (DataPath, error) {
 
 	dataType := dp.GetDataType()
 	if dataType == INVALID {
-		metrics.TaskCount.WithLabelValues(dp.TableBase(), "worker", "BadRequest").Inc()
+		metrics.TaskCount.WithLabelValues(string(dataType), "BadRequest").Inc()
 		log.Printf("Invalid filename: %s\n", path)
 		return DataPath{}, ErrBadDataType
 	}

metrics/metrics.go

@@ -165,6 +165,7 @@ var (
 			Name: "etl_files_processed",
 			Help: "Number of files processed.",
 		},
+		// TODO maybe change to host and exp/type?  Maybe drop day_of_week?
 		[]string{"rsync_host_module", "day_of_week"},
 	)
 
@@ -179,8 +180,8 @@ var (
 			Name: "etl_task_count",
 			Help: "Number of tasks/archive files processed.",
 		},
-		// Go package or filename, and Status
-		[]string{"table", "package", "status"},
+		// table/datatype, and Status
+		[]string{"table", "status"},
 	)
 
 	// TestCount counts the number of tests successfully processed by the parsers.

metrics/metrics_test.go

@@ -99,7 +99,7 @@ func TestMetrics(t *testing.T) {
 	metrics.PTPollutedCount.WithLabelValues("x")
 	metrics.PTTestCount.WithLabelValues("x")
 	metrics.RowSizeHistogram.WithLabelValues("x")
-	metrics.TaskCount.WithLabelValues("x", "x", "x")
+	metrics.TaskCount.WithLabelValues("x", "x")
 	metrics.TestCount.WithLabelValues("x", "x", "x")
 	metrics.WarningCount.WithLabelValues("x", "x", "x")
 	metrics.WorkerCount.WithLabelValues("x")

task/task.go

@@ -61,12 +61,13 @@ func (tt *Task) SetMaxFileSize(max int64) {
 // ProcessAllTests loops through all the tests in a tar file, calls the
 // injected parser to parse them, and inserts them into bigquery. Returns the
 // number of files processed.
+// TODO pass in the datatype label.
 func (tt *Task) ProcessAllTests() (int, error) {
 	if tt.Parser == nil {
 		panic("Parser is nil")
 	}
-	metrics.WorkerState.WithLabelValues(tt.TableName(), "task").Inc()
-	defer metrics.WorkerState.WithLabelValues(tt.TableName(), "task").Dec()
+	metrics.WorkerState.WithLabelValues(tt.Type(), "task").Inc()
+	defer metrics.WorkerState.WithLabelValues(tt.Type(), "task").Dec()
 	files := 0
 	nilData := 0
 	var testname string
@@ -86,7 +87,7 @@ OUTER:
 					tt.meta["filename"], testname, files,
 					time.Since(tt.meta["parse_time"].(time.Time)), err)
 				metrics.TestCount.WithLabelValues(
-					tt.Parser.TableName(), "unknown", "oversize file").Inc()
+					tt.Type(), "unknown", "oversize file").Inc()
 				continue OUTER
 			default:
 				// We are seeing several of these per hour, a little more than
@@ -104,7 +105,7 @@ OUTER:
 					time.Since(tt.meta["parse_time"].(time.Time)), err)
 
 				metrics.TestCount.WithLabelValues(
-					tt.Parser.TableName(), "unknown", "unrecovered").Inc()
+					tt.Type(), "unknown", "unrecovered").Inc()
 				// Since we don't understand these errors, safest thing to do is
 				// stop processing the tar file (and task).
 				break OUTER
@@ -120,18 +121,18 @@ OUTER:
 		kind, parsable := tt.Parser.IsParsable(testname, data)
 		if !parsable {
 			metrics.FileSizeHistogram.WithLabelValues(
-				tt.Parser.TableName(), kind, "ignored").Observe(float64(len(data)))
+				tt.Type(), kind, "ignored").Observe(float64(len(data)))
 			// Don't bother calling ParseAndInsert since this is unparsable.
 			continue
 		} else {
 			metrics.FileSizeHistogram.WithLabelValues(
-				tt.Parser.TableName(), kind, "parsed").Observe(float64(len(data)))
+				tt.Type(), kind, "parsed").Observe(float64(len(data)))
 		}
 		err = tt.Parser.ParseAndInsert(tt.meta, testname, data)
 		// Shouldn't have any of these, as they should be handled in ParseAndInsert.
 		if err != nil {
 			metrics.TaskCount.WithLabelValues(
-				tt.TableName(), "Task", "ParseAndInsertError").Inc()
+				tt.Type(), "ParseAndInsertError").Inc()
 			log.Printf("%v", err)
 			// TODO(dev) Handle this error properly!
 			continue