diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index 61da064..0ecc809 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -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,