Close all open files, clean up fsnotify watcher (#1268)

Author: matoval

pkg/netceptor/conn.go

@@ -151,6 +151,12 @@ func (s *Netceptor) tracer(ctx context.Context, p logging.Perspective, connID qu
 
 			return nil
 		}
+		defer func() {
+			err := f.Close()
+			if err != nil {
+				s.GetLogger().Error("Error closing %s: %s", qlogPath+filename, err)
+			}
+		}()
 
 		return qlog.NewConnectionTracer(f, p, connID)
 	} else {

pkg/workceptor/command.go

@@ -209,6 +209,14 @@ loop:
 			MainInstance.nc.GetLogger().Error("Error updating status file %s: %s", statusFilename, err)
 		}
 	}
+	err = stdin.Close()
+	if err != nil {
+		MainInstance.nc.GetLogger().Error("Error closing %s: %s", path.Join(unitdir, "stdin"), err)
+	}
+	err = stdout.Close()
+	if err != nil {
+		MainInstance.nc.GetLogger().Error("Error closing %s: %s", path.Join(unitdir, "stdout"), err)
+	}
 	os.Exit(cmd.ProcessState.ExitCode())
 
 	return nil

pkg/workceptor/mock_workceptor/workunitbase.go

@@ -198,6 +198,12 @@ func (rw *remoteUnit) startRemoteUnit(ctx context.Context, conn net.Conn, reader
 	if err != nil {
 		return fmt.Errorf("error opening stdin file: %s", err)
 	}
+	defer func() {
+		err := stdin.Close()
+		if err != nil {
+			MainInstance.nc.GetLogger().Error("Error closing %s: %s", path.Join(rw.UnitDir(), "stdin"), err)
+		}
+	}()
 	_, err = io.Copy(conn, stdin)
 	if err != nil {
 		return fmt.Errorf("error sending stdin file: %s", err)
@@ -467,6 +473,12 @@ func (rw *remoteUnit) monitorRemoteStdout(mw *utils.JobContext) {
 
 				return
 			}
+			defer func() {
+				err := stdout.Close()
+				if err != nil {
+					MainInstance.nc.GetLogger().Error("Error closing %s: %s", rw.StdoutFileName(), err)
+				}
+			}()
 			doneChan := make(chan struct{})
 			go func() {
 				select {

pkg/workceptor/remote_work.go

@@ -36,6 +36,7 @@ const (
 // WatcherWrapper is wrapping the fsnofity Watcher struct and exposing the Event chan within.
 type WatcherWrapper interface {
 	Add(name string) error
+	Remove(path string) error
 	Close() error
 	ErrorChannel() chan error
 	EventChannel() chan fsnotify.Event
@@ -49,6 +50,10 @@ func (rw *RealWatcher) Add(name string) error {
 	return rw.watcher.Add(name)
 }
 
+func (rw *RealWatcher) Remove(path string) error {
+	return rw.watcher.Remove(path)
+}
+
 func (rw *RealWatcher) Close() error {
 	return rw.watcher.Close()
 }
@@ -407,9 +412,10 @@ func (bwu *BaseWorkUnit) MonitorLocalStatus() {
 		bwu.statusLock.Unlock()
 		if err == nil {
 			defer func() {
-				werr := bwu.watcher.Close()
-				if werr != nil {
-					bwu.w.nc.GetLogger().Error("Error in defer closing %s: %s", statusFile, err)
+				bwu.watcher.Remove(statusFile)
+				err = bwu.watcher.Close()
+				if err != nil {
+					bwu.w.nc.GetLogger().Error("Error closing watcher: %v", err)
 				}
 			}()
 			watcherEvents = bwu.watcher.EventChannel()

pkg/workceptor/workunitbase.go

@@ -389,6 +389,7 @@ func TestMonitorLocalStatus(t *testing.T) {
 				mockFileSystem.EXPECT().Stat(gomock.Any()).Return(tc.statObjLater, nil).AnyTimes()
 			}
 			mockWatcher.EXPECT().Add(gomock.Any()).Return(tc.addWatcherErr)
+			mockWatcher.EXPECT().Remove(gomock.Any()).AnyTimes()
 			mockWatcher.EXPECT().Close().AnyTimes()
 
 			if tc.fsNotifyEvent != nil {