diff --git a/pkg/workceptor/mock_workceptor/workunitbase.go b/pkg/workceptor/mock_workceptor/workunitbase.go index 5468ee85f..815a6d88d 100644 --- a/pkg/workceptor/mock_workceptor/workunitbase.go +++ b/pkg/workceptor/mock_workceptor/workunitbase.go @@ -68,6 +68,20 @@ func (mr *MockWatcherWrapperMockRecorder) Close() *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockWatcherWrapper)(nil).Close)) } +// ErrorChannel mocks base method. +func (m *MockWatcherWrapper) ErrorChannel() chan error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ErrorChannel") + ret0, _ := ret[0].(chan error) + return ret0 +} + +// ErrorChannel indicates an expected call of ErrorChannel. +func (mr *MockWatcherWrapperMockRecorder) ErrorChannel() *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ErrorChannel", reflect.TypeOf((*MockWatcherWrapper)(nil).ErrorChannel)) +} + // EventChannel mocks base method. func (m *MockWatcherWrapper) EventChannel() chan fsnotify.Event { m.ctrl.T.Helper() diff --git a/pkg/workceptor/workunitbase.go b/pkg/workceptor/workunitbase.go index 47aa21c34..f79ef2c36 100644 --- a/pkg/workceptor/workunitbase.go +++ b/pkg/workceptor/workunitbase.go @@ -37,6 +37,7 @@ const ( type WatcherWrapper interface { Add(name string) error Close() error + ErrorChannel() chan error EventChannel() chan fsnotify.Event } @@ -52,6 +53,10 @@ func (rw *RealWatcher) Close() error { return rw.watcher.Close() } +func (rw *RealWatcher) ErrorChannel() chan error { + return rw.watcher.Errors +} + func (rw *RealWatcher) EventChannel() chan fsnotify.Event { return rw.watcher.Events } @@ -126,6 +131,7 @@ func (bwu *BaseWorkUnit) Init(w *Workceptor, unitID string, workType string, fs if err == nil { bwu.watcher = &RealWatcher{watcher: watcher} } else { + bwu.w.nc.GetLogger().Info("fsnotify.NewWatcher returned %s", err) bwu.watcher = nil } } @@ -392,16 +398,20 @@ func (bwu *BaseWorkUnit) MonitorLocalStatus() { var watcherEvents chan fsnotify.Event watcherEvents = make(chan fsnotify.Event) + var watcherErrors chan error + watcherErrors = make(chan error) + if bwu.watcher != nil { err := bwu.watcher.Add(statusFile) if err == nil { defer func() { werr := bwu.watcher.Close() if werr != nil { - bwu.w.nc.GetLogger().Error("Error closing %s: %s", statusFile, err) + bwu.w.nc.GetLogger().Error("Error in defer closing %s: %s", statusFile, err) } }() watcherEvents = bwu.watcher.EventChannel() + watcherErrors = bwu.watcher.ErrorChannel() } else { werr := bwu.watcher.Close() if werr != nil { @@ -412,6 +422,7 @@ func (bwu *BaseWorkUnit) MonitorLocalStatus() { } fi, err := bwu.fs.Stat(statusFile) if err != nil { + bwu.w.nc.GetLogger().Error("Error retrieving stat for %s: %s", statusFile, err) fi = nil } @@ -424,7 +435,7 @@ loop: if event.Op&fsnotify.Write == fsnotify.Write { err = bwu.Load() if err != nil { - bwu.w.nc.GetLogger().Error("Error reading %s: %s", statusFile, err) + bwu.w.nc.GetLogger().Error("Watcher Events Error reading %s: %s", statusFile, err) } } case <-time.After(time.Second): @@ -433,9 +444,14 @@ loop: fi = newFi err = bwu.Load() if err != nil { - bwu.w.nc.GetLogger().Error("Error reading %s: %s", statusFile, err) + bwu.w.nc.GetLogger().Error("Work unit load Error reading %s: %s", statusFile, err) } } + case err, ok := <-watcherErrors: + if !ok { + return + } + bwu.w.nc.GetLogger().Error("fsnotify Error reading %s: %s", statusFile, err) } complete := IsComplete(bwu.Status().State) if complete { diff --git a/pkg/workceptor/workunitbase_test.go b/pkg/workceptor/workunitbase_test.go index 19ff3bae9..b21c7dfb7 100644 --- a/pkg/workceptor/workunitbase_test.go +++ b/pkg/workceptor/workunitbase_test.go @@ -365,6 +365,10 @@ func TestMonitorLocalStatus(t *testing.T) { eventCh := make(chan fsnotify.Event, 1) mockWatcher.EXPECT().EventChannel().Return(eventCh).AnyTimes() go func() { eventCh <- *tc.fsNotifyEvent }() + + errorCh := make(chan error, 1) + mockWatcher.EXPECT().ErrorChannel().Return(errorCh).AnyTimes() + // go func() { errorCh <- nil }() } go bwu.MonitorLocalStatus()