cipd: Implement local cache for ResolveVersion(...) call with tags.

go/src/infra/tools/cipd/client.go

 // Copyright 2015 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 implements client side of Chrome Infra Package Deployer.
 
 TODO: write more.
 
 Binary package file format (in free form representation):
@@ skipping 23 matching lines @@
 */
 package cipd
 
 import (
         "bufio"
         "errors"
         "fmt"
         "io"
         "net/http"
         "os"
+        "path/filepath"
         "sort"
         "strings"
+        "sync"
         "time"
 
         "github.com/luci/luci-go/common/logging"
 
         "infra/tools/cipd/common"
+        "infra/tools/cipd/internal"
         "infra/tools/cipd/local"
 )
 
 // PackageACLChangeAction defines a flavor of PackageACLChange.
 type PackageACLChangeAction string
 
 const (
         // GrantRole is used in PackageACLChange to request a role to be granted.
         GrantRole PackageACLChangeAction = "GRANT"
         // RevokeRole is used in PackageACLChange to request a role to be revoked.
         RevokeRole PackageACLChangeAction = "REVOKE"
 
         // CASFinalizationTimeout is how long to wait for CAS service to finalize the upload.
         CASFinalizationTimeout = 1 * time.Minute
         // SetRefTimeout is how long to wait for an instance to be processed when setting a ref.
         SetRefTimeout = 1 * time.Minute
         // TagAttachTimeout is how long to wait for an instance to be processed when attaching tags.
         TagAttachTimeout = 1 * time.Minute
 
         // UserAgent is HTTP user agent string for CIPD client.
         UserAgent = "cipd 1.0"
 
         // ServiceURL is URL of a backend to connect to by default.
         ServiceURL = "https://chrome-infra-packages.appspot.com"
 )
 
 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")
+        ErrFinalizationTimeout = errors.New("timeout while waiting for CAS service to finalize the upload")
         // ErrBadUpload is returned when a package file is uploaded, but servers asks us to upload it again.
-        ErrBadUpload = errors.New("Package file is uploaded, but servers asks us to upload it again")
+        ErrBadUpload = errors.New("package file is uploaded, but servers asks us to upload it again")
         // ErrBadUploadSession is returned by UploadToCAS if provided UploadSession is not valid.
-        ErrBadUploadSession = errors.New("UploadURL must be set if UploadSessionID is used")
+        ErrBadUploadSession = errors.New("uploadURL must be set if UploadSessionID is used")
         // ErrUploadSessionDied is returned by UploadToCAS if upload session suddenly disappeared.
-        ErrUploadSessionDied = errors.New("Upload session is unexpectedly missing")
+        ErrUploadSessionDied = errors.New("upload session is unexpectedly missing")
         // ErrNoUploadSessionID is returned by UploadToCAS if server didn't provide upload session ID.
-        ErrNoUploadSessionID = errors.New("Server didn't provide upload session ID")
+        ErrNoUploadSessionID = errors.New("server didn't provide upload session ID")
         // ErrSetRefTimeout is returned when service refuses to move a ref for a long time.
-        ErrSetRefTimeout = errors.New("Timeout while moving a ref")
+        ErrSetRefTimeout = errors.New("timeout while moving a ref")
         // ErrAttachTagsTimeout is returned when service refuses to accept tags for a long time.
-        ErrAttachTagsTimeout = errors.New("Timeout while attaching tags")
+        ErrAttachTagsTimeout = errors.New("timeout while attaching tags")
         // ErrDownloadError is returned by FetchInstance on download errors.
-        ErrDownloadError = errors.New("Failed to download the package file after multiple attempts")
+        ErrDownloadError = errors.New("failed to download the package file after multiple attempts")
         // ErrUploadError is returned by RegisterInstance and UploadToCAS on upload errors.
-        ErrUploadError = errors.New("Failed to upload the package file after multiple attempts")
+        ErrUploadError = errors.New("failed to upload the package file after multiple attempts")
         // ErrAccessDenined is returned by calls talking to backend on 401 or 403 HTTP errors.
