Skip to content

fix: faster update of file tasks for loki.source.file #4611

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 13 commits into from
Closed

fix: faster update of file tasks for loki.source.file #4611

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
6a46dd1
Split consumer loop and target update loop
kalleep Oct 14, 2025
4bfe5eb
Use waitgroup
kalleep Oct 14, 2025
cff887e
fix log
kalleep Oct 14, 2025
ce225c1
file_match: only export targets if anything has changed
kalleep Oct 16, 2025
9ac2016
Update internal/component/loki/source/file/file.go
kalleep Oct 16, 2025
36db133
Update order
kalleep Oct 16, 2025
716a5fb
fix equal check
kalleep Oct 16, 2025
d6aba4f
Rework so we don't stat path twice
kalleep Oct 16, 2025
46103e4
always update state
kalleep Oct 16, 2025
a6e3b56
cleanup
kalleep Oct 16, 2025
2dfe22a
fix
kalleep Oct 16, 2025
494f6a7
Prevent checking targets multiple time in cases where we cannot create a
kalleep Oct 20, 2025
c24f4a7
Add debug logging
kalleep Oct 21, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 44 additions & 21 deletions internal/component/local/file_match/file.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,20 +37,23 @@ var _ component.Component = (*Component)(nil)
type Component struct {
opts component.Options

mut sync.RWMutex
args Arguments
watches []watch
watchDog *time.Ticker
mut sync.Mutex
args Arguments
watches []watch
watchDog *time.Ticker
targetsChanged chan struct{}
}

