Merge pull request #10 from C4T-BuT-S4D/cache_proxy

Cache proxy implementation

docker-compose.yml

@@ -53,6 +53,16 @@ services:
       retries: 5
       start_period: 30s
 
+  cacheproxy:
+    build: httpproxy
+    restart: unless-stopped
+    environment:
+      REDIS_URL: 'redis://redis:6379/2'
+    ports:
+      - "8888:8888"
+    depends_on:
+      - redis
+
   redis:
     image: redislabs/redistimeseries:latest
     restart: unless-stopped

httpproxy/Dockerfile

@@ -0,0 +1,17 @@
+FROM golang:1.18-alpine as build
+
+WORKDIR /app
+COPY go.* ./
+RUN go mod download
+
+COPY proxy proxy
+RUN CGO_ENABLED=0 go build -o cacheproxy proxy/main.go
+
+FROM alpine
+
+COPY --from=build /app/cacheproxy /cacheproxy
+RUN chmod +x /cacheproxy
+
+CMD ["/cacheproxy"]
+
+

httpproxy/go.mod

@@ -0,0 +1,13 @@
+module cacheproxy
+
+go 1.18
+
+require (
+	github.com/elazarl/goproxy v0.0.0-20220529153421-8ea89ba92021
+	github.com/go-redis/redis/v8 v8.11.5
+)
+
+require (
+	github.com/cespare/xxhash/v2 v2.1.2 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
+)

httpproxy/go.sum

@@ -0,0 +1,20 @@
+github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
+github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
+github.com/elazarl/goproxy v0.0.0-20220529153421-8ea89ba92021 h1:EbF0UihnxWRcIMOwoVtqnAylsqcjzqpSvMdjF2Ud4rA=
+github.com/elazarl/goproxy v0.0.0-20220529153421-8ea89ba92021/go.mod h1:Ro8st/ElPeALwNFlcTpWmkr6IoMFfkjXAvTHpevnDsM=
+github.com/elazarl/goproxy/ext v0.0.0-20190711103511-473e67f1d7d2 h1:dWB6v3RcOy03t/bUadywsbyrQwCqZeNIEX6M1OtSZOM=
+github.com/elazarl/goproxy/ext v0.0.0-20190711103511-473e67f1d7d2/go.mod h1:gNh8nYJoAm43RfaxurUnxr+N1PwuFV3ZMl/efxlIlY8=
+github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
+github.com/go-redis/redis/v8 v8.11.5 h1:AcZZR7igkdvfVmQTPnu9WE37LRrO/YrBH5zWyjDC0oI=
+github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo=
+github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
+github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
+github.com/onsi/gomega v1.18.1 h1:M1GfJqGRrBrrGGsbxzV5dqM2U2ApXefZCQpkukxYRLE=
+github.com/rogpeppe/go-charset v0.0.0-20180617210344-2471d30d28b4/go.mod h1:qgYeAmZ5ZIpBWTGllZSQnw97Dj+woV0toclVaRGI8pc=
+golang.org/x/net v0.0.0-20210428140749-89ef3d95e781 h1:DzZ89McO9/gWPsQXS/FVKAlG02ZjaQ6AlZRBimEYOd0=
+golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e h1:fLOSk5Q00efkSvAm+4xcoXD+RRmLmmulPn5I3Y9F2EM=
+golang.org/x/text v0.3.6 h1:aRYxNxv6iGQlyVaZmk6ZgYEDa+Jg18DxebPSrd6bg1M=
+gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
+gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=

httpproxy/proxy/main.go

