adding mtime age while fetching files using doublestar and added refr…

…esh interval to prevent fetching files each pool

go.mod

@@ -2,6 +2,8 @@ module github.com/open-telemetry/opentelemetry-collector-contrib
 
 go 1.21
 
+replace github.com/bmatcuk/doublestar/v4 => github.com/opsramp/doublestar/v4 v4.0.0-20240422124750-f0571f1928b4
+
 require (
 	github.com/open-telemetry/opentelemetry-collector-contrib/connector/countconnector v0.98.0
 	github.com/open-telemetry/opentelemetry-collector-contrib/connector/datadogconnector v0.98.0

pkg/stanza/fileconsumer/matcher/internal/finder/finder.go

@@ -6,6 +6,7 @@ package finder // import "github.com/open-telemetry/opentelemetry-collector-cont
 import (
 	"errors"
 	"fmt"
+	"time"
 
 	"github.com/bmatcuk/doublestar/v4"
 )
@@ -21,16 +22,16 @@ func Validate(globs []string) error {
 }
 
 // FindFiles gets a list of paths given an array of glob patterns to include and exclude
-func FindFiles(includes []string, excludes []string) ([]string, error) {
+func FindFiles(includes []string, excludes []string, maxAge time.Duration) ([]string, error) {
 	var errs error
 	all := make([]string, 0, len(includes))
 	for _, include := range includes {
-		matches, err := doublestar.FilepathGlob(include, doublestar.WithFilesOnly(), doublestar.WithFailOnIOErrors())
+		matches, err := doublestar.FilepathGlob(include, doublestar.WithFilesOnly(), doublestar.WithFailOnIOErrors(), doublestar.WithMaxAge(maxAge))
 		if err != nil {
 			errs = errors.Join(errs, fmt.Errorf("find files with '%s' pattern: %w", include, err))
 			// the same pattern could cause an IO error due to one file or directory,
 			// but also could still find files without `doublestar.WithFailOnIOErrors()`.
-			matches, _ = doublestar.FilepathGlob(include, doublestar.WithFilesOnly())
+			matches, _ = doublestar.FilepathGlob(include, doublestar.WithFilesOnly(), doublestar.WithMaxAge(maxAge))
 		}
 	INCLUDE:
 		for _, match := range matches {

pkg/stanza/fileconsumer/matcher/internal/finder/finder_test.go

@@ -186,7 +186,7 @@ func TestFindFiles(t *testing.T) {
 				require.NoError(t, err)
 				require.NoError(t, file.Close())
 			}
-			files, err := FindFiles(tc.include, tc.exclude)
+			files, err := FindFiles(tc.include, tc.exclude, 0)
 			assert.NoError(t, err)
 			assert.Equal(t, tc.expected, files)
 		})
@@ -249,7 +249,7 @@ func TestFindFilesWithIOErrors(t *testing.T) {
 
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
-			files, err := FindFiles(tc.include, []string{})
+			files, err := FindFiles(tc.include, []string{}, 0)
 			assert.ErrorContains(t, err, tc.failedMsg)
 			assert.Equal(t, tc.expected, files)
 		})

pkg/stanza/fileconsumer/matcher/matcher.go

@@ -43,6 +43,8 @@ type OrderingCriteria struct {
 	Regex  string `mapstructure:"regex,omitempty"`
 	TopN   int    `mapstructure:"top_n,omitempty"`
 	SortBy []Sort `mapstructure:"sort_by,omitempty"`
+
+	RefreshInterval time.Duration `mapstructure:"refresh_interval,omitempty"`
 }
 
 type Sort struct {
@@ -70,13 +72,6 @@ func New(c Criteria) (*Matcher, error) {
 		return nil, fmt.Errorf("exclude: %w", err)
 	}
 
-	if len(c.OrderingCriteria.SortBy) == 0 {
-		return &Matcher{
-			include: c.Include,
-			exclude: c.Exclude,
-		}, nil
-	}
-
 	if c.OrderingCriteria.TopN < 0 {
 		return nil, fmt.Errorf("'top_n' must be a positive integer")
 	}
@@ -85,6 +80,20 @@ func New(c Criteria) (*Matcher, error) {
 		c.OrderingCriteria.TopN = defaultOrderingCriteriaTopN
 	}
 
+	if c.OrderingCriteria.RefreshInterval.Seconds() == 0 {
+		c.OrderingCriteria.RefreshInterval = time.Minute
+	}
+
+	if len(c.OrderingCriteria.SortBy) == 0 {
+		return &Matcher{
+			include:         c.Include,
+			exclude:         c.Exclude,
+			refreshInterval: c.OrderingCriteria.RefreshInterval,
+			topN:            c.OrderingCriteria.TopN,
+			cache:           newCache(),
+		}, nil
+	}
+
 	var regex *regexp.Regexp
 	if orderingCriteriaNeedsRegex(c.OrderingCriteria.SortBy) {
 		if c.OrderingCriteria.Regex == "" {
@@ -98,6 +107,7 @@ func New(c Criteria) (*Matcher, error) {
 		}
 	}
 
+	var maxAge time.Duration
 	var filterOpts []filter.Option
 	for _, sc := range c.OrderingCriteria.SortBy {
 		switch sc.SortType {
@@ -123,18 +133,22 @@ func New(c Criteria) (*Matcher, error) {
 			if !mtimeSortTypeFeatureGate.IsEnabled() {
 				return nil, fmt.Errorf("the %q feature gate must be enabled to use %q sort type", mtimeSortTypeFeatureGate.ID(), sortTypeMtime)
 			}
+			maxAge = sc.MaxTime
 			filterOpts = append(filterOpts, filter.SortMtime(sc.MaxTime))
 		default:
 			return nil, fmt.Errorf("'sort_type' must be specified")
 		}
 	}
 
 	return &Matcher{
-		include:    c.Include,
-		exclude:    c.Exclude,
-		regex:      regex,
-		topN:       c.OrderingCriteria.TopN,
-		filterOpts: filterOpts,
+		include:         c.Include,
+		exclude:         c.Exclude,
+		regex:           regex,
+		refreshInterval: c.OrderingCriteria.RefreshInterval,
+		maxAge:          maxAge,
+		topN:            c.OrderingCriteria.TopN,
+		filterOpts:      filterOpts,
+		cache:           newCache(),
 	}, nil
 }
 
@@ -149,36 +163,76 @@ func orderingCriteriaNeedsRegex(sorts []Sort) bool {
 	return false
 }
 
+// cache stores the matched files and last updated time. No mutex is used since all calls are sequential
+type cache struct {
+	lastUpdatedTime time.Time
+
+	files []string
+}
+
+func newCache() *cache {
+	return &cache{}
+}
+
+func (c *cache) getFiles() []string {
+	return c.files
+}
+
+func (c *cache) update(files []string) {
+	c.files = files
+	c.lastUpdatedTime = time.Now()
+}
+
+func (c *cache) getLastUpdatedTime() time.Time {
+	return c.lastUpdatedTime
+}
+
 type Matcher struct {
 	include    []string
 	exclude    []string
 	regex      *regexp.Regexp
 	topN       int
 	filterOpts []filter.Option
+
+	refreshInterval time.Duration
+	maxAge          time.Duration
+	cache           *cache
 }
 
 // MatchFiles gets a list of paths given an array of glob patterns to include and exclude
 func (m Matcher) MatchFiles() ([]string, error) {
-	var errs error
-	files, err := finder.FindFiles(m.include, m.exclude)
+	var err, errs error
+
+	files := m.cache.getFiles()
+	if time.Since(m.cache.getLastUpdatedTime()) < m.refreshInterval {
+		return files, nil
+	}
+
+	files, err = finder.FindFiles(m.include, m.exclude, m.maxAge)
 	if err != nil {
 		errs = errors.Join(errs, err)
 	}
+
 	if len(files) == 0 {
 		return files, errors.Join(fmt.Errorf("no files match the configured criteria"), errs)
 	}
 	if len(m.filterOpts) == 0 {
 		return files, errs
 	}
 
-	result, err := filter.Filter(files, m.regex, m.filterOpts...)
-	if len(result) == 0 {
-		return result, errors.Join(err, errs)
+	files, err = filter.Filter(files, m.regex, m.filterOpts...)
+	if len(files) == 0 {
+		return files, errors.Join(err, errs)
 	}
 
-	if len(result) <= m.topN {
-		return result, errors.Join(err, errs)
+	if len(files) <= m.topN {
+		m.cache.update(files)
+		return files, errors.Join(err, errs)
 	}
 
-	return result[:m.topN], errors.Join(err, errs)
+	files = files[:m.topN]
+
+	m.cache.update(files)
+
+	return files, errors.Join(err, errs)
 }

pkg/stanza/go.mod

@@ -2,6 +2,8 @@ module github.com/open-telemetry/opentelemetry-collector-contrib/pkg/stanza
 
 go 1.21
 
+replace github.com/bmatcuk/doublestar/v4 => github.com/opsramp/doublestar/v4 v4.0.0-20240422124750-f0571f1928b4
+
 require (
 	github.com/bmatcuk/doublestar/v4 v4.6.1
 	github.com/cespare/xxhash/v2 v2.3.0