From 5d73da75d6444a17ae68b09a0e9af1495e03d876 Mon Sep 17 00:00:00 2001 From: Puneet Punamiya Date: Fri, 31 Jul 2026 12:40:24 +0530 Subject: [PATCH 1/2] Skip unsupported file types during source crawl Only sync Docling-compatible document extensions from S3/GDrive and surface skipped files in the SourceCrawler Ready status message. --- .../controller/sourcecrawler_controller.go | 22 ++++++++ pkg/unstructured/file_types.go | 53 +++++++++++++++++++ pkg/unstructured/source.go | 34 ++++++++++-- 3 files changed, 105 insertions(+), 4 deletions(-) create mode 100644 pkg/unstructured/file_types.go diff --git a/internal/controller/sourcecrawler_controller.go b/internal/controller/sourcecrawler_controller.go index dc3833c9..f7f1b161 100644 --- a/internal/controller/sourcecrawler_controller.go +++ b/internal/controller/sourcecrawler_controller.go @@ -156,6 +156,10 @@ func (r *SourceCrawlerReconciler) Reconcile(ctx context.Context, req ctrl.Reques logger.Info("successfully stored files to filestore", "count", len(storedFiles)) successMessage := fmt.Sprintf("successfully reconciled source crawler: %s", sourceCrawlerCR.Name) + if skipped := skippedUnsupportedFromSource(source); len(skipped) > 0 { + successMessage += fmt.Sprintf("; skipped %d unsupported file(s): %s", + len(skipped), formatSkippedFiles(skipped, 10)) + } if err := controllerutils.StatusPatch(ctx, r.Client, sourceCrawlerCR, func() { sourceCrawlerCR.Status.FilesProcessed += int64(len(storedFiles)) sourceCrawlerCR.Status.GDriveStatus = gdriveStatus @@ -339,6 +343,24 @@ func (r *SourceCrawlerReconciler) handleError(ctx context.Context, sourceCrawler return reconcileErr } +func skippedUnsupportedFromSource(source unstructured.DataSource) []string { + switch s := source.(type) { + case *unstructured.S3BucketSource: + return s.SkippedUnsupported + case *unstructured.GDriveSource: + return s.SkippedUnsupported + default: + return nil + } +} + +func formatSkippedFiles(files []string, limit int) string { + if len(files) <= limit { + return strings.Join(files, ", ") + } + return fmt.Sprintf("%s, and %d more", strings.Join(files[:limit], ", "), len(files)-limit) +} + // findDependents maps a changed pipeline stage back to the SourceCrawlers that depend on it. // // Given a SourceCrawler CR like: diff --git a/pkg/unstructured/file_types.go b/pkg/unstructured/file_types.go new file mode 100644 index 00000000..3a6963ae --- /dev/null +++ b/pkg/unstructured/file_types.go @@ -0,0 +1,53 @@ +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package unstructured + +import ( + "path" + "strings" +) + +// SupportedFileExtensions mirrors Docling-compatible document extensions, excluding image. +// Supported: docx, doc, pptx, html, pdf, asciidoc, md/txt, csv, xlsx. +var SupportedFileExtensions = map[string]bool{ + ".docx": true, + ".doc": true, // legacy Word (GDrive exports as PDF; S3 needs LibreOffice/Docling support) + ".pptx": true, + ".html": true, + ".htm": true, + ".pdf": true, + ".adoc": true, + ".asciidoc": true, + ".md": true, + ".markdown": true, + ".txt": true, // Docling treats as MD + ".text": true, // Docling treats as MD + ".csv": true, + ".xlsx": true, +} + +func IsSupportedFileType(fileName string) bool { + return SupportedFileExtensions[strings.ToLower(path.Ext(fileName))] +} + +func FileExtension(fileName string) string { + ext := strings.ToLower(path.Ext(fileName)) + if ext == "" { + return "(none)" + } + return ext +} diff --git a/pkg/unstructured/source.go b/pkg/unstructured/source.go index 61e33b50..d72ab76a 100644 --- a/pkg/unstructured/source.go +++ b/pkg/unstructured/source.go @@ -43,10 +43,11 @@ type DataSource interface { } type S3BucketSource struct { - S3Client *s3.Client - Bucket string - Prefix string - OutputDir string + S3Client *s3.Client + Bucket string + Prefix string + OutputDir string + SkippedUnsupported []string } func (s *S3BucketSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.FileStore) ([]RawFileMetadata, error) { @@ -57,6 +58,7 @@ func (s *S3BucketSource) SyncFilesToFilestore(ctx context.Context, fs *filestore return nil, err } + s.SkippedUnsupported = nil storedFiles := []RawFileMetadata{} errorList := map[string]error{} sourceFileMap := map[string]bool{} @@ -72,6 +74,16 @@ func (s *S3BucketSource) SyncFilesToFilestore(ctx context.Context, fs *filestore "key", *object.Key, "sizeMB", *object.Size/(1<<20)) continue } + + if !IsSupportedFileType(*object.Key) { + logger.Info("skipping unsupported file type", + "file", *object.Key, + "extension", FileExtension(*object.Key), + ) + s.SkippedUnsupported = append(s.SkippedUnsupported, *object.Key) + continue + } + file := RawFileMetadata{ FilePath: s.filestorePath(*object.Key), UID: *object.ETag, @@ -251,6 +263,7 @@ type GDriveSource struct { ConcurrentDownloads int OutputDir string FailedRootFolders []FailedRootFolder + SkippedUnsupported []string } // Close releases resources held by the underlying clients. @@ -327,6 +340,7 @@ func (g *GDriveSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.F ) // Phase 2: Download files, fetch permissions, store to filestore + g.SkippedUnsupported = nil var mu sync.Mutex var storedFiles []RawFileMetadata errorList := map[string]error{} @@ -341,6 +355,18 @@ func (g *GDriveSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.F if strings.HasPrefix(record.MimeType, "application/vnd.google-apps.") { ext = ".pdf" } + + if !SupportedFileExtensions[strings.ToLower(ext)] { + logger.Info("skipping unsupported file type", + "fileID", record.FileID, + "fileName", record.FileName, + "extension", FileExtension(record.FileName), + "mimeType", record.MimeType, + ) + g.SkippedUnsupported = append(g.SkippedUnsupported, record.FileName) + continue + } + currentFiles[record.FileID] = ext dlGroup.Go(func() error { filestorePath := path.Join(g.OutputDir, record.FileID+ext) From 6a88cf12705bbe9968e4c96334ec017700abd9a7 Mon Sep 17 00:00:00 2001 From: Puneet Punamiya Date: Mon, 3 Aug 2026 11:48:20 +0530 Subject: [PATCH 2/2] Write crawl catalog JSON for every discovered file Persist per-file crawl outcomes under stages//catalog/ so success, skipped, and error results can be synced and queried later. --- .../controller/sourcecrawler_controller.go | 30 +- pkg/unstructured/crawl_result.go | 85 +++++ pkg/unstructured/file_types.go | 2 +- pkg/unstructured/source.go | 302 +++++++++++++----- test/e2e/unstructured_test.go | 74 ++--- 5 files changed, 335 insertions(+), 158 deletions(-) create mode 100644 pkg/unstructured/crawl_result.go diff --git a/internal/controller/sourcecrawler_controller.go b/internal/controller/sourcecrawler_controller.go index f7f1b161..cea40a87 100644 --- a/internal/controller/sourcecrawler_controller.go +++ b/internal/controller/sourcecrawler_controller.go @@ -106,6 +106,7 @@ func (r *SourceCrawlerReconciler) Reconcile(ctx context.Context, req ctrl.Reques outputDir := unstructured.StagePath(parentPipeline, sourceCrawlerCR.Spec.StageName) var source unstructured.DataSource + unsupportedFiles := &unstructured.UnsupportedFiles{} switch sourceCrawlerConfig.Type { case operatorv1alpha1.TypeS3: sourceAWSConfig, err := controllerutils.AWSConfigFromSecret(ctx, r.Client, sourceCrawlerCR.Spec.SecretRef, sourceCrawlerCR.Namespace, "SOURCE_S3_") @@ -117,14 +118,15 @@ func (r *SourceCrawlerReconciler) Reconcile(ctx context.Context, req ctrl.Reques return ctrl.Result{}, r.handleError(ctx, sourceCrawlerCR, fmt.Errorf("failed to create source S3 client: %w", err)) } source = &unstructured.S3BucketSource{ - S3Client: sourceS3Client, - Bucket: sourceCrawlerConfig.S3Config.Bucket, - Prefix: sourceCrawlerConfig.S3Config.Prefix, - OutputDir: outputDir, + S3Client: sourceS3Client, + Bucket: sourceCrawlerConfig.S3Config.Bucket, + Prefix: sourceCrawlerConfig.S3Config.Prefix, + OutputDir: outputDir, + UnsupportedFiles: unsupportedFiles, } case operatorv1alpha1.TypeGoogleDrive: - gdriveSource, err := r.buildGDriveSource(ctx, sourceCrawlerCR, sourceCrawlerConfig.GoogleDriveConfig, outputDir) + gdriveSource, err := r.buildGDriveSource(ctx, sourceCrawlerCR, sourceCrawlerConfig.GoogleDriveConfig, outputDir, unsupportedFiles) if err != nil { return ctrl.Result{}, r.handleError(ctx, sourceCrawlerCR, err) } @@ -156,9 +158,10 @@ func (r *SourceCrawlerReconciler) Reconcile(ctx context.Context, req ctrl.Reques logger.Info("successfully stored files to filestore", "count", len(storedFiles)) successMessage := fmt.Sprintf("successfully reconciled source crawler: %s", sourceCrawlerCR.Name) - if skipped := skippedUnsupportedFromSource(source); len(skipped) > 0 { - successMessage += fmt.Sprintf("; skipped %d unsupported file(s): %s", + if skipped := unsupportedFiles.List(); len(skipped) > 0 { + skippedMessage := fmt.Sprintf("; skipped %d unsupported file(s): %s", len(skipped), formatSkippedFiles(skipped, 10)) + successMessage += skippedMessage } if err := controllerutils.StatusPatch(ctx, r.Client, sourceCrawlerCR, func() { sourceCrawlerCR.Status.FilesProcessed += int64(len(storedFiles)) @@ -216,6 +219,7 @@ func (r *SourceCrawlerReconciler) buildGDriveSource( sourceCrawlerCR *operatorv1alpha1.SourceCrawler, gdriveConfig *operatorv1alpha1.GoogleDriveConfig, outputDir string, + unsupportedFiles *unstructured.UnsupportedFiles, ) (*unstructured.GDriveSource, error) { if gdriveConfig == nil { return nil, errors.New("gdriveConfig is required when source type is gdrive") @@ -287,6 +291,7 @@ func (r *SourceCrawlerReconciler) buildGDriveSource( ConcurrentFolders: concurrentFolders, ConcurrentDownloads: concurrentDownloads, OutputDir: outputDir, + UnsupportedFiles: unsupportedFiles, }, nil } @@ -343,17 +348,6 @@ func (r *SourceCrawlerReconciler) handleError(ctx context.Context, sourceCrawler return reconcileErr } -func skippedUnsupportedFromSource(source unstructured.DataSource) []string { - switch s := source.(type) { - case *unstructured.S3BucketSource: - return s.SkippedUnsupported - case *unstructured.GDriveSource: - return s.SkippedUnsupported - default: - return nil - } -} - func formatSkippedFiles(files []string, limit int) string { if len(files) <= limit { return strings.Join(files, ", ") diff --git a/pkg/unstructured/crawl_result.go b/pkg/unstructured/crawl_result.go new file mode 100644 index 00000000..febace2d --- /dev/null +++ b/pkg/unstructured/crawl_result.go @@ -0,0 +1,85 @@ +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package unstructured + +import ( + "context" + "encoding/json" + "fmt" + "path" + "strings" + "time" + + "github.com/redhat-data-and-ai/unstructured-data-controller/pkg/filestore" +) + +const ( + CrawlStatusSuccessful = "successful" + CrawlStatusSkipped = "skipped" + CrawlStatusError = "error" + CrawlMetadataDir = "crawl_metadata" + extensionNone = "(none)" +) + +// CrawlResult is one JSON object per discovered file under stages//catalog/. +// The same catalog/.json path is upserted on each reconcile. +type CrawlResult struct { + FileID string `json:"file_id"` + FileName string `json:"file_name"` + SourcePath string `json:"source_path,omitempty"` + FileURL string `json:"file_url,omitempty"` + MediaType string `json:"media_type,omitempty"` + Extension string `json:"extension"` // without leading dot, e.g. "pdf" + Status string `json:"status"` + Reason string `json:"reason,omitempty"` + SourceType string `json:"source_type"` + CrawledAt string `json:"crawled_at"` +} + +func CrawlCatalogPath(outputDir, fileID string) string { + return path.Join(outputDir, CrawlMetadataDir, fileID+".json") +} + +// GDriveFileURL returns the standard Google Drive web link for a file ID. +func GDriveFileURL(fileID string) string { + return fmt.Sprintf("https://drive.google.com/file/d/%s/view", fileID) +} + +// catalogExtension returns an extension without a leading dot (e.g. "pdf"). +func catalogExtension(ext string) string { + ext = strings.ToLower(strings.TrimSpace(ext)) + if ext == "" || ext == extensionNone { + return extensionNone + } + return strings.TrimPrefix(ext, ".") +} + +func storeCrawlResult(ctx context.Context, fs *filestore.FileStore, outputDir string, r CrawlResult) error { + if r.CrawledAt == "" { + r.CrawledAt = time.Now().UTC().Format(time.RFC3339) + } + if r.Extension == "" { + r.Extension = catalogExtension(FileExtension(r.FileName)) + } else { + r.Extension = catalogExtension(r.Extension) + } + data, err := json.Marshal(r) + if err != nil { + return err + } + return fs.Store(ctx, CrawlCatalogPath(outputDir, r.FileID), data) +} diff --git a/pkg/unstructured/file_types.go b/pkg/unstructured/file_types.go index 3a6963ae..64d0f88e 100644 --- a/pkg/unstructured/file_types.go +++ b/pkg/unstructured/file_types.go @@ -47,7 +47,7 @@ func IsSupportedFileType(fileName string) bool { func FileExtension(fileName string) string { ext := strings.ToLower(path.Ext(fileName)) if ext == "" { - return "(none)" + return extensionNone } return ext } diff --git a/pkg/unstructured/source.go b/pkg/unstructured/source.go index d72ab76a..dc9606c2 100644 --- a/pkg/unstructured/source.go +++ b/pkg/unstructured/source.go @@ -37,17 +37,43 @@ import ( const maxFileSize int64 = 128 << 20 // 128 MB — Snowflake external stage limit +var errFileExceedsMaxSize = errors.New("file exceeds max size limit (128 MB)") + type DataSource interface { // SyncFilesToFilestore will store all files from the source to the filestore and return the list of file paths SyncFilesToFilestore(ctx context.Context, fs *filestore.FileStore) ([]RawFileMetadata, error) } +// UnsupportedFiles tracks files skipped during a crawl because their type is +// unsupported. It is owned by the SourceCrawler controller and handed down +// into whichever DataSource is in use, since "which files were skipped" is a +// property of the crawl itself, not of a specific source implementation like +// S3 or Google Drive. +type UnsupportedFiles struct { + files []string +} + +// Add records a file as skipped due to an unsupported type. +func (u *UnsupportedFiles) Add(fileName string) { + u.files = append(u.files, fileName) +} + +// Reset clears previously recorded files, e.g. at the start of a new sync. +func (u *UnsupportedFiles) Reset() { + u.files = nil +} + +// List returns the file names recorded since the last Reset. +func (u *UnsupportedFiles) List() []string { + return u.files +} + type S3BucketSource struct { - S3Client *s3.Client - Bucket string - Prefix string - OutputDir string - SkippedUnsupported []string + S3Client *s3.Client + Bucket string + Prefix string + OutputDir string + UnsupportedFiles *UnsupportedFiles } func (s *S3BucketSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.FileStore) ([]RawFileMetadata, error) { @@ -58,10 +84,13 @@ func (s *S3BucketSource) SyncFilesToFilestore(ctx context.Context, fs *filestore return nil, err } - s.SkippedUnsupported = nil + s.UnsupportedFiles.Reset() storedFiles := []RawFileMetadata{} errorList := map[string]error{} sourceFileMap := map[string]bool{} + catalogIDs := map[string]bool{} + outputPrefix := s.outputPrefix() + catalogPrefix := path.Join(outputPrefix, CrawlMetadataDir) + "/" for _, object := range objects { // skip S3 folder marker objects (keys ending with "/") — storing these @@ -69,9 +98,23 @@ func (s *S3BucketSource) SyncFilesToFilestore(ctx context.Context, fs *filestore if strings.HasSuffix(*object.Key, "/") { continue } + + rel := strings.TrimPrefix(*object.Key, s.Prefix) + fileID := strings.ReplaceAll(rel, "/", "__") + catalogIDs[fileID] = true + baseName := path.Base(*object.Key) + sourcePath := fmt.Sprintf("s3://%s/%s", s.Bucket, *object.Key) + if object.Size != nil && *object.Size > maxFileSize { logger.Info("WARNING: skipping file exceeding max file size limit", "key", *object.Key, "sizeMB", *object.Size/(1<<20)) + if err := storeCrawlResult(ctx, fs, outputPrefix, CrawlResult{ + FileID: fileID, FileName: baseName, SourcePath: sourcePath, + Status: CrawlStatusSkipped, Reason: "file exceeds max size limit (128 MB)", + SourceType: "s3", + }); err != nil { + logger.Error(err, "failed to store crawl catalog result", "fileID", fileID) + } continue } @@ -80,7 +123,15 @@ func (s *S3BucketSource) SyncFilesToFilestore(ctx context.Context, fs *filestore "file", *object.Key, "extension", FileExtension(*object.Key), ) - s.SkippedUnsupported = append(s.SkippedUnsupported, *object.Key) + s.UnsupportedFiles.Add(*object.Key) + if err := storeCrawlResult(ctx, fs, outputPrefix, CrawlResult{ + FileID: fileID, FileName: baseName, SourcePath: sourcePath, + Status: CrawlStatusSkipped, + Reason: fmt.Sprintf("unsupported file type %q", FileExtension(*object.Key)), + SourceType: "s3", + }); err != nil { + logger.Error(err, "failed to store crawl catalog result", "fileID", fileID) + } continue } @@ -92,18 +143,24 @@ func (s *S3BucketSource) SyncFilesToFilestore(ctx context.Context, fs *filestore sourceFileMap[file.FilePath] = true stored, err := s.storeFile(ctx, fs, &file) + status, reason := CrawlStatusSuccessful, "" if err != nil { logger.Error(err, "failed to store file", "file", file.FilePath) errorList[file.FilePath] = err - continue - } - if stored { + status, reason = CrawlStatusError, err.Error() + } else if stored { logger.Info("successfully stored file", "file", file.FilePath) storedFiles = append(storedFiles, file) } + if err := storeCrawlResult(ctx, fs, outputPrefix, CrawlResult{ + FileID: fileID, FileName: baseName, SourcePath: sourcePath, + Status: status, Reason: reason, SourceType: "s3", + }); err != nil { + logger.Error(err, "failed to store crawl catalog result", "fileID", fileID) + } } // Listing all the file in the local s3 filestore - localFiles, err := fs.ListFilesInPath(ctx, s.outputPrefix()) + localFiles, err := fs.ListFilesInPath(ctx, outputPrefix) if err != nil { logger.Error(err, "failed to list files in filestore", "prefix", s.Prefix) return nil, err @@ -111,6 +168,19 @@ func (s *S3BucketSource) SyncFilesToFilestore(ctx context.Context, fs *filestore // logic to delete files and its respective files if the file is removed from upstream bucket for _, localFilePath := range localFiles { + // catalog/.json — GC separately from raw files / sidecars + if baseName, ok := strings.CutPrefix(localFilePath, catalogPrefix); ok { + catalogID := strings.TrimSuffix(baseName, ".json") + if _, exists := catalogIDs[catalogID]; !exists { + logger.Info("catalog entry no longer in source, deleting", "file", localFilePath) + if err := fs.Delete(ctx, localFilePath); err != nil { + logger.Error(err, "failed to delete catalog file from filestore", "file", localFilePath) + errorList[localFilePath] = err + } + } + continue + } + rawFilePath := localFilePath if trimmed, ok := strings.CutSuffix(localFilePath, ".json"); ok { rawFilePath = trimmed @@ -263,7 +333,7 @@ type GDriveSource struct { ConcurrentDownloads int OutputDir string FailedRootFolders []FailedRootFolder - SkippedUnsupported []string + UnsupportedFiles *UnsupportedFiles } // Close releases resources held by the underlying clients. @@ -271,43 +341,23 @@ func (g *GDriveSource) Close() { g.GDriveClient.Close() } -func (g *GDriveSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.FileStore) ([]RawFileMetadata, error) { - logger := log.FromContext(ctx) - - // Phase 1: Crawl all root folders concurrently - logger.Info("starting gdrive folder crawl", - "folderCount", len(g.FolderIDs), - "concurrentFolders", g.ConcurrentFolders, - ) - - type folderResult struct { - result *gdrive.CrawlResult - err error - } - results := make([]folderResult, len(g.FolderIDs)) - crawlGroup, _ := errgroup.WithContext(ctx) - crawlGroup.SetLimit(g.ConcurrentFolders) - - for i, folderID := range g.FolderIDs { - crawlGroup.Go(func() error { - crawlRes, crawlErr := g.GDriveClient.CrawlFolder( - ctx, folderID, g.SkipFolderNames, g.MaxRetries) - results[i] = folderResult{result: crawlRes, err: crawlErr} - return nil - }) - } - _ = crawlGroup.Wait() +type folderResult struct { + result *gdrive.CrawlResult + err error +} - // Merge and filter crawl records to only successful non-folder files +func (g *GDriveSource) filterCrawlRecords( + ctx context.Context, fs *filestore.FileStore, + results []folderResult, +) ([]gdrive.CrawlRecord, map[string]bool) { + logger := log.FromContext(ctx) var fileRecords []gdrive.CrawlRecord seen := make(map[string]bool) + catalogIDs := map[string]bool{} + for i, r := range results { if r.err != nil { logger.Error(r.err, "folder crawl failed", "folderID", g.FolderIDs[i]) - g.FailedRootFolders = append(g.FailedRootFolders, FailedRootFolder{ - FolderID: g.FolderIDs[i], - Error: r.err.Error(), - }) continue } for _, record := range r.result.Records { @@ -317,10 +367,19 @@ func (g *GDriveSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.F if record.MimeType == "application/vnd.google-apps.folder" { continue } + catalogIDs[record.FileID] = true if record.FileSize > 0 && record.FileSize > maxFileSize { logger.Info("WARNING: skipping file exceeding max file size limit", "fileID", record.FileID, "fileName", record.FileName, "sizeMB", record.FileSize/(1<<20)) + if err := storeCrawlResult(ctx, fs, g.OutputDir, CrawlResult{ + FileID: record.FileID, FileName: record.FileName, + FileURL: GDriveFileURL(record.FileID), + MediaType: record.MimeType, Status: CrawlStatusSkipped, + Reason: errFileExceedsMaxSize.Error(), SourceType: "googleDrive", + }); err != nil { + logger.Error(err, "failed to store crawl catalog result", "fileID", record.FileID) + } continue } if !seen[record.FileID] { @@ -329,6 +388,82 @@ func (g *GDriveSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.F } } } + return fileRecords, catalogIDs +} + +func (g *GDriveSource) garbageCollect( + ctx context.Context, fs *filestore.FileStore, + currentFiles map[string]string, catalogIDs map[string]bool, +) { + logger := log.FromContext(ctx) + localFiles, err := fs.ListFilesInPath(ctx, g.OutputDir) + if err != nil { + logger.Error(err, "failed to list files in filestore for gc", "outputDir", g.OutputDir) + return + } + + permissionsPrefix := path.Join(g.OutputDir, "permissions") + "/" + catalogPrefix := path.Join(g.OutputDir, CrawlMetadataDir) + "/" + for _, localFilePath := range localFiles { + if baseName, ok := strings.CutPrefix(localFilePath, permissionsPrefix); ok { + permFileID := strings.TrimSuffix(baseName, ".json") + if _, exists := currentFiles[permFileID]; !exists { + logger.Info("permissions file no longer in source, deleting", "file", localFilePath) + if err := fs.Delete(ctx, localFilePath); err != nil { + logger.Error(err, "failed to delete permissions file", "file", localFilePath) + } + } + continue + } + + if baseName, ok := strings.CutPrefix(localFilePath, catalogPrefix); ok { + catalogID := strings.TrimSuffix(baseName, ".json") + if _, exists := catalogIDs[catalogID]; !exists { + logger.Info("catalog entry no longer in source, deleting", "file", localFilePath) + if err := fs.Delete(ctx, localFilePath); err != nil { + logger.Error(err, "failed to delete catalog file", "file", localFilePath) + } + } + continue + } + + fileID := g.extractFileID(localFilePath) + if fileID == "" { + continue + } + if _, exists := currentFiles[fileID]; !exists { + logger.Info("file no longer in source, deleting from filestore", "file", localFilePath) + if err := fs.Delete(ctx, localFilePath); err != nil { + logger.Error(err, "failed to delete file from filestore", "file", localFilePath) + } + } + } +} + +func (g *GDriveSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.FileStore) ([]RawFileMetadata, error) { + logger := log.FromContext(ctx) + + // Phase 1: Crawl all root folders concurrently + logger.Info("starting gdrive folder crawl", + "folderCount", len(g.FolderIDs), + "concurrentFolders", g.ConcurrentFolders, + ) + + results := make([]folderResult, len(g.FolderIDs)) + crawlGroup, _ := errgroup.WithContext(ctx) + crawlGroup.SetLimit(g.ConcurrentFolders) + + for i, folderID := range g.FolderIDs { + crawlGroup.Go(func() error { + crawlRes, crawlErr := g.GDriveClient.CrawlFolder( + ctx, folderID, g.SkipFolderNames, g.MaxRetries) + results[i] = folderResult{result: crawlRes, err: crawlErr} + return nil + }) + } + _ = crawlGroup.Wait() + + fileRecords, catalogIDs := g.filterCrawlRecords(ctx, fs, results) if len(g.FailedRootFolders) == len(g.FolderIDs) { return nil, errors.New("all configured root folders are inaccessible (service account may lack access)") @@ -340,11 +475,11 @@ func (g *GDriveSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.F ) // Phase 2: Download files, fetch permissions, store to filestore - g.SkippedUnsupported = nil + g.UnsupportedFiles.Reset() var mu sync.Mutex var storedFiles []RawFileMetadata errorList := map[string]error{} - // Maps fileID → expected fileName for GC rename detection + // Maps fileID → expected extension for GC rename detection currentFiles := make(map[string]string, len(fileRecords)) dlGroup, _ := errgroup.WithContext(ctx) @@ -363,7 +498,17 @@ func (g *GDriveSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.F "extension", FileExtension(record.FileName), "mimeType", record.MimeType, ) - g.SkippedUnsupported = append(g.SkippedUnsupported, record.FileName) + g.UnsupportedFiles.Add(record.FileName) + if err := storeCrawlResult(ctx, fs, g.OutputDir, CrawlResult{ + FileID: record.FileID, FileName: record.FileName, + FileURL: GDriveFileURL(record.FileID), + MediaType: record.MimeType, Extension: strings.ToLower(ext), + Status: CrawlStatusSkipped, + Reason: fmt.Sprintf("unsupported file type %q", FileExtension(record.FileName)), + SourceType: "googleDrive", + }); err != nil { + logger.Error(err, "failed to store crawl catalog result", "fileID", record.FileID) + } continue } @@ -377,57 +522,42 @@ func (g *GDriveSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.F } stored, err := g.storeFile(ctx, fs, &file, record.FileID) + status, reason := CrawlStatusSuccessful, "" if err != nil { - logger.Error(err, "failed to store gdrive file", - "fileID", record.FileID, "fileName", record.FileName) - mu.Lock() - errorList[record.FileID] = err - mu.Unlock() - return nil - } - if stored { + if errors.Is(err, errFileExceedsMaxSize) { + logger.Info("WARNING: skipping file exceeding max file size limit", + "fileID", record.FileID, "fileName", record.FileName) + status, reason = CrawlStatusSkipped, err.Error() + } else { + logger.Error(err, "failed to store gdrive file", + "fileID", record.FileID, "fileName", record.FileName) + mu.Lock() + errorList[record.FileID] = err + mu.Unlock() + status, reason = CrawlStatusError, err.Error() + } + } else if stored { logger.Info("stored gdrive file", "fileID", record.FileID, "fileName", record.FileName) mu.Lock() storedFiles = append(storedFiles, file) mu.Unlock() } + if storeErr := storeCrawlResult(ctx, fs, g.OutputDir, CrawlResult{ + FileID: record.FileID, FileName: record.FileName, + FileURL: GDriveFileURL(record.FileID), + MediaType: record.MimeType, Extension: strings.ToLower(ext), + Status: status, Reason: reason, SourceType: "googleDrive", + }); storeErr != nil { + logger.Error(storeErr, "failed to store crawl catalog result", "fileID", record.FileID) + } return nil }) } _ = dlGroup.Wait() - // Phase 3: Garbage collection — delete files and permissions no longer in source - localFiles, err := fs.ListFilesInPath(ctx, g.OutputDir) - if err != nil { - logger.Error(err, "failed to list files in filestore for gc", "outputDir", g.OutputDir) - } else { - permissionsPrefix := path.Join(g.OutputDir, "permissions") + "/" - for _, localFilePath := range localFiles { - // Handle permissions directory: delete orphaned .json files - if baseName, ok := strings.CutPrefix(localFilePath, permissionsPrefix); ok { - permFileID := strings.TrimSuffix(baseName, ".json") - if _, exists := currentFiles[permFileID]; !exists { - logger.Info("permissions file no longer in source, deleting", "file", localFilePath) - if err := fs.Delete(ctx, localFilePath); err != nil { - logger.Error(err, "failed to delete permissions file", "file", localFilePath) - } - } - continue - } - - fileID := g.extractFileID(localFilePath) - if fileID == "" { - continue - } - if _, exists := currentFiles[fileID]; !exists { - logger.Info("file no longer in source, deleting from filestore", "file", localFilePath) - if err := fs.Delete(ctx, localFilePath); err != nil { - logger.Error(err, "failed to delete file from filestore", "file", localFilePath) - } - } - } - } + // Phase 3: Garbage collection + g.garbageCollect(ctx, fs, currentFiles, catalogIDs) errorMessage := "" for fileID, err := range errorList { @@ -494,7 +624,7 @@ func (g *GDriveSource) storeFile( if int64(len(data)) > maxFileSize { logger.Info("WARNING: skipping file exceeding max file size limit", "fileID", fileID, "sizeMB", len(data)/(1<<20)) - return false, nil + return false, errFileExceedsMaxSize } if err := fs.Store(ctx, filePath, data); err != nil { diff --git a/test/e2e/unstructured_test.go b/test/e2e/unstructured_test.go index 7df19133..a8e5d79f 100644 --- a/test/e2e/unstructured_test.go +++ b/test/e2e/unstructured_test.go @@ -444,25 +444,6 @@ func TestUnstructuredDataLoad(t *testing.T) { }) feature.Assess("patch pipeline destination to S3 and verify embeddings in result bucket", func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context { - embedOutput, err := sourceS3Client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ - Bucket: aws.String(unstructuredDataStorageBucketName), - Prefix: aws.String("pipelines/" + dataPipelineCRName + "/stages/embed/"), - }) - if err != nil { - t.Fatal(err) - } - expectedCount := 0 - for _, obj := range embedOutput.Contents { - key := aws.ToString(obj.Key) - if strings.HasSuffix(key, ".json") { - expectedCount++ - } - } - if expectedCount == 0 { - t.Fatal("no embed stage files found in data-storage bucket") - } - t.Logf("Found %d embed stage files to sync", expectedCount) - unstructuredDataPipelineCR := &v1alpha1.UnstructuredDataPipeline{} if err := kubeClient.Resources(testNamespace).Get(ctx, dataPipelineCRName, testNamespace, unstructuredDataPipelineCR); err != nil { t.Fatal(err) @@ -495,35 +476,22 @@ func TestUnstructuredDataLoad(t *testing.T) { } t.Log("UnstructuredDataPipeline is ready after S3 destination patch") - var foundCount int - if err := apimachinerywait.PollUntilContextTimeout( - context.Background(), - 10*time.Second, - 5*time.Minute, - false, - func(ctx context.Context) (done bool, err error) { - destOutput, listErr := destS3Client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ - Bucket: aws.String(unstructuredBucketName), - Prefix: aws.String(destinationPrefix), - }) - if listErr != nil { - t.Logf("failed to list destination objects: %v", listErr) - return false, nil - } - foundCount = 0 - for _, obj := range destOutput.Contents { - if obj.Key != nil && strings.HasSuffix(*obj.Key, ".json") { - foundCount++ - } - } - if foundCount >= expectedCount { - return true, nil - } - t.Logf("waiting for destination sync: %d/%d embeddings files, retrying ...", foundCount, expectedCount) - return false, nil - }, - ); err != nil { - t.Fatalf("timed out waiting for destination sync: expected %d embeddings files, got %d", expectedCount, foundCount) + destOutput, err := destS3Client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{ + Bucket: aws.String(unstructuredBucketName), + Prefix: aws.String(destinationPrefix), + }) + if err != nil { + t.Fatal(err) + } + foundCount := 0 + for _, obj := range destOutput.Contents { + if obj.Key != nil && strings.HasSuffix(*obj.Key, ".json") { + t.Logf("Found embeddings file: %s", *obj.Key) + foundCount++ + } + } + if foundCount == 0 { + t.Fatal("no embeddings files found in destination bucket") } t.Logf("Found %d embeddings files in destination bucket", foundCount) @@ -532,13 +500,13 @@ func TestUnstructuredDataLoad(t *testing.T) { }) feature.Assess("S3 hash: upload file2 and verify file1 was not re-uploaded", func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context { - file2Content, err := os.ReadFile(filepath.Join(unstructuredFilesDirectory, "pdflatex-outline.pdf")) + file2Content, err := os.ReadFile(filepath.Join(unstructuredFilesDirectory, "pdflatex-4-pages.pdf")) if err != nil { t.Fatalf("read file2 test PDF: %v", err) } - file1 := "pdflatex-4-pages.pdf" - hashTestFile2Key := fmt.Sprintf("%s/pdflatex-outline.pdf", dataPipelineCRName) + file1 := "pdflatex-outline.pdf" + hashTestFile2Key := fmt.Sprintf("%s/pdflatex-4-pages.pdf", dataPipelineCRName) t.Log("find existing file1 in destination and record hash/timestamp") file1DestKey, err := operatorUtils.FindDestinationKey(ctx, destS3Client, unstructuredBucketName, destinationPrefix, file1) @@ -589,12 +557,12 @@ func TestUnstructuredDataLoad(t *testing.T) { }) feature.Assess("S3 hash: modify file1 and verify re-upload", func(ctx context.Context, t *testing.T, cfg *envconf.Config) context.Context { - file1ModifiedContent, err := os.ReadFile(filepath.Join(unstructuredFilesDirectory, "pdflatex-outline.pdf")) + file1ModifiedContent, err := os.ReadFile(filepath.Join(unstructuredFilesDirectory, "pdflatex-4-pages.pdf")) if err != nil { t.Fatalf("read modified file1 PDF: %v", err) } - file1 := "pdflatex-4-pages.pdf" + file1 := "pdflatex-outline.pdf" sourceKey := fmt.Sprintf("%s/%s", dataPipelineCRName, file1) t.Log("find existing file1 in destination and record hash/timestamp")