Backend for new run_chromium_analysis CT task

ct/go/worker_scripts/run_chromium_analysis/main.go

-// Application that captures SKPs from CT's webpage archives.
+// run_chromium_analysis is an application that runs the specified benchmark over
+// CT's webpage archives. It is intended to be run on swarming bots.
 package main
 
 import (
+        "bytes"
         "flag"
-        "fmt"
         "io/ioutil"
         "os"
-        "path"
         "path/filepath"
-        "strconv"
         "sync"
         "time"
 
         "github.com/skia-dev/glog"
 
+        "strings"
+
         "go.skia.org/infra/ct/go/util"
         "go.skia.org/infra/ct/go/worker_scripts/worker_common"
         "go.skia.org/infra/go/common"
         skutil "go.skia.org/infra/go/util"
 )
 
 const (
-        // The number of goroutines that will run in parallel to capture SKPs.
+        // The number of goroutines that will run in parallel to run benchmarks.
         WORKER_POOL_SIZE = 10
 )
 
 var (
-        startRange         = flag.Int("start_range", 1, "The number this worker will capture SKPs from.")
-        num                = flag.Int("num", 100, "The total number of SKPs to capture starting from the start_range.")
-        pagesetType        = flag.String("pageset_type", util.PAGESET_TYPE_MOBILE_10k, "The type of pagesets to create SKPs from. Eg: 10k, Mobile10k, All.")
-        chromiumBuild      = flag.String("chromium_build", "", "The chromium build that will be used to create the SKPs.")
+        startRange         = flag.Int("start_range", 1, "The number this worker will run benchmarks from.")
+        num                = flag.Int("num", 100, "The total number of benchmarks to run starting from the start_range.")
+        pagesetType        = flag.String("pageset_type", util.PAGESET_TYPE_MOBILE_10k, "The type of pagesets to create from the Alexa CSV list. Eg: 10k, Mobile10k, All.")

[dogben, 2016/05/19 21:06:41]

s/to create from the Alexa CSV list/to analyze/

[rmistry, 2016/05/20 11:07:16]

Done.

+        chromiumBuild      = flag.String("chromium_build", "", "The chromium build to use.")
         runID              = flag.String("run_id", "", "The unique run id (typically requester + timestamp).")
-        targetPlatform     = flag.String("target_platform", util.PLATFORM_LINUX, "The platform the benchmark will run on (Android / Linux).")
-        chromeCleanerTimer = flag.Duration("cleaner_timer", 30*time.Minute, "How often all chrome processes will be killed on this slave.")
+        benchmarkName      = flag.String("benchmark_name", "", "The telemetry benchmark to run on this worker.")
+        benchmarkExtraArgs = flag.String("benchmark_extra_args", "", "The extra arguments that are passed to the specified benchmark.")
+        browserExtraArgs   = flag.String("browser_extra_args", "", "The extra arguments that are passed to the browser while running the benchmark.")
+        chromeCleanerTimer = flag.Duration("cleaner_timer", 15*time.Minute, "How often all chrome processes will be killed on this slave.")
 )
 
 func main() {
         defer common.LogPanic()
         worker_common.Init()
-        defer util.TimeTrack(time.Now(), "Capturing SKPs")
+        defer util.TimeTrack(time.Now(), "Running Chromium Perf")

[dogben, 2016/05/19 21:06:41]

s/Perf/Analysis/

[rmistry, 2016/05/20 11:07:16]

Done.

         defer glog.Flush()
 
         // Validate required arguments.
         if *chromiumBuild == "" {
                 glog.Error("Must specify --chromium_build")
                 return
         }
         if *runID == "" {
                 glog.Error("Must specify --run_id")
                 return
         }
-        if *targetPlatform == util.PLATFORM_ANDROID {
-                glog.Error("Android is not yet supported for capturing SKPs.")
+        if *benchmarkName == "" {
+                glog.Error("Must specify --benchmark_name")
                 return
         }
 
         // Reset the local chromium checkout.
         if err := util.ResetCheckout(util.ChromiumSrcDir); err != nil {
                 glog.Errorf("Could not reset %s: %s", util.ChromiumSrcDir, err)
                 return
         }
         // Sync the local chromium checkout.
         if err := util.SyncDir(util.ChromiumSrcDir); err != nil {
                 glog.Errorf("Could not gclient sync %s: %s", util.ChromiumSrcDir, err)
                 return
         }
 
         // Instantiate GsUtil object.
         gs, err := util.NewGsUtil(nil)
         if err != nil {
                 glog.Error(err)
                 return
         }
 
+        // Download the benchmark patch for this run from Google storage.
+        benchmarkPatchName := *runID + ".benchmark.patch"
+        benchmarkPatchLocalPath := filepath.Join(os.TempDir(), benchmarkPatchName)

[dogben, 2016/05/19 21:06:41]

nit: ioutil.TempDir or ioutil.TempFile

[rmistry, 2016/05/20 11:07:16]

Done.

+        remotePatchesDir := filepath.Join(util.ChromiumPerfRunsDir, *runID)
+        benchmarkPatchRemotePath := filepath.Join(remotePatchesDir, benchmarkPatchName)
+        respBody, err := gs.GetRemoteFileContents(benchmarkPatchRemotePath)
+        if err != nil {
+                glog.Errorf("Could not fetch %s: %s", benchmarkPatchRemotePath, err)
+                return
+        }
+        defer skutil.Close(respBody)
+        buf := new(bytes.Buffer)
+        if _, err := buf.ReadFrom(respBody); err != nil {
+                glog.Errorf("Could not read from %s: %s", benchmarkPatchRemotePath, err)
+                return
+        }
+        if err := ioutil.WriteFile(benchmarkPatchLocalPath, buf.Bytes(), 0666); err != nil {

[dogben, 2016/05/19 21:06:41]

nit: os.Create + io.Copy is probably slightly better.

[rmistry, 2016/05/20 11:07:16]

This is also how the poller creates local patches, leaving it the same to be
slightly consistent.

+                glog.Errorf("Unable to create file %s: %s", benchmarkPatchLocalPath, err)
+                return
+        }
+        defer skutil.Remove(benchmarkPatchLocalPath)
+        // Apply benchmark patch to the local chromium checkout.
+        if buf.Len() > 10 {

[dogben, 2016/05/19 21:06:41]

(io.Copy returns the number of bytes copied)

[rmistry, 2016/05/20 11:07:16]

Acknowledged.

+                if err := util.ApplyPatch(benchmarkPatchLocalPath, util.ChromiumSrcDir); err != nil {
+                        glog.Errorf("Could not apply Telemetry's patch in %s: %s", util.ChromiumSrcDir, err)
+                        return
+                }
+        }
+
         // Download the specified chromium build.
         if err := gs.DownloadChromiumBuild(*chromiumBuild); err != nil {
                 glog.Error(err)
                 return
         }
-        // Delete the chromium build to save space when we are done.
+        //Delete the chromium build to save space when we are done.
         defer skutil.RemoveAll(filepath.Join(util.ChromiumBuildsDir, *chromiumBuild))
+
         chromiumBinary := filepath.Join(util.ChromiumBuildsDir, *chromiumBuild, util.BINARY_CHROME)
-        if *targetPlatform == util.PLATFORM_ANDROID {
-                // Install the APK on the Android device.
-                if err := util.InstallChromeAPK(*chromiumBuild); err != nil {
-                        glog.Errorf("Could not install the chromium APK: %s", err)
-                        return
-                }
-        }
 
         // Download pagesets if they do not exist locally.
         pathToPagesets := filepath.Join(util.PagesetsDir, *pagesetType)
         if _, err := gs.DownloadSwarmingArtifacts(pathToPagesets, util.PAGESETS_DIR_NAME, *pagesetType, *startRange, *num); err != nil {
                 glog.Error(err)
                 return
         }
         defer skutil.RemoveAll(pathToPagesets)
 
         // Download archives if they do not exist locally.
         pathToArchives := filepath.Join(util.WebArchivesDir, *pagesetType)
-        archivesToIndex, err := gs.DownloadSwarmingArtifacts(pathToArchives, util.WEB_ARCHIVES_DIR_NAME, *pagesetType, *startRange, *num)
-        if err != nil {
+        if _, err := gs.DownloadSwarmingArtifacts(pathToArchives, util.WEB_ARCHIVES_DIR_NAME, *pagesetType, *startRange, *num); err != nil {
                 glog.Error(err)
                 return
         }
         defer skutil.RemoveAll(pathToArchives)
 
-        // Create the dir that SKPs will be stored in.
-        pathToSkps := filepath.Join(util.SkpsDir, *pagesetType, *chromiumBuild)
-        // Delete and remake the local SKPs directory.
-        skutil.RemoveAll(pathToSkps)
-        skutil.MkdirAll(pathToSkps, 0700)
-        defer skutil.RemoveAll(pathToSkps)
+        // Establish nopatch output paths.
+        localOutputDir := filepath.Join(util.StorageDir, util.BenchmarkRunsDir, *runID)
+        skutil.RemoveAll(localOutputDir)
+        skutil.MkdirAll(localOutputDir, 0700)
+        defer skutil.RemoveAll(localOutputDir)
+        remoteDir := filepath.Join(util.BenchmarkRunsDir, *runID)
 
-        // Construct path to the ct_run_benchmark python script.
+        // Construct path to CT's python scripts.
         pathToPyFiles := util.GetPathToPyFiles(!*worker_common.Local)
 
-        timeoutSecs := util.PagesetTypeToInfo[*pagesetType].CaptureSKPsTimeoutSecs
         fileInfos, err := ioutil.ReadDir(pathToPagesets)
         if err != nil {
                 glog.Errorf("Unable to read the pagesets dir %s: %s", pathToPagesets, err)
                 return
         }
 
+        glog.Infoln("===== Going to run the task with parallel chrome processes =====")
+
         // Create channel that contains all pageset file names. This channel will
         // be consumed by the worker pool.
         pagesetRequests := util.GetClosedChannelOfPagesets(fileInfos)
 
         var wg sync.WaitGroup
         // Use a RWMutex for the chromeProcessesCleaner goroutine to communicate to
         // the workers (acting as "readers") when it wants to be the "writer" and
         // kill all zombie chrome processes.
         var mutex sync.RWMutex
 
         // Loop through workers in the worker pool.
         for i := 0; i < WORKER_POOL_SIZE; i++ {
                 // Increment the WaitGroup counter.
                 wg.Add(1)
 
                 // Create and run a goroutine closure that captures SKPs.

[dogben, 2016/05/19 21:06:41]

nit: update comment

[rmistry, 2016/05/20 11:07:16]

Done.

                 go func() {
                         // Decrement the WaitGroup counter when the goroutine completes.
                         defer wg.Done()
 
                         for pagesetName := range pagesetRequests {
 
                                 mutex.RLock()
-
-                                // Read the pageset.
-                                pagesetPath := filepath.Join(pathToPagesets, pagesetName)
-                                decodedPageset, err := util.ReadPageset(pagesetPath)
-                                if err != nil {
-                                        glog.Errorf("Could not read %s: %s", pagesetPath, err)
-                                        continue
+                                if err := util.RunBenchmark(pagesetName, pathToPagesets, pathToPyFiles, localOutputDir, *chromiumBuild, chromiumBinary, *runID, *browserExtraArgs, *benchmarkName, "Linux", *benchmarkExtraArgs, *pagesetType, -1); err != nil {
+                                        glog.Errorf("Error while running withpatch benchmark: %s", err)
+                                        return
                                 }
-
-                                glog.Infof("===== Processing %s =====", pagesetPath)
-
-                                skutil.LogErr(os.Chdir(pathToPyFiles))
-                                index, ok := archivesToIndex[decodedPageset.ArchiveDataFile]
-                                if !ok {
-                                        glog.Errorf("%s not found in the archivesToIndex map", decodedPageset.ArchiveDataFile)
-                                        continue
-                                }
-                                args := []string{
-                                        filepath.Join(util.TelemetryBinariesDir, util.BINARY_RUN_BENCHMARK),
-                                        util.BenchmarksToTelemetryName[util.BENCHMARK_SKPICTURE_PRINTER],
-                                        "--also-run-disabled-tests",
-                                        "--page-repeat=1", // Only need one run for SKPs.
-                                        "--skp-outdir=" + path.Join(pathToSkps, strconv.Itoa(index)),
-                                        "--extra-browser-args=" + util.DEFAULT_BROWSER_ARGS,
-                                        "--user-agent=" + decodedPageset.UserAgent,
-                                        "--urls-list=" + decodedPageset.UrlsList,
-                                        "--archive-data-file=" + decodedPageset.ArchiveDataFile,
-                                }
-                                // Figure out which browser and device should be used.
-                                if *targetPlatform == util.PLATFORM_ANDROID {
-                                        args = append(args, "--browser=android-chromium")
-                                } else {
-                                        args = append(args, "--browser=exact", "--browser-executable="+chromiumBinary)
-                                        args = append(args, "--device=desktop")
-                                }
-                                // Set the PYTHONPATH to the pagesets and the telemetry dirs.
-                                env := []string{
-                                        fmt.Sprintf("PYTHONPATH=%s:%s:%s:%s:$PYTHONPATH", pathToPagesets, util.TelemetryBinariesDir, util.TelemetrySrcDir, util.CatapultSrcDir),
-                                        "DISPLAY=:0",
-                                }
-                                skutil.LogErr(
-                                        util.ExecuteCmd("python", args, env, time.Duration(timeoutSecs)*time.Second, nil, nil))
-
                                 mutex.RUnlock()
                         }
                 }()
         }
 
         if !*worker_common.Local {
                 // Start the cleaner.
                 go util.ChromeProcessesCleaner(&mutex, *chromeCleanerTimer)
         }
 
         // Wait for all spawned goroutines to complete.
         wg.Wait()
 
-        // Move and validate all SKP files.
-        if err := util.ValidateSKPs(pathToSkps, pathToPyFiles); err != nil {
-                glog.Error(err)
-                return
-        }
-
-        // Upload SKPs dir to Google Storage.
-        if err := gs.UploadSwarmingArtifacts(util.SKPS_DIR_NAME, *pagesetType); err != nil {
-                glog.Error(err)
-                return
+        // If "--output-format=csv-pivot-table" was specified then merge all CSV files and upload.
+        if strings.Contains(*benchmarkExtraArgs, "--output-format=csv-pivot-table") {
+                if err := util.MergeUploadCSVFilesOnWorkers(localOutputDir, pathToPyFiles, *runID, remoteDir, gs, *startRange); err != nil {
+                        glog.Errorf("Error while processing withpatch CSV files: %s", err)
+                        return
+                }
         }
 }