diff --git a/go.mod b/go.mod index 3664d03..8a83e74 100644 --- a/go.mod +++ b/go.mod @@ -1,15 +1,13 @@ module vegvisir -go 1.13 +go 1.16 require ( github.com/andybalholm/brotli v1.0.3 // indirect github.com/go-redis/redis v6.15.9+incompatible - github.com/google/go-cmp v0.5.6 // indirect - github.com/klauspost/compress v1.13.1 // indirect github.com/onsi/ginkgo v1.15.0 // indirect github.com/onsi/gomega v1.10.5 // indirect - github.com/valyala/fasthttp v1.28.0 + github.com/valyala/fasthttp v1.31.0 golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c // indirect gopkg.in/yaml.v2 v2.4.0 // indirect ) diff --git a/go.sum b/go.sum index 35d2356..1a84498 100644 --- a/go.sum +++ b/go.sum @@ -1,7 +1,9 @@ github.com/andybalholm/brotli v1.0.2/go.mod h1:loMXtMfwqflxFJPmdbJO0a3KNoPuLBgiu3qAvBg8x/Y= +github.com/andybalholm/brotli v1.0.3 h1:fpcw+r1N1h0Poc1F/pHbW40cUm/lMEQslZtCkBQ0UnM= github.com/andybalholm/brotli v1.0.3/go.mod h1:fO7iG3H7G2nSZ7m0zPUDn85XEX2GTukHGRSepvi9Eig= github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo= github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ= +github.com/go-redis/redis v6.15.9+incompatible h1:K0pv1D7EQUjfyoMql+r/jZqCLizCGKFlFgcHWWmHQjg= github.com/go-redis/redis v6.15.9+incompatible/go.mod h1:NAIEuMOZ/fxfXJIrKDQDz8wamY7mA7PouImQ2Jvg6kA= github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= github.com/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8= @@ -14,10 +16,9 @@ github.com/golang/snappy v0.0.3/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEW github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= -github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU= -github.com/klauspost/compress v1.12.2/go.mod h1:8dP1Hq4DHOhN9w426knH3Rhby4rFm6D8eO+e+Dq5Gzg= -github.com/klauspost/compress v1.13.1/go.mod h1:8dP1Hq4DHOhN9w426knH3Rhby4rFm6D8eO+e+Dq5Gzg= +github.com/klauspost/compress v1.13.4 h1:0zhec2I8zGnjWcKyLl6i3gPqKANCCn5e9xmviEEeX6s= +github.com/klauspost/compress v1.13.4/go.mod h1:8dP1Hq4DHOhN9w426knH3Rhby4rFm6D8eO+e+Dq5Gzg= github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A= github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE= github.com/onsi/ginkgo v1.12.1/go.mod h1:zj2OWP4+oCPe1qIXoGWkgMRwljMUYCdkwsT2108oapk= @@ -25,8 +26,10 @@ github.com/onsi/ginkgo v1.15.0/go.mod h1:hF8qUzuuC8DJGygJH3726JnCZX4MYbRB8yFfISq github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY= github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo= github.com/onsi/gomega v1.10.5/go.mod h1:gza4q3jKQJijlu05nKWRCW/GavJumGt8aNRxWg7mt48= +github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw= github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc= -github.com/valyala/fasthttp v1.28.0/go.mod h1:cmWIqlu99AO/RKcp1HWaViTqc57FswJOfYYdPJBl8BA= +github.com/valyala/fasthttp v1.31.0 h1:lrauRLII19afgCs2fnWRJ4M5IkV0lo2FqA61uGkNBfE= +github.com/valyala/fasthttp v1.31.0/go.mod h1:2rsYD01CKFrjjsvFxx75KlEUNpWNBY9JWD3K/7o2Cus= github.com/valyala/tcplisten v1.0.0/go.mod h1:T0xQ8SeCZGxckz9qRXTfG43PvQ/mcWh7FwZEA7Ioqkc= github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= diff --git a/pkg/cache/manager.go b/pkg/cache/manager.go new file mode 100644 index 0000000..6ec8d5c --- /dev/null +++ b/pkg/cache/manager.go @@ -0,0 +1,87 @@ +// ___ ____ ___ ___ +// \ \ / / | _ | __| \ \ / / || | __ || || _ | +// \ \/ / |___ | |__ \ \/ / || |___ || ||___| +// \ / | _ | _ | \ / || __ | || ||\\ +// \/ |___ |___ | \/ || ____| || || \\ +// +// Copyright (c) 2021 Piotr Biernat. https://pbiernat.dev. MIT License +// Repo: https://git.pbiernat.dev/golang/vegvisir + +package cache + +import ( + "encoding/json" + "log" +) + +type Manager struct { + datastore CacheDatastore + prefix string + ttl int +} + +func NewManager(datastore CacheDatastore, prefix string, ttl int) Manager { + return Manager{ + datastore: datastore, + prefix: prefix, + ttl: ttl, + } +} + +func (rm *Manager) Save(name string, r interface{}) bool { // FIXME second argument type( interface{} ) + data, err := json.Marshal(r) + if err != nil { + log.Println("JSON:", err) // FIXME + + return false + } + + name = rm.prefix + name + err = rm.datastore.SetKey(name, string(data), rm.ttl) + if err != nil { + log.Println("Error saving cache item:", name) // FIXME + + return false + } + + return true +} + +func (rm *Manager) Load(name string, output interface{}) (bool, interface{}) { // FIXME second return type( interface{} ) + name = rm.prefix + name + data, err := rm.datastore.GetKey(name) + if err != nil { + log.Println("Cache item:", name, "not found") + + return false, nil + } + + var res interface{} + switch t := output.(type) { + default: + log.Printf("Unsupported cache type: %T", t) + return false, nil + case *ResponseCache: + res = &ResponseCache{} + rm.convertType(data, res) + + return true, res.(*ResponseCache) + case *RouteCache: + res = &RouteCache{} + rm.convertType(data, res) + + return true, res.(*RouteCache) + } +} + +func (rm *Manager) convertType(data interface{}, output interface{}) interface{} { + err := json.Unmarshal([]byte(data.(string)), &output) + + if err != nil { + log.Println("Converting error: ", err) // FIXME + + return nil + } + + return output +} diff --git a/pkg/cache/memo.go b/pkg/cache/memo.go new file mode 100644 index 0000000..8e239a0 --- /dev/null +++ b/pkg/cache/memo.go @@ -0,0 +1,84 @@ +// Copyright © 2016 Alan A. A. Donovan & Brian W. Kernighan. +// License: https://creativecommons.org/licenses/by-nc-sa/4.0/ + +// See page 278. + +// Package memo provides a concurrency-safe non-blocking memoization +// of a function. Requests for different keys proceed in parallel. +// Concurrent requests for the same url block until the first completes. +// This implementation uses a monitor goroutine. + +package cache + +// Func is the type of the function to memoize. +type Func func(url, method string, route *RouteCache) (error, *ResponseCache) + +// A result is the result of calling a Func. +type result struct { + value *ResponseCache + err error +} + +type entry struct { + res result + ready chan struct{} // closed when res is ready +} + +// A request is a message requesting that the Func be applied to url. +type request struct { + url string + method string + route *RouteCache + response chan<- result // the client wants a single result +} + +type Memo struct { + requests chan request +} + +// New returns a memoization of f. Clients must subsequently call Close. +func New(f Func) *Memo { + memo := &Memo{requests: make(chan request)} + go memo.server(f) + return memo +} + +func (memo *Memo) Read(url, method string, route *RouteCache) (*ResponseCache, error) { + response := make(chan result) + memo.requests <- request{url, method, route, response} + res := <-response + return res.value, res.err +} + +func (memo *Memo) Close() { close(memo.requests) } + +func (memo *Memo) server(f Func) { + cache := make(map[string]*entry) + for req := range memo.requests { + key := req.method + "_" + req.url + e := cache[key] + if e == nil { + // This is the first request for this url. + e = &entry{ready: make(chan struct{})} + cache[key] = e + //log.Println("e.call:", req.url, req.method, req.route) + go e.call(f, req.url, req.method, req.route) + } + //log.Println("e.deliver:", req.response) + go e.deliver(req.response) + } +} + +func (e *entry) call(f Func, url, method string, route *RouteCache) { + // Evaluate the function. + e.res.err, e.res.value = f(url, method, route) + // Broadcast the ready condition. + close(e.ready) +} + +func (e *entry) deliver(response chan<- result) { + // Wait for the ready condition. + <-e.ready + // Send the result to the client. + response <- e.res +} diff --git a/pkg/cache/response.go b/pkg/cache/response.go index c213157..cfaa993 100644 --- a/pkg/cache/response.go +++ b/pkg/cache/response.go @@ -9,57 +9,24 @@ package cache -import ( - "encoding/json" - "log" -) +import "github.com/valyala/fasthttp" type ResponseCache struct { - URL string - Body string - // Headers map[string]string + URL string + Method string + Body []byte + Code int + Headers *fasthttp.ResponseHeader } -type ResponseCacheManager struct { - datastore CacheDatastore - prefix string - ttl int +func NewResponseCache(url, method string, body []byte, code int, headers *fasthttp.ResponseHeader) *ResponseCache { + return &ResponseCache{url, method, body, code, headers} } -func NewResponseCacheManager(datastore CacheDatastore, ttl int) ResponseCacheManager { - return ResponseCacheManager{ - datastore: datastore, - prefix: "response_", - ttl: ttl, - } -} - -func (rm *ResponseCacheManager) Save(name string, r ResponseCache) { - data, err := json.Marshal(r) - if err != nil { - log.Println("JSON:", err) // FIXME - } - - name = rm.prefix + name - err = rm.datastore.SetKey(name, string(data), rm.ttl) - if err != nil { - log.Println("REDIS:", err, name) // FIXME - } -} - -func (rm *ResponseCacheManager) Load(name string) (bool, *ResponseCache) { - name = rm.prefix + name - - data, err := rm.datastore.GetKey(name) - if err != nil { - return false, &ResponseCache{} - } - - rc := &ResponseCache{} - err = json.Unmarshal([]byte(data.(string)), &rc) - if err != nil { - log.Println("JSON:", err) // FIXME - } - - return true, rc -} +//func (r *ResponseCache) StatusCode() int { +// return r.Code +//} +// +//func (r *ResponseCache) Body() []byte { +// return r.Body +//} diff --git a/pkg/cache/route.go b/pkg/cache/route.go index ff28a1c..4e470b9 100644 --- a/pkg/cache/route.go +++ b/pkg/cache/route.go @@ -9,55 +9,14 @@ package cache -import ( - "encoding/json" - "log" -) - type RouteCache struct { SourceUrl string TargetUrl string } -type RouteCacheManager struct { - datastore CacheDatastore - ttl int -} - -func NewRouteCacheManager(datastore CacheDatastore, ttl int) RouteCacheManager { - return RouteCacheManager{ - datastore: datastore, - ttl: ttl, +func NewRouteCache(source, target string) *RouteCache { + return &RouteCache{ + SourceUrl: source, + TargetUrl: target, } } - -func (rm *RouteCacheManager) Save(name string, r RouteCache) { - data, err := json.Marshal(r) - if err != nil { - log.Println("JSON:", err) // FIXME - } - - err = rm.datastore.SetKey("route_"+name, data, rm.ttl) - if err != nil { - log.Println("REDIS:", err, name) // FIXME - } -} - -func (rm *RouteCacheManager) Load(name string) (bool, RouteCache) { - name = "route_" + name - - data, err := rm.datastore.GetKey(name) - if err != nil { - log.Println("REDIS:", err, name) // FIXME - - return false, RouteCache{} - } - - rc := RouteCache{} - err = json.Unmarshal([]byte(data.(string)), &rc) - if err != nil { - log.Println("JSON:", err) // FIXME - } - - return true, rc -} diff --git a/pkg/main.go b/pkg/main.go index 27e15bf..a87f1ea 100644 --- a/pkg/main.go +++ b/pkg/main.go @@ -12,11 +12,13 @@ package main import ( "flag" "log" + "net/http" "os" "runtime" "runtime/pprof" "vegvisir/pkg/server" ) +import _ "net/http/pprof" var ( cFile = flag.String("c", "vegvisir.json", "Path to config file") @@ -27,6 +29,9 @@ var ( ) func main() { + go func() { + log.Println(http.ListenAndServe("localhost:6060", nil)) + }() flag.Parse() // cpu profiling diff --git a/pkg/server/server.go b/pkg/server/server.go index ee9f769..a29e9fa 100644 --- a/pkg/server/server.go +++ b/pkg/server/server.go @@ -32,10 +32,10 @@ const ( type Server struct { Config config.Config - cFilePath string - rCache map[string]cache.RouteCache // Internal route cache - routeCM cache.RouteCacheManager // Redis route cache - respCM cache.ResponseCacheManager // Redis response cache + cFilePath string // Path to config file + rCache map[string]*cache.RouteCache // Internal route cache + routeCM cache.Manager // Redis route cache + respCM cache.Manager // Redis response cache } func NewServer(cPath string) *Server { @@ -43,9 +43,9 @@ func NewServer(cPath string) *Server { return &Server{ cFilePath: cPath, - rCache: make(map[string]cache.RouteCache), - routeCM: cache.NewRouteCacheManager(datastore, 30), //FIXME use ttl(seconds) from config or env... - respCM: cache.NewResponseCacheManager(datastore, 30), //FIXME use ttl(seconds) from config or env... + rCache: make(map[string]*cache.RouteCache), + routeCM: cache.NewManager(datastore, "prefix_", 5), //FIXME use ttl(seconds) from config or env... + respCM: cache.NewManager(datastore, "response_", 5), //FIXME use ttl(seconds) from config or env... } } @@ -86,25 +86,75 @@ func (s *Server) mainHandler(ctx *fasthttp.RequestCtx) { ctx.Response.Header.Add(fasthttp.HeaderServer, Name) // move all below logic to concrete handler or sth.... - reqUri, sReqUri, sReqMethod := ctx.RequestURI(), string(ctx.RequestURI()), string(ctx.Method()) - log.Println("Incomming request:", sReqMethod, sReqUri) + reqUrl, sReqUrl, sReqMethod := ctx.RequestURI(), string(ctx.RequestURI()), string(ctx.Method()) + // log.Println("Incoming request:", sReqMethod, sReqUrl) - found, route := s.findRouteByRequestURI(reqUri) + found, route := s.findRouteByRequestUrl(reqUrl) if !found { - // FIXME: return 404/5xx error in repsonse, maybe define it in Backend config? + // FIXME: return 404 or 5xx error in response? Maybe define it in concrete Backend config? ctx.SetStatusCode(fasthttp.StatusNotFound) return } + memo := cache.New(s.processUrl) + defer memo.Close() + response, err := memo.Read(sReqUrl, sReqMethod, route) + //err, response := s.processUrl(sReqUrl, sReqMethod, route) + if err != nil { + // FIXME: Response read error(sending 500 error response) + ctx.SetStatusCode(fasthttp.StatusInternalServerError) + log.Println("Response read error(sending 500 error response)", err) + return + } + + ctx.SetStatusCode(response.Code) + ctx.SetBody(response.Body) + ctx.Response.Header.SetBytesV(fasthttp.HeaderContentType, response.Headers.ContentType()) +} + +func (s *Server) findRouteByRequestUrl(url []byte) (bool, *cache.RouteCache) { + var sUrl = string(url) + + for bId := range s.Config.Backends { + bck := s.Config.Backends[bId] + if !strings.Contains(sUrl, bck.PrefixUrl) { + continue + } + + for rId := range bck.Routes { + route := &bck.Routes[rId] + + if cRoute, ok := s.rCache[sUrl]; ok { + return true, cRoute + } + + rgxp := regexp.MustCompile(fmt.Sprintf("%s%s", bck.PrefixUrl, route.Pattern)) + if rgxp.Match(url) { + targetUrl := bck.BackendAddress + rgxp.ReplaceAllString(sUrl, route.Target) + + cRoute := cache.NewRouteCache(sUrl, targetUrl) + + // s.rCacheManager.Save(sUrl, cRoute) + s.rCache[sUrl] = cRoute + + return true, cRoute + } + } + } + + return false, &cache.RouteCache{} +} + +func (s *Server) processUrl(url, method string, route *cache.RouteCache) (error, *cache.ResponseCache) { // handle response caching - if ok, data := s.respCM.Load(sReqUri); ok { - log.Println("Read resp from cache: ", route.TargetUrl) + cacheKey := method + "_" + url + if ok, data := s.respCM.Load(cacheKey, &cache.ResponseCache{}); ok { + //log.Println("Read resp from cache: ", route.TargetUrl, url) - ctx.SetBody([]byte(data.Body)) + return nil, data.(*cache.ResponseCache) } else { - log.Println("Send req to backend url: ", route.TargetUrl) - - // prepare to send request to backend - separate + //log.Println("Send req to backend url: ", route.TargetUrl, url) + start := time.Now() bckReq := fasthttp.AcquireRequest() bckResp := fasthttp.AcquireResponse() defer fasthttp.ReleaseRequest(bckReq) @@ -112,65 +162,19 @@ func (s *Server) mainHandler(ctx *fasthttp.RequestCtx) { // copy headers from backend response and prepare request for backend - separate bckReq.SetRequestURI(route.TargetUrl) - bckReq.Header.SetMethod(sReqMethod) + bckReq.Header.SetMethod(method) err := fasthttp.Do(bckReq, bckResp) if err != nil { - ctx.SetStatusCode(fasthttp.StatusInternalServerError) - return + return err, nil } - ctx.Response.Header.SetBytesV(fasthttp.HeaderContentType, bckResp.Header.ContentType()) - ctx.SetStatusCode(bckResp.StatusCode()) - ctx.SetBody(bckResp.Body()) - + body, code := bckResp.Body(), bckResp.StatusCode() + log.Printf("%s, %s, %d bytes\n", url, time.Since(start), len(body)) // save response to cache - respCache := cache.ResponseCache{ - URL: sReqUri, - Body: string(bckResp.Body()), - // Headers: [] - } // FIXME: prepare resp cache struct in respCM.Save method or other service... - s.respCM.Save(sReqUri, respCache) + respCache := cache.NewResponseCache(url, method, body, code, &bckResp.Header) + s.respCM.Save(cacheKey, respCache) + + return nil, respCache } } - -func (s *Server) findRouteByRequestURI(uri []byte) (bool, cache.RouteCache) { - var sUri string = string(uri) - - for bId := range s.Config.Backends { - bck := s.Config.Backends[bId] - if !strings.Contains(sUri, bck.PrefixUrl) { - continue - } - - for rId := range bck.Routes { - route := &bck.Routes[rId] - - // if ok, cRoute := s.rCacheManager.Load(sUri); ok { - // return true, cRoute - // } - if cRoute, ok := s.rCache[sUri]; ok { - return true, cRoute - } - - rgxp := regexp.MustCompile(fmt.Sprintf("%s%s", bck.PrefixUrl, route.Pattern)) - if rgxp.Match(uri) { - targetUrl := bck.BackendAddress + rgxp.ReplaceAllString(sUri, route.Target) - - cRoute := cache.RouteCache{ // FIXME: data duplication and use short alias for backend and route! - // Backend: bck, - // Route: *route, - SourceUrl: sUri, - TargetUrl: targetUrl, - } - - // s.rCacheManager.Save(sUri, cRoute) - s.rCache[sUri] = cRoute - - return true, cRoute - } - } - } - - return false, cache.RouteCache{} -} diff --git a/vegvisir.json b/vegvisir.json deleted file mode 100644 index 7bd5829..0000000 --- a/vegvisir.json +++ /dev/null @@ -1,39 +0,0 @@ -{ - "server": { - "address": "127.0.0.1", - "port": 8080 - }, - "backends": { - "news-app": { - "prefixUrl": "news/", - "backendAddress": "http://172.17.0.1:3030", - "routes": [{ - "pattern": "new/(.*)/([0-9]*)", - "target": "new-url/$1/$2" - },{ - "pattern": "new/(.*)", - "target": "$1" - },{ - "pattern": "(.+)", - "target": "url" - }] - }, - "article-app": { - "prefixUrl": "art/", - "backendAddress": "http://172.17.0.1:3030", - "routes": [{ - "pattern": "art1/(.*)", - "target": "art-url/$1" - },{ - "pattern": "([a-zA-Z0-9]*)", - "target": "art-global/$1" - }] - } - }, - "cache": { - "ttl": 300, - "route_ttl": 300, - "response_ttl": 300, - "headers_ttl": 300 - } -} \ No newline at end of file diff --git a/vegvisir.json.dist b/vegvisir.json.dist index 8394cc5..8db1af1 100644 --- a/vegvisir.json.dist +++ b/vegvisir.json.dist @@ -6,28 +6,28 @@ "backends": { "news-app": { "prefixUrl": "news/", - "backendAddress": "http://127.0.0.1:3030", + "backendAddress": "http://172.17.0.1:3030/api/news/", "routes": [{ - "pattern": "new/(.*)", - "target": "$1" + "pattern": "archive/([0-9]*)", + "target": "archive/$1" },{ - "pattern": "new1/(.*)/([0-9]*)", - "target": "new-url1/$1/$2" - },{ - "pattern": "(.+)", - "target": "url" + "pattern": "(.*)", + "target": "new/$1" }] }, "article-app": { - "prefixUrl": "art/", - "backendAddress": "http://127.0.0.1:3030", + "prefixUrl": "article/", + "backendAddress": "http://172.17.0.1:3030/api/articles/", "routes": [{ - "pattern": "art1/(.*)", - "target": "art-url1/$1" - },{ - "pattern": "([a-zA-Z0-9]*)", - "target": "article-global/$1" + "pattern": "(.*)", + "target": "$1" }] } + }, + "cache": { + "ttl": 300, + "route_ttl": 300, + "response_ttl": 300, + "headers_ttl": 300 } } \ No newline at end of file