func DoGKETask(tsk *task.Task, path etl.DataPath) etl.ProcessingError {
	files, err := tsk.ProcessAllTests(true)  
	dateFormat := "20060102"
	date, dateErr := time.Parse(dateFormat, path.PackedDate)
	if dateErr != nil {
		metrics.TaskCount.WithLabelValues(path.DataType, "Bad Date").Inc()
		log.Printf("Error parsing path.PackedDate: %v", err)
		return factory.NewError(
			path.DataType, "PackedDate", http.StatusBadRequest, dateErr)
	}
	metrics.FileCount.WithLabelValues(
		path.Host+"-"+path.Site+"-"+path.Experiment,
		date.Weekday().String()).Add(float64(files))
	if err != nil {
		metrics.TaskCount.WithLabelValues(path.DataType, "TaskError").Inc()
		log.Printf("Error Processing Tests:  %v", err)
		return factory.NewError(
			path.DataType, "TaskError", http.StatusInternalServerError, err)
	}
	metrics.TaskCount.WithLabelValues(path.DataType, "OK").Inc()
	return nil
}
DoGKETask creates task, processes all tests and handle metrics
package worker
import (
	"context"
	"fmt"
	"log"
	"net/http"
	"time"
	"github.com/googleapis/google-cloud-go-testing/storage/stiface"
	"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/parser"
	"github.com/m-lab/etl/storage"
	"github.com/m-lab/etl/task"
)
type nullCloser struct{}
func (nc nullCloser) Close() error { return nil }
func GetSource(client stiface.Client, uri string) (etl.TestSource, etl.DataPath, int, error) {
	path, err := etl.ValidateTestPath(uri)
	label := path.TableBase()  
	if err != nil {
		metrics.TaskCount.WithLabelValues(label, "InvalidFilename").Inc()
		log.Printf("Invalid filename: %v\n", err)
		return nil, etl.DataPath{}, http.StatusBadRequest, err
	}
	dataType := path.GetDataType()
	if dataType == etl.INVALID {
		metrics.TaskCount.WithLabelValues(string(dataType), "SourcePathError").Inc()
		log.Printf("Invalid datatype: %s", path)
		return nil, etl.DataPath{}, http.StatusInternalServerError, err
	}
	tr, err := storage.NewTestSource(client, path, label)
	if err != nil {
		metrics.TaskCount.WithLabelValues(string(dataType), "ETLSourceError").Inc()
		log.Printf("Error opening gcs file: %v", err)
		return nil, etl.DataPath{}, http.StatusInternalServerError, err
	}
	return tr, path, http.StatusOK, nil
}
func ProcessTask(fn string) (int, error) {
	client, err := storage.GetStorageClient(false)
	if err != nil {
		path, _ := etl.ValidateTestPath(fn)
		metrics.TaskCount.WithLabelValues(path.DataType, "ServiceUnavailable").Inc()
		log.Printf("Error getting storage client: %v\n", err)
		return http.StatusServiceUnavailable, err
	}
	return ProcessTaskWithClient(client, fn)
}
func ProcessTaskWithClient(client stiface.Client, fn string) (int, error) {
	tr, path, status, err := GetSource(client, fn)
	if err != nil {
		return status, err
	}
	defer tr.Close()
	return ProcessTestSource(tr, path)
}
func ProcessTestSource(src etl.TestSource, path etl.DataPath) (int, error) {
	label := path.TableBase()  
	metrics.WorkerCount.WithLabelValues(label).Inc()
	defer metrics.WorkerCount.WithLabelValues(label).Dec()
	metrics.WorkerState.WithLabelValues(label, "worker").Inc()
	defer metrics.WorkerState.WithLabelValues(label, "worker").Dec()
	dataType := path.GetDataType()
	dateFormat := "20060102"
	date, err := time.Parse(dateFormat, path.PackedDate)
	ins, err := bq.NewInserter(dataType, date)
	if err != nil {
		metrics.TaskCount.WithLabelValues(label, string(dataType), "NewInserterError").Inc()
		log.Printf("Error creating BQ Inserter:  %v", err)
		return http.StatusInternalServerError, err
	}
	p := parser.NewParser(dataType, ins)
	if p == nil {
		metrics.TaskCount.WithLabelValues(string(dataType), "NewInserterError").Inc()
		log.Printf("Error creating parser for %s", dataType)
		return http.StatusInternalServerError, fmt.Errorf("problem creating parser for %s", dataType)
	}
	tsk := task.NewTask(src.Detail(), src, p, &nullCloser{})
	files, err := tsk.ProcessAllTests(false)  
	metrics.FileCount.WithLabelValues(
		path.Host+"-"+path.Site+"-"+path.Experiment,
		date.Weekday().String()).Add(float64(files))
	metrics.WorkerState.WithLabelValues(label, "finish").Inc()
	defer metrics.WorkerState.WithLabelValues(label, "finish").Dec()
	if err != nil {
		metrics.TaskCount.WithLabelValues(string(dataType), "TaskError").Inc()
		log.Printf("Error Processing Tests:  %v", err)
		return http.StatusInternalServerError, err
	}
	metrics.TaskCount.WithLabelValues(string(dataType), "OK").Inc()
	return http.StatusOK, nil
}
type StandardTaskFactory struct {
	Sink      factory.SinkFactory
	Source    factory.SourceFactory
	Annotator factory.AnnotatorFactory
}
func (tf *StandardTaskFactory) Get(ctx context.Context, dp etl.DataPath) (*task.Task, etl.ProcessingError) {
	sink, err := tf.Sink.Get(ctx, dp)
	if err != nil {
		e := fmt.Errorf("%v creating sink for %s", err, dp.GetDataType())
		log.Println(e, dp.URI)
		return nil, err
	}
	ann, err := tf.Annotator.Get(ctx, dp)
	if err != nil {
		e := fmt.Errorf("%v creating annotator for %s", err, dp.GetDataType())
		log.Println(e, dp.URI)
		return nil, err
	}
	src, err := tf.Source.Get(ctx, dp)
	if err != nil {
		e := fmt.Errorf("%v creating source for %s", err, dp.GetDataType())
		log.Println(e, dp.URI)
		return nil, err
	}
	p := parser.NewSinkParser(dp.GetDataType(), sink, src.Type(), ann)
	if p == nil {
		e := fmt.Errorf("%v creating parser for %s", err, dp.GetDataType())
		log.Println(e, dp.URI)
		return nil, err
	}
	tsk := task.NewTask(dp.URI, src, p, sink)
	return tsk, nil
}
func ProcessGKETask(ctx context.Context, path etl.DataPath, tf task.Factory) etl.ProcessingError {
	metrics.WorkerCount.WithLabelValues(path.DataType).Inc()
	defer metrics.WorkerCount.WithLabelValues(path.DataType).Dec()
	metrics.WorkerState.WithLabelValues(path.DataType, "worker").Inc()
	defer metrics.WorkerState.WithLabelValues(path.DataType, "worker").Dec()
	tsk, err := tf.Get(ctx, path)
	if err != nil {
		metrics.TaskCount.WithLabelValues(err.DataType(), err.Detail()).Inc()
		log.Printf("TaskFactory error: %v", err)
		return err
	}
	defer tsk.Close()
	return DoGKETask(tsk, path)
}