Skip to content

Commit c58e7d8

Browse files
orestisflmergify[bot]
orestisfl
authored and
mergify[bot]
committed
filestream: reduce per-scan allocations (#51863)
Two per-file, per-scan allocations are removed: - fileWatcher.watch resolved the file identity for every file up front. It is now resolved lazily on first use. - fileScanner.GetFiles rebuilt its per-scan maps from empty, forcing repeated grow/rehash allocations on a stable file set. They are now pre-sized from the previous scan's unique-file count. Combined effect vs main (go test -bench 'BenchmarkGetFiles|BenchmarkWatchIdle' -benchmem -benchtime=200ms -count=10): │ main │ this-commit │ │ sec/op │ sec/op vs base │ GetFiles-14 1.660m ± 16% 1.302m ± 5% -21.55% (p=0.000 n=10) GetFilesWithFingerprint-14 5.657m ± 9% 4.898m ± 15% -13.42% (p=0.019 n=10) GetFilesWithFingerprintGrowing/static-14 5.026m ± 14% 4.449m ± 6% -11.49% (p=0.001 n=10) GetFilesWithFingerprintGrowing/growing-14 6.823m ± 11% 7.775m ± 19% +13.96% (p=0.023 n=10) WatchIdle/active/path/100_files-14 160.6µ ± 12% 155.9µ ± 6% ~ (p=0.280 n=10) WatchIdle/active/path/1000_files-14 1.938m ± 12% 1.805m ± 21% ~ (p=0.218 n=10) WatchIdle/active/path/10000_files-14 27.05m ± 12% 24.65m ± 14% -8.87% (p=0.009 n=10) WatchIdle/active/fingerprint/100_files-14 439.9µ ± 2% 413.3µ ± 3% -6.04% (p=0.004 n=10) WatchIdle/active/fingerprint/1000_files-14 5.352m ± 11% 5.300m ± 20% ~ (p=0.853 n=10) WatchIdle/active/fingerprint/10000_files-14 58.20m ± 6% 59.59m ± 12% ~ (p=0.190 n=10) WatchIdle/ignored/path/100_files-14 170.0µ ± 9% 140.3µ ± 10% -17.47% (p=0.000 n=10) WatchIdle/ignored/path/1000_files-14 2.270m ± 9% 1.652m ± 1% -27.23% (p=0.000 n=10) WatchIdle/ignored/path/10000_files-14 25.86m ± 12% 25.09m ± 10% ~ (p=0.218 n=10) WatchIdle/ignored/fingerprint/100_files-14 428.7µ ± 14% 398.9µ ± 7% ~ (p=0.218 n=10) WatchIdle/ignored/fingerprint/1000_files-14 5.107m ± 10% 5.325m ± 15% ~ (p=0.971 n=10) WatchIdle/ignored/fingerprint/10000_files-14 58.54m ± 10% 56.75m ± 5% ~ (p=0.353 n=10) geomean 3.486m 3.230m -7.35% │ main │ this-commit │ │ B/op │ B/op vs base │ GetFiles-14 1.394Mi ± 3% 1.032Mi ± 1% -25.93% (p=0.000 n=10) GetFilesWithFingerprint-14 2.199Mi ± 3% 1.757Mi ± 4% -20.10% (p=0.000 n=10) GetFilesWithFingerprintGrowing/static-14 1.611Mi ± 0% 1.275Mi ± 0% -20.87% (p=0.000 n=10) GetFilesWithFingerprintGrowing/growing-14 5.521Mi ± 0% 5.190Mi ± 0% -5.99% (p=0.000 n=10) WatchIdle/active/path/100_files-14 121.7Ki ± 0% 101.7Ki ± 0% -16.42% (p=0.000 n=10) WatchIdle/active/path/1000_files-14 1.452Mi ± 0% 1.108Mi ± 0% -23.64% (p=0.000 n=10) WatchIdle/active/path/10000_files-14 13.73Mi ± 0% 10.94Mi ± 0% -20.29% (p=0.000 n=10) WatchIdle/active/fingerprint/100_files-14 172.1Ki ± 0% 152.2Ki ± 0% -11.59% (p=0.000 n=10) WatchIdle/active/fingerprint/1000_files-14 1.963Mi ± 0% 1.620Mi ± 0% -17.49% (p=0.000 n=10) WatchIdle/active/fingerprint/10000_files-14 18.72Mi ± 0% 15.94Mi ± 0% -14.88% (p=0.000 n=10) WatchIdle/ignored/path/100_files-14 121.73Ki ± 0% 98.59Ki ± 0% -19.01% (p=0.000 n=10) WatchIdle/ignored/path/1000_files-14 1.452Mi ± 0% 1.078Mi ± 0% -25.76% (p=0.000 n=10) WatchIdle/ignored/path/10000_files-14 13.73Mi ± 0% 10.64Mi ± 0% -22.51% (p=0.000 n=10) WatchIdle/ignored/fingerprint/100_files-14 172.2Ki ± 0% 133.4Ki ± 0% -22.54% (p=0.000 n=10) WatchIdle/ignored/fingerprint/1000_files-14 1.964Mi ± 0% 1.436Mi ± 0% -26.86% (p=0.000 n=10) WatchIdle/ignored/fingerprint/10000_files-14 19.00Mi ± 0% 14.38Mi ± 0% -24.30% (p=0.000 n=10) geomean 1.721Mi 1.376Mi -20.06% │ main │ this-commit │ │ allocs/op │ allocs/op vs base │ GetFiles-14 9.155k ± 0% 9.098k ± 0% -0.62% (p=0.000 n=10) GetFilesWithFingerprint-14 16.34k ± 0% 16.25k ± 0% -0.54% (p=0.000 n=10) GetFilesWithFingerprintGrowing/static-14 16.09k ± 0% 16.05k ± 0% -0.28% (p=0.000 n=10) GetFilesWithFingerprintGrowing/growing-14 18.09k ± 0% 18.05k ± 0% -0.25% (p=0.000 n=10) WatchIdle/active/path/100_files-14 1.452k ± 0% 1.433k ± 0% -1.31% (p=0.000 n=10) WatchIdle/active/path/1000_files-14 14.09k ± 0% 14.05k ± 0% -0.33% (p=0.000 n=10) WatchIdle/active/path/10000_files-14 140.3k ± 0% 140.1k ± 0% -0.10% (p=0.000 n=10) WatchIdle/active/fingerprint/100_files-14 1.956k ± 0% 1.937k ± 0% -0.97% (p=0.000 n=10) WatchIdle/active/fingerprint/1000_files-14 19.10k ± 0% 19.05k ± 0% -0.25% (p=0.000 n=10) WatchIdle/active/fingerprint/10000_files-14 190.3k ± 0% 190.2k ± 0% -0.07% (p=0.000 n=10) WatchIdle/ignored/path/100_files-14 1.452k ± 0% 1.233k ± 0% -15.08% (p=0.000 n=10) WatchIdle/ignored/path/1000_files-14 14.09k ± 0% 12.05k ± 0% -14.52% (p=0.000 n=10) WatchIdle/ignored/path/10000_files-14 140.3k ± 0% 120.1k ± 0% -14.36% (p=0.000 n=10) WatchIdle/ignored/fingerprint/100_files-14 1.956k ± 0% 1.737k ± 0% -11.20% (p=0.000 n=10) WatchIdle/ignored/fingerprint/1000_files-14 19.10k ± 0% 17.05k ± 0% -10.72% (p=0.000 n=10) WatchIdle/ignored/fingerprint/10000_files-14 190.3k ± 0% 170.2k ± 0% -10.58% (p=0.000 n=10) geomean 15.98k 15.14k -5.27% The map pre-sizing (fileScanner.lastCount) in isolation, benchmarked as this commit with and without the field: │ without-lastCount │ with-lastCount │ │ sec/op │ sec/op vs base │ GetFiles-14 1.743m ± 16% 1.293m ± 29% -25.81% (p=0.002 n=10) GetFilesWithFingerprint-14 5.056m ± 16% 4.905m ± 20% ~ (p=0.529 n=10) GetFilesWithFingerprintGrowing/static-14 4.642m ± 16% 4.282m ± 19% ~ (p=0.089 n=10) GetFilesWithFingerprintGrowing/growing-14 8.157m ± 20% 6.981m ± 13% -14.41% (p=0.009 n=10) WatchIdle/active/path/100_files-14 160.9µ ± 3% 142.1µ ± 4% -11.69% (p=0.000 n=10) WatchIdle/active/path/1000_files-14 1.997m ± 15% 1.753m ± 8% -12.21% (p=0.001 n=10) WatchIdle/active/path/10000_files-14 28.54m ± 13% 28.13m ± 17% ~ (p=0.739 n=10) WatchIdle/active/fingerprint/100_files-14 495.4µ ± 10% 398.5µ ± 7% -19.55% (p=0.000 n=10) WatchIdle/active/fingerprint/1000_files-14 6.617m ± 14% 5.502m ± 19% -16.86% (p=0.009 n=10) WatchIdle/active/fingerprint/10000_files-14 58.58m ± 10% 58.93m ± 6% ~ (p=0.796 n=10) WatchIdle/ignored/path/100_files-14 161.9µ ± 10% 151.9µ ± 10% ~ (p=0.089 n=10) WatchIdle/ignored/path/1000_files-14 1.915m ± 7% 1.774m ± 8% -7.38% (p=0.015 n=10) WatchIdle/ignored/path/10000_files-14 25.68m ± 11% 24.15m ± 13% ~ (p=0.218 n=10) WatchIdle/ignored/fingerprint/100_files-14 431.7µ ± 4% 407.7µ ± 5% -5.54% (p=0.019 n=10) WatchIdle/ignored/fingerprint/1000_files-14 5.428m ± 21% 4.832m ± 31% -10.98% (p=0.035 n=10) WatchIdle/ignored/fingerprint/10000_files-14 58.22m ± 7% 55.88m ± 7% ~ (p=0.063 n=10) geomean 3.551m 3.203m -9.78% │ without-lastCount │ with-lastCount │ │ B/op │ B/op vs base │ GetFiles-14 1.401Mi ± 3% 1.037Mi ± 5% -25.96% (p=0.000 n=10) GetFilesWithFingerprint-14 2.117Mi ± 2% 1.755Mi ± 7% -17.11% (p=0.000 n=10) GetFilesWithFingerprintGrowing/static-14 1.611Mi ± 0% 1.275Mi ± 0% -20.86% (p=0.000 n=10) GetFilesWithFingerprintGrowing/growing-14 5.521Mi ± 0% 5.190Mi ± 0% -6.00% (p=0.000 n=10) WatchIdle/active/path/100_files-14 121.7Ki ± 0% 101.7Ki ± 0% -16.45% (p=0.000 n=10) WatchIdle/active/path/1000_files-14 1.452Mi ± 0% 1.108Mi ± 0% -23.65% (p=0.000 n=10) WatchIdle/active/path/10000_files-14 13.73Mi ± 0% 10.94Mi ± 0% -20.29% (p=0.000 n=10) WatchIdle/active/fingerprint/100_files-14 172.1Ki ± 0% 152.2Ki ± 0% -11.59% (p=0.000 n=10) WatchIdle/active/fingerprint/1000_files-14 1.964Mi ± 0% 1.621Mi ± 0% -17.48% (p=0.000 n=10) WatchIdle/active/fingerprint/10000_files-14 18.72Mi ± 0% 15.94Mi ± 0% -14.88% (p=0.000 n=10) WatchIdle/ignored/path/100_files-14 118.60Ki ± 0% 98.60Ki ± 0% -16.86% (p=0.000 n=10) WatchIdle/ignored/path/1000_files-14 1.421Mi ± 0% 1.078Mi ± 0% -24.14% (p=0.000 n=10) WatchIdle/ignored/path/10000_files-14 13.42Mi ± 0% 10.64Mi ± 0% -20.75% (p=0.000 n=10) WatchIdle/ignored/fingerprint/100_files-14 153.3Ki ± 0% 133.4Ki ± 0% -13.00% (p=0.000 n=10) WatchIdle/ignored/fingerprint/1000_files-14 1.779Mi ± 0% 1.436Mi ± 0% -19.29% (p=0.000 n=10) WatchIdle/ignored/fingerprint/10000_files-14 17.17Mi ± 2% 14.38Mi ± 2% -16.23% (p=0.000 n=10) geomean 1.677Mi 1.376Mi -17.93% │ without-lastCount │ with-lastCount │ │ allocs/op │ allocs/op vs base │ GetFiles-14 9.155k ± 0% 9.100k ± 0% -0.61% (p=0.000 n=10) GetFilesWithFingerprint-14 16.31k ± 0% 16.25k ± 0% -0.33% (p=0.009 n=10) GetFilesWithFingerprintGrowing/static-14 16.09k ± 0% 16.05k ± 0% -0.29% (p=0.000 n=10) GetFilesWithFingerprintGrowing/growing-14 18.09k ± 0% 18.05k ± 0% -0.25% (p=0.000 n=10) WatchIdle/active/path/100_files-14 1.452k ± 0% 1.433k ± 0% -1.31% (p=0.000 n=10) WatchIdle/active/path/1000_files-14 14.09k ± 0% 14.05k ± 0% -0.33% (p=0.000 n=10) WatchIdle/active/path/10000_files-14 140.3k ± 0% 140.1k ± 0% -0.10% (p=0.000 n=10) WatchIdle/active/fingerprint/100_files-14 1.956k ± 0% 1.937k ± 0% -0.97% (p=0.000 n=10) WatchIdle/active/fingerprint/1000_files-14 19.10k ± 0% 19.05k ± 0% -0.25% (p=0.000 n=10) WatchIdle/active/fingerprint/10000_files-14 190.3k ± 0% 190.2k ± 0% -0.07% (p=0.000 n=10) WatchIdle/ignored/path/100_files-14 1.252k ± 0% 1.233k ± 0% -1.52% (p=0.000 n=10) WatchIdle/ignored/path/1000_files-14 12.09k ± 0% 12.05k ± 0% -0.38% (p=0.000 n=10) WatchIdle/ignored/path/10000_files-14 120.3k ± 0% 120.1k ± 0% -0.12% (p=0.000 n=10) WatchIdle/ignored/fingerprint/100_files-14 1.756k ± 0% 1.737k ± 0% -1.08% (p=0.000 n=10) WatchIdle/ignored/fingerprint/1000_files-14 17.10k ± 0% 17.05k ± 0% -0.27% (p=0.000 n=10) WatchIdle/ignored/fingerprint/10000_files-14 170.3k ± 0% 170.2k ± 0% -0.08% (p=0.000 n=10) geomean 15.22k 15.14k -0.50% (cherry picked from commit c05890c)
1 parent d3fa6c7 commit c58e7d8

3 files changed

Lines changed: 161 additions & 102 deletions

File tree

‎changelog/fragments/1783612408-filestream-scanner-reduce-scan-allocations.yaml‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,3 @@
1+
kind: enhancement
2+
summary: Reduce filestream scanner per-scan memory allocations for large idle file sets
3+
component: filebeat

‎filebeat/input/filestream/fswatch.go‎

Lines changed: 62 additions & 73 deletions
Original file line numberDiff line numberDiff line change
@@ -206,11 +206,12 @@ func (w *fileWatcher) watch(
206206

207207
// file identity is updated in GetFiles
208208
now := time.Now()
209-
paths, scanMetrics := w.scanner.GetFiles(loginp.FileScanOptions{
209+
scanOpts := loginp.FileScanOptions{
210210
CurrentTime: now,
211211
IgnoreOlder: ignoreOlder,
212212
IgnoreInactiveSince: ignoreInactiveSince,
213-
})
213+
}
214+
paths, scanMetrics := w.scanner.GetFiles(scanOpts)
214215
metrics.UpdateFileScanMetrics(scanMetrics)
215216

216217
// for debugging purposes
@@ -225,10 +226,6 @@ func (w *fileWatcher) watch(
225226
harvesterFiles := make([]loginp.HarvesterFile, 0, len(paths))
226227

227228
for path, fd := range paths {
228-
// srcID is the file identity, it is the same value used to identify
229-
// the harvester and as registry key for the file's state
230-
srcID := w.getFileIdentity(fd)
231-
232229
// if the scanner found a new path or an existing path
233230
// with a different file, it is a new file
234231
prevDesc, ok := w.prev[path]
@@ -239,61 +236,47 @@ func (w *fileWatcher) watch(
239236
continue
240237
}
241238

242-
// If we got notifications about harvesters being closed, update
243-
// the state accordingly.
244-
//
245-
// This is used to prevent a sort of race condition:
246-
// When the reader/harvester reaches EOF, it blocks on a backoff,
247-
// if during this time [logFile.shouldBeClosed] is called, marks the
248-
// file as inactive and closes the reader context, once the backoff
249-
// time expires the reader and harvester are closed without ingesting
250-
// any more data.
251-
//
252-
// If the [fileWatcher] sends a write event while the harvester was blocked
253-
// no new harvester is started because one is already running, however the
254-
// [fileWatcher] updates its internal state and won't send write events until
255-
// more data is added to the file.
256-
//
257-
// This can cause some lines to be missed because the harvester closed
258-
// and the write event was lost.
259-
//
260-
// To prevent this from happening we get notified the offset of the file
261-
// (data ingested) when the harvester closes. If we have this data we
262-
// update our state to the same as the harvester, therefore starting
263-
// a new harvester if needed.
264-
w.closedHarvestersMutex.Lock()
265-
if size, harvesterClosed := w.closedHarvesters[srcID]; harvesterClosed {
266-
w.log.Debugf("Updating previous state because harvester was closed. '%s': %d", srcID, size)
267-
prevDesc.SetBytesIngested(size)
239+
// srcID is the file identity (harvester ID/registry key), resolved lazily via ensureSrcID:
240+
// an unchanged, untracked file (gzip, empty, ignore_older) never needs one, saving allocs.
241+
var srcID string
242+
ensureSrcID := func() string {
243+
if srcID == "" { // getFileIdentity never returns ""
244+
srcID = w.getFileIdentity(fd)
245+
}
246+
return srcID
247+
}
248+
249+
// closedHarvesters is empty in the steady state; this reconciliation is usually skipped.
250+
if w.hasClosedHarvesters() {
251+
w.reconcileClosedHarvester(&prevDesc, ensureSrcID())
268252
}
269-
w.closedHarvestersMutex.Unlock()
270253

271254
var e loginp.FSEvent
272255
switch {
273256
// the new size is smaller, the file was truncated
274257
case prevDesc.Info.Size() > fd.Info.Size():
275-
e = truncateEvent(path, fd, srcID)
258+
e = truncateEvent(path, fd, ensureSrcID())
276259
truncatedCount++
277260

278261
// the size is the same, timestamps are different, the file was touched
279262
case prevDesc.Info.Size() == fd.Info.Size() && prevDesc.Info.ModTime() != fd.Info.ModTime():
280263
if w.cfg.ResendOnModTime {
281-
e = truncateEvent(path, fd, srcID)
264+
e = truncateEvent(path, fd, ensureSrcID())
282265
truncatedCount++
283266
}
284267

285268
// the new size is larger, something was written.
286269
// If a harvester for this file was closed recently,
287270
// we use its state instead of the one we have cached.
288271
case prevDesc.SizeOrBytesIngested() < fd.Info.Size():
289-
e = writeEvent(path, fd, srcID)
272+
e = writeEvent(path, fd, ensureSrcID())
290273
writtenCount++
291274

292275
default:
293276
// For the delete feature we need to run the harvester for
294277
// files that have not changed until they're deleted.
295278
if w.cfg.SendNotChanged {
296-
e = notChangedEvent(path, fd, srcID)
279+
e = notChangedEvent(path, fd, ensureSrcID())
297280
}
298281
}
299282

@@ -306,17 +289,13 @@ func (w *fileWatcher) watch(
306289
}
307290
}
308291

309-
// We want progress metrics for all files except truncated ones
310-
if e.Op != loginp.OpTruncate {
311-
harvesterFiles = appendHarvesterFile(harvesterFiles, fd, srcID, now, ignoreOlder, ignoreInactiveSince)
292+
// Record progress metrics for trackable, non-truncated files (tracksHarvesterProgress).
293+
if e.Op != loginp.OpTruncate && tracksHarvesterProgress(&fd, scanOpts) {
294+
harvesterFiles = append(harvesterFiles, loginp.HarvesterFile{ID: ensureSrcID(), Size: fd.Info.Size()})
312295
}
313296

314297
// delete from previous state to mark that we've seen the existing file again
315298
delete(w.prev, path)
316-
// Delete used state from closedHarvesters
317-
w.closedHarvestersMutex.Lock()
318-
delete(w.closedHarvesters, srcID)
319-
w.closedHarvestersMutex.Unlock()
320299
}
321300

322301
// Remaining files in the prev map are missing from this scan — either
@@ -446,9 +425,10 @@ func (w *fileWatcher) watch(
446425
createdCount++
447426
}
448427

449-
// The previous for loop has an early continue for new files,
450-
// so we need to collect their metrics here.
451-
harvesterFiles = appendHarvesterFile(harvesterFiles, *fd, srcID, now, ignoreOlder, ignoreInactiveSince)
428+
// New files skip the main loop via early continue, so collect their metrics here.
429+
if tracksHarvesterProgress(fd, scanOpts) {
430+
harvesterFiles = append(harvesterFiles, loginp.HarvesterFile{ID: srcID, Size: fd.Info.Size()})
431+
}
452432
}
453433

454434
w.log.Debugw("File scan complete",
@@ -491,27 +471,31 @@ func (w *fileWatcher) watch(
491471
w.prev = paths
492472
}
493473

494-
func appendHarvesterFile(
495-
files []loginp.HarvesterFile,
496-
fd loginp.FileDescriptor,
497-
srcID string,
498-
now time.Time,
499-
ignoreOlder time.Duration,
500-
ignoreInactiveSince time.Time,
501-
) []loginp.HarvesterFile {
502-
opts := loginp.FileScanOptions{
503-
CurrentTime: now,
504-
IgnoreOlder: ignoreOlder,
505-
IgnoreInactiveSince: ignoreInactiveSince,
506-
}
507-
if fd.GZIP || fd.Info.Size() <= 0 || isFileIgnored(fd, opts) {
508-
return files
474+
// hasClosedHarvesters reports whether any harvester-close notification awaits reconciliation.
475+
func (w *fileWatcher) hasClosedHarvesters() bool {
476+
w.closedHarvestersMutex.Lock()
477+
defer w.closedHarvestersMutex.Unlock()
478+
return len(w.closedHarvesters) > 0
479+
}
480+
481+
// reconcileClosedHarvester folds a recently-closed harvester's ingested offset (from
482+
// closedHarvesters) into prevDesc so a restarted harvester resumes from the right position.
483+
// It guards a close-during-backoff race that would otherwise withhold writes and lose lines.
484+
func (w *fileWatcher) reconcileClosedHarvester(prevDesc *loginp.FileDescriptor, id string) {
485+
w.closedHarvestersMutex.Lock()
486+
defer w.closedHarvestersMutex.Unlock()
487+
size, ok := w.closedHarvesters[id]
488+
if !ok {
489+
return
509490
}
491+
w.log.Debugf("Updating previous state because harvester was closed. '%s': %d", id, size)
492+
prevDesc.SetBytesIngested(size)
493+
delete(w.closedHarvesters, id)
494+
}
510495

511-
return append(files, loginp.HarvesterFile{
512-
ID: srcID,
513-
Size: fd.Info.Size(),
514-
})
496+
// tracksHarvesterProgress reports whether a file contributes to the harvester progress metrics.
497+
func tracksHarvesterProgress(fd *loginp.FileDescriptor, opts loginp.FileScanOptions) bool {
498+
return !fd.GZIP && fd.Info.Size() > 0 && !isFileIgnored(*fd, opts)
515499
}
516500

517501
// isFileIgnored returns true when a file is ignored, no matter the reason.
@@ -569,10 +553,10 @@ func (w *fileWatcher) Event() loginp.FSEvent {
569553
}
570554

571555
// GetFiles runs a one-off enumeration scan for the prospector's Init and
572-
// TakeOver phases. It is side-effect free: unlike the watch loop it does not
573-
// advance the scanner's completedFingerprints set, so these pre-watch scans
574-
// cannot suppress the bridging raw header a still-growing entry needs to
575-
// migrate its registry key after a restart.
556+
// TakeOver phases. Unlike the watch loop it does not advance the scanner's
557+
// completedFingerprints set, so these pre-watch scans cannot suppress the
558+
// bridging raw header a still-growing entry needs to migrate its registry key
559+
// after a restart.
576560
func (w *fileWatcher) GetFiles(opts loginp.FileScanOptions) (map[string]loginp.FileDescriptor, loginp.FileScanMetrics) {
577561
return w.scanner.GetFiles(opts)
578562
}
@@ -635,6 +619,9 @@ type fileScanner struct {
635619
// prospector runs in Init/TakeOver cannot suppress the header a
636620
// still-growing entry needs to migrate across a restart.
637621
completedFingerprints map[string]struct{}
622+
623+
// lastCount is the number of unique files the previous scan produced.
624+
lastCount int
638625
}
639626

640627
func newFileScanner(logger *logp.Logger, paths []string, config fileScannerConfig, compression string) (*fileScanner, error) {
@@ -717,11 +704,12 @@ func (s *fileScanner) GetFiles(opts loginp.FileScanOptions) (map[string]loginp.F
717704
opts.CurrentTime = time.Now()
718705
}
719706

720-
fdByName := map[string]loginp.FileDescriptor{}
707+
// Pre-size the per-scan maps from the previous scan's count.
708+
fdByName := make(map[string]loginp.FileDescriptor, s.lastCount)
721709
// used to determine if a symlink resolves in a already known target
722-
uniqueIDs := map[string]string{}
710+
uniqueIDs := make(map[string]string, s.lastCount)
723711
// used to filter out duplicate matches
724-
uniqueFiles := map[string]struct{}{}
712+
uniqueFiles := make(map[string]struct{}, s.lastCount)
725713
scanMetrics := loginp.FileScanMetrics{}
726714

727715
for _, path := range s.paths {
@@ -792,6 +780,7 @@ func (s *fileScanner) GetFiles(opts loginp.FileScanOptions) (map[string]loginp.F
792780
}
793781

794782
scanMetrics.FilesUnique = int64(len(fdByName))
783+
s.lastCount = len(fdByName)
795784
return fdByName, scanMetrics
796785
}
797786

‎filebeat/input/filestream/fswatch_test.go‎

Lines changed: 96 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -1560,17 +1560,20 @@ func (s *testFileScanner) GetFiles(loginp.FileScanOptions) (map[string]loginp.Fi
15601560

15611561
const benchmarkFileCount = 1000
15621562

1563-
func BenchmarkGetFiles(b *testing.B) {
1564-
dir := b.TempDir()
1565-
basenameFormat := "file-%d.log"
1566-
1567-
for i := 0; i < benchmarkFileCount; i++ {
1568-
filename := filepath.Join(dir, fmt.Sprintf(basenameFormat, i))
1563+
// writeBenchmarkFiles creates n log files under dir and returns the glob that matches them.
1564+
func writeBenchmarkFiles(tb testing.TB, dir string, n int) []string {
1565+
tb.Helper()
1566+
for i := range n {
1567+
filename := filepath.Join(dir, fmt.Sprintf("file-%d.log", i))
15691568
content := fmt.Sprintf("content-%d\n", i)
1570-
err := os.WriteFile(filename, []byte(strings.Repeat(content, 1024)), 0777)
1571-
require.NoError(b, err)
1569+
require.NoError(tb, os.WriteFile(
1570+
filename, []byte(strings.Repeat(content, 1024)), 0o644))
15721571
}
1573-
paths := []string{filepath.Join(dir, "*.log")}
1572+
return []string{filepath.Join(dir, "*.log")}
1573+
}
1574+
1575+
func BenchmarkGetFiles(b *testing.B) {
1576+
paths := writeBenchmarkFiles(b, b.TempDir(), benchmarkFileCount)
15741577
cfg := fileScannerConfig{
15751578
Fingerprint: fingerprintConfig{
15761579
Enabled: false,
@@ -1586,16 +1589,7 @@ func BenchmarkGetFiles(b *testing.B) {
15861589
}
15871590

15881591
func BenchmarkGetFilesWithFingerprint(b *testing.B) {
1589-
dir := b.TempDir()
1590-
basenameFormat := "file-%d.log"
1591-
1592-
for i := 0; i < benchmarkFileCount; i++ {
1593-
filename := filepath.Join(dir, fmt.Sprintf(basenameFormat, i))
1594-
content := fmt.Sprintf("content-%d\n", i)
1595-
err := os.WriteFile(filename, []byte(strings.Repeat(content, 1024)), 0777)
1596-
require.NoError(b, err)
1597-
}
1598-
paths := []string{filepath.Join(dir, "*.log")}
1592+
paths := writeBenchmarkFiles(b, b.TempDir(), benchmarkFileCount)
15991593
cfg := fileScannerConfig{
16001594
Fingerprint: fingerprintConfig{
16011595
Enabled: true,
@@ -1637,16 +1631,7 @@ func BenchmarkGetFilesWithFingerprintGrowing(b *testing.B) {
16371631
}
16381632
for _, tc := range cases {
16391633
b.Run(tc.name, func(b *testing.B) {
1640-
dir := b.TempDir()
1641-
for i := range benchmarkFileCount {
1642-
filename := filepath.Join(dir, fmt.Sprintf("file-%d.log", i))
1643-
content := fmt.Sprintf("content-%d\n", i)
1644-
// ~10KB per file: well above the 1024-byte threshold, so every
1645-
// file is fingerprinted with a final SHA-256 on every scan.
1646-
err := os.WriteFile(filename, []byte(strings.Repeat(content, 1024)), 0777)
1647-
require.NoError(b, err)
1648-
}
1649-
paths := []string{filepath.Join(dir, "*.log")}
1634+
paths := writeBenchmarkFiles(b, b.TempDir(), benchmarkFileCount)
16501635
cfg := fileScannerConfig{
16511636
Fingerprint: fingerprintConfig{
16521637
Enabled: true,
@@ -1667,6 +1652,88 @@ func BenchmarkGetFilesWithFingerprintGrowing(b *testing.B) {
16671652
}
16681653
}
16691654

1655+
// BenchmarkWatchIdle measures fileWatcher.watch over an unchanged file set: the mostly-idle steady
1656+
// state where allocs/op is the cost of observing nothing changed (identity cache + pre-sized maps).
1657+
func BenchmarkWatchIdle(b *testing.B) {
1658+
states := []struct {
1659+
name string
1660+
ignoreOlder time.Duration
1661+
aged bool
1662+
}{
1663+
{"active", 0, false},
1664+
{"ignored", time.Hour, true},
1665+
}
1666+
identities := []struct {
1667+
name string
1668+
fingerprint bool
1669+
newID func() fileIdentifier
1670+
}{
1671+
{"path", false, func() fileIdentifier { return mustPathIdentifier(false) }},
1672+
{"fingerprint", true, func() fileIdentifier {
1673+
fi, _ := newFingerprintIdentifier(nil, nil)
1674+
return fi
1675+
}},
1676+
}
1677+
1678+
for _, st := range states {
1679+
for _, id := range identities {
1680+
for _, fileCount := range []int{100, 1000, 10000} {
1681+
b.Run(fmt.Sprintf("%s/%s/%d_files", st.name, id.name, fileCount), func(b *testing.B) {
1682+
paths := writeBenchmarkFiles(b, b.TempDir(), fileCount)
1683+
1684+
if st.aged {
1685+
// Age every file well past ignoreOlder so it is excluded.
1686+
old := time.Now().Add(-48 * time.Hour)
1687+
matches, err := filepath.Glob(paths[0])
1688+
require.NoError(b, err)
1689+
for _, m := range matches {
1690+
require.NoError(b, os.Chtimes(m, old, old))
1691+
}
1692+
}
1693+
1694+
cfg := defaultFileWatcherConfig()
1695+
cfg.Scanner.Fingerprint.Enabled = id.fingerprint
1696+
cfg.Scanner.Fingerprint.Growing = id.fingerprint
1697+
1698+
fw, err := newFileWatcher(
1699+
logp.NewNopLogger(),
1700+
paths,
1701+
cfg,
1702+
CompressionNone,
1703+
false,
1704+
id.newID(),
1705+
mustSourceIdentifier("bench-id"),
1706+
)
1707+
require.NoError(b, err)
1708+
1709+
// A create event is emitted for every file on the first scan; idle rescans emit
1710+
// nothing. Drain in the background so watch's sends never block.
1711+
ctx := b.Context()
1712+
go func() {
1713+
for {
1714+
select {
1715+
case <-ctx.Done():
1716+
return
1717+
case <-fw.events:
1718+
}
1719+
}
1720+
}()
1721+
1722+
metrics := newTestMetrics()
1723+
// Prime w.prev so the benchmarked scans are pure steady state.
1724+
fw.watch(ctx, metrics, st.ignoreOlder, time.Time{})
1725+
1726+
b.ReportAllocs()
1727+
b.ResetTimer()
1728+
for b.Loop() {
1729+
fw.watch(ctx, metrics, st.ignoreOlder, time.Time{})
1730+
}
1731+
})
1732+
}
1733+
}
1734+
}
1735+
}
1736+
16701737
func createWatcherWithConfig(t *testing.T, logger *logp.Logger, paths []string, cfgStr string) *fileWatcher {
16711738
tmpCfg := struct {
16721739
Scaner fileWatcherConfig `config:"scanner"`

0 commit comments

Comments
 (0)