ct/go/util/gs.go - buildbot - Git at Google

 // Google Storage utility that contains methods for both CT master and worker
 // scripts.
 package util

 import (
 	"fmt"
 	"io"
 	"net/http"
 	"os"
 	"path"
 	"path/filepath"
 	"runtime"
 	"strconv"
 	"strings"
 	"sync"
 	"time"

 	"go.skia.org/infra/go/auth"
 	"go.skia.org/infra/go/gcs"
 	"go.skia.org/infra/go/httputils"
 	"go.skia.org/infra/go/sklog"
 	"go.skia.org/infra/go/util"
 	"go.skia.org/infra/go/util/zip"
 	"google.golang.org/api/googleapi"
 	storage "google.golang.org/api/storage/v1"
 )

 const (
 	DOWNLOAD_UPLOAD_GOROUTINE_POOL_SIZE = 30
 	// Use larger pool size for deletions. This is useful when deleting directories
 	// with 1M/1B subdirectories from the master. Google Storage will not be overwhelmed
 	// because all slaves do not do large scale deletions at the same time.
 	DELETE_GOROUTINE_POOL_SIZE = 1000
 )

 type GcsUtil struct {
 	// The client used to connect to Google Storage.
 	client  *http.Client
 	service *storage.Service
 }

 // NewGcsUtil initializes and returns a utility for CT interations with Google
 // Storage. If client is nil then a client is created either from ClientSecretPath or with the
 // default token source.
 func NewGcsUtil(client *http.Client) (*GcsUtil, error) {
 	if client == nil {
 		clientConfig := httputils.DefaultClientConfig().With2xxOnly()
 		// If ClientSecretPath exists then assume that we do not use the default token source.
 		if _, err := os.Stat(ClientSecretPath); err == nil {
 			ts, err := auth.NewLegacyTokenSource(true, GCSTokenPath, ClientSecretPath, auth.SCOPE_FULL_CONTROL)
 			if err != nil {
 				return nil, err
 			}
 			clientConfig = clientConfig.WithTokenSource(ts)
 		} else {
 			ts, err := auth.NewDefaultTokenSource(false, auth.SCOPE_FULL_CONTROL)
 			if err != nil {
 				return nil, fmt.Errorf("Problem setting up default token source: %s", err)
 			}
 			clientConfig = clientConfig.WithTokenSource(ts)
 		}
 		client = clientConfig.Client()
 	}
 	client.Timeout = HTTP_CLIENT_TIMEOUT
 	service, err := storage.New(client)
 	if err != nil {
 		return nil, fmt.Errorf("Failed to create interface to Google Storage: %s", err)
 	}
 	return &GcsUtil{client: client, service: service}, nil
 }

 // Returns the response body of the specified GCS object. Tries MAX_URI_GET_TRIES
 // times if download is unsuccessful. Client must close the response body when
 // finished with it.
 func getRespBody(res *storage.Object, client *http.Client) (io.ReadCloser, error) {
 	for i := 0; i < MAX_URI_GET_TRIES; i++ {
 		request, err := gcs.RequestForStorageURL(res.MediaLink)
 		if err != nil {
 			sklog.Warningf("Unable to create Storage MediaURI request: %s\n", err)
 			continue
 		}

 		resp, err := client.Do(request)
 		if err != nil {
 			sklog.Warningf("Unable to retrieve Storage MediaURI: %s", err)
 			continue
 		}
 		if resp.StatusCode != 200 {
 			sklog.Warningf("Failed to retrieve: %d  %s", resp.StatusCode, resp.Status)
 			util.Close(resp.Body)
 			continue
 		}
 		return resp.Body, nil
 	}
 	return nil, fmt.Errorf("Failed fetching file after %d attempts", MAX_URI_GET_TRIES)
 }

 // Returns the response body of the specified GCS file from the default CT
 // bucket. Client must close the response body when finished with it.
 func (gs *GcsUtil) GetRemoteFileContents(filePath string) (io.ReadCloser, error) {
 	return gs.GetRemoteFileContentsFromBucket(GCSBucketName, filePath)
 }

 // Returns the response body of the specified GCS file. Client must close the
 // response body when finished with it.
 func (gs *GcsUtil) GetRemoteFileContentsFromBucket(bucket, filePath string) (io.ReadCloser, error) {
 	res, err := gs.service.Objects.Get(bucket, filePath).Do()
 	if err != nil {
 		return nil, fmt.Errorf("Could not get %s from GCS: %s", filePath, err)
 	}
 	return getRespBody(res, gs.client)
 }

 type filePathToStorageObject struct {
 	storageObject *storage.Object
 	filePath      string
 }

 // downloadRemoteDir downloads the specified Google Storage dir to the specified
 // local dir. The local dir will be emptied and recreated. Handles multiple levels
 // of directories.
 func (gs *GcsUtil) downloadRemoteDir(localDir, gsDir string) error {
 	// Empty the local dir.
 	util.RemoveAll(localDir)
 	// Create the local dir.
 	MkdirAll(localDir, 0700)
 	// The channel where the storage objects to be downloaded will be sent to.
 	chStorageObjects := make(chan filePathToStorageObject, DOWNLOAD_UPLOAD_GOROUTINE_POOL_SIZE)

 	// Kick off one goroutine to populate the channel.
 	errPopulator := make(chan error, 1)
 	var wgPopulator sync.WaitGroup
 	wgPopulator.Add(1)
 	go func() {
 		defer wgPopulator.Done()
 		defer close(chStorageObjects)
 		req := gs.service.Objects.List(GCSBucketName).Prefix(gsDir + "/")
 		for req != nil {
 			resp, err := req.Do()
 			if err != nil {
 				errPopulator <- fmt.Errorf("Error occured while listing %s: %s", gsDir, err)
 				return
 			}
 			for _, result := range resp.Items {
 				fileName := path.Base(result.Name)
 				// If downloading from subdir then add it to the fileName.
 				fileGsDir := path.Dir(result.Name)
 				subDirs := strings.TrimPrefix(fileGsDir, gsDir)
 				if subDirs != "" {
 					dirTokens := strings.Split(subDirs, "/")
 					for i := range dirTokens {
 						fileName = filepath.Join(dirTokens[len(dirTokens)-i-1], fileName)
 					}
 					// Create the local directory.
 					MkdirAll(filepath.Join(localDir, filepath.Dir(fileName)), 0700)
 				}
 				chStorageObjects <- filePathToStorageObject{storageObject: result, filePath: fileName}
 			}
 			if len(resp.NextPageToken) > 0 {
 				req.PageToken(resp.NextPageToken)
 			} else {
 				req = nil
 			}
 		}
 	}()

 	// Kick off goroutines to download the storage objects.
 	var wgConsumer sync.WaitGroup
 	for i := 0; i < DOWNLOAD_UPLOAD_GOROUTINE_POOL_SIZE; i++ {
 		wgConsumer.Add(1)
 		go func(goroutineNum int) {
 			defer wgConsumer.Done()
 			for obj := range chStorageObjects {
 				if err := downloadStorageObj(obj, gs.client, localDir, goroutineNum); err != nil {
 					sklog.Errorf("Could not download storage object: %s", err)
 					return
 				}
 				// Sleep for a second after downloading file to avoid bombarding Cloud
 				// storage.
 				time.Sleep(time.Second)
 			}
 		}(i + 1)
 	}

 	wgPopulator.Wait()
 	wgConsumer.Wait()
 	// Check if there was an error listing the GCS dir.
 	select {
 	case err, ok := <-errPopulator:
 		if ok {
 			return err
 		}
 	default:
 	}
 	return nil
 }

 func downloadStorageObj(obj filePathToStorageObject, c *http.Client, localDir string, goroutineNum int) error {
 	result := obj.storageObject
 	filePath := obj.filePath
 	respBody, err := getRespBody(result, c)
 	if err != nil {
 		return fmt.Errorf("Could not fetch %s: %s", result.MediaLink, err)
 	}
 	defer util.Close(respBody)
 	outputFile := filepath.Join(localDir, filePath)
 	out, err := os.Create(outputFile)
 	if err != nil {
 		return fmt.Errorf("Unable to create file %s: %s", outputFile, err)
 	}
 	defer util.Close(out)
 	if _, err = io.Copy(out, respBody); err != nil {
 		return err
 	}
 	return nil
 }

 // DownloadChromiumBuild downloads the specified Chromium build from Google
 // Storage to a local dir.
 func (gs *GcsUtil) DownloadChromiumBuild(chromiumBuild string) error {
 	localDir := filepath.Join(ChromiumBuildsDir, chromiumBuild)
 	gsDir := path.Join(CHROMIUM_BUILDS_DIR_NAME, chromiumBuild)
 	sklog.Infof("Downloading %s from Google Storage to %s", gsDir, localDir)
 	if err := gs.downloadRemoteDir(localDir, gsDir); err != nil {
 		return fmt.Errorf("Error downloading %s into %s: %s", gsDir, localDir, err)
 	}

 	// Unzip the build.
 	zipFilePath := filepath.Join(localDir, CHROMIUM_BUILD_ZIP_NAME)
 	if err := zip.UnZip(localDir, zipFilePath); err != nil {
 		return fmt.Errorf("Error when unzipping %s: %s", zipFilePath, err)
 	}

 	if runtime.GOOS != "windows" {
 		// Downloaded chrome binary needs to be set as an executable on linux.
 		util.LogErr(os.Chmod(filepath.Join(localDir, "chrome"), 0777))
 	}

 	return nil
 }

 // DeleteRemoteDirLogErr wraps DeleteRemoteDir and logs an error if one is returned.
 func (gs *GcsUtil) DeleteRemoteDirLogErr(gsDir string) {
 	if err := gs.DeleteRemoteDir(gsDir); err != nil {
 		sklog.Errorf("Failed to DeleteRemoteDir(%s): %s", gsDir, err)
 	}
 }

 func (gs *GcsUtil) DeleteRemoteDir(gsDir string) error {
 	// The channel where the GCS filepaths to be deleted will be sent to.
 	chFilePaths := make(chan string, DELETE_GOROUTINE_POOL_SIZE)

 	// Kick off one goroutine to populate the channel.
 	errPopulator := make(chan error, 1)
 	var wgPopulator sync.WaitGroup
 	wgPopulator.Add(1)
 	go func() {
 		defer wgPopulator.Done()
 		defer close(chFilePaths)
 		req := gs.service.Objects.List(GCSBucketName).Prefix(gsDir + "/")
 		for req != nil {
 			resp, err := req.Do()
 			if err != nil {
 				errPopulator <- fmt.Errorf("Error occured while listing %s: %s", gsDir, err)
 				return
 			}
 			for _, result := range resp.Items {
 				chFilePaths <- result.Name
 			}
 			if len(resp.NextPageToken) > 0 {
 				req.PageToken(resp.NextPageToken)
 			} else {
 				req = nil
 			}
 		}
 	}()

 	// Kick off goroutines to delete the file paths.
 	var wgConsumer sync.WaitGroup
 	for i := 0; i < DELETE_GOROUTINE_POOL_SIZE; i++ {
 		wgConsumer.Add(1)
 		go func(goroutineNum int) {
 			defer wgConsumer.Done()
 			for filePath := range chFilePaths {
 				if err := gs.service.Objects.Delete(GCSBucketName, filePath).Do(); err != nil {
 					sklog.Errorf("Goroutine#%d could not delete %s: %s", goroutineNum, filePath, err)
 					return
 				}
 				// Sleep for a second after deleting file to avoid bombarding Cloud
 				// storage.
 				time.Sleep(time.Second)
 			}
 		}(i + 1)
 	}

 	wgPopulator.Wait()
 	wgConsumer.Wait()
 	// Check if there was an error listing the GCS dir.
 	select {
 	case err, ok := <-errPopulator:
 		if ok {
 			return err
 		}
 	default:
 	}
 	return nil
 }

 // UploadFile calls UploadFileToBucket with CT's default bucket.
 func (gs *GcsUtil) UploadFile(fileName, localDir, gsDir string) error {
 	return gs.UploadFileToBucket(fileName, localDir, gsDir, GCSBucketName)
 }

 // UploadFileToBucket uploads the specified file to the remote dir of the bucket
 // in Google Storage. It also sets the appropriate ACLs on the uploaded file.
 func (gs *GcsUtil) UploadFileToBucket(fileName, localDir, gsDir, bucket string) error {
 	localFile := filepath.Join(localDir, fileName)
 	gsFile := path.Join(gsDir, fileName)
 	object := &storage.Object{
 		Name: gsFile,
 		// All objects uploaded to CT's bucket via this util must be readable by
 		// the google.com domain. This will be fine tuned later if required.
 		Acl: []*storage.ObjectAccessControl{
 			{
 				Bucket: bucket,
 				Entity: "domain-google.com",
 				Object: gsFile,
 				Role:   "READER",
 			},
 		},
 	}
 	f, err := os.Open(localFile)
 	if err != nil {
 		return fmt.Errorf("Error opening %s: %s", localFile, err)
 	}
 	defer util.Close(f)
 	// TODO(rmistry): gs api now enables resumable uploads by default. Handle 308
 	// response codes.
 	fi, err := f.Stat()
 	if err != nil {
 		return fmt.Errorf("Error stating %s: %s", localFile, err)
 	}
 	mediaOption := googleapi.ChunkSize(int(fi.Size()))
 	if _, err := gs.service.Objects.Insert(bucket, object).Media(f, mediaOption).Do(); err != nil {
 		return fmt.Errorf("Objects.Insert failed: %s", err)
 	}
 	sklog.Infof("Copied %s to %s", localFile, fmt.Sprintf("gs://%s/%s", bucket, gsFile))
 	return nil
 }

 // UploadSwarmingArtifact uploads the specified local artifacts to Google Storage.
 func (gs *GcsUtil) UploadSwarmingArtifacts(dirName, pagesetType string) error {
 	localDir := path.Join(StorageDir, dirName, pagesetType)
 	gsDir := path.Join(SWARMING_DIR_NAME, dirName, pagesetType)

 	return gs.UploadDir(localDir, gsDir, false)
 }

 // DownloadSwarmingArtifacts downloads the specified artifacts from Google Storage to a local dir.
 // The Google storage directory is assumed to have numerical subdirs Eg: {1..1000}. This function
 // downloads the contents of those directories into a local directory without the numerical
 // subdirs.
 // Returns the ranking/index of the downloaded artifact.
 func (gs *GcsUtil) DownloadSwarmingArtifacts(localDir, remoteDirName, pagesetType string, startRange, num int) (map[string]int, error) {
 	// Empty the local dir.
 	util.RemoveAll(localDir)
 	// Create the local dir.
 	MkdirAll(localDir, 0700)

 	gsDir := path.Join(SWARMING_DIR_NAME, remoteDirName, pagesetType)
 	endRange := num + startRange - 1
 	// The channel where remote files to be downloaded will be sent to.
 	chRemoteDirs := make(chan string, num)
 	for i := startRange; i <= endRange; i++ {
 		chRemoteDirs <- path.Join(gsDir, strconv.Itoa(i))
 	}
 	close(chRemoteDirs)

 	// Dictionary of artifacts to its rank/index.
 	artifactToIndex := map[string]int{}
 	// Mutex to control access to the above dictionary.
 	var mtx sync.Mutex
 	// Kick off goroutines to download artifacts and populate the artifactToIndex dictionary.
 	var wg sync.WaitGroup
 	for i := 0; i < DOWNLOAD_UPLOAD_GOROUTINE_POOL_SIZE; i++ {
 		wg.Add(1)
 		go func(goroutineNum int) {
 			defer wg.Done()
 			for remoteDir := range chRemoteDirs {
 				if err := gs.downloadFromSwarmingDir(remoteDir, gsDir, localDir, goroutineNum, &mtx, artifactToIndex); err != nil {
 					sklog.Error(err)
 					return
 				}
 			}
 		}(i + 1)
 	}
 	wg.Wait()
 	if len(chRemoteDirs) != 0 {
 		return artifactToIndex, fmt.Errorf("Unable to download all artifacts.")
 	}
 	return artifactToIndex, nil
 }

 // GetRemoteDirCount returns the number of objects in the specified dir.
 func (gs *GcsUtil) GetRemoteDirCount(gsDir string) (int, error) {
 	req := gs.service.Objects.List(GCSBucketName).Prefix(gsDir + "/")
 	count := 0
 	for req != nil {
 		resp, err := req.Do()
 		if err != nil {
 			return -1, fmt.Errorf("Error occured while listing %s: %s", gsDir, err)
 		}
 		count += len(resp.Items)
 		if len(resp.NextPageToken) > 0 {
 			req.PageToken(resp.NextPageToken)
 		} else {
 			req = nil
 		}
 	}
 	return count, nil
 }

 func (gs *GcsUtil) downloadFromSwarmingDir(remoteDir, gsDir, localDir string, runID int, mtx *sync.Mutex, artifactToIndex map[string]int) error {
 	req := gs.service.Objects.List(GCSBucketName).Prefix(remoteDir + "/")
 	for req != nil {
 		resp, err := req.Do()
 		if err != nil {
 			return fmt.Errorf("Error occured while listing %s: %s", gsDir, err)
 		}
 		for _, result := range resp.Items {
 			fileName := path.Base(result.Name)
 			fileGsDir := path.Dir(result.Name)
 			index, err := strconv.Atoi(path.Base(fileGsDir))
 			if err != nil {
 				return fmt.Errorf("%s was not in expected format: %s", fileGsDir, err)
 			}
 			respBody, err := getRespBody(result, gs.client)
 			if err != nil {
 				return fmt.Errorf("Could not fetch %s: %s", result.MediaLink, err)
 			}
 			defer util.Close(respBody)
 			outputFile := filepath.Join(localDir, fileName)
 			out, err := os.Create(outputFile)
 			if err != nil {
 				return fmt.Errorf("Unable to create file %s: %s", outputFile, err)
 			}
 			defer util.Close(out)
 			if _, err = io.Copy(out, respBody); err != nil {
 				return err
 			}
 			// Sleep for a second after downloading file to avoid bombarding Cloud
 			// storage.
 			time.Sleep(time.Second)
 			mtx.Lock()
 			artifactToIndex[path.Join(localDir, fileName)] = index
 			mtx.Unlock()
 		}
 		if len(resp.NextPageToken) > 0 {
 			req.PageToken(resp.NextPageToken)
 		} else {
 			req = nil
 		}
 	}
 	return nil
 }

 // UploadDir uploads the specified local dir into the specified Google Storage dir.
 func (gs *GcsUtil) UploadDir(localDir, gsDir string, cleanDir bool) error {
 	if cleanDir {
 		// Empty the remote dir before uploading to it.
 		util.LogErr(gs.DeleteRemoteDir(gsDir))
 	}

 	// Construct a dictionary of file paths to their file infos.
 	pathsToFileInfos := map[string]os.FileInfo{}
 	visit := func(path string, f os.FileInfo, err error) error {
 		if f.IsDir() {
 			return nil
 		}
 		pathsToFileInfos[path] = f
 		return nil
 	}
 	if err := filepath.Walk(localDir, visit); err != nil {
 		return fmt.Errorf("Unable to read the local dir %s: %s", localDir, err)
 	}

 	// The channel where the filepaths to be uploaded will be sent to.
 	chFilePaths := make(chan string, len(pathsToFileInfos))
 	// File filepaths and send it to the above channel.
 	for path, fileInfo := range pathsToFileInfos {
 		fileName := fileInfo.Name()
 		containingDir := strings.TrimSuffix(path, fileName)
 		subDirs := strings.TrimPrefix(containingDir, localDir)
 		if subDirs != "" {
 			dirTokens := strings.Split(subDirs, "/")
 			for i := range dirTokens {
 				fileName = filepath.Join(dirTokens[len(dirTokens)-i-1], fileName)
 			}
 		}
 		chFilePaths <- fileName
 	}
 	close(chFilePaths)

 	// Kick off goroutines to upload the file paths.
 	var wg sync.WaitGroup
 	for i := 0; i < DOWNLOAD_UPLOAD_GOROUTINE_POOL_SIZE; i++ {
 		wg.Add(1)
 		go func(goroutineNum int) {
 			defer wg.Done()
 			for filePath := range chFilePaths {
 				if err := gs.UploadFile(filePath, localDir, gsDir); err != nil {
 					sklog.Errorf("Goroutine#%d could not upload %s to %s: %s", goroutineNum, filePath, localDir, err)
 				}
 				// Sleep for a second after uploading file to avoid bombarding Cloud
 				// storage.
 				time.Sleep(time.Second)
 			}
 		}(i + 1)
 	}
 	wg.Wait()
 	return nil
 }

 // DownloadRemoteFile calls DownloadRemoteFileFromBucket with CT's default bucket.
 func (gs *GcsUtil) DownloadRemoteFile(remotePath, localPath string) error {
 	return gs.DownloadRemoteFileFromBucket(GCSBucketName, remotePath, localPath)
 }

 // DownloadRemoteFileFromBucket downloads the specified remote path into the specified
 // local file. This function has been tested to download very large files (~33GB).
 // TODO(rmistry): Update all code that downloads remote files to use this or the
 // DownloadRemoteFile method.
 func (gs *GcsUtil) DownloadRemoteFileFromBucket(bucket, remotePath, localPath string) error {
 	respBody, err := gs.GetRemoteFileContentsFromBucket(bucket, remotePath)
 	if err != nil {
 		return err
 	}
 	defer util.Close(respBody)
 	out, err := os.Create(localPath)
 	if err != nil {
 		return err
 	}

 	bufferSize := int64(1024 * 1024 * 1024)
 	for {
 		_, err := io.CopyN(out, respBody, bufferSize)
 		if err == io.EOF {
 			break
 		} else if err != nil {
 			defer util.Close(out)
 			return err
 		}
 		// Sleep for 30 seconds. Bots run out of memory without this.
 		// Eg: https://chromium-swarm.appspot.com/task?id=2fba9fba3d553510
 		// Maybe this sleep gives Golang time to clear some caches.
 		time.Sleep(30 * time.Second)
 	}
 	if err := out.Close(); err != nil {
 		return err
 	}
 	return nil
 }
	// Google Storage utility that contains methods for both CT master and worker
	// scripts.
	package util

	import (
	"fmt"
	"io"
	"net/http"
	"os"
	"path"
	"path/filepath"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"time"

	"go.skia.org/infra/go/auth"
	"go.skia.org/infra/go/gcs"
	"go.skia.org/infra/go/httputils"
	"go.skia.org/infra/go/sklog"
	"go.skia.org/infra/go/util"
	"go.skia.org/infra/go/util/zip"
	"google.golang.org/api/googleapi"
	storage "google.golang.org/api/storage/v1"
	)

	const (
	DOWNLOAD_UPLOAD_GOROUTINE_POOL_SIZE = 30
	// Use larger pool size for deletions. This is useful when deleting directories
	// with 1M/1B subdirectories from the master. Google Storage will not be overwhelmed
	// because all slaves do not do large scale deletions at the same time.
	DELETE_GOROUTINE_POOL_SIZE = 1000
	)

	type GcsUtil struct {
	// The client used to connect to Google Storage.
	client *http.Client
	service *storage.Service
	}

	// NewGcsUtil initializes and returns a utility for CT interations with Google
	// Storage. If client is nil then a client is created either from ClientSecretPath or with the
	// default token source.
	func NewGcsUtil(client http.Client) (GcsUtil, error) {
	if client == nil {
	clientConfig := httputils.DefaultClientConfig().With2xxOnly()
	// If ClientSecretPath exists then assume that we do not use the default token source.
	if _, err := os.Stat(ClientSecretPath); err == nil {
	ts, err := auth.NewLegacyTokenSource(true, GCSTokenPath, ClientSecretPath, auth.SCOPE_FULL_CONTROL)
	if err != nil {
	return nil, err
	}
	clientConfig = clientConfig.WithTokenSource(ts)
	} else {
	ts, err := auth.NewDefaultTokenSource(false, auth.SCOPE_FULL_CONTROL)
	if err != nil {
	return nil, fmt.Errorf("Problem setting up default token source: %s", err)
	}
	clientConfig = clientConfig.WithTokenSource(ts)
	}
	client = clientConfig.Client()
	}
	client.Timeout = HTTP_CLIENT_TIMEOUT
	service, err := storage.New(client)
	if err != nil {
	return nil, fmt.Errorf("Failed to create interface to Google Storage: %s", err)
	}
	return &GcsUtil{client: client, service: service}, nil
	}

	// Returns the response body of the specified GCS object. Tries MAX_URI_GET_TRIES
	// times if download is unsuccessful. Client must close the response body when
	// finished with it.
	func getRespBody(res storage.Object, client http.Client) (io.ReadCloser, error) {
	for i := 0; i < MAX_URI_GET_TRIES; i++ {
	request, err := gcs.RequestForStorageURL(res.MediaLink)
	if err != nil {
	sklog.Warningf("Unable to create Storage MediaURI request: %s\n", err)
	continue
	}

	resp, err := client.Do(request)
	if err != nil {
	sklog.Warningf("Unable to retrieve Storage MediaURI: %s", err)
	continue
	}
	if resp.StatusCode != 200 {
	sklog.Warningf("Failed to retrieve: %d %s", resp.StatusCode, resp.Status)
	util.Close(resp.Body)
	continue
	}
	return resp.Body, nil
	}
	return nil, fmt.Errorf("Failed fetching file after %d attempts", MAX_URI_GET_TRIES)
	}

	// Returns the response body of the specified GCS file from the default CT
	// bucket. Client must close the response body when finished with it.
	func (gs *GcsUtil) GetRemoteFileContents(filePath string) (io.ReadCloser, error) {
	return gs.GetRemoteFileContentsFromBucket(GCSBucketName, filePath)
	}

	// Returns the response body of the specified GCS file. Client must close the
	// response body when finished with it.
	func (gs *GcsUtil) GetRemoteFileContentsFromBucket(bucket, filePath string) (io.ReadCloser, error) {
	res, err := gs.service.Objects.Get(bucket, filePath).Do()
	if err != nil {
	return nil, fmt.Errorf("Could not get %s from GCS: %s", filePath, err)
	}
	return getRespBody(res, gs.client)
	}

	type filePathToStorageObject struct {
	storageObject *storage.Object
	filePath string
	}

	// downloadRemoteDir downloads the specified Google Storage dir to the specified
	// local dir. The local dir will be emptied and recreated. Handles multiple levels
	// of directories.
	func (gs *GcsUtil) downloadRemoteDir(localDir, gsDir string) error {
	// Empty the local dir.
	util.RemoveAll(localDir)
	// Create the local dir.
	MkdirAll(localDir, 0700)
	// The channel where the storage objects to be downloaded will be sent to.
	chStorageObjects := make(chan filePathToStorageObject, DOWNLOAD_UPLOAD_GOROUTINE_POOL_SIZE)

	// Kick off one goroutine to populate the channel.
	errPopulator := make(chan error, 1)
	var wgPopulator sync.WaitGroup
	wgPopulator.Add(1)
	go func() {
	defer wgPopulator.Done()
	defer close(chStorageObjects)
	req := gs.service.Objects.List(GCSBucketName).Prefix(gsDir + "/")
	for req != nil {
	resp, err := req.Do()
	if err != nil {
	errPopulator <- fmt.Errorf("Error occured while listing %s: %s", gsDir, err)
	return
	}
	for _, result := range resp.Items {
	fileName := path.Base(result.Name)
	// If downloading from subdir then add it to the fileName.
	fileGsDir := path.Dir(result.Name)
	subDirs := strings.TrimPrefix(fileGsDir, gsDir)
	if subDirs != "" {
	dirTokens := strings.Split(subDirs, "/")
	for i := range dirTokens {
	fileName = filepath.Join(dirTokens[len(dirTokens)-i-1], fileName)
	}
	// Create the local directory.
	MkdirAll(filepath.Join(localDir, filepath.Dir(fileName)), 0700)
	}
	chStorageObjects <- filePathToStorageObject{storageObject: result, filePath: fileName}
	}
	if len(resp.NextPageToken) > 0 {
	req.PageToken(resp.NextPageToken)
	} else {
	req = nil
	}
	}
	}()

	// Kick off goroutines to download the storage objects.
	var wgConsumer sync.WaitGroup
	for i := 0; i < DOWNLOAD_UPLOAD_GOROUTINE_POOL_SIZE; i++ {
	wgConsumer.Add(1)
	go func(goroutineNum int) {
	defer wgConsumer.Done()
	for obj := range chStorageObjects {
	if err := downloadStorageObj(obj, gs.client, localDir, goroutineNum); err != nil {
	sklog.Errorf("Could not download storage object: %s", err)
	return
	}
	// Sleep for a second after downloading file to avoid bombarding Cloud
	// storage.
	time.Sleep(time.Second)
	}
	}(i + 1)
	}

	wgPopulator.Wait()
	wgConsumer.Wait()
	// Check if there was an error listing the GCS dir.
	select {
	case err, ok := <-errPopulator:
	if ok {
	return err
	}
	default:
	}
	return nil
	}

	func downloadStorageObj(obj filePathToStorageObject, c *http.Client, localDir string, goroutineNum int) error {
	result := obj.storageObject
	filePath := obj.filePath
	respBody, err := getRespBody(result, c)
	if err != nil {
	return fmt.Errorf("Could not fetch %s: %s", result.MediaLink, err)
	}
	defer util.Close(respBody)
	outputFile := filepath.Join(localDir, filePath)
	out, err := os.Create(outputFile)
	if err != nil {
	return fmt.Errorf("Unable to create file %s: %s", outputFile, err)
	}
	defer util.Close(out)
	if _, err = io.Copy(out, respBody); err != nil {
	return err
	}
	return nil
	}

	// DownloadChromiumBuild downloads the specified Chromium build from Google
	// Storage to a local dir.
	func (gs *GcsUtil) DownloadChromiumBuild(chromiumBuild string) error {
	localDir := filepath.Join(ChromiumBuildsDir, chromiumBuild)
	gsDir := path.Join(CHROMIUM_BUILDS_DIR_NAME, chromiumBuild)
	sklog.Infof("Downloading %s from Google Storage to %s", gsDir, localDir)
	if err := gs.downloadRemoteDir(localDir, gsDir); err != nil {
	return fmt.Errorf("Error downloading %s into %s: %s", gsDir, localDir, err)
	}

	// Unzip the build.
	zipFilePath := filepath.Join(localDir, CHROMIUM_BUILD_ZIP_NAME)
	if err := zip.UnZip(localDir, zipFilePath); err != nil {
	return fmt.Errorf("Error when unzipping %s: %s", zipFilePath, err)
	}

	if runtime.GOOS != "windows" {
	// Downloaded chrome binary needs to be set as an executable on linux.
	util.LogErr(os.Chmod(filepath.Join(localDir, "chrome"), 0777))
	}

	return nil
	}

	// DeleteRemoteDirLogErr wraps DeleteRemoteDir and logs an error if one is returned.
	func (gs *GcsUtil) DeleteRemoteDirLogErr(gsDir string) {
	if err := gs.DeleteRemoteDir(gsDir); err != nil {
	sklog.Errorf("Failed to DeleteRemoteDir(%s): %s", gsDir, err)
	}
	}

	func (gs *GcsUtil) DeleteRemoteDir(gsDir string) error {
	// The channel where the GCS filepaths to be deleted will be sent to.
	chFilePaths := make(chan string, DELETE_GOROUTINE_POOL_SIZE)

	// Kick off one goroutine to populate the channel.
	errPopulator := make(chan error, 1)
	var wgPopulator sync.WaitGroup
	wgPopulator.Add(1)
	go func() {
	defer wgPopulator.Done()
	defer close(chFilePaths)
	req := gs.service.Objects.List(GCSBucketName).Prefix(gsDir + "/")
	for req != nil {
	resp, err := req.Do()
	if err != nil {
	errPopulator <- fmt.Errorf("Error occured while listing %s: %s", gsDir, err)
	return
	}
	for _, result := range resp.Items {
	chFilePaths <- result.Name
	}
	if len(resp.NextPageToken) > 0 {
	req.PageToken(resp.NextPageToken)
	} else {
	req = nil
	}
	}
	}()

	// Kick off goroutines to delete the file paths.
	var wgConsumer sync.WaitGroup
	for i := 0; i < DELETE_GOROUTINE_POOL_SIZE; i++ {
	wgConsumer.Add(1)
	go func(goroutineNum int) {
	defer wgConsumer.Done()
	for filePath := range chFilePaths {
	if err := gs.service.Objects.Delete(GCSBucketName, filePath).Do(); err != nil {
	sklog.Errorf("Goroutine#%d could not delete %s: %s", goroutineNum, filePath, err)
	return
	}
	// Sleep for a second after deleting file to avoid bombarding Cloud
	// storage.
	time.Sleep(time.Second)
	}
	}(i + 1)
	}

	wgPopulator.Wait()
	wgConsumer.Wait()
	// Check if there was an error listing the GCS dir.
	select {
	case err, ok := <-errPopulator:
	if ok {
	return err
	}
	default:
	}
	return nil
	}

	// UploadFile calls UploadFileToBucket with CT's default bucket.
	func (gs *GcsUtil) UploadFile(fileName, localDir, gsDir string) error {
	return gs.UploadFileToBucket(fileName, localDir, gsDir, GCSBucketName)
	}

	// UploadFileToBucket uploads the specified file to the remote dir of the bucket
	// in Google Storage. It also sets the appropriate ACLs on the uploaded file.
	func (gs *GcsUtil) UploadFileToBucket(fileName, localDir, gsDir, bucket string) error {
	localFile := filepath.Join(localDir, fileName)
	gsFile := path.Join(gsDir, fileName)
	object := &storage.Object{
	Name: gsFile,
	// All objects uploaded to CT's bucket via this util must be readable by
	// the google.com domain. This will be fine tuned later if required.
	Acl: []*storage.ObjectAccessControl{
	{
	Bucket: bucket,
	Entity: "domain-google.com",
	Object: gsFile,
	Role: "READER",
	},
	},
	}
	f, err := os.Open(localFile)
	if err != nil {
	return fmt.Errorf("Error opening %s: %s", localFile, err)
	}
	defer util.Close(f)
	// TODO(rmistry): gs api now enables resumable uploads by default. Handle 308
	// response codes.
	fi, err := f.Stat()
	if err != nil {
	return fmt.Errorf("Error stating %s: %s", localFile, err)
	}
	mediaOption := googleapi.ChunkSize(int(fi.Size()))
	if _, err := gs.service.Objects.Insert(bucket, object).Media(f, mediaOption).Do(); err != nil {
	return fmt.Errorf("Objects.Insert failed: %s", err)
	}
	sklog.Infof("Copied %s to %s", localFile, fmt.Sprintf("gs://%s/%s", bucket, gsFile))
	return nil
	}

	// UploadSwarmingArtifact uploads the specified local artifacts to Google Storage.
	func (gs *GcsUtil) UploadSwarmingArtifacts(dirName, pagesetType string) error {
	localDir := path.Join(StorageDir, dirName, pagesetType)
	gsDir := path.Join(SWARMING_DIR_NAME, dirName, pagesetType)

	return gs.UploadDir(localDir, gsDir, false)
	}

	// DownloadSwarmingArtifacts downloads the specified artifacts from Google Storage to a local dir.
	// The Google storage directory is assumed to have numerical subdirs Eg: {1..1000}. This function
	// downloads the contents of those directories into a local directory without the numerical
	// subdirs.
	// Returns the ranking/index of the downloaded artifact.
	func (gs *GcsUtil) DownloadSwarmingArtifacts(localDir, remoteDirName, pagesetType string, startRange, num int) (map[string]int, error) {
	// Empty the local dir.
	util.RemoveAll(localDir)
	// Create the local dir.
	MkdirAll(localDir, 0700)

	gsDir := path.Join(SWARMING_DIR_NAME, remoteDirName, pagesetType)
	endRange := num + startRange - 1
	// The channel where remote files to be downloaded will be sent to.
	chRemoteDirs := make(chan string, num)
	for i := startRange; i <= endRange; i++ {
	chRemoteDirs <- path.Join(gsDir, strconv.Itoa(i))
	}
	close(chRemoteDirs)

	// Dictionary of artifacts to its rank/index.
	artifactToIndex := map[string]int{}
	// Mutex to control access to the above dictionary.
	var mtx sync.Mutex
	// Kick off goroutines to download artifacts and populate the artifactToIndex dictionary.
	var wg sync.WaitGroup
	for i := 0; i < DOWNLOAD_UPLOAD_GOROUTINE_POOL_SIZE; i++ {
	wg.Add(1)
	go func(goroutineNum int) {
	defer wg.Done()
	for remoteDir := range chRemoteDirs {
	if err := gs.downloadFromSwarmingDir(remoteDir, gsDir, localDir, goroutineNum, &mtx, artifactToIndex); err != nil {
	sklog.Error(err)
	return
	}
	}
	}(i + 1)
	}
	wg.Wait()
	if len(chRemoteDirs) != 0 {
	return artifactToIndex, fmt.Errorf("Unable to download all artifacts.")
	}
	return artifactToIndex, nil
	}

	// GetRemoteDirCount returns the number of objects in the specified dir.
	func (gs *GcsUtil) GetRemoteDirCount(gsDir string) (int, error) {
	req := gs.service.Objects.List(GCSBucketName).Prefix(gsDir + "/")
	count := 0
	for req != nil {
	resp, err := req.Do()
	if err != nil {
	return -1, fmt.Errorf("Error occured while listing %s: %s", gsDir, err)
	}
	count += len(resp.Items)
	if len(resp.NextPageToken) > 0 {
	req.PageToken(resp.NextPageToken)
	} else {
	req = nil
	}
	}
	return count, nil
	}

	func (gs GcsUtil) downloadFromSwarmingDir(remoteDir, gsDir, localDir string, runID int, mtx sync.Mutex, artifactToIndex map[string]int) error {
	req := gs.service.Objects.List(GCSBucketName).Prefix(remoteDir + "/")
	for req != nil {
	resp, err := req.Do()
	if err != nil {
	return fmt.Errorf("Error occured while listing %s: %s", gsDir, err)
	}
	for _, result := range resp.Items {
	fileName := path.Base(result.Name)
	fileGsDir := path.Dir(result.Name)
	index, err := strconv.Atoi(path.Base(fileGsDir))
	if err != nil {
	return fmt.Errorf("%s was not in expected format: %s", fileGsDir, err)
	}
	respBody, err := getRespBody(result, gs.client)
	if err != nil {
	return fmt.Errorf("Could not fetch %s: %s", result.MediaLink, err)
	}
	defer util.Close(respBody)
	outputFile := filepath.Join(localDir, fileName)
	out, err := os.Create(outputFile)
	if err != nil {
	return fmt.Errorf("Unable to create file %s: %s", outputFile, err)
	}
	defer util.Close(out)
	if _, err = io.Copy(out, respBody); err != nil {
	return err
	}
	// Sleep for a second after downloading file to avoid bombarding Cloud
	// storage.
	time.Sleep(time.Second)
	mtx.Lock()
	artifactToIndex[path.Join(localDir, fileName)] = index
	mtx.Unlock()
	}
	if len(resp.NextPageToken) > 0 {
	req.PageToken(resp.NextPageToken)
	} else {
	req = nil
	}
	}
	return nil
	}

	// UploadDir uploads the specified local dir into the specified Google Storage dir.
	func (gs *GcsUtil) UploadDir(localDir, gsDir string, cleanDir bool) error {
	if cleanDir {
	// Empty the remote dir before uploading to it.
	util.LogErr(gs.DeleteRemoteDir(gsDir))
	}

	// Construct a dictionary of file paths to their file infos.
	pathsToFileInfos := map[string]os.FileInfo{}
	visit := func(path string, f os.FileInfo, err error) error {
	if f.IsDir() {
	return nil
	}
	pathsToFileInfos[path] = f
	return nil
	}
	if err := filepath.Walk(localDir, visit); err != nil {
	return fmt.Errorf("Unable to read the local dir %s: %s", localDir, err)
	}

	// The channel where the filepaths to be uploaded will be sent to.
	chFilePaths := make(chan string, len(pathsToFileInfos))
	// File filepaths and send it to the above channel.
	for path, fileInfo := range pathsToFileInfos {
	fileName := fileInfo.Name()
	containingDir := strings.TrimSuffix(path, fileName)
	subDirs := strings.TrimPrefix(containingDir, localDir)
	if subDirs != "" {
	dirTokens := strings.Split(subDirs, "/")
	for i := range dirTokens {
	fileName = filepath.Join(dirTokens[len(dirTokens)-i-1], fileName)
	}
	}
	chFilePaths <- fileName
	}
	close(chFilePaths)

	// Kick off goroutines to upload the file paths.
	var wg sync.WaitGroup
	for i := 0; i < DOWNLOAD_UPLOAD_GOROUTINE_POOL_SIZE; i++ {
	wg.Add(1)
	go func(goroutineNum int) {
	defer wg.Done()
	for filePath := range chFilePaths {
	if err := gs.UploadFile(filePath, localDir, gsDir); err != nil {
	sklog.Errorf("Goroutine#%d could not upload %s to %s: %s", goroutineNum, filePath, localDir, err)
	}
	// Sleep for a second after uploading file to avoid bombarding Cloud
	// storage.
	time.Sleep(time.Second)
	}
	}(i + 1)
	}
	wg.Wait()
	return nil
	}

	// DownloadRemoteFile calls DownloadRemoteFileFromBucket with CT's default bucket.
	func (gs *GcsUtil) DownloadRemoteFile(remotePath, localPath string) error {
	return gs.DownloadRemoteFileFromBucket(GCSBucketName, remotePath, localPath)
	}

	// DownloadRemoteFileFromBucket downloads the specified remote path into the specified
	// local file. This function has been tested to download very large files (~33GB).
	// TODO(rmistry): Update all code that downloads remote files to use this or the
	// DownloadRemoteFile method.
	func (gs *GcsUtil) DownloadRemoteFileFromBucket(bucket, remotePath, localPath string) error {
	respBody, err := gs.GetRemoteFileContentsFromBucket(bucket, remotePath)
	if err != nil {
	return err
	}
	defer util.Close(respBody)
	out, err := os.Create(localPath)
	if err != nil {
	return err
	}

	bufferSize := int64(1024 * 1024 * 1024)
	for {
	_, err := io.CopyN(out, respBody, bufferSize)
	if err == io.EOF {
	break
	} else if err != nil {
	defer util.Close(out)
	return err
	}
	// Sleep for 30 seconds. Bots run out of memory without this.
	// Eg: https://chromium-swarm.appspot.com/task?id=2fba9fba3d553510
	// Maybe this sleep gives Golang time to clear some caches.
	time.Sleep(30 * time.Second)
	}
	if err := out.Close(); err != nil {
	return err
	}
	return nil
	}