Merge branch 'upgrade-sarama-v1.38.1' into upgrade-sarama-v1.38.1

Author: axfor

cmd/query/app/mocks/Watcher.go

@@ -27,7 +27,6 @@ import (
 	"strings"
 	"sync/atomic"
 
-	"github.com/fsnotify/fsnotify"
 	"github.com/gorilla/mux"
 	"go.uber.org/zap"
 
@@ -61,10 +60,10 @@ func RegisterStaticHandler(r *mux.Router, logger *zap.Logger, qOpts *QueryOption
 
 // StaticAssetsHandler handles static assets
 type StaticAssetsHandler struct {
-	options    StaticAssetsHandlerOptions
-	indexHTML  atomic.Value // stores []byte
-	assetsFS   http.FileSystem
-	newWatcher func() (fswatcher.Watcher, error)
+	options   StaticAssetsHandlerOptions
+	indexHTML atomic.Value // stores []byte
+	assetsFS  http.FileSystem
+	watcher   *fswatcher.FSWatcher
 }
 
 // StaticAssetsHandlerOptions defines options for NewStaticAssetsHandler
@@ -73,7 +72,6 @@ type StaticAssetsHandlerOptions struct {
 	UIConfigPath string
 	LogAccess    bool
 	Logger       *zap.Logger
-	NewWatcher   func() (fswatcher.Watcher, error)
 }
 
 type loadedConfig struct {
@@ -92,23 +90,23 @@ func NewStaticAssetsHandler(staticAssetsRoot string, options StaticAssetsHandler
 		options.Logger = zap.NewNop()
 	}
 
-	if options.NewWatcher == nil {
-		options.NewWatcher = fswatcher.NewWatcher
-	}
-
 	indexHTML, err := loadAndEnrichIndexHTML(assetsFS.Open, options)
 	if err != nil {
 		return nil, err
 	}
 
 	h := &StaticAssetsHandler{
-		options:    options,
-		assetsFS:   assetsFS,
-		newWatcher: options.NewWatcher,
+		options:  options,
+		assetsFS: assetsFS,
 	}
 
+	watcher, err := fswatcher.New([]string{options.UIConfigPath}, h.reloadUIConfig, h.options.Logger)
+	if err != nil {
+		return nil, err
+	}
+	h.watcher = watcher
+
 	h.indexHTML.Store(indexHTML)
-	h.watch()
 
 	return h, nil
 }
@@ -142,67 +140,14 @@ func loadAndEnrichIndexHTML(open func(string) (http.File, error), options Static
 	return indexBytes, nil
 }
 
-func (sH *StaticAssetsHandler) configListener(watcher fswatcher.Watcher) {
-	for {
-		select {
-		case event := <-watcher.Events():
-			// ignore if the event filename is not the UI configuration
-			if filepath.Base(event.Name) != filepath.Base(sH.options.UIConfigPath) {
-				continue
-			}
-			// ignore if the event is a chmod event (permission or owner changes)
-			if event.Op&fsnotify.Chmod == fsnotify.Chmod {
-				continue
-			}
-			if event.Op&fsnotify.Remove == fsnotify.Remove {
-				sH.options.Logger.Warn("the UI config file has been removed, using the last known version")
-				continue
-			}
-			// this will catch events for all files inside the same directory, which is OK if we don't have many changes
-			sH.options.Logger.Info("reloading UI config", zap.String("filename", sH.options.UIConfigPath))
-			content, err := loadAndEnrichIndexHTML(sH.assetsFS.Open, sH.options)
-			if err != nil {
-				sH.options.Logger.Error("error while reloading the UI config", zap.Error(err))
-			}
-			sH.indexHTML.Store(content)
-			sH.options.Logger.Info("reloaded UI config", zap.String("filename", sH.options.UIConfigPath))
-		case err, ok := <-watcher.Errors():
-			if !ok {
-				return
-			}
-			sH.options.Logger.Error("event", zap.Error(err))
-		}
-	}
-}
-
-func (sH *StaticAssetsHandler) watch() {
-	if sH.options.UIConfigPath == "" {
-		sH.options.Logger.Info("UI config path not provided, config file will not be watched")
-		return
-	}
-
-	watcher, err := sH.newWatcher()
+func (sH *StaticAssetsHandler) reloadUIConfig() {
+	sH.options.Logger.Info("reloading UI config", zap.String("filename", sH.options.UIConfigPath))
+	content, err := loadAndEnrichIndexHTML(sH.assetsFS.Open, sH.options)
 	if err != nil {
-		sH.options.Logger.Error("failed to create a new watcher for the UI config", zap.Error(err))
-		return
-	}
-
-	go func() {
-		sH.configListener(watcher)
-	}()
-
-	if err := watcher.Add(sH.options.UIConfigPath); err != nil {
-		sH.options.Logger.Error("error adding watcher to file", zap.String("file", sH.options.UIConfigPath), zap.Error(err))
-	} else {
-		sH.options.Logger.Info("watching", zap.String("file", sH.options.UIConfigPath))
-	}
-
-	dir := filepath.Dir(sH.options.UIConfigPath)
-	if err := watcher.Add(dir); err != nil {
-		sH.options.Logger.Error("error adding watcher to dir", zap.String("dir", dir), zap.Error(err))
-	} else {
-		sH.options.Logger.Info("watching", zap.String("dir", dir))
+		sH.options.Logger.Error("error while reloading the UI config", zap.Error(err))
 	}
+	sH.indexHTML.Store(content)
+	sH.options.Logger.Info("reloaded UI config", zap.String("filename", sH.options.UIConfigPath))
 }
 
 func loadIndexHTML(open func(string) (http.File, error)) ([]byte, error) {

cmd/query/app/static_handler.go

@@ -26,17 +26,13 @@ import (
 	"testing"
 	"time"
 
-	"github.com/fsnotify/fsnotify"
 	"github.com/gorilla/mux"
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"go.uber.org/zap/zaptest/observer"
 
-	"github.com/jaegertracing/jaeger/cmd/query/app/mocks"
-	"github.com/jaegertracing/jaeger/pkg/fswatcher"
 	"github.com/jaegertracing/jaeger/pkg/testutils"
 )
 
@@ -151,100 +147,6 @@ func TestNewStaticAssetsHandlerErrors(t *testing.T) {
 	}
 }
 
-func TestWatcherError(t *testing.T) {
-	const totalWatcherAddCalls = 2
-
-	for _, tc := range []struct {
-		name                string
-		errorOnNthAdd       int
-		newWatcherErr       error
-		watcherAddErr       error
-		wantWatcherAddCalls int
-	}{
-		{
-			name:          "NewWatcher error",
-			newWatcherErr: fmt.Errorf("new watcher error"),
-		},
-		{
-			name:                "Watcher.Add first call error",
-			errorOnNthAdd:       0,
-			watcherAddErr:       fmt.Errorf("add first error"),
-			wantWatcherAddCalls: 2,
-		},
-		{
-			name:                "Watcher.Add second call error",
-			errorOnNthAdd:       1,
-			watcherAddErr:       fmt.Errorf("add second error"),
-			wantWatcherAddCalls: 2,
-		},
-	} {
-		t.Run(tc.name, func(t *testing.T) {
-			// Prepare
-			zcore, logObserver := observer.New(zapcore.InfoLevel)
-			logger := zap.New(zcore)
-			defer func() {
-				if r := recover(); r != nil {
-					// Select loop exits without logging error, only containing previous error log.
-					assert.Equal(t, logObserver.FilterMessage("event").Len(), 1)
-					assert.Equal(t, "send on closed channel", fmt.Sprint(r))
-				}
-			}()
-
-			watcher := &mocks.Watcher{}
-			for i := 0; i < totalWatcherAddCalls; i++ {
-				var err error
-				if i == tc.errorOnNthAdd {
-					err = tc.watcherAddErr
-				}
-				watcher.On("Add", mock.Anything).Return(err).Once()
-			}
-			watcher.On("Events").Return(make(chan fsnotify.Event))
-			errChan := make(chan error)
-			watcher.On("Errors").Return(errChan)
-
-			// Test
-			_, err := NewStaticAssetsHandler("fixture", StaticAssetsHandlerOptions{
-				UIConfigPath: "fixture/ui-config-hotreload.json",
-				NewWatcher: func() (fswatcher.Watcher, error) {
-					return watcher, tc.newWatcherErr
-				},
-				Logger: logger,
-			})
-
-			// Validate
-
-			// Error logged but not returned
-			assert.NoError(t, err)
-			if tc.newWatcherErr != nil {
-				assert.Equal(t, logObserver.FilterField(zap.Error(tc.newWatcherErr)).Len(), 1)
-			} else {
-				assert.Zero(t, logObserver.FilterField(zap.Error(tc.newWatcherErr)).Len())
-			}
-
-			if tc.watcherAddErr != nil {
-				assert.Equal(t, logObserver.FilterField(zap.Error(tc.watcherAddErr)).Len(), 1)
-			} else {
-				assert.Zero(t, logObserver.FilterField(zap.Error(tc.watcherAddErr)).Len())
-			}
-
-			watcher.AssertNumberOfCalls(t, "Add", tc.wantWatcherAddCalls)
-
-			// Validate Events and Errors channels
-			if tc.newWatcherErr == nil {
-				errChan <- fmt.Errorf("first error")
-
-				waitUntil(t, func() bool {
-					return logObserver.FilterMessage("event").Len() > 0
-				}, 100, 10*time.Millisecond, "timed out waiting for error")
-				assert.Equal(t, logObserver.FilterMessage("event").Len(), 1)
-
-				close(errChan)
-				errChan <- fmt.Errorf("second error on closed chan")
-			}
-		})
-	}
-}
-
 func TestHotReloadUIConfig(t *testing.T) {
 	dir, err := os.MkdirTemp("", "ui-config-hotreload-*")
 	require.NoError(t, err)