mirror of
https://github.com/StarFleetCPTN/GoMFT.git
synced 2026-09-08 15:41:20 +02:00
Enhance logging in scheduler for better debugging
- Added debug logging for job scheduling, execution, and configuration processing. - Included detailed log messages for cron expression validation and file transfer commands. - Improved visibility into job history and webhook notifications with additional debug information.
This commit is contained in:
@@ -173,6 +173,11 @@ func NewLogger() *Logger {
|
||||
filepath.Join(logsDir, "scheduler.log"), maxSize, maxBackups, maxAge, compress, logLevel.String())
|
||||
}
|
||||
|
||||
if logLevel >= LogLevelDebug {
|
||||
logger.Debug.Printf("Log rotation details: file=%s, maxSize=%dMB, maxBackups=%d, maxAge=%d days, compress=%v",
|
||||
filepath.Join(logsDir, "scheduler.log"), maxSize, maxBackups, maxAge, compress)
|
||||
}
|
||||
|
||||
return logger
|
||||
}
|
||||
|
||||
@@ -258,6 +263,8 @@ func (s *Scheduler) loadJobs() {
|
||||
}
|
||||
|
||||
func (s *Scheduler) ScheduleJob(job *db.Job) error {
|
||||
s.log.LogDebug("Attempting to schedule job ID %d: %+v", job.ID, job)
|
||||
|
||||
s.log.LogInfo("Scheduling job %d: %s with schedule %s", job.ID, job.Name, job.Schedule)
|
||||
|
||||
// Remove existing job if it exists
|
||||
@@ -279,6 +286,8 @@ func (s *Scheduler) ScheduleJob(job *db.Job) error {
|
||||
schedule = "0 " + schedule
|
||||
}
|
||||
|
||||
s.log.LogDebug("Converted schedule from '%s' to '%s'", job.Schedule, schedule)
|
||||
|
||||
// Validate cron expression
|
||||
parser := cron.NewParser(cron.Second | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow)
|
||||
_, err := parser.Parse(schedule)
|
||||
@@ -286,6 +295,8 @@ func (s *Scheduler) ScheduleJob(job *db.Job) error {
|
||||
return fmt.Errorf("invalid cron expression '%s': %w", job.Schedule, err)
|
||||
}
|
||||
|
||||
s.log.LogDebug("Validated cron expression '%s' for job %d", schedule, job.ID)
|
||||
|
||||
// Schedule the job
|
||||
entryID, err := s.cron.AddFunc(job.Schedule, func() {
|
||||
s.executeJob(job.ID)
|
||||
@@ -296,6 +307,8 @@ func (s *Scheduler) ScheduleJob(job *db.Job) error {
|
||||
return err
|
||||
}
|
||||
|
||||
s.log.LogDebug("Scheduled job %d with cron entry ID %d", job.ID, entryID)
|
||||
|
||||
// Store mapping of job ID to cron entry ID
|
||||
s.jobMutex.Lock()
|
||||
s.jobs[job.ID] = entryID
|
||||
@@ -313,6 +326,9 @@ func (s *Scheduler) ScheduleJob(job *db.Job) error {
|
||||
}
|
||||
|
||||
func (s *Scheduler) executeJob(jobID uint) {
|
||||
s.log.LogDebug("Entering executeJob for job ID %d", jobID)
|
||||
defer s.log.LogDebug("Exiting executeJob for job ID %d", jobID)
|
||||
|
||||
s.log.LogInfo("Starting execution of job %d", jobID)
|
||||
|
||||
// Get job details
|
||||
@@ -322,6 +338,8 @@ func (s *Scheduler) executeJob(jobID uint) {
|
||||
return
|
||||
}
|
||||
|
||||
s.log.LogDebug("Loaded job details: %+v", job)
|
||||
|
||||
// Get all configurations associated with this job
|
||||
configs, err := s.db.GetConfigsForJob(jobID)
|
||||
if err != nil {
|
||||
@@ -329,6 +347,8 @@ func (s *Scheduler) executeJob(jobID uint) {
|
||||
return
|
||||
}
|
||||
|
||||
s.log.LogDebug("Processing %d configurations: %+v", len(configs), configs)
|
||||
|
||||
if len(configs) == 0 {
|
||||
s.log.LogError("Error: job %d has no associated configurations", jobID)
|
||||
return
|
||||
@@ -366,6 +386,8 @@ func (s *Scheduler) executeJob(jobID uint) {
|
||||
|
||||
// processConfiguration processes a single configuration for a job
|
||||
func (s *Scheduler) processConfiguration(job *db.Job, config *db.TransferConfig, index int, totalConfigs int) {
|
||||
s.log.LogDebug("Processing configuration %d: %+v", config.ID, config)
|
||||
|
||||
s.log.LogInfo("Processing configuration %d (%d/%d) for job %d: source=%s:%s, dest=%s:%s",
|
||||
config.ID,
|
||||
index,
|
||||
@@ -392,12 +414,16 @@ func (s *Scheduler) processConfiguration(job *db.Job, config *db.TransferConfig,
|
||||
return
|
||||
}
|
||||
|
||||
s.log.LogDebug("Creating job history record: %+v", history)
|
||||
|
||||
// Execute the configuration transfer
|
||||
s.executeConfigTransfer(*job, *config, history)
|
||||
}
|
||||
|
||||
// executeConfigTransfer performs the actual file transfer for a single configuration
|
||||
func (s *Scheduler) executeConfigTransfer(job db.Job, config db.TransferConfig, history *db.JobHistory) {
|
||||
s.log.LogDebug("Starting transfer for config %d with params: %+v", config.ID, config)
|
||||
|
||||
// Track files already processed in this job execution to prevent duplicates
|
||||
processedFiles := make(map[string]bool)
|
||||
|
||||
@@ -447,7 +473,7 @@ func (s *Scheduler) executeConfigTransfer(job db.Job, config db.TransferConfig,
|
||||
listArgs = append(listArgs, sourceListPath)
|
||||
|
||||
// Execute lsjson command
|
||||
s.log.LogInfo("Listing files with metadata for job %d, config %d: rclone %s", job.ID, config.ID, strings.Join(listArgs, " "))
|
||||
s.log.LogDebug("Full lsjson command: %s %v", os.Getenv("RCLONE_PATH"), listArgs)
|
||||
rclonePath := os.Getenv("RCLONE_PATH")
|
||||
if rclonePath == "" {
|
||||
rclonePath = "rclone"
|
||||
@@ -455,6 +481,19 @@ func (s *Scheduler) executeConfigTransfer(job db.Job, config db.TransferConfig,
|
||||
listCmd := exec.Command(rclonePath, listArgs...)
|
||||
listOutput, listErr := listCmd.CombinedOutput()
|
||||
|
||||
// Add debug logging of raw output
|
||||
if listErr == nil {
|
||||
s.log.LogDebug("Raw lsjson output for job %d config %d:\n%s",
|
||||
job.ID,
|
||||
config.ID,
|
||||
string(listOutput))
|
||||
} else {
|
||||
s.log.LogDebug("Raw lsjson output (error case) for job %d config %d:\n%s",
|
||||
job.ID,
|
||||
config.ID,
|
||||
string(listOutput))
|
||||
}
|
||||
|
||||
if listErr != nil {
|
||||
s.log.LogError("Error listing files for job %d, config %d: %v", job.ID, config.ID, listErr)
|
||||
// s.log.Debug.Printf("Output: %s", string(listOutput))
|
||||
@@ -550,7 +589,7 @@ func (s *Scheduler) executeConfigTransfer(job db.Job, config db.TransferConfig,
|
||||
concurrencySemaphore := make(chan struct{}, maxConcurrent)
|
||||
|
||||
// Process each file individually
|
||||
for _, fileEntry := range files {
|
||||
for i, fileEntry := range files {
|
||||
fileName, ok := fileEntry["Path"].(string)
|
||||
if !ok || fileName == "" {
|
||||
continue
|
||||
@@ -666,7 +705,8 @@ func (s *Scheduler) executeConfigTransfer(job db.Job, config db.TransferConfig,
|
||||
currentModTime := modTime
|
||||
|
||||
// Log the file information that will be processed
|
||||
s.log.LogDebug("Processing file: %s, size: %d, hash: %s", currentFileName, currentFileSize, currentFileHash)
|
||||
s.log.LogDebug("Processing file %d/%d: %s (Size: %d, Hash: %s)",
|
||||
i+1, len(files), currentFileName, currentFileSize, currentFileHash)
|
||||
|
||||
// Start goroutine for concurrent processing
|
||||
go func() {
|
||||
@@ -740,13 +780,8 @@ func (s *Scheduler) executeConfigTransfer(job db.Job, config db.TransferConfig,
|
||||
transferArgs = append(transferArgs, sourcePath, destPath)
|
||||
|
||||
// Execute transfer for this file
|
||||
s.log.LogInfo("Executing rclone transfer command for job %d, config %d, file %s: rclone %s",
|
||||
job.ID, config.ID, currentFileName, strings.Join(transferArgs, " "))
|
||||
// Get the rclone path from the environment variable or use the default path
|
||||
rclonePath := os.Getenv("RCLONE_PATH")
|
||||
if rclonePath == "" {
|
||||
rclonePath = "rclone"
|
||||
}
|
||||
s.log.LogDebug("Full transfer command: %s %v", rclonePath, transferArgs)
|
||||
s.log.LogDebug("Environment: RCLONE_PATH=%s", os.Getenv("RCLONE_PATH"))
|
||||
cmd := exec.Command(rclonePath, transferArgs...)
|
||||
fileOutput, fileErr := cmd.CombinedOutput()
|
||||
|
||||
@@ -1087,6 +1122,8 @@ func (s *Scheduler) sendWebhookNotification(job *db.Job, history *db.JobHistory,
|
||||
return
|
||||
}
|
||||
|
||||
s.log.LogDebug("Webhook payload: %s", string(jsonPayload))
|
||||
|
||||
// Create HTTP request
|
||||
req, err := http.NewRequest("POST", job.WebhookURL, bytes.NewBuffer(jsonPayload))
|
||||
if err != nil {
|
||||
@@ -1116,6 +1153,8 @@ func (s *Scheduler) sendWebhookNotification(job *db.Job, history *db.JobHistory,
|
||||
}
|
||||
}
|
||||
|
||||
s.log.LogDebug("Webhook headers: %+v", req.Header)
|
||||
|
||||
// Send the request with a timeout
|
||||
client := &http.Client{
|
||||
Timeout: 10 * time.Second,
|
||||
|
||||
Reference in New Issue
Block a user