diff --git a/api.go b/api.go index 2bd8eb2e4..4b27e0766 100644 --- a/api.go +++ b/api.go @@ -816,30 +816,48 @@ func (api *API) Node() *Node { // NodeUsage represents all usage measurements for one node. type NodeUsage struct { - Disk DiskUsage `json:"bytesOnDisk"` + Disk DiskUsage `json:"diskUsage"` } // DiskUsage represents the storage space used on disk by one node. type DiskUsage struct { - Capacity uint64 `json:"capacity,omitempty"` - TotalUse int64 `json:"totalInUse"` - Indexes map[string]int64 `json:"indexes"` + Capacity uint64 `json:"capacity,omitempty"` + TotalUse uint64 `json:"totalInUse"` + IndexUsage map[string]IndexUsage `json:"indexes"` } -// Usage gets the disk usage per index, in a map[nodeID]NodeUsage +// IndexUsage represents the storage space used on disk by one index, on one node. +type IndexUsage struct { + Total uint64 `json:"total"` + IndexKeys uint64 `json:"indexKeys"` + FieldKeysTotal uint64 `json:"fieldKeysTotal"` + Fragments uint64 `json:"fragments"` + Metadata uint64 `json:"metadata"` + Fields map[string]FieldUsage `json:"fields"` +} + +// FieldUsage represents the storage space used on disk by one field, on one node +type FieldUsage struct { + Total uint64 `json:"total"` + Fragments uint64 `json:"fragments"` + Keys uint64 `json:"keys"` + Metadata uint64 `json:"metadata"` +} + +// Usage gets the disk usage, in a map[nodeID]NodeUsage. 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) - indexSizes, err := api.holder.Txf().IndexSizes() + indexDetails, nodeMetadataBytes, err := api.holder.Txf().IndexUsageDetails() if err != nil { - return nil, errors.Wrap(err, "getting index usage") + return nil, errors.Wrap(err, "getting node usage") } - var totalSize int64 - for _, s := range indexSizes { - totalSize += s + totalSize := nodeMetadataBytes + for _, s := range indexDetails { + totalSize += s.Total } capacity, err := api.server.systemInfo.DiskCapacity(api.holder.path) @@ -850,14 +868,14 @@ func (api *API) Usage(ctx context.Context, remote bool) (map[string]NodeUsage, e // Insert into result. nodeUsage := NodeUsage{ Disk: DiskUsage{ - Capacity: capacity, - TotalUse: totalSize, - Indexes: indexSizes, + Capacity: capacity, + TotalUse: totalSize, + IndexUsage: indexDetails, }, } nodeUsages[api.server.nodeID] = nodeUsage - // Collect size on disk from remote nodes + // Collect diskUsage from remote nodes if !remote { nodes := api.cluster.Nodes() for _, node := range nodes { diff --git a/bluegreentx.go b/bluegreentx.go index 62f1c0bd8..e0d9ddee9 100644 --- a/bluegreentx.go +++ b/bluegreentx.go @@ -663,6 +663,10 @@ func (c *blueGreenTx) ContainerIterator(index, field, view string, shard uint64, return bgi, bfound, errB } +func (tx *blueGreenTx) GetFieldSizeBytes(index, field string) (uint64, error) { + return 0, nil +} + func NewBlueGreenIterator(tx *blueGreenTx, ait, bit roaring.ContainerIterator) *blueGreenIterator { return &blueGreenIterator{ tx: tx, diff --git a/bolt.go b/bolt.go index b8a3f25be..3b37a2bec 100644 --- a/bolt.go +++ b/bolt.go @@ -759,6 +759,10 @@ func (tx *BoltTx) ContainerIterator(index, field, view string, shard uint64, fir return bi, bytes.Equal(bi.lastKey, needle), nil } +func (tx *BoltTx) GetFieldSizeBytes(index, field string) (uint64, error) { + return 0, nil +} + // BoltIterator is the iterator returned from a BoltTx.ContainerIterator() call. // It implements the roaring.ContainerIterator interface. type BoltIterator struct { diff --git a/catcher.go b/catcher.go index 671ee7662..09f24ce93 100644 --- a/catcher.go +++ b/catcher.go @@ -317,3 +317,7 @@ func (c *catcherTx) ApplyFilter(index, field, view string, shard uint64, ckey ui func (c *catcherTx) GetSortedFieldViewList(idx *Index, shard uint64) (fvs []txkey.FieldView, err error) { return c.b.GetSortedFieldViewList(idx, shard) } + +func (tx *catcherTx) GetFieldSizeBytes(index, field string) (uint64, error) { + return 0, nil +} diff --git a/holder.go b/holder.go index 256871058..3e40ea14b 100644 --- a/holder.go +++ b/holder.go @@ -1277,10 +1277,8 @@ func (h *Holder) LoadNodeID() (string, error) { // Log startup time and version to $DATA_DIR/.startup.log func (h *Holder) logStartup() error { - time, err := time.Now().MarshalText() - if err != nil { - return errors.Wrap(err, "creating timestamp") - } + RFC3339NanoFixedWidth := "2006-01-02T15:04:05.000000 07:00" + time := time.Now().Format(RFC3339NanoFixedWidth) logLine := fmt.Sprintf("%s\t%s\n", time, Version) f, err := os.OpenFile(h.path+"/.startup.log", os.O_APPEND|os.O_WRONLY|os.O_CREATE, 0600) diff --git a/lattice b/lattice index 28c2313ec..2f0302c1d 160000 --- a/lattice +++ b/lattice @@ -1 +1 @@ -Subproject commit 28c2313ecfcd7e083d42d4e409483e968b4c421b +Subproject commit 2f0302c1d124433f0e1af5ae6c3bb7e4a64ca520 diff --git a/rbf.go b/rbf.go index dff5a1123..4fa2f8900 100644 --- a/rbf.go +++ b/rbf.go @@ -435,6 +435,10 @@ func (tx *RBFTx) GetSortedFieldViewList(idx *Index, shard uint64) (fvs []txkey.F return tx.tx.GetSortedFieldViewList() } +func (tx *RBFTx) GetFieldSizeBytes(index, field string) (uint64, error) { + return tx.tx.GetSizeBytesWithPrefix(string(txkey.FieldPrefix(index, field))) +} + // rbfName returns a NULL-separated key used for identifying bitmap maps in RBF. func rbfName(index, field, view string, shard uint64) string { return string(txkey.Prefix(index, field, view, shard)) diff --git a/rbf/tx.go b/rbf/tx.go index a952104ec..5e7c35001 100644 --- a/rbf/tx.go +++ b/rbf/tx.go @@ -839,6 +839,33 @@ func (tx *Tx) inusePageSet() (map[uint32]struct{}, error) { return m, nil } +// GetSizeBytesWithPrefix returns the size of bitmaps with a given key prefix. +func (tx *Tx) GetSizeBytesWithPrefix(prefix string) (n uint64, err error) { + records, err := tx.RootRecords() + if err != nil { + return 0, err + } + + // Loop over each bitmap in the database. + for itr := records.Iterator(); !itr.Done(); { + name, pgno := itr.Next() + + // Skip over any bitmaps that don't have a matching prefix. + if !strings.HasPrefix(name.(string), prefix) { + continue + } + + // Traverse the bitmap's b-tree and count the bytes for each page. + if err := tx.walkTree(pgno.(uint32), 0, func(pgno, parent, typ uint32) error { + n += PageSize + return nil + }); err != nil { + return 0, err + } + } + return n, nil +} + // walkTree recursively iterates over a page and all its children. func (tx *Tx) walkTree(pgno, parent uint32, fn func(pgno, parent, typ uint32) error) error { // Read page and iterate over children. @@ -964,6 +991,7 @@ func (tx *Tx) deallocateTree(pgno uint32) error { func (tx *Tx) readPage(pgno uint32) (_ []byte, isHeap bool, err error) { // Meta page is always cached on the transaction. + //fmt.Printf("readPage %d\n", pgno) if pgno == 0 { return tx.meta[:], false, nil } @@ -971,7 +999,7 @@ func (tx *Tx) readPage(pgno uint32) (_ []byte, isHeap bool, err error) { // Verify page number requested is within current size of database. pageN := readMetaPageN(tx.meta[:]) if pgno > pageN { - return nil, false, fmt.Errorf("rbf: page read out of bounds: pgno=%d max=%d", pgno, pageN) + return nil, false, fmt.Errorf("rbf: page read out of bounds: pgno=%d max=%d", pgno, pageN-1) } // Check if page has been updated in this tx. diff --git a/rrtx.go b/rrtx.go index 8c75a2924..ca89b6d3d 100644 --- a/rrtx.go +++ b/rrtx.go @@ -589,6 +589,10 @@ func (tx *RoaringTx) GetSortedFieldViewList(idx *Index, shard uint64) (fvs []txk return } +func (tx *RoaringTx) GetFieldSizeBytes(index, field string) (uint64, error) { + return 0, nil +} + //////// registrar and wrapper machinery // roaringRegistrar mirrors the machinery expected diff --git a/server/grpc.go b/server/grpc.go index c2e68bb1f..f7fe1192c 100644 --- a/server/grpc.go +++ b/server/grpc.go @@ -20,6 +20,7 @@ import ( "fmt" "net" "net/http" + "strconv" "strings" "sync" "time" @@ -34,6 +35,7 @@ import ( "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials" + "google.golang.org/grpc/metadata" "google.golang.org/grpc/reflection" "google.golang.org/grpc/status" ) @@ -146,6 +148,13 @@ func (h *GRPCHandler) QuerySQL(req *pb.QuerySQLRequest, stream pb.Pilosa_QuerySQ return err } + err = stream.SendHeader(metadata.New(map[string]string{ + "duration": strconv.Itoa(int(duration)), + })) + if err != nil { + return errors.Wrap(err, "sending header") + } + err = newDurationRowser(results, duration).ToRows(stream.Send) if err != nil { return errors.Wrap(err, "streaming result") @@ -183,7 +192,15 @@ func (h *GRPCHandler) QuerySQLUnary(ctx context.Context, req *pb.QuerySQLRequest if err != nil { return nil, err } - table.Duration = int64(time.Since(start)) + duration := time.Since(start) + table.Duration = int64(duration) + err = grpc.SendHeader(ctx, metadata.New(map[string]string{ + "duration": strconv.Itoa(int(duration)), + })) + if err != nil { + return nil, errors.Wrap(err, "sending header") + } + return table, nil } @@ -197,6 +214,7 @@ func (h *GRPCHandler) QueryPQL(req *pb.QueryPQLRequest, stream pb.Pilosa_QueryPQ t := time.Now() resp, err := h.api.Query(stream.Context(), &query) durQuery := time.Since(t) + // TODO: what about resp.CollumnAttrSets? if err != nil { return errToStatusError(err) @@ -215,6 +233,13 @@ func (h *GRPCHandler) QueryPQL(req *pb.QueryPQLRequest, stream pb.Pilosa_QueryPQ return errors.Wrap(err, "wrapping as type ToRowser") } + err = stream.SendHeader(metadata.New(map[string]string{ + "duration": strconv.Itoa(int(durQuery)), + })) + if err != nil { + return errors.Wrap(err, "sending header") + } + t = time.Now() if err := newDurationRowser(toRowser, durQuery).ToRows(stream.Send); err != nil { return errToStatusError(err) @@ -262,7 +287,14 @@ func (h *GRPCHandler) QueryPQLUnary(ctx context.Context, req *pb.QueryPQLRequest } durFormat := time.Since(t) - table.Duration = int64(durQuery + durFormat) + duration := durQuery + durFormat + table.Duration = int64(duration) + err = grpc.SendHeader(ctx, metadata.New(map[string]string{ + "duration": strconv.Itoa(int(duration)), + })) + if err != nil { + return nil, errors.Wrap(err, "sending header") + } h.stats.Timing(pilosa.MetricGRPCUnaryQueryDurationSeconds, durQuery, 0.1) h.stats.Timing(pilosa.MetricGRPCUnaryFormatDurationSeconds, durFormat, 0.1) diff --git a/server/grpc_test.go b/server/grpc_test.go index 3cc808bb2..d24aa0cc3 100644 --- a/server/grpc_test.go +++ b/server/grpc_test.go @@ -29,7 +29,9 @@ import ( "github.com/pilosa/pilosa/v2/sql" "github.com/pilosa/pilosa/v2/test" "github.com/pkg/errors" + "google.golang.org/grpc" "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" "google.golang.org/grpc/status" ) @@ -354,9 +356,11 @@ func TestQueryPQLUnary(t *testing.T) { i := m.MustCreateIndex(t, "i", pilosa.IndexOptions{}) m.MustCreateField(t, i.Name(), "f", pilosa.OptFieldKeys()) - ctx := context.Background() gh := server.NewGRPCHandler(m.API) + stream := &MockServerTransportStream{} + ctx := grpc.NewContextWithServerTransportStream(context.Background(), stream) + resp, err := gh.QueryPQLUnary(ctx, &pb.QueryPQLRequest{ Index: i.Name(), Pql: `Set(0, f="zero")`, @@ -369,6 +373,10 @@ func TestQueryPQLUnary(t *testing.T) { if resp.Duration == 0 { t.Fatal("duration not recorded") } + duration, err := stream.GetDuration() + if duration == 0 || err != nil { + t.Fatal("duration header not recorded") + } _, err = gh.QueryPQLUnary(ctx, &pb.QueryPQLRequest{ Index: i.Name(), @@ -400,6 +408,11 @@ func TestQueryPQL(t *testing.T) { t.Fatal(err) } + duration, err := mock.GetDuration() + if duration == 0 || err != nil { + t.Fatal("duration header not recorded") + } + if len(mock.Results) != 1 { t.Fatal("expecting one result") } @@ -481,7 +494,9 @@ type ( func TestQuerySQL(t *testing.T) { - ctx := context.Background() + stream := &MockServerTransportStream{} + ctx := grpc.NewContextWithServerTransportStream(context.Background(), stream) + gh, tearDownFunc := setUpTestQuerySQLUnary(ctx, t) defer tearDownFunc() @@ -924,6 +939,11 @@ func TestQuerySQL(t *testing.T) { if resp.Duration == 0 { t.Fatal("duration not recorded") } + duration, err := stream.GetDuration() + if duration == 0 || err != nil { + t.Fatal("duration header not recorded") + } + stream.ClearMD() tr := toTableResponse(resp) if err := test.eq(test.exp, tr); err != nil { t.Fatalf("sql: %s, error: %+v", test.sql, err) @@ -942,6 +962,10 @@ func TestQuerySQL(t *testing.T) { if mock.Results[0].Duration == 0 { t.Fatal("duration not recorded") } + duration, err := mock.GetDuration() + if duration == 0 || err != nil { + t.Fatal("duration header not recorded") + } if len(mock.Results) > 1 && mock.Results[1].Duration != 0 { t.Fatal("duration on second result expected to be zero") } @@ -953,7 +977,8 @@ func TestQuerySQL(t *testing.T) { func TestQuerySQLUnaryWithError(t *testing.T) { - ctx := context.Background() + stream := &MockServerTransportStream{} + ctx := grpc.NewContextWithServerTransportStream(context.Background(), stream) gh, tearDownFunc := setUpTestQuerySQLUnary(ctx, t) defer tearDownFunc() @@ -999,7 +1024,8 @@ func TestCRUDIndexes(t *testing.T) { m := test.RunCommand(t) defer m.Close() - ctx := context.Background() + stream := &MockServerTransportStream{} + ctx := grpc.NewContextWithServerTransportStream(context.Background(), stream) gh := server.NewGRPCHandler(m.API) t.Run("CreateIndex", func(t *testing.T) { @@ -1383,7 +1409,43 @@ func equalUnordered(exp tableResponse, got tableResponse) error { return nil } +type MockServerTransportStream struct { + header metadata.MD +} + +func (stream *MockServerTransportStream) Method() string { + return "" +} + +func (stream *MockServerTransportStream) SetHeader(md metadata.MD) error { + // Should probably merge md with value of stream.header, but this works since we have only one metadata value + stream.header = md + return nil +} + +func (stream *MockServerTransportStream) SendHeader(md metadata.MD) error { + stream.header = md + return nil +} + +func (stream *MockServerTransportStream) SetTrailer(md metadata.MD) error { + return nil +} + +func (stream *MockServerTransportStream) GetDuration() (int, error) { + duration, ok := stream.header["duration"] + if ok { + return strconv.Atoi(duration[0]) + } + return 0, errors.New("duration not recorded") +} + +func (stream *MockServerTransportStream) ClearMD() { + stream.header = metadata.New(map[string]string{}) +} + type mockPilosa_QuerySQLServer struct { + MockServerTransportStream pb.Pilosa_QuerySQLServer Results []*pb.RowResponse } @@ -1393,6 +1455,18 @@ func (m *mockPilosa_QuerySQLServer) Send(result *pb.RowResponse) error { return nil } +func (m *mockPilosa_QuerySQLServer) SendHeader(md metadata.MD) error { + return m.MockServerTransportStream.SendHeader(md) +} + +func (m *mockPilosa_QuerySQLServer) SetHeader(md metadata.MD) error { + return m.MockServerTransportStream.SetHeader(md) +} + +func (m *mockPilosa_QuerySQLServer) SetTrailer(md metadata.MD) { + _ = m.MockServerTransportStream.SetTrailer(md) +} + func (m *mockPilosa_QuerySQLServer) Context() context.Context { return context.Background() } diff --git a/server/handler_test.go b/server/handler_test.go index 705cfdb5c..834399587 100644 --- a/server/handler_test.go +++ b/server/handler_test.go @@ -40,6 +40,7 @@ import ( pb "github.com/pilosa/pilosa/v2/proto" "github.com/pilosa/pilosa/v2/server" "github.com/pilosa/pilosa/v2/test" + "google.golang.org/grpc" ) func TestHandler_PostSchemaCluster(t *testing.T) { @@ -396,10 +397,22 @@ func TestHandler_Endpoints(t *testing.T) { } for _, nodeUsage := range nodeUsages { - if len(nodeUsage.Disk.Indexes) != 2 { - t.Fatalf("wrong length index size list: %#v", nodeUsage.Disk.Indexes) + numIndexes := len(nodeUsage.Disk.IndexUsage) + if nodeUsage.Disk.TotalUse < 75000 || nodeUsage.Disk.TotalUse > 300000 { + // 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 < 300k, got %d", nodeUsage.Disk.TotalUse) + } + if numIndexes != 2 { + t.Fatalf("wrong length index usage list: expected %d, got %d", 2, 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) { @@ -1505,7 +1518,9 @@ func TestQueryHistory(t *testing.T) { test.Do(t, "POST", cmd.URL()+"/index/i0/field/f0", "") gh := server.NewGRPCHandler(cmd.API) - _, err = gh.QuerySQLUnary(context.Background(), &pb.QuerySQLRequest{ + stream := &MockServerTransportStream{} + ctx := grpc.NewContextWithServerTransportStream(context.Background(), stream) + _, err = gh.QuerySQLUnary(ctx, &pb.QuerySQLRequest{ Sql: `select * from i0`, }) diff --git a/stattx.go b/stattx.go index 02b60d6e1..7009de9ac 100644 --- a/stattx.go +++ b/stattx.go @@ -681,3 +681,7 @@ func (c *statTx) Sn() int64 { func (c *statTx) GetSortedFieldViewList(idx *Index, shard uint64) (fvs []txkey.FieldView, err error) { return c.b.GetSortedFieldViewList(idx, shard) } + +func (tx *statTx) GetFieldSizeBytes(index, field string) (uint64, error) { + return 0, nil +} diff --git a/test/pilosa.go b/test/pilosa.go index 264e3f4c9..7cdac8a10 100644 --- a/test/pilosa.go +++ b/test/pilosa.go @@ -41,6 +41,13 @@ type Command struct { commandOptions []server.CommandOption } +func OptTxSrc(src string) server.CommandOption { + return func(m *server.Command) error { + m.Config.Txsrc = src + return nil + } +} + func OptAllowedOrigins(origins []string) server.CommandOption { return func(m *server.Command) error { m.Config.Handler.AllowedOrigins = origins diff --git a/tx.go b/tx.go index 771e94ff8..f33648cb0 100644 --- a/tx.go +++ b/tx.go @@ -219,6 +219,8 @@ type Tx interface { // GetSortedFieldViewList gets the set of FieldView(s) GetSortedFieldViewList(idx *Index, shard uint64) (fvs []txkey.FieldView, err error) + + GetFieldSizeBytes(index, field string) (uint64, error) } // Closer is used by Finders diff --git a/txfactory.go b/txfactory.go index d8aaf3aee..811c33570 100644 --- a/txfactory.go +++ b/txfactory.go @@ -593,40 +593,156 @@ func (f *TxFactory) DumpAll() { f.dbPerShard.DumpAll() } -func (f *TxFactory) IndexSizes() (index2bytes map[string]int64, err error) { - // Open storage directory. - index2bytes = make(map[string]int64) - dirName, err := expandDirName(f.holder.path) +// 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) { + indexUsage := make(map[string]IndexUsage) + holderPath, err := expandDirName(f.holder.path) if err != nil { - return index2bytes, errors.Wrap(err, "expanding data directory") + return indexUsage, 0, errors.Wrap(err, "expanding data directory") } idxs := f.holder.Indexes() + qcx := f.NewQcx() + defer qcx.Abort() for _, idx := range idxs { index := idx.name - fullName := path.Join(dirName, index) - roaringAndMeta, err := directoryUsage(fullName) - if err != nil { - return index2bytes, errors.Wrap(err, "getting disk usage for roaring and meta") + indexPath := path.Join(holderPath, index) + + // field usage + fieldUsages := make(map[string]FieldUsage) + fragmentsTotal := uint64(0) + fieldKeysTotal := uint64(0) + fieldMetaBytesTotal := uint64(0) + fieldsTotal := uint64(0) + flds := idx.Fields() + for _, fld := range flds { + field := fld.Name() + if field == "_keys" { + continue + } + fUsage, err := f.fieldUsage(indexPath, fld) + if err != nil { + return indexUsage, 0, errors.Wrapf(err, "getting disk usage for index (%s)", index) + } + + // non-roaring field usage + fragmentUsage := uint64(0) + + for _, shard := range fld.AvailableShards(true).Slice() { + if err := func() error { + tx, finisher, err := qcx.GetTx(Txo{Write: !writable, Index: idx, Shard: shard}) + if err != nil { + return errors.Wrap(err, "qcx.GetTx") + } + defer finisher(nil) + + fieldBytes, err := tx.GetFieldSizeBytes(index, field) + if err != nil { + return errors.Wrapf(err, "getting disk usage for non-roaring fragments (%s)", field) + } + fragmentUsage += fieldBytes + return nil + }(); err != nil { + return indexUsage, 0, err + } + } + + // add non-roaring to roaring + fUsage.Fragments += fragmentUsage + fUsage.Total += fragmentUsage + + // add to running total + fieldMetaBytesTotal += fUsage.Metadata + fieldKeysTotal += fUsage.Keys + fragmentsTotal += fUsage.Fragments + fieldsTotal += fUsage.Total + + fieldUsages[field] = fUsage } - fullName += ".index.txstores@@@" - rbfOrLmdb, err := directoryUsage(fullName) + + // index metadata, e.g. columnAttrs + indexMetaBytes, err := directoryUsage(indexPath, false) if err != nil { - return index2bytes, errors.Wrap(err, "getting disk usage for backend") + return indexUsage, 0, errors.Wrapf(err, "getting disk usage for index metadata (%s)", index) + } + + // index keys usage + indexKeysBytes := uint64(0) + if idx.keys { + keysPath := path.Join(indexPath, translateStoreDir) + indexKeysBytes, _ = directoryUsage(keysPath, true) // if directory doesn't exist, size = 0 + } + + indexUsage[index] = IndexUsage{ + Total: indexMetaBytes + indexKeysBytes + fieldsTotal, + Metadata: indexMetaBytes + fieldMetaBytesTotal, + IndexKeys: indexKeysBytes, + FieldKeysTotal: fieldKeysTotal, + Fragments: fragmentsTotal, + Fields: fieldUsages, } - index2bytes[index] = roaringAndMeta + rbfOrLmdb } - return index2bytes, nil + // node metadata, e.g. id allocator + nodeMetaBytes, err := directoryUsage(holderPath, false) + if err != nil { + return indexUsage, 0, errors.Wrapf(err, "getting disk usage for node metadata") + } + + return indexUsage, nodeMetaBytes, nil } -func directoryUsage(fname string) (int64, error) { - if !DirExists(fname) { - return 0, nil +// fieldUsage computes the sum of filesizes used by a field in +// the filesystem tree (roaring storage), broken down by keys and fragments. +func (f *TxFactory) fieldUsage(indexPath string, fld *Field) (FieldUsage, error) { + fieldUsage := FieldUsage{} + + field := fld.name + + // row keys + keysBytes := int64(0) + var err error + keysBytes, err = fileSize(fld.TranslateStorePath()) + if err != nil { + // if file doesn't exist, size = 0 + keysBytes = 0 } - var size int64 + // field metadata, e.g. rowAttrs + fieldPath := path.Join(indexPath, field) + metaBytes, err := directoryUsage(fieldPath, false) // this includes keys + if err != nil { + return fieldUsage, errors.Wrapf(err, "getting disk usage for field meta (%s)", field) + } + + // fragment data + viewsPath := path.Join(fieldPath, "views") + fragmentBytes := uint64(0) + if dirExists(viewsPath) { + fragmentBytes, err = directoryUsage(viewsPath, true) + if err != nil { + return fieldUsage, errors.Wrapf(err, "getting disk usage for field fragments (%s)", field) + } + } + + fieldUsage = FieldUsage{ + Total: metaBytes + fragmentBytes, // metaBytes includes keys + Metadata: metaBytes - uint64(keysBytes), + Fragments: fragmentBytes, + Keys: uint64(keysBytes), + } + + return fieldUsage, nil +} + +func directoryUsage(fname string, recursive bool) (uint64, error) { + if !dirExists(fname) { + return 0, errors.Errorf("directory does not exist (%s)", fname) + } + + var size uint64 dir, err := os.Open(fname) if err != nil { @@ -640,14 +756,14 @@ func directoryUsage(fname string) (int64, error) { } for _, file := range files { - if file.IsDir() { - sz, err := directoryUsage(path.Join(fname, file.Name())) + if recursive && file.IsDir() { + sz, err := directoryUsage(path.Join(fname, file.Name()), true) if err != nil { return 0, err } size += sz } else { - size += file.Size() + size += uint64(file.Size()) // NOTE this cast is safe for regular files, not necessarily others } }