缓存一些看板数据，以加快看板打开速度

2025-11-09 03:50:26 +08:00 · 2022-03-07 11:45:58 +08:00
parent b66b8d198a
commit 95de3b12e2
4 changed files with 540 additions and 299 deletions
--- a/internal/rpc/tasks/cache_task_manager.go
+++ b/internal/rpc/tasks/cache_task_manager.go
@@ -0,0 +1,494 @@
+// Copyright 2022 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+
+package tasks
+
+import (
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models/stats"
+	"github.com/TeaOSLab/EdgeAPI/internal/goman"
+	"github.com/TeaOSLab/EdgeAPI/internal/remotelogs"
+	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
+	"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
+	"github.com/iwind/TeaGo/dbs"
+	"github.com/iwind/TeaGo/lists"
+	"github.com/iwind/TeaGo/types"
+	timeutil "github.com/iwind/TeaGo/utils/time"
+	"sync"
+	"time"
+)
+
+var SharedCacheTaskManager = NewCacheTaskManager()
+
+const (
+	// global keys
+
+	CacheKeyFindAllMetricDataCharts = "findAllMetricDataCharts"
+	CacheKeyFindGlobalTopDomains    = "globalFindTopDomains"
+
+	// cluster keys
+
+	CacheKeyFindNodeClusterMetricDataCharts = "findNodeClusterMetricDataCharts"
+
+	// node keys
+
+	CacheKeyFindNodeMetricDataCharts = "findNodeMetricDataCharts"
+
+	// server keys
+
+	CacheKeyFindServerMetricDataCharts = "findServerMetricDataCharts"
+)
+
+func init() {
+	dbs.OnReadyDone(func() {
+		goman.New(func() {
+			_ = SharedCacheTaskManager.Loop()
+			SharedCacheTaskManager.Start()
+		})
+	})
+}
+
+type CacheTaskManager struct {
+	cacheMap map[string]interface{}
+	locker   sync.Mutex
+
+	nodeTasksTooSlow bool // 记录节点任务是否太慢，如果太慢就不再后台执行
+
+	ticker *time.Ticker
+}
+
+func NewCacheTaskManager() *CacheTaskManager {
+	return &CacheTaskManager{
+		cacheMap: map[string]interface{}{},
+		ticker:   time.NewTicker(5 * time.Minute),
+	}
+}
+
+func (this *CacheTaskManager) Start() {
+	for range this.ticker.C {
+		err := this.Loop()
+		if err != nil {
+			remotelogs.Error("CACHE_TASK_MANAGER", err.Error())
+		}
+	}
+}
+
+func (this *CacheTaskManager) Loop() error {
+	var tx *dbs.Tx
+
+	// Admin看板指标数据
+	{
+		value, err := this.findAllMetricDataCharts(tx)
+		if err != nil {
+			return err
+		}
+		this.locker.Lock()
+		this.cacheMap[CacheKeyFindAllMetricDataCharts] = value
+		this.locker.Unlock()
+	}
+
+	{
+		var hourFrom = timeutil.Format("YmdH", time.Now().Add(-23*time.Hour))
+		var hourTo = timeutil.Format("YmdH")
+
+		value, err := stats.SharedServerDomainHourlyStatDAO.FindTopDomainStats(tx, hourFrom, hourTo, 10)
+		if err != nil {
+			return err
+		}
+
+		var composedKey = CacheKeyFindGlobalTopDomains
+
+		this.locker.Lock()
+		this.cacheMap[composedKey] = value
+		this.locker.Unlock()
+	}
+
+	// 集群数据
+	clusterIds, err := models.SharedNodeClusterDAO.FindAllEnabledNodeClusterIds(tx)
+	if err != nil {
+		return err
+	}
+	for _, clusterId := range clusterIds {
+		// metric charts
+		{
+			var category = serverconfigs.MetricItemCategoryHTTP
+			var composedKey = CacheKeyFindNodeClusterMetricDataCharts + "@" + types.String(clusterId) + "@" + category
+
+			value, err := this.findNodeClusterMetricDataCharts(tx, clusterId, 0, 0, category)
+			if err != nil {
+				return err
+			}
+			this.locker.Lock()
+			this.cacheMap[composedKey] = value
+			this.locker.Unlock()
+		}
+
+		// nodes
+		var before = time.Now()
+		if !this.nodeTasksTooSlow {
+			nodeIds, err := models.SharedNodeDAO.FindEnabledNodeIdsWithClusterId(tx, clusterId)
+			if err != nil {
+				return err
+			}
+			for _, nodeId := range nodeIds {
+				// metric
+				{
+					var category = serverconfigs.MetricItemCategoryHTTP
+					var composedKey = CacheKeyFindNodeMetricDataCharts + "@" + types.String(clusterId) + "@" + types.String(nodeId) + "@" + category
+
+					value, err := this.findNodeClusterMetricDataCharts(tx, clusterId, nodeId, 0, category)
+					if err != nil {
+						return err
+					}
+					this.locker.Lock()
+					this.cacheMap[composedKey] = value
+					this.locker.Unlock()
+				}
+			}
+		}
+		var cost = time.Since(before).Seconds()
+		if cost > 600 {
+			this.nodeTasksTooSlow = true
+		}
+	}
+
+	return nil
+}
+
+func (this *CacheTaskManager) Get(key string) (value interface{}, ok bool) {
+	this.locker.Lock()
+	defer this.locker.Unlock()
+
+	if key == CacheKeyFindAllMetricDataCharts {
+		value, ok = this.cacheMap[key]
+		return
+	}
+
+	return
+}
+
+func (this *CacheTaskManager) GetCluster(key string, clusterId int64, category string) (value interface{}, ok bool) {
+	this.locker.Lock()
+	defer this.locker.Unlock()
+
+	var composedKey = key + "@" + types.String(clusterId) + "@" + category
+	value, ok = this.cacheMap[composedKey]
+	return
+}
+
+func (this *CacheTaskManager) GetNode(key string, clusterId int64, nodeId int64, category string) (value interface{}, ok bool) {
+	this.locker.Lock()
+	var composedKey = key + "@" + types.String(clusterId) + "@" + types.String(nodeId) + "@" + category
+	value, ok = this.cacheMap[composedKey]
+	this.locker.Unlock()
+
+	if ok {
+		result, err := this.findNodeClusterMetricDataCharts(nil, clusterId, nodeId, 0, category)
+		if err == nil {
+			value = result
+		}
+	}
+
+	return
+}
+
+func (this *CacheTaskManager) GetServer(key string, clusterId int64, serverId int64, category string) (value interface{}, ok bool) {
+	switch key {
+	case CacheKeyFindServerMetricDataCharts:
+		var tx *dbs.Tx
+		v, err := this.findNodeClusterMetricDataCharts(tx, clusterId, 0, serverId, category)
+		if err != nil {
+			return nil, false
+		}
+		return v, true
+	}
+
+	return
+}
+
+func (this *CacheTaskManager) GetGlobalTopDomains() (value interface{}, ok bool) {
+	this.locker.Lock()
+	defer this.locker.Unlock()
+
+	var composedKey = CacheKeyFindGlobalTopDomains
+	value, ok = this.cacheMap[composedKey]
+	return
+}
+
+// 所有集群的指标统计
+func (this *CacheTaskManager) findAllMetricDataCharts(tx *dbs.Tx) (result []*pb.MetricDataChart, err error) {
+	// 集群指标
+	items, err := models.SharedMetricItemDAO.FindAllPublicItems(tx, serverconfigs.MetricItemCategoryHTTP, nil)
+	if err != nil {
+		return nil, err
+	}
+	var pbMetricCharts = []*pb.MetricDataChart{}
+	for _, item := range items {
+		var itemId = int64(item.Id)
+		charts, err := models.SharedMetricChartDAO.FindAllEnabledCharts(tx, itemId)
+		if err != nil {
+			return nil, err
+		}
+
+		for _, chart := range charts {
+			if chart.IsOn == 0 {
+				continue
+			}
+
+			var pbChart = &pb.MetricChart{
+				Id:         int64(chart.Id),
+				Name:       chart.Name,
+				Type:       chart.Type,
+				WidthDiv:   chart.WidthDiv,
+				ParamsJSON: nil,
+				IsOn:       chart.IsOn == 1,
+				MaxItems:   types.Int32(chart.MaxItems),
+				MetricItem: &pb.MetricItem{
+					Id:         itemId,
+					PeriodUnit: item.PeriodUnit,
+					Period:     types.Int32(item.Period),
+					Name:       item.Name,
+					Value:      item.Value,
+					Category:   item.Category,
+					Keys:       item.DecodeKeys(),
+					Code:       item.Code,
+					IsOn:       item.IsOn == 1,
+				},
+			}
+			var pbStats = []*pb.MetricStat{}
+			switch chart.Type {
+			case serverconfigs.MetricChartTypeTimeLine:
+				itemStats, err := models.SharedMetricStatDAO.FindLatestItemStats(tx, itemId, chart.IgnoreEmptyKeys == 1, chart.DecodeIgnoredKeys(), types.Int32(item.Version), 10)
+				if err != nil {
+					return nil, err
+				}
+
+				for _, stat := range itemStats {
+					// 当前时间总和
+					count, total, err := models.SharedMetricSumStatDAO.FindSumAtTime(tx, stat.Time, itemId, types.Int32(item.Version))
+					if err != nil {
+						return nil, err
+					}
+
+					pbStats = append(pbStats, &pb.MetricStat{
+						Id:          int64(stat.Id),
+						Hash:        stat.Hash,
+						ServerId:    0,
+						ItemId:      0,
+						Keys:        stat.DecodeKeys(),
+						Value:       types.Float32(stat.Value),
+						Time:        stat.Time,
+						Version:     0,
+						NodeCluster: nil,
+						Node:        nil,
+						Server:      nil,
+						SumCount:    count,
+						SumTotal:    total,
+					})
+				}
+			default:
+				itemStats, err := models.SharedMetricStatDAO.FindItemStatsAtLastTime(tx, itemId, chart.IgnoreEmptyKeys == 1, chart.DecodeIgnoredKeys(), types.Int32(item.Version), 10)
+				if err != nil {
+					return nil, err
+				}
+				for _, stat := range itemStats {
+					count, total, err := models.SharedMetricSumStatDAO.FindSumAtTime(tx, stat.Time, itemId, types.Int32(item.Version))
+					if err != nil {
+						return nil, err
+					}
+
+					pbStats = append(pbStats, &pb.MetricStat{
+						Id:          int64(stat.Id),
+						Hash:        stat.Hash,
+						ServerId:    0,
+						ItemId:      0,
+						Keys:        stat.DecodeKeys(),
+						Value:       types.Float32(stat.Value),
+						Time:        stat.Time,
+						Version:     0,
+						NodeCluster: nil,
+						Node:        nil,
+						Server:      nil,
+						SumCount:    count,
+						SumTotal:    total,
+					})
+				}
+			}
+			pbMetricCharts = append(pbMetricCharts, &pb.MetricDataChart{
+				MetricChart: pbChart,
+				MetricStats: pbStats,
+			})
+		}
+	}
+	return pbMetricCharts, nil
+}
+
+// 某个集群、节点或者服务的指标统计
+func (this *CacheTaskManager) findNodeClusterMetricDataCharts(tx *dbs.Tx, clusterId int64, nodeId int64, serverId int64, category string) (result []*pb.MetricDataChart, err error) {
+	// 集群指标
+	clusterMetricItems, err := models.SharedNodeClusterMetricItemDAO.FindAllClusterItems(tx, clusterId, category)
+	if err != nil {
+		return nil, err
+	}
+	var pbMetricCharts = []*pb.MetricDataChart{}
+	var metricItemIds = []int64{}
+	var items = []*models.MetricItem{}
+	for _, clusterMetricItem := range clusterMetricItems {
+		if clusterMetricItem.IsOn != 1 {
+			continue
+		}
+		var itemId = int64(clusterMetricItem.ItemId)
+		item, err := models.SharedMetricItemDAO.FindEnabledMetricItem(tx, itemId)
+		if err != nil {
+			return nil, err
+		}
+		if item == nil || item.IsOn == 0 {
+			continue
+		}
+		items = append(items, item)
+		metricItemIds = append(metricItemIds, itemId)
+	}
+
+	publicMetricItems, err := models.SharedMetricItemDAO.FindAllPublicItems(tx, category, nil)
+	if err != nil {
+		return nil, err
+	}
+	for _, item := range publicMetricItems {
+		if item.IsOn != 1 {
+			continue
+		}
+		if lists.ContainsInt64(metricItemIds, int64(item.Id)) {
+			continue
+		}
+		items = append(items, item)
+	}
+
+	for _, item := range items {
+		var itemId = int64(item.Id)
+		charts, err := models.SharedMetricChartDAO.FindAllEnabledCharts(tx, itemId)
+		if err != nil {
+			return nil, err
+		}
+
+		for _, chart := range charts {
+			if chart.IsOn == 0 {
+				continue
+			}
+
+			var pbChart = &pb.MetricChart{
+				Id:         int64(chart.Id),
+				Name:       chart.Name,
+				Type:       chart.Type,
+				WidthDiv:   chart.WidthDiv,
+				ParamsJSON: nil,
+				IsOn:       chart.IsOn == 1,
+				MaxItems:   types.Int32(chart.MaxItems),
+				MetricItem: &pb.MetricItem{
+					Id:         itemId,
+					PeriodUnit: item.PeriodUnit,
+					Period:     types.Int32(item.Period),
+					Name:       item.Name,
+					Value:      item.Value,
+					Category:   item.Category,
+					Keys:       item.DecodeKeys(),
+					Code:       item.Code,
+					IsOn:       item.IsOn == 1,
+				},
+			}
+			var pbStats = []*pb.MetricStat{}
+			switch chart.Type {
+			case serverconfigs.MetricChartTypeTimeLine:
+				var itemStats []*models.MetricStat
+				if serverId > 0 {
+					itemStats, err = models.SharedMetricStatDAO.FindLatestItemStatsWithServerId(tx, serverId, itemId, chart.IgnoreEmptyKeys == 1, chart.DecodeIgnoredKeys(), types.Int32(item.Version), 10)
+				} else if nodeId > 0 {
+					itemStats, err = models.SharedMetricStatDAO.FindLatestItemStatsWithNodeId(tx, nodeId, itemId, chart.IgnoreEmptyKeys == 1, chart.DecodeIgnoredKeys(), types.Int32(item.Version), 10)
+				} else {
+					itemStats, err = models.SharedMetricStatDAO.FindLatestItemStatsWithClusterId(tx, clusterId, itemId, chart.IgnoreEmptyKeys == 1, chart.DecodeIgnoredKeys(), types.Int32(item.Version), 10)
+				}
+				if err != nil {
+					return nil, err
+				}
+
+				for _, stat := range itemStats {
+					// 当前时间总和
+					var count int64
+					var total float32
+					if serverId > 0 {
+						count, total, err = models.SharedMetricSumStatDAO.FindServerSum(tx, serverId, stat.Time, itemId, types.Int32(item.Version))
+					} else if nodeId > 0 {
+						count, total, err = models.SharedMetricSumStatDAO.FindNodeSum(tx, nodeId, stat.Time, itemId, types.Int32(item.Version))
+					} else {
+						count, total, err = models.SharedMetricSumStatDAO.FindClusterSum(tx, clusterId, stat.Time, itemId, types.Int32(item.Version))
+					}
+					if err != nil {
+						return nil, err
+					}
+
+					pbStats = append(pbStats, &pb.MetricStat{
+						Id:          int64(stat.Id),
+						Hash:        stat.Hash,
+						ServerId:    0,
+						ItemId:      0,
+						Keys:        stat.DecodeKeys(),
+						Value:       types.Float32(stat.Value),
+						Time:        stat.Time,
+						Version:     0,
+						NodeCluster: nil,
+						Node:        nil,
+						Server:      nil,
+						SumCount:    count,
+						SumTotal:    total,
+					})
+				}
+			default:
+				var itemStats []*models.MetricStat
+				if serverId > 0 {
+					itemStats, err = models.SharedMetricStatDAO.FindItemStatsWithServerIdAndLastTime(tx, serverId, itemId, chart.IgnoreEmptyKeys == 1, chart.DecodeIgnoredKeys(), types.Int32(item.Version), 10)
+				} else if nodeId > 0 {
+					itemStats, err = models.SharedMetricStatDAO.FindItemStatsWithNodeIdAndLastTime(tx, nodeId, itemId, chart.IgnoreEmptyKeys == 1, chart.DecodeIgnoredKeys(), types.Int32(item.Version), 10)
+				} else {
+					itemStats, err = models.SharedMetricStatDAO.FindItemStatsWithClusterIdAndLastTime(tx, clusterId, itemId, chart.IgnoreEmptyKeys == 1, chart.DecodeIgnoredKeys(), types.Int32(item.Version), 10)
+				}
+				if err != nil {
+					return nil, err
+				}
+				for _, stat := range itemStats {
+					// 当前时间总和
+					var count int64
+					var total float32
+					if serverId > 0 {
+						count, total, err = models.SharedMetricSumStatDAO.FindServerSum(tx, serverId, stat.Time, itemId, types.Int32(item.Version))
+					} else if nodeId > 0 {
+						count, total, err = models.SharedMetricSumStatDAO.FindNodeSum(tx, nodeId, stat.Time, itemId, types.Int32(item.Version))
+					} else {
+						count, total, err = models.SharedMetricSumStatDAO.FindClusterSum(tx, clusterId, stat.Time, itemId, types.Int32(item.Version))
+					}
+					if err != nil {
+						return nil, err
+					}
+
+					pbStats = append(pbStats, &pb.MetricStat{
+						Id:          int64(stat.Id),
+						Hash:        stat.Hash,
+						ServerId:    0,
+						ItemId:      0,
+						Keys:        stat.DecodeKeys(),
+						Value:       types.Float32(stat.Value),
+						Time:        stat.Time,
+						Version:     0,
+						NodeCluster: nil,
+						Node:        nil,
+						Server:      nil,
+						SumCount:    count,
+						SumTotal:    total,
+					})
+				}
+			}
+			pbMetricCharts = append(pbMetricCharts, &pb.MetricDataChart{
+				MetricChart: pbChart,
+				MetricStats: pbStats,
+			})
+		}
+	}
+	return pbMetricCharts, nil
+}