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"
@@ skipping 17 matching lines @@
         "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
                 },
         }
         fr, err := swarmingFetch(c, svc, taskID, fetchParams)
         if err != nil {
                 return nil, err
         }
 
         var build resp.MiloBuild
         var s *miloProto.Step
         var lds *logdog.Streams
         var ub logdog.URLBuilder
 
         // Load the build from the available data.
         //
         // If the Swarming task explicitly specifies its log location, we prefer that.
         // As a fallback, we will try and parse the Swarming task's output for
         // annotations.
         switch {
         case logDogStreamAddr != nil:
+                logging.Infof(c, "Loading build from LogDog stream at: %s", logDogStreamAddr)
+
                 // 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,
                 }
 
+                if s, err = as.Fetch(c); err != nil {
+                        switch errors.Unwrap(err) {
+                        case coordinator.ErrNoSuchStream:
+                                logging.WithError(err).Errorf(c, "User cannot access stream.")
+                                build.Components = append(build.Components, infoComponent(resp.Running,
+                                        "Waiting...", "waiting for annotation stream"))
+
+                        case coordinator.ErrNoAccess:
+                                logging.WithError(err).Errorf(c, "User cannot access stream.")
+                                build.Components = append(build.Components, infoComponent(resp.Failure,
+                                        "No Access", "no access to annotation stream"))
+
+                        default:
+                                logging.WithError(err).Errorf(c, "Failed to load LogDog annotation stream.")
+                                build.Components = append(build.Components, infoComponent(resp.InfraFailure,
+                                        "Error", "failed to load annotation stream"))
+                        }
+                }
+
         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 := infoComponent(resp.InfraFailure, "Milo annotation parser", err.Error())
+                        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
+        if s != nil {
+                if err := addTaskToMiloStep(c, svc.getHost(), fr.res, s); err != nil {
+                        return nil, err
+                }
+                logdog.AddLogDogToBuild(c, ub, s, &build)
         }
-        logdog.AddLogDogToBuild(c, ub, s, &build)
 
         if err := addTaskToBuild(c, svc.getHost(), fr.res, &build); err != nil {
                 return nil, err
         }
 
         return &build, nil
 }
 
+func infoComponent(st resp.Status, label, text string) *resp.BuildComponent {
+        return &resp.BuildComponent{
+                Type:   resp.Summary,
+                Label:  label,
+                Text:   []string{text},
+                Status: st,
+        }
+}
+
 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
 }