Refactor distributor API so that methods always get the Quest_Desc too.

dm/appengine/distributor/swarming/v1/distributor.go

 // Copyright 2016 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 swarming
 
 import (
         "encoding/json"
         "fmt"
         "net/http"
@@ skipping 62 matching lines @@
         _, authC := httpClients(c)
         svc, err := swarm.New(authC)
         if err != nil {
                 // can only happen with nil client
                 panic(err)
         }
         svc.BasePath = cfg.Swarming.Url + "/_ah/api/swarming/v1/"
         return svc
 }
 
-func (d *swarmingDist) Run(tsk *distributor.TaskDescription) (tok distributor.Token, _ time.Duration, err error) {
-        auth := tsk.ExecutionAuth()
-        id := auth.Id
-        desc := tsk.Payload()
-
-        params := &sv1.Parameters{}
-        if err = jsonpb.UnmarshalString(desc.DistributorParameters, params); err != nil {
+func parseParams(desc *dm.Quest_Desc) (ret *sv1.Parameters, err error) {
+        ret = &sv1.Parameters{}
+        if err = jsonpb.UnmarshalString(desc.DistributorParameters, ret); err != nil {
                 err = errors.Annotate(err).
                         Reason("unmarshalling DistributorParameters").
                         InternalReason("These paramaeters were already validated?").
                         Err()
                 return
         }
-        if err = params.Normalize(); err != nil {
+        if err = ret.Normalize(); err != nil {
                 err = errors.Annotate(err).
                         Reason("normalizing DistributorParameters").
                         InternalReason("These paramaeters were already normalized successfully once?").
                         Err()
                 return
         }
+        return
+}
+
+func (d *swarmingDist) Run(desc *dm.Quest_Desc, auth *dm.Execution_Auth, prev *dm.JsonResult) (tok distributor.Token, _ time.Duration, err error) {
+        id := auth.Id
+
+        params, err := parseParams(desc)
+        if err != nil {
+                return
+        }
 
         isoCtx, _ := context.WithTimeout(d, 30*time.Second)
-        iso, err := prepIsolate(isoCtx, d.sCfg.Isolate.Url, tsk, params)
+        iso, err := prepIsolate(isoCtx, d.sCfg.Isolate.Url, desc, auth, prev, params)
         if err != nil {
                 err = errors.Annotate(err).Reason("prepping Isolated").Err()
                 return
         }
 
-        topic, token, err := tsk.PrepareTopic()
+        topic, token, err := d.cfg.PrepareTopic(d, auth.Id)
         if err != nil {
                 err = errors.Annotate(err).Reason("preparing topic").Err()
                 return
         }
 
         cipdInput := (*swarm.SwarmingRpcsCipdInput)(nil)
         if len(params.Job.Inputs.Packages) > 0 {
                 cipdInput := &swarm.SwarmingRpcsCipdInput{
                         Server: params.Job.Inputs.CipdServer,
                 }
@@ skipping 54 matching lines @@
         }, retry.LogCallback(d, "swarm.Tasks.New"))
         if err != nil {
                 err = errors.Annotate(err).Reason("calling swarm.Tasks.New").Err()
                 return
         }
 
         tok = distributor.Token(rslt.TaskId)
         return
 }
 
-func (d *swarmingDist) Cancel(tok distributor.Token) error {
+func (d *swarmingDist) Cancel(q *dm.Quest_Desc, tok distributor.Token) error {
         return retry.Retry(d, retry.Default, func() (err error) {
                 ctx, _ := context.WithTimeout(d, 10*time.Second)
                 _, err = newSwarmClient(ctx, d.sCfg).Task.Cancel(string(tok)).Context(ctx).Do()
                 return
         }, retry.LogCallback(d, "swarm.Task.Cancel"))
 }
 
-func (d *swarmingDist) GetStatus(tok distributor.Token) (*dm.Result, error) {
+func (d *swarmingDist) GetStatus(_ *dm.Quest_Desc, tok distributor.Token) (*dm.Result, error) {
         rslt := (*swarm.SwarmingRpcsTaskResult)(nil)
 
         err := retry.Retry(d, retry.Default, func() (err error) {
                 ctx, _ := context.WithTimeout(d, 10*time.Second)
                 rslt, err = newSwarmClient(ctx, d.sCfg).Task.Result(string(tok)).Context(ctx).Do()
                 return
         }, retry.LogCallback(d, fmt.Sprintf("swarm.Task.Result(%s)", tok)))
         if err != nil {
                 if gerr := err.(*googleapi.Error); gerr != nil {
                         if gerr.Code == http.StatusNotFound {
@@ skipping 56 matching lines @@
                 }
         }
 
         return ret, nil
 }
 
 func (d *swarmingDist) InfoURL(tok distributor.Token) string {
         return fmt.Sprintf("%s/user/task/%s", d.sCfg.Swarming.Url, tok)
 }
 
-func (d *swarmingDist) HandleNotification(notification *distributor.Notification) (*dm.Result, error) {
+func (d *swarmingDist) HandleNotification(q *dm.Quest_Desc, notification *distributor.Notification) (*dm.Result, error) {
         type Data struct {
                 TaskID distributor.Token `json:"task_id"`
         }
         dat := &Data{}
         if err := json.Unmarshal(notification.Data, dat); err != nil {
                 logging.Fields{"payload": notification.Data}.Errorf(
                         d, "Could not unmarshal swarming payload! relying on timeout.")
                 return nil, nil
         }
-        return d.GetStatus(dat.TaskID)
+        return d.GetStatus(q, dat.TaskID)
 }
 
 func (*swarmingDist) HandleTaskQueueTask(r *http.Request) ([]*distributor.Notification, error) {
         return nil, nil
 }
 
 func (*swarmingDist) Validate(payload string) error {
         msg := &sv1.Parameters{}
         if err := jsonpb.UnmarshalString(payload, msg); err != nil {
                 return errors.Annotate(err).Reason("unmarshal").D("payload", payload).Err()
         }
         return errors.Annotate(msg.Normalize()).Reason("normalize").D("payload", payload).Err()
 }
 
 func factory(c context.Context, cfg *distributor.Config) (distributor.D, error) {
         return &swarmingDist{c, cfg, cfg.Content.(*sv1.Config)}, nil
 }
 
 // AddFactory adds this distributor implementation into the distributor
 // Registry.
 func AddFactory(m distributor.FactoryMap) {
         m[(*sv1.Config)(nil)] = factory
 }