diff --git a/pkg/sources/filesystem/filesystem.go b/pkg/sources/filesystem/filesystem.go index b56d3bc67..e97592571 100644 --- a/pkg/sources/filesystem/filesystem.go +++ b/pkg/sources/filesystem/filesystem.go @@ -83,11 +83,14 @@ func (s *Source) Init(aCtx trContext.Context, name string, jobId sources.JobID, if err != nil { return fmt.Errorf("unable to create filter: %w", err) } + s.filter = filter + err = s.setMaxSymlinkDepth(&conn) if err != nil { return err } + return nil } @@ -100,7 +103,9 @@ func (s *Source) setMaxSymlinkDepth(conn *sourcespb.Filesystem) error { defaultMaxSymlinkDepth, ) } + s.maxSymlinkDepth = depth + return nil } @@ -133,28 +138,31 @@ func (s *Source) Chunks(ctx trContext.Context, chunksChan chan *sources.Chunk, _ continue } + initialDepth := 0 + if fileInfo.Mode()&os.ModeSymlink != 0 { - // if the root path is a symlink we scan the symlink lctx.Logger().V(5).Info("Root path is a symlink") - initialDepth := 0 - err = s.scanSymlink(ctx, chunksChan, rootPath, initialDepth, cleanRootPath) + if err := s.scanSymlink(ctx, chunksChan, rootPath, initialDepth, cleanRootPath); err != nil { + lctx.Logger().Error(err, "error scanning root path symlink") + } + s.ClearEncodedResumeInfoFor(rootPath) } else if fileInfo.IsDir() { lctx.Logger().V(5).Info("Root path is a dir") - initialDepth := 0 - err = s.scanDir(ctx, chunksChan, rootPath, initialDepth, cleanRootPath) - s.ClearEncodedResumeInfoFor(rootPath) - } else { - if !fileInfo.Mode().IsRegular() { - lctx.Logger().V(2).Info("Root path is a non-regular file; skipping") - continue + if err := s.scanDir(ctx, chunksChan, rootPath, initialDepth, cleanRootPath); err != nil { + lctx.Logger().Error(err, "error scanning root path directory") } - lctx.Logger().V(5).Info("Root path is a file") - err = s.scanFile(ctx, chunksChan, cleanRootPath) - } - if err != nil && !errors.Is(err, io.EOF) { - lctx.Logger().Error(err, "error scanning filesystem") + s.ClearEncodedResumeInfoFor(rootPath) + } else if !fileInfo.Mode().IsRegular() { + lctx.Logger().V(2).Info("Root path is a non-regular file; skipping") + } else { + lctx.Logger().V(5).Info("Root path is a file") + if err := s.scanFile(ctx, chunksChan, cleanRootPath); err != nil { + if !errors.Is(err, io.EOF) { + lctx.Logger().Error(err, "error scanning filesystem") + } + } } } @@ -176,7 +184,6 @@ func (s *Source) scanSymlink( lctx.Logger().V(5).Info("scanSymlink") if !s.canFollowSymlinks() { - // If the file or directory is a symlink but the followSymlinks is disable ignore the path lctx.Logger().V(2).Info("Path is a symlink, but following symlinks is not allowed; skipping") return nil } @@ -259,8 +266,6 @@ func (s *Source) scanDir( lctx.Logger().V(5).Info("scanDir") - // check if the full path is not matching any pattern in include - // FilterRuleSet and matching any exclude FilterRuleSet. if s.filter != nil && s.filter.ShouldExclude(path) { lctx.Logger().V(2).Info("Path was filtered out by filter.ShouldExclude; skipping") return nil @@ -379,7 +384,6 @@ func (s *Source) scanDir( func (s *Source) scanFile(ctx trContext.Context, chunksChan chan *sources.Chunk, path string) error { ctx.Logger().V(5).Info("scanFile") - // Check if file is binary and should be skipped if (s.skipBinaries || feature.ForceSkipBinaries.Load()) && common.IsBinary(path) { ctx.Logger().V(5).Info("File is binary; skipping") return nil @@ -454,31 +458,35 @@ func (s *Source) ChunkUnit(ctx trContext.Context, unit sources.SourceUnit, repor } ch := make(chan *sources.Chunk) - var scanErr error + var reportErr error + initialDepth := 0 + go func() { defer close(ch) if fileInfo.Mode()&os.ModeSymlink != 0 { - // if the root path is a symlink we scan the symlink lctx.Logger().V(5).Info("Root unit is a symlink") - initialDepth := 0 - scanErr = s.scanSymlink(ctx, ch, rootPath, initialDepth, cleanPath) + if err := s.scanSymlink(ctx, ch, rootPath, initialDepth, cleanPath); err != nil { + lctx.Logger().Error(err, "error scanning root unit symlink") + reportErr = reporter.ChunkErr(ctx, err) + } + s.ClearEncodedResumeInfoFor(rootPath) } else if fileInfo.IsDir() { lctx.Logger().V(5).Info("Root unit is a directory") - initialDepth := 0 - // TODO: Finer grain error tracking of individual chunks. - scanErr = s.scanDir(ctx, ch, rootPath, initialDepth, cleanPath) - s.ClearEncodedResumeInfoFor(rootPath) - } else { - // TODO: Finer grain error tracking of individual - // chunks (in the case of archives). - if !fileInfo.Mode().IsRegular() { - lctx.Logger().V(2).Info("Root unit is a non-regular file; skipping") - return + if err := s.scanDir(ctx, ch, rootPath, initialDepth, cleanPath); err != nil { + lctx.Logger().Error(err, "error scanning root unit directory") + reportErr = reporter.ChunkErr(ctx, err) } + s.ClearEncodedResumeInfoFor(rootPath) + } else if !fileInfo.Mode().IsRegular() { + lctx.Logger().V(2).Info("Root unit is a non-regular file; skipping") + } else { lctx.Logger().V(5).Info("Root unit is a file") - scanErr = s.scanFile(ctx, ch, cleanPath) + if err := s.scanFile(ctx, ch, cleanPath); err != nil && !errors.Is(err, io.EOF) { + lctx.Logger().Error(err, "error scanning root unit file") + reportErr = reporter.ChunkErr(ctx, err) + } } }() @@ -492,10 +500,5 @@ func (s *Source) ChunkUnit(ctx trContext.Context, unit sources.SourceUnit, repor } } - if scanErr != nil && !errors.Is(scanErr, io.EOF) { - lctx.Logger().Error(scanErr, "error scanning filesystem") - return reporter.ChunkErr(ctx, scanErr) - } - - return nil + return reportErr }