foomo/contentfulproxy
1
0
Fork 0
mirror of https://github.com/foomo/contentfulproxy.git synced 2026-08-12 12:30:20 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

ECOM-2313 feat:

Browse Source 
- add cache flush webhook
- add callback to external webhooks
This commit is contained in:
Daniel Thomas
2021-09-30 08:16:26 +02:00
parent a64aad38f1
commit d72c8fee26
5 changed files with 91 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go.mod
+6 -2
View File
@@ -2,13 +2,17 @@ module github.com/foomo/contentfulproxy
go 1.17
require github.com/stretchr/testify v1.7.0
require (
github.com/stretchr/testify v1.7.0
go.uber.org/zap v1.19.1
)
require (
github.com/davecgh/go-spew v1.1.1 // indirect
github.com/pmezard/go-difflib v1.0.0 // indirect
go.uber.org/atomic v1.7.0 // indirect
go.uber.org/multierr v1.6.0 // indirect
go.uber.org/zap v1.19.1 // indirect
golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
github.com/foomo/keel v0.2.9
)
main.go
+10 -2
View File
@@ -12,7 +12,7 @@ import (
)
func main() {
flagAddr := flag.String("addr", ":80", "address to listen to")
flagAddr := flag.String("addr", ":8888", "address to listen to")
flag.Parse()
l, err := zap.NewProduction()
if err != nil {
@@ -23,7 +23,15 @@ func main() {
if len(args) != 1 {
l.Error("unexpected number of args - must be exactly one for backendURL")
}
p := proxy.NewProxy(context.Background(), l, args[0])
p := proxy.NewProxy(
context.Background(),
l,
args[0],
[]proxy.WebHookURL{
"https://www.bestbytes.com",
"https://www.spiegel.de",
},
)
l.Info("starting proxy for", zap.String("backendURL", args[0]), zap.String("addr", *flagAddr))
l.Error("http listen failed", zap.Error(http.ListenAndServe(*flagAddr, p)))
}
proxy/cache.go
+25 -1
View File
@@ -1,6 +1,7 @@
package proxy
import (
"go.uber.org/zap"
"io/ioutil"
"net/http"
"sort"
@@ -10,6 +11,8 @@ import (
type cacheID string
type requestFlush string
type cachedResponse struct {
header http.Header
response []byte
@@ -19,6 +22,8 @@ type cacheMap map[cacheID]*cachedResponse
type cache struct {
sync.RWMutex
cacheMap cacheMap
webHooks []WebHookURL
l *zap.Logger
}
func (c *cache) set(id cacheID, response *http.Response) (*cachedResponse, error) {
@@ -26,7 +31,10 @@ func (c *cache) set(id cacheID, response *http.Response) (*cachedResponse, error
if err != nil {
return nil, err
}
response.Body.Close()
err = response.Body.Close()
if err != nil {
return nil, err
}
c.Lock()
defer c.Unlock()
cr := &cachedResponse{
@@ -44,6 +52,22 @@ func (c *cache) get(id cacheID) (*cachedResponse, bool) {
return response, ok
}
func (c *cache) flush() {
c.RLock()
defer c.RUnlock()
c.cacheMap = cacheMap{}
}
func (c *cache) callWebHooks() {
for _, url := range c.webHooks {
c.l.Info("call webhook", zap.String("url", string(url)))
_, err := http.Get(string(url))
if err != nil {
c.l.Error("could not call webhook", zap.String("url", string(url)))
}
}
}
func getCacheIDForRequest(r *http.Request) cacheID {
id := r.URL.RequestURI()
keys := make([]string, len(r.Header))
proxy/jobs.go
+13 -1
View File
@@ -17,7 +17,19 @@ type jobRunner func(job requestJob, id cacheID)
func getJobRunner(c *cache, backendURL string, chanJobDone chan requestJobDone) jobRunner {
return func(job requestJob, id cacheID) {
resp, err := http.Get(backendURL + job.request.URL.RequestURI())
req, err := http.NewRequest("GET", backendURL+job.request.URL.RequestURI(), nil)
if err != nil {
chanJobDone <- requestJobDone{
id: id,
err: err,
}
return
}
for k, v := range job.request.Header {
req.Header.Set(k, string(v[0]))
}
client := http.Client{}
resp, err := client.Do(req)
if err != nil {
chanJobDone <- requestJobDone{
id: id,
proxy/proxy.go
+37 -5
View File
@@ -7,33 +7,51 @@ import (
"go.uber.org/zap"
)
type WebHookURL string
type Proxy struct {
l *zap.Logger
cache *cache
backendURL string
chanRequestJob chan requestJob
l *zap.Logger
chanFlushJob chan requestFlush
}
func NewProxy(ctx context.Context, l *zap.Logger, backendURL string) *Proxy {
func NewProxy(ctx context.Context, l *zap.Logger, backendURL string, webHooks []WebHookURL) *Proxy {
chanRequest := make(chan requestJob)
chanFlush := make(chan requestFlush)
c := &cache{
cacheMap: cacheMap{},
webHooks: webHooks,
l: l,
}
go getLoop(ctx, l, backendURL, c, chanRequest)
go getLoop(ctx, l, backendURL, c, chanRequest, chanFlush)
return &Proxy{
l: l,
cache: c,
backendURL: backendURL,
chanRequestJob: chanRequest,
chanFlushJob: chanFlush,
}
}
func getLoop(ctx context.Context, l *zap.Logger, backendURL string, c *cache, chanRequestJob chan requestJob) {
func getLoop(
ctx context.Context,
l *zap.Logger,
backendURL string,
c *cache,
chanRequestJob chan requestJob,
chanFlush chan requestFlush,
) {
pendingRequests := map[cacheID][]chan requestJobDone{}
chanJobDone := make(chan requestJobDone)
jobRunner := getJobRunner(c, backendURL, chanJobDone)
for {
select {
case command := <-chanFlush:
c.flush()
l.Info("cache flush command coming in", zap.String("flushCommand", string(command)))
c.callWebHooks()
case nextJob := <-chanRequestJob:
id := getCacheIDForRequest(nextJob.request)
pendingRequests[id] = append(pendingRequests[id], nextJob.chanDone)
@@ -55,8 +73,16 @@ func getLoop(ctx context.Context, l *zap.Logger, backendURL string, c *cache, ch
}
func (p *Proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {
case "/api/flush":
command := requestFlush("doit")
p.chanFlushJob <- command
return
}
switch r.Method {
case http.MethodGet:
p.l.Info("serve get request", zap.String("url", r.RequestURI))
cacheID := getCacheIDForRequest(r)
cachedResponse, ok := p.cache.get(cacheID)
if !ok {
@@ -71,13 +97,19 @@ func (p *Proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
return
}
cachedResponse = jobDone.cachedResponse
p.l.Info("serve response after cache creation", zap.String("url", r.RequestURI))
} else {
p.l.Info("serve response from cache", zap.String("url", r.RequestURI))
}
for key, values := range cachedResponse.header {
for _, value := range values {
w.Header().Set(key, value)
}
}
w.Write(cachedResponse.response)
_, err := w.Write(cachedResponse.response)
if err != nil {
p.l.Info("writing cached response failed", zap.String("url", r.RequestURI))
}
default:
http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
}