Add a hot reloading RIE side implementation

cmd/localstack/awsutil.go

@@ -10,16 +10,20 @@ import (
 	"fmt"
 	"github.com/jessevdk/go-flags"
 	log "github.com/sirupsen/logrus"
+	"go.amzn.com/cmd/localstack/filenotify"
 	"go.amzn.com/lambda/interop"
 	"go.amzn.com/lambda/rapidcore"
 	"io"
+	"io/fs"
 	"math"
 	"net/http"
 	"os"
 	"path"
 	"path/filepath"
 	"strings"
 	"time"
+
+	"github.com/fsnotify/fsnotify"
 )
 
 const (
@@ -196,6 +200,135 @@ func DownloadCodeArchive(url string) {
 
 }
 
+func RunFileWatcher(server *CustomInteropServer, targetPaths []string, opts *LsOpts, done <-chan bool) {
+	if !opts.HotReloading {
+		return
+	}
+	defaultDuration := 500 * time.Millisecond
+	log.Infoln("Hot reloading enabled, starting filewatcher.", targetPaths)
+	watcher, err := filenotify.New(200 * time.Millisecond)
+	if err != nil {
+		log.Errorln("Hot reloading disabled due to filewatcher error.")
+		log.Errorln(err)
+		return
+	}
+	defer watcher.Close()
+
+	changeChannel := make(chan string, 10)
+	defer close(changeChannel)
+	// Start listening for events.
+	go func(channel chan<- string) {
+		var watchedFolders []string
+		for {
+			select {
+			case event, ok := <-watcher.Events():
+				if !ok {
+					return
+				}
+				log.Debugln("FileWatcher got event: ", event)
+				if event.Has(fsnotify.Create) {
+					stat, err := os.Stat(event.Name)
+					if err != nil {
+						log.Errorln("Error stating event file: ", event.Name, err)
+					} else if stat.IsDir() {
+						subfolders := getSubFolders(event.Name)
+						for _, folder := range subfolders {
+							err = watcher.Add(folder)
+							watchedFolders = append(watchedFolders, folder)
+							if err != nil {
+								log.Errorln("Error watching folder: ", folder, err)
+							}
+						}
+					}
+					// remove in case of remove / rename (rename within the folder will trigger a separate create event)
+				} else if event.Has(fsnotify.Remove) || event.Has(fsnotify.Rename) {
+					// remove all file watchers if it is in our folders list
+					toBeRemovedDirs, newWatchedFolders := getSubFoldersInList(event.Name, watchedFolders)
+					watchedFolders = newWatchedFolders
+					for _, dir := range toBeRemovedDirs {
+						err = watcher.Remove(dir)
+						if err != nil {
+							log.Warnln("Error removing path: ", event.Name, err)
+						}
+					}
+				}
+				channel <- event.Name
+			case err, ok := <-watcher.Errors():
+				if !ok {
+					log.Println("error:", err)
+					return
+				}
+				log.Println("error:", err)
+			}
+		}
+	}(changeChannel)
+
+	// debouncer to limit restarts
+	go func(channel <-chan string, duration time.Duration) {
+		timer := time.NewTimer(duration)
+		// immediately stop the timer, since we do not want to reload right at the startup
+		if !timer.Stop() {
+			// we have to drain the channel in case the timer already fired
+			<-timer.C
+		}
+		for {
+			select {
+			case _, more := <-channel:
+				if !more {
+					timer.Stop()
+					return
+				}
+				timer.Reset(duration)
+			case <-timer.C:
+				log.Println("Resetting environment...")
+				server.Reset("HotReload", 2000)
+			}
+		}
+
+	}(changeChannel, defaultDuration)
+
+	// Add all target paths and subfolders
+	for _, targetPath := range targetPaths {
+		subfolders := getSubFolders(targetPath)
+		log.Infoln("Subfolders: ", subfolders)
+		for _, target := range subfolders {
+			err = watcher.Add(target)
+			if err != nil {
+				log.Fatal(err)
+			}
+		}
+	}
+	<-done
+	log.Infoln("Closing down filewatcher.")
+
+}
+
+func getSubFolders(dirPath string) []string {
+	var subfolders []string
+	err := filepath.WalkDir(dirPath, func(path string, d fs.DirEntry, err error) error {
+		if err == nil && d.IsDir() {
+			subfolders = append(subfolders, path)
+		}
+		return err
+	})
+	if err != nil {
+		log.Errorln("Error listing directory contents: ", err)
+		return subfolders
+	}
+	return subfolders
+}
+
+func getSubFoldersInList(prefix string, pathList []string) (old_folders []string, new_folders []string) {
+	for _, item := range pathList {
+		if strings.HasPrefix(item, prefix) {
+			old_folders = append(old_folders, item)
+		} else {
+			new_folders = append(new_folders, item)
+		}
+	}
+	return
+}
+
 func InitHandler(sandbox Sandbox, functionVersion string, timeout int64) (time.Time, time.Time) {
 	additionalFunctionEnvironmentVariables := map[string]string{}
 

cmd/localstack/filenotify/filenotify.go

@@ -0,0 +1,64 @@
+// This package is adapted from https://github.com/gohugoio/hugo/tree/master/watcher/filenotify, Apache-2.0 License.
+
+// Package filenotify provides a mechanism for watching file(s) for changes.
+// Generally leans on fsnotify, but provides a poll-based notifier which fsnotify does not support.
+// These are wrapped up in a common interface so that either can be used interchangeably in your code.
+//
+// This package is adapted from https://github.com/moby/moby/tree/master/pkg/filenotify, Apache-2.0 License.
+// Hopefully this can be replaced with an external package sometime in the future, see https://github.com/fsnotify/fsnotify/issues/9
+package filenotify
+
+import (
+	log "github.com/sirupsen/logrus"
+	"golang.org/x/sys/unix"
+	"strings"
+	"time"
+
+	"github.com/fsnotify/fsnotify"
+)
+
+// FileWatcher is an interface for implementing file notification watchers
+type FileWatcher interface {
+	Events() <-chan fsnotify.Event
+	Errors() <-chan error
+	Add(name string) error
+	Remove(name string) error
+	Close() error
+}
+
+// New tries to use a fs-event watcher, and falls back to the poller if there is an error
+func New(interval time.Duration) (FileWatcher, error) {
+	// cheap check if we are in Docker desktop or not.
+	// We could also inspect the mounts, but that would be more complicated and needs more parsing
+	var utsname unix.Utsname
+	err := unix.Uname(&utsname)
+	release := strings.TrimRight(string(utsname.Release[:]), "\x00")
+	log.Println("Release detected: ", release)
+	if err == nil && !(strings.Contains(release, "linuxkit") || strings.Contains(release, "WSL2")) {
+		if watcher, err := NewEventWatcher(); err == nil {
+			log.Debugln("Using event based filewatcher")
+			return watcher, nil
+		}
+	}
+	log.Debugln("Using polling based filewatcher")
+	return NewPollingWatcher(interval), nil
+}
+
+// NewPollingWatcher returns a poll-based file watcher
+func NewPollingWatcher(interval time.Duration) FileWatcher {
+	return &filePoller{
+		interval: interval,
+		done:     make(chan struct{}),
+		events:   make(chan fsnotify.Event),
+		errors:   make(chan error),
+	}
+}
+
+// NewEventWatcher returns a fs-event based file watcher
+func NewEventWatcher() (FileWatcher, error) {
+	watcher, err := fsnotify.NewWatcher()
+	if err != nil {
+		return nil, err
+	}
+	return &fsNotifyWatcher{watcher}, nil
+}

cmd/localstack/filenotify/fsnotify.go

@@ -0,0 +1,22 @@
+// This package is adapted from https://github.com/gohugoio/hugo/tree/master/watcher/filenotify, Apache-2.0 License.
+
+// Package filenotify is adapted from https://github.com/moby/moby/tree/master/pkg/filenotify, Apache-2.0 License.
+// Hopefully this can be replaced with an external package sometime in the future, see https://github.com/fsnotify/fsnotify/issues/9
+package filenotify
+
+import "github.com/fsnotify/fsnotify"
+
+// fsNotifyWatcher wraps the fsnotify package to satisfy the FileNotifier interface
+type fsNotifyWatcher struct {
+	*fsnotify.Watcher
+}
+
+// Events returns the fsnotify event channel receiver
+func (w *fsNotifyWatcher) Events() <-chan fsnotify.Event {
+	return w.Watcher.Events
+}
+
+// Errors returns the fsnotify error channel receiver
+func (w *fsNotifyWatcher) Errors() <-chan error {
+	return w.Watcher.Errors
+}