cipd: Refactor client to make it more readable.

go/src/infra/tools/cipd/remote.go

 // Copyright 2014 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 package cipd
 
 import (
         "bytes"
         "encoding/json"
         "errors"
         "fmt"
         "io"
         "io/ioutil"
         "net/http"
         "net/url"
         "strconv"
         "time"
 
-        "infra/libs/logging"
+        "infra/tools/cipd/common"
 )
 
-// remoteService is a wrapper around Cloud Endpoints APIs exposed by backend.
-// See //appengine/chrome_infra_packages.
-type remoteService struct {
-        client     *http.Client
-        serviceURL string
-        log        logging.Logger
-}
+// remoteMaxRetries is how many times to retry transient HTTP errors.
+const remoteMaxRetries = 10
 
-type uploadSession struct {
-        ID  string
-        URL string
-}
-
-// packageInstanceMsg corresponds to PackageInstance message on the backend.
 type packageInstanceMsg struct {
         PackageName  string `json:"package_name"`
         InstanceID   string `json:"instance_id"`
         RegisteredBy string `json:"registered_by"`
         RegisteredTs string `json:"registered_ts"`
 }
 
-// packageInstance is almost like packageInstanceMsg, but with timestamp as
-// time.Time. See also makePackageInstanceInfo.
-type packageInstanceInfo struct {
-        PackageName  string
-        InstanceID   string
-        RegisteredBy string
-        RegisteredTs time.Time
-}
-
-type registerInstanceResponse struct {
-        UploadSession     *uploadSession
-        AlreadyRegistered bool
-        Info              packageInstanceInfo
-}
-
-type fetchInstanceResponse struct {
-        FetchURL string
-        Info     packageInstanceInfo
-}
-
 // roleChangeMsg corresponds to RoleChange proto message on backend.
 type roleChangeMsg struct {
         Action    string `json:"action"`
         Role      string `json:"role"`
         Principal string `json:"principal"`
 }
 
 // pendingProcessingError is returned by attachTags if package instance is not
 // yet ready and the call should be retried later.
 type pendingProcessingError struct {
         message string
 }
 
 func (e *pendingProcessingError) Error() string {
         return e.message
 }
 
