milo: Use service interface for swarming.

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"
-        "encoding/json"
         "fmt"
-        "io/ioutil"
         "net/http"
         "net/url"
-        "path/filepath"
         "strings"
-        "sync"
         "time"
 
         "golang.org/x/net/context"
 
         swarming "github.com/luci/luci-go/common/api/swarming/swarming/v1"
+        "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/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"
 
 // 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.
@@ skipping 15 matching lines @@
                 return nil, err
         }
         sc, err := swarming.New(&http.Client{Transport: t})
         if err != nil {
                 return nil, err
         }
         sc.BasePath = fmt.Sprintf("https://%s/_ah/api/swarming/v1/", server)
         return sc, nil
 }
 
-func getDebugTaskOutput(taskID string) (string, error) {
-        // Read the debug file instead.
+// swarmingService is an interface that fetches data from Swarming.
+//
+// In production, this is fetched from a Swarming server. For testing, this can
+// be replaced with a mock.
+type swarmingService interface {
+        getHost() string
+        getSwarmingResult(c context.Context, taskID string) (*swarming.SwarmingRpcsTaskResult, error)
+        getTaskOutput(c context.Context, taskID string) (string, error)
+}
 
-        // ../swarming below assumes that
-        // - this code is not executed by tests outside of this dir
-        // - this dir is a sibling of frontend dir
-        logFilename := filepath.Join("..", "swarming", "testdata", taskID)
-        b, err := ioutil.ReadFile(logFilename)
+type prodSwarmingService struct {
+        host   string
+        client *swarming.Service
+}
+
+func newProdService(c context.Context, server string) (*prodSwarmingService, error) {
+        client, err := getSwarmingClient(c, server)
+        if err != nil {
+                return nil, err
+        }
+        return &prodSwarmingService{
+                host:   server,
+                client: client,
+        }, nil
+}
+
+func (svc *prodSwarmingService) getHost() string { return svc.host }
+
+func (svc *prodSwarmingService) getSwarmingResult(c context.Context, taskID string) (*swarming.SwarmingRpcsTaskResult, error) {
+        return svc.client.Task.Result(taskID).Context(c).Do()
+}
+
+func (svc *prodSwarmingService) getTaskOutput(c context.Context, taskID string) (string, error) {
+        stdout, err := svc.client.Task.Stdout(taskID).Context(c).Do()
         if err != nil {
                 return "", err
         }
-        return string(b), nil
+        return stdout.Output, nil
 }
 
-func getTaskOutput(sc *swarming.Service, taskID string) (string, error) {
-        res, err := sc.Task.Stdout(taskID).Do()
-        if err != nil {
-                return "", err
-        }
-        return res.Output, nil
+type swarmingFetchParams struct {
+        fetchRes bool
+        fetchLog bool
 }
 
-func getDebugSwarmingResult(
-        taskID string) (*swarming.SwarmingRpcsTaskResult, error) {
+type swarmingFetchResult struct {
+        res *swarming.SwarmingRpcsTaskResult
+        log string
+}
 
-        // ../swarming below assumes that
-        // - this code is not executed by tests outside of this dir
-        // - this dir is a sibling of frontend dir
-        logFilename := filepath.Join("..", "swarming", "testdata", taskID)
-        swarmFilename := fmt.Sprintf("%s.swarm", logFilename)
-        s, err := ioutil.ReadFile(swarmFilename)
+// 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
+
+        err := parallel.FanOutIn(func(workC chan<- func() error) {
+                if req.fetchRes {
+                        workC <- func() (err error) {
+                                fr.res, err = svc.getSwarmingResult(c, taskID)
+                                return
+                        }
+                }
+
+                if req.fetchLog {
+                        workC <- func() error {
+                                fr.log, logErr = svc.getTaskOutput(c, taskID)
+                                return nil
+                        }
+                }
+        })
         if err != nil {
                 return nil, err
         }
-        sr := &swarming.SwarmingRpcsTaskResult{}
-        if err := json.Unmarshal(s, sr); err != nil {
-                return nil, err
-        }
-        return sr, nil
-}
 
-func getSwarming(c context.Context, server string, taskID string) (
-        *swarming.SwarmingRpcsTaskResult, string, error) {
-
-        var log string
-        var sr *swarming.SwarmingRpcsTaskResult
-        var errLog, errRes error
-
-        // Detour: Return debugging results, useful for development.
-        if server == "debug" {
-                sr, errRes = getDebugSwarmingResult(taskID)
-                log, errLog = getDebugTaskOutput(taskID)
-        } else {
-                sc, err := getSwarmingClient(c, server)
-                if err != nil {
-                        return nil, "", err
+        // Current ACL implementation: error if this is not a Milo job.
+        switch {
+        case req.fetchRes:
+                if !isMiloJob(fr.res.Tags) {
+                        return nil, errNotMiloJob
                 }
-
-                var wg sync.WaitGroup
-                wg.Add(2) // Getting log and result can happen concurrently.  Wait for both.
-
-                go func() {
-                        defer wg.Done()
-                        log, errLog = getTaskOutput(sc, taskID)
-                }()
-                go func() {
-                        defer wg.Done()
-                        sr, errRes = sc.Task.Result(taskID).Do()
-                }()
-                wg.Wait()
+        default:
+                // No metadata to decide if this is a Milo job, so assume that it is not.
+                return nil, errNotMiloJob
         }
 
-        if errRes != nil {
-                // Swarming result errors take priority.
-                return sr, log, errRes
-        }
-
-        switch sr.State {
-        case TaskCompleted, TaskRunning, TaskCanceled:
-        default:
-                //  Ignore log errors if the task might be pending, timed out, expired, etc.
-                if errLog != nil {
-                        errLog = nil
-                        log = ""
+        if req.fetchRes && logErr != nil {
+                switch fr.res.State {
+                case TaskCompleted, TaskRunning, TaskCanceled:
+                default:
+                        //  Ignore log errors if the task might be pending, timed out, expired, etc.
+                        if err != nil {
+                                fr.log = ""
+                                logErr = nil
+                        }
                 }
         }
-        return sr, log, errLog
+        return &fr, logErr
 }
 
 func taskProperties(sr *swarming.SwarmingRpcsTaskResult) *resp.PropertyGroup {
         props := &resp.PropertyGroup{GroupName: "Swarming"}
         if len(sr.CostsUsd) == 1 {
                 props.Property = append(props.Property, &resp.Property{
                         Key:   "Cost of job (USD)",
                         Value: fmt.Sprintf("$%.2f", sr.CostsUsd[0]),
                 })
         }
@@ skipping 232 matching lines @@
         }
         // If this ever has more than one stream then memoryClient needs to become
         // goroutine safe
         if err := p.RunStreams([]*annotee.Stream{&is}); err != nil {
                 return nil, err
         }
         p.Finish()
         return c.ToLogDogStreams()
 }
 
-func swarmingBuildImpl(c context.Context, linkBase, server, taskID string) (*resp.MiloBuild, error) {
+func swarmingBuildImpl(c context.Context, svc swarmingService, linkBase, taskID string) (*resp.MiloBuild, error) {
         // Fetch the data from Swarming
-        sr, body, err := getSwarming(c, server, taskID)
+        fetchParams := swarmingFetchParams{
+                fetchRes: true,
+                fetchLog: true,
+        }
+        fr, err := swarmingFetch(c, svc, taskID, fetchParams)
         if err != nil {
                 return nil, err
         }
 
-        allowMilo := false
-        for _, t := range sr.Tags {
-                if t == "allow_milo:1" {
-                        allowMilo = true
-                        break
-                }
-        }
-        if !allowMilo {
-                return nil, fmt.Errorf("Not A Milo Job")
-        }
-
         var build resp.MiloBuild
         var s *miloProto.Step
         var lds *logdog.Streams
 
         // 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.
-        if body != "" {
-                lds, err = streamsFromAnnotatedLog(c, body)
+        if fr.log != "" {
+                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(server, taskID),
+                                        URL:   taskPageURL(svc.getHost(), taskID),
                                 }},
                         }}
                 }
         }
 
         if lds != nil && lds.MainStream != nil && lds.MainStream.Data != nil {
                 s = lds.MainStream.Data
         } else {
                 s = &miloProto.Step{}
         }
 
-        if err := addTaskToMiloStep(c, server, sr, s); err != nil {
+        if err := addTaskToMiloStep(c, svc.getHost(), fr.res, s); err != nil {
                 return nil, err
         }
         logdog.AddLogDogToBuild(c, swarmingURLBuilder(linkBase), s, &build)
 
-        if err := addTaskToBuild(c, server, sr, &build); err != nil {
+        if err := addTaskToBuild(c, svc.getHost(), fr.res, &build); err != nil {
                 return nil, err
         }
 
         return &build, nil
 }
 
+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.
 // Supports server aliases.
 func taskPageURL(swarmingHostname, taskID string) string {
         return fmt.Sprintf("https://%s/user/task/%s", swarmingHostname, taskID)
 }
 
 // botPageURL returns a URL to a human-consumable page of a swarming bot.
 // Supports server aliases.
 func botPageURL(swarmingHostname, botID string) string {
         return fmt.Sprintf("https://%s/restricted/bot/%s", swarmingHostname, botID)
@@ skipping 33 matching lines @@
         case *miloProto.Link_Url:
                 return &resp.Link{
                         Label: l.Label,
                         URL:   t.Url,
                 }
 
         default:
                 return nil
         }
 }