mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-07 00:55:55 +00:00
Merge pull request #1622 from 54mir/lattice-fix
CORE-557 UI/Usage Cache with Lattice Fix
This commit is contained in:
Samir Patel
GitHub
commit
b886d1e571
9 changed files with 194 additions and 71 deletions
202
api.go
202
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.
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
}
|
||||
|
|
|
|||
4
go.mod
4
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
|
||||
|
|
|
|||
10
go.sum
10
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=
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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) {
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue