Initial distributor implementation

appengine/cmd/dm/deps/ensure_graph_data.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 (
-        "errors"
         "fmt"
 
         "golang.org/x/net/context"
         "google.golang.org/grpc/codes"
 
         "github.com/luci/gae/service/datastore"
 
         "github.com/luci/luci-go/common/grpcutil"
         "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/parallel"
         "github.com/luci/luci-go/common/stringset"
 
-        "github.com/luci/luci-go/common/api/dm/service/v1"
+        dm "github.com/luci/luci-go/common/api/dm/service/v1"
         "github.com/luci/luci-go/common/api/dm/template"
 
         "github.com/luci/luci-go/appengine/tumble"
 
+        "github.com/luci/luci-go/appengine/cmd/dm/distributor"
         "github.com/luci/luci-go/appengine/cmd/dm/model"
         "github.com/luci/luci-go/appengine/cmd/dm/mutate"
 )
 
 func (d *deps) runEnsureGraphDepsWalk(c context.Context, req *dm.EnsureGraphDataReq, newAttempts *dm.AttemptList) (*dm.GraphData, error) {
         // first lets run a query to load all of the proposed attempts.
         wgreq := &dm.WalkGraphReq{
                 Query: dm.AttemptListQuery(newAttempts),
                 Limit: &dm.WalkGraphReq_Limit{
                         MaxDepth:    1,
@@ skipping 180 matching lines @@
         }
 
         // not all of the attemps exist/are finished, we have to block.
         rsp.Result = nil
         rsp.ShouldHalt = true
 
         return tumbleNow(c, &mutate.AddDeps{
                 Auth:   req.ForExecution,
                 Quests: newQuests,
                 // Attempts we think are missing
-                Atmpts: newAttempts,
+                Attempts: newAttempts,
                 // Deps we think are missing (>= newAttempts)
                 Deps: missingDeps,
         })
 }
 
 type templateFileKey struct {
         project, ref string
 }
 
 type templateFile struct {
@@ skipping 26 matching lines @@
                 if err == nil {
                         return false
                 }
                 if rsp.TemplateError == nil {
                         rsp.TemplateError = make([]string, len(req.TemplateQuest))
                 }
                 rsp.TemplateError[i] = err.Error()
                 return true
         }
 
+        dists := map[string]distributor.D{}
+
         newQuests = make(map[string]*model.Quest, len(req.Quest)+len(req.TemplateQuest))
         newAttempts = dm.NewAttemptList(nil)
 
+        reg := distributor.GetRegistry(c)
+
         // render all quest descriptions
         for i, qDesc := range req.Quest {
-                var q *model.Quest
-                if q, err = model.NewQuest(c, qDesc); err != nil {
-                        err = grpcutil.MaybeLogErr(c, err, codes.InvalidArgument, "bad quest description")
+                q := model.NewQuest(c, qDesc)
+
+                d, ok := dists[qDesc.DistributorConfigName]
+                if !ok {
+                        if d, _, err = reg.MakeDistributor(c, qDesc.DistributorConfigName); err != nil {
+                                return
+                        }
+                        dists[qDesc.DistributorConfigName] = d
+                }
+
+                if err = d.Validate(qDesc.JsonPayload); err != nil {
+                        err = grpcutil.MaybeLogErr(c, err, codes.InvalidArgument,
+                                "JSON payload is invalid for this distributor configuration.")
                         return
                 }
 
                 // all provided quest descriptions MUST include at least one attempt
                 if _, ok := req.Attempts.To[q.ID]; !ok {
                         c = logging.SetFields(c, logging.Fields{"id": q.ID, "idx": i})
                         err = grpcutil.MaybeLogErr(c,
-                                errors.New("Quest entries must have a matching Attempts entry"),
+                                fmt.Errorf("Quest %d:%q must have a matching Attempts entry", i, q.ID),
                                 codes.InvalidArgument, "no matches")
                         return
                 }
 
                 if _, ok := newQuests[q.ID]; !ok {
                         newQuests[q.ID] = q
                 }
         }
 
         // copy all normal attempt descriptions
         for qid, nums := range req.Attempts.To {
                 newNums := &dm.AttemptList_Nums{Nums: make([]uint32, len(nums.Nums))}
                 copy(newNums.Nums, nums.Nums)
                 newAttempts.To[qid] = newNums
         }
 
         // render all templates and template attempts into newQuests
         templateFiles := templateFileCache{}
         for i := 0; i < len(req.TemplateQuest); i++ {
                 inst := req.TemplateQuest[i]
 
                 var vers string
                 var desc *dm.Quest_Desc
                 if desc, vers, err = templateFiles.render(c, inst); setTemplateErr(i, err) {
                         continue
                 }
-
-                var q *model.Quest
-                q, err = model.NewQuest(c, desc)
-                if setTemplateErr(i, err) {
+                if setTemplateErr(i, desc.Normalize()) {
                         continue
                 }
 
+                q := model.NewQuest(c, desc)
+
                 rsp.TemplateIds = append(rsp.TemplateIds, dm.NewQuestID(q.ID))
 
                 // if we have any errors going on, might as well skip the rest
                 if len(rsp.TemplateError) > 0 {
                         continue
                 }
 
                 anums := newAttempts.To[q.ID]
                 anums.Nums = append(anums.Nums, req.TemplateAttempt[i].Nums...)
                 if err := anums.Normalize(); err != nil {
@@ skipping 10 matching lines @@
                         Project: inst.Project, Ref: inst.Ref, Version: vers,
                         Name: inst.Specifier.TemplateName})
         }
 
         return
 }
 
 func (d *deps) EnsureGraphData(c context.Context, req *dm.EnsureGraphDataReq) (rsp *dm.EnsureGraphDataRsp, err error) {
         // TODO(riannucci): real non-execution authentication
         if req.ForExecution != nil {
+                logging.Fields{"execution": req.ForExecution.Id}.Infof(c, "on behalf of")
                 _, _, err := model.AuthenticateExecution(c, req.ForExecution)
                 if err != nil {
                         return nil, grpcutil.MaybeLogErr(c, err, codes.Unauthenticated, "bad execution auth")
                 }
         }
 
         // render any quest descirptions, templates and template attempts into
         // a single merged set of new quests and new attempts
         rsp, newQuests, newAttempts, err := renderRequest(c, req)
         if err != nil || len(rsp.TemplateError) > 0 {
                 return
         }
 
         newQuestList := make([]*model.Quest, 0, len(newQuests))
         for _, q := range newQuests {
                 newQuestList = append(newQuestList, q)
         }
 
         err = d.ensureGraphData(c, req, newQuestList, newAttempts, rsp)
 
         return
 }