-
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.
Move everything common between cla and mesh cache in a common cache i…
…mplementation
- Loading branch information
Charly Molter
committed
Apr 28, 2021
1 parent
18baef9
commit ec2794b
Showing
6 changed files
with
266 additions
and
81 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,57 @@ | ||
| package once | ||
|
|
||
| import ( | ||
| "context" | ||
| "github.com/kumahq/kuma/pkg/metrics" | ||
| "github.com/patrickmn/go-cache" | ||
| "github.com/prometheus/client_golang/prometheus" | ||
| "time" | ||
| ) | ||
|
|
||
| type Cache struct { | ||
| cache *cache.Cache | ||
| onceMap *omap | ||
| metrics *prometheus.GaugeVec | ||
| } | ||
|
|
||
| func New(expirationTime time.Duration, name string, metrics metrics.Metrics) (*Cache, error) { | ||
| metric := prometheus.NewGaugeVec(prometheus.GaugeOpts{ | ||
| Name: name, | ||
| Help: "Summary of " + name, | ||
| }, []string{"operation", "result"}) | ||
| if err := metrics.Register(metric); err != nil { | ||
| return nil, err | ||
| } | ||
| return &Cache{ | ||
| cache: cache.New(expirationTime, time.Duration(int64(float64(expirationTime)*0.9))), | ||
| onceMap: newMap(), | ||
| metrics: metric, | ||
| }, nil | ||
| } | ||
|
|
||
| func (c *Cache) Get(ctx context.Context, key string, fn func(context.Context, string) (interface{}, error)) (interface{}, error) { | ||
| v, found := c.cache.Get(key) | ||
| if found { | ||
| c.metrics.WithLabelValues("get", "hit").Inc() | ||
| return v, nil | ||
| } | ||
| o := c.onceMap.Get(key) | ||
| c.metrics.WithLabelValues("get", "hit-wait").Inc() | ||
| o.Do(func() (interface{}, error) { | ||
| defer c.onceMap.Delete(key) | ||
| c.metrics.WithLabelValues("get", "hit-wait").Dec() | ||
| c.metrics.WithLabelValues("get", "miss").Inc() | ||
| res, err := fn(ctx, key) | ||
| if err != nil { | ||
| c.metrics.WithLabelValues("get", "error").Inc() | ||
| return nil, err | ||
| } | ||
| c.cache.SetDefault(key, res) | ||
| return res, nil | ||
| }) | ||
| if o.Err != nil { | ||
| return "", o.Err | ||
| } | ||
| return o.Value, 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,160 @@ | ||
| package once_test | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "github.com/kumahq/kuma/pkg/xds/cache/once" | ||
| "sync" | ||
| "sync/atomic" | ||
| "time" | ||
|
|
||
| core_metrics "github.com/kumahq/kuma/pkg/metrics" | ||
| test_metrics "github.com/kumahq/kuma/pkg/test/metrics" | ||
|
|
||
| . "github.com/onsi/ginkgo" | ||
| . "github.com/onsi/gomega" | ||
| ) | ||
|
|
||
| var _ = Describe("ClusterLoadAssignment Cache", func() { | ||
| var metrics core_metrics.Metrics | ||
| var cache *once.Cache | ||
|
|
||
| expiration := time.Millisecond * 200 | ||
|
|
||
| BeforeEach(func() { | ||
| var err error | ||
| metrics, err = core_metrics.NewMetrics("Standalone") | ||
| Expect(err).ToNot(HaveOccurred()) | ||
|
|
||
| cache, err = once.New(expiration, "cache", metrics) | ||
| Expect(err).ToNot(HaveOccurred()) | ||
| }) | ||
|
|
||
| It("should cache Get() queries", func() { | ||
| var count int32 = 0 | ||
| var val int32 = 1 | ||
| fn := func(ctx context.Context, s string) (interface{}, error) { | ||
| atomic.AddInt32(&count, 1) | ||
| v := atomic.LoadInt32(&val) | ||
| return v, nil | ||
| } | ||
| By("getting item for the first time") | ||
| out, err := cache.Get(context.Background(), "k1", fn) | ||
| Expect(err).ToNot(HaveOccurred()) | ||
| Expect(out.(int32)).To(Equal(int32(1))) | ||
| Expect(atomic.LoadInt32(&count)).To(Equal(int32(1))) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "miss").Gauge.GetValue()).To(Equal(1.0)) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "hit")).To(BeNil()) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "error")).To(BeNil()) | ||
|
|
||
| By("getting cached item") | ||
| out, err = cache.Get(context.Background(), "k1", fn) | ||
| Expect(err).ToNot(HaveOccurred()) | ||
| Expect(out.(int32)).To(Equal(int32(1))) | ||
| Expect(atomic.LoadInt32(&count)).To(Equal(int32(1))) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "miss").Gauge.GetValue()).To(Equal(1.0)) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "hit").Gauge.GetValue()).To(Equal(1.0)) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "error")).To(BeNil()) | ||
|
|
||
| By("updating Dataplane in store") | ||
| atomic.StoreInt32(&val, 2) | ||
|
|
||
| By("cached value hasn't changed") | ||
| out, err = cache.Get(context.Background(), "k1", fn) | ||
| Expect(err).ToNot(HaveOccurred()) | ||
| Expect(atomic.LoadInt32(&count)).To(Equal(int32(1))) | ||
| Expect(out.(int32)).To(Equal(int32(1))) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "miss").Gauge.GetValue()).To(Equal(1.0)) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "hit").Gauge.GetValue()).To(Equal(2.0)) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "error")).To(BeNil()) | ||
|
|
||
| By("wait for invalidation") | ||
| time.Sleep(expiration) | ||
|
|
||
| By("get new value") | ||
| out, err = cache.Get(context.Background(), "k1", fn) | ||
| Expect(err).ToNot(HaveOccurred()) | ||
| Expect(out.(int32)).To(Equal(int32(2))) | ||
| Expect(atomic.LoadInt32(&count)).To(Equal(int32(2))) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "miss").Gauge.GetValue()).To(Equal(2.0)) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "hit").Gauge.GetValue()).To(Equal(2.0)) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "error")).To(BeNil()) | ||
| }) | ||
|
|
||
| It("should cache concurrent Get() requests", func() { | ||
| var count int32 = 0 | ||
| var val int32 = 1 | ||
| fn := func(ctx context.Context, s string) (interface{}, error) { | ||
| atomic.AddInt32(&count, 1) | ||
| v := atomic.LoadInt32(&val) | ||
| return v, nil | ||
| } | ||
| var wg sync.WaitGroup | ||
| for i := 0; i < 100; i++ { | ||
| wg.Add(1) | ||
| go func() { | ||
| out, err := cache.Get(context.Background(), "key", fn) | ||
| Expect(err).ToNot(HaveOccurred()) | ||
|
|
||
| Expect(out.(int32)).To(Equal(atomic.LoadInt32(&val))) | ||
| wg.Done() | ||
| }() | ||
| } | ||
| wg.Wait() | ||
|
|
||
| Expect(atomic.LoadInt32(&count)).To(Equal(int32(1))) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "miss").Gauge.GetValue()).To(Equal(1.0)) | ||
| hitWaits := 0.0 | ||
| if hw := test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "hit-wait"); hw != nil { | ||
| hitWaits = hw.Gauge.GetValue() | ||
| } | ||
| hits := 0.0 | ||
| if h := test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "hit"); h != nil { | ||
| hits = h.Gauge.GetValue() | ||
| } | ||
| Expect(hitWaits + hits + 1).To(Equal(100.0)) | ||
| }) | ||
|
|
||
| It("should retry previously failed Get() requests", func() { | ||
| var count int32 = 0 | ||
| var hasError int32 = 1 | ||
| fn := func(ctx context.Context, s string) (interface{}, error) { | ||
| atomic.AddInt32(&count, 1) | ||
| if atomic.LoadInt32(&hasError) != 0 { | ||
| return "", errors.New("It's an error!") | ||
| } | ||
| return "hello", nil | ||
| } | ||
| By("getting Hash for the first time") | ||
| out, err := cache.Get(context.Background(), "key-1", fn) | ||
| Expect(err).To(HaveOccurred()) | ||
| Expect(out).To(BeEmpty()) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "miss").Gauge.GetValue()).To(Equal(1.0)) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "hit")).To(BeNil()) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "error").Gauge.GetValue()).To(Equal(1.0)) | ||
|
|
||
| By("getting Hash calls again") | ||
| out, err = cache.Get(context.Background(), "key-1", fn) | ||
| Expect(err).To(HaveOccurred()) | ||
| Expect(out).To(BeEmpty()) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "miss").Gauge.GetValue()).To(Equal(2.0)) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "hit")).To(BeNil()) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "error").Gauge.GetValue()).To(Equal(2.0)) | ||
|
|
||
| By("Getting the hash once manager is fixed") | ||
| atomic.StoreInt32(&hasError, 0) | ||
| out, err = cache.Get(context.Background(), "key-1", fn) | ||
| Expect(err).ToNot(HaveOccurred()) | ||
| Expect(out).ToNot(BeEmpty()) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "miss").Gauge.GetValue()).To(Equal(3.0)) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "hit")).To(BeNil()) | ||
|
|
||
| By("Now it should cache the hash once manager is fixed") | ||
| err = nil | ||
| out, err = cache.Get(context.Background(), "key-1", fn) | ||
| Expect(err).ToNot(HaveOccurred()) | ||
| Expect(out).ToNot(BeNil()) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "miss").Gauge.GetValue()).To(Equal(3.0)) | ||
| Expect(test_metrics.FindMetric(metrics, "cache", "operation", "get", "result", "hit").Gauge.GetValue()).To(Equal(1.0)) | ||
| }) | ||
| }) |
Oops, something went wrong.