test-results: package frontend: Add delete keys task queue

go/src/infra/appengine/test-results/frontend/upload.go

 package frontend
 
 import (
         "bytes"
         "encoding/json"
         "errors"
         "io"
         "mime/multipart"
         "net/http"
+        "net/url"
         "strconv"
         "sync"
+        "time"
 
         "golang.org/x/net/context"
 
         "github.com/luci/gae/service/datastore"
+        "github.com/luci/gae/service/taskqueue"
+        "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/server/router"
 
+        "infra/appengine/test-results/builderstate"
         "infra/appengine/test-results/model"
 )
 
 type statusError struct {
         error
         code int
 }
 
 // MarshalJSON marshals status error to JSON.
 func (se *statusError) MarshalJSON() ([]byte, error) {
@@ skipping 111 matching lines @@
         }
         defer file.Close()
 
         switch fh.Filename {
         case "incremental_results.json":
                 var incr model.AggregateResult
                 if err := json.NewDecoder(file).Decode(&incr); err != nil {
                         return statusError{err, http.StatusBadRequest}
                 }
                 return updateIncremental(c, &incr)
+
         case "full_results.json":
-                return updateFullResults(c, file)
+                bn, data, err := extractBuildNumber(file)
+                if err != nil {
+                        if err == ErrInvalidBuildNumber {
+                                return statusError{err, http.StatusBadRequest}
+                        }
+                        return statusError{err, http.StatusInternalServerError}
+                }
+                if err := updateFullResults(c, data); err != nil {
+                        return err
+                }
+
+                p := GetUploadParams(c)
+                wg := sync.WaitGroup{}
+
+                wg.Add(1)
+                go func() {
+                        defer wg.Done()
+                        if err := builderstate.Update(c, p.Master, p.Builder, p.TestType, time.Now().UTC()); err != nil {
+                                logging.WithError(err).Errorf(c, "builderstate update")
+                        }
+                }()
+
+                wg.Add(1)
+                go func() {
+                        defer wg.Done()
+                        if err := taskqueue.Get(c).Add(
+                                taskqueue.NewPOSTTask("/internal/monitoring/upload", url.Values{
+                                        "master":       {p.Master},
+                                        "builder":      {p.Builder},
+                                        "build_number": {strconv.Itoa(bn)},
+                                        "test_type":    {p.TestType},
+                                }),
+                                defaultQueueName,
+                        ); err != nil {
+                                logging.WithError(err).Errorf(c, "taskqueue add: /internal/monitoring/upload")
+                        }
+                }()
+
+                wg.Wait()
+                return nil

[Vadim Sh., 2016/08/16 21:27:12]

no reporting errors back to the user?

[nishanths, 2016/08/16 21:29:54]

These errors are not critical to the POST request.
(The python implementation also does not report these errors, but let me know if
we should be.)

[Vadim Sh., 2016/08/16 21:34:46]

I have no idea how it all works on high level, I just review local details :)

Beware that in Python stuff (datastore requests) can throw DeadlineExceeded and
other exceptions at any time, that (unless caught) result in HTTP 500, that is
usually retried by the client. In Go these exceptions manifest as errors from
datastore.Put etc, so ignoring them is not the same as not explicitly checking
exceptions in Python.

+
         default:
                 return uploadTestFile(c, file, fh.Filename)
         }
 }
 
