Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Per shard per namespace RPS warning log #4525

Merged
merged 8 commits into from
Jun 27, 2023
Merged

Per shard per namespace RPS warning log #4525

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
364d15f
add warning log for per shard per ns rps
pdoerner Jun 21, 2023
d6ff63e
Merge branch 'master' into per-shard-per-ns-warning
pdoerner Jun 21, 2023
29b9f72
handle nil map key
pdoerner Jun 21, 2023
f937a1b
Merge branch 'master' into per-shard-per-ns-warning
pdoerner Jun 21, 2023
dc0c575
feedback
pdoerner Jun 23, 2023
81a9591
Merge branch 'master' into per-shard-per-ns-warning
pdoerner Jun 23, 2023
b6e73c5
config name
pdoerner Jun 27, 2023
a59801a
Merge branch 'master' into per-shard-per-ns-warning
pdoerner Jun 27, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions common/dynamicconfig/constants.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -116,6 +116,8 @@ const (
PersistenceHealthSignalBufferSize = "system.persistenceHealthSignalBufferSize"
// ShardRPSWarnLimit is the per-shard RPS limit for warning
ShardRPSWarnLimit = "system.shardRPSWarnLimit"
// ShardPerNsRPSWarnLimit is the per-shard per-namespace RPS limit for warning
ShardPerNsRPSWarnLimit = "system.shardPerNsRPSWarnLimit"

// Whether the deadlock detector should dump goroutines
DeadlockDumpGoroutines = "system.deadlock.DumpGoroutines"
Expand Down
1 change: 1 addition & 0 deletions common/persistence/client/fx.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,7 @@ func HealthSignalAggregatorProvider(
dynamicCollection.GetIntProperty(dynamicconfig.PersistenceHealthSignalBufferSize, 5000)(),
metricsHandler,
dynamicCollection.GetIntProperty(dynamicconfig.ShardRPSWarnLimit, 50),
dynamicCollection.GetIntProperty(dynamicconfig.ShardPerNsRPSWarnLimit, 50),
logger,
)
}
Expand Down
60 changes: 38 additions & 22 deletions common/persistence/health_signal_aggregator.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ const (
type (
HealthSignalAggregator interface {
common.Daemon
Record(callerSegment int32, latency time.Duration, err error)
Record(callerSegment int32, namespace string, latency time.Duration, err error)
yycptt marked this conversation as resolved.
Show resolved Hide resolved
AverageLatency() float64
ErrorRatio() float64
}
Expand All @@ -53,16 +53,18 @@ type (
status int32
shutdownCh chan struct{}

requestsPerShard map[int32]int64
requestsLock sync.Mutex
// map of shardID -> map of namespace -> request count
requestCounts map[int32]map[string]int64
requestsLock sync.Mutex

aggregationEnabled bool
latencyAverage aggregate.MovingWindowAverage
errorRatio aggregate.MovingWindowAverage

metricsHandler metrics.Handler
emitMetricsTimer *time.Ticker
perShardRPSWarnLimit dynamicconfig.IntPropertyFn
metricsHandler metrics.Handler
emitMetricsTimer *time.Ticker
perShardRPSWarnLimit dynamicconfig.IntPropertyFn
perShardPerNsRPSWarnLimit dynamicconfig.IntPropertyFn

logger log.Logger
}
Expand All @@ -74,17 +76,19 @@ func NewHealthSignalAggregatorImpl(
maxBufferSize int,
metricsHandler metrics.Handler,
perShardRPSWarnLimit dynamicconfig.IntPropertyFn,
perShardPerNsRPSWarnLimit dynamicconfig.IntPropertyFn,
logger log.Logger,
) *HealthSignalAggregatorImpl {
ret := &HealthSignalAggregatorImpl{
status: common.DaemonStatusInitialized,
shutdownCh: make(chan struct{}),
requestsPerShard: make(map[int32]int64),
metricsHandler: metricsHandler,
emitMetricsTimer: time.NewTicker(emitMetricsInterval),
perShardRPSWarnLimit: perShardRPSWarnLimit,
logger: logger,
aggregationEnabled: aggregationEnabled,
status: common.DaemonStatusInitialized,
shutdownCh: make(chan struct{}),
requestCounts: make(map[int32]map[string]int64),
metricsHandler: metricsHandler,
emitMetricsTimer: time.NewTicker(emitMetricsInterval),
perShardRPSWarnLimit: perShardRPSWarnLimit,
perShardPerNsRPSWarnLimit: perShardPerNsRPSWarnLimit,
logger: logger,
aggregationEnabled: aggregationEnabled,
}

if aggregationEnabled {
Expand Down Expand Up @@ -113,7 +117,7 @@ func (s *HealthSignalAggregatorImpl) Stop() {
s.emitMetricsTimer.Stop()
}

func (s *HealthSignalAggregatorImpl) Record(callerSegment int32, latency time.Duration, err error) {
func (s *HealthSignalAggregatorImpl) Record(callerSegment int32, namespace string, latency time.Duration, err error) {
if s.aggregationEnabled {
s.latencyAverage.Record(latency.Milliseconds())

Expand All @@ -125,7 +129,7 @@ func (s *HealthSignalAggregatorImpl) Record(callerSegment int32, latency time.Du
}

if callerSegment != CallerSegmentMissing {
s.incrementShardRequestCount(callerSegment)
s.incrementShardRequestCount(callerSegment, namespace)
}
}

Expand All @@ -137,10 +141,13 @@ func (s *HealthSignalAggregatorImpl) ErrorRatio() float64 {
return s.errorRatio.Average()
}

func (s *HealthSignalAggregatorImpl) incrementShardRequestCount(shardID int32) {
func (s *HealthSignalAggregatorImpl) incrementShardRequestCount(shardID int32, namespace string) {
s.requestsLock.Lock()
defer s.requestsLock.Unlock()
s.requestsPerShard[shardID]++
if s.requestCounts[shardID] == nil {
s.requestCounts[shardID] = make(map[string]int64)
}
s.requestCounts[shardID][namespace]++
}

func (s *HealthSignalAggregatorImpl) emitMetricsLoop() {
Expand All @@ -150,12 +157,21 @@ func (s *HealthSignalAggregatorImpl) emitMetricsLoop() {
return
case <-s.emitMetricsTimer.C:
s.requestsLock.Lock()
requestCounts := s.requestsPerShard
s.requestsPerShard = make(map[int32]int64, len(requestCounts))
requestCounts := s.requestCounts
s.requestCounts = make(map[int32]map[string]int64, len(requestCounts))
s.requestsLock.Unlock()

for shardID, count := range requestCounts {
shardRPS := int64(float64(count) / emitMetricsInterval.Seconds())
for shardID, requestCountPerNS := range requestCounts {
shardRequestCount := int64(0)
for namespace, count := range requestCountPerNS {
shardRequestCount += count
shardRPSPerNS := int64(float64(count) / emitMetricsInterval.Seconds())
if shardRPSPerNS > int64(s.perShardPerNsRPSWarnLimit()) {
yycptt marked this conversation as resolved.
Show resolved Hide resolved
s.logger.Warn("Per shard per namespace RPS warn limit exceeded", tag.ShardID(shardID), tag.WorkflowNamespace(namespace))
yycptt marked this conversation as resolved.
Show resolved Hide resolved
}
}

shardRPS := int64(float64(shardRequestCount) / emitMetricsInterval.Seconds())
s.metricsHandler.Histogram(metrics.PersistenceShardRPS.GetMetricName(), metrics.PersistenceShardRPS.GetMetricUnit()).Record(shardRPS)
if shardRPS > int64(s.perShardRPSWarnLimit()) {
s.logger.Warn("Per shard RPS warn limit exceeded", tag.ShardID(shardID))
Expand Down
2 changes: 1 addition & 1 deletion common/persistence/noop_health_signal_aggregator.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ func (a *noopSignalAggregator) Start() {}

func (a *noopSignalAggregator) Stop() {}

func (a *noopSignalAggregator) Record(_ int32, _ time.Duration, _ error) {}
func (a *noopSignalAggregator) Record(_ int32, _ string, _ time.Duration, _ error) {}

func (a *noopSignalAggregator) AverageLatency() float64 {
return 0
Expand Down
Loading