实现自动清理服务访问日志

internal/db/db_test.go

@@ -0,0 +1,13 @@
+package db
+
+import (
+	"github.com/iwind/TeaGo/Tea"
+	_ "github.com/iwind/TeaGo/bootstrap"
+	"github.com/iwind/TeaGo/dbs"
+	"testing"
+)
+
+func TestDB(t *testing.T) {
+	Tea.Env = "prod"
+	t.Log(dbs.Default())
+}

internal/db/models/db_node_model_ext.go

@@ -1 +1,17 @@
 package models
+
+import (
+	"fmt"
+	"github.com/iwind/TeaGo/dbs"
+)
+
+// 获取数据库配置
+func (this *DBNode) DBConfig() *dbs.DBConfig {
+	dsn := this.Username + ":" + this.Password + "@tcp(" + this.Host + ":" + fmt.Sprintf("%d", this.Port) + ")/" + this.Database + "?charset=utf8mb4&timeout=10s"
+
+	return &dbs.DBConfig{
+		Driver: "mysql",
+		Dsn:    dsn,
+		Prefix: "edge",
+	}
+}

internal/db/models/ip_list_dao.go

@@ -4,6 +4,7 @@ import (
 	"github.com/TeaOSLab/EdgeAPI/internal/errors"
 	"github.com/TeaOSLab/EdgeAPI/internal/utils/numberutils"
 	"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs/ipconfigs"
+	"github.com/TeaOSLab/EdgeCommon/pkg/systemconfigs"
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/iwind/TeaGo/Tea"
 	"github.com/iwind/TeaGo/dbs"
@@ -114,12 +115,12 @@ func (this *IPListDAO) UpdateIPList(tx *dbs.Tx, listId int64, name string, code
 
 // 增加版本
 func (this *IPListDAO) IncreaseVersion(tx *dbs.Tx) (int64, error) {
-	valueJSON, err := SharedSysSettingDAO.ReadSetting(tx, SettingCodeIPListVersion)
+	valueJSON, err := SharedSysSettingDAO.ReadSetting(tx, systemconfigs.SettingCodeIPListVersion)
 	if err != nil {
 		return 0, err
 	}
 	if len(valueJSON) == 0 {
-		err = SharedSysSettingDAO.UpdateSetting(tx, SettingCodeIPListVersion, []byte("1"))
+		err = SharedSysSettingDAO.UpdateSetting(tx, systemconfigs.SettingCodeIPListVersion, []byte("1"))
 		if err != nil {
 			return 0, err
 		}
@@ -127,7 +128,7 @@ func (this *IPListDAO) IncreaseVersion(tx *dbs.Tx) (int64, error) {
 	}
 
 	value := types.Int64(string(valueJSON)) + 1
-	err = SharedSysSettingDAO.UpdateSetting(tx, SettingCodeIPListVersion, []byte(numberutils.FormatInt64(value)))
+	err = SharedSysSettingDAO.UpdateSetting(tx, systemconfigs.SettingCodeIPListVersion, []byte(numberutils.FormatInt64(value)))
 	return value, nil
 }
 

internal/db/models/node_dao.go

@@ -8,6 +8,7 @@ import (
 	"github.com/TeaOSLab/EdgeCommon/pkg/configutils"
 	"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
 	"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
+	"github.com/TeaOSLab/EdgeCommon/pkg/systemconfigs"
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/iwind/TeaGo/Tea"
 	"github.com/iwind/TeaGo/dbs"
@@ -467,7 +468,7 @@ func (this *NodeDAO) ComposeNodeConfig(tx *dbs.Tx, nodeId int64) (*nodeconfigs.N
 
 	// 全局设置
 	// TODO 根据用户的不同读取不同的全局设置
-	settingJSON, err := SharedSysSettingDAO.ReadSetting(tx, SettingCodeServerGlobalConfig)
+	settingJSON, err := SharedSysSettingDAO.ReadSetting(tx, systemconfigs.SettingCodeServerGlobalConfig)
 	if err != nil {
 		return nil, err
 	}

internal/db/models/server_dao.go

@@ -9,6 +9,7 @@ import (
 	"github.com/TeaOSLab/EdgeCommon/pkg/configutils"
 	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
 	"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
+	"github.com/TeaOSLab/EdgeCommon/pkg/systemconfigs"
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/iwind/TeaGo/Tea"
 	"github.com/iwind/TeaGo/dbs"
@@ -295,7 +296,7 @@ func (this *ServerDAO) UpdateServerConfig(tx *dbs.Tx, serverId int64, configJSON
 		return false, err
 	}
 
-	globalConfig, err := SharedSysSettingDAO.ReadSetting(tx, SettingCodeServerGlobalConfig)
+	globalConfig, err := SharedSysSettingDAO.ReadSetting(tx, systemconfigs.SettingCodeServerGlobalConfig)
 	if err != nil {
 		return false, err
 	}

internal/db/models/sys_setting_dao.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
+	"github.com/TeaOSLab/EdgeCommon/pkg/systemconfigs"
 	_ "github.com/go-sql-driver/mysql"
 	"github.com/iwind/TeaGo/Tea"
 	"github.com/iwind/TeaGo/dbs"
@@ -12,16 +13,6 @@ import (
 
 type SysSettingDAO dbs.DAO
 
-type SettingCode = string
-
-const (
-	SettingCodeServerGlobalConfig  SettingCode = "serverGlobalConfig"  // 服务相关全局设置
-	SettingCodeNodeMonitor         SettingCode = "nodeMonitor"         // 监控节点状态
-	SettingCodeClusterHealthCheck  SettingCode = "clusterHealthCheck"  // 集群健康检查
-	SettingCodeIPListVersion       SettingCode = "ipListVersion"       // IP名单的版本号
-	SettingCodeAdminSecurityConfig SettingCode = "adminSecurityConfig" // 管理员安全设置
-)
-
 func NewSysSettingDAO() *SysSettingDAO {
 	return dbs.NewDAO(&SysSettingDAO{
 		DAOObject: dbs.DAOObject{
@@ -116,7 +107,7 @@ func (this *SysSettingDAO) CompareInt64Setting(tx *dbs.Tx, code string, anotherV
 
 // 读取全局配置
 func (this *SysSettingDAO) ReadGlobalConfig(tx *dbs.Tx) (*serverconfigs.GlobalConfig, error) {
-	globalConfigData, err := this.ReadSetting(tx, SettingCodeServerGlobalConfig)
+	globalConfigData, err := this.ReadSetting(tx, systemconfigs.SettingCodeServerGlobalConfig)
 	if err != nil {
 		return nil, err
 	}

internal/tasks/health_check_cluster_task.go

@@ -7,6 +7,7 @@ import (
 	"github.com/TeaOSLab/EdgeAPI/internal/utils"
 	"github.com/TeaOSLab/EdgeAPI/internal/utils/numberutils"
 	"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
+	"github.com/TeaOSLab/EdgeCommon/pkg/systemconfigs"
 	"github.com/iwind/TeaGo/logs"
 	"github.com/iwind/TeaGo/maps"
 	"time"
@@ -87,7 +88,7 @@ func (this *HealthCheckClusterTask) Stop() {
 // 单个循环任务
 func (this *HealthCheckClusterTask) loop(seconds int64) error {
 	// 检查上次运行时间，防止重复运行
-	settingKey := models.SettingCodeClusterHealthCheck + "Loop" + numberutils.FormatInt64(this.clusterId)
+	settingKey := systemconfigs.SettingCodeClusterHealthCheck + "Loop" + numberutils.FormatInt64(this.clusterId)
 	timestamp := time.Now().Unix()
 	c, err := models.SharedSysSettingDAO.CompareInt64Setting(nil, settingKey, timestamp-seconds)
 	if err != nil {

internal/tasks/node_monitor_task.go

@@ -3,6 +3,7 @@ package tasks
 import (
 	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
 	"github.com/TeaOSLab/EdgeAPI/internal/utils/numberutils"
+	"github.com/TeaOSLab/EdgeCommon/pkg/systemconfigs"
 	"github.com/iwind/TeaGo/dbs"
 	"github.com/iwind/TeaGo/logs"
 	"time"
@@ -40,7 +41,7 @@ func (this *NodeMonitorTask) Run() {
 
 func (this *NodeMonitorTask) loop() error {
 	// 检查上次运行时间，防止重复运行
-	settingKey := models.SettingCodeNodeMonitor + "Loop"
+	settingKey := systemconfigs.SettingCodeNodeMonitor + "Loop"
 	timestamp := time.Now().Unix()
 	c, err := models.SharedSysSettingDAO.CompareInt64Setting(nil, settingKey, timestamp-int64(this.intervalSeconds))
 	if err != nil {

internal/tasks/server_access_log_cleaner.go

@@ -0,0 +1,128 @@
+package tasks
+
+import (
+	"encoding/json"
+	"github.com/TeaOSLab/EdgeAPI/internal/db/models"
+	"github.com/TeaOSLab/EdgeAPI/internal/errors"
+	"github.com/TeaOSLab/EdgeCommon/pkg/systemconfigs"
+	"github.com/iwind/TeaGo/dbs"
+	"github.com/iwind/TeaGo/logs"
+	timeutil "github.com/iwind/TeaGo/utils/time"
+	"regexp"
+	"strings"
+	"time"
+)
+
+func init() {
+	dbs.OnReady(func() {
+		task := NewServerAccessLogCleaner()
+		go task.Start()
+	})
+}
+
+// 服务访问日志自动清理
+type ServerAccessLogCleaner struct {
+}
+
+func NewServerAccessLogCleaner() *ServerAccessLogCleaner {
+	return &ServerAccessLogCleaner{}
+}
+
+func (this *ServerAccessLogCleaner) Start() {
+	ticker := time.NewTicker(12 * time.Hour)
+	for range ticker.C {
+		err := this.Loop()
+		if err != nil {
+			logs.Println("[TASK][ServerAccessLogCleaner]Error: " + err.Error())
+		}
+	}
+}
+
+func (this *ServerAccessLogCleaner) Loop() error {
+	// 当前设置
+	configJSON, err := models.SharedSysSettingDAO.ReadSetting(nil, systemconfigs.SettingCodeDatabaseConfigSetting)
+	if err != nil {
+		return err
+	}
+	if len(configJSON) == 0 {
+		return nil
+	}
+	config := &systemconfigs.DatabaseConfig{}
+	err = json.Unmarshal(configJSON, config)
+	if err != nil {
+		return err
+	}
+	if config.ServerAccessLog.Clean.Days <= 0 {
+		return nil
+	}
+	days := config.ServerAccessLog.Clean.Days
+	endDay := timeutil.Format("Ymd", time.Now().AddDate(0, 0, -days+1))
+
+	// 当前连接的数据库
+	db, err := dbs.Default()
+	if err != nil {
+		return err
+	}
+	err = this.cleanDB(db, endDay)
+	if err != nil {
+		return err
+	}
+
+	// 日志数据库节点
+	nodes, err := models.SharedDBNodeDAO.FindAllEnabledAndOnDBNodes(nil)
+	if err != nil {
+		return err
+	}
+	for _, node := range nodes {
+		dbConfig := node.DBConfig()
+		db, err := dbs.NewInstanceFromConfig(dbConfig)
+		if err != nil {
+			return err
+		}
+		err = this.cleanDB(db, endDay)
+		if err != nil {
+			_ = db.Close()
+			return err
+		}
+
+		_ = db.Close()
+	}
+
+	return nil
+}
+
+func (this *ServerAccessLogCleaner) cleanDB(db *dbs.DB, endDay string) error {
+	ones, columnNames, err := db.FindOnes("SHOW TABLES")
+	if err != nil {
+		return err
+	}
+	if len(columnNames) != 1 {
+		return errors.New("invalid column names: " + strings.Join(columnNames, ", "))
+	}
+	columnName := columnNames[0]
+	for _, one := range ones {
+		tableName := one.GetString(columnName)
+		if len(tableName) == 0 {
+			continue
+		}
+		ok, err := regexp.MatchString(`^edgeHTTPAccessLogs_(\d{8})$`, tableName)
+		if err != nil {
+			return err
+		}
+		if !ok {
+			continue
+		}
+		index := strings.LastIndex(tableName, "_")
+		if index < 0 {
+			continue
+		}
+		day := tableName[index+1:]
+		if day < endDay {
+			_, err = db.Exec("DROP TABLE " + tableName)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}

internal/tasks/server_access_log_cleaner_test.go

@@ -0,0 +1,17 @@
+package tasks
+
+import (
+	"github.com/iwind/TeaGo/dbs"
+	"testing"
+)
+
+func TestServerAccessLogCleaner_Loop(t *testing.T) {
+	dbs.NotifyReady()
+
+	task := NewServerAccessLogCleaner()
+	err := task.Loop()
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Log("ok")
+}