mirror of
https://github.com/fnproject/fn.git
synced 2022-10-28 21:29:17 +03:00
Expiring cache
This commit is contained in:
committed by
Derek Schultz
parent
2be96689d4
commit
c3630eaa41
@@ -20,10 +20,6 @@ var (
|
||||
code: http.StatusGatewayTimeout,
|
||||
error: errors.New("Timed out"),
|
||||
}
|
||||
ErrRunnerRouteNotFound = err{
|
||||
code: http.StatusNotFound,
|
||||
error: errors.New("Route not found on that application"),
|
||||
}
|
||||
ErrAppsValidationMissingName = err{
|
||||
code: http.StatusBadRequest,
|
||||
error: errors.New("Missing app name"),
|
||||
|
||||
@@ -1,88 +0,0 @@
|
||||
// Package routecache is meant to assist in resolving the most used routes at
|
||||
// an application. Implemented as a LRU, it returns always its full context for
|
||||
// iteration at the router handler.
|
||||
package routecache
|
||||
|
||||
// based on groupcache's LRU
|
||||
|
||||
import (
|
||||
"container/list"
|
||||
|
||||
"gitlab-odx.oracle.com/odx/functions/api/models"
|
||||
)
|
||||
|
||||
// Cache holds an internal linkedlist for hotness management. It is not safe
|
||||
// for concurrent use, must be guarded externally.
|
||||
type Cache struct {
|
||||
MaxEntries int
|
||||
|
||||
ll *list.List
|
||||
cache map[string]*list.Element
|
||||
}
|
||||
|
||||
// New returns a route cache.
|
||||
func New(maxentries int) *Cache {
|
||||
return &Cache{
|
||||
MaxEntries: maxentries,
|
||||
ll: list.New(),
|
||||
cache: make(map[string]*list.Element),
|
||||
}
|
||||
}
|
||||
|
||||
// Refresh updates internal linkedlist either adding a new route to the front,
|
||||
// or moving it to the front when used. It will discard seldom used routes.
|
||||
func (c *Cache) Refresh(route *models.Route) {
|
||||
if c.cache == nil {
|
||||
return
|
||||
}
|
||||
|
||||
if ee, ok := c.cache[route.AppName+route.Path]; ok {
|
||||
c.ll.MoveToFront(ee)
|
||||
ee.Value = route
|
||||
return
|
||||
}
|
||||
|
||||
ele := c.ll.PushFront(route)
|
||||
c.cache[route.AppName+route.Path] = ele
|
||||
if c.MaxEntries != 0 && c.ll.Len() > c.MaxEntries {
|
||||
c.removeOldest()
|
||||
}
|
||||
}
|
||||
|
||||
// Get looks up a path's route from the cache.
|
||||
func (c *Cache) Get(appname, path string) (route *models.Route, ok bool) {
|
||||
if c.cache == nil {
|
||||
return
|
||||
}
|
||||
if ele, hit := c.cache[appname+path]; hit {
|
||||
c.ll.MoveToFront(ele)
|
||||
return ele.Value.(*models.Route), true
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// Delete removes the element for the given appname and path from the cache.
|
||||
func (c *Cache) Delete(appname, path string) {
|
||||
if ele, hit := c.cache[appname+path]; hit {
|
||||
c.removeElement(ele)
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cache) removeOldest() {
|
||||
if c.cache == nil {
|
||||
return
|
||||
}
|
||||
if ele := c.ll.Back(); ele != nil {
|
||||
c.removeElement(ele)
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cache) removeElement(e *list.Element) {
|
||||
c.ll.Remove(e)
|
||||
kv := e.Value.(*models.Route)
|
||||
delete(c.cache, kv.AppName+kv.Path)
|
||||
}
|
||||
|
||||
func (c *Cache) Len() int {
|
||||
return len(c.cache)
|
||||
}
|
||||
@@ -47,7 +47,7 @@ func (s *Server) handleRouteCreateOrUpdate(c *gin.Context) {
|
||||
return
|
||||
}
|
||||
|
||||
s.cacheRefresh(resp.Route)
|
||||
s.cachedelete(resp.Route.AppName, resp.Route.Path)
|
||||
|
||||
c.JSON(http.StatusOK, resp)
|
||||
}
|
||||
|
||||
@@ -14,6 +14,7 @@ import (
|
||||
"github.com/Sirupsen/logrus"
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/go-openapi/strfmt"
|
||||
cache "github.com/patrickmn/go-cache"
|
||||
"gitlab-odx.oracle.com/odx/functions/api"
|
||||
"gitlab-odx.oracle.com/odx/functions/api/id"
|
||||
"gitlab-odx.oracle.com/odx/functions/api/models"
|
||||
@@ -44,7 +45,7 @@ func (s *Server) handleSpecial(c *gin.Context) {
|
||||
c.Request = r
|
||||
c.Set(api.AppName, r.Context().Value(api.AppName).(string))
|
||||
if c.MustGet(api.AppName).(string) == "" {
|
||||
handleErrorResponse(c, models.ErrRunnerRouteNotFound)
|
||||
handleErrorResponse(c, models.ErrRoutesNotFound)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -111,40 +112,45 @@ func (s *Server) handleRequest(c *gin.Context, enqueue models.Enqueue) {
|
||||
}
|
||||
|
||||
log.WithFields(logrus.Fields{"app": appName, "path": path}).Debug("Finding route on datastore")
|
||||
routes, err := s.loadroutes(ctx, models.RouteFilter{AppName: appName, Path: path})
|
||||
route, err := s.loadroute(ctx, appName, path)
|
||||
if err != nil {
|
||||
handleErrorResponse(c, err)
|
||||
return
|
||||
}
|
||||
|
||||
if len(routes) == 0 {
|
||||
handleErrorResponse(c, models.ErrRunnerRouteNotFound)
|
||||
if route == nil {
|
||||
handleErrorResponse(c, models.ErrRoutesNotFound)
|
||||
return
|
||||
}
|
||||
|
||||
log.WithField("routes", len(routes)).Debug("Got routes from datastore")
|
||||
route := routes[0]
|
||||
log = log.WithFields(logrus.Fields{"app": appName, "path": route.Path, "image": route.Image})
|
||||
log.Debug("Got route from datastore")
|
||||
|
||||
if s.serve(ctx, c, appName, route, app, path, reqID, payload, enqueue) {
|
||||
s.FireAfterDispatch(ctx, reqRoute)
|
||||
return
|
||||
}
|
||||
|
||||
handleErrorResponse(c, models.ErrRunnerRouteNotFound)
|
||||
handleErrorResponse(c, models.ErrRoutesNotFound)
|
||||
}
|
||||
|
||||
func (s *Server) loadroutes(ctx context.Context, filter models.RouteFilter) ([]*models.Route, error) {
|
||||
if route, ok := s.cacheget(filter.AppName, filter.Path); ok {
|
||||
return []*models.Route{route}, nil
|
||||
func (s *Server) loadroute(ctx context.Context, appName, path string) (*models.Route, error) {
|
||||
if route, ok := s.cacheget(appName, path); ok {
|
||||
return route, nil
|
||||
}
|
||||
key := routeCacheKey(appName, path)
|
||||
resp, err := s.singleflight.do(
|
||||
filter,
|
||||
key,
|
||||
func() (interface{}, error) {
|
||||
return s.Datastore.GetRoutesByApp(ctx, filter.AppName, &filter)
|
||||
return s.Datastore.GetRoute(ctx, appName, path)
|
||||
},
|
||||
)
|
||||
return resp.([]*models.Route), err
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
route := resp.(*models.Route)
|
||||
s.routeCache.Set(key, route, cache.DefaultExpiration)
|
||||
return route, nil
|
||||
}
|
||||
|
||||
// TODO: Should remove *gin.Context from these functions, should use only context.Context
|
||||
|
||||
@@ -7,25 +7,26 @@ import (
|
||||
"net/http"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
cache "github.com/patrickmn/go-cache"
|
||||
"gitlab-odx.oracle.com/odx/functions/api/datastore"
|
||||
"gitlab-odx.oracle.com/odx/functions/api/models"
|
||||
"gitlab-odx.oracle.com/odx/functions/api/mqs"
|
||||
"gitlab-odx.oracle.com/odx/functions/api/runner"
|
||||
"gitlab-odx.oracle.com/odx/functions/api/server/internal/routecache"
|
||||
)
|
||||
|
||||
func testRouterAsync(ds models.Datastore, mq models.MessageQueue, rnr *runner.Runner, enqueue models.Enqueue) *gin.Engine {
|
||||
ctx := context.Background()
|
||||
|
||||
s := &Server{
|
||||
Runner: rnr,
|
||||
Router: gin.New(),
|
||||
Datastore: ds,
|
||||
MQ: mq,
|
||||
Enqueue: enqueue,
|
||||
hotroutes: routecache.New(10),
|
||||
Runner: rnr,
|
||||
Router: gin.New(),
|
||||
Datastore: ds,
|
||||
MQ: mq,
|
||||
Enqueue: enqueue,
|
||||
routeCache: cache.New(60*time.Second, 5*time.Minute),
|
||||
}
|
||||
|
||||
r := s.Router
|
||||
|
||||
@@ -45,7 +45,7 @@ func TestRouteRunnerGet(t *testing.T) {
|
||||
}{
|
||||
{"/route", "", http.StatusNotFound, nil},
|
||||
{"/r/app/route", "", http.StatusNotFound, models.ErrAppsNotFound},
|
||||
{"/r/myapp/route", "", http.StatusNotFound, models.ErrRunnerRouteNotFound},
|
||||
{"/r/myapp/route", "", http.StatusNotFound, models.ErrRoutesNotFound},
|
||||
} {
|
||||
_, rec := routerRequest(t, srv.Router, "GET", test.path, nil)
|
||||
|
||||
@@ -60,8 +60,8 @@ func TestRouteRunnerGet(t *testing.T) {
|
||||
|
||||
if !strings.Contains(resp.Error.Message, test.expectedError.Error()) {
|
||||
t.Log(buf.String())
|
||||
t.Errorf("Test %d: Expected error message to have `%s`",
|
||||
i, test.expectedError.Error())
|
||||
t.Errorf("Test %d: Expected error message to have `%s`, but got `%s`",
|
||||
i, test.expectedError.Error(), resp.Error.Message)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -89,7 +89,7 @@ func TestRouteRunnerPost(t *testing.T) {
|
||||
}{
|
||||
{"/route", `{ "payload": "" }`, http.StatusNotFound, nil},
|
||||
{"/r/app/route", `{ "payload": "" }`, http.StatusNotFound, models.ErrAppsNotFound},
|
||||
{"/r/myapp/route", `{ "payload": "" }`, http.StatusNotFound, models.ErrRunnerRouteNotFound},
|
||||
{"/r/myapp/route", `{ "payload": "" }`, http.StatusNotFound, models.ErrRoutesNotFound},
|
||||
} {
|
||||
body := bytes.NewBuffer([]byte(test.body))
|
||||
_, rec := routerRequest(t, srv.Router, "POST", test.path, body)
|
||||
|
||||
@@ -10,11 +10,12 @@ import (
|
||||
"net/http"
|
||||
"os"
|
||||
"path"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/Sirupsen/logrus"
|
||||
"github.com/ccirello/supervisor"
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/patrickmn/go-cache"
|
||||
"github.com/spf13/viper"
|
||||
"gitlab-odx.oracle.com/odx/functions/api"
|
||||
"gitlab-odx.oracle.com/odx/functions/api/datastore"
|
||||
@@ -24,7 +25,6 @@ import (
|
||||
"gitlab-odx.oracle.com/odx/functions/api/mqs"
|
||||
"gitlab-odx.oracle.com/odx/functions/api/runner"
|
||||
"gitlab-odx.oracle.com/odx/functions/api/runner/common"
|
||||
"gitlab-odx.oracle.com/odx/functions/api/server/internal/routecache"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -51,8 +51,7 @@ type Server struct {
|
||||
middlewares []Middleware
|
||||
runnerListeners []RunnerListener
|
||||
|
||||
mu sync.Mutex // protects hotroutes
|
||||
hotroutes *routecache.Cache
|
||||
routeCache *cache.Cache
|
||||
singleflight singleflight // singleflight assists Datastore
|
||||
}
|
||||
|
||||
@@ -95,14 +94,14 @@ func New(ctx context.Context, ds models.Datastore, mq models.MessageQueue, logDB
|
||||
}
|
||||
|
||||
s := &Server{
|
||||
Runner: rnr,
|
||||
Router: gin.New(),
|
||||
Datastore: ds,
|
||||
MQ: mq,
|
||||
hotroutes: routecache.New(cacheSize),
|
||||
LogDB: logDB,
|
||||
Enqueue: DefaultEnqueue,
|
||||
apiURL: apiURL,
|
||||
Runner: rnr,
|
||||
Router: gin.New(),
|
||||
Datastore: ds,
|
||||
MQ: mq,
|
||||
routeCache: cache.New(5*time.Second, 5*time.Minute),
|
||||
LogDB: logDB,
|
||||
Enqueue: DefaultEnqueue,
|
||||
apiURL: apiURL,
|
||||
}
|
||||
|
||||
setMachineId()
|
||||
@@ -110,6 +109,9 @@ func New(ctx context.Context, ds models.Datastore, mq models.MessageQueue, logDB
|
||||
s.bindHandlers(ctx)
|
||||
|
||||
for _, opt := range opts {
|
||||
if opt == nil {
|
||||
continue
|
||||
}
|
||||
opt(s)
|
||||
}
|
||||
return s
|
||||
@@ -171,26 +173,19 @@ func DefaultEnqueue(ctx context.Context, mq models.MessageQueue, task *models.Ta
|
||||
return mq.Push(ctx, task)
|
||||
}
|
||||
|
||||
func routeCacheKey(appname, path string) string {
|
||||
return fmt.Sprintf("%s_%s", appname, path)
|
||||
}
|
||||
func (s *Server) cacheget(appname, path string) (*models.Route, bool) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
route, ok := s.hotroutes.Get(appname, path)
|
||||
route, ok := s.routeCache.Get(routeCacheKey(appname, path))
|
||||
if !ok {
|
||||
return nil, false
|
||||
}
|
||||
return route, ok
|
||||
}
|
||||
|
||||
func (s *Server) cacheRefresh(route *models.Route) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.hotroutes.Refresh(route)
|
||||
return route.(*models.Route), ok
|
||||
}
|
||||
|
||||
func (s *Server) cachedelete(appname, path string) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.hotroutes.Delete(appname, path)
|
||||
s.routeCache.Delete(routeCacheKey(appname, path))
|
||||
}
|
||||
|
||||
func (s *Server) handleRunnerRequest(c *gin.Context) {
|
||||
|
||||
@@ -10,13 +10,14 @@ import (
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/gin-gonic/gin"
|
||||
cache "github.com/patrickmn/go-cache"
|
||||
"gitlab-odx.oracle.com/odx/functions/api/datastore"
|
||||
"gitlab-odx.oracle.com/odx/functions/api/models"
|
||||
"gitlab-odx.oracle.com/odx/functions/api/mqs"
|
||||
"gitlab-odx.oracle.com/odx/functions/api/runner"
|
||||
"gitlab-odx.oracle.com/odx/functions/api/server/internal/routecache"
|
||||
)
|
||||
|
||||
var tmpDatastoreTests = "/tmp/func_test_datastore.db"
|
||||
@@ -25,13 +26,13 @@ func testServer(ds models.Datastore, mq models.MessageQueue, logDB models.FnLog,
|
||||
ctx := context.Background()
|
||||
|
||||
s := &Server{
|
||||
Runner: rnr,
|
||||
Router: gin.New(),
|
||||
Datastore: ds,
|
||||
LogDB: nil,
|
||||
MQ: mq,
|
||||
Enqueue: DefaultEnqueue,
|
||||
hotroutes: routecache.New(2),
|
||||
Runner: rnr,
|
||||
Router: gin.New(),
|
||||
Datastore: ds,
|
||||
LogDB: nil,
|
||||
MQ: mq,
|
||||
Enqueue: DefaultEnqueue,
|
||||
routeCache: cache.New(60*time.Second, 5*time.Minute),
|
||||
}
|
||||
|
||||
r := s.Router
|
||||
@@ -102,7 +103,6 @@ func TestFullStack(t *testing.T) {
|
||||
defer rnrcancel()
|
||||
|
||||
srv := testServer(ds, &mqs.Mock{}, logDB, rnr)
|
||||
srv.hotroutes = routecache.New(2)
|
||||
|
||||
for _, test := range []struct {
|
||||
name string
|
||||
@@ -115,13 +115,15 @@ func TestFullStack(t *testing.T) {
|
||||
{"create my app", "POST", "/v1/apps", `{ "app": { "name": "myapp" } }`, http.StatusOK, 0},
|
||||
{"list apps", "GET", "/v1/apps", ``, http.StatusOK, 0},
|
||||
{"get app", "GET", "/v1/apps/myapp", ``, http.StatusOK, 0},
|
||||
{"add myroute", "POST", "/v1/apps/myapp/routes", `{ "route": { "name": "myroute", "path": "/myroute", "image": "funcy/hello" } }`, http.StatusOK, 1},
|
||||
{"add myroute2", "POST", "/v1/apps/myapp/routes", `{ "route": { "name": "myroute2", "path": "/myroute2", "image": "funcy/error" } }`, http.StatusOK, 2},
|
||||
{"get myroute", "GET", "/v1/apps/myapp/routes/myroute", ``, http.StatusOK, 2},
|
||||
{"get myroute2", "GET", "/v1/apps/myapp/routes/myroute2", ``, http.StatusOK, 2},
|
||||
{"get all routes", "GET", "/v1/apps/myapp/routes", ``, http.StatusOK, 2},
|
||||
{"execute myroute", "POST", "/r/myapp/myroute", `{ "name": "Teste" }`, http.StatusOK, 2},
|
||||
// NOTE: cache is lazy, loads when a request comes in for the route, not when added
|
||||
{"add myroute", "POST", "/v1/apps/myapp/routes", `{ "route": { "name": "myroute", "path": "/myroute", "image": "funcy/hello" } }`, http.StatusOK, 0},
|
||||
{"add myroute2", "POST", "/v1/apps/myapp/routes", `{ "route": { "name": "myroute2", "path": "/myroute2", "image": "funcy/error" } }`, http.StatusOK, 0},
|
||||
{"get myroute", "GET", "/v1/apps/myapp/routes/myroute", ``, http.StatusOK, 0},
|
||||
{"get myroute2", "GET", "/v1/apps/myapp/routes/myroute2", ``, http.StatusOK, 0},
|
||||
{"get all routes", "GET", "/v1/apps/myapp/routes", ``, http.StatusOK, 0},
|
||||
{"execute myroute", "POST", "/r/myapp/myroute", `{ "name": "Teste" }`, http.StatusOK, 1},
|
||||
{"execute myroute2", "POST", "/r/myapp/myroute2", `{ "name": "Teste" }`, http.StatusInternalServerError, 2},
|
||||
{"get myroute2", "GET", "/v1/apps/myapp/routes/myroute2", ``, http.StatusOK, 2},
|
||||
{"delete myroute", "DELETE", "/v1/apps/myapp/routes/myroute", ``, http.StatusOK, 1},
|
||||
{"delete app (fail)", "DELETE", "/v1/apps/myapp", ``, http.StatusConflict, 1},
|
||||
{"delete myroute2", "DELETE", "/v1/apps/myapp/routes/myroute2", ``, http.StatusOK, 0},
|
||||
@@ -136,10 +138,10 @@ func TestFullStack(t *testing.T) {
|
||||
t.Errorf("Test \"%s\": Expected status code to be %d but was %d",
|
||||
test.name, test.expectedCode, rec.Code)
|
||||
}
|
||||
if srv.hotroutes.Len() != test.expectedCacheSize {
|
||||
if srv.routeCache.ItemCount() != test.expectedCacheSize {
|
||||
t.Log(buf.String())
|
||||
t.Errorf("Test \"%s\": Expected cache size to be %d but was %d",
|
||||
test.name, test.expectedCacheSize, srv.hotroutes.Len())
|
||||
test.name, test.expectedCacheSize, srv.routeCache.ItemCount())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -4,8 +4,6 @@ package server
|
||||
|
||||
import (
|
||||
"sync"
|
||||
|
||||
"gitlab-odx.oracle.com/odx/functions/api/models"
|
||||
)
|
||||
|
||||
// call is an in-flight or completed do call
|
||||
@@ -16,18 +14,18 @@ type call struct {
|
||||
}
|
||||
|
||||
type singleflight struct {
|
||||
mu sync.Mutex // protects m
|
||||
m map[models.RouteFilter]*call // lazily initialized
|
||||
mu sync.Mutex // protects m
|
||||
m map[interface{}]*call // lazily initialized
|
||||
}
|
||||
|
||||
// do executes and returns the results of the given function, making
|
||||
// sure that only one execution is in-flight for a given key at a
|
||||
// time. If a duplicate comes in, the duplicate caller waits for the
|
||||
// original to complete and receives the same results.
|
||||
func (g *singleflight) do(key models.RouteFilter, fn func() (interface{}, error)) (interface{}, error) {
|
||||
func (g *singleflight) do(key interface{}, fn func() (interface{}, error)) (interface{}, error) {
|
||||
g.mu.Lock()
|
||||
if g.m == nil {
|
||||
g.m = make(map[models.RouteFilter]*call)
|
||||
g.m = make(map[interface{}]*call)
|
||||
}
|
||||
if c, ok := g.m[key]; ok {
|
||||
g.mu.Unlock()
|
||||
|
||||
Reference in New Issue
Block a user