Merge pull request #1331 from alanbernstein/field-usage

Show disk usage broken down by field and keys

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 {

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,

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 {

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

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)

lattice

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

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

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.

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

server/handler_test.go

@@ -396,10 +396,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) {

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

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

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

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