diff --git a/api.go b/api.go index 7bc31142e..ecb9783bd 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 float64 muCalculate sync.Mutex muAssign sync.Mutex @@ -950,17 +952,22 @@ 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 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() + if api.usageCache.lastCalcDuration < (time.Second * 5) { + err := api.ResetUsageCache() + if err != nil { + api.server.logger.Infof("could not reset usageCache: %s", err) + } + } + api.usageCache.muAssign.Lock() lastUpdated := api.usageCache.lastUpdated api.usageCache.muAssign.Unlock() - - var t time.Time - if lastUpdated == t { + if lastUpdated == (time.Time{}) { api.calculateUsage() } @@ -990,14 +997,17 @@ 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() 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 { @@ -1050,17 +1060,29 @@ func (api *API) calculateUsage() { } } -// Periodically calculates disk usage -func (api *API) RefreshUsageCache(refresh time.Duration) { +// 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) + + if dutyCycle <= 0 { + dutyCycle = 20 + } + multiplier := 100/dutyCycle - 1 + api.usageCache = &usageCache{ - data: make(map[string]NodeUsage), - refreshInterval: refresh, - resetTrigger: trigger, + data: make(map[string]NodeUsage), + refreshInterval: time.Hour, + resetTrigger: trigger, + lastCalcDuration: 0, + waitMultiplier: multiplier, } for { + start := time.Now() api.calculateUsage() + api.setRefreshInterval(time.Since(start)) select { case <-trigger: continue @@ -1072,6 +1094,18 @@ func (api *API) RefreshUsageCache(refresh time.Duration) { } } +// 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 + } + 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 { diff --git a/ctl/server.go b/ctl/server.go index 250902fd1..98472ea95 100644 --- a/ctl/server.go +++ b/ctl/server.go @@ -111,6 +111,6 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) { 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") + // 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 7871364db..e744b86e2 100644 --- a/server/config.go +++ b/server/config.go @@ -225,10 +225,8 @@ 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"` - } + // 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. @@ -364,7 +362,8 @@ func NewConfig() *Config { c.Etcd.PeerCertFile = "" c.Etcd.PeerKeyFile = "" - c.Usage.Interval = toml.Duration(6 * 60 * time.Minute) // 6 hours + // Disk and Memory Usage + c.UsageDutyCycle = 20.0 return c } diff --git a/server/server.go b/server/server.go index a8d5ecfc4..11ff5bf1b 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(m.Config.UsageDutyCycle) _ = testhook.Opened(pilosa.NewAuditor(), m, nil) close(m.Started)