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)
+                go builderstate.Update(c, p.Master, p.Builder, p.TestType, time.Now().UTC())

[Vadim Sh., 2016/08/16 18:56:13]

you need to wait for these goroutines to finish somewhere, since GAE doesn't
guarantee that it won't kill the process right after handling the request.

Also, it's good to check errors.

[nishanths, 2016/08/16 20:35:13]

Done.

+                go taskqueue.Get(c).Add(
+                        taskqueue.NewPOSTTask("/internal/monitoring/upload", url.Values{

[Vadim Sh., 2016/08/16 18:56:13]

does this handler exist?

[nishanths, 2016/08/16 20:35:13]

It exists in the python implementation.

+                                "master":       {p.Master},
+                                "builder":      {p.Builder},
+                                "build_number": {strconv.Itoa(bn)},
+                                "test_type":    {p.TestType},
+                        }),
+                        defaultQueueName,
+                )
+                return nil
+
         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)
 
         wg.Add(1)
         go func() {
                 defer wg.Done()
-                errCh <- uploadTestFile(
-                        c, io.MultiReader(buf, dec.Buffered()), "full_results.json",
-                )
+                errCh <- uploadTestFile(c, io.MultiReader(buf, dec.Buffered()), "full_results.json")
         }()
 
         wg.Add(1)
         go func() {
                 defer wg.Done()
                 incr, err := f.AggregateResult()
                 if err != nil {
                         errCh <- statusError{err, http.StatusBadRequest}
                         return
                 }
@@ skipping 24 matching lines @@
         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()
                 // Prioritize returning http.StatusInternalServerError status
                 // code errors over other errors.
                 var e error
                 for _, err := range errs {
                         se, ok := err.(statusError)
                         if ok && se.code == http.StatusInternalServerError {
                                 return se

[Vadim Sh., 2016/08/16 18:56:13]

beware that RunInTransaction can return an error on its own, even if the
callback is successful (if transaction can't be committed).

[nishanths, 2016/08/16 20:35:13]

in that case, the error will be propagated at line 163 and handled at line 127
(let me know if i'm missing something.)

                         }
                         e = err
                 }
                 return e
         }, &datastore.TransactionOptions{XG: true})

[Vadim Sh., 2016/08/16 18:56:13]

how many entity groups does this transaction modify?

[nishanths, 2016/08/16 20:35:13]

Two entity groups. (1) TestFile (2) DataEntry

 }
 
 // getTestFileAlt returns the the first TestFile in the datastore for
 // the query formed by calling p.Query().
 //
 // The function tries to find the first TestFile using p. If no such TestFile
 // exists the function sets p.Master to altMaster and tries again.
 // If altMaster is empty, the function does not perform the additional try.
 func getTestFileAlt(c context.Context, p model.TestFileParams, altMaster string) (ret *model.TestFile, err error) {
         a, err := getFirstTestFile(c, p.Query())
@@ skipping 10 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 err

[Vadim Sh., 2016/08/16 18:56:13]

statusError{err, http.StatusInternalServerError} ?

[nishanths, 2016/08/16 20:35:13]

Done.

+        }
+        if err := deleteKeys(c, tf.OldDataKeys); err != nil {

[Vadim Sh., 2016/08/16 18:56:13]

ignoring this error?

[nishanths, 2016/08/16 20:35:13]

yes, only logging the error. 

Returning an HTTP error 500 to the client may result in the client retrying the
POST request, which would not be idempotent.

+                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)
+}