From f6102bde70d344de1b80f30b1333766b0e8e207a Mon Sep 17 00:00:00 2001 From: Jake Sanders Date: Mon, 2 Mar 2020 18:14:25 +0000 Subject: [PATCH] Add a cache to workers KV storage implementation (#1284) * cloudflare-cache * go mod tidy --- go.mod | 1 + go.sum | 5 ++++- store/cloudflare/cloudflare.go | 30 ++++++++++++++++++++++++++++- store/cloudflare/cloudflare_test.go | 1 + store/cloudflare/options.go | 10 ++++++++++ 5 files changed, 45 insertions(+), 2 deletions(-) diff --git a/go.mod b/go.mod index 3e58e15b..33c35ba1 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,7 @@ go 1.13 require ( github.com/BurntSushi/toml v0.3.1 + github.com/ReneKroon/ttlcache v1.6.0 github.com/beevik/ntp v0.2.0 github.com/bitly/go-simplejson v0.5.0 github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 // indirect diff --git a/go.sum b/go.sum index 3c3fc7aa..c0cf2229 100644 --- a/go.sum +++ b/go.sum @@ -35,6 +35,8 @@ github.com/Microsoft/go-winio v0.4.15-0.20190919025122-fc70bd9a86b5/go.mod h1:tT github.com/Microsoft/hcsshim v0.8.7-0.20191101173118-65519b62243c h1:YMP6olTU903X3gxQJckdmiP8/zkSMq4kN3uipsU9XjU= github.com/Microsoft/hcsshim v0.8.7-0.20191101173118-65519b62243c/go.mod h1:7xhjOwRV2+0HXGmM0jxaEu+ZiXJFoVZOTfL/dmqbrD8= github.com/OpenDNS/vegadns2client v0.0.0-20180418235048-a3fa4a771d87/go.mod h1:iGLljf5n9GjT6kc0HBvyI1nOKnGQbNB66VzSNbK5iks= +github.com/ReneKroon/ttlcache v1.6.0 h1:aO+GDNVKTQmcuI0H78PXCR9E59JMiGfSXHAkVBUlzbA= +github.com/ReneKroon/ttlcache v1.6.0/go.mod h1:DG6nbhXKUQhrExfwwLuZUdH7UnRDDRA1IW+nBuCssvs= github.com/Shopify/sarama v1.19.0/go.mod h1:FVkBWblsNy7DGZRfXLU0O9RCGt5g3g3yEuWXgklEdEo= github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI= github.com/akamai/AkamaiOPEN-edgegrid-golang v0.9.0/go.mod h1:zpDJeKyp9ScW4NNrbdr+Eyxvry3ilGPewKoXw3XGN1k= @@ -284,7 +286,6 @@ github.com/mholt/certmagic v0.9.3 h1:RmzuNJ5mpFplDbyS41z+gGgE/py24IX6m0nHZ0yNTQU github.com/mholt/certmagic v0.9.3/go.mod h1:nu8jbsbtwK4205EDH/ZUMTKsfYpJA1Q7MKXHfgTihNw= github.com/micro/cli/v2 v2.1.2 h1:43J1lChg/rZCC1rvdqZNFSQDrGT7qfMrtp6/ztpIkEM= github.com/micro/cli/v2 v2.1.2/go.mod h1:EguNh6DAoWKm9nmk+k/Rg0H3lQnDxqzu5x5srOtGtYg= -github.com/micro/go-micro v1.18.0 h1:gP70EZVHpJuUIT0YWth192JmlIci+qMOEByHm83XE9E= github.com/micro/mdns v0.3.0 h1:bYycYe+98AXR3s8Nq5qvt6C573uFTDPIYzJemWON0QE= github.com/micro/mdns v0.3.0/go.mod h1:KJ0dW7KmicXU2BV++qkLlmHYcVv7/hHnbtguSWt9Aoc= github.com/miekg/dns v1.1.3/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg= @@ -435,6 +436,8 @@ go.uber.org/atomic v1.3.2/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= go.uber.org/atomic v1.5.0 h1:OI5t8sDa1Or+q8AeE+yKeB/SDYioSHAgcVljj9JIETY= go.uber.org/atomic v1.5.0/go.mod h1:sABNBOSYdrvTF6hTgEIbc7YasKWGhgEQZyfxyTvoXHQ= +go.uber.org/goleak v0.10.0 h1:G3eWbSNIskeRqtsN/1uI5B+eP73y3JUuBsv9AZjehb4= +go.uber.org/goleak v0.10.0/go.mod h1:VCZuO8V8mFPlL0F5J5GK1rtHV3DrFcQ1R8ryq7FK0aI= go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0= go.uber.org/multierr v1.3.0 h1:sFPn2GLc3poCkfrpIXGhBD2X0CMIo4Q/zSULXrj/+uc= go.uber.org/multierr v1.3.0/go.mod h1:VgVr7evmIr6uPjLBxg28wmKNXyqE9akIJ5XnfpiKl+4= diff --git a/store/cloudflare/cloudflare.go b/store/cloudflare/cloudflare.go index 7feac885..cbc0a365 100644 --- a/store/cloudflare/cloudflare.go +++ b/store/cloudflare/cloudflare.go @@ -19,6 +19,8 @@ import ( "github.com/micro/go-micro/v2/store" "github.com/pkg/errors" + + "github.com/ReneKroon/ttlcache" ) const ( @@ -35,6 +37,8 @@ type workersKV struct { namespace string // http client to use httpClient *http.Client + // cache + cache *ttlcache.Cache } // apiResponse is a cloudflare v4 api response @@ -103,6 +107,16 @@ func (w *workersKV) Init(opts ...store.Option) error { if len(w.options.Namespace) > 0 { w.namespace = w.options.Namespace } + ttl := w.options.Context.Value("STORE_CACHE_TTL") + if ttl != nil { + ttlint64, ok := ttl.(int64) + if !ok { + log.Fatal("STORE_CACHE_TTL from context must be type int64") + } + w.cache = ttlcache.NewCache() + w.cache.SetTTL(time.Duration(ttlint64)) + w.cache.SkipTtlExtensionOnHit(true) + } return nil } @@ -191,6 +205,15 @@ func (w *workersKV) Read(key string, opts ...store.ReadOption) ([]*store.Record, var records []*store.Record for _, k := range keys { + if w.cache != nil { + if resp, hit := w.cache.Get(k); hit { + if record, ok := resp.(*store.Record); ok { + records = append(records, record) + continue + } + } + } + path := fmt.Sprintf("accounts/%s/storage/kv/namespaces/%s/values/%s", w.account, w.namespace, url.PathEscape(k)) response, headers, status, err := w.request(ctx, http.MethodGet, path, nil, make(http.Header)) if err != nil { @@ -210,6 +233,7 @@ func (w *workersKV) Read(key string, opts ...store.ReadOption) ([]*store.Record, } record.Expiry = time.Until(time.Unix(expiryUnix, 0)) } + w.cache.Set(record.Key, record) records = append(records, record) } @@ -217,6 +241,10 @@ func (w *workersKV) Read(key string, opts ...store.ReadOption) ([]*store.Record, } func (w *workersKV) Write(r *store.Record) error { + // Set it in local cache, with the global TTL from options + if w.cache != nil { + w.cache.Set(r.Key, r) + } ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) defer cancel() @@ -336,7 +364,7 @@ func (w *workersKV) String() string { return "cloudflare" } -// New returns a cloudflare Store implementation. +// NewStore returns a cloudflare Store implementation. // Account ID, Token and Namespace must either be passed as options or // environment variables. If set as env vars we expect the following; // CF_API_TOKEN to a cloudflare API token scoped to Workers KV. diff --git a/store/cloudflare/cloudflare_test.go b/store/cloudflare/cloudflare_test.go index ee5deee1..dbf78c26 100644 --- a/store/cloudflare/cloudflare_test.go +++ b/store/cloudflare/cloudflare_test.go @@ -24,6 +24,7 @@ func TestCloudflare(t *testing.T) { Token(apiToken), Account(accountID), Namespace(kvID), + CacheTTL(60000000000), ) records, err := wkv.List() diff --git a/store/cloudflare/options.go b/store/cloudflare/options.go index 8b307024..71331d12 100644 --- a/store/cloudflare/options.go +++ b/store/cloudflare/options.go @@ -51,3 +51,13 @@ func Namespace(ns string) store.Option { o.Namespace = ns } } + +// CacheTTL sets the timeout in nanoseconds of the read/write cache +func CacheTTL(ttl int64) store.Option { + return func(o *store.Options) { + if o.Context == nil { + o.Context = context.Background() + } + o.Context = context.WithValue(o.Context, "STORE_CACHE_TTL", ttl) + } +}