Initial distributor implementation

appengine/cmd/dm/deps/walk_graph.go

 // Copyright 2015 The LUCI Authors. All rights reserved.
 // Use of this source code is governed under the Apache License, Version 2.0
 // that can be found in the LICENSE file.
 
 package deps
 
 import (
         "fmt"
         "time"
 
+        "golang.org/x/net/context"
+        "google.golang.org/grpc/codes"
+
         "github.com/luci/gae/service/datastore"
+        "github.com/luci/luci-go/appengine/cmd/dm/distributor"
         "github.com/luci/luci-go/appengine/cmd/dm/model"
         dm "github.com/luci/luci-go/common/api/dm/service/v1"
         "github.com/luci/luci-go/common/clock"
         "github.com/luci/luci-go/common/grpcutil"
         "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/parallel"
-        "golang.org/x/net/context"
-        "google.golang.org/grpc/codes"
 )
 
 const numWorkers = 16
 
 const maxTimeout = 55 * time.Second // GAE limit is 60s
 
 type node struct {
         aid                 *dm.Attempt_ID
         depth               int64
         canSeeAttemptResult bool
 }
 
 func runSearchQuery(c context.Context, send func(*dm.Attempt_ID) error, s *dm.GraphQuery_Search) func() error {
         return func() error {
                 logging.Errorf(c, "SearchQuery not implemented")
                 return grpcutil.Errf(codes.Unimplemented, "GraphQuery.Search is not implemented")
         }
 }
 
 func isCtxErr(err error) bool {
         return err == context.Canceled || err == context.DeadlineExceeded
 }
 
 func runAttemptListQuery(c context.Context, includeExpired bool, send func(*dm.Attempt_ID) error, al *dm.AttemptList) func() error {
         return func() error {
                 for qst, anum := range al.To {
                         if len(anum.Nums) == 0 {
                                 qry := model.QueryAttemptsForQuest(c, qst)
                                 if !includeExpired {
-                                        qry = qry.Eq("Expired", false)
+                                        qry = qry.Eq("ResultExpired", false)
                                 }
                                 err := datastore.Get(c).Run(qry, func(k *datastore.Key) error {
                                         aid := &dm.Attempt_ID{}
                                         if err := aid.SetDMEncoded(k.StringID()); err != nil {
                                                 logging.WithError(err).Errorf(c, "Attempt_ID.SetDMEncoded returned an error with input: %q", k.StringID())
                                                 panic(fmt.Errorf("in AttemptListQuery: %s", err))
                                         }
                                         return send(aid)
                                 })
                                 if err != nil {
@@ skipping 58 matching lines @@
                 }
                 if doSend {
                         if err := send(aid); err != nil {
                                 return err
                         }
                 }
                 return nil
         })
 }
 
-func loadExecutions(c context.Context, includeID bool, atmpt *model.Attempt, akey *datastore.Key, numEx int64, dst *dm.Attempt) error {
+func maybeLoadInfoURL(c context.Context, reg distributor.Registry, dat *dm.Execution_Data_DistributorInfo) {
+        d, _, err := reg.MakeDistributor(c, dat.ConfigName)
+        if err != nil {
+                logging.Fields{
+                        ek: err, "cfgName": dat.ConfigName,
+                }.Errorf(c, "unable to load distributor")
+        } else {
+                dat.Url = d.InfoURL(distributor.Token(dat.Token))
+        }
+}
+
+func loadExecutions(c context.Context, includeID, includeURL bool, atmpt *model.Attempt, akey *datastore.Key, numEx int64, dst *dm.Attempt) error {
         start := int64(atmpt.CurExecution) - numEx
         if start <= 0 {
                 start = 1
         }
         toLoad := (int64(atmpt.CurExecution) - start) + 1
         if toLoad <= 0 {
                 return nil
         }
+        reg := distributor.GetRegistry(c)
         dst.Executions = make(map[uint32]*dm.Execution, toLoad)
         ds := datastore.Get(c)
         q := datastore.NewQuery("Execution").Ancestor(akey).Gte(
                 "__key__", ds.MakeKey("Attempt", akey.StringID(), "Execution", start))
         return ds.Run(q, func(e *model.Execution) error {
                 if c.Err() != nil {
                         return datastore.Stop
                 }
-                dst.Executions[e.ID] = e.ToProto(includeID)
+                p := e.ToProto(includeID)
+                if includeURL {
+                        maybeLoadInfoURL(c, reg, p.Data.DistributorInfo)
+                }
+                dst.Executions[e.ID] = p
                 return nil
         })
 }
 
-func attemptResultLoader(c context.Context, aid *dm.Attempt_ID, authedForResult bool, rsltSize uint32, lim *sizeLimit, akey *datastore.Key, auth *dm.Execution_Auth, dst *dm.Attempt) func() error {
+func attemptResultLoader(c context.Context, aid *dm.Attempt_ID, authedForResult bool, persistentState []byte, rsltSize uint32, lim *sizeLimit, akey *datastore.Key, auth *dm.Execution_Auth, dst *dm.Attempt) func() error {
         return func() error {
                 ds := datastore.Get(c)
                 if auth != nil && !authedForResult {
                         // we need to prove that the currently authed execution depends on this
                         // attempt.
                         from := auth.Id.AttemptID().DMEncoded()
                         to := aid.DMEncoded()
                         fdepKey := ds.MakeKey("Attempt", from, "FwdDep", to)
                         exist, err := ds.Exists(fdepKey)
                         if err != nil {
                                 logging.Fields{ek: err, "key": fdepKey}.Errorf(c, "failed to determine if FwdDep exists")
                                 dst.Partial.Result = dm.Attempt_Partial_NOT_AUTHORIZED
                                 return err
                         }
                         if !exist.All() {
                                 dst.Partial.Result = dm.Attempt_Partial_NOT_AUTHORIZED
                                 return nil
                         }
                 }
 
-                if !lim.PossiblyOK(rsltSize) {
+                siz := rsltSize + uint32(len(persistentState))
+                if !lim.PossiblyOK(siz) {
                         dst.Partial.Result = dm.Attempt_Partial_DATA_SIZE_LIMIT
                         logging.Infof(c, "skipping load of AttemptResult %s (size limit)", aid)
                         return nil
                 }
                 r := &model.AttemptResult{Attempt: akey}
                 if err := ds.Get(r); err != nil {
                         logging.Fields{ek: err, "aid": aid}.Errorf(c, "failed to load AttemptResult")
                         return err
                 }
-                if lim.Add(r.Size) {
+                if lim.Add(siz) {
                         dst.Data.GetFinished().JsonResult = r.Data
+                        dst.Data.GetFinished().PersistentStateResult = persistentState
                         dst.Partial.Result = dm.Attempt_Partial_LOADED
                 } else {
                         dst.Partial.Result = dm.Attempt_Partial_DATA_SIZE_LIMIT
                         logging.Infof(c, "loaded AttemptResult %s, but hit size limit after", aid)
                 }
                 return nil
         }
 }
 
 func attemptLoader(c context.Context, req *dm.WalkGraphReq, aid *dm.Attempt_ID, authedForResult bool, lim *sizeLimit, dst *dm.Attempt, send func(aid *dm.Attempt_ID, fwd bool) error) func() error {
         return func() error {
                 ds := datastore.Get(c)
 
                 atmpt := &model.Attempt{ID: *aid}
                 akey := ds.KeyForObj(atmpt)
                 if err := ds.Get(atmpt); err != nil {
                         if err == datastore.ErrNoSuchEntity {
                                 dst.DNE = true
                                 dst.Partial = nil
                                 return nil
                         }
                         return err
                 }
-                if !req.Include.ExpiredAttempts && atmpt.Expired {
+                if !req.Include.ExpiredAttempts && atmpt.ResultExpired {
                         return nil
                 }
+
+                persistentState := []byte(nil)
                 if req.Include.AttemptData {
                         dst.Data = atmpt.DataProto()
                         dst.Partial.Data = false
+                        if fin := dst.Data.GetFinished(); fin != nil {
+                                // if we're including data and finished, we only add the persistentState
+                                // if we could see the attempt result. Save it off here, and restore it
+                                // in attemptResultLoader, only if we're able to load the actual result.
+                                //
+                                // This is done because for some jobs the persistentState is
+                                // almost-as-good as the actual result, and we want to prevent
+                                // false/accidental dependencies where a job is able to 'depend' on the
+                                // results without actually emitting a dependency on them.
+                                persistentState = fin.PersistentStateResult
+                                fin.PersistentStateResult = nil
+                        }
                 }
 
                 errChan := parallel.Run(0, func(ch chan<- func() error) {
                         if req.Include.AttemptResult {
                                 if atmpt.State == dm.Attempt_FINISHED {
-                                        ch <- attemptResultLoader(c, aid, authedForResult, atmpt.ResultSize, lim, akey, req.Auth, dst)
+                                        ch <- attemptResultLoader(c, aid, authedForResult, persistentState, atmpt.ResultSize, lim, akey, req.Auth, dst)
                                 } else {
                                         dst.Partial.Result = dm.Attempt_Partial_LOADED
                                 }
                         }
 
                         if req.Include.NumExecutions > 0 {
                                 ch <- func() error {
-                                        err := loadExecutions(c, req.Include.ObjectIds, atmpt, akey, int64(req.Include.NumExecutions), dst)
+                                        err := loadExecutions(c, req.Include.ObjectIds, req.Include.ExecutionInfoUrl, atmpt, akey, int64(req.Include.NumExecutions), dst)
                                         if err != nil {
                                                 logging.Fields{ek: err, "aid": aid}.Errorf(c, "error loading executions")
                                         } else {
                                                 dst.Partial.Executions = false
                                         }
                                         return err
                                 }
                         }
 
                         writeFwd := req.Include.FwdDeps
@@ skipping 89 matching lines @@
 //     a = get_attempt()
 //     if a exists {
 //       FanOutIn(
 //         maybeLoadAttemptResult,
 //         maybeLoadExecutions,
 //         maybeLoadFwdDeps, // sends to nodeChan if walking direction Fwd|Both
 //         maybeLoadBackDeps // sends to nodeChan if walking direction Back|Both
 //       )
 //     }
 func (d *deps) WalkGraph(c context.Context, req *dm.WalkGraphReq) (rsp *dm.GraphData, err error) {
+        if req.Auth != nil {
+                logging.Fields{"execution": req.Auth.Id}.Debugf(c, "on behalf of")
+        } else {
+                logging.Debugf(c, "for user")
+        }
+
         cncl := (func())(nil)
         timeoutProto := req.Limit.MaxTime
         timeout := timeoutProto.Duration() // .Duration on nil is OK
         if timeoutProto == nil || timeout > maxTimeout {
                 timeout = maxTimeout
         }
         c, cncl = clock.WithTimeout(c, timeout)
         defer cncl()
 
         // nodeChan recieves attempt nodes to process. If it recieves the
@@ skipping 66 matching lines @@
                         outstandingJobs--
                         if err == nil {
                                 break
                         }
                         if !isCtxErr(err) {
                                 rsp.HadErrors = true
                         }
                         // assume that contextualized logging already happened
 
                 case n := <-nodeChan:
-                        logging.Fields{"aid": n.aid.DMEncoded(), "depth": n.depth}.Infof(c, "got node")
                         qst, ok := rsp.GetQuest(n.aid.Quest)
                         if !ok {
                                 if !req.Include.ObjectIds {
                                         qst.Id = nil
                                 }
                                 if req.Include.QuestData {
                                         addJob(questDataLoader(c, n.aid.Quest, qst))
                                 }
                         }
                         if _, ok := qst.Attempts[n.aid.Id]; !ok {
@@ skipping 17 matching lines @@
                                                 sendNodeAuthed(n.depth+1)))
                                 }
                         }
                         // otherwise, we've dealt with this attempt before, so ignore it.
                 }
         }
 
         if c.Err() != nil {
                 rsp.HadMore = true
         }
-
         return
 }