Add back GlobalNamespaceRPS
* Use GlobalNamespaceRPS for additional QPS allocated per namespace per host.
* Per frontend per namespace QPS: `hostRPS+globalRPS*math.Exp((1.0-hosts)/8.0)`
NOTE: above is not distributing tokens among all frontend hosts, but rather a temporary hack.
package frontend
import (
+ "math"
"os"
"sync/atomic"
"time"
@@ -35,6 +36,8 @@ import (
"google.golang.org/grpc/keepalive"
"google.golang.org/grpc/reflection"
+ "go.temporal.io/server/common/membership"
+
"go.temporal.io/server/service/frontend/configs"
"go.temporal.io/server/common/quotas"
@@ -273,7 +276,11 @@ func NewService(
quotas.NewNamespaceRateLimiter(
func(req quotas.Request) quotas.RequestRateLimiter {
return configs.NewRequestToRateLimiter(func() float64 {
- return float64(serviceConfig.MaxNamespaceRPSPerInstance(req.Caller))
+ return namespaceRPS(
+ serviceConfig,
+ serviceResource.GetFrontendServiceResolver(),
+ req.Caller,
+ )
})
},
),
@@ -407,3 +414,32 @@ func (s *Service) Stop() {
logger.Info("frontend stopped")
}
+
+func namespaceRPS(
+ config *Config,
+ frontendResolver membership.ServiceResolver,
+ namespace string,
+) float64 {
+ hostRPS := float64(config.MaxNamespaceRPSPerInstance(namespace))
+ globalRPS := float64(config.GlobalNamespaceRPS(namespace))
+ hosts := float64(numFrontendHosts(frontendResolver))
+
+ rps := hostRPS + globalRPS*math.Exp((1.0-hosts)/8.0)
+ return rps
+}
+
+func numFrontendHosts(
+ frontendResolver membership.ServiceResolver,
+) int {
+
+ defaultHosts := 1
+ if frontendResolver == nil {
+ return defaultHosts
+ }
+
+ ringSize := frontendResolver.MemberCount()
+ if ringSize < defaultHosts {
+ return defaultHosts
+ }
+ return ringSize
+}
