server: Add global LRU for routes with keys being the appname + path (#472)

* Add global lru for routes with keys being the appname + path

* minor comment fixes

* remove duplicate entires from THIRD_PARTY

* Make sure that we lock and unlock on get, refresh and delete on the cache
This commit is contained in:
Seif Lotfy سيف لطفي
2017-01-05 19:08:46 +01:00
committed by C Cirello
parent b8d01817b8
commit 6f8e94629f
11 changed files with 193 additions and 59 deletions

View File

@@ -58,7 +58,8 @@ func TestAppCreate(t *testing.T) {
{&datastore.Mock{}, "/v1/apps", `{ "app": { "name": "teste" } }`, http.StatusOK, nil},
} {
rnr, cancel := testRunner(t)
router := testRouter(test.mock, &mqs.Mock{}, rnr, tasks)
srv := testServer(test.mock, &mqs.Mock{}, rnr, tasks)
router := srv.Router
body := bytes.NewBuffer([]byte(test.body))
_, rec := routerRequest(t, router, "POST", test.path, body)
@@ -102,9 +103,9 @@ func TestAppDelete(t *testing.T) {
}, "/v1/apps/myapp", "", http.StatusOK, nil},
} {
rnr, cancel := testRunner(t)
router := testRouter(test.ds, &mqs.Mock{}, rnr, tasks)
srv := testServer(test.ds, &mqs.Mock{}, rnr, tasks)
_, rec := routerRequest(t, router, "DELETE", test.path, nil)
_, rec := routerRequest(t, srv.Router, "DELETE", test.path, nil)
if rec.Code != test.expectedCode {
t.Log(buf.String())
@@ -132,7 +133,7 @@ func TestAppList(t *testing.T) {
rnr, cancel := testRunner(t)
defer cancel()
router := testRouter(&datastore.Mock{}, &mqs.Mock{}, rnr, tasks)
srv := testServer(&datastore.Mock{}, &mqs.Mock{}, rnr, tasks)
for i, test := range []struct {
path string
@@ -142,7 +143,7 @@ func TestAppList(t *testing.T) {
}{
{"/v1/apps", "", http.StatusOK, nil},
} {
_, rec := routerRequest(t, router, "GET", test.path, nil)
_, rec := routerRequest(t, srv.Router, "GET", test.path, nil)
if rec.Code != test.expectedCode {
t.Log(buf.String())
@@ -169,7 +170,7 @@ func TestAppGet(t *testing.T) {
rnr, cancel := testRunner(t)
defer cancel()
router := testRouter(&datastore.Mock{}, &mqs.Mock{}, rnr, tasks)
srv := testServer(&datastore.Mock{}, &mqs.Mock{}, rnr, tasks)
for i, test := range []struct {
path string
@@ -179,7 +180,7 @@ func TestAppGet(t *testing.T) {
}{
{"/v1/apps/myapp", "", http.StatusNotFound, nil},
} {
_, rec := routerRequest(t, router, "GET", test.path, nil)
_, rec := routerRequest(t, srv.Router, "GET", test.path, nil)
if rec.Code != test.expectedCode {
t.Log(buf.String())
@@ -229,10 +230,10 @@ func TestAppUpdate(t *testing.T) {
}, "/v1/apps/myapp", `{ "app": { "name": "othername" } }`, http.StatusBadRequest, nil},
} {
rnr, cancel := testRunner(t)
router := testRouter(test.mock, &mqs.Mock{}, rnr, tasks)
srv := testServer(test.mock, &mqs.Mock{}, rnr, tasks)
body := bytes.NewBuffer([]byte(test.body))
_, rec := routerRequest(t, router, "PATCH", test.path, body)
_, rec := routerRequest(t, srv.Router, "PATCH", test.path, body)
if rec.Code != test.expectedCode {
t.Log(buf.String())

View File

@@ -0,0 +1,88 @@
// 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"
"github.com/iron-io/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)
}

View File

@@ -100,5 +100,7 @@ func (s *Server) handleRouteCreate(c *gin.Context) {
return
}
s.cacherefresh(route)
c.JSON(http.StatusOK, routeResponse{"Route successfully created", route})
}

View File

@@ -29,5 +29,6 @@ func (s *Server) handleRouteDelete(c *gin.Context) {
return
}
s.cachedelete(appName, routePath)
c.JSON(http.StatusOK, gin.H{"message": "Route deleted"})
}

View File

@@ -37,10 +37,10 @@ func TestRouteCreate(t *testing.T) {
{&datastore.Mock{}, "/v1/apps/a/routes", `{ "route": { "image": "iron/hello", "path": "/myroute" } }`, http.StatusOK, nil},
} {
rnr, cancel := testRunner(t)
router := testRouter(test.mock, &mqs.Mock{}, rnr, tasks)
srv := testServer(test.mock, &mqs.Mock{}, rnr, tasks)
body := bytes.NewBuffer([]byte(test.body))
_, rec := routerRequest(t, router, "POST", test.path, body)
_, rec := routerRequest(t, srv.Router, "POST", test.path, body)
if rec.Code != test.expectedCode {
t.Log(buf.String())
@@ -81,8 +81,8 @@ func TestRouteDelete(t *testing.T) {
}, "/v1/apps/a/routes/myroute", "", http.StatusOK, nil},
} {
rnr, cancel := testRunner(t)
router := testRouter(test.ds, &mqs.Mock{}, rnr, tasks)
_, rec := routerRequest(t, router, "DELETE", test.path, nil)
srv := testServer(test.ds, &mqs.Mock{}, rnr, tasks)
_, rec := routerRequest(t, srv.Router, "DELETE", test.path, nil)
if rec.Code != test.expectedCode {
t.Log(buf.String())
@@ -110,7 +110,7 @@ func TestRouteList(t *testing.T) {
rnr, cancel := testRunner(t)
defer cancel()
router := testRouter(&datastore.Mock{}, &mqs.Mock{}, rnr, tasks)
srv := testServer(&datastore.Mock{}, &mqs.Mock{}, rnr, tasks)
for i, test := range []struct {
path string
@@ -120,7 +120,7 @@ func TestRouteList(t *testing.T) {
}{
{"/v1/apps/a/routes", "", http.StatusOK, nil},
} {
_, rec := routerRequest(t, router, "GET", test.path, nil)
_, rec := routerRequest(t, srv.Router, "GET", test.path, nil)
if rec.Code != test.expectedCode {
t.Log(buf.String())
@@ -148,7 +148,7 @@ func TestRouteGet(t *testing.T) {
rnr, cancel := testRunner(t)
defer cancel()
router := testRouter(&datastore.Mock{}, &mqs.Mock{}, rnr, tasks)
srv := testServer(&datastore.Mock{}, &mqs.Mock{}, rnr, tasks)
for i, test := range []struct {
path string
@@ -158,7 +158,7 @@ func TestRouteGet(t *testing.T) {
}{
{"/v1/apps/a/routes/myroute", "", http.StatusNotFound, nil},
} {
_, rec := routerRequest(t, router, "GET", test.path, nil)
_, rec := routerRequest(t, srv.Router, "GET", test.path, nil)
if rec.Code != test.expectedCode {
t.Log(buf.String())
@@ -215,11 +215,11 @@ func TestRouteUpdate(t *testing.T) {
}, "/v1/apps/a/routes/myroute/do", `{ "route": { "path": "/otherpath" } }`, http.StatusBadRequest, nil},
} {
rnr, cancel := testRunner(t)
router := testRouter(test.ds, &mqs.Mock{}, rnr, tasks)
srv := testServer(test.ds, &mqs.Mock{}, rnr, tasks)
body := bytes.NewBuffer([]byte(test.body))
_, rec := routerRequest(t, router, "PATCH", test.path, body)
_, rec := routerRequest(t, srv.Router, "PATCH", test.path, body)
if rec.Code != test.expectedCode {
t.Log(buf.String())

View File

@@ -62,5 +62,7 @@ func (s *Server) handleRouteUpdate(c *gin.Context) {
return
}
s.cacherefresh(route)
c.JSON(http.StatusOK, routeResponse{"Route successfully updated", route})
}

View File

@@ -130,6 +130,9 @@ func (s *Server) handleRequest(c *gin.Context, enqueue models.Enqueue) {
}
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
}
resp, err := s.singleflight.do(
filter,
func() (interface{}, error) {

View File

@@ -30,7 +30,7 @@ func TestRouteRunnerGet(t *testing.T) {
rnr, cancel := testRunner(t)
defer cancel()
router := testRouter(&datastore.Mock{
srv := testServer(&datastore.Mock{
Apps: []*models.App{
{Name: "myapp", Config: models.Config{}},
},
@@ -46,7 +46,7 @@ func TestRouteRunnerGet(t *testing.T) {
{"/r/app/route", "", http.StatusNotFound, models.ErrAppsNotFound},
{"/r/myapp/route", "", http.StatusNotFound, models.ErrRunnerRouteNotFound},
} {
_, rec := routerRequest(t, router, "GET", test.path, nil)
_, rec := routerRequest(t, srv.Router, "GET", test.path, nil)
if rec.Code != test.expectedCode {
t.Log(buf.String())
@@ -73,7 +73,7 @@ func TestRouteRunnerPost(t *testing.T) {
rnr, cancel := testRunner(t)
defer cancel()
router := testRouter(&datastore.Mock{
srv := testServer(&datastore.Mock{
Apps: []*models.App{
{Name: "myapp", Config: models.Config{}},
},
@@ -90,7 +90,7 @@ func TestRouteRunnerPost(t *testing.T) {
{"/r/myapp/route", `{ "payload": "" }`, http.StatusNotFound, models.ErrRunnerRouteNotFound},
} {
body := bytes.NewBuffer([]byte(test.body))
_, rec := routerRequest(t, router, "POST", test.path, body)
_, rec := routerRequest(t, srv.Router, "POST", test.path, body)
if rec.Code != test.expectedCode {
t.Log(buf.String())
@@ -123,7 +123,7 @@ func TestRouteRunnerExecution(t *testing.T) {
go runner.StartWorkers(ctx, rnr, tasks)
router := testRouter(&datastore.Mock{
srv := testServer(&datastore.Mock{
Apps: []*models.App{
{Name: "myapp", Config: models.Config{}},
},
@@ -148,7 +148,7 @@ func TestRouteRunnerExecution(t *testing.T) {
{"/r/myapp/myerror", ``, "GET", http.StatusInternalServerError, map[string][]string{"X-Function": {"Test"}}},
} {
body := strings.NewReader(test.body)
_, rec := routerRequest(t, router, test.method, test.path, body)
_, rec := routerRequest(t, srv.Router, test.method, test.path, body)
if rec.Code != test.expectedCode {
t.Log(buf.String())
@@ -181,7 +181,7 @@ func TestRouteRunnerTimeout(t *testing.T) {
defer cancelrnr()
go runner.StartWorkers(ctx, rnr, tasks)
router := testRouter(&datastore.Mock{
srv := testServer(&datastore.Mock{
Apps: []*models.App{
{Name: "myapp", Config: models.Config{}},
},
@@ -201,7 +201,7 @@ func TestRouteRunnerTimeout(t *testing.T) {
{"/r/myapp/sleeper", `{"sleep": 2}`, "POST", http.StatusGatewayTimeout, nil},
} {
body := strings.NewReader(test.body)
_, rec := routerRequest(t, router, test.method, test.path, body)
_, rec := routerRequest(t, srv.Router, test.method, test.path, body)
if rec.Code != test.expectedCode {
t.Log(buf.String())

View File

@@ -7,6 +7,7 @@ import (
"io/ioutil"
"net/http"
"path"
"sync"
"github.com/Sirupsen/logrus"
"github.com/ccirello/supervisor"
@@ -15,6 +16,7 @@ import (
"github.com/iron-io/functions/api/models"
"github.com/iron-io/functions/api/runner"
"github.com/iron-io/functions/api/runner/task"
"github.com/iron-io/functions/api/server/internal/routecache"
"github.com/iron-io/runner/common"
)
@@ -41,10 +43,14 @@ type Server struct {
appDeleteListeners []AppDeleteListener
runnerListeners []RunnerListener
mu sync.Mutex // protects hotroutes
hotroutes *routecache.Cache
tasks chan task.Request
singleflight singleflight // singleflight assists Datastore
}
const cacheSize = 1024
func New(ctx context.Context, ds models.Datastore, mq models.MessageQueue, apiURL string, opts ...ServerOption) *Server {
metricLogger := runner.NewMetricLogger()
funcLogger := runner.NewFuncLogger()
@@ -56,12 +62,12 @@ func New(ctx context.Context, ds models.Datastore, mq models.MessageQueue, apiUR
}
tasks := make(chan task.Request)
s := &Server{
Runner: rnr,
Router: gin.New(),
Datastore: ds,
MQ: mq,
hotroutes: routecache.New(cacheSize),
tasks: tasks,
Enqueue: DefaultEnqueue,
apiURL: apiURL,
@@ -72,7 +78,6 @@ func New(ctx context.Context, ds models.Datastore, mq models.MessageQueue, apiUR
for _, opt := range opts {
opt(s)
}
return s
}
@@ -87,7 +92,6 @@ func prepareMiddleware(ctx context.Context) gin.HandlerFunc {
if routePath := c.Param("route"); routePath != "" {
c.Set(api.Path, routePath)
}
c.Set("ctx", ctx)
c.Next()
}
@@ -98,6 +102,28 @@ func DefaultEnqueue(ctx context.Context, mq models.MessageQueue, task *models.Ta
return mq.Push(ctx, task)
}
func (s *Server) cacheget(appname, path string) (*models.Route, bool) {
s.mu.Lock()
defer s.mu.Unlock()
route, ok := s.hotroutes.Get(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)
}
func (s *Server) cachedelete(appname, path string) {
s.mu.Lock()
defer s.mu.Unlock()
s.hotroutes.Delete(appname, path)
}
func (s *Server) handleRunnerRequest(c *gin.Context) {
s.handleRequest(c, s.Enqueue)
}

View File

@@ -17,11 +17,12 @@ import (
"github.com/iron-io/functions/api/mqs"
"github.com/iron-io/functions/api/runner"
"github.com/iron-io/functions/api/runner/task"
"github.com/iron-io/functions/api/server/internal/routecache"
)
var tmpBolt = "/tmp/func_test_bolt.db"
func testRouter(ds models.Datastore, mq models.MessageQueue, rnr *runner.Runner, tasks chan task.Request) *gin.Engine {
func testServer(ds models.Datastore, mq models.MessageQueue, rnr *runner.Runner, tasks chan task.Request) *Server {
ctx := context.Background()
s := &Server{
@@ -31,14 +32,15 @@ func testRouter(ds models.Datastore, mq models.MessageQueue, rnr *runner.Runner,
MQ: mq,
tasks: tasks,
Enqueue: DefaultEnqueue,
hotroutes: routecache.New(2),
}
r := s.Router
r.Use(gin.Logger())
r.Use(prepareMiddleware(ctx))
s.Router.Use(prepareMiddleware(ctx))
s.bindHandlers()
return r
return s
}
func routerRequest(t *testing.T, router *gin.Engine, method, path string, body io.Reader) (*http.Request, *httptest.ResponseRecorder) {
@@ -104,38 +106,45 @@ func TestFullStack(t *testing.T) {
go runner.StartWorkers(ctx, rnr, tasks)
router := testRouter(ds, &mqs.Mock{}, rnr, tasks)
srv := testServer(ds, &mqs.Mock{}, rnr, tasks)
srv.hotroutes = routecache.New(2)
for _, test := range []struct {
name string
method string
path string
body string
expectedCode int
name string
method string
path string
body string
expectedCode int
expectedCacheSize int
}{
{"create my app", "POST", "/v1/apps", `{ "app": { "name": "myapp" } }`, http.StatusOK},
{"list apps", "GET", "/v1/apps", ``, http.StatusOK},
{"get app", "GET", "/v1/apps/myapp", ``, http.StatusOK},
{"add myroute", "POST", "/v1/apps/myapp/routes", `{ "route": { "name": "myroute", "path": "/myroute", "image": "iron/hello" } }`, http.StatusOK},
{"add myroute2", "POST", "/v1/apps/myapp/routes", `{ "route": { "name": "myroute2", "path": "/myroute2", "image": "iron/error" } }`, http.StatusOK},
{"get myroute", "GET", "/v1/apps/myapp/routes/myroute", ``, http.StatusOK},
{"get myroute2", "GET", "/v1/apps/myapp/routes/myroute2", ``, http.StatusOK},
{"get all routes", "GET", "/v1/apps/myapp/routes", ``, http.StatusOK},
{"execute myroute", "POST", "/r/myapp/myroute", `{ "name": "Teste" }`, http.StatusOK},
{"execute myroute2", "POST", "/r/myapp/myroute2", `{ "name": "Teste" }`, http.StatusInternalServerError},
{"delete myroute", "DELETE", "/v1/apps/myapp/routes/myroute", ``, http.StatusOK},
{"delete app (fail)", "DELETE", "/v1/apps/myapp", ``, http.StatusBadRequest},
{"delete myroute2", "DELETE", "/v1/apps/myapp/routes/myroute2", ``, http.StatusOK},
{"delete app (success)", "DELETE", "/v1/apps/myapp", ``, http.StatusOK},
{"get deleted app", "GET", "/v1/apps/myapp", ``, http.StatusNotFound},
{"get delete route on deleted app", "GET", "/v1/apps/myapp/routes/myroute", ``, http.StatusInternalServerError},
{"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": "iron/hello" } }`, http.StatusOK, 1},
{"add myroute2", "POST", "/v1/apps/myapp/routes", `{ "route": { "name": "myroute2", "path": "/myroute2", "image": "iron/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},
{"execute myroute2", "POST", "/r/myapp/myroute2", `{ "name": "Teste" }`, http.StatusInternalServerError, 2},
{"delete myroute", "DELETE", "/v1/apps/myapp/routes/myroute", ``, http.StatusOK, 1},
{"delete app (fail)", "DELETE", "/v1/apps/myapp", ``, http.StatusBadRequest, 1},
{"delete myroute2", "DELETE", "/v1/apps/myapp/routes/myroute2", ``, http.StatusOK, 0},
{"delete app (success)", "DELETE", "/v1/apps/myapp", ``, http.StatusOK, 0},
{"get deleted app", "GET", "/v1/apps/myapp", ``, http.StatusNotFound, 0},
{"get delete route on deleted app", "GET", "/v1/apps/myapp/routes/myroute", ``, http.StatusInternalServerError, 0},
} {
_, rec := routerRequest(t, router, test.method, test.path, bytes.NewBuffer([]byte(test.body)))
_, rec := routerRequest(t, srv.Router, test.method, test.path, bytes.NewBuffer([]byte(test.body)))
if rec.Code != test.expectedCode {
t.Log(buf.String())
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 {
t.Log(buf.String())
t.Errorf("Test \"%s\": Expected cache size to be %d but was %d",
test.name, test.expectedCacheSize, srv.hotroutes.Len())
}
}
}

View File

@@ -12,6 +12,7 @@ import (
"github.com/iron-io/functions/api/mqs"
"github.com/iron-io/functions/api/runner"
"github.com/iron-io/functions/api/runner/task"
"github.com/iron-io/functions/api/server/internal/routecache"
)
type testSpecialHandler struct{}
@@ -44,9 +45,10 @@ func TestSpecialHandlerSet(t *testing.T) {
{Path: "/test", Image: "iron/hello", AppName: "test"},
},
},
MQ: &mqs.Mock{},
tasks: tasks,
Enqueue: DefaultEnqueue,
MQ: &mqs.Mock{},
tasks: tasks,
Enqueue: DefaultEnqueue,
hotroutes: routecache.New(2),
}
router := s.Router