[supervisor] analyze image file changes

Author: akosyakov

components/supervisor/pkg/config/gitpod-config.go

@@ -7,13 +7,14 @@ package config
 import (
 	"context"
 	"os"
+	"path/filepath"
 	"sync"
 	"time"
 
 	"github.com/fsnotify/fsnotify"
-	"github.com/sirupsen/logrus"
-	"gopkg.in/yaml.v2"
+	"gopkg.in/yaml.v3"
 
+	"github.com/gitpod-io/gitpod/common-go/log"
 	gitpod "github.com/gitpod-io/gitpod/gitpod-protocol"
 )
 
@@ -23,142 +24,231 @@ type ConfigInterface interface {
 	Watch(ctx context.Context)
 	// Observe provides channels triggered whenever the config is changed
 	Observe(ctx context.Context) <-chan *gitpod.GitpodConfig
+	// Observe provides channels triggered whenever the image file is changed
+	ObserveImageFile(ctx context.Context) <-chan *struct{}
 }
 
 // ConfigService provides access to the gitpod config file.
 type ConfigService struct {
-	location      string
 	locationReady <-chan struct{}
 
-	cond   *sync.Cond
-	config *gitpod.GitpodConfig
+	configLocation string
+	configWatcher  *fileWatcher[gitpod.GitpodConfig]
 
-	pollTimer *time.Timer
+	imageWatcher *fileWatcher[struct{}]
+}
 
-	log *logrus.Entry
+// NewConfigService creates a new instance of ConfigService.
+func NewConfigService(configLocation string, locationReady <-chan struct{}) *ConfigService {
+	return &ConfigService{
+		locationReady:  locationReady,
+		configLocation: configLocation,
+		configWatcher: newFileWatcher(func(data []byte) (*gitpod.GitpodConfig, error) {
+			var config *gitpod.GitpodConfig
+			err := yaml.Unmarshal(data, &config)
+			return config, err
+		}),
+		imageWatcher: newFileWatcher(func(data []byte) (*struct{}, error) {
+			return &struct{}{}, nil
+		}),
+	}
+}
+
+// Observe provides channels triggered whenever the config is changed.
+func (service *ConfigService) Observe(ctx context.Context) <-chan *gitpod.GitpodConfig {
+	return service.configWatcher.observe(ctx)
+}
+
+// Observe provides channels triggered whenever the image file is changed
+func (service *ConfigService) ObserveImageFile(ctx context.Context) <-chan *struct{} {
+	return service.imageWatcher.observe(ctx)
+}
+
+// Watch starts the config watching.
+func (service *ConfigService) Watch(ctx context.Context) {
+	select {
+	case <-service.locationReady:
+	case <-ctx.Done():
+		return
+	}
+	go service.watchImageFile(ctx)
+	service.configWatcher.watch(ctx, service.configLocation)
+}
+
+func (service *ConfigService) watchImageFile(ctx context.Context) {
+	var (
+		imageLocation string
+		cancelWatch   func()
+	)
+	defer func() {
+		if cancelWatch != nil {
+			cancelWatch()
+		}
+	}()
+	cfgs := service.configWatcher.observe(ctx)
+	for {
+		select {
+		case cfg, ok := <-cfgs:
+			if !ok {
+				return
+			}
+			var currentImageLocation string
+			if cfg != nil {
+				switch img := cfg.Image.(type) {
+				case map[string]interface{}:
+					currentImageLocation = filepath.Join(filepath.Dir(service.configLocation), img["file"].(string))
+				}
+			}
+			if imageLocation == currentImageLocation {
+				continue
+			}
+			if cancelWatch != nil {
+				cancelWatch()
+				cancelWatch = nil
+				service.imageWatcher.reset()
+			}
+			imageLocation = currentImageLocation
+			if imageLocation == "" {
+				continue
+			}
+			watchCtx, cancel := context.WithCancel(ctx)
+			cancelWatch = cancel
+			go service.imageWatcher.watch(watchCtx, imageLocation)
+		case <-ctx.Done():
+			return
+		}
+	}
+}
+
+type fileWatcher[T any] struct {
+	unmarshal func(data []byte) (*T, error)
+
+	cond *sync.Cond
+	data *T
+
+	pollTimer *time.Timer
 
 	ready     chan struct{}
 	readyOnce sync.Once
 
 	debounceDuration time.Duration
 }
 
