lb read cache everything under it (#1471)

* add everything else to read cache

* finagle front end configuration

* add metrics to read data access thing

also fixes metrics, need to use the req context instead of the gin context in
order to get spans. more thumbs down for ginny.

this wraps the read data access on the whole, since the singleflight thing can
show up as latency around accessing the datastore, but they are not
necessarily the same thing (eg if 2 reqs come in, 1 waits for the other, but
won't have a latency for actually hitting the datastore, it's just waiting for
the cache to fill - instead of showing a void, show this to make it clear
where the request is).

Author: rdallman

api/agent/data_access.go

@@ -7,6 +7,7 @@ import (
 	"github.com/fnproject/fn/api/models"
 	"github.com/golang/groupcache/singleflight"
 	"github.com/patrickmn/go-cache"
+	"go.opencensus.io/trace"
 )
 
 // XXX(reed): this is only used by the front end now, this should be in the server/ package
@@ -25,6 +26,39 @@ type DataAccess interface {
 	ReadDataAccess
 }
 
+// NewMetricReadDataAccess adds metrics to a ReadDataAccess
+func NewMetricReadDataAccess(rda ReadDataAccess) ReadDataAccess {
+	return &metricda{rda}
+}
+
+type metricda struct {
+	rda ReadDataAccess
+}
+
+func (m *metricda) GetTriggerBySource(ctx context.Context, appID string, triggerType, source string) (*models.Trigger, error) {
+	ctx, span := trace.StartSpan(ctx, "rda_get_trigger_by_source")
+	defer span.End()
+	return m.rda.GetTriggerBySource(ctx, appID, triggerType, source)
+}
+
+func (m *metricda) GetAppID(ctx context.Context, appName string) (string, error) {
+	ctx, span := trace.StartSpan(ctx, "rda_get_app_id")
+	defer span.End()
+	return m.rda.GetAppID(ctx, appName)
+}
+
+func (m *metricda) GetAppByID(ctx context.Context, appID string) (*models.App, error) {
+	ctx, span := trace.StartSpan(ctx, "rda_get_app_by_id")
+	defer span.End()
+	return m.rda.GetAppByID(ctx, appID)
+}
+
+func (m *metricda) GetFnByID(ctx context.Context, fnID string) (*models.Fn, error) {
+	ctx, span := trace.StartSpan(ctx, "rda_get_fn_by_id")
+	defer span.End()
+	return m.rda.GetFnByID(ctx, fnID)
+}
+
 // CachedDataAccess wraps a DataAccess and caches the results of GetApp.
 type cachedDataAccess struct {
 	ReadDataAccess
@@ -42,12 +76,31 @@ func NewCachedDataAccess(da ReadDataAccess) ReadDataAccess {
 	return cda
 }
 
-func appIDCacheKey(appID string) string {
-	return "a:" + appID
+func appIDCacheKey(appID string) string     { return "a:" + appID }
+func appNameCacheKey(appName string) string { return "n:" + appName }
+func fnCacheKey(fnID string) string         { return "f:" + fnID }
+func trigSourceCacheKey(app, typ, source string) string {
+	return "t:" + app + string('\x00') + typ + string('\x00') + source
 }
 
 func (da *cachedDataAccess) GetAppID(ctx context.Context, appName string) (string, error) {
-	return da.ReadDataAccess.GetAppID(ctx, appName)
+	key := appNameCacheKey(appName)
+	app, ok := da.cache.Get(key)
+	if ok {
+		return app.(string), nil
+	}
+
+	resp, err := da.singleflight.Do(key,
+		func() (interface{}, error) {
+			return da.ReadDataAccess.GetAppID(ctx, appName)
+		})
+
+	if err != nil {
+		return "", err
+	}
+	app = resp.(string)
+	da.cache.Set(key, app, cache.DefaultExpiration)
+	return app.(string), nil
 }
 
 func (da *cachedDataAccess) GetAppByID(ctx context.Context, appID string) (*models.App, error) {
@@ -69,3 +122,43 @@ func (da *cachedDataAccess) GetAppByID(ctx context.Context, appID string) (*mode
 	da.cache.Set(key, app, cache.DefaultExpiration)
 	return app.(*models.App), nil
 }
+
+func (da *cachedDataAccess) GetTriggerBySource(ctx context.Context, appID string, triggerType, source string) (*models.Trigger, error) {
+	key := trigSourceCacheKey(appID, triggerType, source)
+	trigger, ok := da.cache.Get(key)
+	if ok {
+		return trigger.(*models.Trigger), nil
+	}
+
+	resp, err := da.singleflight.Do(key,
+		func() (interface{}, error) {
+			return da.ReadDataAccess.GetTriggerBySource(ctx, appID, triggerType, source)
+		})
+
+	if err != nil {
+		return nil, err
+	}
+	trigger = resp.(*models.Trigger)
+	da.cache.Set(key, trigger, cache.DefaultExpiration)
+	return trigger.(*models.Trigger), nil
+}
+
+func (da *cachedDataAccess) GetFnByID(ctx context.Context, fnID string) (*models.Fn, error) {
+	key := fnCacheKey(fnID)
+	fn, ok := da.cache.Get(key)
+	if ok {
+		return fn.(*models.Fn), nil
+	}
+
+	resp, err := da.singleflight.Do(key,
+		func() (interface{}, error) {
+			return da.ReadDataAccess.GetFnByID(ctx, fnID)
+		})
+
+	if err != nil {
+		return nil, err
+	}
+	fn = resp.(*models.Fn)
+	da.cache.Set(key, fn, cache.DefaultExpiration)
+	return fn.(*models.Fn), nil
+}

api/datastore/internal/datastoreutil/metrics.go

@@ -3,9 +3,8 @@ package datastoreutil
 import (
 	"context"
 
-	"go.opencensus.io/trace"
-
 	"github.com/fnproject/fn/api/models"
+	"go.opencensus.io/trace"
 )
 
 func MetricDS(ds models.Datastore) models.Datastore {

api/server/runner_fninvoke.go

@@ -57,12 +57,13 @@ func (s *Server) handleFnInvokeCall(c *gin.Context) {
 // handleTriggerHTTPFunctionCall2 executes the function and returns an error
 // Requires the following in the context:
 func (s *Server) handleFnInvokeCall2(c *gin.Context) error {
-	fn, err := s.lbReadAccess.GetFnByID(c, c.Param(api.FnID))
+	ctx := c.Request.Context()
+	fn, err := s.lbReadAccess.GetFnByID(ctx, c.Param(api.FnID))
 	if err != nil {
 		return err
 	}
 
-	app, err := s.lbReadAccess.GetAppByID(c, fn.AppID)
+	app, err := s.lbReadAccess.GetAppByID(ctx, fn.AppID)
 	if err != nil {
 		return err
 	}
@@ -71,7 +72,7 @@ func (s *Server) handleFnInvokeCall2(c *gin.Context) error {
 	if models.IsFuncError(err) || err == nil {
 		// report all user-directed errors and function responses from here, after submit has run.
 		// this is our never ending attempt to distinguish user and platform errors.
-		ctx, err := tag.New(c.Request.Context(),
+		ctx, err := tag.New(ctx,
 			tag.Insert(whodunitKey, "user"),
 		)
 		if err != nil {

api/server/server.go

@@ -201,7 +201,6 @@ type Server struct {
 	// TODO: extend this to cover gRPC options.
 	svcConfigs map[string]*http.Server
 
-	// Agent enqueue  and read stores
 	lbReadAccess           agent.ReadDataAccess
 	noHTTTPTriggerEndpoint bool
 	noFnInvokeEndpoint     bool
@@ -335,8 +334,7 @@ func WithDBURL(dbURL string) Option {
 			if err != nil {
 				return err
 			}
-			s.datastore = ds
-			s.lbReadAccess = agent.NewCachedDataAccess(s.datastore)
+			return WithDatastore(ds)(ctx, s)
 		}
 		return nil
 	}
@@ -361,7 +359,7 @@ func WithTLS(service string, tlsCfg *tls.Config) Option {
 // WithReadDataAccess overrides the LB read DataAccess for a server
 func WithReadDataAccess(ds agent.ReadDataAccess) Option {
 	return func(ctx context.Context, s *Server) error {
-		s.lbReadAccess = ds
+		s.lbReadAccess = agent.NewMetricReadDataAccess(ds)
 		return nil
 	}
 }
@@ -370,8 +368,10 @@ func WithReadDataAccess(ds agent.ReadDataAccess) Option {
 func WithDatastore(ds models.Datastore) Option {
 	return func(ctx context.Context, s *Server) error {
 		s.datastore = ds
+		s.datastore = datastore.Wrap(s.datastore)
+		s.datastore = fnext.NewDatastore(s.datastore, s.appListeners, s.fnListeners, s.triggerListeners)
 		if s.lbReadAccess == nil {
-			s.lbReadAccess = agent.NewCachedDataAccess(ds)
+			return WithReadDataAccess(agent.NewCachedDataAccess(s.datastore))(ctx, s)
 		}
 		return nil
 	}
@@ -397,10 +397,6 @@ func (s *Server) defaultRunnerPool() (pool.RunnerPool, error) {
 func WithFullAgent() Option {
 	return func(ctx context.Context, s *Server) error {
 		s.nodeType = ServerTypeFull
-
-		if s.datastore == nil {
-			return errors.New("full nodes must configure FN_DB_URL")
-		}
 		s.agent = agent.New()
 		return nil
 	}
@@ -414,9 +410,6 @@ func WithAgentFromEnv() Option {
 		case ServerTypeAPI:
 			return errors.New("should not initialize an agent for an Fn API node")
 		case ServerTypePureRunner:
-			if s.datastore != nil {
-				return errors.New("pure runner nodes must not be configured with a datastore (FN_DB_URL)")
-			}
 			cancelCtx, cancel := context.WithCancel(ctx)
 			prAgent, err := agent.DefaultPureRunner(cancel, s.svcConfigs[GRPCServer].Addr, s.svcConfigs[GRPCServer].TLSConfig)
 			if err != nil {
@@ -430,9 +423,6 @@ func WithAgentFromEnv() Option {
 			if runnerURL == "" {
 				return errors.New("no FN_RUNNER_API_URL provided for an Fn NuLB node")
 			}
-			if s.datastore != nil {
-				return errors.New("lb nodes must not be configured with a datastore (FN_DB_URL)")
-			}
 
 			cl, err := hybrid.NewClient(runnerURL)
 			if err != nil {
@@ -454,7 +444,10 @@ func WithAgentFromEnv() Option {
 				placer = pool.NewNaivePlacer(&placerCfg)
 			}
 
-			s.lbReadAccess = agent.NewCachedDataAccess(cl)
+			err = WithReadDataAccess(agent.NewCachedDataAccess(cl))(ctx, s)
+			if err != nil {
+				return errors.New("LBAgent creation failed")
+			}
 			s.agent, err = agent.NewLBAgent(runnerPool, placer)
 			if err != nil {
 				return errors.New("LBAgent creation failed")
@@ -499,6 +492,7 @@ func WithAdminServer(port int) Option {
 	}
 }
 
+// WithHTTPConfig allows configuring specific http servers
 func WithHTTPConfig(service string, cfg *http.Server) Option {
 	return func(ctx context.Context, s *Server) error {
 		s.svcConfigs[service] = cfg
@@ -528,6 +522,11 @@ func New(ctx context.Context, opts ...Option) *Server {
 			AdminServer: &http.Server{},
 			GRPCServer:  &http.Server{},
 		},
+		// MUST initialize these before opts
+		appListeners:     new(appListeners),
+		fnListeners:      new(fnListeners),
+		triggerListeners: new(triggerListeners),
+
 		// Almost everything else is configured through opts (see NewFromEnv for ex.) or below
 	}
 
@@ -596,14 +595,6 @@ func New(ctx context.Context, opts ...Option) *Server {
 	s.AdminRouter.Use(panicWrap)
 	s.bindHandlers(ctx)
 
-	s.appListeners = new(appListeners)
-	s.fnListeners = new(fnListeners)
-	s.triggerListeners = new(triggerListeners)
-
-	// TODO it's not clear that this is always correct as the read store  won't  get wrapping
-	s.datastore = datastore.Wrap(s.datastore)
-	s.datastore = fnext.NewDatastore(s.datastore, s.appListeners, s.fnListeners, s.triggerListeners)
-
 	return s
 }