-
Notifications
You must be signed in to change notification settings - Fork 339
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(kuma-cp) read only cache manager
- Loading branch information
1 parent
079dbbd
commit e330872
Showing
27 changed files
with
345 additions
and
139 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,73 @@ | ||
| package manager | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "time" | ||
|
|
||
| "github.com/patrickmn/go-cache" | ||
|
|
||
| "github.com/Kong/kuma/pkg/core/resources/model" | ||
| "github.com/Kong/kuma/pkg/core/resources/store" | ||
| ) | ||
|
|
||
| // Cached version of the ReadOnlyResourceManager designed to be used only for use cases of eventual consistency. | ||
| // | ||
| // This cache is NOT consistent across instances of the control plane. | ||
| // This cache is mutex free for performance with consideration that: values can be overridden by other goroutine. | ||
| // * if cache expires and multiple goroutines tries to fetch the resource, they may fetch underlying manager multiple times. | ||
| // * if cache expires and multiple goroutines tries to fetch the resource, they fetch underlying manager multiple times | ||
| // and the value returned is different, the older value may be persisted. This is ok, since this cache is designed | ||
| // to have low expiration time (like 1s) and having old value just extends propagation of new config for 1 more second. | ||
| type cachedManager struct { | ||
| delegate ReadOnlyResourceManager | ||
| cache *cache.Cache | ||
| } | ||
|
|
||
| var _ ReadOnlyResourceManager = &cachedManager{} | ||
|
|
||
| func NewCachedManager(delegate ReadOnlyResourceManager, expirationTime time.Duration) ReadOnlyResourceManager { | ||
| return &cachedManager{ | ||
| delegate: delegate, | ||
| cache: cache.New(expirationTime, time.Duration(int64(float64(expirationTime)*0.9))), | ||
| } | ||
| } | ||
|
|
||
| func (c cachedManager) Get(ctx context.Context, res model.Resource, fs ...store.GetOptionsFunc) error { | ||
| opts := store.NewGetOptions(fs...) | ||
| cacheKey := fmt.Sprintf("GET:%s:%s", res.GetType(), opts.HashCode()) | ||
| obj, found := c.cache.Get(cacheKey) | ||
| if !found { | ||
| if err := c.delegate.Get(ctx, res, fs...); err != nil { | ||
| return err | ||
| } | ||
| c.cache.SetDefault(cacheKey, res) | ||
| } else { | ||
| cached := obj.(model.Resource) | ||
| if err := res.SetSpec(cached.GetSpec()); err != nil { | ||
| return err | ||
| } | ||
| res.SetMeta(cached.GetMeta()) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (c cachedManager) List(ctx context.Context, list model.ResourceList, fs ...store.ListOptionsFunc) error { | ||
| opts := store.NewListOptions(fs...) | ||
| cacheKey := fmt.Sprintf("LIST:%s:%s", list.GetItemType(), opts.HashCode()) | ||
| obj, found := c.cache.Get(cacheKey) | ||
| if !found { | ||
| if err := c.delegate.List(ctx, list, fs...); err != nil { | ||
| return err | ||
| } | ||
| c.cache.SetDefault(cacheKey, list.GetItems()) | ||
| } else { | ||
| resources := obj.([]model.Resource) | ||
| for _, res := range resources { | ||
| if err := list.AddItem(res); err != nil { | ||
| return err | ||
| } | ||
| } | ||
| } | ||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,121 @@ | ||
| package manager_test | ||
|
|
||
| import ( | ||
| "context" | ||
| "time" | ||
|
|
||
| mesh_proto "github.com/Kong/kuma/api/mesh/v1alpha1" | ||
| core_mesh "github.com/Kong/kuma/pkg/core/resources/apis/mesh" | ||
| core_manager "github.com/Kong/kuma/pkg/core/resources/manager" | ||
| core_model "github.com/Kong/kuma/pkg/core/resources/model" | ||
| core_store "github.com/Kong/kuma/pkg/core/resources/store" | ||
| "github.com/Kong/kuma/pkg/plugins/resources/memory" | ||
|
|
||
| . "github.com/onsi/ginkgo" | ||
| . "github.com/onsi/gomega" | ||
| ) | ||
|
|
||
| type countingResourcesManager struct { | ||
| store core_store.ResourceStore | ||
| getQueries int | ||
| listQueries int | ||
| } | ||
|
|
||
| func (c *countingResourcesManager) Get(ctx context.Context, res core_model.Resource, fn ...core_store.GetOptionsFunc) error { | ||
| c.getQueries++ | ||
| return c.store.Get(ctx, res, fn...) | ||
| } | ||
|
|
||
| func (c *countingResourcesManager) List(ctx context.Context, list core_model.ResourceList, fn ...core_store.ListOptionsFunc) error { | ||
| c.listQueries++ | ||
| return c.store.List(ctx, list, fn...) | ||
| } | ||
|
|
||
| var _ core_manager.ReadOnlyResourceManager = &countingResourcesManager{} | ||
|
|
||
| var _ = Describe("Cached Resource Manager", func() { | ||
|
|
||
| var store core_store.ResourceStore | ||
| var cachedManager core_manager.ReadOnlyResourceManager | ||
| var countingManager *countingResourcesManager | ||
| var res *core_mesh.DataplaneResource | ||
| expiration := 100 * time.Millisecond | ||
|
|
||
| BeforeEach(func() { | ||
| // given | ||
| store = memory.NewStore() | ||
| countingManager = &countingResourcesManager{ | ||
| store: store, | ||
| } | ||
| cachedManager = core_manager.NewCachedManager(countingManager, expiration) | ||
|
|
||
| // and created resources | ||
| res = &core_mesh.DataplaneResource{ | ||
| Spec: mesh_proto.Dataplane{ | ||
| Networking: &mesh_proto.Dataplane_Networking{ | ||
| Address: "127.0.0.1", | ||
| Inbound: []*mesh_proto.Dataplane_Networking_Inbound{ | ||
| { | ||
| Port: 80, | ||
| ServicePort: 8080, | ||
| }, | ||
| }, | ||
| }, | ||
| }, | ||
| } | ||
| err := store.Create(context.Background(), res, core_store.CreateByKey("dp-1", "default")) | ||
| Expect(err).ToNot(HaveOccurred()) | ||
| }) | ||
|
|
||
| It("should cache Get() queries", func() { | ||
| // when fetched resources multiple times | ||
| fetch := func() core_mesh.DataplaneResource { | ||
| fetched := core_mesh.DataplaneResource{} | ||
| err := cachedManager.Get(context.Background(), &fetched, core_store.GetByKey("dp-1", "default")) | ||
| Expect(err).ToNot(HaveOccurred()) | ||
| return fetched | ||
| } | ||
|
|
||
| for i := 0; i < 100; i++ { | ||
| fetch() | ||
| } | ||
|
|
||
| // then real manager should be called only once | ||
| Expect(fetch().Spec).To(Equal(res.Spec)) | ||
| Expect(countingManager.getQueries).To(Equal(1)) | ||
|
|
||
| // when | ||
| time.Sleep(expiration) | ||
|
|
||
| // then | ||
| Expect(fetch().Spec).To(Equal(res.Spec)) | ||
| Expect(countingManager.getQueries).To(Equal(2)) | ||
| }) | ||
|
|
||
| It("should cache List() queries", func() { | ||
| // when fetched resources multiple times | ||
| fetch := func() core_mesh.DataplaneResourceList { | ||
| fetched := core_mesh.DataplaneResourceList{} | ||
| err := cachedManager.List(context.Background(), &fetched, core_store.ListByMesh("default")) | ||
| Expect(err).ToNot(HaveOccurred()) | ||
| return fetched | ||
| } | ||
|
|
||
| for i := 0; i < 100; i++ { | ||
| fetch() | ||
| } | ||
|
|
||
| // then real manager should be called only once | ||
| Expect(fetch().Items).To(HaveLen(1)) | ||
| Expect(fetch().Items[0].GetSpec()).To(Equal(&res.Spec)) | ||
| Expect(countingManager.listQueries).To(Equal(1)) | ||
|
|
||
| // when | ||
| time.Sleep(expiration) | ||
|
|
||
| // then | ||
| Expect(fetch().Items).To(HaveLen(1)) | ||
| Expect(fetch().Items[0].GetSpec()).To(Equal(&res.Spec)) | ||
| Expect(countingManager.listQueries).To(Equal(2)) | ||
| }) | ||
| }) |
Oops, something went wrong.