上传统计数据时自动清理旧数据

internal/db/models/metric_item_dao.go

@@ -195,7 +195,7 @@ func (this *MetricItemDAO) UpdateItem(tx *dbs.Tx, itemId int64, name string, key
 
 	// 删除旧数据
 	if versionChanged {
-		err := SharedMetricStatDAO.DeleteOldItemStats(tx, itemId, types.Int32(oldItem.Version+1))
+		err := SharedMetricStatDAO.DeleteOldVersionItemStats(tx, itemId, types.Int32(oldItem.Version+1))
 		if err != nil {
 			return err
 		}

internal/db/models/metric_stat_dao.go

@@ -81,8 +81,8 @@ func (this *MetricStatDAO) CreateStat(tx *dbs.Tx, hash string, clusterId int64,
 		})
 }
 
-// DeleteOldItemStats 删除以前版本的统计数据
+// DeleteOldVersionItemStats 删除以前版本的统计数据
-func (this *MetricStatDAO) DeleteOldItemStats(tx *dbs.Tx, itemId int64, version int32) error {
+func (this *MetricStatDAO) DeleteOldVersionItemStats(tx *dbs.Tx, itemId int64, version int32) error {
 	_, err := this.Query(tx).
 		Attr("itemId", itemId).
 		Where("version<:version").
@@ -99,6 +99,17 @@ func (this *MetricStatDAO) DeleteItemStats(tx *dbs.Tx, itemId int64) error {
 	return err
 }
 
+// DeleteNodeItemStats 删除某个节点的统计数据
+func (this *MetricStatDAO) DeleteNodeItemStats(tx *dbs.Tx, nodeId int64, serverId int64, itemId int64, time string) error {
+	_, err := this.Query(tx).
+		Attr("nodeId", nodeId).
+		Attr("serverId", serverId).
+		Attr("itemId", itemId).
+		Attr("time", time).
+		Delete()
+	return err
+}
+
 // CountItemStats 计算统计数据数量
 func (this *MetricStatDAO) CountItemStats(tx *dbs.Tx, itemId int64, version int32) (int64, error) {
 	return this.Query(tx).
@@ -134,6 +145,7 @@ func (this *MetricStatDAO) FindItemStatsAtLastTime(tx *dbs.Tx, itemId int64, ign
 	if err != nil {
 		return nil, err
 	}
+
 	if statOne == nil {
 		return nil, nil
 	}

internal/db/models/metric_stat_dao_test.go

@@ -3,4 +3,16 @@ package models
 import (
 	_ "github.com/go-sql-driver/mysql"
 	_ "github.com/iwind/TeaGo/bootstrap"
+	"github.com/iwind/TeaGo/types"
+	"testing"
 )
+
+func TestNewMetricStatDAO_InsertMany(t *testing.T) {
+	for i := 0; i <= 1; i++ {
+		err := NewMetricStatDAO().CreateStat(nil, types.String(i) + "_v1", 18, 48, 23, 25, []string{"/html" + types.String(i)}, 1, "20210728", 0)
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+	t.Log("done")
+}

internal/db/models/node_value_dao_test.go

@@ -14,7 +14,7 @@ func TestNodeValueDAO_CreateValue(t *testing.T) {
 	m := maps.Map{
 		"hello": "world12344",
 	}
-	err := dao.CreateValue(nil, nodeconfigs.NodeRoleNode, 1, "test", m.AsJSON(), time.Now().Unix())
+	err := dao.CreateValue(nil, 1, nodeconfigs.NodeRoleNode, 1, "test", m.AsJSON(), time.Now().Unix())
 	if err != nil {
 		t.Fatal(err)
 	}

internal/rpc/services/service_metric_stat.go

@@ -27,6 +27,12 @@ func (this *MetricStatService) UploadMetricStats(ctx context.Context, req *pb.Up
 		return nil, err
 	}
 
+	// 删除旧的数据
+	err = models.SharedMetricStatDAO.DeleteNodeItemStats(tx, nodeId, req.ServerId, req.ItemId, req.Time)
+	if err != nil {
+		return nil, err
+	}
+
 	for _, stat := range req.MetricStats {
 		err := models.SharedMetricStatDAO.CreateStat(tx, stat.Hash, clusterId, nodeId, req.ServerId, req.ItemId, stat.Keys, float64(stat.Value), req.Time, req.Version)
 		if err != nil {