cipd: instance cache

client/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.
 //
 // Binary package file format (in free form representation):
 //   <binary package> := <zipped data>
 //   <zipped data> := DeterministicZip(<all input files> + <manifest json>)
 //   <manifest json> := File{
@@ skipping 319 matching lines @@
         // RPC requests.
         AuthenticatedClientFactory HTTPClientFactory
 
         // 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
 
-        // CacheDir is a directory for shared cache. If empty, tags are cached
-        // inside the site root. If both Root and CacheDir are empty, tag cache
-        // is disabled.
+        // CacheDir is a directory for shared cache. If empty, instances are not
+        // cached and tags are cached inside the site root. If both Root and
+        // CacheDir are empty, tag cache is disabled.
         CacheDir string
 }
 
 // NewClient initializes CIPD client object.
 func NewClient(opts ClientOptions) Client {
         if opts.ServiceURL == "" {
                 opts.ServiceURL = ServiceURL
         }
         if opts.Logger == nil {
                 opts.Logger = logging.Null()
@@ skipping 34 matching lines @@
         storage storage
 
         // deployer knows how to install packages to local file system. Thread safe.
         deployer local.Deployer
 
         // tagCache is used to cache (pkgname, tag) -> instanceID mapping.
         // Thread safe, but lazily initialized under lock.
         tagCache     *internal.TagCache
         tagCacheInit sync.Once
 
+        // instanceCache is a file-system based cache of instances.
+        instanceCache     *internal.InstanceCache
+        instanceCacheInit sync.Once
+
         // 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.
@@ skipping 73 matching lines @@
         out, err := client.tagCache.Save()
         if err == nil {
                 err = local.EnsureFile(fs, path, bytes.NewReader(out))
         }
         if err != nil {
                 client.Logger.Warningf("cipd: failed to update tag cache - %s", err)
         }
         client.tagCache = nil
 }
 
+// instanceCachePath returns path to the instance cache directory or "" if
+// instance cache is disabled.
+func (client *clientImpl) instanceCachePath() string {
+        if client.CacheDir == "" {
+                return ""
+        }
+        return filepath.Join(client.CacheDir, "instances")
+}
+
+// getInstanceCache lazy-initializes instanceCache and returns it.
+func (client *clientImpl) getInstanceCache() *internal.InstanceCache {
+        client.instanceCacheInit.Do(func() {
+                if path := client.instanceCachePath(); path != "" {
+                        cachePath := local.NewFileSystem(path, client.Logger)
+                        client.instanceCache = internal.LoadInstanceCache(cachePath, client.Logger, client.clock.now())
+                }
+        })
+        return client.instanceCache
+}
+
+// closeInstanceCache dumps any changes made to instance cache to disk, if necessary.
+func (client *clientImpl) closeInstanceCache() {
+        if client.instanceCache == nil || !client.instanceCache.Dirty() {
+                return
+        }
+
+        if err := client.instanceCache.Save(); err != nil {
+                client.Logger.Warningf("cipd: failed to save instance cache - %s", err)
+        }
+        client.instanceCache = 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)
 }
 
 func (client *clientImpl) ListPackages(path string, recursive bool) ([]string, error) {
         pkgs, dirs, err := client.remote.listPackages(path, recursive)
@@ skipping 242 matching lines @@
 }
 
 func (client *clientImpl) FetchInstanceRefs(pin common.Pin, refs []string) ([]RefInfo, error) {
         err := common.ValidatePin(pin)
         if err != nil {
                 return nil, err
         }
         return client.remote.fetchRefs(pin, refs)
 }
 
 func (client *clientImpl) FetchInstance(pin common.Pin, output io.WriteSeeker) error {

[Vadim Sh., 2016/04/11 23:47:50]

I think I prefer 

func ... FetchInstance(...) ... {
   ..

  cache := ...
  if cache == nil {
    return client.fetchInstanceNoCache(...)
  } else {
    return client.fetchInstanceWithCache(...)
  }
}

instead of bunch of back-and-forth ifs.

[nodir, 2016/04/13 02:05:30]

Done.

         err := common.ValidatePin(pin)
         if err != nil {
                 return err
         }
+
+        now := client.clock.now()
+
+        cache := client.getInstanceCache()
+        if cache != nil {
+                switch err := cache.Get(pin, output, now); {
+                case os.IsNotExist(err):
+                        // output is not corrupted.
+
+                case err != nil:
+                        client.Logger.Warningf("cipd: could not get %s from cache - %s", pin, err)
+                        // Output may be corrupted. Rewinding back.
+                        if _, err := output.Seek(0, 0); err != nil {
+                                return err
+                        }
+
+                default:
+                        client.Logger.Infof("cipd: cache hit for %s", pin)
+                        return nil
+                }
+        }
+
         client.Logger.Infof("cipd: resolving fetch URL for %s", pin)
         fetchInfo, err := client.remote.fetchInstance(pin)
-        if err == nil {
-                err = client.storage.download(fetchInfo.fetchURL, output)
+        fetchFailed := func(err error) {
+                client.Logger.Errorf("cipd: failed to fetch %s - %s", pin, err)
         }
         if err != nil {
-                client.Logger.Errorf("cipd: failed to fetch %s - %s", pin, err)
+                fetchFailed(err)
                 return err
         }
+
+        if cache == nil {
+                if err := client.storage.download(fetchInfo.fetchURL, output); err != nil {
+                        fetchFailed(err)
+                        return err
+                }
+        } else {
+                // With cache, we need to download the instance to two places.
+                // Download to the cache and then read from there.
+                err := cache.Put(pin, now, func(f *os.File) error {
+                        err := client.storage.download(fetchInfo.fetchURL, f)
+                        if err != nil {
+                                fetchFailed(err)
+                        }
+                        return err
+                })
+                if err != nil {
+                        return err
+                }
+
+                // Now read it back.
+                if err := cache.Get(pin, output, now); err != nil {
+                        return err
+                }
+        }
+
         client.Logger.Infof("cipd: successfully fetched %s", pin)
         return nil
 }
 
 func (client *clientImpl) FetchAndDeployInstance(pin common.Pin) error {
         err := common.ValidatePin(pin)
         if err != nil {
                 return err
         }
 
@@ skipping 166 matching lines @@
                 client.Logger.Infof("All changes applied.")
                 return actions, nil
         }
         return actions, ErrEnsurePackagesFailed
 }
 
 func (client *clientImpl) Close() {
         client.lock.Lock()
         defer client.lock.Unlock()
         client.closeTagCache()
+        client.closeInstanceCache()
         client.authClient = nil
         client.anonClient = nil
 }
 
 ////////////////////////////////////////////////////////////////////////////////
 // Private structs and interfaces.
 
 type clock interface {
         now() time.Time
         sleep(time.Duration)
@@ skipping 71 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
 }