Repository: matrixorigin/matrixone
Pull request #12675 (2023-11-10 12:05:21)

sync `adding metrics for TN mpool and transfer page`
## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [x] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #

## What this PR does / why we need it:
see #12674


Patch:
diff --git a/pkg/util/metric/v2/dashboard/grafana_dashboard.go b/pkg/util/metric/v2/dashboard/grafana_dashboard.go
index 3aa92d89e..bcaab532c 100644
--- a/pkg/util/metric/v2/dashboard/grafana_dashboard.go
+++ b/pkg/util/metric/v2/dashboard/grafana_dashboard.go
@@ -90,6 +90,10 @@ func (c *DashboardCreator) Create() error {
 		return err
 	}
 
+	if err := c.initMemDashboard(); err != nil {
+		return err
+	}
+
 	return nil
 }
 
diff --git a/pkg/util/metric/v2/dashboard/grafana_dashboard_mem.go b/pkg/util/metric/v2/dashboard/grafana_dashboard_mem.go
new file mode 100644
index 000000000..38746f5fa
--- /dev/null
+++ b/pkg/util/metric/v2/dashboard/grafana_dashboard_mem.go
@@ -0,0 +1,67 @@
+// Copyright 2023 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package dashboard
+
+import (
+	"context"
+	"github.com/K-Phoen/grabana/dashboard"
+)
+
+func (c *DashboardCreator) initMemDashboard() error {
+	folder, err := c.createFolder(moFolderName)
+	if err != nil {
+		return err
+	}
+
+	build, err := dashboard.New(
+		"Memory Metrics",
+		c.withRowOptions(
+			c.initTAEMpoolAllocatorRow(),
+		)...)
+	if err != nil {
+		return err
+	}
+	_, err = c.cli.UpsertDashboard(context.Background(), folder, build)
+	return err
+}
+
+func (c *DashboardCreator) initTAEMpoolAllocatorRow() dashboard.Option {
+	return dashboard.Row(
+		"TAE Mpool Allocator",
+		c.withGraph(
+			"TAE Defaulter Allocator",
+			3,
+			`sum(`+c.getMetricWithFilter("mo_mem_mpool_allocated_size", `type="tae_default"`)+`) by (name)`,
+			"{{ name }}"),
+
+		c.withGraph(
+			"TAE Small Allocator",
+			3,
+			`sum(`+c.getMetricWithFilter("mo_mem_mpool_allocated_size", `type="tae_small"`)+`) by (name)`,
+			"{{ name }}"),
+
+		c.withGraph(
+			"TAE Mutable Memory Allocator",
+			3,
+			`sum(`+c.getMetricWithFilter("mo_mem_mpool_allocated_size", `type="tae_mutable"`)+`) by (name)`,
+			"{{ name }}"),
+
+		c.withGraph(
+			"VectorPool Default Allocator",
+			3,
+			`sum(`+c.getMetricWithFilter("mo_mem_mpool_allocated_size", `type="vectorpool_default"`)+`) by (name)`,
+			"{{ name }}"),
+	)
+}
diff --git a/pkg/util/metric/v2/dashboard/grafana_dashboard_task.go b/pkg/util/metric/v2/dashboard/grafana_dashboard_task.go
index f576af3d3..a93a061ab 100644
--- a/pkg/util/metric/v2/dashboard/grafana_dashboard_task.go
+++ b/pkg/util/metric/v2/dashboard/grafana_dashboard_task.go
@@ -33,6 +33,7 @@ func (c *DashboardCreator) initTaskDashboard() error {
 			c.initTaskFlushTableTailRow(),
 			c.initTaskCkpEntryPendingRow(),
 			c.initTaskMergeRow(),
+			c.initTaskMergeTransferPageRow(),
 		)...)
 
 	if err != nil {
@@ -42,6 +43,17 @@ func (c *DashboardCreator) initTaskDashboard() error {
 	return err
 }
 
+func (c *DashboardCreator) initTaskMergeTransferPageRow() dashboard.Option {
+	return dashboard.Row(
+		"Task Merge Transfer Page Length",
+		c.withGraph(
+			"Transfer Page Length",
+			12,
+			`sum(`+c.getMetricWithFilter("mo_task_merge_transfer_page_size", ``)+`)`,
+			"{{ "+c.by+" }}"),
+	)
+}
+
 func (c *DashboardCreator) initTaskFlushTableTailRow() dashboard.Option {
 	return dashboard.Row(
 		"Flush Table Tail Duration",
diff --git a/pkg/util/metric/v2/dashboard/grafana_dashboard_txn.go b/pkg/util/metric/v2/dashboard/grafana_dashboard_txn.go
index d6c2e7e06..64723a4b2 100644
--- a/pkg/util/metric/v2/dashboard/grafana_dashboard_txn.go
+++ b/pkg/util/metric/v2/dashboard/grafana_dashboard_txn.go
@@ -44,6 +44,7 @@ func (c *DashboardCreator) initTxnDashboard() error {
 			c.initTxnDequeuePreparedRow(),
 			c.initTxnDequeuePreparingRow(),
 			c.initTxnRangesLoadedObjectMetaRow(),
+			c.initCNCommittedObjectQuantityRow(),
 		)...)
 	if err != nil {
 		return err
@@ -52,6 +53,23 @@ func (c *DashboardCreator) initTxnDashboard() error {
 	return err
 }
 
+func (c *DashboardCreator) initCNCommittedObjectQuantityRow() dashboard.Option {
+	return dashboard.Row(
+		"Quantity of Object Location The CN Have Committed to TN",
+		c.withGraph(
+			"meta location",
+			6,
+			`sum(`+c.getMetricWithFilter("mo_txn_cn_committed_location_quantity_size", `type="meta_location"`)+`)`,
+			""),
+
+		c.withGraph(
+			"delta location",
+			6,
+			`sum(`+c.getMetricWithFilter("mo_txn_cn_committed_location_quantity_size", `type="delta_location"`)+`)`,
+			""),
+	)
+}
+
 func (c *DashboardCreator) initTxnOverviewRow() dashboard.Option {
 	return dashboard.Row(
 		"Txn overview",
diff --git a/pkg/util/metric/v2/mem.go b/pkg/util/metric/v2/mem.go
new file mode 100644
index 000000000..bd5c45c78
--- /dev/null
+++ b/pkg/util/metric/v2/mem.go
@@ -0,0 +1,32 @@
+// Copyright 2023 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package v2
+
+import "github.com/prometheus/client_golang/prometheus"
+
+var (
+	memMPoolAllocatedSizeGauge = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: "mo",
+			Subsystem: "mem",
+			Name:      "mpool_allocated_size",
+			Help:      "Size of mpool have allocated.",
+		}, []string{"type"})
+
+	MemTAEDefaultAllocatorGauge  = memMPoolAllocatedSizeGauge.WithLabelValues("tae_default")
+	MemTAEMutableAllocatorGauge  = memMPoolAllocatedSizeGauge.WithLabelValues("tae_mutable")
+	MemTAESmallAllocatorGauge    = memMPoolAllocatedSizeGauge.WithLabelValues("tae_small")
+	MemTAEDefaultVectorPoolGauge = memMPoolAllocatedSizeGauge.WithLabelValues("vectorpool_default")
+)
diff --git a/pkg/util/metric/v2/metrics.go b/pkg/util/metric/v2/metrics.go
index 8a51388c7..08f8d576b 100644
--- a/pkg/util/metric/v2/metrics.go
+++ b/pkg/util/metric/v2/metrics.go
@@ -29,19 +29,25 @@ func init() {
 	initTaskMetrics()
 	initRPCMetrics()
 	initPipelineMetrics()
-
+	initMemMetrics()
 	registry.MustRegister(HeartbeatHistogram)
 	registry.MustRegister(HeartbeatFailureCounter)
 	registry.MustRegister(HeartbeatRecvHistogram)
 	registry.MustRegister(HeartbeatRecvFailureCounter)
 }
 
+func initMemMetrics() {
+	registry.MustRegister(memMPoolAllocatedSizeGauge)
+}
+
 func initTaskMetrics() {
 	registry.MustRegister(taskShortDurationHistogram)
 	registry.MustRegister(taskLongDurationHistogram)
 
 	registry.MustRegister(taskScheduledByCounter)
 	registry.MustRegister(taskGeneratedStuffCounter)
+
+	registry.MustRegister(TaskMergeTransferPageLengthGauge)
 }
 
 func initFileServiceMetrics() {
@@ -95,6 +101,7 @@ func initTxnMetrics() {
 
 	registry.MustRegister(TxnRangesLoadedObjectMetaTotalCounter)
 	registry.MustRegister(TxnRangesLoadedObjectHistogram)
+	registry.MustRegister(txnCNCommittedLocationQuantityGauge)
 }
 
 func initRPCMetrics() {
diff --git a/pkg/util/metric/v2/task.go b/pkg/util/metric/v2/task.go
index 3aae05f02..0db1ae37a 100644
--- a/pkg/util/metric/v2/task.go
+++ b/pkg/util/metric/v2/task.go
@@ -64,3 +64,13 @@ var (
 	TaskMergedBlocksCounter = taskGeneratedStuffCounter.WithLabelValues("merged_block")
 	TasKMergedSizeCounter   = taskGeneratedStuffCounter.WithLabelValues("merged_size")
 )
+
+var (
+	TaskMergeTransferPageLengthGauge = prometheus.NewGauge(
+		prometheus.GaugeOpts{
+			Namespace: "mo",
+			Subsystem: "task",
+			Name:      "merge_transfer_page_size",
+			Help:      "Size of merge generated transfer page",
+		})
+)
diff --git a/pkg/util/metric/v2/txn.go b/pkg/util/metric/v2/txn.go
index d9adb3719..9124cdda0 100644
--- a/pkg/util/metric/v2/txn.go
+++ b/pkg/util/metric/v2/txn.go
@@ -89,6 +89,17 @@ var (
 	TxnCommitQueueSizeGauge     = txnQueueSizeGauge.WithLabelValues("commit")
 	TxnWaitActiveQueueSizeGauge = txnQueueSizeGauge.WithLabelValues("wait-active")
 	TxnActiveQueueSizeGauge     = txnQueueSizeGauge.WithLabelValues("active")
+
+	txnCNCommittedLocationQuantityGauge = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: "mo",
+			Subsystem: "txn",
+			Name:      "cn_committed_location_quantity_size",
+			Help:      "Quantity of object location the cn have committed to tn.",
+		}, []string{"type"})
+
+	TxnCNCommittedMetaLocationQuantityGauge  = txnCNCommittedLocationQuantityGauge.WithLabelValues("meta_location")
+	TxnCNCommittedDeltaLocationQuantityGauge = txnCNCommittedLocationQuantityGauge.WithLabelValues("delta_location")
 )
 
 var (
diff --git a/pkg/vm/engine/disttae/txn.go b/pkg/vm/engine/disttae/txn.go
index 6dfbbcaed..4c2565350 100644
--- a/pkg/vm/engine/disttae/txn.go
+++ b/pkg/vm/engine/disttae/txn.go
@@ -287,7 +287,8 @@ func (txn *Transaction) WriteFileLocked(
 		newBat.SetAttributes([]string{catalog.BlockMeta_MetaLoc})
 		newBat.SetVector(0, vector.NewVec(types.T_text.ToType()))
 
-		for _, blk := range vector.MustBytesCol(bat.GetVector(0)) {
+		blkInfos := vector.MustBytesCol(bat.GetVector(0))
+		for _, blk := range blkInfos {
 			blkInfo := *catalog.DecodeBlockInfo(blk)
 			vector.AppendBytes(
 				newBat.GetVector(0),
diff --git a/pkg/vm/engine/tae/common/utils/types.go b/pkg/vm/engine/tae/common/utils/types.go
index 1c301ac99..4fc9fcafc 100644
--- a/pkg/vm/engine/tae/common/utils/types.go
+++ b/pkg/vm/engine/tae/common/utils/types.go
@@ -68,3 +68,5 @@ type SimpleSensor struct {
 	name  string
 	state atomic.Int32
 }
+
+var TransferPageCounter atomic.Int64
diff --git a/pkg/vm/engine/tae/containers/pool.go b/pkg/vm/engine/tae/containers/pool.go
index 717307766..7beadb56d 100644
--- a/pkg/vm/engine/tae/containers/pool.go
+++ b/pkg/vm/engine/tae/containers/pool.go
@@ -44,6 +44,10 @@ func init() {
 	}
 }
 
+func GetDefaultVectorPoolALLocator() *mpool.MPool {
+	return _vectorPoolAlloactor
+}
+
 type VectorPoolOption func(*VectorPool)
 
 func WithAllocationLimit(maxv int) VectorPoolOption {
diff --git a/pkg/vm/engine/tae/db/checkpoint/info.go b/pkg/vm/engine/tae/db/checkpoint/info.go
index 7cf099b87..2e0bd2a98 100644
--- a/pkg/vm/engine/tae/db/checkpoint/info.go
+++ b/pkg/vm/engine/tae/db/checkpoint/info.go
@@ -16,11 +16,10 @@ package checkpoint
 
 import (
 	"context"
-	"time"
-
 	"github.com/matrixorigin/matrixone/pkg/container/types"
 	"github.com/matrixorigin/matrixone/pkg/logutil"
 	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/containers"
+	"time"
 )
 
 type RunnerReader interface {
diff --git a/pkg/vm/engine/tae/db/open.go b/pkg/vm/engine/tae/db/open.go
index 9f55c36ad..60138526d 100644
--- a/pkg/vm/engine/tae/db/open.go
+++ b/pkg/vm/engine/tae/db/open.go
@@ -16,6 +16,10 @@ package db
 
 import (
 	"context"
+	"fmt"
+	v2 "github.com/matrixorigin/matrixone/pkg/util/metric/v2"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common/utils"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/containers"
 	"path"
 	"sync/atomic"
 	"time"
@@ -246,7 +250,39 @@ func Open(ctx context.Context, dirname string, opts *options.Options) (db *DB, e
 
 	db.GCManager.Start()
 
+	go TaeMetricsTask(ctx, db.Runtime)
+
 	// For debug or test
 	// logutil.Info(db.Catalog.SimplePPString(common.PPL2))
 	return
 }
+
+func TaeMetricsTask(ctx context.Context, rt *dbutils.Runtime) {
+	logutil.Info("tae metrics task started")
+	defer logutil.Info("tae metrics task exit")
+
+	timer := time.NewTicker(time.Second * 10)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-timer.C:
+			transferPageSubTask()
+			mpoolAllocatorSubTask(rt)
+		}
+	}
+
+}
+
+func mpoolAllocatorSubTask(rt *dbutils.Runtime) {
+	v2.MemTAEDefaultAllocatorGauge.Set(float64(common.DefaultAllocator.CurrNB()))
+	v2.MemTAEMutableAllocatorGauge.Set(float64(common.MutMemAllocator.CurrNB()))
+	v2.MemTAESmallAllocatorGauge.Set(float64(common.SmallAllocator.CurrNB()))
+	v2.MemTAEDefaultVectorPoolGauge.Set(float64(containers.GetDefaultVectorPoolALLocator().CurrNB()))
+}
+
+func transferPageSubTask() {
+	length := utils.TransferPageCounter.Load()
+	mb_size := length * (4 + 24) * 3 / 2 / (1024 * 1024)
+	logutil.Info(fmt.Sprintf("current transfer page length = %d, mb_size = %d", length, mb_size))
+}
diff --git a/pkg/vm/engine/tae/model/pages.go b/pkg/vm/engine/tae/model/pages.go
index 9684ce275..83a715613 100644
--- a/pkg/vm/engine/tae/model/pages.go
+++ b/pkg/vm/engine/tae/model/pages.go
@@ -55,6 +55,10 @@ func (page *TransferHashPage) Close() {
 	page.hashmap = make(map[uint32]types.Rowid)
 }
 
+func (page *TransferHashPage) Length() int {
+	return len(page.hashmap)
+}
+
 func (page *TransferHashPage) String() string {
 	var w bytes.Buffer
 	_, _ = w.WriteString(fmt.Sprintf("hashpage[%s][%s][Len=%d]",
diff --git a/pkg/vm/engine/tae/model/transfer.go b/pkg/vm/engine/tae/model/transfer.go
index e10688f5a..3727fedb0 100644
--- a/pkg/vm/engine/tae/model/transfer.go
+++ b/pkg/vm/engine/tae/model/transfer.go
@@ -15,6 +15,8 @@
 package model
 
 import (
+	v2 "github.com/matrixorigin/matrixone/pkg/util/metric/v2"
+	"github.com/matrixorigin/matrixone/pkg/vm/engine/tae/common/utils"
 	"sync"
 	"time"
 
@@ -27,6 +29,7 @@ type PageT[T common.IRef] interface {
 	Pin() *common.PinnedItem[T]
 	TTL(time.Time, time.Duration) bool
 	ID() *common.ID
+	Length() int
 }
 
 type TransferTable[T PageT[T]] struct {
@@ -73,14 +76,21 @@ func (table *TransferTable[T]) executeTTL(items []*common.PinnedItem[T]) {
 	if len(items) == 0 {
 		return
 	}
+
+	cnt := 0
+
 	table.Lock()
 	for _, pinned := range items {
+		cnt += pinned.Val.Length()
 		delete(table.pages, *pinned.Item().ID())
 	}
 	table.Unlock()
 	for _, pinned := range items {
 		pinned.Close()
 	}
+
+	v2.TaskMergeTransferPageLengthGauge.Sub(float64(cnt))
+	utils.TransferPageCounter.Add(-int64(cnt))
 }
 
 func (table *TransferTable[T]) RunTTL(now time.Time) {
@@ -103,6 +113,10 @@ func (table *TransferTable[T]) AddPage(page T) (dup bool) {
 		return
 	}
 	table.pages[id] = pinned
+
+	cnt := page.Length()
+	utils.TransferPageCounter.Add(int64(cnt))
+	v2.TaskMergeTransferPageLengthGauge.Add(float64(cnt))
 	return
 }
 
@@ -113,6 +127,15 @@ func (table *TransferTable[T]) DeletePage(id *common.ID) (deleted bool) {
 		return
 	}
 	delete(table.pages, *id)
+
+	// to pass ut
+	if len(table.pages) == 0 || table.pages[*id] == nil {
+		return
+	}
+
+	cnt := table.pages[*id].Val.Length()
+	utils.TransferPageCounter.Add(-int64(cnt))
+	v2.TaskMergeTransferPageLengthGauge.Sub(float64(cnt))
 	return
 }
 
diff --git a/pkg/vm/engine/tae/rpc/handle.go b/pkg/vm/engine/tae/rpc/handle.go
index fd9cb926f..4144c837d 100644
--- a/pkg/vm/engine/tae/rpc/handle.go
+++ b/pkg/vm/engine/tae/rpc/handle.go
@@ -586,8 +586,7 @@ func (h *Handle) HandleInspectTN(
 	return nil, nil
 }
 
-func (h *Handle) prefetchDeleteRowID(ctx context.Context,
-	req *db.WriteReq) error {
+func (h *Handle) prefetchDeleteRowID(ctx context.Context, req *db.WriteReq) error {
 	if len(req.DeltaLocs) == 0 {
 		return nil
 	}
@@ -615,56 +614,72 @@ func (h *Handle) prefetchDeleteRowID(ctx context.Context,
 }
 
 func (h *Handle) prefetchMetadata(ctx context.Context,
-	req *db.WriteReq) error {
+	req *db.WriteReq) (int, error) {
 	if len(req.MetaLocs) == 0 {
-		return nil
+		return 0, nil
 	}
 	//start loading jobs asynchronously,should create a new root context.
+	objCnt := 0
 	var objectName objectio.ObjectNameShort
 	for _, meta := range req.MetaLocs {
 		loc, err := blockio.EncodeLocationFromString(meta)
 		if err != nil {
-			return err
+			return 0, err
 		}
 		if !objectio.IsSameObjectLocVsShort(loc, &objectName) {
 			err := blockio.PrefetchMeta(h.db.Runtime.Fs.Service, loc)
 			if err != nil {
-				return err
+				return 0, err
 			}
+			objCnt++
 			objectName = *loc.Name().Short()
 		}
 	}
-	return nil
+	return objCnt, nil
 }
 
 // EvaluateTxnRequest only evaluate the request ,do not change the state machine of TxnEngine.
 func (h *Handle) EvaluateTxnRequest(
 	ctx context.Context,
 	meta txn.TxnMeta,
-) (err error) {
+) error {
 	h.mu.RLock()
 	txnCtx := h.mu.txnCtxs[string(meta.GetID())]
 	h.mu.RUnlock()
+
+	metaLocCnt := 0
+	deltaLocCnt := 0
+
+	defer func() {
+		if metaLocCnt != 0 {
+			v2.TxnCNCommittedMetaLocationQuantityGauge.Set(float64(metaLocCnt))
+		}
+
+		if deltaLocCnt != 0 {
+			v2.TxnCNCommittedDeltaLocationQuantityGauge.Set(float64(deltaLocCnt))
+		}
+	}()
+
 	for _, e := range txnCtx.reqs {
 		if r, ok := e.(*db.WriteReq); ok {
 			if r.FileName != "" {
 				if r.Type == db.EntryDelete {
 					// start to load deleted row ids
-					err = h.prefetchDeleteRowID(ctx, r)
-					if err != nil {
-						return
+					deltaLocCnt += len(r.DeltaLocs)
+					if err := h.prefetchDeleteRowID(ctx, r); err != nil {
+						return err
 					}
 				} else if r.Type == db.EntryInsert {
-					err = h.prefetchMetadata(ctx, r)
+					objCnt, err := h.prefetchMetadata(ctx, r)
 					if err != nil {
-						return
+						return err
 					}
-
+					metaLocCnt += objCnt
 				}
 			}
 		}
 	}
-	return
+	return nil
 }
 
 func (h *Handle) CacheTxnRequest(


Test patch:
diff --git a/pkg/util/metric/v2/dashboard/grafana_dashboard_test.go b/pkg/util/metric/v2/dashboard/grafana_dashboard_test.go
index ef91cfc7e..3bd3acbb1 100644
--- a/pkg/util/metric/v2/dashboard/grafana_dashboard_test.go
+++ b/pkg/util/metric/v2/dashboard/grafana_dashboard_test.go
@@ -26,6 +26,6 @@ func TestCreateCloudDashboard(t *testing.T) {
 		return
 	}
 
-	c := NewCloudDashboardCreator("http://127.0.0.1", "admin", "admin", "Prometheus")
+	c := NewLocalDashboardCreator("http://127.0.0.1:3000", "admin", "admin", "Prometheus")
 	require.NoError(t, c.Create())
 }