From b7898b0a22dcf6a109ea27e2b804b6ebf35badc0 Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Fri, 2 Jul 2021 11:31:37 -0500 Subject: [PATCH 1/5] Allow usage-duty-cycle < 20%, and 0 disables --- api.go | 3 --- server/server.go | 4 +++- 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/api.go b/api.go index ecb9783bd..2df5eb14f 100644 --- a/api.go +++ b/api.go @@ -1067,9 +1067,6 @@ func (api *API) RefreshUsageCache(dutyCycle float64) { trigger := make(chan bool) defer close(trigger) - if dutyCycle <= 0 { - dutyCycle = 20 - } multiplier := 100/dutyCycle - 1 api.usageCache = &usageCache{ diff --git a/server/server.go b/server/server.go index 84378f00b..11fbe1522 100644 --- a/server/server.go +++ b/server/server.go @@ -277,7 +277,9 @@ func (m *Command) Start() (err error) { } } - go m.API.RefreshUsageCache(m.Config.UsageDutyCycle) + if m.Config.UsageDutyCycle > 0 { + go m.API.RefreshUsageCache(m.Config.UsageDutyCycle) + } _ = testhook.Opened(pilosa.NewAuditor(), m, nil) close(m.Started) From f1e12567a3f25fead8c5bf1a0485e730e4f117cf Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Wed, 7 Jul 2021 13:59:45 -0500 Subject: [PATCH 2/5] Define some constants for usageCache --- api.go | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/api.go b/api.go index 2df5eb14f..83022dcf6 100644 --- a/api.go +++ b/api.go @@ -914,6 +914,11 @@ type usageCache struct { muAssign sync.Mutex } +var usageCacheMinDuration = 5 * time.Second // If usage takes less than this duration to calculate, don't use the cache. +var usageCacheMinInterval = time.Hour // Refresh interval is forced to be >= this duration. +var usageCacheInitialInterval = time.Hour // Refresh interval starts with this duration. +var usageCacheDebugSleep = 10 * time.Second + // NodeUsage represents all usage measurements for one node. type NodeUsage struct { Disk DiskUsage `json:"diskUsage"` @@ -957,7 +962,7 @@ func (api *API) Usage(ctx context.Context, remote bool) (map[string]NodeUsage, e span, _ := tracing.StartSpanFromContext(ctx, "API.Usage") defer span.Finish() - if api.usageCache.lastCalcDuration < (time.Second * 5) { + if api.usageCache.lastCalcDuration < usageCacheMinDuration { err := api.ResetUsageCache() if err != nil { api.server.logger.Infof("could not reset usageCache: %s", err) @@ -974,6 +979,7 @@ func (api *API) Usage(ctx context.Context, remote bool) (map[string]NodeUsage, e if !remote { api.requestUsageOfNodes() } + time.Sleep(usageCacheDebugSleep) return api.usageCache.data, nil } @@ -1071,7 +1077,7 @@ func (api *API) RefreshUsageCache(dutyCycle float64) { api.usageCache = &usageCache{ data: make(map[string]NodeUsage), - refreshInterval: time.Hour, + refreshInterval: usageCacheInitialInterval, resetTrigger: trigger, lastCalcDuration: 0, waitMultiplier: multiplier, @@ -1094,8 +1100,8 @@ func (api *API) RefreshUsageCache(dutyCycle float64) { // Refresh interval set in relation to how long the last calculation took. func (api *API) setRefreshInterval(dur time.Duration) { refresh := time.Duration(float64(dur) * api.usageCache.waitMultiplier) - if refresh < time.Hour { - refresh = time.Hour + if refresh < usageCacheMinInterval { + refresh = usageCacheMinInterval } api.usageCache.muAssign.Lock() api.usageCache.refreshInterval = refresh From 46df93c52173571789fdad37ecba6a4a594dd504 Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Wed, 7 Jul 2021 14:00:16 -0500 Subject: [PATCH 3/5] Unindent --- api.go | 99 +++++++++++++++++++++++++++++----------------------------- 1 file changed, 50 insertions(+), 49 deletions(-) diff --git a/api.go b/api.go index 83022dcf6..3361780fd 100644 --- a/api.go +++ b/api.go @@ -1014,56 +1014,57 @@ func (api *API) calculateUsage() { lastUpdated := api.usageCache.lastUpdated api.usageCache.muAssign.Unlock() - if time.Since(lastUpdated) > api.usageCache.refreshInterval { - indexDetails, nodeMetadataBytes, err := api.holder.Txf().IndexUsageDetails(api.isClosing) - if err != nil { - api.server.logger.Infof("couldn't get index usage details: %s", err) - } - if api.isClosing() { - return - } - - totalSize := nodeMetadataBytes - for _, s := range indexDetails { - totalSize += s.Total - } - - // NOTE: these errors are ignored in api.Info(), but checked here - si := api.server.systemInfo - diskCapacity, err := si.DiskCapacity(api.holder.path) - if err != nil { - api.server.logger.Infof("couldn't read disk capacity: %s", err) - } - - memoryCapacity, err := si.MemTotal() - if err != nil { - api.server.logger.Infof("couldn't read memory capacity: %s", err) - } - memoryUse, err := si.MemUsed() - if err != nil { - api.server.logger.Infof("couldn't read memory usage: %s", err) - } - - lastUpdated = time.Now() - // Insert into result. - nodeUsage := NodeUsage{ - Disk: DiskUsage{ - Capacity: diskCapacity, - TotalUse: totalSize, - IndexUsage: indexDetails, - }, - Memory: MemoryUsage{ - Capacity: memoryCapacity, - TotalUse: memoryUse, - }, - LastUpdated: lastUpdated, - } - api.usageCache.muAssign.Lock() - api.usageCache.data = make(map[string]NodeUsage) - api.usageCache.data[api.server.nodeID] = nodeUsage - api.usageCache.lastUpdated = lastUpdated - api.usageCache.muAssign.Unlock() + if time.Since(lastUpdated) <= api.usageCache.refreshInterval { + return } + indexDetails, nodeMetadataBytes, err := api.holder.Txf().IndexUsageDetails(api.isClosing) + if err != nil { + api.server.logger.Infof("couldn't get index usage details: %s", err) + } + if api.isClosing() { + return + } + + totalSize := nodeMetadataBytes + for _, s := range indexDetails { + totalSize += s.Total + } + + // NOTE: these errors are ignored in api.Info(), but checked here + si := api.server.systemInfo + diskCapacity, err := si.DiskCapacity(api.holder.path) + if err != nil { + api.server.logger.Infof("couldn't read disk capacity: %s", err) + } + + memoryCapacity, err := si.MemTotal() + if err != nil { + api.server.logger.Infof("couldn't read memory capacity: %s", err) + } + memoryUse, err := si.MemUsed() + if err != nil { + api.server.logger.Infof("couldn't read memory usage: %s", err) + } + + lastUpdated = time.Now() + // Insert into result. + nodeUsage := NodeUsage{ + Disk: DiskUsage{ + Capacity: diskCapacity, + TotalUse: totalSize, + IndexUsage: indexDetails, + }, + Memory: MemoryUsage{ + Capacity: memoryCapacity, + TotalUse: memoryUse, + }, + LastUpdated: lastUpdated, + } + api.usageCache.muAssign.Lock() + api.usageCache.data = make(map[string]NodeUsage) + api.usageCache.data[api.server.nodeID] = nodeUsage + api.usageCache.lastUpdated = lastUpdated + api.usageCache.muAssign.Unlock() } // Periodically calculates disk/memory usage in terms of the duty cycle. The duty cycle represents the percentage of From 60f2893152d947b15a23138dd2518100560ec41a Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Wed, 7 Jul 2021 14:34:12 -0500 Subject: [PATCH 4/5] Add usageCache logging --- api.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/api.go b/api.go index 3361780fd..cc9633d86 100644 --- a/api.go +++ b/api.go @@ -915,9 +915,8 @@ type usageCache struct { } var usageCacheMinDuration = 5 * time.Second // If usage takes less than this duration to calculate, don't use the cache. -var usageCacheMinInterval = time.Hour // Refresh interval is forced to be >= this duration. -var usageCacheInitialInterval = time.Hour // Refresh interval starts with this duration. -var usageCacheDebugSleep = 10 * time.Second +var usageCacheMinInterval = time.Hour // Refresh interval is forced to be >= this duration. +var usageCacheInitialInterval = time.Hour // Refresh interval starts with this duration. // NodeUsage represents all usage measurements for one node. type NodeUsage struct { @@ -979,7 +978,6 @@ func (api *API) Usage(ctx context.Context, remote bool) (map[string]NodeUsage, e if !remote { api.requestUsageOfNodes() } - time.Sleep(usageCacheDebugSleep) return api.usageCache.data, nil } @@ -1083,10 +1081,12 @@ func (api *API) RefreshUsageCache(dutyCycle float64) { lastCalcDuration: 0, waitMultiplier: multiplier, } + api.server.logger.Infof("monitoring resource usage with duty cycle %v%%\n", dutyCycle) for { start := time.Now() api.calculateUsage() api.setRefreshInterval(time.Since(start)) + api.server.logger.Infof("updated resource usage cache at %v, took %v, next update in %v\n", api.usageCache.lastUpdated.Format(time.RFC3339), api.usageCache.lastCalcDuration, api.usageCache.refreshInterval) select { case <-trigger: continue From 641506720017e5f3ea160cb87682f53a99a64b3a Mon Sep 17 00:00:00 2001 From: Alan Bernstein Date: Mon, 12 Jul 2021 15:27:51 -0700 Subject: [PATCH 5/5] Make duration print format more readable --- api.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/api.go b/api.go index cc9633d86..7cfa814e7 100644 --- a/api.go +++ b/api.go @@ -1086,7 +1086,7 @@ func (api *API) RefreshUsageCache(dutyCycle float64) { start := time.Now() api.calculateUsage() api.setRefreshInterval(time.Since(start)) - api.server.logger.Infof("updated resource usage cache at %v, took %v, next update in %v\n", api.usageCache.lastUpdated.Format(time.RFC3339), api.usageCache.lastCalcDuration, api.usageCache.refreshInterval) + api.server.logger.Infof("updated resource usage cache at %v, took %v, next update in %v\n", api.usageCache.lastUpdated.Format(time.RFC3339), api.usageCache.lastCalcDuration.Truncate(time.Millisecond), api.usageCache.refreshInterval.Truncate(100*time.Millisecond)) select { case <-trigger: continue