Milo: Handle missing / transient LogDog failures.

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"
         "net/url"
         "strings"
         "time"
 
         "golang.org/x/net/context"
 
         swarming "github.com/luci/luci-go/common/api/swarming/swarming/v1"
+        "github.com/luci/luci-go/common/clock"
         "github.com/luci/luci-go/common/errors"
         "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/proto/google"
         miloProto "github.com/luci/luci-go/common/proto/milo"
         "github.com/luci/luci-go/common/sync/parallel"
         "github.com/luci/luci-go/logdog/client/annotee"
         "github.com/luci/luci-go/logdog/client/coordinator"
         "github.com/luci/luci-go/logdog/common/types"
         "github.com/luci/luci-go/milo/api/resp"
         "github.com/luci/luci-go/milo/appengine/logdog"
         "github.com/luci/luci-go/server/auth"
 )
 
 // errNotMiloJob is returned if a Swarming task is fetched that does not self-
 // identify as a Milo job.
 var errNotMiloJob = errors.New("Not a Milo Job")
 
 // SwarmingTimeLayout is time layout used by swarming.
 const SwarmingTimeLayout = "2006-01-02T15:04:05.999999999"
 
+// logDogFetchTimeout is the amount of time to wait while fetching a LogDog
+// stream before we time out the fetch.
+const logDogFetchTimeout = 30 * time.Second
+
 // Swarming task states..
 const (
         // TaskRunning means task is running.
         TaskRunning = "RUNNING"
         // TaskPending means task didn't start yet.
         TaskPending = "PENDING"
         // TaskExpired means task expired and did not start.
         TaskExpired = "EXPIRED"
         // TaskTimedOut means task started, but took too long.
         TaskTimedOut = "TIMED_OUT"
@@ skipping 457 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.
                 taskResCallback: func(res *swarming.SwarmingRpcsTaskResult) (cancelLogs bool) {
+                        // If the build hasn't started yet, then there is no LogDog log stream to
+                        // render.
+                        switch res.State {
+                        case TaskPending, TaskExpired:
+                                return false
+
+                        case TaskCanceled:
+                                // If the task wasn't created, then it wasn't started.
+                                if res.CreatedTs == "" {
+                                        return false
+                                }
+                        }
+
+                        // The task started ... is it using LogDog for logging?
                         tags := swarmingTags(res.Tags)
 
                         var err error
                         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
                 },
         }
@@ skipping 13 matching lines @@
         // As a fallback, we will try and parse the Swarming task's output for
         // annotations.
         switch {
         case logDogStreamAddr != nil:
                 // If the LogDog stream is available, load the step from that.
                 as, err := bl.newEmptyAnnotationStream(c, logDogStreamAddr)
                 if err != nil {
                         return nil, errors.Annotate(err).Reason("failed to create LogDog annotation stream").Err()
                 }
 
-                if s, err = as.Fetch(c); err != nil {
-                        return nil, errors.Annotate(err).Reason("failed to load LogDog annotation stream").Err()
-                }
-
                 prefix, _ := logDogStreamAddr.Path.Split()
                 ub = &logdog.ViewerURLBuilder{
                         Host:    logDogStreamAddr.Host,
                         Prefix:  prefix,
                         Project: logDogStreamAddr.Project,
                 }
 
+                fetchCtx, cancelFunc := clock.WithTimeout(c, logDogFetchTimeout)
+                defer cancelFunc()
+
+                if s, err = as.Fetch(fetchCtx); err != nil {
+                        logging.Fields{
+                                logging.ErrorKey: err,
+                                "addr":           logDogStreamAddr,
+                        }.Errorf(c, "Failed to load LogDog annotation stream.")
+                        build.Components = append(build.Components, infraFailureComponent("LogDog load error", err))
+                        break
+                }
+
         case fr.log != "":
                 // Decode the data using annotee. The logdog stream returned here is assumed
                 // to be consistent, which is why the following block of code are not
                 // expected to ever err out.
                 var err error
                 lds, err = streamsFromAnnotatedLog(c, fr.log)
                 if err != nil {
-                        build.Components = []*resp.BuildComponent{{
-                                Type:   resp.Summary,
-                                Label:  "Milo annotation parser",
-                                Text:   []string{err.Error()},
-                                Status: resp.InfraFailure,
-                                SubLink: []*resp.Link{{
-                                        Label: "swarming task",
-                                        URL:   taskPageURL(svc.getHost(), taskID),
-                                }},
-                        }}
+                        comp := infraFailureComponent("Milo annotation parser", err)
+                        comp.SubLink = append(comp.SubLink, &resp.Link{
+                                Label: "swarming task",
+                                URL:   taskPageURL(svc.getHost(), taskID),
+                        })
+                        build.Components = append(build.Components, comp)
                 }
 
                 if lds != nil && lds.MainStream != nil && lds.MainStream.Data != nil {
                         s = lds.MainStream.Data
                 }
                 ub = swarmingURLBuilder(linkBase)
 
         default:
                 s = &miloProto.Step{}
                 ub = swarmingURLBuilder(linkBase)
         }
 
         if err := addTaskToMiloStep(c, svc.getHost(), fr.res, s); err != nil {
                 return nil, err
         }
         logdog.AddLogDogToBuild(c, ub, s, &build)
 
         if err := addTaskToBuild(c, svc.getHost(), fr.res, &build); err != nil {
                 return nil, err
         }
 
         return &build, nil
 }
 
+func infraFailureComponent(label string, err error) *resp.BuildComponent {
+        return &resp.BuildComponent{
+                Type:   resp.Summary,
+                Label:  label,
+                Text:   []string{err.Error()},
+                Status: resp.InfraFailure,
+        }
+}
+
 func isMiloJob(tags []string) bool {
         for _, t := range tags {
                 if t == "allow_milo:1" {
                         return true
                 }
         }
         return false
 }
 
 // taskPageURL returns a URL to a human-consumable page of a swarming task.
@@ skipping 55 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
 }