Clean up

Signed-off-by: chiragkyal <ckyal@redhat.com>

Author: chiragkyal

internal/ansible/runner/eventapi/fileapi.go

@@ -90,7 +90,7 @@ func (f *FileEventReceiver) watchJobEvents() {
 	defer f.wg.Done()
 	defer close(f.Events)
 
-	// Watch the job_events directory, not individual files
+	// Watch the job_events directory
 	// Ansible-runner writes artifacts to {inputDir}/artifacts/{ident}/job_events
 	jobEventsDir := filepath.Join(f.ArtifactPath, "artifacts", f.ident, "job_events")
 
@@ -102,9 +102,6 @@ func (f *FileEventReceiver) watchJobEvents() {
 
 	f.logger.Info("Starting file-based event receiver", "jobEventsDir", jobEventsDir)
 
-	// Process any existing files first
-	// f.processExistingFiles(jobEventsDir)
-
 	// Watch for new files
 	watcher, err := fsnotify.NewWatcher()
 	if err != nil {
@@ -118,10 +115,6 @@ func (f *FileEventReceiver) watchJobEvents() {
 		return
 	}
 
-	// // Ticker to periodically check for new files (in case we miss file events)
-	// ticker := time.NewTicker(1 * time.Second)
-	// defer ticker.Stop()
-
 	for {
 		select {
 		case event, ok := <-watcher.Events:
@@ -140,63 +133,20 @@ func (f *FileEventReceiver) watchJobEvents() {
 				return
 			}
 			f.errChan <- fmt.Errorf("file watcher error: %v", err)
-		// case <-ticker.C:
-		// 	// Periodically scan for any missed files
-		// 	f.processExistingFiles(jobEventsDir)
 		case <-f.ctx.Done():
-			// When context is cancelled, do a final scan for any remaining files
-			f.logger.V(1).Info("Context cancelled, doing final scan for events")
-			// f.processExistingFiles(jobEventsDir)
+			f.logger.V(1).Info("Context cancelled")
 			return
 		}
 	}
 }
 
-// processExistingFiles processes any existing event files in the directory
-// func (f *FileEventReceiver) processExistingFiles(jobEventsDir string) {
-// 	files, err := os.ReadDir(jobEventsDir)
-// 	if err != nil {
-// 		f.logger.Info("Could not read job_events directory", "dir", jobEventsDir, "error", err)
-// 		return
-// 	}
-
-// 	f.logger.Info("Scanning for existing event files", "dir", jobEventsDir, "fileCount", len(files))
-
-// 	// Sort files by name to process in order (ansible-runner prefixes with sequence number)
-// 	var jsonFiles []string
-// 	for _, file := range files {
-// 		if !file.IsDir() && filepath.Ext(file.Name()) == ".json" {
-// 			fullPath := filepath.Join(jobEventsDir, file.Name())
-
-// 			f.processMutex.Lock()
-// 			alreadyProcessed := f.processedFiles[fullPath]
-// 			f.processMutex.Unlock()
-
-// 			if !alreadyProcessed {
-// 				jsonFiles = append(jsonFiles, fullPath)
-// 			}
-// 		}
-// 	}
-
-// 	f.logger.Info("Found event files to process", "count", len(jsonFiles))
-
-// 	// Sort files to process in order
-// 	sort.Strings(jsonFiles)
-
-// 	// Process in order
-// 	for _, filePath := range jsonFiles {
-// 		f.logger.V(2).Info("Processing event file", "file", filePath)
-// 		f.processEventFile(filePath)
-// 	}
-// }
-
 // processEventFile reads and parses a single event file
 func (r *FileEventReceiver) processEventFile(filename string) bool {
 	// Check if already processed
 	r.processMutex.Lock()
 	if r.processedFiles[filename] {
 		r.processMutex.Unlock()
-		r.logger.Info("already processed file", "file", filename)
+		r.logger.Info("Already processed file", "file", filename)
 		return true
 	}
 	r.processMutex.Unlock()
@@ -248,7 +198,7 @@ func (r *FileEventReceiver) processEventFile(filename string) bool {
 		return true
 	case <-timeout.C:
 		r.logger.Info("Timed out writing event to channel")
-		return true // Consider it processed to avoid retry loops
+		return true
 	case <-r.ctx.Done():
 		r.logger.V(1).Info("Context cancelled while writing event")
 		return false

internal/ansible/runner/runner.go

@@ -352,14 +352,14 @@ func (r *runner) Run(ident string, u *unstructured.Unstructured, kubeconfig stri
 
 		// For file-based API, give some time to process final event files
 		if _, isFileReceiver := receiver.(*fileReceiverWrapper); isFileReceiver {
-			logger.Info("Waiting for file-based event receiver to process final events")
+			logger.V(1).Info("Waiting for file-based event receiver to process final events")
 
 			// Check if artifacts directory exists and list contents
 			artifactsDir := filepath.Join(inputDir.Path, "artifacts", ident)
 			if entries, err := os.ReadDir(artifactsDir); err == nil {
-				logger.Info("Artifacts directory contents", "dir", artifactsDir, "entries", len(entries))
+				logger.V(1).Info("Artifacts directory contents", "dir", artifactsDir, "entries", len(entries))
 				for _, entry := range entries {
-					logger.Info("Artifact entry", "name", entry.Name(), "isDir", entry.IsDir())
+					logger.V(1).Info("Artifact entry", "name", entry.Name(), "isDir", entry.IsDir())
 				}
 			} else {
 				logger.Info("Could not read artifacts directory", "dir", artifactsDir, "error", err)
@@ -368,15 +368,15 @@ func (r *runner) Run(ident string, u *unstructured.Unstructured, kubeconfig stri
 			// Check job_events directory specifically
 			jobEventsDir := filepath.Join(inputDir.Path, "artifacts", ident, "job_events")
 			if entries, err := os.ReadDir(jobEventsDir); err == nil {
-				logger.Info("Job events directory contents", "dir", jobEventsDir, "entries", len(entries))
+				logger.V(1).Info("Job events directory contents", "dir", jobEventsDir, "entries", len(entries))
 				for _, entry := range entries {
-					logger.Info("Job event file", "name", entry.Name())
+					logger.V(1).Info("Job event file", "name", entry.Name())
 				}
 			} else {
 				logger.Info("Could not read job_events directory", "dir", jobEventsDir, "error", err)
 			}
 
-			time.Sleep(5 * time.Second) // Increased delay for final file processing
+			time.Sleep(10 * time.Second) // Increased delay for final file processing
 		}
 
 		receiver.Close()