Change max namespace count limit to per poller type
@@ -51,7 +51,7 @@ type (
tokens map[string]int
sync.Mutex
- namespaceToCount map[namespace.Name]*int32
+ activeTokensCount map[string]*int32
}
)
@@ -68,8 +68,7 @@ func NewNamespaceCountLimitInterceptor(
logger: logger,
countFn: countFn,
tokens: tokens,
-
- namespaceToCount: make(map[namespace.Name]*int32),
+ activeTokensCount: make(map[string]*int32),
}
}
@@ -84,13 +83,15 @@ func (ni *NamespaceCountLimitInterceptor) Intercept(
token := ni.tokens[methodName]
if token != 0 {
nsName := GetNamespace(ni.namespaceRegistry, req)
- counter := ni.counter(nsName)
+ counter := ni.counter(nsName, methodName)
count := atomic.AddInt32(counter, int32(token))
defer atomic.AddInt32(counter, -int32(token))
handler := GetMetricsHandlerFromContext(ctx, ni.logger)
handler.Gauge(metrics.ServicePendingRequests.GetMetricName()).Record(float64(count))
+ // frontend.namespaceCount is applied per poller type temporarily to prevent
+ // one poller type to take all token waiting in the long poll.
if int(count) > ni.countFn(nsName.String()) {
return nil, ErrNamespaceCountLimitServerBusy
}
@@ -101,15 +102,24 @@ func (ni *NamespaceCountLimitInterceptor) Intercept(
func (ni *NamespaceCountLimitInterceptor) counter(
namespace namespace.Name,
+ methodName string,
) *int32 {
+ key := ni.getTokenKey(namespace, methodName)
+
ni.Lock()
defer ni.Unlock()
- count, ok := ni.namespaceToCount[namespace]
+ counter, ok := ni.activeTokensCount[key]
if !ok {
- counter := int32(0)
- count = &counter
- ni.namespaceToCount[namespace] = count
+ counter = new(int32)
+ ni.activeTokensCount[key] = counter
+ }
+ return counter
}
- return count
+
+func (ni *NamespaceCountLimitInterceptor) getTokenKey(
+ namespace namespace.Name,
+ methodName string,
+) string {
+ return namespace.String() + "/" + methodName
}
