Initial distributor implementation

appengine/cmd/dm/deps/ensure_graph_data.go

Index: appengine/cmd/dm/deps/ensure_graph_data.go
diff --git a/appengine/cmd/dm/deps/ensure_graph_data.go b/appengine/cmd/dm/deps/ensure_graph_data.go
index 41b198063b6e3a1d91faf7adfdd9e32fde069393..2aa5446d4ae84ba6702db2abfaf52d54caeb1a9e 100644
--- a/appengine/cmd/dm/deps/ensure_graph_data.go
+++ b/appengine/cmd/dm/deps/ensure_graph_data.go
@@ -5,7 +5,6 @@
 package deps
 
 import (
-	"errors"
 	"fmt"
 
 	"golang.org/x/net/context"
@@ -18,11 +17,12 @@ import (
 	"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"
 )
@@ -221,7 +221,7 @@ func (d *deps) ensureGraphData(c context.Context, req *dm.EnsureGraphDataReq, ne
 		Auth:   req.ForExecution,
 		Quests: newQuests,
 		// Attempts we think are missing
-		Atmpts: newAttempts,
+		Attempts: newAttempts,
 		// Deps we think are missing (>= newAttempts)
 		Deps: missingDeps,
 	})
@@ -268,14 +268,28 @@ func renderRequest(c context.Context, req *dm.EnsureGraphDataReq) (rsp *dm.Ensur
 		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
 		}
 
@@ -283,7 +297,7 @@ func renderRequest(c context.Context, req *dm.EnsureGraphDataReq) (rsp *dm.Ensur
 		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),

[dnj (Google), 2016/06/09 18:00:54]

Maybe use annotated error?

[iannucci, 2016/06/15 00:45:58]

I will do so in a followup for sure!

 				codes.InvalidArgument, "no matches")
 			return
 		}
@@ -310,13 +324,12 @@ func renderRequest(c context.Context, req *dm.EnsureGraphDataReq) (rsp *dm.Ensur
 		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
@@ -347,6 +360,7 @@ func renderRequest(c context.Context, req *dm.EnsureGraphDataReq) (rsp *dm.Ensur
 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")