milo: Add SwarmBucket templating.

milo/appengine/swarming/build.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 swarming
 
 import (
         "bytes"
         "fmt"
         "net/http"
@@ skipping 99 matching lines @@
 
 func (svc *prodSwarmingService) getSwarmingRequest(c context.Context, taskID string) (*swarming.SwarmingRpcsTaskRequest, error) {
         return svc.client.Task.Request(taskID).Context(c).Do()
 }
 
 type swarmingFetchParams struct {
         fetchReq bool
         fetchRes bool
         fetchLog bool
 
-        // taskTagCallback, if not nil, is a callback that will be invoked after
+        // taskResCallback, if not nil, is a callback that will be invoked after
         // fetching the result, if fetchRes is true. It will be passed a key/value map
         // of the Swarming result's tags.
         //
-        // If taskTagCallback returns true, any pending log fetch will be cancelled
+        // If taskResCallback returns true, any pending log fetch will be cancelled
         // without error.
-        taskTagCallback func(map[string]string) bool
+        taskResCallback func(*swarming.SwarmingRpcsTaskResult) bool
 }
 
 type swarmingFetchResult struct {
         req *swarming.SwarmingRpcsTaskRequest
         res *swarming.SwarmingRpcsTaskResult
 
         // log is the log data content. If no log data was fetched, this will empty.
         // If the log fetch was cancelled, this is undefined.
         log string
 }
 
 // swarmingFetch fetches (in parallel) the components that it is configured to
 // fetch.
 //
 // After fetching, an ACL check is performed to confirm that the user is
 // permitted to view the resulting data. If this check fails, get returns
 // errNotMiloJob.
 //
 // TODO(hinoka): Make this ACL check something more specific than the
 // resence of the "allow_milo" dimension.
 func swarmingFetch(c context.Context, svc swarmingService, taskID string, req swarmingFetchParams) (
         *swarmingFetchResult, error) {
 
         // logErr is managed separately from other fetch errors, since in some
         // situations it's acceptable to not have a log stream.
         var logErr error
         var fr swarmingFetchResult
-        var resTags map[string]string
 
         // Special Context to enable the cancellation of log fetching.
         logsCancelled := false
         logCtx, cancelLogs := context.WithCancel(c)
         defer cancelLogs()
 
         err := parallel.FanOutIn(func(workC chan<- func() error) {
                 if req.fetchReq {
                         workC <- func() (err error) {
                                 fr.req, err = svc.getSwarmingRequest(c, taskID)
                                 return
                         }
                 }
 
                 if req.fetchRes {
                         workC <- func() (err error) {
                                 if fr.res, err = svc.getSwarmingResult(c, taskID); err == nil {
-                                        resTags = swarmingTags(fr.res.Tags)
-                                        if req.taskTagCallback != nil && req.taskTagCallback(resTags) {
+                                        if req.taskResCallback != nil && req.taskResCallback(fr.res) {
                                                 logsCancelled = true
                                                 cancelLogs()
                                         }
                                 }
                                 return
                         }
                 }
 
                 if req.fetchLog {
                         workC <- func() error {
@@ skipping 322 matching lines @@
 
 func (bl *buildLoader) swarmingBuildImpl(c context.Context, svc swarmingService, linkBase, taskID string) (*resp.MiloBuild, error) {
         // Fetch the data from Swarming
         var logDogStreamAddr *types.StreamAddr
 
         fetchParams := swarmingFetchParams{
                 fetchRes: true,
                 fetchLog: true,
 
                 // Cancel if LogDog annotation stream parameters are present in the tag set.
-                taskTagCallback: func(tags map[string]string) (cancelLogs bool) {
+                taskResCallback: func(res *swarming.SwarmingRpcsTaskResult) (cancelLogs bool) {
+                        tags := swarmingTags(res.Tags)
+
                         var err error
-                        if logDogStreamAddr, err = resolveLogDogStreamAddrFromTags(tags); err != nil {
+                        if logDogStreamAddr, err = resolveLogDogStreamAddrFromTags(tags, res.TaskId, res.TryNumber); err != nil {
                                 logging.WithError(err).Debugf(c, "Not using LogDog annotation stream.")
                                 return false
                         }
                         return true
                 },
         }
         fr, err := swarmingFetch(c, svc, taskID, fetchParams)
         if err != nil {
                 return nil, err
         }
@@ skipping 136 matching lines @@
         for _, tag := range v {
                 var value string
                 parts := strings.SplitN(tag, ":", 2)
                 if len(parts) == 2 {
                         value = parts[1]
                 }
                 res[parts[0]] = value
         }
         return res
 }