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