cipd: Add -tag option to 'create' and 'pkg-register' subcomands.

go/src/infra/tools/cipd/uploader.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 (
         "errors"
         "fmt"
         "io"
         "net/http"
         "os"
         "time"
 
         "infra/libs/logging"
 )
 
 var (
         // ErrFinalizationTimeout is returned if CAS service can not finalize upload fast enough.
         ErrFinalizationTimeout = errors.New("Timeout while waiting for CAS service to finalize the upload")
+        // ErrAttachTagsTimeout is returned when service refuses to accept tags for a long time.
+        ErrAttachTagsTimeout = errors.New("Timeout while attaching tags")
 )
 
 // UploadOptions contains upload related parameters shared by UploadToCAS and
 // RegisterInstance functions.
 type UploadOptions struct {
         // ServiceURL is root URL of the backend service, or "" to use default service.
         ServiceURL string
         // FinalizationTimeout is how long to wait for CAS service to finalize the upload, default is 1 min.
         FinalizationTimeout time.Duration
         // Client is http.Client to use to make requests, default is http.DefaultClient.
@@ skipping 97 matching lines @@
                 if delay < 4*time.Second {
                         delay += 500 * time.Millisecond
                 }
         }
 }
 
 // RegisterInstanceOptions contains parameters for RegisterInstance function.
 type RegisterInstanceOptions struct {
         UploadOptions
 
-        // PackageInstance is a package to upload.
+        // PackageInstance is a package instance to register.
         PackageInstance PackageInstance
+        // Tags is a list of tags to attach to an instance. Will be attached even if
+        // the instance already existed before.
+        Tags []string
 }
 
 // RegisterInstance makes the package instance available for clients by
 // uploading it to the storage and registering it in the package repository.
 func RegisterInstance(options RegisterInstanceOptions) error {
         // Fill in default options.
         if options.ServiceURL == "" {
                 options.ServiceURL = DefaultServiceURL()
         }
         if options.Client == nil {
@@ skipping 37 matching lines @@
 
         if result.AlreadyRegistered {
                 log.Infof(
                         "cipd: instance %s:%s is already registered by %s on %s",
                         inst.PackageName(), inst.InstanceID(),
                         result.Info.RegisteredBy, result.Info.RegisteredTs)
         } else {
                 log.Infof("cipd: instance %s:%s was successfully registered", inst.PackageName(), inst.InstanceID())
         }
 
-        return nil
+        return attachTagsWhenReady(remote, inst.PackageName(), inst.InstanceID(), options.Tags, log)
 }
 
 ////////////////////////////////////////////////////////////////////////////////
 // Google Storage resumable upload protocol.
 // See https://cloud.google.com/storage/docs/concepts-techniques#resumable
 
 // errTransientError is returned by getNextOffset in case of retryable error.
 var errTransientError = errors.New("Transient error in getUploadedOffset")
 
 // resumableUpload is mocked in tests.
@@ skipping 109 matching lines @@
                                 offset++
                         }
                 }
         } else if resp.StatusCode < 500 && resp.StatusCode != 408 {
                 err = fmt.Errorf("Unexpected response (HTTP %d) when querying for uploaded offset", resp.StatusCode)
         } else {
                 err = errTransientError
         }
         return
 }
+
+////////////////////////////////////////////////////////////////////////////////
+// Tags related functions.
+
+// attachTagsWhenReady attaches tags to an instance retrying when receiving
+// PROCESSING_NOT_FINISHED_YET errors.
+func attachTagsWhenReady(remote *remoteService, packageName, instanceID string, tags []string, log logging.Logger) error {
+        if len(tags) == 0 {
+                return nil
+        }
+        for _, tag := range tags {
+                log.Infof("cipd: attaching tag %s", tag)
+        }
+        deadline := clock.Now().Add(60 * time.Second)
+        for clock.Now().Before(deadline) {
+                err := remote.attachTags(packageName, instanceID, tags)
+                if err == nil {
+                        log.Infof("cipd: all tags attached")
+                        return nil
+                }
+                if _, ok := err.(*pendingProcessingError); ok {
+                        log.Warningf("cipd: package instance is not ready yet - %s", err)
+                        clock.Sleep(5 * time.Second)
+                } else {
+                        log.Errorf("cipd: failed to attach tags - %s", err)
+                        return err
+                }
+        }
+        log.Errorf("cipd: failed to attach tags - deadline exceeded")
+        return ErrAttachTagsTimeout
+}