add cache view (#12163)

增加cache视图，讨论见: https://github.com/matrixorigin/docs/blob/main/design/system_view.md#cache

方案：
1，用到fileservice目前只有cn, tn。

2，cn上已经有queryservice。因此在queryservice上加取cache视图的接口即可。
tn目前没启动queryservice的。因此需要先在tn上加queryservice，之后与cn上的做法一致。
在tn上加queryservice需要在tn和hakeeper上心跳上加tn.queryservice的地址。用于后续的查询。

3，standalone启动时，cn，tn在一个进程。cn已经有了query service。tn就无需再增加queryservice了。
分布式部署时，tn需要启动query service。

Approved by: @volgariver6, @zhangxu19830126, @nnsgmsone, @triump2020, @reusee, @heni02, @m-schen, @XuPeng-SH, @badboynt1, @aunjgr, @ouyuanning, @w-zr, @qingxinhome, @sukki37

cmd/mo-service/config.go

@@ -134,6 +134,13 @@ type Config struct {
 
 	// MetaCache the config for objectio metacache
 	MetaCache objectio.CacheConfig `toml:"metacache"`
+
+	// IsStandalone denotes the matrixone is running in standalone mode
+	// For the tn does not boost an independent queryservice.
+	// cn,tn shares the same queryservice in standalone mode.
+	// Under distributed deploy mode, cn,tn are independent os process.
+	// they have their own queryservice.
+	IsStandalone bool
 }
 
 // NewConfig return Config with default values.
@@ -263,6 +270,7 @@ func (c *Config) defaultFileServiceDataDir(name string) string {
 
 func (c *Config) createFileService(
 	ctx context.Context,
+	st metadata.ServiceType,
 	defaultName string,
 	perfCounterSet *perfcounter.CounterSet,
 	serviceType metadata.ServiceType,
@@ -358,7 +366,7 @@ func (c *Config) createFileService(
 		// set shared fs perf counter as node perf counter
 		if service.Name() == defines.SharedFileServiceName {
 			perfcounter.Named.Store(
-				perfcounter.NameForNode(nodeUUID),
+				perfcounter.NameForNode(st.String(), nodeUUID),
 				counterSet,
 			)
 		}

cmd/mo-service/config_test.go

@@ -16,6 +16,7 @@ package main
 
 import (
 	"context"
+	"github.com/matrixorigin/matrixone/pkg/pb/metadata"
 	"reflect"
 	"testing"
 
@@ -95,7 +96,7 @@ func TestFileServiceFactory(t *testing.T) {
 		Backend: "DISK-ETL",
 	})
 
-	fs, err := c.createFileService(ctx, "A", globalCounterSet, 0, "")
+	fs, err := c.createFileService(ctx, metadata.ServiceType_CN, "A", globalCounterSet, 0, "")
 	assert.NoError(t, err)
 	assert.NotNil(t, fs)
 }

cmd/mo-service/launch.go

@@ -117,6 +117,8 @@ func startTNServiceCluster(
 
 	for _, file := range files {
 		cfg := NewConfig()
+		// mo boosting in standalone mode
+		cfg.IsStandalone = true
 		if err := parseConfigFromFile(file, cfg); err != nil {
 			return err
 		}

cmd/mo-service/main.go

@@ -183,7 +183,7 @@ func startService(
 		}
 	}
 
-	fs, err := cfg.createFileService(ctx, defines.LocalFileServiceName, globalCounterSet, st, uuid)
+	fs, err := cfg.createFileService(ctx, st, defines.LocalFileServiceName, globalCounterSet, st, uuid)
 	if err != nil {
 		return err
 	}
@@ -282,6 +282,8 @@ func startTNService(
 		ctx = perfcounter.WithCounterSet(ctx, perfCounterSet)
 		cfg.initMetaCache()
 		c := cfg.getTNServiceConfig()
+		//notify the tn service it is in the standalone cluster
+		c.InStandalone = cfg.IsStandalone
 		commonConfigKVMap, _ := dumpCommonConfig(*cfg)
 		s, err := tnservice.NewService(
 			perfCounterSet,

pkg/clusterservice/cluster.go

@@ -276,6 +276,7 @@ func newTNService(tn logpb.TNStore) metadata.TNService {
 		LogTailServiceAddress: tn.LogtailServerAddress,
 		LockServiceAddress:    tn.LockServiceAddress,
 		CtlAddress:            tn.CtlAddress,
+		QueryAddress:          tn.QueryAddress,
 	}
 	v.Shards = make([]metadata.TNShard, 0, len(tn.Shards))
 	for _, s := range tn.Shards {

pkg/cnservice/server_query.go

@@ -21,6 +21,7 @@ import (
 	pblock "github.com/matrixorigin/matrixone/pkg/pb/lock"
 	"github.com/matrixorigin/matrixone/pkg/pb/query"
 	"github.com/matrixorigin/matrixone/pkg/pb/txn"
+	"github.com/matrixorigin/matrixone/pkg/perfcounter"
 	"github.com/matrixorigin/matrixone/pkg/queryservice"
 	"github.com/matrixorigin/matrixone/pkg/sql/plan/function/ctl"
 	"github.com/matrixorigin/matrixone/pkg/txn/client"
@@ -42,6 +43,7 @@ func (s *service) initQueryCommandHandler() {
 	s.queryService.AddHandleFunc(query.CmdMethod_TraceSpan, s.handleTraceSpan, false)
 	s.queryService.AddHandleFunc(query.CmdMethod_GetLockInfo, s.handleGetLockInfo, false)
 	s.queryService.AddHandleFunc(query.CmdMethod_GetTxnInfo, s.handleGetTxnInfo, false)
+	s.queryService.AddHandleFunc(query.CmdMethod_GetCacheInfo, s.handleGetCacheInfo, false)
 }
 
 func (s *service) handleKillConn(ctx context.Context, req *query.Request, resp *query.Response) error {
@@ -198,3 +200,17 @@ func copyTxnInfo(src client.Lock) *query.TxnLockInfo {
 	}
 	return dst
 }
+
+func (s *service) handleGetCacheInfo(ctx context.Context, req *query.Request, resp *query.Response) error {
+	resp.GetCacheInfoResponse = new(query.GetCacheInfoResponse)
+
+	perfcounter.GetCacheStats(func(infos []*query.CacheInfo) {
+		for _, info := range infos {
+			if info != nil {
+				resp.GetCacheInfoResponse.CacheInfoList = append(resp.GetCacheInfoResponse.CacheInfoList, info)
+			}
+		}
+	})
+
+	return nil
+}

pkg/cnservice/upgrader/new_add_table.go

@@ -314,8 +314,25 @@ var MoTransactionsView = &table.Table{
 	CreateTableSql: "drop view if exists `mo_catalog`.`mo_transactions`;",
 }
 
-var registeredViews = []*table.Table{processlistView, MoLocksView, MoVariablesView, MoTransactionsView}
-var needUpgradeNewView = []*table.Table{PARTITIONSView, STATISTICSView, MoSessionsView, SqlStatementHotspotView, MoLocksView, MoConfigurationsView, MoVariablesView, MoTransactionsView}
+var MoCacheView = &table.Table{
+	Account:  table.AccountAll,
+	Database: catalog.MO_CATALOG,
+	Table:    "mo_cache",
+	Columns: []table.Column{
+		table.StringColumn("node_type", "the type of the node. cn,tn"),
+		table.StringColumn("node_id", "the id of node"),
+		table.StringColumn("type", "the type of fileservice cache. memory, disk_cache"),
+		table.StringColumn("used", "used bytes of the cache"),
+		table.StringColumn("free", "free bytes of the cache"),
+		table.StringColumn("hit_ratio", "the hit ratio of the cache"),
+	},
+	CreateViewSql: "CREATE VIEW IF NOT EXISTS `mo_catalog`.`mo_cache` AS SELECT * FROM mo_cache() AS mo_cache_tmp;",
+	//actually drop view here
+	CreateTableSql: "drop view if exists `mo_catalog`.`mo_cache`;",
+}
+
+var registeredViews = []*table.Table{processlistView, MoLocksView, MoVariablesView, MoTransactionsView, MoCacheView}
+var needUpgradeNewView = []*table.Table{PARTITIONSView, STATISTICSView, MoSessionsView, SqlStatementHotspotView, MoLocksView, MoConfigurationsView, MoVariablesView, MoTransactionsView, MoCacheView}
 
 var InformationSchemaSCHEMATA = &table.Table{
 	Account:  table.AccountAll,

pkg/frontend/authenticate.go

@@ -824,6 +824,7 @@ var (
 		"mo_locks":                    0,
 		"mo_variables":                0,
 		"mo_transactions":             0,
+		"mo_cache":                    0,
 	}
 	configInitVariables = map[string]int8{
 		"save_query_result":      0,
@@ -855,6 +856,7 @@ var (
 		"mo_locks":                    0,
 		"mo_variables":                0,
 		"mo_transactions":             0,
+		"mo_cache":                    0,
 	}
 	createDbInformationSchemaSql = "create database information_schema;"
 	createAutoTableSql           = fmt.Sprintf(`create table if not exists %s (
@@ -1034,6 +1036,7 @@ var (
 		`CREATE VIEW IF NOT EXISTS mo_locks AS SELECT * FROM mo_locks() AS mo_locks_tmp;`,
 		`CREATE VIEW IF NOT EXISTS mo_variables AS SELECT * FROM mo_catalog.mo_mysql_compatibility_mode;`,
 		`CREATE VIEW IF NOT EXISTS mo_transactions AS SELECT * FROM mo_transactions() AS mo_transactions_tmp;`,
+		`CREATE VIEW IF NOT EXISTS mo_cache AS SELECT * FROM mo_cache() AS mo_cache_tmp;`,
 	}
 
 	//drop tables for the tenant
@@ -1052,6 +1055,7 @@ var (
 		`drop view if exists mo_catalog.mo_locks;`,
 		`drop view if exists mo_catalog.mo_variables;`,
 		`drop view if exists mo_catalog.mo_transactions;`,
+		`drop view if exists mo_catalog.mo_cache;`,
 	}
 	dropMoPubsSql         = `drop table if exists mo_catalog.mo_pubs;`
 	dropAutoIcrColSql     = fmt.Sprintf("drop table if exists mo_catalog.`%s`;", catalog.MOAutoIncrTable)

pkg/hakeeper/rsm.go

@@ -721,6 +721,7 @@ func (s *stateMachine) handleClusterDetailsQuery(cfg Config) *pb.ClusterDetails
 			LockServiceAddress:   info.LockServiceAddress,
 			CtlAddress:           info.CtlAddress,
 			ConfigData:           info.ConfigData,
+			QueryAddress:         info.QueryAddress,
 		}
 		cd.TNStores = append(cd.TNStores, n)
 	}

pkg/pb/logservice/logservice.go

@@ -169,6 +169,7 @@ func (s *TNState) Update(hb TNStoreHeartbeat, tick uint64) {
 	if hb.ConfigData != nil {
 		storeInfo.ConfigData = hb.ConfigData
 	}
+	storeInfo.QueryAddress = hb.QueryAddress
 	s.Stores[hb.UUID] = storeInfo
 }