push app/route cache down to datastore (#303)

cache now implements models.Datastore by just embedding one and then changing
GetApp and GetRoute to have the cache inside. this makes it really flexible
for things like testing, so now the agent doesn't automagically do caching,
now it must be passed a datastore that was wrapped with a cache datastore.
the datastore in the server can remain separate and not use the cache still,
and then now the agent when running fn 'for real' is configured with the cache
baked in. this seems a lot cleaner than what we had and gets the cache out of
the way and it's easier to swap in / out / extend.

api/agent/agent.go

@@ -14,11 +14,9 @@ import (
 	"github.com/fnproject/fn/api/agent/drivers/docker"
 	"github.com/fnproject/fn/api/agent/protocol"
 	"github.com/fnproject/fn/api/common"
-	"github.com/fnproject/fn/api/common/singleflight"
 	"github.com/fnproject/fn/api/id"
 	"github.com/fnproject/fn/api/models"
 	"github.com/opentracing/opentracing-go"
-	"github.com/patrickmn/go-cache"
 	"github.com/sirupsen/logrus"
 )
 
@@ -57,7 +55,6 @@ import (
 // TODO the log api should be plaintext (or at least offer it)
 // TODO func logger needs to be hanged, dragged and quartered. in reverse order.
 // TODO we should probably differentiate ran-but-timeout vs timeout-before-run
-// TODO push down the app/route cache into Datastore that decorates
 // TODO between calls, logs and stderr can contain output/ids from previous call. need elegant solution. grossness.
 // TODO if async would store requests (or interchange format) it would be slick, but
 // if we're going to store full calls in db maybe we should only queue pointers to ids?
@@ -127,11 +124,6 @@ type agent struct {
 	hMu sync.RWMutex // protects hot
 	hot map[string]chan slot
 
-	// TODO could make this separate too...
-	// cache for apps and routes
-	cache        *cache.Cache
-	singleflight singleflight.SingleFlight // singleflight assists Datastore // TODO rename
-
 	// TODO we could make a separate struct for the memory stuff
 	// cond protects access to ramUsed
 	cond *sync.Cond
@@ -157,7 +149,6 @@ func New(ds models.Datastore, mq models.MessageQueue) Agent {
 		mq:       mq,
 		driver:   driver,
 		hot:      make(map[string]chan slot),
-		cache:    cache.New(5*time.Second, 1*time.Minute),
 		cond:     sync.NewCond(new(sync.Mutex)),
 		ramTotal: getAvailableMemory(),
 		shutdown: make(chan struct{}),

api/agent/call.go

@@ -13,7 +13,6 @@ import (
 	"github.com/fnproject/fn/api/models"
 	"github.com/go-openapi/strfmt"
 	"github.com/opentracing/opentracing-go"
-	"github.com/patrickmn/go-cache"
 	"github.com/sirupsen/logrus"
 )
 
@@ -43,13 +42,12 @@ type CallOpt func(a *agent, c *call) error
 
 func FromRequest(appName, path string, req *http.Request) CallOpt {
 	return func(a *agent, c *call) error {
-		// TODO we need to add a little timeout to these 2 things
+		app, err := a.ds.GetApp(req.Context(), appName)
-		app, err := a.app(req.Context(), appName)
 		if err != nil {
 			return err
 		}
 
-		route, err := a.route(req.Context(), appName, path)
+		route, err := a.ds.GetRoute(req.Context(), appName, path)
 		if err != nil {
 			return err
 		}
@@ -295,50 +293,6 @@ func (c *call) End(ctx context.Context, err error) {
 	}
 }
 
-func (a *agent) route(ctx context.Context, appName, path string) (*models.Route, error) {
-	key := routeCacheKey(appName, path)
-	route, ok := a.cache.Get(key)
-	if ok {
-		return route.(*models.Route), nil
-	}
-
-	resp, err := a.singleflight.Do(key,
-		func() (interface{}, error) { return a.ds.GetRoute(ctx, appName, path) },
-	)
-	if err != nil {
-		return nil, err
-	}
-	route = resp.(*models.Route)
-	a.cache.Set(key, route, cache.DefaultExpiration)
-	return route.(*models.Route), nil
-}
-
-func (a *agent) app(ctx context.Context, appName string) (*models.App, error) {
-	key := appCacheKey(appName)
-	app, ok := a.cache.Get(key)
-	if ok {
-		return app.(*models.App), nil
-	}
-
-	resp, err := a.singleflight.Do(key,
-		func() (interface{}, error) { return a.ds.GetApp(ctx, appName) },
-	)
-	if err != nil {
-		return nil, err
-	}
-	app = resp.(*models.App)
-	a.cache.Set(key, app, cache.DefaultExpiration)
-	return app.(*models.App), nil
-}
-
-func routeCacheKey(appname, path string) string {
-	return "r:" + appname + "\x00" + path
-}
-
-func appCacheKey(appname string) string {
-	return "a:" + appname
-}
-
 func fakeHandler(http.ResponseWriter, *http.Request, Params) {}
 
 // TODO what is this stuff anyway?

api/datastore/cache/cache.go

@@ -0,0 +1,71 @@
+package cache
+
+import (
+	"context"
+	"time"
+
+	"github.com/fnproject/fn/api/common/singleflight"
+	"github.com/fnproject/fn/api/models"
+	"github.com/patrickmn/go-cache"
+)
+
+type cacheDB struct {
+	models.Datastore
+
+	cache        *cache.Cache
+	singleflight singleflight.SingleFlight // singleflight assists Datastore
+}
+
+// Wrap implements models.Datastore by wrapping an existing datastore and
+// adding caching around certain methods. At present, GetApp and GetRoute add
+// caching.
+func Wrap(ds models.Datastore) models.Datastore {
+	return &cacheDB{
+		Datastore: ds,
+		cache:     cache.New(5*time.Second, 1*time.Minute), // TODO configurable from env
+	}
+}
+
+func (c *cacheDB) GetApp(ctx context.Context, appName string) (*models.App, error) {
+	key := appCacheKey(appName)
+	app, ok := c.cache.Get(key)
+	if ok {
+		return app.(*models.App), nil
+	}
+
+	resp, err := c.singleflight.Do(key,
+		func() (interface{}, error) { return c.Datastore.GetApp(ctx, appName) },
+	)
+	if err != nil {
+		return nil, err
+	}
+	app = resp.(*models.App)
+	c.cache.Set(key, app, cache.DefaultExpiration)
+	return app.(*models.App), nil
+}
+
+func (c *cacheDB) GetRoute(ctx context.Context, appName, path string) (*models.Route, error) {
+	key := routeCacheKey(appName, path)
+	route, ok := c.cache.Get(key)
+	if ok {
+		return route.(*models.Route), nil
+	}
+
+	resp, err := c.singleflight.Do(key,
+		func() (interface{}, error) { return c.Datastore.GetRoute(ctx, appName, path) },
+	)
+	if err != nil {
+		return nil, err
+	}
+	route = resp.(*models.Route)
+	c.cache.Set(key, route, cache.DefaultExpiration)
+	return route.(*models.Route), nil
+}
+
+func routeCacheKey(appname, path string) string {
+	return "r:" + appname + "\x00" + path
+}
+
+func appCacheKey(appname string) string {
+	return "a:" + appname
+}

api/server/server.go

@@ -14,6 +14,7 @@ import (
 	"github.com/fnproject/fn/api/agent"
 	"github.com/fnproject/fn/api/common"
 	"github.com/fnproject/fn/api/datastore"
+	"github.com/fnproject/fn/api/datastore/cache"
 	"github.com/fnproject/fn/api/id"
 	"github.com/fnproject/fn/api/logs"
 	"github.com/fnproject/fn/api/models"
@@ -74,7 +75,7 @@ func NewFromEnv(ctx context.Context) *Server {
 // New creates a new Functions server with the passed in datastore, message queue and API URL
 func New(ctx context.Context, ds models.Datastore, mq models.MessageQueue, logDB models.LogStore, opts ...ServerOption) *Server {
 	s := &Server{
-		Agent:     agent.New(ds, mq),
+		Agent:     agent.New(cache.Wrap(ds), mq), // only add datastore caching to agent
 		Router:    gin.New(),
 		Datastore: ds,
 		MQ:        mq,