Migrate parser to Gardener v2 Jobs API (#1098)

* Migrate to the v2 gardener Jobs API
* Update go mod and build to go1.18

Author: stephen-soltesz

.travis.yml

@@ -8,7 +8,7 @@
 dist: bionic
 language: go
 go:
- - "1.15.8"
+ - 1.18
 
 env:
 - PATH=$PATH:$HOME/gopath/bin
@@ -17,8 +17,8 @@ before_install:
 - echo Branch is ${TRAVIS_BRANCH} and Tag is ${TRAVIS_TAG}
 
 # Install coverage tools.
-- go get github.com/mattn/goveralls
-- go get github.com/wadey/gocovmerge
+- go install github.com/mattn/goveralls@latest
+- go install github.com/wadey/gocovmerge@latest
 
 # Install gcloud, for integration tests.
 - $TRAVIS_BUILD_DIR/travis/install_gcloud.sh

Dockerfile

@@ -1,12 +1,12 @@
 # Dockerfile to contain the generate_schema_docs CLI.
 
-FROM golang:1.16.7-alpine3.14 AS build
+FROM golang:1.18-alpine3.14 AS build
 RUN apk update
 RUN apk add --virtual build-dependencies build-base gcc wget git linux-headers
 # Build the command.
 COPY . /go/src/github.com/m-lab/etl
 WORKDIR /go/src/github.com/m-lab/etl
-RUN go get -v github.com/m-lab/etl/cmd/generate_schema_docs
+RUN go install -v github.com/m-lab/etl/cmd/generate_schema_docs
 
 # Now copy the resulting command into the minimal base image.
 FROM alpine:3.14

Dockerfile.testing

@@ -1,10 +1,9 @@
 FROM gcr.io/google.com/cloudsdktool/cloud-sdk:331.0.0
 
-
 # Fetch recent go version.
-ENV GOLANG_VERSION 1.15.8
+ENV GOLANG_VERSION 1.18.3
 ENV GOLANG_DOWNLOAD_URL https://golang.org/dl/go$GOLANG_VERSION.linux-amd64.tar.gz
-ENV GOLANG_DOWNLOAD_SHA256 d3379c32a90fdf9382166f8f48034c459a8cc433730bc9476d39d9082c94583b
+ENV GOLANG_DOWNLOAD_SHA256 956f8507b302ab0bb747613695cdae10af99bbd39a90cae522b7c0302cc27245
 
 RUN curl -fsSL "$GOLANG_DOWNLOAD_URL" -o golang.tar.gz \
     && echo "$GOLANG_DOWNLOAD_SHA256  golang.tar.gz" | sha256sum -c - \
@@ -19,6 +18,6 @@ WORKDIR $GOPATH
 # Install binaries needed for builds and testing.
 RUN apt-get --allow-releaseinfo-change update
 RUN apt-get install -y jq gcc netcat
-RUN go get -v github.com/m-lab/gcp-config/cmd/cbif
+RUN go install -v github.com/m-lab/gcp-config/cmd/cbif@v1.3.12
 
 ENTRYPOINT ["/go/bin/cbif"]

active/poller.go

@@ -2,9 +2,7 @@ package active
 
 import (
 	"context"
-	"errors"
 	"fmt"
-	"io/ioutil"
 	"log"
 	"net/http"
 	"net/url"
@@ -19,7 +17,7 @@ import (
 	"google.golang.org/api/iterator"
 	"google.golang.org/api/option"
 
-	job "github.com/m-lab/etl-gardener/client"
+	gardener "github.com/m-lab/etl-gardener/client/v2"
 	"github.com/m-lab/etl-gardener/tracker"
 	"github.com/m-lab/go/cloud/gcs"
 	"github.com/m-lab/go/rtx"
@@ -46,11 +44,13 @@ var JobFailures = promauto.NewCounterVec(
 type GardenerAPI struct {
 	trackerBase url.URL
 	gcs         stiface.Client
+	jobs        *gardener.JobClient
 }
 
 // NewGardenerAPI creates a GardenerAPI.
 func NewGardenerAPI(trackerBase url.URL, gcs stiface.Client) *GardenerAPI {
-	return &GardenerAPI{trackerBase: trackerBase, gcs: gcs}
+	c := gardener.NewJobClient(trackerBase)
+	return &GardenerAPI{trackerBase: trackerBase, gcs: gcs, jobs: c}
 }
 
 // MustStorageClient creates a default GCS client.
@@ -60,43 +60,12 @@ func MustStorageClient(ctx context.Context) stiface.Client {
 	return stiface.AdaptClient(c)
 }
 
-// TODO migrate this to m-lab/go
-func post(ctx context.Context, url url.URL) ([]byte, int, error) {
-	ctx, cancel := context.WithTimeout(ctx, time.Minute)
-	defer cancel()
-	req, reqErr := http.NewRequestWithContext(ctx, "POST", url.String(), nil)
-	if reqErr != nil {
-		return nil, 0, reqErr
-	}
-	resp, postErr := http.DefaultClient.Do(req)
-	if postErr != nil {
-		return nil, 0, postErr // Documentation says we can ignore body.
-	}
-
-	// Gauranteed to have a non-nil response and body.
-	defer resp.Body.Close()
-	b, err := ioutil.ReadAll(resp.Body) // Documentation recommends reading body.
-	return b, resp.StatusCode, err
-}
-
-// TODO add retry in case gardener is offline (during redeployment)
-// TODO add metrics to track latency, retries, and errors.
-func postAndIgnoreResponse(ctx context.Context, url url.URL) error {
-	_, status, err := post(ctx, url)
-	if err != nil {
-		return err
-	}
-	if status != http.StatusOK {
-		return errors.New(http.StatusText(status))
-	}
-	return nil
-}
-
 // RunAll will execute functions provided by Next() until there are no more,
 // or the context is canceled.
-func (g *GardenerAPI) RunAll(ctx context.Context, rSrc RunnableSource, job tracker.Job) (*errgroup.Group, error) {
+func (g *GardenerAPI) RunAll(ctx context.Context, rSrc RunnableSource, jt *tracker.JobWithTarget) (*errgroup.Group, error) {
 	eg := &errgroup.Group{}
 	count := 0
+	job := jt.Job
 	for {
 		run, err := rSrc.Next(ctx)
 		if err != nil {
@@ -111,9 +80,8 @@ func (g *GardenerAPI) RunAll(ctx context.Context, rSrc RunnableSource, job track
 			}
 		}
 
-		heartbeat := tracker.HeartbeatURL(g.trackerBase, job)
-		if postErr := postAndIgnoreResponse(ctx, *heartbeat); postErr != nil {
-			log.Println(postErr, "on heartbeat for", job.Path())
+		if err := g.jobs.Heartbeat(ctx, jt.ID); err != nil {
+			log.Println(err, "on heartbeat for", job.Path())
 		}
 
 		debug.Println("Starting func")
@@ -131,9 +99,8 @@ func (g *GardenerAPI) RunAll(ctx context.Context, rSrc RunnableSource, job track
 
 			err = run.Run(ctx)
 			if err == nil {
-				update := tracker.UpdateURL(g.trackerBase, job, tracker.Parsing, run.Info())
-				if postErr := postAndIgnoreResponse(ctx, *update); postErr != nil {
-					log.Println(postErr, "on update for", job.Path())
+				if err := g.jobs.Update(ctx, jt.ID, tracker.Parsing, run.Info()); err != nil {
+					log.Println(err, "on update for", job.Path())
 				}
 			}
 			return
@@ -178,52 +145,48 @@ func (g *GardenerAPI) JobFileSource(ctx context.Context, job tracker.Job,
 }
 
 // NextJob requests a new job from Gardener service.
-func (g *GardenerAPI) NextJob(ctx context.Context) (tracker.JobWithTarget, error) {
-	return job.NextJob(ctx, g.trackerBase)
+func (g *GardenerAPI) NextJob(ctx context.Context) (*tracker.JobWithTarget, error) {
+	return g.jobs.Next(ctx)
 }
 
 func (g *GardenerAPI) pollAndRun(ctx context.Context,
 	toRunnable func(o *storage.ObjectAttrs) Runnable, tokens TokenSource) error {
-	job, err := g.NextJob(ctx)
+	jt, err := g.jobs.Next(ctx)
 	if err != nil {
 		log.Println(err, "on Gardener client.NextJob()")
 		return err
 	}
 
-	log.Println(job, "filter:", job.Filter)
-	gcsSource, err := g.JobFileSource(ctx, job.Job, toRunnable)
+	log.Println(jt, "filter:", jt.Job.Filter)
+	gcsSource, err := g.JobFileSource(ctx, jt.Job, toRunnable)
 	if err != nil {
 		log.Println(err, "on JobFileSource")
 		return err
 	}
 	src := Throttle(gcsSource, tokens)
 
-	log.Println("Running", job.Path())
-
-	update := tracker.UpdateURL(g.trackerBase, job.Job, tracker.Parsing, "starting tasks")
-	if postErr := postAndIgnoreResponse(ctx, *update); postErr != nil {
-		log.Println(postErr)
+	log.Println("Running", jt.Job.Path())
+	if err := g.jobs.Update(ctx, jt.ID, tracker.Parsing, "starting tasks"); err != nil {
+		log.Println(err)
 	}
 
-	eg, err := g.RunAll(ctx, src, job.Job)
+	eg, err := g.RunAll(ctx, src, jt)
 	if err != nil {
 		log.Println(err)
 	}
 
 	// Once all are dispatched, we want to wait until all have completed
 	// before posting the state change.
 	go func() {
-		log.Println("all tasks dispatched for", job.Path())
+		log.Println("all tasks dispatched for", jt.Job.Path())
 		err := eg.Wait()
 		if err != nil {
-			log.Println(err, "on wait for", job.Path())
+			log.Println(err, "on wait for", jt.Job.Path())
 		} else {
-			log.Println("finished", job.Path())
+			log.Println("finished", jt.Job.Path())
 		}
-		update := tracker.UpdateURL(g.trackerBase, job.Job, tracker.ParseComplete, "")
-		// TODO - should this have a retry?
-		if postErr := postAndIgnoreResponse(ctx, *update); postErr != nil {
-			log.Println(postErr)
+		if err := g.jobs.Update(ctx, jt.ID, tracker.ParseComplete, ""); err != nil {
+			log.Println(err)
 		}
 	}()
 

active/poller_test.go

@@ -28,13 +28,17 @@ type fakeGardener struct {
 	t *testing.T // for logging
 
 	lock       sync.Mutex
-	jobs       []tracker.Job
+	jobs       []*tracker.JobWithTarget
 	heartbeats int
 	updates    int
 }
 
 func (g *fakeGardener) AddJob(job tracker.Job) {
-	g.jobs = append(g.jobs, job)
+	jt := &tracker.JobWithTarget{
+		ID:  job.Key(),
+		Job: job,
+	}
+	g.jobs = append(g.jobs, jt)
 }
 
 func (g *fakeGardener) ServeHTTP(w http.ResponseWriter, r *http.Request) {
@@ -43,21 +47,21 @@ func (g *fakeGardener) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		log.Fatal("Should be POST") // Not t.Fatal because this is asynchronous.
 	}
 	g.lock.Lock()
-	g.lock.Unlock()
-	switch r.URL.Path {
-	case "/job":
+	defer g.lock.Unlock()
+	switch {
+	case r.URL.Path == "/v2/job/next":
 		if len(g.jobs) < 1 {
 			w.WriteHeader(http.StatusInternalServerError)
 			return
 		}
 		j := g.jobs[0]
 		g.jobs = g.jobs[1:]
 		w.Write(j.Marshal())
-	case "/heartbeat":
+	case r.URL.Path == "/v2/job/heartbeat":
 		g.t.Log(r.URL.Path, r.URL.Query())
 		g.heartbeats++
 
-	case "/update":
+	case r.URL.Path == "/v2/job/update":
 		g.t.Log(r.URL.Path, r.URL.Query())
 		g.updates++
 
@@ -75,7 +79,7 @@ func TestGardenerAPI_JobFileSource(t *testing.T) {
 	c := testClient()
 
 	// set up a fake gardener service.
-	fg := fakeGardener{t: t, jobs: make([]tracker.Job, 0)}
+	fg := fakeGardener{t: t, jobs: make([]*tracker.JobWithTarget, 0)}
 	fg.AddJob(tracker.NewJob("foobar", "ndt", "ndt5", time.Date(2019, 01, 01, 0, 0, 0, 0, time.UTC)))
 	tracker := httptest.NewServer(&fg)
 	defer tracker.Close()
@@ -85,12 +89,12 @@ func TestGardenerAPI_JobFileSource(t *testing.T) {
 	// Set up GardenerAPI using the fakes.
 	g := active.NewGardenerAPI(*tkURL, c)
 
-	job, err := g.NextJob(ctx)
+	jt, err := g.NextJob(ctx)
 	rtx.Must(err, "next job")
 
 	// The test counter creates runnables for the jobs.
 	p := newCounter(t)
-	src, err := g.JobFileSource(ctx, job.Job, p.toRunnable)
+	src, err := g.JobFileSource(ctx, jt.Job, p.toRunnable)
 	rtx.Must(err, "file source")
 	log.Println(src)
 
@@ -119,7 +123,7 @@ func TestGardenerAPI_RunAll(t *testing.T) {
 	c := testClient()
 
 	// set up a fake gardener service.
-	fg := fakeGardener{t: t, jobs: make([]tracker.Job, 0)}
+	fg := fakeGardener{t: t, jobs: make([]*tracker.JobWithTarget, 0)}
 	fg.AddJob(tracker.NewJob("foobar", "ndt", "ndt5", time.Date(2019, 01, 01, 0, 0, 0, 0, time.UTC)))
 	tracker := httptest.NewServer(&fg)
 	defer tracker.Close()
@@ -129,14 +133,14 @@ func TestGardenerAPI_RunAll(t *testing.T) {
 	// Set up GardenerAPI using the fakes.
 	g := active.NewGardenerAPI(*tkURL, c)
 
-	job, err := g.NextJob(ctx)
+	jt, err := g.NextJob(ctx)
 	rtx.Must(err, "next job")
 
 	// The test counter creates runnables for the jobs.
 	p := newCounter(t)
-	src, err := g.JobFileSource(ctx, job.Job, p.toRunnable)
+	src, err := g.JobFileSource(ctx, jt.Job, p.toRunnable)
 	rtx.Must(err, "file source")
-	eg, err := g.RunAll(ctx, src, job.Job)
+	eg, err := g.RunAll(ctx, src, jt)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -168,7 +172,7 @@ func TestGardenerAPI_Poll(t *testing.T) {
 	c := testClient()
 
 	// set up a fake gardener service.
-	fg := fakeGardener{t: t, jobs: make([]tracker.Job, 0)}
+	fg := fakeGardener{t: t, jobs: make([]*tracker.JobWithTarget, 0)}
 	fg.AddJob(tracker.NewJob("foobar", "ndt", "ndt5", time.Date(2019, 01, 01, 0, 0, 0, 0, time.UTC)))
 	tracker := httptest.NewServer(&fg)
 	defer tracker.Close()

cmd/etl_worker/Dockerfile.k8s

@@ -1,28 +1,21 @@
-FROM golang:1.15.8 as builder
-
+FROM golang:1.18 as builder
 ARG VERSION
-
 WORKDIR /go/src/github.com/m-lab/etl
 COPY . .
-
 RUN go get -v ./...
-
 # This leave some dynamically linked lookups, but seems to work anyway.  But
 # it is unclear whether we might see random segfaults.
-RUN go get \
+RUN go install \
     -tags netgo -a -v \
     -ldflags "-linkmode external -extldflags -static \
       -X github.com/m-lab/go/prometheusx.GitShortCommit=$(git log -1 --format=%h) \
       -X github.com/m-lab/etl/etl.Version=$VERSION \
       -X github.com/m-lab/etl/etl.GitCommit=$(git log -1 --format=%H)" \
     ./cmd/etl_worker
 
-FROM alpine:3.12
+FROM alpine:3.14
 RUN apk add --no-cache ca-certificates
-
 COPY --from=builder /go/bin/etl_worker /bin/etl_worker
-
 EXPOSE 9090 8080
-
 WORKDIR /
 ENTRYPOINT [ "/bin/etl_worker" ]