implement review changes

filebeat/prospector/prospector_log.go

@@ -96,8 +96,12 @@ func (l *Log) Run() {
 			// os.Stat will return an error in case the file does not exist
 			stat, err := os.Stat(state.Source)
 			if err != nil {
-				l.removeState(state)
-				logp.Debug("prospector", "Remove state for file as file removed: %s", state.Source)
+				if os.IsNotExist(err) {
+					l.removeState(state)
+					logp.Debug("prospector", "Remove state for file as file removed: %s", state.Source)
+				} else {
+					logp.Err("Prospector state for %s was not removed: %s", state.Source, err)
+				}
 			} else {
 				// Check if existing source on disk and state are the same. Remove if not the case.
 				newState := file.NewState(stat, state.Source)
@@ -111,16 +115,19 @@ func (l *Log) Run() {
 }
 
 func (l *Log) removeState(state file.State) {
+
 	// Only clean up files where state is Finished
-	if state.Finished {
-		state.TTL = 0
-		err := l.Prospector.updateState(input.NewEvent(state))
-		if err != nil {
-			logp.Err("File cleanup state update error: %s", err)
-		}
-	} else {
+	if !state.Finished {
 		logp.Debug("prospector", "State for file not removed because harvester not finished: %s", state.Source)
+		return
 	}
+
+	state.TTL = 0
+	err := l.Prospector.updateState(input.NewEvent(state))
+	if err != nil {
+		logp.Err("File cleanup state update error: %s", err)
+	}
+
 }
 
 // getFiles returns all files which have to be harvested