-        ErrAccessDenined = errors.New("Access denied (not authenticated or not enough permissions)")
+        ErrAccessDenined = errors.New("access denied (not authenticated or not enough permissions)")
         // ErrBackendInaccessible is returned by calls talking to backed if it doesn't response.
-        ErrBackendInaccessible = errors.New("Request to the backend failed after multiple attempts")
+        ErrBackendInaccessible = errors.New("request to the backend failed after multiple attempts")
         // ErrEnsurePackagesFailed is returned by EnsurePackages if something is not right.
-        ErrEnsurePackagesFailed = errors.New("Failed to update packages, see the log")
+        ErrEnsurePackagesFailed = errors.New("failed to update packages, see the log")
 )
 
 // PackageACL is per package path per role access control list that is a part of
 // larger overall ACL: ACL for package "a/b/c" is a union of PackageACLs for "a"
 // "a/b" and "a/b/c".
 type PackageACL struct {
         // PackagePath is a package subpath this ACL is defined for.
         PackagePath string
         // Role is a role that listed users have, e.g. 'READER', 'WRITER', ...
         Role string
@@ skipping 16 matching lines @@
 }
 
 // UploadSession describes open CAS upload session.
 type UploadSession struct {
         // ID identifies upload session in the backend.
         ID string
         // URL is where to upload the data to.
         URL string
 }
 
