Merge pull request micro#1656 from micro/client-cache

Client Cache

auth/service/service.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"sort"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/micro/go-micro/v2/auth"
@@ -23,9 +22,6 @@ type svc struct {
 	auth    pb.AuthService
 	rule    pb.RulesService
 	jwt     token.Provider
-
-	rules []*pb.Rule
-	sync.Mutex
 }
 
 func (s *svc) String() string {
@@ -53,8 +49,6 @@ func (s *svc) Init(opts ...auth.Option) {
 }
 
 func (s *svc) Options() auth.Options {
-	s.Lock()
-	defer s.Unlock()
 	return s.options
 }
 
@@ -110,9 +104,6 @@ func (s *svc) Revoke(role string, res *auth.Resource) error {
 
 // Verify an account has access to a resource
 func (s *svc) Verify(acc *auth.Account, res *auth.Resource) error {
-	// load the rules if none are loaded
-	s.loadRulesIfEmpty()
-
 	// set the namespace on the resource
 	if len(res.Namespace) == 0 {
 		res.Namespace = s.Options().Namespace
@@ -230,11 +221,14 @@ func accessForRule(rule *pb.Rule, acc *auth.Account, res *auth.Resource) pb.Acce
 // listRules gets all the rules from the store which match the filters.
 // filters are namespace, type, name and then endpoint.
 func (s *svc) listRules(filters ...string) []*pb.Rule {
-	s.Lock()
-	defer s.Unlock()
+	// load rules using the client cache
+	allRules, err := s.loadRules()
+	if err != nil {
+		return []*pb.Rule{}
+	}
 
 	var rules []*pb.Rule
-	for _, r := range s.rules {
+	for _, r := range allRules {
 		if len(filters) > 0 && r.Resource.Namespace != filters[0] {
 			continue
 		}
@@ -260,27 +254,15 @@ func (s *svc) listRules(filters ...string) []*pb.Rule {
 }
 
 // loadRules retrieves the rules from the auth service
-func (s *svc) loadRules() {
-	rsp, err := s.rule.List(context.TODO(), &pb.ListRequest{})
-	s.Lock()
-	defer s.Unlock()
+func (s *svc) loadRules() ([]*pb.Rule, error) {
+	rsp, err := s.rule.List(context.TODO(), &pb.ListRequest{}, client.WithCache(time.Minute))
 
 	if err != nil {
-		log.Errorf("Error listing rules: %v", err)
-		return
+		log.Debugf("Error listing rules: %v", err)
+		return nil, err
 	}
 
-	s.rules = rsp.Rules
-}
-
-func (s *svc) loadRulesIfEmpty() {
-	s.Lock()
-	rules := s.rules
-	s.Unlock()
-
-	if len(rules) == 0 {
-		s.loadRules()
-	}
+	return rsp.Rules, nil
 }
 
 func serializeToken(t *pb.Token) *auth.Token {

client/cache.go

@@ -0,0 +1,66 @@
+package client
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"hash/fnv"
+	"time"
+
+	"github.com/micro/go-micro/v2/metadata"
+	cache "github.com/patrickmn/go-cache"
+)
+
+// NewCache returns an initialised cache.
+func NewCache() *Cache {
+	return &Cache{
+		cache: cache.New(cache.NoExpiration, 30*time.Second),
+	}
+}
+
+// Cache for responses
+type Cache struct {
+	cache *cache.Cache
+}
+
+// Get a response from the cache
+func (c *Cache) Get(ctx context.Context, req *Request) (interface{}, bool) {
+	return c.cache.Get(key(ctx, req))
+}
+
+// Set a response in the cache
+func (c *Cache) Set(ctx context.Context, req *Request, rsp interface{}, expiry time.Duration) {
+	c.cache.Set(key(ctx, req), rsp, expiry)
+}
+
+// List the key value pairs in the cache
+func (c *Cache) List() map[string]string {
+	items := c.cache.Items()
+
+	rsp := make(map[string]string, len(items))
+	for k, v := range items {
+		bytes, _ := json.Marshal(v.Object)
+		rsp[k] = string(bytes)
+	}
+
+	return rsp
+}
+
+// key returns a hash for the context and request
+func key(ctx context.Context, req *Request) string {
+	md, _ := metadata.FromContext(ctx)
+
+	bytes, _ := json.Marshal(map[string]interface{}{
+		"metadata": md,
+		"request": map[string]interface{}{
+			"service":  (*req).Service(),
+			"endpoint": (*req).Endpoint(),
+			"method":   (*req).Method(),
+			"body":     (*req).Body(),
+		},
+	})
+
+	h := fnv.New64()
+	h.Write(bytes)
+	return fmt.Sprintf("%x", h.Sum(nil))
+}

client/cache_test.go

@@ -0,0 +1,76 @@
+package client
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/micro/go-micro/v2/metadata"
+)
+
+func TestCache(t *testing.T) {
+	ctx := context.TODO()
+	req := NewRequest("go.micro.service.foo", "Foo.Bar", nil)
+
+	t.Run("CacheMiss", func(t *testing.T) {
+		if _, ok := NewCache().Get(ctx, &req); ok {
+			t.Errorf("Expected to get no result from Get")
+		}
+	})
+
+	t.Run("CacheHit", func(t *testing.T) {
+		c := NewCache()
+
+		rsp := "theresponse"
+		c.Set(ctx, &req, rsp, time.Minute)
+
+		if res, ok := c.Get(ctx, &req); !ok {
+			t.Errorf("Expected a result, got nothing")
+		} else if res != rsp {
+			t.Errorf("Expected '%v' result, got '%v'", rsp, res)
+		}
+	})
+}
+
+func TestCacheKey(t *testing.T) {
+	ctx := context.TODO()
+	req1 := NewRequest("go.micro.service.foo", "Foo.Bar", nil)
+	req2 := NewRequest("go.micro.service.foo", "Foo.Baz", nil)
+	req3 := NewRequest("go.micro.service.foo", "Foo.Baz", "customquery")
+
+	t.Run("IdenticalRequests", func(t *testing.T) {
+		key1 := key(ctx, &req1)
+		key2 := key(ctx, &req1)
+		if key1 != key2 {
+			t.Errorf("Expected the keys to match for identical requests and context")
+		}
+	})
+
+	t.Run("DifferentRequestEndpoints", func(t *testing.T) {
+		key1 := key(ctx, &req1)
+		key2 := key(ctx, &req2)
+
+		if key1 == key2 {
+			t.Errorf("Expected the keys to differ for different request endpoints")
+		}
+	})
+
+	t.Run("DifferentRequestBody", func(t *testing.T) {
+		key1 := key(ctx, &req2)
+		key2 := key(ctx, &req3)
+
+		if key1 == key2 {
+			t.Errorf("Expected the keys to differ for different request bodies")
+		}
+	})
+
+	t.Run("DifferentMetadata", func(t *testing.T) {
+		mdCtx := metadata.Set(context.TODO(), "foo", "bar")
+		key1 := key(mdCtx, &req1)
+		key2 := key(ctx, &req1)
+
+		if key1 == key2 {
+			t.Errorf("Expected the keys to differ for different metadata")
+		}
+	})
+}

client/options.go

@@ -29,6 +29,9 @@ type Options struct {
 	PoolSize int
 	PoolTTL  time.Duration
 
+	// Response cache
+	Cache *Cache
+
 	// Middleware for client
 	Wrappers []Wrapper
 
@@ -59,6 +62,8 @@ type CallOptions struct {
 	StreamTimeout time.Duration
 	// Use the services own auth token
 	ServiceToken bool
+	// Duration to cache the response for
+	CacheExpiry time.Duration
 
 	// Middleware for low level call func
 	CallWrappers []CallWrapper
@@ -91,6 +96,7 @@ type RequestOptions struct {
 
 func NewOptions(options ...Option) Options {
 	opts := Options{
+		Cache:       NewCache(),
 		Context:     context.Background(),
 		ContentType: DefaultContentType,
 		Codecs:      make(map[string]codec.NewCodec),
@@ -324,6 +330,14 @@ func WithServiceToken() CallOption {
 	}
 }
 
+// WithCache is a CallOption which sets the duration the response
+// shoull be cached for
+func WithCache(c time.Duration) CallOption {
+	return func(o *CallOptions) {
+		o.CacheExpiry = c
+	}
+}
+
 func WithMessageContentType(ct string) MessageOption {
 	return func(o *MessageOptions) {
 		o.ContentType = ct

config/cmd/cmd.go

@@ -471,9 +471,13 @@ func (c *cmd) Before(ctx *cli.Context) error {
 	var serverOpts []server.Option
 	var clientOpts []client.Option
 
-	// setup a client to use when calling the runtime
+	// setup a client to use when calling the runtime. It is important the auth client is wrapped
+	// after the cache client since the wrappers are applied in reverse order and the cache will use
+	// some of the headers set by the auth client.
 	authFn := func() auth.Auth { return *c.opts.Auth }
-	microClient := wrapper.AuthClient(authFn, grpc.NewClient())
+	cacheFn := func() *client.Cache { return (*c.opts.Client).Options().Cache }
+	microClient := wrapper.CacheClient(cacheFn, grpc.NewClient())
+	microClient = wrapper.AuthClient(authFn, microClient)
 
 	// Set the store
 	if name := ctx.String("store"); len(name) > 0 {

debug/service/handler/debug.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"time"
 
+	"github.com/micro/go-micro/v2/client"
 	"github.com/micro/go-micro/v2/debug/log"
 	proto "github.com/micro/go-micro/v2/debug/service/proto"
 	"github.com/micro/go-micro/v2/debug/stats"
@@ -13,11 +14,12 @@ import (
 )
 
 // NewHandler returns an instance of the Debug Handler
-func NewHandler() *Debug {
+func NewHandler(c client.Client) *Debug {
 	return &Debug{
 		log:   log.DefaultLog,
 		stats: stats.DefaultStats,
 		trace: trace.DefaultTracer,
+		cache: c.Options().Cache,
 	}
 }
 
@@ -30,6 +32,8 @@ type Debug struct {
 	stats stats.Stats
 	// the tracer
 	trace trace.Tracer
+	// the cache
+	cache *client.Cache
 }
 
 func (d *Debug) Health(ctx context.Context, req *proto.HealthRequest, rsp *proto.HealthResponse) error {
@@ -164,3 +168,9 @@ func (d *Debug) Log(ctx context.Context, stream server.Stream) error {
 
 	return nil
 }
+
+// Cache returns all the key value pairs in the client cache
+func (d *Debug) Cache(ctx context.Context, req *proto.CacheRequest, rsp *proto.CacheResponse) error {
+	rsp.Values = d.cache.List()
+	return nil
+}