Added a new metric to the workflow cache

common/metrics/defs.go

@@ -2458,6 +2458,7 @@ const (
  UpdateWorkflowExecutionCount
  WorkflowIDCacheSizeGauge
  WorkflowIDCacheRequestsExternalRatelimitedCounter
+ WorkflowIDCacheRequestsExternalMaxRequestsPerSecondsTimer
  WorkflowIDCacheRequestsInternalRatelimitedCounter
  NumHistoryMetrics
 )
@@ -3093,6 +3094,7 @@ var MetricDefs = map[ServiceIdx]map[int]metricDefinition{
  UpdateWorkflowExecutionCount: {metricName: "update_workflow_execution_count", metricType: Counter},
  WorkflowIDCacheSizeGauge: {metricName: "workflow_id_cache_size", metricType: Gauge},
  WorkflowIDCacheRequestsExternalRatelimitedCounter: {metricName: "workflow_id_external_requests_ratelimited", metricType: Counter},
+ WorkflowIDCacheRequestsExternalMaxRequestsPerSecondsTimer: {metricName: "workflow_id_external_requests_max_requests_per_seconds", metricType: Timer},
  WorkflowIDCacheRequestsInternalRatelimitedCounter: {metricName: "workflow_id_internal_requests_ratelimited", metricType: Counter},
  },
  Matching: {

service/history/workflowcache/cache.go

@@ -29,6 +29,7 @@ import (
  "time"
 
  "github.com/uber/cadence/common/cache"
+ "github.com/uber/cadence/common/clock"
  "github.com/uber/cadence/common/dynamicconfig"
  "github.com/uber/cadence/common/log"
  "github.com/uber/cadence/common/log/tag"
@@ -55,7 +56,10 @@ type wfCache struct {
  domainCache cache.DomainCache
  metricsClient metrics.Client
  logger log.Logger
- getCacheItemFn func(domainName string, workflowID string) (*cacheValue, error)
+ timeSource clock.TimeSource
+
+ // we use functions to get cache items, and the current time, so we can mock it in unit tests
+ getCacheItemFn func(domainName string, workflowID string) (*cacheValue, error)
 }
 
 type cacheKey struct {
@@ -66,6 +70,7 @@ type cacheKey struct {
 type cacheValue struct {
  externalRateLimiter quotas.Limiter
  internalRateLimiter quotas.Limiter
+ countMetric workflowIDCountMetric
 }
 
 // Params is the parameters for a new WFCache
@@ -96,6 +101,7 @@ func New(params Params) WFCache {
  workflowIDCacheInternalEnabled: params.WorkflowIDCacheInternalEnabled,
  domainCache: params.DomainCache,
  metricsClient: params.MetricsClient,
+ timeSource: clock.NewRealTimeSource(),
  logger: params.Logger,
  }
  // We set getCacheItemFn to cache.getCacheItem so that we can mock it in unit tests
@@ -138,6 +144,7 @@ func (c *wfCache) allow(domainID string, workflowID string, rateLimitType rateLi
 
  switch rateLimitType {
  case external:
+ c.updatePerDomainMaxWFRequestCount(domainName, value)
  if !value.externalRateLimiter.Allow() {
  c.emitRateLimitMetrics(domainID, workflowID, domainName, "external", metrics.WorkflowIDCacheRequestsExternalRatelimitedCounter)
  return false

service/history/workflowcache/metrics.go

@@ -0,0 +1,56 @@
+// The MIT License (MIT)
+
+// Copyright (c) 2017-2020 Uber Technologies Inc.
+
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+package workflowcache
+
+import (
+ "sync"
+ "time"
+
+ "github.com/uber/cadence/common/metrics"
+)
+
+type workflowIDCountMetric struct {
+ sync.Mutex
+
+ startingSecond time.Time
+ count int
+}
+
+func (w *workflowIDCountMetric) reset(now time.Time) {
+ w.startingSecond = now
+ w.count = 0
+}
+
+func (c *wfCache) updatePerDomainMaxWFRequestCount(domainName string, value *cacheValue) {
+ value.countMetric.Lock()
+ defer value.countMetric.Unlock()
+
+ if c.timeSource.Since(value.countMetric.startingSecond) > time.Second {
+ value.countMetric.reset(c.timeSource.Now().UTC())
+ }
+ value.countMetric.count++
+
+ // We can just use the upper of the metric, so it is not an issue to emit all the counts
+ c.metricsClient.Scope(metrics.HistoryClientWfIDCacheScope, metrics.DomainTag(domainName)).
+ RecordTimer(metrics.WorkflowIDCacheRequestsExternalMaxRequestsPerSecondsTimer, time.Duration(value.countMetric.count))
+}

service/history/workflowcache/metrics_test.go

@@ -0,0 +1,110 @@
+// The MIT License (MIT)
+
+// Copyright (c) 2017-2020 Uber Technologies Inc.
+
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+package workflowcache
+
+import (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/uber-go/tally"
+
+ "github.com/uber/cadence/common/clock"
+ "github.com/uber/cadence/common/metrics"
+)
+
+func TestUpdatePerDomainMaxWFRequestCount(t *testing.T) {
+ domainName := "some domain name"
+
+ cases := []struct {
+ name string
+ updatePerDomainMaxWFRequestCount func(wfc *wfCache, source clock.MockedTimeSource)
+ expecetMetrics []time.Duration
+ }{
+ {
+ name: "Single workflowID",
+ updatePerDomainMaxWFRequestCount: func(wfc *wfCache, timeSource clock.MockedTimeSource) {
+ workflowID1 := &cacheValue{}
+ wfc.updatePerDomainMaxWFRequestCount(domainName, workflowID1) // Emits 1
+ wfc.updatePerDomainMaxWFRequestCount(domainName, workflowID1) // Emits 2
+ },
+ expecetMetrics: []time.Duration{1, 2},
+ },
+ {
+ name: "Separate workflowIDs",
+ updatePerDomainMaxWFRequestCount: func(wfc *wfCache, timeSource clock.MockedTimeSource) {
+ workflowID1 := &cacheValue{}
+ wfc.updatePerDomainMaxWFRequestCount(domainName, workflowID1) // Emits 1
+
+ workflowID2 := &cacheValue{}
+ wfc.updatePerDomainMaxWFRequestCount(domainName, workflowID2) // Emits 1
+ wfc.updatePerDomainMaxWFRequestCount(domainName, workflowID2) // Emits 2
+ wfc.updatePerDomainMaxWFRequestCount(domainName, workflowID2) // Emits 3
+
+ wfc.updatePerDomainMaxWFRequestCount(domainName, workflowID1) // Emits 2
+
+ },
+ expecetMetrics: []time.Duration{1, 1, 2, 3, 2},
+ },
+ {
+ name: "Reset",
+ updatePerDomainMaxWFRequestCount: func(wfc *wfCache, timeSource clock.MockedTimeSource) {
+ workflowID1 := &cacheValue{}
+ wfc.updatePerDomainMaxWFRequestCount(domainName, workflowID1) // Emits 1
+ wfc.updatePerDomainMaxWFRequestCount(domainName, workflowID1) // Emits 2
+
+ timeSource.Advance(1100 * time.Millisecond)
+ wfc.updatePerDomainMaxWFRequestCount(domainName, workflowID1) // Emits 1
+ },
+ expecetMetrics: []time.Duration{1, 2, 1},
+ },
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ testScope := tally.NewTestScope("", make(map[string]string))
+ timeSource := clock.NewMockedTimeSourceAt(time.Unix(123, 456))
+
+ wfc := &wfCache{
+ metricsClient: metrics.NewClient(testScope, metrics.History),
+ timeSource: timeSource,
+ }
+
+ tc.updatePerDomainMaxWFRequestCount(wfc, timeSource)
+
+ // We expect the domain tag to be set to "all" and "some domain name", we don't know the order, so use a set
+ expectedDomainTags := map[string]struct{}{"all": {}, "some domain name": {}}
+ actualDomainTags := map[string]struct{}{}
+
+ timers := testScope.Snapshot().Timers()
+ assert.Equal(t, 2, len(timers))
+ for _, v := range timers {
+ actualDomainTags[v.Tags()["domain"]] = struct{}{}
+ assert.Equal(t, tc.expecetMetrics, v.Values())
+ }
+
+ assert.Equal(t, expectedDomainTags, actualDomainTags)
+ })
+ }
+
+}