-// Client provides high-level CIPD client interface.
+// Client provides high-level CIPD client interface. Thread safe.
 type Client interface {
         // FetchACL returns a list of PackageACL objects (parent paths first) that
         // together define the access control list for the given package subpath.
         FetchACL(packagePath string) ([]PackageACL, error)
 
         // ModifyACL applies a set of PackageACLChanges to a package path.
         ModifyACL(packagePath string, changes []PackageACLChange) error
 
         // UploadToCAS uploads package data blob to Content Addressed Store if it is
         // not there already. The data is addressed by SHA1 hash (also known as
@@ skipping 39 matching lines @@
         // <package name> <desired version>. Whitespaces are ignored. Lines that start
         // with '#' are ignored. Version can be specified as instance ID, tag or ref.
         // Will resolve tags and refs to concrete instance IDs by calling the backend.
         ProcessEnsureFile(r io.Reader) ([]common.Pin, error)
 
         // EnsurePackages is high-level interface for installation, removal and update
         // of packages inside the installation site root. Given a description of
         // what packages (and versions) should be installed it will do all necessary
         // actions to bring the state of the site root to the desired one.
         EnsurePackages(pins []common.Pin) error
+
+        // Close should be called to dump any cached state to disk.
+        Close()
 }
 
 // HTTPClientFactory lazily creates http.Client to use for making requests.
 type HTTPClientFactory func() (*http.Client, error)
 
 // ClientOptions is passed to NewClient factory function.
 type ClientOptions struct {
         // ServiceURL is root URL of the backend service.
         ServiceURL string
-        // Root is a site root directory (where packages will be installed). It can
-        // be empty string if client is not going to be used to deploy or remove local packages.
+
+        // Root is a site root directory (a directory where packages will be
+        // installed to). It also hosts .cipd/* directory that tracks internal state
+        // of installed packages and keeps various cache files. 'Root' can be an empty
+        // string if the client is not going to be used to deploy or remove local
+        // packages. In that case caches are also disabled.
         Root string
+
         // Logger is a logger to use for logs (null-logger by default).
         Logger logging.Logger
-        // AuthenticatedClientFactory lazily creates http.Client to use for making RPC requests.
+
+        // AuthenticatedClientFactory lazily creates http.Client to use for making
+        // RPC requests.
         AuthenticatedClientFactory HTTPClientFactory
-        // AnonymousClientFactory lazily creates http.Client to use for making requests to storage.
+
+        // AnonymousClientFactory lazily creates http.Client to use for making
+        // requests to storage.
         AnonymousClientFactory HTTPClientFactory
+
         // UserAgent is put into User-Agent HTTP header with each request.
         UserAgent string
 }
 
 // NewClient initializes CIPD client object.
 func NewClient(opts ClientOptions) Client {
         if opts.ServiceURL == "" {
                 opts.ServiceURL = ServiceURL
         }
         if opts.Logger == nil {
@@ skipping 14 matching lines @@
         }
         c.remote = &remoteImpl{c}
         c.storage = &storageImpl{c, uploadChunkSize}
         c.deployer = local.NewDeployer(opts.Root, opts.Logger)
         return c
 }
 
 type clientImpl struct {
         ClientOptions
 
-        // clock provides current time and ability to sleep.
+        // lock protects lazily initialized portions of the client.
+        lock sync.Mutex
+
+        // clock provides current time and ability to sleep. Thread safe.
         clock clock
-        // remote knows how to call backend REST API.
+
+        // remote knows how to call backend REST API. Thread safe.
         remote remote
+
         // storage knows how to upload and download raw binaries using signed URLs.
+        // Thread safe.
         storage storage
-        // deployer knows how to install packages to local file system.
+
+        // deployer knows how to install packages to local file system. Thread safe.
         deployer local.Deployer
 
-        // authClient is a lazily created http.Client to use for authenticated requests.
+        // tagCache is used to cache (pkgname, tag) -> instanceID mapping.
+        // Thread safe, but lazily initialized under lock.
+        tagCache *internal.TagCache
+
+        // authClient is a lazily created http.Client to use for authenticated
+        // requests. Thread safe, but lazily initialized under lock.
         authClient *http.Client
+
         // anonClient is a lazily created http.Client to use for anonymous requests.
+        // Thread safe, but lazily initialized under lock.
         anonClient *http.Client
 }
 
 // doAuthenticatedHTTPRequest is used by remote implementation to make HTTP calls.
 func (client *clientImpl) doAuthenticatedHTTPRequest(req *http.Request) (*http.Response, error) {
-        if client.authClient == nil {
-                var err error
-                client.authClient, err = client.AuthenticatedClientFactory()
-                if err != nil {
-                        return nil, err
-                }
-        }
-        return client.authClient.Do(req)
+        return client.doRequest(req, &client.authClient, client.AuthenticatedClientFactory)
 }
 
 // doAnonymousHTTPRequest is used by storage implementation to make HTTP calls.
 func (client *clientImpl) doAnonymousHTTPRequest(req *http.Request) (*http.Response, error) {
-        if client.anonClient == nil {
+        return client.doRequest(req, &client.anonClient, client.AnonymousClientFactory)
+}
+
+// doRequest lazy-initializes http.Client using provided factory and then
+// executes the request.
+func (client *clientImpl) doRequest(req *http.Request, c **http.Client, fac HTTPClientFactory) (*http.Response, error) {
+        httpClient, err := func() (*http.Client, error) {
+                client.lock.Lock()
+                defer client.lock.Unlock()
                 var err error
-                client.anonClient, err = client.AnonymousClientFactory()
-                if err != nil {
-                        return nil, err
+                if *c == nil {
+                        *c, err = fac()
+                }
+                return *c, err
+        }()
+        if err != nil {
+                return nil, err
+        }
+        return httpClient.Do(req)
+}
+
+// tagCachePath returns path to a tag cache file or "" if no root dir.
+func (client *clientImpl) tagCachePath() string {
+        if client.Root == "" {
+                return ""
+        }
+        return filepath.Join(client.Root, local.SiteServiceDir, "tagcache.db")
+}
+
+// getTagCache lazy-initializes tagCache instance and returns it.
+func (client *clientImpl) getTagCache() *internal.TagCache {
+        client.lock.Lock()
+        defer client.lock.Unlock()
+        if client.tagCache == nil {
+                if path := client.tagCachePath(); path != "" {
+                        var err error
+                        client.tagCache, err = internal.LoadTagCacheFromFile(path)
+                        if err != nil {
+                                client.Logger.Warningf("cipd: failed to load tag cache - %s", err)
+                        }
+                }
+                if client.tagCache == nil {
+                        client.tagCache = &internal.TagCache{}
                 }
         }
-        return client.anonClient.Do(req)
+        return client.tagCache
+}
+
+// closeTagCache dumps any changes made to tag cache to disk, if necessary.
+// Must be called under lock.
+func (client *clientImpl) closeTagCache() {
+        path := client.tagCachePath()
+        if client.tagCache == nil || path == "" || !client.tagCache.Dirty() {
+                client.tagCache = nil
+                return
+        }
+        // It's tiny in size (and protobuf can't serialize to io.Reader anyway). Dump
+        // it to disk via FileSystem object to deal with possible concurrent updates,
+        // missing directories, etc.
+        fs := local.NewFileSystem(filepath.Dir(path), client.Logger)
+        out, err := client.tagCache.Save()
+        if err == nil {
+                err = fs.EnsureFile(path, out, 0666)
+        }
+        if err != nil {
+                client.Logger.Warningf("cipd: failed to update tag cache - %s", err)
+        }
+        client.tagCache = nil
 }
 
 func (client *clientImpl) FetchACL(packagePath string) ([]PackageACL, error) {
         return client.remote.fetchACL(packagePath)
 }
 
 func (client *clientImpl) ModifyACL(packagePath string, changes []PackageACLChange) error {
         return client.remote.modifyACL(packagePath, changes)
 }
 
@@ skipping 68 matching lines @@
         if err := common.ValidatePackageName(packageName); err != nil {
                 return common.Pin{}, err
         }
         // Is it instance ID already? Don't bother calling the backend.
         if common.ValidateInstanceID(version) == nil {
                 return common.Pin{PackageName: packageName, InstanceID: version}, nil
         }
         if err := common.ValidateInstanceVersion(version); err != nil {
                 return common.Pin{}, err
         }
-        return client.remote.resolveVersion(packageName, version)
+        // Use local cache when resolving tags to avoid round trips to backend when
+        // calling same 'cipd ensure' command again and again.
+        isTag := common.ValidateInstanceTag(version) == nil
+        if isTag {
+                cached := client.getTagCache().ResolveTag(packageName, version)
+                if cached.InstanceID != "" {
+                        return cached, nil
+                }
+        }
+        pin, err := client.remote.resolveVersion(packageName, version)
+        if err != nil {
+                return pin, err
+        }
+        if isTag {
+                client.getTagCache().AddTag(pin, version)
+        }
+        return pin, nil
 }
 
 func (client *clientImpl) RegisterInstance(instance local.PackageInstance) error {
         // Attempt to register.
         client.Logger.Infof("cipd: registering %s", instance.Pin())
         result, err := client.remote.registerInstance(instance.Pin())
         if err != nil {
                 return err
         }
 
@@ skipping 250 matching lines @@
                 }
         }
 
         if !fail {
                 client.Logger.Infof("All changes applied.")
                 return nil
         }
         return ErrEnsurePackagesFailed
 }
 
+func (client *clientImpl) Close() {
+        client.lock.Lock()
+        defer client.lock.Unlock()
+        client.closeTagCache()
+        client.authClient = nil
+        client.anonClient = nil
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 // Private structs and interfaces.
 
 type clock interface {
         now() time.Time
         sleep(time.Duration)
 }
 
 type remote interface {
         fetchACL(packagePath string) ([]PackageACL, error)
@@ skipping 60 matching lines @@
 }
 
 // buildInstanceIDMap builds mapping {package name -> instance ID}.
 func buildInstanceIDMap(pins []common.Pin) map[string]string {
         out := map[string]string{}
         for _, p := range pins {
                 out[p.PackageName] = p.InstanceID
         }
         return out
 }