diff --git a/internal/controller/sourcecrawler_controller.go b/internal/controller/sourcecrawler_controller.go index dc3833c9..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,6 +158,11 @@ 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 := 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)) sourceCrawlerCR.Status.GDriveStatus = gdriveStatus @@ -212,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") @@ -283,6 +291,7 @@ func (r *SourceCrawlerReconciler) buildGDriveSource( ConcurrentFolders: concurrentFolders, ConcurrentDownloads: concurrentDownloads, OutputDir: outputDir, + UnsupportedFiles: unsupportedFiles, }, nil } @@ -339,6 +348,13 @@ func (r *SourceCrawlerReconciler) handleError(ctx context.Context, sourceCrawler return reconcileErr } +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/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 new file mode 100644 index 00000000..64d0f88e --- /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 extensionNone + } + return ext +} diff --git a/pkg/unstructured/source.go b/pkg/unstructured/source.go index 61e33b50..dc9606c2 100644 --- a/pkg/unstructured/source.go +++ b/pkg/unstructured/source.go @@ -37,16 +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 + S3Client *s3.Client + Bucket string + Prefix string + OutputDir string + UnsupportedFiles *UnsupportedFiles } func (s *S3BucketSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.FileStore) ([]RawFileMetadata, error) { @@ -57,9 +84,13 @@ func (s *S3BucketSource) SyncFilesToFilestore(ctx context.Context, fs *filestore return nil, err } + 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 @@ -67,11 +98,43 @@ 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 + } + + if !IsSupportedFileType(*object.Key) { + logger.Info("skipping unsupported file type", + "file", *object.Key, + "extension", FileExtension(*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 } + file := RawFileMetadata{ FilePath: s.filestorePath(*object.Key), UID: *object.ETag, @@ -80,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 @@ -99,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 @@ -251,6 +333,7 @@ type GDriveSource struct { ConcurrentDownloads int OutputDir string FailedRootFolders []FailedRootFolder + UnsupportedFiles *UnsupportedFiles } // Close releases resources held by the underlying clients. @@ -258,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 { @@ -304,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] { @@ -316,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)") @@ -327,10 +475,11 @@ func (g *GDriveSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.F ) // Phase 2: Download files, fetch permissions, store to filestore + 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) @@ -341,6 +490,28 @@ 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.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 + } + currentFiles[record.FileID] = ext dlGroup.Go(func() error { filestorePath := path.Join(g.OutputDir, record.FileID+ext) @@ -351,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 { @@ -468,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")