// New creates a new local.file_match component.
func New(o component.Options, args Arguments) (*Component, error) {
c := &Component{
opts: o,
mut: sync.RWMutex{},
mut: sync.Mutex{},
args: args,
watches: make([]watch, 0),
watchDog: time.NewTicker(args.SyncPeriod),
// Buffered channel to avoid blocking
targetsChanged: make(chan struct{}, 1),
}

if err := c.Update(args); err != nil {
Expand All @@ -73,11 +76,16 @@ func (c *Component) Update(args component.Arguments) error {
c.mut.Lock()
defer c.mut.Unlock()

newArgs := args.(Arguments)

// Check to see if our ticker timer needs to be reset.
if args.(Arguments).SyncPeriod != c.args.SyncPeriod {
c.watchDog.Reset(c.args.SyncPeriod)
if newArgs.SyncPeriod != c.args.SyncPeriod {
c.watchDog.Reset(newArgs.SyncPeriod)
}
c.args = args.(Arguments)

c.args = newArgs

// Rebuild watches
c.watches = c.watches[:0]
for _, v := range c.args.PathTargets {
c.watches = append(c.watches, watch{
Expand All @@ -87,35 +95,47 @@ func (c *Component) Update(args component.Arguments) error {
})
}

// Always trigger immediate check when Update is called
select {
case c.targetsChanged <- struct{}{}:
default:
}

return nil
}

// Run satisfies the component interface.
func (c *Component) Run(ctx context.Context) error {
update := func() {
c.mut.Lock()
defer c.mut.Unlock()

paths := c.getWatchedFiles()
// The component node checks to see if exports have actually changed.
c.opts.OnStateChange(discovery.Exports{Targets: paths})
}
// Trigger initial check
update()
defer c.watchDog.Stop()
for {
select {
case <-c.targetsChanged:
// When we get a signal that we have new targets we will get all watched files and
// reset the timer.
c.mut.Lock()
c.watchDog.Reset(c.args.SyncPeriod)
targets := c.getWatchedFiles(true)
c.mut.Unlock()
c.opts.OnStateChange(discovery.Exports{Targets: targets})
case <-c.watchDog.C:
// This triggers a check for any new paths, along with pushing new targets.
update()
// If we have not received a signal that we have new targets watch job will periodically
// get all files that we should watch.
c.mut.Lock()
targets := c.getWatchedFiles(false)
c.mut.Unlock()
c.opts.OnStateChange(discovery.Exports{Targets: targets})
case <-ctx.Done():
return nil
}
}
}

func (c *Component) getWatchedFiles() []discovery.Target {
func (c *Component) getWatchedFiles(targetsUpdated bool) []discovery.Target {
start := time.Now()
paths := make([]discovery.Target, 0)

level.Info(c.opts.Logger).Log("msg", "expanding paths", "targets_updated", targetsUpdated)

// See if there is anything new we need to check.
for _, w := range c.watches {
newPaths, err := w.getPaths()
Expand All @@ -124,5 +144,8 @@ func (c *Component) getWatchedFiles() []discovery.Target {
}
paths = append(paths, newPaths...)
}

level.Info(c.opts.Logger).Log("msg", "finish expanding paths", "paths", len(paths), "duration", time.Since(start))

return paths
}
22 changes: 11 additions & 11 deletions internal/component/local/file_match/file_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@ func TestFile(t *testing.T) {
go c.Run(ct)
time.Sleep(20 * time.Millisecond)
ct.Done()
foundFiles := c.getWatchedFiles()
foundFiles := c.getWatchedFiles(false)
require.Len(t, foundFiles, 1)
require.True(t, contains(foundFiles, "t1.txt"))
}
Expand All @@ -58,7 +58,7 @@ func TestDirectoryFile(t *testing.T) {
go c.Run(ct)
time.Sleep(20 * time.Millisecond)
ct.Done()
foundFiles := c.getWatchedFiles()
foundFiles := c.getWatchedFiles(false)
require.Len(t, foundFiles, 1)
require.True(t, contains(foundFiles, "t1.txt"))
}
Expand All @@ -80,14 +80,14 @@ func TestFileIgnoreOlder(t *testing.T) {
c.Update(c.args)
go c.Run(ct)

foundFiles := c.getWatchedFiles()
foundFiles := c.getWatchedFiles(false)
require.Len(t, foundFiles, 1)
require.True(t, contains(foundFiles, "t1.txt"))
time.Sleep(150 * time.Millisecond)

writeFile(t, dir, "t2.txt")
ct.Done()
foundFiles = c.getWatchedFiles()
foundFiles = c.getWatchedFiles(false)
require.Len(t, foundFiles, 1)
require.True(t, contains(foundFiles, "t2.txt"))
}
Expand All @@ -110,7 +110,7 @@ func TestAddingFile(t *testing.T) {
time.Sleep(20 * time.Millisecond)
writeFile(t, dir, "t2.txt")
ct.Done()
foundFiles := c.getWatchedFiles()
foundFiles := c.getWatchedFiles(false)
require.Len(t, foundFiles, 2)
require.True(t, contains(foundFiles, "t1.txt"))
require.True(t, contains(foundFiles, "t2.txt"))
Expand Down Expand Up @@ -138,7 +138,7 @@ func TestAddingFileInSubDir(t *testing.T) {
require.NoError(t, err)
time.Sleep(20 * time.Millisecond)
ct.Done()
foundFiles := c.getWatchedFiles()
foundFiles := c.getWatchedFiles(false)
require.Len(t, foundFiles, 3)
require.True(t, contains(foundFiles, "t1.txt"))
require.True(t, contains(foundFiles, "t2.txt"))
Expand Down Expand Up @@ -175,7 +175,7 @@ func TestAddingFileInAnExcludedSubDir(t *testing.T) {
require.NoError(t, err)
time.Sleep(20 * time.Millisecond)
ct.Done()
foundFiles := c.getWatchedFiles()
foundFiles := c.getWatchedFiles(false)
require.Len(t, foundFiles, 3)
require.True(t, contains(foundFiles, "t1.txt"))
require.True(t, contains(foundFiles, "t2.txt"))
Expand Down Expand Up @@ -204,7 +204,7 @@ func TestAddingRemovingFileInSubDir(t *testing.T) {
err := os.WriteFile(path.Join(subdir, "t3.txt"), []byte("asdf"), 0664)
require.NoError(t, err)
time.Sleep(100 * time.Millisecond)
foundFiles := c.getWatchedFiles()
foundFiles := c.getWatchedFiles(false)
require.Len(t, foundFiles, 3)
require.True(t, contains(foundFiles, "t1.txt"))
require.True(t, contains(foundFiles, "t2.txt"))
Expand All @@ -213,7 +213,7 @@ func TestAddingRemovingFileInSubDir(t *testing.T) {
err = os.RemoveAll(subdir)
require.NoError(t, err)
time.Sleep(1000 * time.Millisecond)
foundFiles = c.getWatchedFiles()
foundFiles = c.getWatchedFiles(false)
require.Len(t, foundFiles, 2)
require.True(t, contains(foundFiles, "t1.txt"))
require.True(t, contains(foundFiles, "t2.txt"))
Expand All @@ -237,7 +237,7 @@ func TestExclude(t *testing.T) {
os.Mkdir(subdir, 0755)
writeFile(t, subdir, "t3.txt")
time.Sleep(100 * time.Millisecond)
foundFiles := c.getWatchedFiles()
foundFiles := c.getWatchedFiles(false)
require.Len(t, foundFiles, 2)
require.True(t, contains(foundFiles, "t1.txt"))
require.True(t, contains(foundFiles, "t3.txt"))
Expand All @@ -263,7 +263,7 @@ func TestMultiLabels(t *testing.T) {
c.args.SyncPeriod = 10 * time.Millisecond
go c.Run(ct)
time.Sleep(100 * time.Millisecond)
foundFiles := c.getWatchedFiles()
foundFiles := c.getWatchedFiles(false)
require.Len(t, foundFiles, 2)
require.True(t, contains([]discovery.Target{foundFiles[0]}, "t1.txt"))
require.True(t, contains([]discovery.Target{foundFiles[1]}, "t1.txt"))
Expand Down
Loading