Use swarming in create_pagesets CT task

ct/go/master_scripts/create_pagesets_on_workers/main.go

 // create_pagesets_on_workers is an application that creates pagesets on all CT
 // workers and uploads it to Google Storage. The requester is emailed when the task
 // is done.
 package main
 
 import (
         "flag"
         "fmt"
-        "strings"
+        "path/filepath"
         "time"
 
         "github.com/skia-dev/glog"
         "go.skia.org/infra/ct/go/ctfe/admin_tasks"
         "go.skia.org/infra/ct/go/frontend"
         "go.skia.org/infra/ct/go/master_scripts/master_common"
         "go.skia.org/infra/ct/go/util"
         "go.skia.org/infra/go/common"
+        "go.skia.org/infra/go/swarming"
         skutil "go.skia.org/infra/go/util"
 )
 
+const (
+        MAX_PAGES_PER_SWARMING_BOT = 1000
+)
+
 var (
         emails      = flag.String("emails", "", "The comma separated email addresses to notify when the task is picked up and completes.")
         gaeTaskID   = flag.Int64("gae_task_id", -1, "The key of the App Engine task. This task will be updated when the task is completed.")
         pagesetType = flag.String("pageset_type", "", "The type of pagesets to create from the Alexa CSV list. Eg: 10k, Mobile10k, All.")
         runID       = flag.String("run_id", "", "The unique run id (typically requester + timestamp).")
 
         taskCompletedSuccessfully = new(bool)
 )
 
 func sendEmail(recipients []string) {
@@ skipping 33 matching lines @@
         emailsArr = append(emailsArr, util.CtAdmins...)
         if len(emailsArr) == 0 {
                 glog.Error("At least one email address must be specified")
                 return
         }
         skutil.LogErr(frontend.UpdateWebappTaskSetStarted(&admin_tasks.RecreatePageSetsUpdateVars{}, *gaeTaskID))
         skutil.LogErr(util.SendTaskStartEmail(emailsArr, "Creating pagesets", *runID, ""))
         // Ensure webapp is updated and completion email is sent even if task fails.
         defer updateWebappTask()
         defer sendEmail(emailsArr)
-        if !*master_common.Local {
-                // Cleanup tmp files after the run.
-                defer util.CleanTmpDir()
-        }
         // Finish with glog flush and how long the task took.
         defer util.TimeTrack(time.Now(), "Creating Pagesets on Workers")
         defer glog.Flush()
 
         if *pagesetType == "" {
                 glog.Error("Must specify --pageset_type")
                 return
         }
 
-        cmd := append(master_common.WorkerSetupCmds(),
-                // The main command that runs create_pagesets on all workers.
-                fmt.Sprintf(
-                        "create_pagesets --worker_num=%s --log_dir=%s --log_id=%s --pageset_type=%s --local=%t;",
-                        util.WORKER_NUM_KEYWORD, util.GLogDir, *runID, *pagesetType, *master_common.Local))
-
-        _, err := util.SSH(strings.Join(cmd, " "), util.Slaves, util.CREATE_PAGESETS_TIMEOUT)
+        // Empty the remote dir before the workers upload to it.
+        gs, err := util.NewGsUtil(nil)
         if err != nil {
-                glog.Errorf("Error while running cmd %s: %s", cmd, err)
+                glog.Error(err)
                 return
         }
+        gsBaseDir := filepath.Join(util.SWARMING_DIR_NAME, util.PAGESETS_DIR_NAME, *pagesetType)
+        skutil.LogErr(gs.DeleteRemoteDir(gsBaseDir))
+
+        // Archive, trigger and collect swarming tasks.
+        if err := util.TriggerSwarmingTask(*pagesetType, "create_pagesets", util.CREATE_PAGESETS_ISOLATE, swarming.RECOMMENDED_HARD_TIMEOUT, swarming.RECOMMENDED_IO_TIMEOUT, MAX_PAGES_PER_SWARMING_BOT, map[string]string{}); err != nil {
+                glog.Errorf("Error encountered when swarming tasks: %s", err)
+                return
+        }
+
         *taskCompletedSuccessfully = true
 }