diff --git a/api.go b/api.go index 341e766ca..7bc31142e 100644 --- a/api.go +++ b/api.go @@ -59,6 +59,8 @@ type API struct { importWorkerPoolSize int importWork chan importJob + usageCache *usageCache + Serializer Serializer } @@ -899,10 +901,22 @@ func (api *API) PrimaryNode() *topology.Node { return snap.PrimaryFieldTranslationNode() } +// Cache of disk usage statistics +type usageCache struct { + data map[string]NodeUsage + refreshInterval time.Duration + lastUpdated time.Time + resetTrigger chan bool + + muCalculate sync.Mutex + muAssign sync.Mutex +} + // NodeUsage represents all usage measurements for one node. type NodeUsage struct { - Disk DiskUsage `json:"diskUsage"` - Memory MemoryUsage `json:"memoryUsage"` + Disk DiskUsage `json:"diskUsage"` + Memory MemoryUsage `json:"memoryUsage"` + LastUpdated time.Time `json:"lastUpdated"` } // DiskUsage represents the storage space used on disk by one node. @@ -936,67 +950,149 @@ type MemoryUsage struct { TotalUse uint64 `json:"totalInUse"` } -// Usage gets the resource usage per index, in a map[nodeID]NodeUsage +// Returns disk usage from cache. Waits for calculation if cache is empty. func (api *API) Usage(ctx context.Context, remote bool) (map[string]NodeUsage, error) { span, _ := tracing.StartSpanFromContext(ctx, "API.Usage") defer span.Finish() - nodeUsages := make(map[string]NodeUsage) + api.usageCache.muAssign.Lock() + lastUpdated := api.usageCache.lastUpdated + api.usageCache.muAssign.Unlock() - indexDetails, nodeMetadataBytes, err := api.holder.Txf().IndexUsageDetails() - if err != nil { - return nil, errors.Wrap(err, "getting node usage") - } - totalSize := nodeMetadataBytes - for _, s := range indexDetails { - totalSize += s.Total + var t time.Time + if lastUpdated == t { + api.calculateUsage() } - // 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) - } - - // Insert into result. - nodeUsage := NodeUsage{ - Disk: DiskUsage{ - Capacity: diskCapacity, - TotalUse: totalSize, - IndexUsage: indexDetails, - }, - Memory: MemoryUsage{ - Capacity: memoryCapacity, - TotalUse: memoryUse, - }, - } - nodeUsages[api.server.nodeID] = nodeUsage - - // Collect usage from remote nodes if !remote { - nodes := api.cluster.Nodes() - for _, node := range nodes { - if node.ID == api.server.nodeID { - continue - } - nodeUsage, err := api.server.defaultClient.GetNodeUsage(ctx, &node.URI) - if err != nil { - return nil, errors.Wrapf(err, "collecting disk usage from %s", node.URI) - } - nodeUsages[node.ID] = nodeUsage[node.ID] + api.requestUsageOfNodes() + } + + return api.usageCache.data, nil +} + +// Makes a ui/usage request for each node in cluster to calculates its usage and adds it to the cache +func (api *API) requestUsageOfNodes() { + nodes := api.cluster.Nodes() + for _, node := range nodes { + if node.ID == api.server.nodeID { + continue + } + + nodeUsage, err := api.server.defaultClient.GetNodeUsage(context.Background(), &node.URI) + if err != nil { + api.server.logger.Infof("couldn't collect disk usage from %s: %s", node.URI, err) + } + + api.usageCache.muAssign.Lock() + api.usageCache.data[node.ID] = nodeUsage[node.ID] + api.usageCache.muAssign.Unlock() + } +} + +// Calculates disk usage from scratch 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() + + lastUpdated := api.usageCache.lastUpdated + 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() + } +} + +// Periodically calculates disk usage +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, + } + for { + api.calculateUsage() + select { + case <-trigger: + continue + case <-api.server.closing: + return + case <-time.After(api.usageCache.refreshInterval): + continue } } - return nodeUsages, nil +} + +// Resets the lastUpdated time and awakens RefreshUsageCache() +func (api *API) ResetUsageCache() error { + if api.usageCache != nil { + api.usageCache.muAssign.Lock() + api.usageCache.lastUpdated = time.Time{} + api.usageCache.muAssign.Unlock() + } else { + return errors.New("invalidating cache: cache not initialized") + } + api.usageCache.resetTrigger <- true + return nil +} + +// isClosing returns true if the server is shutting down. +func (api *API) isClosing() bool { + select { + case <-api.server.closing: + return true + default: + return false + } } // RecalculateCaches forces all TopN caches to be updated. diff --git a/ctl/server.go b/ctl/server.go index 6c885d213..250902fd1 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/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/go.mod b/go.mod index 3fb7f5f96..7af634a50 100644 --- a/go.mod +++ b/go.mod @@ -47,9 +47,9 @@ require ( go.etcd.io/etcd v0.0.0-20201125193152-8a03d2e9614b golang.org/x/exp v0.0.0-20201008143054-e3b2a7f2fdc7 golang.org/x/mod v0.4.2 - golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4 // indirect + golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4 // indirect golang.org/x/sync v0.0.0-20210220032951-036812b2e83c - golang.org/x/sys v0.0.0-20210320140829-1e4c9ba3b0c4 // indirect + golang.org/x/sys v0.0.0-20210510120138-977fb7262007 // indirect golang.org/x/text v0.3.5 // indirect google.golang.org/grpc v1.28.0 gopkg.in/yaml.v2 v2.3.0 // indirect diff --git a/go.sum b/go.sum index 56cd62a14..296acadcb 100644 --- a/go.sum +++ b/go.sum @@ -408,8 +408,8 @@ golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLL golang.org/x/net v0.0.0-20190813141303-74dc4d7220e7/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= -golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4 h1:b0LrWgu8+q7z4J+0Y3Umo5q1dL7NXBkKBWkaVkAq17E= -golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4/go.mod h1:RBQZq4jEuRlivfhVLdyRGr576XBO4/greRjx4P4O3yc= +golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4 h1:4nGaVu0QrbjT/AK2PRLuQfQuh6DJve+pELhqTdAj3x0= +golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= @@ -444,9 +444,9 @@ golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7w golang.org/x/sys v0.0.0-20201014080544-cc95f250f6bc/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201024232916-9f70ab9862d5/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20210315160823-c6e025ad8005/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20210320140829-1e4c9ba3b0c4 h1:EZ2mChiOa8udjfp6rRmswTbtZN/QzUQp4ptM4rnjHvc= -golang.org/x/sys v0.0.0-20210320140829-1e4c9ba3b0c4/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210510120138-977fb7262007 h1:gG67DSER+11cZvqIMb8S8bt0vZtiN6xWYARwirrOSfE= +golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= diff --git a/server.go b/server.go index bb7f85c94..f824253bf 100644 --- a/server.go +++ b/server.go @@ -601,7 +601,6 @@ func (s *Server) Open() error { // bring up the background tasks for the holder. s.holder.SnapshotQueue = s.snapshotQueue s.holder.Activate() - // if we joined existing cluster then broadcast "resize on add" message if initState == disco.InitialClusterStateExisting { if err := s.cluster.addNode(s.nodeID); err != nil { diff --git a/server/config.go b/server/config.go index f59040271..7871364db 100644 --- a/server/config.go +++ b/server/config.go @@ -224,6 +224,11 @@ 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. @@ -359,6 +364,8 @@ 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/handler_test.go b/server/handler_test.go index f49b61624..639611ac8 100644 --- a/server/handler_test.go +++ b/server/handler_test.go @@ -502,11 +502,17 @@ func TestHandler_Endpoints(t *testing.T) { } }) + // UI/usage returns disk and memory usage from a precalculated cache. + // Since the cache calculates the cache on server startup, and tests create indexes thereafter + // the cache initially has 0 indexes when the test suite is ran. Therefore, this test first + // resets the cache. t.Run("UI/usage", func(t *testing.T) { + if cmd.API.ResetUsageCache() != nil { + t.Fatal(err) + } w := httptest.NewRecorder() h.ServeHTTP(w, test.MustNewHTTPRequest("GET", "/ui/usage", nil)) if w.Code != gohttp.StatusOK { - fmt.Printf("%+v\n", w.Body) t.Fatalf("unexpected status code: %d", w.Code) } nodeUsages := make(map[string]pilosa.NodeUsage) @@ -515,22 +521,27 @@ func TestHandler_Endpoints(t *testing.T) { } for _, nodeUsage := range nodeUsages { - numIndexes := len(nodeUsage.Disk.IndexUsage) - if nodeUsage.Disk.TotalUse < 75000 || nodeUsage.Disk.TotalUse > 700000 { - // Usage measurements are not consistent between machines, or - // over time, as features and implementations change, so checking - // for a range of sizes may be most useful way to test the details of this. - t.Fatalf("expected 75k < total < 500k, got %d", nodeUsage.Disk.TotalUse) + if nodeUsage.Disk.TotalUse < 1 { + t.Fatalf("expected some disk use, got %d", nodeUsage.Disk.TotalUse) } + if nodeUsage.Disk.Capacity < 1 { + t.Fatalf("expected some disk capacity, got %d", nodeUsage.Disk.Capacity) + } + if nodeUsage.Memory.TotalUse < 1 { + t.Fatalf("expected some memory use, got %d", nodeUsage.Memory.TotalUse) + } + if nodeUsage.Memory.Capacity < 1 { + t.Fatalf("expected some memory capacity, got %d", nodeUsage.Memory.Capacity) + } + numIndexes := len(nodeUsage.Disk.IndexUsage) if numIndexes != 3 { - t.Fatalf("wrong length index usage list: expected %d, got %d", 2, numIndexes) + t.Fatalf("wrong length index usage list: expected %d, got %d", 3, numIndexes) } numFields := len(nodeUsage.Disk.IndexUsage["i1"].Fields) if numFields != len(i1.Fields()) { t.Fatalf("wrong length field usage list: expected %d, got %d", len(i1.Fields()), numFields) } } - }) t.Run("UI/shard-distribution", func(t *testing.T) { diff --git a/server/server.go b/server/server.go index ea8de2110..a8d5ecfc4 100644 --- a/server/server.go +++ b/server/server.go @@ -277,6 +277,8 @@ func (m *Command) Start() (err error) { } } + go m.API.RefreshUsageCache(time.Duration(m.Config.Usage.Interval)) + _ = testhook.Opened(pilosa.NewAuditor(), m, nil) close(m.Started) return nil diff --git a/txfactory.go b/txfactory.go index 8a2b987cb..482b90f85 100644 --- a/txfactory.go +++ b/txfactory.go @@ -572,7 +572,7 @@ func (f *TxFactory) DumpAll() { // IndexUsageDetails computes the sum of filesizes used by the node, broken down // by index, field, fragments and keys. -func (f *TxFactory) IndexUsageDetails() (map[string]IndexUsage, uint64, error) { +func (f *TxFactory) IndexUsageDetails(isClosing func() bool) (map[string]IndexUsage, uint64, error) { indexUsage := make(map[string]IndexUsage) holderPath, err := expandDirName(f.holder.path) if err != nil { @@ -612,6 +612,9 @@ func (f *TxFactory) IndexUsageDetails() (map[string]IndexUsage, uint64, error) { fragmentUsage := uint64(0) for _, shard := range fld.AvailableShards(true).Slice() { + if isClosing() { + return nil, 0, nil + } if err := func() error { tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) if err != nil { @@ -718,6 +721,8 @@ func (f *TxFactory) fieldUsage(indexPath string, fld *Field) (FieldUsage, error) return fieldUsage, nil } +// NOTE: Go 1.16 introduced a new Readdir() method that is supposed to be more performant. +// Not yet upgraded b/c new method is not compatible with older versions of Go. func directoryUsage(fname string, recursive bool) (uint64, error) { if !dirExists(fname) { return 0, errors.Errorf("directory does not exist (%s)", fname)