@@ -0,0 +1,168 @@
+package main
+
+import (
+	"bufio"
+	"context"
+	"fmt"
+	"log"
+	"net/http"
+	"net/http/httputil"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/elazarl/goproxy"
+	"github.com/go-redis/redis/v8"
+)
+
+const RedisDeadline = time.Second * 30
+
+type cachingData struct {
+	cacheKey      string
+	cacheFor      time.Duration
+	alreadyCached bool
+}
+
+type cachingHandler struct {
+	cli *redis.Client
+}
+
+func (c *cachingHandler) getFromCache(key string) (string, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), RedisDeadline)
+	defer cancel()
+
+	return c.cli.Get(ctx, key).Result()
+}
+
+func (c *cachingHandler) storeInCache(key string, value []byte, d time.Duration) error {
+	ctx, cancel := context.WithTimeout(context.Background(), RedisDeadline)
+	defer cancel()
+
+	_, err := c.cli.SetNX(ctx, key, value, d).Result()
+	return err
+}
+
+func (c *cachingHandler) getCacheDuration(r *http.Request) time.Duration {
+	defer r.Header.Del("X-CBSProxy-Cache-Duration")
+
+	val := r.Header.Get("X-CBSProxy-Cache-Duration")
+	if val == "" {
+		return 0
+	}
+
+	// Try parse duration first.
+	if dur, err := time.ParseDuration(val); err == nil {
+		return dur
+	}
+
+	// Parse number of seconds
+	if durS, err := strconv.Atoi(val); err == nil {
+		return time.Second * time.Duration(durS)
+	}
+
+	return 0
+}
+
+func (c *cachingHandler) overrideCacheFlag(r *http.Request) bool {
+	defer r.Header.Del("X-CBSProxy-Cache-Override")
+
+	return r.Header.Get("X-CBSProxy-Cache-Override") != ""
+}
+
+func (c *cachingHandler) validateDuration(d time.Duration) time.Duration {
+	if d <= 0 {
+		return d
+	}
+
+	if d < time.Second {
+		return time.Second
+	}
+
+	// Don't do stupid things.
+	if d > time.Minute*5 {
+		return time.Minute * 5
+	}
+	return d
+}
+
+func (c *cachingHandler) OnRequest(r *http.Request, ctx *goproxy.ProxyCtx) (*http.Request, *http.Response) {
+	cd := c.getCacheDuration(r)
+	cd = c.validateDuration(cd)
+
+	cacheKey := r.URL.String()
+	ctx.UserData = cachingData{cacheKey: cacheKey, cacheFor: cd, alreadyCached: false}
+
+	cachedResponseString, err := c.getFromCache(cacheKey)
+	if err == nil && len(cachedResponseString) > 0 {
+		if c.overrideCacheFlag(r) {
+			return r, nil
+		}
+
+		// Found response in cache. Try to decode.
+		cachedResp, err := http.ReadResponse(bufio.NewReader(strings.NewReader(cachedResponseString)), r)
+		if err != nil {
+			fmt.Printf("Failed to parse cached response: %v\n", err)
+			// Failed to decode will proxy the request.
+			return r, nil
+		}
+
+		ctx.UserData = cachingData{cacheKey: cacheKey, cacheFor: cd, alreadyCached: true}
+		return r, cachedResp
+	}
+
+	return r, nil
+}
+
+func (c *cachingHandler) OnResponse(resp *http.Response, ctx *goproxy.ProxyCtx) (out *http.Response) {
+	cd, ok := ctx.UserData.(cachingData)
+	if !ok {
+		fmt.Printf("Failed to parse context: should be 'cachingData', got %T\n", cd)
+		// Do not cache since we can't.
+		return resp
+	}
+
+	if cd.alreadyCached {
+		resp.Header.Set("X-CBSProxy-Cached", "true")
+		return resp
+	}
+
+	defer func() {
+		out.Header.Set("X-CBSProxy-Cached", "false")
+	}()
+
+	if cd.cacheFor == 0 {
+		return resp
+	}
+
+	dump, err := httputil.DumpResponse(resp, true)
+	if err != nil {
+		fmt.Printf("Failed to dump HTTP response: %v \n", err)
+		return resp
+	}
+
+	if err := c.storeInCache(cd.cacheKey, dump, cd.cacheFor); err != nil {
+		fmt.Printf("Failed to store result in cache: %v \n", err)
+	}
+	return resp
+}
+
+func main() {
+	redopts, err := redis.ParseURL(os.Getenv("REDIS_URL"))
+	if err != nil {
+		log.Fatal("Failed to parse REDIS_URL")
+		return
+	}
+
+	handler := cachingHandler{redis.NewClient(redopts)}
+
+	proxy := goproxy.NewProxyHttpServer()
+	proxy.Verbose = true
+
+	proxy.OnRequest().HandleConnect(goproxy.AlwaysMitm)
+	proxy.OnRequest().DoFunc(handler.OnRequest)
+	proxy.OnResponse().DoFunc(handler.OnResponse)
+
+	fmt.Println("Proxy started on port 8888")
+	log.Fatal(http.ListenAndServe(":8888", proxy))
+}