Add StatsReporter component to estimate QPS

common/metrics/defs.go

@@ -2582,6 +2582,7 @@ const (
 	PollActivityTaskAlreadyStartedCounterPerTaskList
 	TaskListReadWritePartitionMismatchGauge
 	TaskListPollerPartitionMismatchGauge
+	EstimatedAddTaskQPSGauge
 
 	NumMatchingMetrics
 )
@@ -2824,7 +2825,6 @@ var MetricDefs = map[ServiceIdx]map[int]metricDefinition{
 		MatchingClientInvalidTaskListName:                         {metricName: "invalid_task_list_name", metricType: Counter},
 
 		// per task list common metrics
-
 		CadenceRequestsPerTaskList: {
 			metricName: "cadence_requests_per_tl", metricRollupName: "cadence_requests", metricType: Counter,
 		},
@@ -3262,6 +3262,7 @@ var MetricDefs = map[ServiceIdx]map[int]metricDefinition{
 		PollActivityTaskAlreadyStartedCounterPerTaskList:        {metricName: "poll_activity_task_already_started_per_tl", metricType: Counter},
 		TaskListReadWritePartitionMismatchGauge:                 {metricName: "tasklist_read_write_partition_mismatch", metricType: Gauge},
 		TaskListPollerPartitionMismatchGauge:                    {metricName: "tasklist_poller_partition_mismatch", metricType: Gauge},
+		EstimatedAddTaskQPSGauge:                                {metricName: "estimated_add_task_qps_per_tl", metricType: Gauge},
 	},
 	Worker: {
 		ReplicatorMessages:                            {metricName: "replicator_messages"},

common/stats/interfaces.go

@@ -0,0 +1,37 @@
+// 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 stats
+
+import (
+	"github.com/uber/cadence/common"
+)
+
+// QPSTracker is an interface for reporting statistics related to quotas.
+type QPSTracker interface {
+	common.Daemon
+	// ReportCounter reports the value of a counter.
+	ReportCounter(int64)
+
+	// QPS returns the current queries per second (QPS) value.
+	QPS() float64
+}

common/stats/stats.go

@@ -0,0 +1,115 @@
+// 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 stats
+
+import (
+	"sync"
+	"time"
+
+	"go.uber.org/atomic"
+
+	"github.com/uber/cadence/common"
+	"github.com/uber/cadence/common/clock"
+)
+
+type (
+	// emaFixedWindowQPSTracker is a QPSTracker that uses a fixed time period to calculate QPS and an exponential moving average algorithm to estimate QPS.
+	emaFixedWindowQPSTracker struct {
+		timeSource            clock.TimeSource
+		exp                   float64
+		bucketInterval        time.Duration
+		bucketIntervalSeconds float64
+		wg                    sync.WaitGroup
+		done                  chan struct{}
+		status                *atomic.Int32
+		firstBucket           bool
+
+		qps     *atomic.Float64
+		counter *atomic.Int64
+	}
+)
+
+func NewEmaFixedWindowQPSTracker(timeSource clock.TimeSource, exp float64, bucketInterval time.Duration) QPSTracker {
+	return &emaFixedWindowQPSTracker{
+		timeSource:            timeSource,
+		exp:                   exp,
+		bucketInterval:        bucketInterval,
+		bucketIntervalSeconds: float64(bucketInterval) / float64(time.Second),
+		done:                  make(chan struct{}),
+		status:                atomic.NewInt32(common.DaemonStatusInitialized),
+		firstBucket:           true,
+		counter:               atomic.NewInt64(0),
+		qps:                   atomic.NewFloat64(0),
+	}
+}
+
+func (r *emaFixedWindowQPSTracker) Start() {
+	if !r.status.CompareAndSwap(common.DaemonStatusInitialized, common.DaemonStatusStarted) {
+		return
+	}
+	r.wg.Add(1)
+	go r.reportLoop()
+}
+
+func (r *emaFixedWindowQPSTracker) reportLoop() {
+	defer r.wg.Done()
+	ticker := r.timeSource.NewTicker(r.bucketInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.Chan():
+			r.report()
+		case <-r.done:
+			return
+		}
+	}
+}
+
+func (r *emaFixedWindowQPSTracker) report() {
+	if r.firstBucket {
+		counter := r.counter.Swap(0)
+		r.qps.Store(float64(counter) / r.bucketIntervalSeconds)
+		r.firstBucket = false
+		return
+	}
+	counter := r.counter.Swap(0)
+	qps := r.qps.Load()
+	r.qps.Store(qps*(1-r.exp) + float64(counter)*r.exp/r.bucketIntervalSeconds)
+}
+
+func (r *emaFixedWindowQPSTracker) Stop() {
+	if !r.status.CompareAndSwap(common.DaemonStatusStarted, common.DaemonStatusStopped) {
+		return
+	}
+	close(r.done)
+	r.wg.Wait()
+}
+
+func (r *emaFixedWindowQPSTracker) ReportCounter(delta int64) {
+	r.counter.Add(delta)
+}
+
+func (r *emaFixedWindowQPSTracker) QPS() float64 {
+	return r.qps.Load()
+}

common/stats/stats_benchmark_test.go

@@ -0,0 +1,111 @@
+// 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 stats
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/uber/cadence/common/clock"
+)
+
+// Benchmark the ReportCounter function to see how it handles frequent updates
+func BenchmarkReportCounter(b *testing.B) {
+	timeSource := clock.NewRealTimeSource()
+	// Initialize the QPS reporter with a smoothing factor and a 1 second bucket interval
+	reporter := NewEmaFixedWindowQPSTracker(timeSource, 0.5, time.Second)
+	reporter.Start()
+
+	// Run the benchmark for b.N iterations
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		reporter.ReportCounter(1)
+	}
+
+	// Stop the reporter after the benchmark
+	b.StopTimer()
+	reporter.Stop()
+}
+
+// Benchmark the QPS calculation function under high load
+func BenchmarkQPS(b *testing.B) {
+	timeSource := clock.NewRealTimeSource()
+	// Initialize the QPS reporter
+	reporter := NewEmaFixedWindowQPSTracker(timeSource, 0.5, time.Second)
+	reporter.Start()
+
+	// Simulate a number of report updates before calling QPS
+	for i := 0; i < 1000; i++ {
+		reporter.ReportCounter(1)
+	}
+
+	// Benchmark QPS retrieval
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_ = reporter.QPS()
+	}
+
+	// Stop the reporter
+	b.StopTimer()
+	reporter.Stop()
+}
+
+// Benchmark the full reporting loop, simulating a real-time system.
+func BenchmarkFullReport(b *testing.B) {
+	timeSource := clock.NewRealTimeSource()
+	// Initialize the QPS reporter
+	reporter := NewEmaFixedWindowQPSTracker(timeSource, 0.5, time.Millisecond*100) // 100ms bucket interval
+	reporter.Start()
+
+	var wg sync.WaitGroup
+	// Number of goroutines for each task
+	numReporters := 10
+	numQPSQueries := 10
+	b.ResetTimer()
+
+	for i := 0; i < numReporters; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for j := 0; j < b.N; j++ {
+				// Report random counter value (simulate workload)
+				reporter.ReportCounter(1)
+			}
+		}()
+	}
+	for i := 0; i < numQPSQueries; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for j := 0; j < b.N; j++ {
+				// Query QPS value (simulate workload)
+				_ = reporter.QPS()
+			}
+		}()
+	}
+	wg.Wait()
+	// Stop the reporter after the benchmark
+	b.StopTimer()
+	reporter.Stop()
+}