-// uploadTestFile creates a new TestFile from the values in context
-// and supplied data, and puts it to the datastore.
-func uploadTestFile(c context.Context, data io.Reader, filename string) error {
+// ErrInvalidBuildNumber is returned when the extractBuildNumber fails
+// to convert the build number value to an int.
+var ErrInvalidBuildNumber = errors.New("invalid build_number: cannot convert to int")
+
+// extractBuildNumber extracts the value of "build_number" key from
+// the supplied JSON encoded data. The returned io.Reader will have
+// the same contents as the supplied io.Reader.
+//
+// The error is ErrInvalidBuildNumber if the build number value
+// could not be converted to an int.
+func extractBuildNumber(data io.Reader) (int, io.Reader, error) {
         var buf bytes.Buffer
         tee := io.TeeReader(data, &buf)
 
         aux := struct {
                 N string `json:"build_number,omitempty"`
         }{}
-
         dec := json.NewDecoder(tee)
         if err := dec.Decode(&aux); err != nil {
-                return statusError{err, http.StatusInternalServerError}
+                return 0, io.MultiReader(&buf, dec.Buffered()), err
         }
 
-        bn := 0
-
+        var bn int
         if aux.N != "" {
                 n, err := strconv.Atoi(aux.N)
                 if err != nil {
-                        return statusError{errors.New("invalid build_number"), http.StatusBadRequest}
+                        return 0, io.MultiReader(&buf, dec.Buffered()), ErrInvalidBuildNumber
                 }
                 bn = n
         }
 
+        return bn, io.MultiReader(&buf, dec.Buffered()), nil
+}
+
+// uploadTestFile creates a new TestFile from the values in context
+// and supplied data, and puts it to the datastore.
+func uploadTestFile(c context.Context, data io.Reader, filename string) error {
+        bn, data, err := extractBuildNumber(data)
+        if err != nil {
+                if err == ErrInvalidBuildNumber {
+                        return statusError{err, http.StatusBadRequest}
+                }
+                return statusError{err, http.StatusInternalServerError}
+        }
+
         p := GetUploadParams(c)
         tf := model.TestFile{
                 Master:      p.Master,
                 Builder:     p.Builder,
                 TestType:    p.TestType,
                 BuildNumber: model.BuildNum(bn),
                 Name:        filename,
-                Data:        io.MultiReader(&buf, dec.Buffered()),
+                Data:        data,
         }
         if err := tf.PutData(c); err != nil {
                 return statusError{err, http.StatusInternalServerError}
         }
-        return nil
+
+        return datastore.Get(c).Put(&tf)
 }
 
 // updateFullResults puts the supplied data as "full_results.json"
 // to the datastore, and updates corresponding "results.json" and
 // "results-small.json" files in the datastore.
 //
 // The supplied data should unmarshal into model.FullResults.
 // Otherwise, an error is returned.
 func updateFullResults(c context.Context, data io.Reader) error {
         buf := &bytes.Buffer{}
         tee := io.TeeReader(data, buf)
         dec := json.NewDecoder(tee)
 
         var f model.FullResult
         if err := dec.Decode(&f); err != nil {
                 return statusError{err, http.StatusBadRequest}
         }
 
-        wg := sync.WaitGroup{}
-        errCh := make(chan error, 2)
+        if err := uploadTestFile(c, io.MultiReader(buf, dec.Buffered()), "full_results.json"); err != nil {
+                return statusError{err, http.StatusInternalServerError}
+        }
 
-        wg.Add(1)
-        go func() {
-                defer wg.Done()
-                errCh <- uploadTestFile(
-                        c, io.MultiReader(buf, dec.Buffered()), "full_results.json",
-                )
-        }()
+        incr, err := f.AggregateResult()
+        if err != nil {
+                return statusError{err, http.StatusBadRequest}
+        }
+        if err := updateIncremental(c, &incr); err != nil {
+                return statusError{err, http.StatusInternalServerError}
+        }
 
-        wg.Add(1)
-        go func() {
-                defer wg.Done()
-                incr, err := f.AggregateResult()
-                if err != nil {
-                        errCh <- statusError{err, http.StatusBadRequest}
-                        return
-                }
-                errCh <- updateIncremental(c, &incr)
-        }()
-
-        wg.Wait()
-        close(errCh)
-        for err := range errCh {
-                if err != nil {
-                        return err
-                }
-        }
         return nil
 }
 
 // updateIncremental gets "results.json" and "results-small.json"
 // for values in context, merges incr into them, and puts the updated
 // files to the datastore.
 func updateIncremental(c context.Context, incr *model.AggregateResult) error {
         u := GetUploadParams(c)
         p := model.TestFileParams{
                 Master:   u.Master,
                 Builder:  u.Builder,
                 TestType: u.TestType,
         }
 
         names := []string{"results.json", "results-small.json"}
         files := make([]struct {
                 tf   *model.TestFile
                 aggr *model.AggregateResult
                 err  error
         }, len(names))
 
         wg := sync.WaitGroup{}
 
         for i, name := range names {
-                i, name := i, name
+                i, name, p := i, name, p
                 wg.Add(1)
+
                 go func() {
                         defer wg.Done()
+                        p.Name = name
                         tf, err := getTestFileAlt(c, p, u.DeprecatedMaster)
                         if err != nil {
                                 if _, ok := err.(ErrNoMatches); ok {
                                         files[i].tf = &model.TestFile{
                                                 Master:      p.Master,
                                                 Builder:     p.Builder,
                                                 TestType:    p.TestType,
                                                 BuildNumber: -1,
                                                 Name:        name,
                                         }
-                                } else {
-                                        files[i].err = err
+                                        return
                                 }
+                                files[i].err = err
                                 return
                         }
-                        files[i].tf = tf
                         if err := tf.GetData(c); err != nil {
                                 files[i].err = err
                                 return
                         }
-                        if err := json.NewDecoder(tf.Data).Decode(files[i].aggr); err != nil {
+                        var a model.AggregateResult
+                        if err := json.NewDecoder(tf.Data).Decode(&a); err != nil {
                                 files[i].err = err
                                 return
                         }
+                        files[i].tf = tf
+                        files[i].aggr = &a
                 }()
         }
 
         wg.Wait()
         for _, file := range files {
                 if file.err != nil {
                         return file.err
                 }
         }
 
-        wg = sync.WaitGroup{}
-        errs := make([]error, len(files))
+        return datastore.Get(c).RunInTransaction(func(c context.Context) error {
+                wg = sync.WaitGroup{}
+                errs := make([]error, len(files))
 
-        return datastore.Get(c).RunInTransaction(func(c context.Context) error {
                 for i, file := range files {
                         i, file := i, file
                         wg.Add(1)
                         go func() {
                                 defer wg.Done()
                                 errs[i] = updateAggregate(c, file.tf, file.aggr, incr)
                         }()
                 }
 
                 wg.Wait()
@@ skipping 32 matching lines @@
         a, err = getFirstTestFile(c, p.Query())
         if err == nil {
                 a.Master = origMaster
                 return a, nil
         }
 
         return nil, err
 }
 
 // updateAggregate updates tf with the result of merging incr into
-// aggr.
+// aggr, and updates tf in datastore.
 func updateAggregate(c context.Context, tf *model.TestFile, aggr, incr *model.AggregateResult) error {
         if !model.IsAggregateTestFile(tf.Name) {
                 return errors.New("frontend: tf should be an aggregate test file")
         }
 
         size := model.ResultsSize
         if tf.Name == "results-small.json" {
                 size = model.ResultsSmallSize
         }
 
@@ skipping 19 matching lines @@
         b := &bytes.Buffer{}
         if err := json.NewEncoder(b).Encode(&aggr); err != nil {
                 return statusError{err, http.StatusInternalServerError}
         }
 
         tf.Data = b
         if err := tf.PutData(c); err != nil {
                 return statusError{err, http.StatusInternalServerError}
         }
 
+        if err := datastore.Get(c).Put(tf); err != nil {
+                return statusError{err, http.StatusInternalServerError}
+        }
+        if err := deleteKeys(c, tf.OldDataKeys); err != nil {
+                logging.Fields{
+                        logging.ErrorKey: err,
+                        "keys":           tf.OldDataKeys,
+                }.Errorf(c, "upload: failed to delete keys")
+        }
         return nil
 }
+
+func deleteKeys(c context.Context, k []*datastore.Key) error {
+        if len(k) == 0 {
+                return nil
+        }
+
+        keys := make([]string, 0, len(k))
+        for _, key := range k {
+                keys = append(keys, key.Encode())
+        }
+
+        payload, err := json.Marshal(struct {
+                Keys []string `json:"keys"`
+        }{
+                keys,
+        })
+        if err != nil {
+                return err
+        }
+
+        h := make(http.Header)
+        h.Set("Content-Type", "application/json")
+
+        logging.Fields{
+                "keys": keys,
+        }.Infof(c, "deleteKeys: enqueing")
+
+        return taskqueue.Get(c).Add(&taskqueue.Task{
+                Path:    "/internal/delete",
+                Payload: payload,
+                Header:  h,
+                Method:  "POST",
+                Delay:   time.Duration(30) * time.Minute,
+        }, deleteKeysQueueName)
+}