Implement config service cache on top of datastore

luci_config/appengine/backend/datastore/ds.go

Index: luci_config/appengine/backend/datastore/ds.go
diff --git a/luci_config/appengine/backend/datastore/ds.go b/luci_config/appengine/backend/datastore/ds.go
new file mode 100644
index 0000000000000000000000000000000000000000..d1f99b4ddb6e670656ba3e6d85c54d9cd4886bfa
--- /dev/null
+++ b/luci_config/appengine/backend/datastore/ds.go
@@ -0,0 +1,289 @@
+// Copyright 2016 The LUCI Authors. All rights reserved.
+// Use of this source code is governed under the Apache License, Version 2.0
+// that can be found in the LICENSE file.
+
+package datastore
+
+import (
+	"time"
+
+	"github.com/luci/luci-go/appengine/datastorecache"
+	"github.com/luci/luci-go/common/clock"
+	"github.com/luci/luci-go/common/errors"
+	log "github.com/luci/luci-go/common/logging"
+	"github.com/luci/luci-go/luci_config/common/cfgtypes"
+	"github.com/luci/luci-go/luci_config/server/cfgclient"
+	"github.com/luci/luci-go/luci_config/server/cfgclient/access"
+	"github.com/luci/luci-go/luci_config/server/cfgclient/backend"
+	"github.com/luci/luci-go/luci_config/server/cfgclient/backend/caching"
+
+	"golang.org/x/net/context"
+)
+
+const (
+	dsCacheSchema = "v1"
+
+	// RPCDeadline is the deadline applied to config service RPCs.
+	RPCDeadline = 10 * time.Minute
+)
+
+var dsHandlerKey = "github.com/luci/luci-go/appengine/gaeconfig.dsHandlerKey"
+
+func getCacheHandler(c context.Context) datastorecache.Handler {
+	v, _ := c.Value(&dsHandlerKey).(datastorecache.Handler)
+	return v
+}
+
+// Cache is our registered datastore cache, bound to our Config Handler. The
+// generator function is used by the cache manager task to get a Handler
+// instance during refresh.
+var Cache = datastorecache.Cache{
+	Name:                 "github.com/luci/luci-go/appengine/gaeconfig",
+	AccessUpdateInterval: 24 * time.Hour,
+	PruneFactor:          4,
+	Parallel:             16,
+	HandlerFunc:          getCacheHandler,
+}
+
+// dsCacheBackend is an interface around datastoreCache.Cache functionality used
+// by dsCacheFilter.
+//
+// We specialize this in testing to swap in other cache backends.
+type dsCacheBackend interface {
+	Get(c context.Context, key []byte) (datastorecache.Value, error)
+}
+
+// Config is a datastore cache configuration.
+//
+// Cache parameters can be extracted from a config instance.
+type Config struct {
+	// RefreshInterval is the cache entry refresh interval.
+	RefreshInterval time.Duration
+	// FailOpen, if true, means that a cache miss or failure should be passed
+	// through to the underlying cache.
+	FailOpen bool
+
+	// userProjAccess is cache of the current user's project access lookups.
+	userProjAccess map[string]bool
+	// anonProjAccess is cache of the current user's project access lookups.
+	anonProjAccess map[string]bool
+
+	// cache, if nil, is the datastore cache. This can be set for testing.
+	cache dsCacheBackend
+}
+
+// Backend wraps the specified Backend in a datastore-backed cache.
+func (dc *Config) Backend(base backend.B) backend.B {
+	return &caching.Backend{
+		B:           base,
+		FailOnError: !dc.FailOpen,
+		CacheGet:    dc.cacheGet,
+	}
+}
+
+// WithHandler installs a datastorecache.Handler into our Context. This is
+// used during datastore cache's Refresh calls.
+//
+// The Handler binds the parameters and Loader to the resolution call. For
+// service resolution, this will be the Loader that is provided by the caching
+// layer. For cron refresh, this will be the generic Loader provided by
+// CronLoader.
+func (dc *Config) WithHandler(c context.Context, l caching.Loader, timeout time.Duration) context.Context {
+	handler := dsCacheHandler{
+		refreshInterval: dc.RefreshInterval,
+		failOpen:        dc.FailOpen,
+		loader:          l,
+		loaderTimeout:   timeout,
+	}
+	return context.WithValue(c, &dsHandlerKey, &handler)
+}
+
+func (dc *Config) cacheGet(c context.Context, key caching.Key, l caching.Loader) (
+	*caching.Value, error) {
+
+	cache := dc.cache
+	if cache == nil {
+		cache = &Cache
+	}
+
+	// Modify our cache key to always refresh AsService (ACLs will be asserted on
+	// load) and request full-content.
+	origAuthority := key.Authority
+	key.Authority = backend.AsService
+
+	// If we can assert access based on the operation's target ConfigSet, do that
+	// before actually performing the action.
+	switch key.Op {
+	case caching.OpGet, caching.OpConfigSetURL:
+		// We can deny access to this operation based solely on the requested
+		// ConfigSet.
+		if err := access.Check(c, origAuthority, cfgtypes.ConfigSet(key.ConfigSet)); err != nil {
+			// Access check failed, simulate ErrNoSuchConfig.
+			//
+			// OpGet: Indicate no content via empty Items.
+			// OpConfigSetURL: Indicate no content via empty "URL" field in value.
+			return &caching.Value{}, nil
+		}
+	}
+
+	// For content-fetching operations, always request full content.
+	switch key.Op {
+	case caching.OpGet, caching.OpGetAll:
+		// Always ask for full content.
+		key.Content = true
+	}
+
+	// Encode our caching key, and use this for our datastore cache key.
+	//
+	// This gets recoded in dsCacheHandler's "Refresh" to identify the cache
+	// operation that is being performed.
+	encKey, err := caching.Encode(&key)
+	if err != nil {
+		return nil, errors.Annotate(err).Reason("failed to encode cache key").Err()
+	}
+
+	// Construct a cache handler.
+	v, err := cache.Get(dc.WithHandler(c, l, 0), encKey)
+	if err != nil {
+		return nil, err
+	}
+
+	// Decode our response.
+	if v.Schema != dsCacheSchema {
+		return nil, errors.Reason("response schema (%(resp)q) doesn't match current (%(cur)q)").
+			D("resp", v.Schema).D("cur", dsCacheSchema).Err()
+	}
+
+	cacheValue, err := caching.DecodeValue(v.Data)
+	if err != nil {
+		return nil, errors.Annotate(err).Reason("failed to decode cached value").Err()
+	}
+
+	// Prune any responses that are not permitted for the supplied Authority.
+	switch key.Op {
+	case caching.OpGetAll:
+		if len(cacheValue.Items) > 0 {
+			// Shift over any elements that can't be accessed.
+			ptr := 0
+			for _, itm := range cacheValue.Items {
+				if dc.accessConfigSet(c, origAuthority, itm.ConfigSet) {
+					cacheValue.Items[ptr] = itm
+					ptr++
+				}
+			}
+			cacheValue.Items = cacheValue.Items[:ptr]
+		}
+	}
+
+	return cacheValue, nil
+}
+
+func (dc *Config) accessConfigSet(c context.Context, a backend.Authority, configSet string) bool {
+	var cacheMap *map[string]bool
+	switch a {
+	case backend.AsService:
+		return true
+	case backend.AsUser:
+		cacheMap = &dc.userProjAccess
+	default:
+		cacheMap = &dc.anonProjAccess
+	}
+
+	// If we've already cached this project access, return the cached value.
+	if v, ok := (*cacheMap)[configSet]; ok {
+		return v
+	}
+
+	// Perform a soft access check.
+	canAccess := false
+	switch err := access.Check(c, a, cfgtypes.ConfigSet(configSet)); err {
+	case nil:
+		canAccess = true
+
+	case access.ErrNoAccess:
+		// No access.
+		break
+	case cfgclient.ErrNoConfig:
+		log.Fields{
+			"configSet": configSet,
+		}.Debugf(c, "Checking access to project without a config.")
+	default:
+		log.Fields{
+			log.ErrorKey: err,
+			"configSet":  configSet,
+		}.Warningf(c, "Error checking for project access.")
+	}
+
+	// Cache the result for future lookups.
+	if *cacheMap == nil {
+		*cacheMap = make(map[string]bool)
+	}
+	(*cacheMap)[configSet] = canAccess
+	return canAccess
+}
+
+type dsCacheValue struct {
+	// Key is the cache Key.
+	Key caching.Key `json:"k"`
+	// Value is the cache Value.
+	Value *caching.Value `json:"v"`
+}
+
+type dsCacheHandler struct {
+	failOpen        bool
+	refreshInterval time.Duration
+	loader          caching.Loader
+
+	// loaderTimeout, if >0, will be applied prior to performing the loader
+	// operation. This is used for cron operations.
+	loaderTimeout time.Duration
+}
+
+func (dch *dsCacheHandler) FailOpen() bool                       { return dch.failOpen }
+func (dch *dsCacheHandler) RefreshInterval([]byte) time.Duration { return dch.refreshInterval }
+
+func (dch *dsCacheHandler) Refresh(c context.Context, key []byte, v datastorecache.Value) (datastorecache.Value, error) {
+	// Decode the key into our caching key.
+	var ck caching.Key
+	if err := caching.Decode(key, &ck); err != nil {
+		return v, errors.Annotate(err).Reason("failed to decode cache key").Err()
+	}
+
+	var cv *caching.Value
+	if v.Schema == dsCacheSchema && len(v.Data) > 0 {
+		// We have a currently-cached value, so decode it into "cv".
+		var err error
+		if cv, err = caching.DecodeValue(v.Data); err != nil {
+			return v, errors.Annotate(err).Reason("failed to decode cache value").Err()
+		}
+	}
+
+	// Apply our timeout, if configured (influences urlfetch).
+	if dch.loaderTimeout > 0 {
+		var cancelFunc context.CancelFunc
+		c, cancelFunc = clock.WithTimeout(c, dch.loaderTimeout)
+		defer cancelFunc()
+	}
+
+	// Perform a cache load on this value.
+	cv, err := dch.loader(c, ck, cv)
+	if err != nil {
+		return v, errors.Annotate(err).Reason("failed to load cache value").Err()
+	}
+
+	// Encode the resulting cache value.
+	if v.Data, err = cv.Encode(); err != nil {
+		return v, errors.Annotate(err).Reason("failed to encode cache value").Err()
+	}
+	v.Schema = dsCacheSchema
+	v.Description = ck.String()
+	return v, nil
+}
+
+// CronLoader returns a caching.Loader implementation to be used
+// by the Cron task.
+func CronLoader(b backend.B) caching.Loader {
+	return func(c context.Context, k caching.Key, v *caching.Value) (*caching.Value, error) {
+		return caching.CacheLoad(c, b, k, v)
+	}
+}