DeforaNetworks/corteza
3
0
Fork 0
Code Activity

Tweakup performance monitoring

Browse Source
This commit is contained in:
Tomaž Jerman
2024-12-03 13:44:48 +01:00
parent 08c8f29dca
commit ac57c68fcf
2 changed files with 156 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

105
server/pkg/rbac/service.go
View File

@@ -102,18 +102,22 @@ type (
}
Stats struct {
CacheHits uint `json:"cacheHits"`
CacheMisses uint `json:"cacheMisses"`
CacheUpdates uint `json:"cacheUpdates"`
AvgTiming time.Duration `json:"avgTiming"`
MinTiming time.Duration `json:"minTiming"`
MaxTiming time.Duration `json:"maxTiming"`
CacheHits uint `json:"cacheHits"`
CacheMisses uint `json:"cacheMisses"`
CacheUpdates uint `json:"cacheUpdates"`
AvgDbTiming time.Duration `json:"avgDbTiming"`
MinDbTiming time.Duration `json:"minDbTiming"`
MaxDbTiming time.Duration `json:"maxDbTiming"`
AvgIndexTiming time.Duration `json:"avgIndexTiming"`
MinIndexTiming time.Duration `json:"minIndexTiming"`
MaxIndexTiming time.Duration `json:"maxIndexTiming"`
IndexSize int `json:"indexSize"`
LastHits []string `json:"lastHits"`
LastMisses []string `json:"lastMisses"`
LastTimings []time.Duration `json:"lastTimings"`
LastHits []string `json:"lastHits"`
LastMisses []string `json:"lastMisses"`
LastDbTimings []time.Duration `json:"lastDbTimings"`
LastIndexTimings []time.Duration `json:"lastIndexTimings"`
Counters []expCtrItem `json:"counters"`
}
@@ -232,10 +236,11 @@ func initUsageCounter(ctx context.Context, cc Config) (svc *usageCounter[string]
func initStatsLogger(ctx context.Context, l *zap.Logger) (svc *StatsLogger) {
svc = &StatsLogger{
log: l.Named("rbac stats logger"),
cacheHitChan: make(chan statsWrap, 1024),
cacheMissChan: make(chan statsWrap, 1024),
timingChan: make(chan time.Duration, 1024),
log: l.Named("rbac stats logger"),
cacheHitChan: make(chan statsWrap, 1024),
cacheMissChan: make(chan statsWrap, 1024),
timingDatabaseChan: make(chan time.Duration, 1024),
timingIndexChan: make(chan time.Duration, 1024),
}
svc.watch(ctx)
@@ -427,12 +432,16 @@ func (svc *Service) Stats() (out Stats, err error) {
out.CacheHits,
out.CacheMisses,
out.CacheUpdates,
out.AvgTiming,
out.MinTiming,
out.MaxTiming,
out.AvgDbTiming,
out.MinDbTiming,
out.MaxDbTiming,
out.AvgIndexTiming,
out.MinIndexTiming,
out.MaxIndexTiming,
out.LastHits,
out.LastMisses,
out.LastTimings = svc.StatLogger.Stats()
out.LastDbTimings,
out.LastIndexTimings = svc.StatLogger.Stats()
out.IndexSize = svc.index.getSize()
@@ -576,7 +585,7 @@ func (svc *Service) check(ctx context.Context, rolesByKind partRoles, op, res st
return Inherit, err
}
svc.logDbTiming(timing)
svc.logDatabaseTiming(timing)
a, err = svc.evaluate(
[]roleKind{ContextRole, CommonRole, AuthenticatedRole, AnonymousRole},
@@ -833,7 +842,10 @@ func (svc *Service) getMatchingRule(st evaluationState, kind roleKind, role uint
)
// Indexed
now := time.Now()
aux = svc.index.get(role, st.op, st.res)
svc.logIndexTiming(time.Since(now))
rules = append(rules, aux...)
// Unindexed
@@ -1078,21 +1090,39 @@ func (svc *Service) swapIndexes(auxIndex *wrapperIndex) {
}
// Performance monitoring
func (svc *Service) logDbTiming(timing time.Duration) {
func (svc *Service) logDatabaseTiming(timing time.Duration) {
if svc.cfg.Synchronous {
svc.logAccessSync(timing)
svc.logDatabaseSync(timing)
} else {
svc.logAccessAsync(timing)
svc.logDatabaseAsync(timing)
}
}
func (svc *Service) logAccessSync(timing time.Duration) {
svc.StatLogger.Timing(timing)
func (svc *Service) logDatabaseSync(timing time.Duration) {
svc.StatLogger.TimingDatabase(timing)
}
func (svc *Service) logAccessAsync(timing time.Duration) {
if svc.StatLogger != nil && svc.StatLogger.timingChan != nil {
svc.StatLogger.timingChan <- timing
func (svc *Service) logDatabaseAsync(timing time.Duration) {
if svc.StatLogger != nil && svc.StatLogger.timingDatabaseChan != nil {
svc.StatLogger.timingDatabaseChan <- timing
}
}
func (svc *Service) logIndexTiming(timing time.Duration) {
if svc.cfg.Synchronous {
svc.logIndexSync(timing)
} else {
svc.logIndexAsync(timing)
}
}
func (svc *Service) logIndexSync(timing time.Duration) {
svc.StatLogger.TimingIndex(timing)
}
func (svc *Service) logIndexAsync(timing time.Duration) {
if svc.StatLogger != nil && svc.StatLogger.timingIndexChan != nil {
svc.StatLogger.timingIndexChan <- timing
}
}
@@ -1200,37 +1230,26 @@ func (svc *Service) DebuggerAddIndex(role uint64, resource string, rules ...*Rul
func (svc *Service) watch(ctx context.Context) {
tck := time.NewTicker(time.Minute * 5)
tInt := svc.cfg.IndexFlushInterval
if tInt == 0 {
tInt = time.Minute * 5
}
tTck := time.NewTicker(tInt)
_ = tTck
flushInt := svc.cfg.IndexFlushInterval
if flushInt == 0 {
flushInt = time.Minute * 30
}
flushTck := time.NewTicker(flushInt)
_ = flushTck
rexInt := svc.cfg.ReindexInterval
if rexInt == 0 {
rexInt = time.Minute * 30
}
rexTck := time.NewTicker(rexInt)
_ = rexTck
defer func() {
tck.Stop()
tTck.Stop()
flushTck.Stop()
rexTck.Stop()
}()
lg := svc.logger.Named("rbac service wrapper")
go func() {
defer func() {
tck.Stop()
flushTck.Stop()
rexTck.Stop()
}()
for {
select {
case <-tck.C:

128
server/pkg/rbac/stats.go
View File

@@ -17,23 +17,28 @@ type (
log *zap.Logger
// Channels for async comms
cacheHitChan chan statsWrap
cacheMissChan chan statsWrap
timingChan chan time.Duration
cacheHitChan chan statsWrap
cacheMissChan chan statsWrap
timingDatabaseChan chan time.Duration
timingIndexChan chan time.Duration
// Counters
cacheHits uint
cacheMisses uint
cacheUpdates uint
avgTiming time.Duration
minTiming time.Duration
maxTiming time.Duration
cacheHits uint
cacheMisses uint
cacheUpdates uint
avgDatabaseTiming time.Duration
minDatabaseTiming time.Duration
maxDatabaseTiming time.Duration
avgIndexTiming time.Duration
minIndexTiming time.Duration
maxIndexTiming time.Duration
// Track a limited set of things
// Using a circular buffer we can easily not consume too much data
lastHits *slice.Circular[string]
lastMisses *slice.Circular[string]
lastTimings *slice.Circular[time.Duration]
lastHits *slice.Circular[string]
lastMisses *slice.Circular[string]
lastDatabaseTimings *slice.Circular[time.Duration]
lastIndexTimings *slice.Circular[time.Duration]
}
// statsWrap wraps the state to log
@@ -45,56 +50,108 @@ type (
)
// Stats returns the tracked stats
func (l *StatsLogger) Stats() (cacheHit uint, cacheMiss uint, cacheUpdates uint, avgTiming, minTiming, maxTiming time.Duration, lastHits []string, lastMisses []string, lastTimings []time.Duration) {
func (l *StatsLogger) Stats() (
cacheHit uint,
cacheMiss uint,
cacheUpdates uint,
avgDbTiming, minDbTiming, maxDbTiming time.Duration,
avgIndexTiming, minIndexTiming, maxIndexTiming time.Duration,
lastHits []string,
lastMisses []string,
lastDbTimings []time.Duration,
lastIndexTimings []time.Duration,
) {
l.lock.RLock()
defer l.lock.RUnlock()
return l.cacheHits,
l.cacheMisses,
l.cacheUpdates,
l.avgTiming,
l.minTiming,
l.maxTiming,
l.avgDatabaseTiming,
l.minDatabaseTiming,
l.maxDatabaseTiming,
l.avgIndexTiming,
l.minIndexTiming,
l.maxIndexTiming,
l.lastHits.Slice(),
l.lastMisses.Slice(),
l.lastTimings.Slice()
l.lastDatabaseTimings.Slice(),
l.lastIndexTimings.Slice()
}
// Timing logs the giving duration
func (l *StatsLogger) Timing(timing time.Duration) {
// TimingDatabase logs the giving duration
func (l *StatsLogger) TimingDatabase(timing time.Duration) {
l.lock.Lock()
defer l.lock.Unlock()
l.log.Info("record timing", zap.Duration("timing", timing))
l.log.Info("record database timing", zap.Duration("timing", timing))
{
l.avgTiming = (l.avgTiming + timing) / 2
l.avgDatabaseTiming = (l.avgDatabaseTiming + timing) / 2
}
{
if l.minTiming == 0 {
l.minTiming = timing
if l.minDatabaseTiming == 0 {
l.minDatabaseTiming = timing
}
if timing < l.minTiming {
l.minTiming = timing
if timing < l.minDatabaseTiming {
l.minDatabaseTiming = timing
}
}
{
if l.maxTiming == 0 {
l.maxTiming = timing
if l.maxDatabaseTiming == 0 {
l.maxDatabaseTiming = timing
}
if timing > l.maxTiming {
l.maxTiming = timing
if timing > l.maxDatabaseTiming {
l.maxDatabaseTiming = timing
}
}
{
if l.lastTimings == nil {
l.lastTimings = slice.NewCircular[time.Duration](500)
if l.lastDatabaseTimings == nil {
l.lastDatabaseTimings = slice.NewCircular[time.Duration](500)
}
l.lastTimings.Add(timing)
l.lastDatabaseTimings.Add(timing)
}
}
// TimingIndex logs the giving duration
func (l *StatsLogger) TimingIndex(timing time.Duration) {
l.lock.Lock()
defer l.lock.Unlock()
l.log.Info("record index timing", zap.Duration("timing", timing))
{
l.avgIndexTiming = (l.avgIndexTiming + timing) / 2
}
{
if l.minIndexTiming == 0 {
l.minIndexTiming = timing
}
if timing < l.minIndexTiming {
l.minIndexTiming = timing
}
}
{
if l.maxIndexTiming == 0 {
l.maxIndexTiming = timing
}
if timing > l.maxIndexTiming {
l.maxIndexTiming = timing
}
}
{
if l.lastIndexTimings == nil {
l.lastIndexTimings = slice.NewCircular[time.Duration](500)
}
l.lastIndexTimings.Add(timing)
}
}
@@ -157,8 +214,11 @@ func (l *StatsLogger) watch(ctx context.Context) {
case rs := <-l.cacheHitChan:
l.CacheHit(rs.roles, rs.resource, rs.op)
case tt := <-l.timingChan:
l.Timing(tt)
case tt := <-l.timingDatabaseChan:
l.TimingDatabase(tt)
case tt := <-l.timingIndexChan:
l.TimingIndex(tt)
case <-ctx.Done():
l.log.Info("terminating watcher")