Merge pull request #1633 from 54mir/reset-cache-on-schema

CORE-642 Reset Usage Cache If Outdated On Request

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 {

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.")
 }

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
 }

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)