-// NewConfigService creates a new instance of ConfigService.
-func NewConfigService(configLocation string, locationReady <-chan struct{}, log *logrus.Entry) *ConfigService {
-	return &ConfigService{
-		location:         configLocation,
-		locationReady:    locationReady,
+func newFileWatcher[T any](unmarshal func(data []byte) (*T, error)) *fileWatcher[T] {
+	return &fileWatcher[T]{
+		unmarshal:        unmarshal,
 		cond:             sync.NewCond(&sync.Mutex{}),
-		log:              log.WithField("location", configLocation),
 		ready:            make(chan struct{}),
 		debounceDuration: 100 * time.Millisecond,
 	}
 }
 
-// Observe provides channels triggered whenever the config is changed.
-func (service *ConfigService) Observe(ctx context.Context) <-chan *gitpod.GitpodConfig {
-	configs := make(chan *gitpod.GitpodConfig)
+func (service *fileWatcher[T]) observe(ctx context.Context) <-chan *T {
+	results := make(chan *T)
 	go func() {
-		defer close(configs)
+		defer close(results)
 
 		<-service.ready
 
 		service.cond.L.Lock()
 		defer service.cond.L.Unlock()
 		for {
-			configs <- service.config
+			results <- service.data
 
 			service.cond.Wait()
 			if ctx.Err() != nil {
 				return
 			}
 		}
 	}()
-	return configs
+	return results
 }
 
-// Watch starts the config watching.
-func (service *ConfigService) Watch(ctx context.Context) {
-	service.log.Info("gitpod config watcher: starting...")
+func (service *fileWatcher[T]) markReady() {
+	service.readyOnce.Do(func() {
+		close(service.ready)
+	})
+}
 
-	select {
-	case <-service.locationReady:
-	case <-ctx.Done():
-		return
-	}
+func (service *fileWatcher[T]) reset() {
+	service.cond.L.Lock()
+	defer service.cond.L.Unlock()
 
-	_, err := os.Stat(service.location)
-	if os.IsNotExist(err) {
-		service.poll(ctx)
+	if service.data != nil {
+		service.data = nil
+		service.cond.Broadcast()
 	}
-	service.watch(ctx)
 }
 
-func (service *ConfigService) markReady() {
-	service.readyOnce.Do(func() {
-		close(service.ready)
-	})
+func (service *fileWatcher[T]) watch(ctx context.Context, location string) {
+	log.WithField("location", location).Info("file watcher: starting...")
+
+	_, err := os.Stat(location)
+	if os.IsNotExist(err) {
+		service.poll(ctx, location)
+	}
+	service.doWatch(ctx, location)
 }
 
-func (service *ConfigService) watch(ctx context.Context) {
+func (service *fileWatcher[T]) doWatch(ctx context.Context, location string) {
 	watcher, err := fsnotify.NewWatcher()
 	defer func() {
 		if err != nil {
-			service.log.WithError(err).Error("gitpod config watcher: failed to start")
+			log.WithField("location", location).WithError(err).Error("file watcher: failed to start")
 			return
 		}
 
-		service.log.Info("gitpod config watcher: started")
+		log.WithField("location", location).Info("file watcher: started")
 	}()
 	if err != nil {
 		return
 	}
 
-	err = watcher.Add(service.location)
+	err = watcher.Add(location)
 	if err != nil {
 		watcher.Close()
 		return
 	}
 
 	go func() {
-		defer service.log.Info("gitpod config watcher: stopped")
+		defer log.WithField("location", location).Info("file watcher: stopped")
 		defer watcher.Close()
 
 		polling := make(chan struct{}, 1)
-		service.scheduleUpdateConfig(ctx, polling)
+		service.scheduleUpdate(ctx, polling, location)
 		for {
 			select {
 			case <-polling:
 				return
 			case <-ctx.Done():
 				return
 			case err := <-watcher.Errors:
-				service.log.WithError(err).Error("gitpod config watcher: failed to watch")
+				log.WithField("location", location).WithError(err).Error("file watcher: failed to watch")
 			case <-watcher.Events:
-				service.scheduleUpdateConfig(ctx, polling)
+				service.scheduleUpdate(ctx, polling, location)
 			}
 		}
 	}()
 }
 
-func (service *ConfigService) scheduleUpdateConfig(ctx context.Context, polling chan<- struct{}) {
+func (service *fileWatcher[T]) scheduleUpdate(ctx context.Context, polling chan<- struct{}, location string) {
 	service.cond.L.Lock()
 	defer service.cond.L.Unlock()
 	if service.pollTimer != nil {
 		service.pollTimer.Stop()
 	}
 	service.pollTimer = time.AfterFunc(service.debounceDuration, func() {
-		err := service.updateConfig()
+		err := service.update(location)
 		if os.IsNotExist(err) {
 			polling <- struct{}{}
-			go service.poll(ctx)
+			go service.poll(ctx, location)
 		} else if err != nil {
-			service.log.WithError(err).Error("gitpod config watcher: failed to parse")
+			log.WithField("location", location).WithError(err).Error("file watcher: failed to parse")
 		}
 	})
 }
 
-func (service *ConfigService) poll(ctx context.Context) {
+func (service *fileWatcher[T]) poll(ctx context.Context, location string) {
 	service.markReady()
 
 	timer := time.NewTicker(2 * time.Second)
@@ -171,35 +261,33 @@ func (service *ConfigService) poll(ctx context.Context) {
 		case <-timer.C:
 		}
 
-		if _, err := os.Stat(service.location); !os.IsNotExist(err) {
-			service.watch(ctx)
+		if _, err := os.Stat(location); !os.IsNotExist(err) {
+			service.doWatch(ctx, location)
 			return
 		}
 	}
 }
 
-func (service *ConfigService) updateConfig() error {
+func (service *fileWatcher[T]) update(location string) error {
 	service.cond.L.Lock()
 	defer service.cond.L.Unlock()
 
-	config, err := service.parse()
+	data, err := service.parse(location)
 	if err == nil || os.IsNotExist(err) {
-		service.config = config
+		service.data = data
 		service.markReady()
 		service.cond.Broadcast()
 
-		service.log.WithField("config", service.config).Debug("gitpod config watcher: updated")
+		log.WithField("location", location).WithField("data", service.data).Debug("file watcher: updated")
 	}
 
 	return err
 }
 
-func (service *ConfigService) parse() (*gitpod.GitpodConfig, error) {
-	data, err := os.ReadFile(service.location)
+func (service *fileWatcher[T]) parse(location string) (*T, error) {
+	data, err := os.ReadFile(location)
 	if err != nil {
 		return nil, err
 	}
-	var config *gitpod.GitpodConfig
-	err = yaml.Unmarshal(data, &config)
-	return config, err
+	return service.unmarshal(data)
 }

components/supervisor/pkg/config/gitpod-config_analytics_test.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 
+	"github.com/gitpod-io/gitpod/common-go/log"
 	gitpod "github.com/gitpod-io/gitpod/gitpod-protocol"
 )
 
@@ -65,7 +66,7 @@ func TestAnalyzeGitpodConfig(t *testing.T) {
 	for _, test := range tests {
 		t.Run(test.Desc, func(t *testing.T) {
 			var fields []string
-			analyzer := NewConfigAnalyzer(log, 100*time.Millisecond, func(field string) {
+			analyzer := NewConfigAnalyzer(log.Log, 100*time.Millisecond, func(field string) {
 				fields = append(fields, field)
 			}, test.Prev)
 			<-analyzer.Analyse(test.Current)