Add global ratelimiter for frontend API (cadence-workflow#3161)

* Add global ratelimiter for frontend API

common/membership/interfaces.go

@@ -70,6 +70,9 @@ type (
 		RemoveListener(service string, name string) error
 		// GetReachableMembers returns addresses of all members of the ring
 		GetReachableMembers() ([]string, error)
+		// GetMemberCount returns the number of reachable members
+		// currently in this node's membership list for the given role
+		GetMemberCount(role string) (int, error)
 	}
 
 	// ServiceResolver provides membership information for a specific cadence service.

common/membership/rpMonitor.go

@@ -154,3 +154,11 @@ func (rpo *ringpopMonitor) RemoveListener(service string, name string) error {
 func (rpo *ringpopMonitor) GetReachableMembers() ([]string, error) {
 	return rpo.rp.GetReachableMembers()
 }
+
+func (rpo *ringpopMonitor) GetMemberCount(service string) (int, error) {
+	ring, err := rpo.GetResolver(service)
+	if err != nil {
+		return 0, err
+	}
+	return ring.MemberCount(), nil
+}

common/service/dynamicconfig/constants.go

@@ -89,7 +89,8 @@ var keys = map[Key]string{
 	FrontendESIndexMaxResultWindow:        "frontend.esIndexMaxResultWindow",
 	FrontendHistoryMaxPageSize:            "frontend.historyMaxPageSize",
 	FrontendRPS:                           "frontend.rps",
-	FrontendDomainRPS:                     "frontend.domainrps",
+	FrontendMaxDomainRPSPerInstance:       "frontend.domainrps",
+	FrontendGlobalDomainRPS:               "frontend.globalDomainrps",
 	FrontendHistoryMgrNumConns:            "frontend.historyMgrNumConns",
 	FrontendShutdownDrainDuration:         "frontend.shutdownDrainDuration",
 	DisableListVisibilityByFilter:         "frontend.disableListVisibilityByFilter",
@@ -352,8 +353,10 @@ const (
 	FrontendHistoryMaxPageSize
 	// FrontendRPS is workflow rate limit per second
 	FrontendRPS
-	// FrontendDomainRPS is workflow domain rate limit per second
-	FrontendDomainRPS
+	// FrontendMaxDomainRPSPerInstance is workflow domain rate limit per second
+	FrontendMaxDomainRPSPerInstance
+	// FrontendGlobalDomainRPS is workflow domain rate limit per second for the whole Cadence cluster
+	FrontendGlobalDomainRPS
 	// FrontendHistoryMgrNumConns is for persistence cluster.NumConns
 	FrontendHistoryMgrNumConns
 	// FrontendThrottledLogRPS is the rate limit on number of log messages emitted per second for throttled logger

host/simpleMonitor.go

@@ -78,3 +78,7 @@ func (s *simpleMonitor) RemoveListener(service string, name string) error {
 func (s *simpleMonitor) GetReachableMembers() ([]string, error) {
 	return nil, nil
 }
+
+func (s *simpleMonitor) GetMemberCount(service string) (int, error) {
+	return 0, nil
+}

service/frontend/service.go

@@ -55,7 +55,8 @@ type Config struct {
 	ESIndexMaxResultWindow          dynamicconfig.IntPropertyFn
 	HistoryMaxPageSize              dynamicconfig.IntPropertyFnWithDomainFilter
 	RPS                             dynamicconfig.IntPropertyFn
-	DomainRPS                       dynamicconfig.IntPropertyFnWithDomainFilter
+	MaxDomainRPSPerInstance         dynamicconfig.IntPropertyFnWithDomainFilter
+	GlobalDomainRPS                 dynamicconfig.IntPropertyFnWithDomainFilter
 	MaxIDLengthLimit                dynamicconfig.IntPropertyFn
 	EnableClientVersionCheck        dynamicconfig.BoolPropertyFn
 	MinRetentionDays                dynamicconfig.IntPropertyFn
@@ -107,7 +108,8 @@ func NewConfig(dc *dynamicconfig.Collection, numHistoryShards int, enableReadFro
 		ESIndexMaxResultWindow:              dc.GetIntProperty(dynamicconfig.FrontendESIndexMaxResultWindow, 10000),
 		HistoryMaxPageSize:                  dc.GetIntPropertyFilteredByDomain(dynamicconfig.FrontendHistoryMaxPageSize, common.GetHistoryMaxPageSize),
 		RPS:                                 dc.GetIntProperty(dynamicconfig.FrontendRPS, 1200),
-		DomainRPS:                           dc.GetIntPropertyFilteredByDomain(dynamicconfig.FrontendDomainRPS, 1200),
+		MaxDomainRPSPerInstance:             dc.GetIntPropertyFilteredByDomain(dynamicconfig.FrontendMaxDomainRPSPerInstance, 1200),
+		GlobalDomainRPS:                     dc.GetIntPropertyFilteredByDomain(dynamicconfig.FrontendGlobalDomainRPS, 0),
 		MaxIDLengthLimit:                    dc.GetIntProperty(dynamicconfig.MaxIDLengthLimit, 1000),
 		HistoryMgrNumConns:                  dc.GetIntProperty(dynamicconfig.FrontendHistoryMgrNumConns, 10),
 		MaxBadBinaries:                      dc.GetIntPropertyFilteredByDomain(dynamicconfig.FrontendMaxBadBinaries, domain.MaxBadBinaries),

service/frontend/workflowHandler.go

@@ -160,7 +160,14 @@ func NewWorkflowHandler(
 				return float64(config.RPS())
 			},
 			func(domain string) float64 {
-				return float64(config.DomainRPS(domain))
+				if monitor := resource.GetMembershipMonitor(); monitor != nil && config.GlobalDomainRPS(domain) > 0 {
+					ringSize, err := monitor.GetMemberCount(common.FrontendServiceName)
+					if err == nil && ringSize > 0 {
+						avgQuota := common.MaxInt(config.GlobalDomainRPS(domain)/ringSize, 1)
+						return float64(common.MinInt(avgQuota, config.MaxDomainRPSPerInstance(domain)))
+					}
+				}
+				return float64(config.MaxDomainRPSPerInstance(domain))
 			},
 		),
 		versionChecker: client.NewVersionChecker(),

service/frontend/workflowHandler_test.go

@@ -115,6 +115,10 @@ func (s *workflowHandlerSuite) SetupTest() {
 	s.mockMessagingClient = mocks.NewMockMessagingClient(s.mockProducer, nil)
 	s.mockHistoryArchiver = &archiver.HistoryArchiverMock{}
 	s.mockVisibilityArchiver = &archiver.VisibilityArchiverMock{}
+
+	mockMonitor := s.mockResource.MembershipMonitor
+	mockMonitor.EXPECT().GetMemberCount(common.FrontendServiceName).Return(5, nil).AnyTimes()
+
 }
 
 func (s *workflowHandlerSuite) TearDownTest() {