From a64aad38f1adaab2e86e45b9d4800e67cbe0cdaa Mon Sep 17 00:00:00 2001 From: Jan Halfar Date: Mon, 27 Sep 2021 16:05:59 +0200 Subject: [PATCH] initial draft --- .gitignore | 3 ++ go.mod | 14 ++++++++ go.sum | 55 +++++++++++++++++++++++++++++ main.go | 29 ++++++++++++++++ proxy/cache.go | 61 ++++++++++++++++++++++++++++++++ proxy/jobs.go | 41 ++++++++++++++++++++++ proxy/proxy.go | 84 +++++++++++++++++++++++++++++++++++++++++++++ proxy/proxy_test.go | 83 ++++++++++++++++++++++++++++++++++++++++++++ 8 files changed, 370 insertions(+) create mode 100644 .gitignore create mode 100644 go.mod create mode 100644 go.sum create mode 100644 main.go create mode 100644 proxy/cache.go create mode 100644 proxy/jobs.go create mode 100644 proxy/proxy.go create mode 100644 proxy/proxy_test.go diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..01c3a9a --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +.* +!.git* + diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..9802cd4 --- /dev/null +++ b/go.mod @@ -0,0 +1,14 @@ +module github.com/foomo/contentfulproxy + +go 1.17 + +require github.com/stretchr/testify v1.7.0 + +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 + gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..5d87e1a --- /dev/null +++ b/go.sum @@ -0,0 +1,55 @@ +github.com/benbjohnson/clock v1.1.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA= +github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= +github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= +github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= +github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY= +github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k= +go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw= +go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc= +go.uber.org/goleak v1.1.11-0.20210813005559-691160354723/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ= +go.uber.org/multierr v1.6.0 h1:y6IPFStTAIT5Ytl7/XYmHvzXQ7S3g/IeZW9hyZ5thw4= +go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU= +go.uber.org/zap v1.19.1 h1:ue41HOKd1vGURxrmeKIgELGb3jPW9DMUDGtsinblHwI= +go.uber.org/zap v1.19.1/go.mod h1:j3DNczoxDZroyBnOT1L/Q79cfUMGZxlv/9dzN7SM1rI= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= +golang.org/x/lint v0.0.0-20190930215403-16217165b5de/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc= +golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20190311212946-11955173bddd/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.1.5/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY= +gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/main.go b/main.go new file mode 100644 index 0000000..36c4183 --- /dev/null +++ b/main.go @@ -0,0 +1,29 @@ +package main + +import ( + "context" + "flag" + "log" + "net/http" + + "go.uber.org/zap" + + "github.com/foomo/contentfulproxy/proxy" +) + +func main() { + flagAddr := flag.String("addr", ":80", "address to listen to") + flag.Parse() + l, err := zap.NewProduction() + if err != nil { + log.Fatal("could not initialize zap logger", err) + } + defer l.Sync() + args := flag.Args() + if len(args) != 1 { + l.Error("unexpected number of args - must be exactly one for backendURL") + } + p := proxy.NewProxy(context.Background(), l, args[0]) + 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))) +} diff --git a/proxy/cache.go b/proxy/cache.go new file mode 100644 index 0000000..1e114cf --- /dev/null +++ b/proxy/cache.go @@ -0,0 +1,61 @@ +package proxy + +import ( + "io/ioutil" + "net/http" + "sort" + "strings" + "sync" +) + +type cacheID string + +type cachedResponse struct { + header http.Header + response []byte +} +type cacheMap map[cacheID]*cachedResponse + +type cache struct { + sync.RWMutex + cacheMap cacheMap +} + +func (c *cache) set(id cacheID, response *http.Response) (*cachedResponse, error) { + responseBytes, err := ioutil.ReadAll(response.Body) + if err != nil { + return nil, err + } + response.Body.Close() + c.Lock() + defer c.Unlock() + cr := &cachedResponse{ + header: response.Header, + response: responseBytes, + } + c.cacheMap[id] = cr + return cr, nil +} + +func (c *cache) get(id cacheID) (*cachedResponse, bool) { + c.RLock() + defer c.RUnlock() + response, ok := c.cacheMap[id] + return response, ok +} + +func getCacheIDForRequest(r *http.Request) cacheID { + id := r.URL.RequestURI() + keys := make([]string, len(r.Header)) + i := 0 + for k := range r.Header { + keys[i] = k + i++ + } + sort.Strings(keys) + for _, k := range keys { + id += k + strings.Join(r.Header[k], "-") + } + // hash it here maybe, to keep it shorter + return cacheID(id) +} diff --git a/proxy/jobs.go b/proxy/jobs.go new file mode 100644 index 0000000..76bcdc1 --- /dev/null +++ b/proxy/jobs.go @@ -0,0 +1,41 @@ +package proxy + +import "net/http" + +type requestJobDone struct { + cachedResponse *cachedResponse + err error + id cacheID +} + +type requestJob struct { + request *http.Request + chanDone chan requestJobDone +} + +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()) + if err != nil { + chanJobDone <- requestJobDone{ + id: id, + err: err, + } + return + } + cachedResponse, err := c.set(id, resp) + if err != nil { + chanJobDone <- requestJobDone{ + id: id, + err: err, + } + return + } + chanJobDone <- requestJobDone{ + id: id, + cachedResponse: cachedResponse, + } + } +} diff --git a/proxy/proxy.go b/proxy/proxy.go new file mode 100644 index 0000000..0df4052 --- /dev/null +++ b/proxy/proxy.go @@ -0,0 +1,84 @@ +package proxy + +import ( + "context" + "net/http" + + "go.uber.org/zap" +) + +type Proxy struct { + cache *cache + backendURL string + chanRequestJob chan requestJob + l *zap.Logger +} + +func NewProxy(ctx context.Context, l *zap.Logger, backendURL string) *Proxy { + chanRequest := make(chan requestJob) + c := &cache{ + cacheMap: cacheMap{}, + } + go getLoop(ctx, l, backendURL, c, chanRequest) + return &Proxy{ + l: l, + cache: c, + backendURL: backendURL, + chanRequestJob: chanRequest, + } +} + +func getLoop(ctx context.Context, l *zap.Logger, backendURL string, c *cache, chanRequestJob chan requestJob) { + pendingRequests := map[cacheID][]chan requestJobDone{} + chanJobDone := make(chan requestJobDone) + jobRunner := getJobRunner(c, backendURL, chanJobDone) + for { + select { + case nextJob := <-chanRequestJob: + id := getCacheIDForRequest(nextJob.request) + pendingRequests[id] = append(pendingRequests[id], nextJob.chanDone) + requests := pendingRequests[id] + if len(requests) == 1 { + l.Info("starting jobrunner for", zap.String("uri", nextJob.request.RequestURI), zap.String("id", string(id))) + go jobRunner(nextJob, id) + } + case jobDone := <-chanJobDone: + l.Info("request complete", zap.String("id", string(jobDone.id)), zap.Int("num-waiting-clients", len(pendingRequests[jobDone.id]))) + for _, chanPending := range pendingRequests[jobDone.id] { + chanPending <- jobDone + } + delete(pendingRequests, jobDone.id) + case <-ctx.Done(): + return + } + } +} + +func (p *Proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodGet: + cacheID := getCacheIDForRequest(r) + cachedResponse, ok := p.cache.get(cacheID) + if !ok { + chanDone := make(chan requestJobDone) + p.chanRequestJob <- requestJob{ + request: r, + chanDone: chanDone, + } + jobDone := <-chanDone + if jobDone.err != nil { + http.Error(w, "cache / job error", http.StatusInternalServerError) + return + } + cachedResponse = jobDone.cachedResponse + } + for key, values := range cachedResponse.header { + for _, value := range values { + w.Header().Set(key, value) + } + } + w.Write(cachedResponse.response) + default: + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + } +} diff --git a/proxy/proxy_test.go b/proxy/proxy_test.go new file mode 100644 index 0000000..4189c66 --- /dev/null +++ b/proxy/proxy_test.go @@ -0,0 +1,83 @@ +package proxy + +import ( + "context" + "io/ioutil" + "net/http" + "net/http/httptest" + "sync" + "testing" + + "github.com/stretchr/testify/assert" +) + +const ( + responseFoo = `i am a foo response` + responseBar = `i am bar` +) + +type getStats func(path string) int + +func GetBackend() (getStats, http.HandlerFunc) { + stats := map[string]int{} + statLock := sync.RWMutex{} + return func(path string) int { + statLock.RLock() + defer statLock.RUnlock() + count, ok := stats[path] + if !ok { + return -1 + } + return count + }, func(w http.ResponseWriter, r *http.Request) { + statLock.Lock() + stats[r.URL.Path]++ + statLock.Unlock() + + switch r.URL.Path { + case "/foo": + w.Write([]byte(responseFoo)) + return + case "/bar": + w.Write([]byte(responseBar)) + return + } + http.Error(w, "not found", http.StatusNotFound) + } +} + +func getTestServer(t *testing.T) (gs func(path string) int, s *httptest.Server) { + gs, backendHandler := GetBackend() + + p := NewProxy(context.Background(), httptest.NewServer(backendHandler).URL) + s = httptest.NewServer(p) + t.Log("we have a proxy in front of it running on", s.URL) + return gs, s + +} + +func TestProxy(t *testing.T) { + gs, server := getTestServer(t) + + get := func(path string) string { + resp, err := http.Get(server.URL + "/foo") + assert.NoError(t, err) + defer resp.Body.Close() + responseBytes, err := ioutil.ReadAll(resp.Body) + assert.NoError(t, err) + return string(responseBytes) + } + for j := 0; j < 10; j++ { + wg := sync.WaitGroup{} + for i := 0; i < 128; i++ { + wg.Add(1) + go func() { + get("/foo") + wg.Done() + }() + } + wg.Wait() + } + assert.Equal(t, 1, gs("/foo")) + +}