Improvements to the way delete_after_read works (#432)

- Wait until end of poll cycle and then delete files from the primary
  goroutine
- Update checkpoints, even though in thoery files are consumed fully
- If file.Close fails, log at Warn level instead of Debug

operator/builtin/input/file/file.go

@@ -162,29 +162,27 @@ OUTER:
 		go func(r *Reader) {
 			defer wg.Done()
 			r.ReadToEnd(ctx)
-			if f.deleteAfterRead {
-				r.Close()
-				if err := os.Remove(r.file.Name()); err != nil {
-					f.Errorf("could not delete %s", r.file.Name())
-				}
-			}
 		}(reader)
 	}
 
 	// Wait until all the reader goroutines are finished
 	wg.Wait()
 
 	if f.deleteAfterRead {
-		// No need to track files, since we only consume them once
-		return
-	}
-
-	for _, reader := range f.lastPollReaders {
-		reader.Close()
+		f.Debug("cleaning up log files that have been consumed")
+		for _, reader := range readers {
+			reader.Close()
+			if err := os.Remove(reader.file.Name()); err != nil {
+				f.Errorf("could not delete %s", reader.file.Name())
+			}
+		}
+	} else {
+		for _, reader := range f.lastPollReaders {
+			reader.Close()
+		}
+		f.lastPollReaders = readers
 	}
 
-	f.lastPollReaders = readers
-
 	f.saveCurrent(readers)
 	f.syncLastPollFiles()
 }

operator/builtin/input/file/reader.go

@@ -174,7 +174,7 @@ func (f *Reader) readHeaders(ctx context.Context, msgBuf []byte) error {
 func (f *Reader) Close() {
 	if f.file != nil {
 		if err := f.file.Close(); err != nil {
-			f.Debugf("Problem closing reader", "Error", err.Error())
+			f.Warnf("Problem closing reader", "Error", err.Error())
 		}
 	}
 }