Milo: Fix swarmbucket kitchen crashes

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 515 matching lines @@
                 Project: addr.Project,
                 Path:    addr.Path,
         }
         if err := as.Normalize(); err != nil {
                 return nil, errors.Annotate(err).Reason("failed to normalize annotation stream parameters").Err()
         }
 
         return &as, nil
 }
 
+// failedToStart is called in the case where logdog-only mode is on but the
+// stream doesn't exist and the swarming job is complete.  It modifies the build
+// to add information that would've otherwise been in the annotation stream.
+func failedToStart(c context.Context, build *resp.MiloBuild, res *swarming.SwarmingRpcsTaskResult, host string) error {
+        var err error
+        build.Summary.Status = resp.InfraFailure
+        build.Summary.Started, err = time.Parse(SwarmingTimeLayout, res.StartedTs)
+        if err != nil {
+                return err
+        }
+        build.Summary.Finished, err = time.Parse(SwarmingTimeLayout, res.CompletedTs)
+        if err != nil {
+                return err
+        }
+        build.Summary.Duration = build.Summary.Finished.Sub(build.Summary.Started)
+        infoComp := infoComponent(resp.InfraFailure,
+                "LogDog stream not found", "Job likely failed to start.")
+        infoComp.Started = build.Summary.Started
+        infoComp.Finished = build.Summary.Finished
+        infoComp.Duration = build.Summary.Duration
+        infoComp.Verbosity = resp.Interesting
+        build.Components = append(build.Components, infoComp)
+        return addTaskToBuild(c, host, res, build)
+}
+
 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) {
@@ skipping 49 matching lines @@
                 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:
+                                // The stream was not found.  This could be due to one of two things:
+                                // 1. The step just started and we're just waiting for the logs
+                                // to propogage to logdog.
+                                // 2. The bootsrap on the client failed, and never sent data to logdog.
+                                // This would be evident because the swarming result would be a failure.
+                                if fr.res.State == TaskCompleted {
+                                        err = failedToStart(c, &build, fr.res, svc.getHost())
+                                        return &build, err
+                                }
                                 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:
@@ skipping 138 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
 }