common/stats/stats_test.go

@@ -0,0 +1,71 @@
+// 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 stats
+
+import (
+	"testing"
+	"time"
+
+	"github.com/uber/cadence/common/clock"
+)
+
+func TestEmaFixedWindowQPSTracker(t *testing.T) {
+	timeSource := clock.NewMockedTimeSourceAt(time.Now())
+	exp := 0.4
+	bucketInterval := time.Second
+
+	r := NewEmaFixedWindowQPSTracker(timeSource, exp, bucketInterval)
+	r.Start()
+
+	// Test ReportCounter
+	r.ReportCounter(10)
+	r.ReportCounter(20)
+
+	qps := r.QPS()
+	if qps != 0 {
+		t.Errorf("QPS mismatch, expected: 0, got: %f", qps)
+	}
+
+	timeSource.BlockUntil(1)
+	timeSource.Advance(bucketInterval)
+	time.Sleep(10 * time.Millisecond)
+	// Test QPS
+	qps = r.QPS()
+	expectedQPS := float64(30) / (float64(bucketInterval) / float64(time.Second))
+	if qps != expectedQPS {
+		t.Errorf("QPS mismatch, expected: %f, got: %f", expectedQPS, qps)
+	}
+
+	r.ReportCounter(10)
+	timeSource.BlockUntil(1)
+	timeSource.Advance(bucketInterval)
+	time.Sleep(10 * time.Millisecond)
+	// Test QPS
+	qps = r.QPS()
+	expectedQPS = float64(22) / (float64(bucketInterval) / float64(time.Second))
+	if qps != expectedQPS {
+		t.Errorf("QPS mismatch, expected: %f, got: %f", expectedQPS, qps)
+	}
+
+	r.Stop()
+}