From 5fe0df7db1048979dba74823b635eb7b3970d989 Mon Sep 17 00:00:00 2001 From: Dmitriy Aleksandrov Date: Mon, 19 Oct 2026 12:19:44 +0300 Subject: [PATCH] src: watch the files by tree.WatchFiles, for the drivers that read files without the tree Signed-off-by: Dmitriy Aleksandrov --- drivers/internal/tree/tree.go | 98 ++++++++++++++++++++--------------- 1 file changed, 56 insertions(+), 42 deletions(-) diff --git a/drivers/internal/tree/tree.go b/drivers/internal/tree/tree.go index 23a0d7b..858dba0 100644 --- a/drivers/internal/tree/tree.go +++ b/drivers/internal/tree/tree.go @@ -213,16 +213,14 @@ var ( eventsDelay = 100 * time.Millisecond ) -// fileEvents returns the channel of the events of the directories with the files of the sources and the function -// that stops watching. Nil channel is returned if the directories can not be watched, the files are checked by polling then. +// fileEvents returns the channel of the events of the directories with the files and the function that stops watching. +// Nil channel is returned if the directories can not be watched, the files are checked by polling then. // Directories are watched instead of the files, so the replaced files and symlinks are watched too: // editors, atomic renames, updates of ConfigMap volumes in Kubernetes. -func (d *Driver) fileEvents() (<-chan fsnotify.Event, func()) { +func fileEvents(files []string) (<-chan fsnotify.Event, func()) { dirs := make(map[string]bool) - for _, source := range d.sources { - if source.File != "" { - dirs[filepath.Dir(source.File)] = true - } + for _, file := range files { + dirs[filepath.Dir(file)] = true } if len(dirs) == 0 { return nil, func() {} @@ -250,41 +248,16 @@ func (d *Driver) fileEvents() (<-chan fsnotify.Event, func()) { return watcher.Events, func() { _ = watcher.Close() } } -// Watch checks the sources for changes and calls onChange if any source is changed. -// Files are checked after the events of the file system and with the watch interval, if the events are lost. -// Sources with Watch report their changes by themselves. It blocks until ctx is done. -func (d *Driver) Watch(ctx context.Context, onChange func()) error { - // watchers of the sources are stopped and finished before return - var wg sync.WaitGroup - defer wg.Wait() - ctx, cancel := context.WithCancel(ctx) - defer cancel() - errs := make(chan error, len(d.sources)) - for _, source := range d.sources { - if source.Watch == nil { - continue - } - wg.Add(1) - go func(source Source) { - defer wg.Done() - if err := source.Watch(ctx, onChange); err != nil && ctx.Err() == nil { - errs <- fmt.Errorf("failed to watch %s: %w", source.Name, err) - } - }(source) - } - ticker := time.NewTicker(d.watchInterval) +// WatchFiles calls onChange every time the state of the files returned by stat differs from the previous one, +// starting from state. The state is checked after the events of the file system in the directories of the files +// and with the interval, if the events are lost. It blocks until ctx is done. +func WatchFiles(ctx context.Context, files []string, interval time.Duration, state string, stat func() string, onChange func()) { + ticker := time.NewTicker(interval) defer ticker.Stop() - events, stopEvents := d.fileEvents() + events, stopEvents := fileEvents(files) defer stopEvents() - // changes made after the last Load should be detected too - d.mu.RLock() - state, loaded := d.state, d.loaded - d.mu.RUnlock() - if !loaded { - state = d.stat(ctx) - } check := func() { - if newState := d.stat(ctx); newState != state { + if newState := stat(); newState != state { state = newState onChange() } @@ -293,9 +266,7 @@ func (d *Driver) Watch(ctx context.Context, onChange func()) error { for { select { case <-ctx.Done(): - return nil - case err := <-errs: - return err + return case <-ticker.C: check() case _, ok := <-events: @@ -312,6 +283,49 @@ func (d *Driver) Watch(ctx context.Context, onChange func()) error { } } +// Watch checks the sources for changes and calls onChange if any source is changed. +// Files are checked after the events of the file system and with the watch interval, if the events are lost. +// Sources with Watch report their changes by themselves. It blocks until ctx is done. +func (d *Driver) Watch(ctx context.Context, onChange func()) error { + ctx, cancel := context.WithCancel(ctx) + defer cancel() + // the first error of the sources stops watching, it is returned after the watchers of the sources are finished + var ( + wg sync.WaitGroup + errOnce sync.Once + watchErr error + ) + for _, source := range d.sources { + if source.Watch == nil { + continue + } + wg.Add(1) + go func(source Source) { + defer wg.Done() + if err := source.Watch(ctx, onChange); err != nil && ctx.Err() == nil { + errOnce.Do(func() { watchErr = fmt.Errorf("failed to watch %s: %w", source.Name, err) }) + cancel() + } + }(source) + } + var files []string + for _, source := range d.sources { + if source.File != "" { + files = append(files, source.File) + } + } + // changes made after the last Load should be detected too + d.mu.RLock() + state, loaded := d.state, d.loaded + d.mu.RUnlock() + if !loaded { + state = d.stat(ctx) + } + WatchFiles(ctx, files, d.watchInterval, state, func() string { return d.stat(ctx) }, onChange) + wg.Wait() + return watchErr +} + func (d *Driver) getTree() (map[string]any, error) { d.mu.RLock() tree, loaded := d.tree, d.loaded