scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'master' into bench

Browse source
This commit is contained in:
Ben Johnson 2021-02-01 14:10:15 -07:00 • committed by GitHub
commit 0900e7b9d2
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
16 changed files with 364 additions and 50 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

46
api.go
View file

@ -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 {

4
bluegreentx.go
View file

@ -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,

4
bolt.go
View file

@ -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 {

4
catcher.go
View file

@ -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
}

6
holder.go
View file

@ -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)

2
lattice

@ -1 +1 @@
Subproject commit 28c2313ecfcd7e083d42d4e409483e968b4c421b
Subproject commit 2f0302c1d124433f0e1af5ae6c3bb7e4a64ca520

4
rbf.go
View file

@ -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))

30
rbf/tx.go
View file

@ -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.

4
rrtx.go
View file

@ -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

36
server/grpc.go
View file

@ -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)

82
server/grpc_test.go
View file

@ -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()
}

21
server/handler_test.go
View file

@ -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`,
})

4
stattx.go
View file

@ -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
}

7
test/pilosa.go
View file

@ -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

2
tx.go
View file

@ -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

158
txfactory.go
View file

@ -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
}
}