feat: 备份网元Log文件到网管支持ftp转存

src/modules/crontask/processor/backup_export_log_ne/backup_export_log_ne.go

@@ -0,0 +1,140 @@
+package backup_export_log_ne
+
+import (
+	"fmt"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"be.ems/src/framework/cron"
+	"be.ems/src/framework/database/redis"
+	"be.ems/src/framework/logger"
+	"be.ems/src/framework/ssh"
+	neDataService "be.ems/src/modules/network_data/service"
+	neModel "be.ems/src/modules/network_element/model"
+	neService "be.ems/src/modules/network_element/service"
+)
+
+var NewProcessor = &BackupExportLogNEProcessor{
+	count:         0,
+	backupService: neDataService.NewBackup,
+	neInfoService: neService.NewNeInfo,
+	logFilePath:   "/var/log",
+}
+
+// BackupExportLogNEProcessor 备份导出日志任务处理
+type BackupExportLogNEProcessor struct {
+	count         int                   // 执行次数
+	backupService *neDataService.Backup // 备份相关服务
+	neInfoService *neService.NeInfo     // 网元信息服务
+	logFilePath   string                // 日志文件目录
+}
+
+func (s *BackupExportLogNEProcessor) Execute(data any) (any, error) {
+	s.count++ // 执行次数加一
+	options := data.(cron.JobData)
+	sysJob := options.SysJob
+	logger.Infof("重复:%v 任务ID:%d  执行次数:%d", options.Repeat, sysJob.JobId, s.count)
+	// 返回结果，用于记录执行结果
+	result := map[string]any{
+		"count": s.count,
+	}
+
+	// 分布式锁，防止多个任务同时执行
+	lockKey := fmt.Sprintf("processor:backup_export_log_ne:%d", sysJob.JobId)
+	if ok := redis.SetNX("", lockKey, time.Minute); !ok {
+		return nil, cron.ErrTaskRunning
+	}
+	defer redis.Del("", lockKey)
+
+	// 时间目录
+	timeDir := time.Now().Format("20060102150405")
+
+	neList := s.neInfoService.Find(neModel.NeInfo{}, false, false)
+	for _, neInfo := range neList {
+		if neInfo.IP == "" {
+			continue
+		}
+
+		// 网元主机的SSH客户端
+		sshClient, err := s.neInfoService.NeRunSSHClient(neInfo.NeType, neInfo.NeId)
+		if err != nil {
+			result[neInfo.NeName] = err.Error()
+			continue
+		}
+		defer sshClient.Close()
+		// 网元主机的SSH客户端进行文件传输
+		sftpClient, err := sshClient.NewClientSFTP()
+		if err != nil {
+			result[neInfo.NeName] = err.Error()
+			continue
+		}
+		defer sftpClient.Close()
+
+		// 备份导出日志
+		if neInfo.NeType == "IMS" {
+			imsDirArr := [...]string{"icscf", "bgcf", "mmtel", "pcscf", "scscf", "iwf", "bsf", "ismc"}
+			for _, dirPath := range imsDirArr {
+				logFilePathIMS := filepath.ToSlash(filepath.Join(s.logFilePath, "ims", dirPath))
+				output := s.backup(timeDir, logFilePathIMS, neInfo, sshClient, sftpClient)
+				for k, v := range output {
+					result[k+"-"+dirPath] = v
+				}
+			}
+		} else {
+			output := s.backup(timeDir, s.logFilePath, neInfo, sshClient, sftpClient)
+			for k, v := range output {
+				result[k] = v
+			}
+		}
+
+	}
+
+	// 返回结果，用于记录执行结果
+	return result, nil
+}
+
+// backup 备份导出日志
+func (s *BackupExportLogNEProcessor) backup(timeDir string, logFilePath string, neInfo neModel.NeInfo, sshClient *ssh.ConnSSH, sftpClient *ssh.SSHClientSFTP) map[string]string {
+	var result = map[string]string{}
+
+	// 获取文件列表
+	search := strings.ToLower(neInfo.NeType)
+	if neInfo.NeType == "IMS" {
+		search = ""
+	}
+	rows, err := ssh.FileList(sshClient, logFilePath, search)
+	if err != nil {
+		result[neInfo.NeName] = err.Error()
+		return result
+	}
+
+	// 计算今天00:00:00的时间戳（秒）
+	today := time.Now()
+	todayStart := time.Date(today.Year(), today.Month(), today.Day(), 0, 0, 0, 0, today.Location())
+	todayStartTimestamp := todayStart.Unix()
+	for _, row := range rows {
+		if row.ModifiedTime < todayStartTimestamp || row.FileType != "file" {
+			continue
+		}
+		// 网元日志文件路径
+		neFilePath := fmt.Sprintf("%s/%s", logFilePath, row.FileName)
+		// 复制到备份目录
+		neDirPath := fmt.Sprintf("%s/%s", search, row.FileName)
+		if neInfo.NeType == "IMS" {
+			neDirPath = strings.Replace(neFilePath, s.logFilePath, "", 1)
+		}
+		localFilePath := filepath.Join(s.backupService.BACKUP_DIR, "/log/ne_log", timeDir, neDirPath)
+		if err := sftpClient.CopyFileRemoteToLocal(neFilePath, localFilePath); err != nil {
+			result[neInfo.NeName+"-copy"] = err.Error()
+			continue
+		}
+		// 上传到FTP服务器
+		if err := s.backupService.FTPPushFile(localFilePath, "log"); err != nil {
+			result[neInfo.NeName+"-copy-ftp"] = "ok, ftp err:" + err.Error()
+			continue
+		}
+	}
+
+	return result
+}

src/modules/crontask/processor/processor.go

@@ -5,6 +5,7 @@ import (
 	processorBackupExportCDR "be.ems/src/modules/crontask/processor/backup_export_cdr"
 	processorBackupExportKPI "be.ems/src/modules/crontask/processor/backup_export_kpi"
 	processorBackupExportLog "be.ems/src/modules/crontask/processor/backup_export_log"
+	processorBackupExportLogNE "be.ems/src/modules/crontask/processor/backup_export_log_ne"
 	processorBackupExportTable "be.ems/src/modules/crontask/processor/backup_export_table"
 	processorBackupExportUDM "be.ems/src/modules/crontask/processor/backup_export_udm"
 	processorBackupRemoveFile "be.ems/src/modules/crontask/processor/backup_remove_file"
@@ -59,8 +60,10 @@ func InitCronQueue() {
 	cron.CreateQueue("backup_export_udm", processorBackupExportUDM.NewProcessor)
 	// 备份-导出CDR数据
 	cron.CreateQueue("backup_export_cdr", processorBackupExportCDR.NewProcessor)
-	// 备份-导出Log数据
+	// 备份-导出系统Log数据
 	cron.CreateQueue("backup_export_log", processorBackupExportLog.NewProcessor)
 	// 备份-导出KPI数据
 	cron.CreateQueue("backup_export_kpi", processorBackupExportKPI.NewProcessor)
+	// 备份-导出网元Log数据
+	cron.CreateQueue("backup_export_log_ne", processorBackupExportLogNE.NewProcessor)
 }