From 486244ab58cd2fdd5658d9deab2ed2eee66ea269 Mon Sep 17 00:00:00 2001 From: Samir Patel <48686912+54mir@users.noreply.github.com> Date: Thu, 10 Jun 2021 10:48:44 -0500 Subject: [PATCH 01/12] reset cache if outdated on call --- api.go | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) diff --git a/api.go b/api.go index 7bc31142e..59cde8d1f 100644 --- a/api.go +++ b/api.go @@ -957,7 +957,17 @@ func (api *API) Usage(ctx context.Context, remote bool) (map[string]NodeUsage, e api.usageCache.muAssign.Lock() lastUpdated := api.usageCache.lastUpdated + cacheN := len(api.usageCache.data[api.server.nodeID].Disk.IndexUsage) api.usageCache.muAssign.Unlock() + holderN := len(api.holder.Indexes()) + + // reset cache if server was started with no data, and data has subsequently been added + if cacheN == 0 && holderN > 0 { + err := api.ResetUsageCache() + if err != nil { + api.server.logger.Infof("data detected but could not recalculate cache: %s", err) + } + } var t time.Time if lastUpdated == t { @@ -968,6 +978,15 @@ func (api *API) Usage(ctx context.Context, remote bool) (map[string]NodeUsage, e api.requestUsageOfNodes() } + // triggers recalculation of cache in background, ahead of schedule, if number of indexes in + // cache differs from number of indexes in holder + if cacheN > 0 && cacheN != holderN { + err := api.ResetUsageCache() + if err != nil { + api.server.logger.Infof("resetting cache in background: %s", err) + } + } + return api.usageCache.data, nil } From a7c24c6745e1c207ed7282947982443bcf256522 Mon Sep 17 00:00:00 2001 From: Samir Patel <48686912+54mir@users.noreply.github.com> Date: Thu, 10 Jun 2021 11:28:23 -0500 Subject: [PATCH 02/12] lock read of lastUpdated --- api.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/api.go b/api.go index 59cde8d1f..7709ba8d2 100644 --- a/api.go +++ b/api.go @@ -1016,7 +1016,10 @@ func (api *API) calculateUsage() { api.server.wg.Add(1) defer api.server.wg.Done() + api.usageCache.muAssign.Lock() 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 { From 9b11ded9e3370dba7123c62ec43579613686ec6e Mon Sep 17 00:00:00 2001 From: Samir Patel <48686912+54mir@users.noreply.github.com> Date: Thu, 10 Jun 2021 16:30:22 -0500 Subject: [PATCH 03/12] scale refresh rate by last calculation time --- api.go | 55 +++++++++++++++++++++++++++++-------------------------- 1 file changed, 29 insertions(+), 26 deletions(-) diff --git a/api.go b/api.go index 7709ba8d2..8211d55ee 100644 --- a/api.go +++ b/api.go @@ -903,10 +903,12 @@ func (api *API) PrimaryNode() *topology.Node { // Cache of disk usage statistics type usageCache struct { - data map[string]NodeUsage - refreshInterval time.Duration - lastUpdated time.Time - resetTrigger chan bool + data map[string]NodeUsage + refreshInterval time.Duration + lastUpdated time.Time + resetTrigger chan bool + lastCalcDuration time.Duration + waitMultiplier time.Duration muCalculate sync.Mutex muAssign sync.Mutex @@ -955,22 +957,17 @@ func (api *API) Usage(ctx context.Context, remote bool) (map[string]NodeUsage, e span, _ := tracing.StartSpanFromContext(ctx, "API.Usage") defer span.Finish() - api.usageCache.muAssign.Lock() - lastUpdated := api.usageCache.lastUpdated - cacheN := len(api.usageCache.data[api.server.nodeID].Disk.IndexUsage) - api.usageCache.muAssign.Unlock() - holderN := len(api.holder.Indexes()) - - // reset cache if server was started with no data, and data has subsequently been added - if cacheN == 0 && holderN > 0 { + if api.usageCache.lastCalcDuration < (time.Second * 5) { err := api.ResetUsageCache() if err != nil { api.server.logger.Infof("data detected but could not recalculate cache: %s", err) } } - var t time.Time - if lastUpdated == t { + api.usageCache.muAssign.Lock() + lastUpdated := api.usageCache.lastUpdated + api.usageCache.muAssign.Unlock() + if lastUpdated == (time.Time{}) { api.calculateUsage() } @@ -978,15 +975,6 @@ func (api *API) Usage(ctx context.Context, remote bool) (map[string]NodeUsage, e api.requestUsageOfNodes() } - // triggers recalculation of cache in background, ahead of schedule, if number of indexes in - // cache differs from number of indexes in holder - if cacheN > 0 && cacheN != holderN { - err := api.ResetUsageCache() - if err != nil { - api.server.logger.Infof("resetting cache in background: %s", err) - } - } - return api.usageCache.data, nil } @@ -1077,12 +1065,16 @@ func (api *API) RefreshUsageCache(refresh time.Duration) { trigger := make(chan bool) defer close(trigger) api.usageCache = &usageCache{ - data: make(map[string]NodeUsage), - refreshInterval: refresh, - resetTrigger: trigger, + data: make(map[string]NodeUsage), + refreshInterval: refresh, + resetTrigger: trigger, + lastCalcDuration: 0, + waitMultiplier: time.Duration(5), } for { + start := time.Now() api.calculateUsage() + api.setRefreshInterval(time.Since(start)) select { case <-trigger: continue @@ -1094,6 +1086,17 @@ func (api *API) RefreshUsageCache(refresh time.Duration) { } } +func (api *API) setRefreshInterval(dur time.Duration) { + refresh := dur * api.usageCache.waitMultiplier + if refresh < time.Hour { + refresh = time.Hour + } + api.usageCache.muAssign.Lock() + api.usageCache.refreshInterval = refresh + api.usageCache.lastCalcDuration = dur + api.usageCache.muAssign.Unlock() +} + // Resets the lastUpdated time and awakens RefreshUsageCache() func (api *API) ResetUsageCache() error { if api.usageCache != nil { From 268e710e6922c32a86fddedca0ad17888edce806 Mon Sep 17 00:00:00 2001 From: Samir Patel <48686912+54mir@users.noreply.github.com> Date: Thu, 10 Jun 2021 17:36:03 -0500 Subject: [PATCH 04/12] update comment --- api.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/api.go b/api.go index 8211d55ee..08523b4f9 100644 --- a/api.go +++ b/api.go @@ -952,7 +952,7 @@ type MemoryUsage struct { TotalUse uint64 `json:"totalInUse"` } -// Returns disk usage from cache. Waits for calculation if cache is empty. +// Returns disk usage from cache if cache is large. It will recalculate on the spot of the last cacluation was under 5 seconds. func (api *API) Usage(ctx context.Context, remote bool) (map[string]NodeUsage, error) { span, _ := tracing.StartSpanFromContext(ctx, "API.Usage") defer span.Finish() From b46b5fc134da61a875120938028dc58e15984e5e Mon Sep 17 00:00:00 2001 From: Samir Patel <48686912+54mir@users.noreply.github.com> Date: Fri, 11 Jun 2021 11:16:32 -0500 Subject: [PATCH 05/12] remove usage-interval flag --- api.go | 4 ++-- ctl/server.go | 3 --- server/config.go | 7 ------- server/server.go | 2 +- 4 files changed, 3 insertions(+), 13 deletions(-) diff --git a/api.go b/api.go index 08523b4f9..79c6573ee 100644 --- a/api.go +++ b/api.go @@ -1061,12 +1061,12 @@ func (api *API) calculateUsage() { } // Periodically calculates disk usage -func (api *API) RefreshUsageCache(refresh time.Duration) { +func (api *API) RefreshUsageCache() { trigger := make(chan bool) defer close(trigger) api.usageCache = &usageCache{ data: make(map[string]NodeUsage), - refreshInterval: refresh, + refreshInterval: time.Hour, resetTrigger: trigger, lastCalcDuration: 0, waitMultiplier: time.Duration(5), diff --git a/ctl/server.go b/ctl/server.go index 250902fd1..6c885d213 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -110,7 +110,4 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.DurationVar((*time.Duration)(&srv.Config.Postgres.WriteTimeout), "postgres.write-timeout", time.Duration(srv.Config.Postgres.WriteTimeout), "Timeout for writes on a postgres connection. (set 0 to disable)") flags.Uint32Var(&srv.Config.Postgres.MaxStartupSize, "postgres.max-startup-size", srv.Config.Postgres.MaxStartupSize, "Maximum acceptable size of a postgres startup packet, in bytes. (set 0 to disable)") flags.Uint16Var(&srv.Config.Postgres.ConnectionLimit, "postgres.connection-limit", srv.Config.Postgres.ConnectionLimit, "Maximum number of simultaneous postgres connections to allow. (set 0 to disable)") - - // Disk/Memory Usage refresh rate in minutes for ui/usage http endpoint - flags.DurationVar((*time.Duration)(&srv.Config.Usage.Interval), "usage-interval", time.Duration(srv.Config.Usage.Interval), "Number in minutes between recalculations of disk/memory usage cache") } diff --git a/server/config.go b/server/config.go index 7871364db..f59040271 100644 --- a/server/config.go +++ b/server/config.go @@ -224,11 +224,6 @@ type Config struct { // LookupDBDSN is an external database to connect to for `ExternalLookup` queries. LookupDBDSN string `toml:"lookup-db-dsn"` - - // Disk Usage refresh interval for ui/usage http endpoint - Usage struct { - Interval toml.Duration `toml:"usage-interval"` - } } // MustValidate checks that all ports in a Config are unique and not zero. @@ -364,8 +359,6 @@ func NewConfig() *Config { c.Etcd.PeerCertFile = "" c.Etcd.PeerKeyFile = "" - c.Usage.Interval = toml.Duration(6 * 60 * time.Minute) // 6 hours - return c } diff --git a/server/server.go b/server/server.go index a8d5ecfc4..febdb13ee 100644 --- a/server/server.go +++ b/server/server.go @@ -277,7 +277,7 @@ func (m *Command) Start() (err error) { } } - go m.API.RefreshUsageCache(time.Duration(m.Config.Usage.Interval)) + go m.API.RefreshUsageCache() _ = testhook.Opened(pilosa.NewAuditor(), m, nil) close(m.Started) From 011291f19bdce0800a2a6391e4eb7076cd285ed7 Mon Sep 17 00:00:00 2001 From: Samir Patel <48686912+54mir@users.noreply.github.com> Date: Fri, 11 Jun 2021 11:28:39 -0500 Subject: [PATCH 06/12] update comments --- api.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/api.go b/api.go index 79c6573ee..0ff042165 100644 --- a/api.go +++ b/api.go @@ -952,7 +952,7 @@ type MemoryUsage struct { TotalUse uint64 `json:"totalInUse"` } -// Returns disk usage from cache if cache is large. It will recalculate on the spot of the last cacluation was under 5 seconds. +// Returns disk usage from cache if cache is large. It will recalculate on the spot if the last cacluation was under 5 seconds. func (api *API) Usage(ctx context.Context, remote bool) (map[string]NodeUsage, error) { span, _ := tracing.StartSpanFromContext(ctx, "API.Usage") defer span.Finish() @@ -960,7 +960,7 @@ func (api *API) Usage(ctx context.Context, remote bool) (map[string]NodeUsage, e if api.usageCache.lastCalcDuration < (time.Second * 5) { err := api.ResetUsageCache() if err != nil { - api.server.logger.Infof("data detected but could not recalculate cache: %s", err) + api.server.logger.Infof("could not reset cache: %s", err) } } @@ -997,7 +997,7 @@ func (api *API) requestUsageOfNodes() { } } -// Calculates disk usage from scratch for each index and stores the results in the usage cache +// Calculates disk usage from scratch if cache has expired for each index and stores the results in the usage cache func (api *API) calculateUsage() { api.usageCache.muCalculate.Lock() defer api.usageCache.muCalculate.Unlock() @@ -1086,6 +1086,7 @@ func (api *API) RefreshUsageCache() { } } +// Refresh interval set in relation to how long the last calculation took. func (api *API) setRefreshInterval(dur time.Duration) { refresh := dur * api.usageCache.waitMultiplier if refresh < time.Hour { From 8bf472bb75aa7cbb747aa4876517f97db948babe Mon Sep 17 00:00:00 2001 From: Samir Patel <48686912+54mir@users.noreply.github.com> Date: Mon, 14 Jun 2021 10:48:52 -0500 Subject: [PATCH 07/12] add duty cycle config flag --- api.go | 10 ++++++++-- ctl/server.go | 3 +++ server/config.go | 6 ++++++ server/server.go | 2 +- 4 files changed, 18 insertions(+), 3 deletions(-) diff --git a/api.go b/api.go index 0ff042165..c9766e27c 100644 --- a/api.go +++ b/api.go @@ -1061,15 +1061,21 @@ func (api *API) calculateUsage() { } // Periodically calculates disk usage -func (api *API) RefreshUsageCache() { +func (api *API) RefreshUsageCache(dutyCycle float64) { trigger := make(chan bool) defer close(trigger) + + if dutyCycle <= 0 { + dutyCycle = 20 + } + multiplier := int(math.Ceil(100/dutyCycle)) - 1 + api.usageCache = &usageCache{ data: make(map[string]NodeUsage), refreshInterval: time.Hour, resetTrigger: trigger, lastCalcDuration: 0, - waitMultiplier: time.Duration(5), + waitMultiplier: time.Duration(multiplier), } for { start := time.Now() diff --git a/ctl/server.go b/ctl/server.go index 6c885d213..98472ea95 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -110,4 +110,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { flags.DurationVar((*time.Duration)(&srv.Config.Postgres.WriteTimeout), "postgres.write-timeout", time.Duration(srv.Config.Postgres.WriteTimeout), "Timeout for writes on a postgres connection. (set 0 to disable)") flags.Uint32Var(&srv.Config.Postgres.MaxStartupSize, "postgres.max-startup-size", srv.Config.Postgres.MaxStartupSize, "Maximum acceptable size of a postgres startup packet, in bytes. (set 0 to disable)") flags.Uint16Var(&srv.Config.Postgres.ConnectionLimit, "postgres.connection-limit", srv.Config.Postgres.ConnectionLimit, "Maximum number of simultaneous postgres connections to allow. (set 0 to disable)") + + // Disk and Memory usage cache for ui/usage endpoint + flags.Float64Var(&srv.Config.UsageDutyCycle, "usage-duty-cycle", srv.Config.UsageDutyCycle, "Sets the percentage of time that is spent recalculating the disk and memory usage cache. 100.0 for always-running, must be > 0.") } diff --git a/server/config.go b/server/config.go index f59040271..e744b86e2 100644 --- a/server/config.go +++ b/server/config.go @@ -224,6 +224,9 @@ type Config struct { // LookupDBDSN is an external database to connect to for `ExternalLookup` queries. LookupDBDSN string `toml:"lookup-db-dsn"` + + // The percentage of time spent recalculating the disk and memory usage cache. + UsageDutyCycle float64 `toml:"usage-duty-cycle"` } // MustValidate checks that all ports in a Config are unique and not zero. @@ -359,6 +362,9 @@ func NewConfig() *Config { c.Etcd.PeerCertFile = "" c.Etcd.PeerKeyFile = "" + // Disk and Memory Usage + c.UsageDutyCycle = 20.0 + return c } diff --git a/server/server.go b/server/server.go index febdb13ee..11ff5bf1b 100644 --- a/server/server.go +++ b/server/server.go @@ -277,7 +277,7 @@ func (m *Command) Start() (err error) { } } - go m.API.RefreshUsageCache() + go m.API.RefreshUsageCache(m.Config.UsageDutyCycle) _ = testhook.Opened(pilosa.NewAuditor(), m, nil) close(m.Started) From 11207fc589e4e65da68a25802099189359040470 Mon Sep 17 00:00:00 2001 From: Samir Patel <48686912+54mir@users.noreply.github.com> Date: Mon, 14 Jun 2021 11:12:12 -0500 Subject: [PATCH 08/12] Update api.go Co-authored-by: Alan Bernstein --- api.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/api.go b/api.go index c9766e27c..84dd64943 100644 --- a/api.go +++ b/api.go @@ -960,7 +960,7 @@ func (api *API) Usage(ctx context.Context, remote bool) (map[string]NodeUsage, e if api.usageCache.lastCalcDuration < (time.Second * 5) { err := api.ResetUsageCache() if err != nil { - api.server.logger.Infof("could not reset cache: %s", err) + api.server.logger.Infof("could not reset usageCache: %s", err) } } From 6bdf685471889a431c50724a69a6c51abaa59044 Mon Sep 17 00:00:00 2001 From: Samir Patel <48686912+54mir@users.noreply.github.com> Date: Mon, 14 Jun 2021 11:43:49 -0500 Subject: [PATCH 09/12] change waitMultiplier to float64 --- api.go | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/api.go b/api.go index c9766e27c..9884416c7 100644 --- a/api.go +++ b/api.go @@ -908,7 +908,7 @@ type usageCache struct { lastUpdated time.Time resetTrigger chan bool lastCalcDuration time.Duration - waitMultiplier time.Duration + waitMultiplier float64 muCalculate sync.Mutex muAssign sync.Mutex @@ -1068,14 +1068,14 @@ func (api *API) RefreshUsageCache(dutyCycle float64) { if dutyCycle <= 0 { dutyCycle = 20 } - multiplier := int(math.Ceil(100/dutyCycle)) - 1 + multiplier := 100 / dutyCycle api.usageCache = &usageCache{ data: make(map[string]NodeUsage), refreshInterval: time.Hour, resetTrigger: trigger, lastCalcDuration: 0, - waitMultiplier: time.Duration(multiplier), + waitMultiplier: multiplier, } for { start := time.Now() @@ -1094,7 +1094,7 @@ 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 := dur * api.usageCache.waitMultiplier + refresh := time.Duration(float64(dur) * api.usageCache.waitMultiplier) if refresh < time.Hour { refresh = time.Hour } From d83a502657d8ba74ef5dc9c3632179e6a11e8e39 Mon Sep 17 00:00:00 2001 From: Samir Patel <48686912+54mir@users.noreply.github.com> Date: Mon, 14 Jun 2021 11:55:23 -0500 Subject: [PATCH 10/12] change comment --- api.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/api.go b/api.go index e8fa2eea8..efba38a18 100644 --- a/api.go +++ b/api.go @@ -1060,7 +1060,7 @@ func (api *API) calculateUsage() { } } -// Periodically calculates disk usage +// Periodically calculates disk/memory usage func (api *API) RefreshUsageCache(dutyCycle float64) { trigger := make(chan bool) defer close(trigger) From fb11d7f65666af7cf13c632df8c9a6ee74e20ebf Mon Sep 17 00:00:00 2001 From: Samir Patel <48686912+54mir@users.noreply.github.com> Date: Tue, 15 Jun 2021 09:17:53 -0500 Subject: [PATCH 11/12] Update api.go Co-authored-by: Alan Bernstein --- api.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/api.go b/api.go index efba38a18..f86140366 100644 --- a/api.go +++ b/api.go @@ -1068,7 +1068,7 @@ func (api *API) RefreshUsageCache(dutyCycle float64) { if dutyCycle <= 0 { dutyCycle = 20 } - multiplier := 100 / dutyCycle + multiplier := 100 / dutyCycle - 1 api.usageCache = &usageCache{ data: make(map[string]NodeUsage), From 6b45325ca7a81f5e74f26d610f911c5528db86d4 Mon Sep 17 00:00:00 2001 From: Samir Patel <48686912+54mir@users.noreply.github.com> Date: Tue, 15 Jun 2021 09:37:42 -0500 Subject: [PATCH 12/12] expand comment on duty cycle --- api.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/api.go b/api.go index f86140366..ecb9783bd 100644 --- a/api.go +++ b/api.go @@ -1060,7 +1060,9 @@ func (api *API) calculateUsage() { } } -// Periodically calculates disk/memory usage +// Periodically calculates disk/memory usage in terms of the duty cycle. The duty cycle represents the percentage of +// time that is spent recalculating this cache. It is specified relatively, rather than by a set interval, because +// scans can take an unpredictably long time. func (api *API) RefreshUsageCache(dutyCycle float64) { trigger := make(chan bool) defer close(trigger) @@ -1068,7 +1070,7 @@ func (api *API) RefreshUsageCache(dutyCycle float64) { if dutyCycle <= 0 { dutyCycle = 20 } - multiplier := 100 / dutyCycle - 1 + multiplier := 100/dutyCycle - 1 api.usageCache = &usageCache{ data: make(map[string]NodeUsage),