-// newRemoteService is mocked in tests.
-var newRemoteService = func(client *http.Client, url string, log logging.Logger) *remoteService {
-        log.Infof("cipd: service URL is %s", url)
-        return &remoteService{
-                client:     client,
-                serviceURL: url,
-                log:        log,
-        }
+// remoteImpl implements remote on top of real HTTP calls.
+type remoteImpl struct {
+        client *Client
 }
 
-// makeRequest sends POST or GET request with retries.
-func (r *remoteService) makeRequest(path, method string, request, response interface{}) error {
+func isTemporaryNetError(err error) bool {
+        // TODO(vadimsh): Figure out how to recognize dial timeouts, read timeouts,
+        // etc. For now all errors that end up here are considered temporary.
+        return true
+}
+
+// isTemporaryHTTPError returns true for HTTP status codes that indicate
+// a temporary error that may go away if request is retried.
+func isTemporaryHTTPError(statusCode int) bool {
+        return statusCode >= 500 || statusCode == 408 || statusCode == 429
+}
+
+// makeRequest sends POST or GET REST JSON requests with retries.
+func (r *remoteImpl) makeRequest(path, method string, request, response interface{}) error {
         var body []byte
-        var err error
         if request != nil {
-                body, err = json.Marshal(request)
+                b, err := json.Marshal(request)
                 if err != nil {
                         return err
                 }
+                body = b
         }
-        url := fmt.Sprintf("%s/_ah/api/%s", r.serviceURL, path)
-        for attempt := 0; attempt < 10; attempt++ {
+
+        url := fmt.Sprintf("%s/_ah/api/%s", r.client.ServiceURL, path)
+        for attempt := 0; attempt < remoteMaxRetries; attempt++ {
                 if attempt != 0 {
-                        r.log.Warningf("cipd: retrying request to %s", url)
-                        clock.Sleep(2 * time.Second)
+                        r.client.Log.Warningf("cipd: retrying request to %s", url)
+                        r.client.clock.sleep(2 * time.Second)
                 }
+
+                // Prepare request.
                 var bodyReader io.Reader
                 if body != nil {
                         bodyReader = bytes.NewReader(body)
                 }
                 req, err := http.NewRequest(method, url, bodyReader)
                 if err != nil {
                         return err
                 }
                 if body != nil {
                         req.Header.Set("Content-Type", "application/json")
                 }
-                req.Header.Set("User-Agent", userAgent())
-                resp, err := r.client.Do(req)
+                req.Header.Set("User-Agent", r.client.UserAgent)
+
+                // Connect, read response.
+                r.client.Log.Debugf("cipd: %s %s", method, url)
+                resp, err := r.client.doAuthenticatedHTTPRequest(req)
                 if err != nil {
+                        if isTemporaryNetError(err) {
+                                r.client.Log.Warningf("cipd: connectivity error (%s)", err)
+                                continue
+                        }
                         return err
                 }
+                responseBody, err := ioutil.ReadAll(resp.Body)
+                resp.Body.Close()
+                if err != nil {
+                        if isTemporaryNetError(err) {
+                                r.client.Log.Warningf("cipd: temporary error when reading response (%s)", err)
+                                continue
+                        }
+                        return err
+                }
+                r.client.Log.Debugf("cipd: http %d: %s", resp.StatusCode, body)
+                if isTemporaryHTTPError(resp.StatusCode) {
+                        continue
+                }
+
                 // Success?
                 if resp.StatusCode < 300 {
-                        defer resp.Body.Close()
-                        return json.NewDecoder(resp.Body).Decode(response)
+                        return json.Unmarshal(responseBody, response)
                 }
+
                 // Fatal error?
-                if resp.StatusCode >= 300 && resp.StatusCode < 500 {
-                        defer resp.Body.Close()
-                        body, _ := ioutil.ReadAll(resp.Body)
-                        return fmt.Errorf("Unexpected reply (HTTP %d):\n%s", resp.StatusCode, string(body))
+                if resp.StatusCode == 403 || resp.StatusCode == 401 {
+                        return ErrAccessDenined
                 }
-                // Retry.
-                resp.Body.Close()
+                return fmt.Errorf("Unexpected reply (HTTP %d):\n%s", resp.StatusCode, string(body))
         }
-        return fmt.Errorf("Request to %s failed after 10 attempts", url)
+
+        return ErrBackendInaccessible
 }
 
-func (r *remoteService) initiateUpload(sha1 string) (s *uploadSession, err error) {
+func (r *remoteImpl) initiateUpload(sha1 string) (s *UploadSession, err error) {
         var reply struct {
                 Status          string `json:"status"`
                 UploadSessionID string `json:"upload_session_id"`
                 UploadURL       string `json:"upload_url"`
                 ErrorMessage    string `json:"error_message"`
         }
         err = r.makeRequest("cas/v1/upload/SHA1/"+sha1, "POST", nil, &reply)
         if err != nil {
                 return
         }
         switch reply.Status {
         case "ALREADY_UPLOADED":
                 return
         case "SUCCESS":
-                s = &uploadSession{
-                        ID:  reply.UploadSessionID,
-                        URL: reply.UploadURL,
-                }
+                s = &UploadSession{reply.UploadSessionID, reply.UploadURL}
         case "ERROR":
                 err = fmt.Errorf("Server replied with error: %s", reply.ErrorMessage)
         default:
                 err = fmt.Errorf("Unexpected status: %s", reply.Status)
         }
         return
 }
 
-func (r *remoteService) finalizeUpload(sessionID string) (finished bool, err error) {
+func (r *remoteImpl) finalizeUpload(sessionID string) (finished bool, err error) {
         var reply struct {
                 Status       string `json:"status"`
                 ErrorMessage string `json:"error_message"`
         }
         err = r.makeRequest("cas/v1/finalize/"+sessionID, "POST", nil, &reply)
         if err != nil {
                 return
         }
         switch reply.Status {
         case "MISSING":
-                err = errors.New("Upload session is unexpectedly missing")
+                err = ErrUploadSessionDied
         case "UPLOADING", "VERIFYING":
                 finished = false
         case "PUBLISHED":
                 finished = true
         case "ERROR":
                 err = errors.New(reply.ErrorMessage)
         default:
                 err = fmt.Errorf("Unexpected upload session status: %s", reply.Status)
         }
         return
 }
 
-func (r *remoteService) registerInstance(packageName, instanceID string) (*registerInstanceResponse, error) {
-        endpoint, err := instanceEndpoint(packageName, instanceID)
+func (r *remoteImpl) registerInstance(pin common.Pin) (*registerInstanceResponse, error) {
+        endpoint, err := instanceEndpoint(pin)
         if err != nil {
                 return nil, err
         }
         var reply struct {
                 Status          string             `json:"status"`
                 Instance        packageInstanceMsg `json:"instance"`
                 UploadSessionID string             `json:"upload_session_id"`
                 UploadURL       string             `json:"upload_url"`
                 ErrorMessage    string             `json:"error_message"`
         }
         err = r.makeRequest(endpoint, "POST", nil, &reply)
         if err != nil {
                 return nil, err
         }
         switch reply.Status {
         case "REGISTERED", "ALREADY_REGISTERED":
-                info, err := makePackageInstanceInfo(reply.Instance)
+                ts, err := convertTimestamp(reply.Instance.RegisteredTs)
                 if err != nil {
                         return nil, err
                 }
                 return &registerInstanceResponse{
-                        AlreadyRegistered: reply.Status == "ALREADY_REGISTERED",
-                        Info:              info,
+                        alreadyRegistered: reply.Status == "ALREADY_REGISTERED",
+                        registeredBy:      reply.Instance.RegisteredBy,
+                        registeredTs:      ts,
                 }, nil
         case "UPLOAD_FIRST":
                 if reply.UploadSessionID == "" {
-                        return nil, errors.New("Server didn't provide upload session ID")
+                        return nil, ErrNoUploadSessionID
                 }
                 return &registerInstanceResponse{
-                        UploadSession: &uploadSession{
-                                ID:  reply.UploadSessionID,
-                                URL: reply.UploadURL,
-                        },
+                        uploadSession: &UploadSession{reply.UploadSessionID, reply.UploadURL},
                 }, nil
         case "ERROR":
                 return nil, errors.New(reply.ErrorMessage)
         }
         return nil, fmt.Errorf("Unexpected register package status: %s", reply.Status)
 }
 
-func (r *remoteService) fetchInstance(packageName, instanceID string) (*fetchInstanceResponse, error) {
-        endpoint, err := instanceEndpoint(packageName, instanceID)
+func (r *remoteImpl) fetchInstance(pin common.Pin) (*fetchInstanceResponse, error) {
+        endpoint, err := instanceEndpoint(pin)
         if err != nil {
                 return nil, err
         }
         var reply struct {
                 Status       string             `json:"status"`
                 Instance     packageInstanceMsg `json:"instance"`
                 FetchURL     string             `json:"fetch_url"`
                 ErrorMessage string             `json:"error_message"`
         }
         err = r.makeRequest(endpoint, "GET", nil, &reply)
         if err != nil {
                 return nil, err
         }
         switch reply.Status {
         case "SUCCESS":
-                info, err := makePackageInstanceInfo(reply.Instance)
+                ts, err := convertTimestamp(reply.Instance.RegisteredTs)
                 if err != nil {
                         return nil, err
                 }
                 return &fetchInstanceResponse{
-                        FetchURL: reply.FetchURL,
-                        Info:     info,
+                        fetchURL:     reply.FetchURL,
+                        registeredBy: reply.Instance.RegisteredBy,
+                        registeredTs: ts,
                 }, nil
         case "PACKAGE_NOT_FOUND":
-                return nil, fmt.Errorf("Package '%s' is not registered or you do not have permission to fetch it", packageName)
+                return nil, fmt.Errorf("Package '%s' is not registered or you do not have permission to fetch it", pin.PackageName)
         case "INSTANCE_NOT_FOUND":
-                return nil, fmt.Errorf("Package '%s' doesn't have instance '%s'", packageName, instanceID)
+                return nil, fmt.Errorf("Package '%s' doesn't have instance '%s'", pin.PackageName, pin.InstanceID)
         case "ERROR":
                 return nil, errors.New(reply.ErrorMessage)
         }
         return nil, fmt.Errorf("Unexpected reply status: %s", reply.Status)
 }
 
-func (r *remoteService) fetchACL(packagePath string) ([]PackageACL, error) {
+func (r *remoteImpl) fetchACL(packagePath string) ([]PackageACL, error) {
         endpoint, err := aclEndpoint(packagePath)
         if err != nil {
                 return nil, err
         }
         var reply struct {
                 Status       string `json:"status"`
                 ErrorMessage string `json:"error_message"`
                 Acls         struct {
                         Acls []struct {
                                 PackagePath string   `json:"package_path"`
@@ skipping 24 matching lines @@
                                 ModifiedTs:  ts,
                         })
                 }
                 return out, nil
         case "ERROR":
                 return nil, errors.New(reply.ErrorMessage)
         }
         return nil, fmt.Errorf("Unexpected reply status: %s", reply.Status)
 }
 
-func (r *remoteService) modifyACL(packagePath string, changes []PackageACLChange) error {
+func (r *remoteImpl) modifyACL(packagePath string, changes []PackageACLChange) error {
         endpoint, err := aclEndpoint(packagePath)
         if err != nil {
                 return err
         }
         var request struct {
                 Changes []roleChangeMsg `json:"changes"`
         }
         for _, c := range changes {
                 action := ""
                 if c.Action == GrantRole {
@@ skipping 19 matching lines @@
         }
         switch reply.Status {
         case "SUCCESS":
                 return nil
         case "ERROR":
                 return errors.New(reply.ErrorMessage)
         }
         return fmt.Errorf("Unexpected reply status: %s", reply.Status)
 }
 
-func (r *remoteService) attachTags(packageName, instanceID string, tags []string) error {
+func (r *remoteImpl) attachTags(pin common.Pin, tags []string) error {
         // Tags will be passed in the request body, not via URL.
-        endpoint, err := tagsEndpoint(packageName, instanceID, nil)
+        endpoint, err := tagsEndpoint(pin, nil)
         if err != nil {
                 return err
         }
-
-        if len(tags) == 0 {
-                return errors.New("At least one tag must be provided")
-        }
         for _, tag := range tags {
-                err = ValidateInstanceTag(tag)
+                err = common.ValidateInstanceTag(tag)
                 if err != nil {
                         return err
                 }
         }
+
         var request struct {
                 Tags []string `json:"tags"`
         }
         request.Tags = tags
 
         var reply struct {
                 Status       string `json:"status"`
                 ErrorMessage string `json:"error_message"`
         }
         err = r.makeRequest(endpoint, "POST", &request, &reply)
         if err != nil {
                 return err
         }
         switch reply.Status {
         case "SUCCESS":
                 return nil
         case "PROCESSING_NOT_FINISHED_YET":
                 return &pendingProcessingError{reply.ErrorMessage}
         case "ERROR", "PROCESSING_FAILED":
                 return errors.New(reply.ErrorMessage)
         }
         return fmt.Errorf("Unexpected status when attaching tags: %s", reply.Status)
 }
 
 ////////////////////////////////////////////////////////////////////////////////
 
-func instanceEndpoint(packageName, instanceID string) (string, error) {
-        err := ValidatePackageName(packageName)
-        if err != nil {
-                return "", err
-        }
-        err = ValidateInstanceID(instanceID)
+func instanceEndpoint(pin common.Pin) (string, error) {
+        err := common.ValidatePin(pin)
         if err != nil {
                 return "", err
         }
         params := url.Values{}
-        params.Add("package_name", packageName)
-        params.Add("instance_id", instanceID)
+        params.Add("package_name", pin.PackageName)
+        params.Add("instance_id", pin.InstanceID)
         return "repo/v1/instance?" + params.Encode(), nil
 }
 
 func aclEndpoint(packagePath string) (string, error) {
-        err := ValidatePackageName(packagePath)
+        err := common.ValidatePackageName(packagePath)
         if err != nil {
                 return "", err
         }
         params := url.Values{}
         params.Add("package_path", packagePath)
         return "repo/v1/acl?" + params.Encode(), nil
 }
 
-func tagsEndpoint(packageName, instanceID string, tags []string) (string, error) {
-        err := ValidatePackageName(packageName)
-        if err != nil {
-                return "", err
-        }
-        err = ValidateInstanceID(instanceID)
+func tagsEndpoint(pin common.Pin, tags []string) (string, error) {
+        err := common.ValidatePin(pin)
         if err != nil {
                 return "", err
         }
         for _, tag := range tags {
-                err = ValidateInstanceTag(tag)
+                err = common.ValidateInstanceTag(tag)
                 if err != nil {
                         return "", err
                 }
         }
         params := url.Values{}
-        params.Add("package_name", packageName)
-        params.Add("instance_id", instanceID)
+        params.Add("package_name", pin.PackageName)
+        params.Add("instance_id", pin.InstanceID)
         for _, tag := range tags {
                 params.Add("tag", tag)
         }
         return "repo/v1/tags?" + params.Encode(), nil
 }
 
 // convertTimestamp coverts string with int64 timestamp in microseconds since
 // to time.Time
 func convertTimestamp(ts string) (time.Time, error) {
         i, err := strconv.ParseInt(ts, 10, 64)
         if err != nil {
                 return time.Time{}, fmt.Errorf("Unexpected timestamp value '%s' in the server response", ts)
         }
         return time.Unix(0, i*1000), nil
 }
-
-func makePackageInstanceInfo(msg packageInstanceMsg) (pi packageInstanceInfo, err error) {
-        ts, err := convertTimestamp(msg.RegisteredTs)
-        if err != nil {
-                return
-        }
-        pi = packageInstanceInfo{
-                PackageName:  msg.PackageName,
-                InstanceID:   msg.InstanceID,
-                RegisteredBy: msg.RegisteredBy,
-                RegisteredTs: ts,
-        }
-        return
-}