From aeece599037bb9f4633cf6b000508d94a0bdc999 Mon Sep 17 00:00:00 2001 From: Shikhar Gupta Date: Mon, 20 Jul 2026 15:43:00 +0530 Subject: [PATCH 1/4] Surface inaccessible GDrive items in SourceCrawler status When crawling Google Drive folders, files and folders the service account cannot access are now reported in the SourceCrawler CR status with full detail (folder IDs, file IDs, shortcut targets) instead of being silently discarded --- api/v1alpha1/sourcecrawler_types.go | 32 ++++++ api/v1alpha1/zz_generated.deepcopy.go | 100 ++++++++++++++++++ ...r.dataverse.redhat.com_sourcecrawlers.yaml | 63 +++++++++++ .../controller/sourcecrawler_controller.go | 43 +++++++- pkg/unstructured/source.go | 25 +++-- 5 files changed, 253 insertions(+), 10 deletions(-) diff --git a/api/v1alpha1/sourcecrawler_types.go b/api/v1alpha1/sourcecrawler_types.go index e61df118..f2210b7a 100644 --- a/api/v1alpha1/sourcecrawler_types.go +++ b/api/v1alpha1/sourcecrawler_types.go @@ -60,11 +60,43 @@ type SourceCrawlerSpec struct { SourceCrawlerConfig SourceCrawlerConfig `json:"sourceCrawlerConfig,omitempty"` } +type InaccessibleFolder struct { + FolderID string `json:"folderID"` + FolderName string `json:"folderName"` + RootFolderID string `json:"rootFolderID"` +} + +type InaccessibleFile struct { + FileID string `json:"fileID"` + ParentFolderID string `json:"parentFolderID"` + RootFolderID string `json:"rootFolderID"` +} + +type InaccessibleShortcutFolder struct { + ShortcutFileID string `json:"shortcutFileID"` + TargetFolderID string `json:"targetFolderID"` + RootFolderID string `json:"rootFolderID"` +} + +type InaccessibleShortcutFile struct { + ShortcutFileID string `json:"shortcutFileID"` + TargetFileID string `json:"targetFileID"` + RootFolderID string `json:"rootFolderID"` +} + +type InaccessibleItems struct { + Folders []InaccessibleFolder `json:"folders,omitempty"` + Files []InaccessibleFile `json:"files,omitempty"` + ShortcutTargetFolders []InaccessibleShortcutFolder `json:"shortcutTargetFolders,omitempty"` + ShortcutTargetFiles []InaccessibleShortcutFile `json:"shortcutTargetFiles,omitempty"` +} + // SourceCrawlerStatus defines the observed state of SourceCrawler. type SourceCrawlerStatus struct { LastAppliedGeneration int64 `json:"lastAppliedGeneration,omitempty"` Conditions []metav1.Condition `json:"conditions,omitempty"` FilesProcessed int64 `json:"filesProcessed,omitempty"` + InaccessibleItems *InaccessibleItems `json:"inaccessibleItems,omitempty"` } // +kubebuilder:object:root=true diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index a055fc17..7f0ca7dc 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -634,6 +634,101 @@ func (in *GoogleDriveFolders) DeepCopy() *GoogleDriveFolders { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *InaccessibleFile) DeepCopyInto(out *InaccessibleFile) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InaccessibleFile. +func (in *InaccessibleFile) DeepCopy() *InaccessibleFile { + if in == nil { + return nil + } + out := new(InaccessibleFile) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *InaccessibleFolder) DeepCopyInto(out *InaccessibleFolder) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InaccessibleFolder. +func (in *InaccessibleFolder) DeepCopy() *InaccessibleFolder { + if in == nil { + return nil + } + out := new(InaccessibleFolder) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *InaccessibleItems) DeepCopyInto(out *InaccessibleItems) { + *out = *in + if in.Folders != nil { + in, out := &in.Folders, &out.Folders + *out = make([]InaccessibleFolder, len(*in)) + copy(*out, *in) + } + if in.Files != nil { + in, out := &in.Files, &out.Files + *out = make([]InaccessibleFile, len(*in)) + copy(*out, *in) + } + if in.ShortcutTargetFolders != nil { + in, out := &in.ShortcutTargetFolders, &out.ShortcutTargetFolders + *out = make([]InaccessibleShortcutFolder, len(*in)) + copy(*out, *in) + } + if in.ShortcutTargetFiles != nil { + in, out := &in.ShortcutTargetFiles, &out.ShortcutTargetFiles + *out = make([]InaccessibleShortcutFile, len(*in)) + copy(*out, *in) + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InaccessibleItems. +func (in *InaccessibleItems) DeepCopy() *InaccessibleItems { + if in == nil { + return nil + } + out := new(InaccessibleItems) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *InaccessibleShortcutFile) DeepCopyInto(out *InaccessibleShortcutFile) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InaccessibleShortcutFile. +func (in *InaccessibleShortcutFile) DeepCopy() *InaccessibleShortcutFile { + if in == nil { + return nil + } + out := new(InaccessibleShortcutFile) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *InaccessibleShortcutFolder) DeepCopyInto(out *InaccessibleShortcutFolder) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InaccessibleShortcutFolder. +func (in *InaccessibleShortcutFolder) DeepCopy() *InaccessibleShortcutFolder { + if in == nil { + return nil + } + out := new(InaccessibleShortcutFolder) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *Job) DeepCopyInto(out *Job) { *out = *in @@ -961,6 +1056,11 @@ func (in *SourceCrawlerStatus) DeepCopyInto(out *SourceCrawlerStatus) { (*in)[i].DeepCopyInto(&(*out)[i]) } } + if in.InaccessibleItems != nil { + in, out := &in.InaccessibleItems, &out.InaccessibleItems + *out = new(InaccessibleItems) + (*in).DeepCopyInto(*out) + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new SourceCrawlerStatus. diff --git a/config/crd/bases/operator.dataverse.redhat.com_sourcecrawlers.yaml b/config/crd/bases/operator.dataverse.redhat.com_sourcecrawlers.yaml index 1c13ac15..b4b38743 100644 --- a/config/crd/bases/operator.dataverse.redhat.com_sourcecrawlers.yaml +++ b/config/crd/bases/operator.dataverse.redhat.com_sourcecrawlers.yaml @@ -176,6 +176,69 @@ spec: filesProcessed: format: int64 type: integer + inaccessibleItems: + properties: + files: + items: + properties: + fileID: + type: string + parentFolderID: + type: string + rootFolderID: + type: string + required: + - fileID + - parentFolderID + - rootFolderID + type: object + type: array + folders: + items: + properties: + folderID: + type: string + folderName: + type: string + rootFolderID: + type: string + required: + - folderID + - folderName + - rootFolderID + type: object + type: array + shortcutTargetFiles: + items: + properties: + rootFolderID: + type: string + shortcutFileID: + type: string + targetFileID: + type: string + required: + - rootFolderID + - shortcutFileID + - targetFileID + type: object + type: array + shortcutTargetFolders: + items: + properties: + rootFolderID: + type: string + shortcutFileID: + type: string + targetFolderID: + type: string + required: + - rootFolderID + - shortcutFileID + - targetFolderID + type: object + type: array + type: object lastAppliedGeneration: format: int64 type: integer diff --git a/internal/controller/sourcecrawler_controller.go b/internal/controller/sourcecrawler_controller.go index 2feeb387..bedea2b9 100644 --- a/internal/controller/sourcecrawler_controller.go +++ b/internal/controller/sourcecrawler_controller.go @@ -138,15 +138,23 @@ func (r *SourceCrawlerReconciler) Reconcile(ctx context.Context, req ctrl.Reques return ctrl.Result{}, r.handleError(ctx, sourceCrawlerCR, fmt.Errorf("unsupported source type: %s", sourceCrawlerConfig.Type)) } - storedFiles, err := source.SyncFilesToFilestore(ctx, r.fileStore) + syncResult, err := source.SyncFilesToFilestore(ctx, r.fileStore) if err != nil { return ctrl.Result{}, r.handleError(ctx, sourceCrawlerCR, fmt.Errorf("failed to store files to filestore: %w", err)) } - logger.Info("successfully stored files to filestore", "count", len(storedFiles)) + logger.Info("successfully stored files to filestore", "count", len(syncResult.StoredFiles)) successMessage := fmt.Sprintf("successfully reconciled source crawler: %s", sourceCrawlerCR.Name) + inaccessible := toAPIInaccessibleItems(syncResult.InaccessibleItems) + if inaccessible != nil { + total := len(syncResult.InaccessibleItems.Folders) + len(syncResult.InaccessibleItems.Files) + + len(syncResult.InaccessibleItems.ShortcutTargetFolders) + len(syncResult.InaccessibleItems.ShortcutTargetFiles) + successMessage += fmt.Sprintf(", %d inaccessible items (service account may lack access)", total) + } + if err := controllerutils.StatusPatch(ctx, r.Client, sourceCrawlerCR, func() { - sourceCrawlerCR.Status.FilesProcessed += int64(len(storedFiles)) + sourceCrawlerCR.Status.FilesProcessed += int64(len(syncResult.StoredFiles)) + sourceCrawlerCR.Status.InaccessibleItems = inaccessible sourceCrawlerCR.UpdateStatus(successMessage, nil) }); err != nil { logger.Error(err, "failed to update SourceCrawler CR status") @@ -308,6 +316,35 @@ func (r *SourceCrawlerReconciler) handleError(ctx context.Context, sourceCrawler return reconcileErr } +func toAPIInaccessibleItems(items gdrive.InaccessibleItems) *operatorv1alpha1.InaccessibleItems { + total := len(items.Folders) + len(items.Files) + len(items.ShortcutTargetFolders) + len(items.ShortcutTargetFiles) + if total == 0 { + return nil + } + result := &operatorv1alpha1.InaccessibleItems{} + for _, f := range items.Folders { + result.Folders = append(result.Folders, operatorv1alpha1.InaccessibleFolder{ + FolderID: f.FolderID, FolderName: f.FolderName, RootFolderID: f.RootFolderID, + }) + } + for _, f := range items.Files { + result.Files = append(result.Files, operatorv1alpha1.InaccessibleFile{ + FileID: f.FileID, ParentFolderID: f.ParentFolderID, RootFolderID: f.RootFolderID, + }) + } + for _, f := range items.ShortcutTargetFolders { + result.ShortcutTargetFolders = append(result.ShortcutTargetFolders, operatorv1alpha1.InaccessibleShortcutFolder{ + ShortcutFileID: f.ShortcutFileID, TargetFolderID: f.TargetFolderID, RootFolderID: f.RootFolderID, + }) + } + for _, f := range items.ShortcutTargetFiles { + result.ShortcutTargetFiles = append(result.ShortcutTargetFiles, operatorv1alpha1.InaccessibleShortcutFile{ + ShortcutFileID: f.ShortcutFileID, TargetFileID: f.TargetFileID, RootFolderID: f.RootFolderID, + }) + } + return result +} + // findDependents maps a changed pipeline stage back to the SourceCrawlers that depend on it. // // Given a SourceCrawler CR like: diff --git a/pkg/unstructured/source.go b/pkg/unstructured/source.go index 2d31391e..327965cc 100644 --- a/pkg/unstructured/source.go +++ b/pkg/unstructured/source.go @@ -37,9 +37,13 @@ import ( const maxFileSize int64 = 128 << 20 // 128 MB — Snowflake external stage limit +type SyncResult struct { + StoredFiles []RawFileMetadata + InaccessibleItems gdrive.InaccessibleItems +} + 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) + SyncFilesToFilestore(ctx context.Context, fs *filestore.FileStore) (*SyncResult, error) } type S3BucketSource struct { @@ -49,7 +53,7 @@ type S3BucketSource struct { OutputDir string } -func (s *S3BucketSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.FileStore) ([]RawFileMetadata, error) { +func (s *S3BucketSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.FileStore) (*SyncResult, error) { logger := log.FromContext(ctx) logger.Info("listing objects in prefix", "bucket", s.Bucket, "prefix", s.Prefix) objects, err := awsclienthandler.ListObjectsInPrefix(ctx, s.S3Client, s.Bucket, s.Prefix) @@ -123,7 +127,7 @@ func (s *S3BucketSource) SyncFilesToFilestore(ctx context.Context, fs *filestore return nil, errors.New(errorMessage) } - return storedFiles, nil + return &SyncResult{StoredFiles: storedFiles}, nil } // storeFile will store the given file to the filestore @@ -252,7 +256,7 @@ func (g *GDriveSource) Close() { g.GDriveClient.Close() } -func (g *GDriveSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.FileStore) ([]RawFileMetadata, error) { +func (g *GDriveSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.FileStore) (*SyncResult, error) { logger := log.FromContext(ctx) // Phase 1: Crawl all root folders concurrently @@ -281,12 +285,19 @@ func (g *GDriveSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.F // Merge and filter crawl records to only successful non-folder files var fileRecords []gdrive.CrawlRecord + var mergedInaccessible gdrive.InaccessibleItems seen := make(map[string]bool) for i, r := range results { if r.err != nil { logger.Error(r.err, "folder crawl failed", "folderID", g.FolderIDs[i]) continue } + mergedInaccessible.Folders = append(mergedInaccessible.Folders, r.result.InaccessibleItems.Folders...) + mergedInaccessible.Files = append(mergedInaccessible.Files, r.result.InaccessibleItems.Files...) + mergedInaccessible.ShortcutTargetFolders = append(mergedInaccessible.ShortcutTargetFolders, + r.result.InaccessibleItems.ShortcutTargetFolders...) + mergedInaccessible.ShortcutTargetFiles = append(mergedInaccessible.ShortcutTargetFiles, + r.result.InaccessibleItems.ShortcutTargetFiles...) for _, record := range r.result.Records { if record.Status != "successful" { continue @@ -394,10 +405,10 @@ func (g *GDriveSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.F errorMessage += fmt.Sprintf("fileID: %s, error: %v\n", fileID, err) } if len(errorMessage) > 0 { - return storedFiles, errors.New(errorMessage) + return &SyncResult{StoredFiles: storedFiles, InaccessibleItems: mergedInaccessible}, errors.New(errorMessage) } - return storedFiles, nil + return &SyncResult{StoredFiles: storedFiles, InaccessibleItems: mergedInaccessible}, nil } // storeFile downloads a Google Drive file and stores it with metadata, From 336a92b7676c965d17f63a438196e284dafe0b37 Mon Sep 17 00:00:00 2001 From: Shikhar Gupta Date: Mon, 20 Jul 2026 17:25:25 +0530 Subject: [PATCH 2/4] Address comments --- .../controller/sourcecrawler_controller.go | 27 ++++++++++++------- 1 file changed, 18 insertions(+), 9 deletions(-) diff --git a/internal/controller/sourcecrawler_controller.go b/internal/controller/sourcecrawler_controller.go index bedea2b9..c2e64f67 100644 --- a/internal/controller/sourcecrawler_controller.go +++ b/internal/controller/sourcecrawler_controller.go @@ -138,17 +138,26 @@ func (r *SourceCrawlerReconciler) Reconcile(ctx context.Context, req ctrl.Reques return ctrl.Result{}, r.handleError(ctx, sourceCrawlerCR, fmt.Errorf("unsupported source type: %s", sourceCrawlerConfig.Type)) } - syncResult, err := source.SyncFilesToFilestore(ctx, r.fileStore) - if err != nil { - return ctrl.Result{}, r.handleError(ctx, sourceCrawlerCR, fmt.Errorf("failed to store files to filestore: %w", err)) + syncResult, syncErr := source.SyncFilesToFilestore(ctx, r.fileStore) + if syncErr != nil { + if syncResult != nil { + inaccessible, _ := toAPIInaccessibleItems(syncResult.InaccessibleItems) + if err := controllerutils.StatusPatch(ctx, r.Client, sourceCrawlerCR, func() { + sourceCrawlerCR.Status.FilesProcessed += int64(len(syncResult.StoredFiles)) + sourceCrawlerCR.Status.InaccessibleItems = inaccessible + sourceCrawlerCR.UpdateStatus("", syncErr) + }); err != nil { + logger.Error(err, "failed to update SourceCrawler CR status with partial results") + return ctrl.Result{}, err + } + } + return ctrl.Result{}, r.handleError(ctx, sourceCrawlerCR, syncErr) } logger.Info("successfully stored files to filestore", "count", len(syncResult.StoredFiles)) successMessage := fmt.Sprintf("successfully reconciled source crawler: %s", sourceCrawlerCR.Name) - inaccessible := toAPIInaccessibleItems(syncResult.InaccessibleItems) + inaccessible, total := toAPIInaccessibleItems(syncResult.InaccessibleItems) if inaccessible != nil { - total := len(syncResult.InaccessibleItems.Folders) + len(syncResult.InaccessibleItems.Files) + - len(syncResult.InaccessibleItems.ShortcutTargetFolders) + len(syncResult.InaccessibleItems.ShortcutTargetFiles) successMessage += fmt.Sprintf(", %d inaccessible items (service account may lack access)", total) } @@ -316,10 +325,10 @@ func (r *SourceCrawlerReconciler) handleError(ctx context.Context, sourceCrawler return reconcileErr } -func toAPIInaccessibleItems(items gdrive.InaccessibleItems) *operatorv1alpha1.InaccessibleItems { +func toAPIInaccessibleItems(items gdrive.InaccessibleItems) (*operatorv1alpha1.InaccessibleItems, int) { total := len(items.Folders) + len(items.Files) + len(items.ShortcutTargetFolders) + len(items.ShortcutTargetFiles) if total == 0 { - return nil + return nil, 0 } result := &operatorv1alpha1.InaccessibleItems{} for _, f := range items.Folders { @@ -342,7 +351,7 @@ func toAPIInaccessibleItems(items gdrive.InaccessibleItems) *operatorv1alpha1.In ShortcutFileID: f.ShortcutFileID, TargetFileID: f.TargetFileID, RootFolderID: f.RootFolderID, }) } - return result + return result, total } // findDependents maps a changed pipeline stage back to the SourceCrawlers that depend on it. From 7fd1d326703431ec8c3137396fc17011041a1d7c Mon Sep 17 00:00:00 2001 From: Shikhar Gupta Date: Mon, 27 Jul 2026 17:29:25 +0530 Subject: [PATCH 3/4] Address comments --- api/v1alpha1/sourcecrawler_types.go | 44 ++++------ api/v1alpha1/zz_generated.deepcopy.go | 84 +++---------------- ...r.dataverse.redhat.com_sourcecrawlers.yaml | 60 +++---------- .../controller/sourcecrawler_controller.go | 44 +++++----- pkg/unstructured/source.go | 46 ++++++++-- 5 files changed, 97 insertions(+), 181 deletions(-) diff --git a/api/v1alpha1/sourcecrawler_types.go b/api/v1alpha1/sourcecrawler_types.go index f2210b7a..aed12de6 100644 --- a/api/v1alpha1/sourcecrawler_types.go +++ b/api/v1alpha1/sourcecrawler_types.go @@ -60,43 +60,27 @@ type SourceCrawlerSpec struct { SourceCrawlerConfig SourceCrawlerConfig `json:"sourceCrawlerConfig,omitempty"` } -type InaccessibleFolder struct { - FolderID string `json:"folderID"` - FolderName string `json:"folderName"` +type InaccessibleItem struct { + Kind string `json:"kind"` + ItemID string `json:"itemID"` + Name string `json:"name,omitempty"` + TargetID string `json:"targetID,omitempty"` RootFolderID string `json:"rootFolderID"` + Reason string `json:"reason,omitempty"` } -type InaccessibleFile struct { - FileID string `json:"fileID"` - ParentFolderID string `json:"parentFolderID"` - RootFolderID string `json:"rootFolderID"` -} - -type InaccessibleShortcutFolder struct { - ShortcutFileID string `json:"shortcutFileID"` - TargetFolderID string `json:"targetFolderID"` - RootFolderID string `json:"rootFolderID"` -} - -type InaccessibleShortcutFile struct { - ShortcutFileID string `json:"shortcutFileID"` - TargetFileID string `json:"targetFileID"` - RootFolderID string `json:"rootFolderID"` -} - -type InaccessibleItems struct { - Folders []InaccessibleFolder `json:"folders,omitempty"` - Files []InaccessibleFile `json:"files,omitempty"` - ShortcutTargetFolders []InaccessibleShortcutFolder `json:"shortcutTargetFolders,omitempty"` - ShortcutTargetFiles []InaccessibleShortcutFile `json:"shortcutTargetFiles,omitempty"` +type InaccessibleSummary struct { + // +listType=atomic + Items []InaccessibleItem `json:"items,omitempty"` + Count int `json:"count"` } // SourceCrawlerStatus defines the observed state of SourceCrawler. type SourceCrawlerStatus struct { - LastAppliedGeneration int64 `json:"lastAppliedGeneration,omitempty"` - Conditions []metav1.Condition `json:"conditions,omitempty"` - FilesProcessed int64 `json:"filesProcessed,omitempty"` - InaccessibleItems *InaccessibleItems `json:"inaccessibleItems,omitempty"` + LastAppliedGeneration int64 `json:"lastAppliedGeneration,omitempty"` + Conditions []metav1.Condition `json:"conditions,omitempty"` + FilesProcessed int64 `json:"filesProcessed,omitempty"` + InaccessibleItems *InaccessibleSummary `json:"inaccessibleItems,omitempty"` } // +kubebuilder:object:root=true diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index 7f0ca7dc..3a1a8911 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -635,96 +635,36 @@ func (in *GoogleDriveFolders) DeepCopy() *GoogleDriveFolders { } // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. -func (in *InaccessibleFile) DeepCopyInto(out *InaccessibleFile) { +func (in *InaccessibleItem) DeepCopyInto(out *InaccessibleItem) { *out = *in } -// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InaccessibleFile. -func (in *InaccessibleFile) DeepCopy() *InaccessibleFile { +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InaccessibleItem. +func (in *InaccessibleItem) DeepCopy() *InaccessibleItem { if in == nil { return nil } - out := new(InaccessibleFile) + out := new(InaccessibleItem) in.DeepCopyInto(out) return out } // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. -func (in *InaccessibleFolder) DeepCopyInto(out *InaccessibleFolder) { +func (in *InaccessibleSummary) DeepCopyInto(out *InaccessibleSummary) { *out = *in -} - -// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InaccessibleFolder. -func (in *InaccessibleFolder) DeepCopy() *InaccessibleFolder { - if in == nil { - return nil - } - out := new(InaccessibleFolder) - in.DeepCopyInto(out) - return out -} - -// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. -func (in *InaccessibleItems) DeepCopyInto(out *InaccessibleItems) { - *out = *in - if in.Folders != nil { - in, out := &in.Folders, &out.Folders - *out = make([]InaccessibleFolder, len(*in)) - copy(*out, *in) - } - if in.Files != nil { - in, out := &in.Files, &out.Files - *out = make([]InaccessibleFile, len(*in)) - copy(*out, *in) - } - if in.ShortcutTargetFolders != nil { - in, out := &in.ShortcutTargetFolders, &out.ShortcutTargetFolders - *out = make([]InaccessibleShortcutFolder, len(*in)) - copy(*out, *in) - } - if in.ShortcutTargetFiles != nil { - in, out := &in.ShortcutTargetFiles, &out.ShortcutTargetFiles - *out = make([]InaccessibleShortcutFile, len(*in)) + if in.Items != nil { + in, out := &in.Items, &out.Items + *out = make([]InaccessibleItem, len(*in)) copy(*out, *in) } } -// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InaccessibleItems. -func (in *InaccessibleItems) DeepCopy() *InaccessibleItems { - if in == nil { - return nil - } - out := new(InaccessibleItems) - in.DeepCopyInto(out) - return out -} - -// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. -func (in *InaccessibleShortcutFile) DeepCopyInto(out *InaccessibleShortcutFile) { - *out = *in -} - -// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InaccessibleShortcutFile. -func (in *InaccessibleShortcutFile) DeepCopy() *InaccessibleShortcutFile { - if in == nil { - return nil - } - out := new(InaccessibleShortcutFile) - in.DeepCopyInto(out) - return out -} - -// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. -func (in *InaccessibleShortcutFolder) DeepCopyInto(out *InaccessibleShortcutFolder) { - *out = *in -} - -// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InaccessibleShortcutFolder. -func (in *InaccessibleShortcutFolder) DeepCopy() *InaccessibleShortcutFolder { +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InaccessibleSummary. +func (in *InaccessibleSummary) DeepCopy() *InaccessibleSummary { if in == nil { return nil } - out := new(InaccessibleShortcutFolder) + out := new(InaccessibleSummary) in.DeepCopyInto(out) return out } @@ -1058,7 +998,7 @@ func (in *SourceCrawlerStatus) DeepCopyInto(out *SourceCrawlerStatus) { } if in.InaccessibleItems != nil { in, out := &in.InaccessibleItems, &out.InaccessibleItems - *out = new(InaccessibleItems) + *out = new(InaccessibleSummary) (*in).DeepCopyInto(*out) } } diff --git a/config/crd/bases/operator.dataverse.redhat.com_sourcecrawlers.yaml b/config/crd/bases/operator.dataverse.redhat.com_sourcecrawlers.yaml index b4b38743..f5bf26e1 100644 --- a/config/crd/bases/operator.dataverse.redhat.com_sourcecrawlers.yaml +++ b/config/crd/bases/operator.dataverse.redhat.com_sourcecrawlers.yaml @@ -178,66 +178,32 @@ spec: type: integer inaccessibleItems: properties: - files: + count: + type: integer + items: items: properties: - fileID: + itemID: type: string - parentFolderID: + kind: type: string - rootFolderID: - type: string - required: - - fileID - - parentFolderID - - rootFolderID - type: object - type: array - folders: - items: - properties: - folderID: + name: type: string - folderName: + reason: type: string rootFolderID: type: string - required: - - folderID - - folderName - - rootFolderID - type: object - type: array - shortcutTargetFiles: - items: - properties: - rootFolderID: - type: string - shortcutFileID: - type: string - targetFileID: - type: string - required: - - rootFolderID - - shortcutFileID - - targetFileID - type: object - type: array - shortcutTargetFolders: - items: - properties: - rootFolderID: - type: string - shortcutFileID: - type: string - targetFolderID: + targetID: type: string required: + - itemID + - kind - rootFolderID - - shortcutFileID - - targetFolderID type: object type: array + x-kubernetes-list-type: atomic + required: + - count type: object lastAppliedGeneration: format: int64 diff --git a/internal/controller/sourcecrawler_controller.go b/internal/controller/sourcecrawler_controller.go index c2e64f67..08ca3dcf 100644 --- a/internal/controller/sourcecrawler_controller.go +++ b/internal/controller/sourcecrawler_controller.go @@ -141,7 +141,7 @@ func (r *SourceCrawlerReconciler) Reconcile(ctx context.Context, req ctrl.Reques syncResult, syncErr := source.SyncFilesToFilestore(ctx, r.fileStore) if syncErr != nil { if syncResult != nil { - inaccessible, _ := toAPIInaccessibleItems(syncResult.InaccessibleItems) + inaccessible, _ := buildInaccessibleSummary(syncResult.InaccessibleItems) if err := controllerutils.StatusPatch(ctx, r.Client, sourceCrawlerCR, func() { sourceCrawlerCR.Status.FilesProcessed += int64(len(syncResult.StoredFiles)) sourceCrawlerCR.Status.InaccessibleItems = inaccessible @@ -156,7 +156,7 @@ func (r *SourceCrawlerReconciler) Reconcile(ctx context.Context, req ctrl.Reques logger.Info("successfully stored files to filestore", "count", len(syncResult.StoredFiles)) successMessage := fmt.Sprintf("successfully reconciled source crawler: %s", sourceCrawlerCR.Name) - inaccessible, total := toAPIInaccessibleItems(syncResult.InaccessibleItems) + inaccessible, total := buildInaccessibleSummary(syncResult.InaccessibleItems) if inaccessible != nil { successMessage += fmt.Sprintf(", %d inaccessible items (service account may lack access)", total) } @@ -325,33 +325,29 @@ func (r *SourceCrawlerReconciler) handleError(ctx context.Context, sourceCrawler return reconcileErr } -func toAPIInaccessibleItems(items gdrive.InaccessibleItems) (*operatorv1alpha1.InaccessibleItems, int) { - total := len(items.Folders) + len(items.Files) + len(items.ShortcutTargetFolders) + len(items.ShortcutTargetFiles) +const maxInaccessibleItems = 100 + +func buildInaccessibleSummary(items []unstructured.InaccessibleItem) (*operatorv1alpha1.InaccessibleSummary, int) { + total := len(items) if total == 0 { return nil, 0 } - result := &operatorv1alpha1.InaccessibleItems{} - for _, f := range items.Folders { - result.Folders = append(result.Folders, operatorv1alpha1.InaccessibleFolder{ - FolderID: f.FolderID, FolderName: f.FolderName, RootFolderID: f.RootFolderID, - }) - } - for _, f := range items.Files { - result.Files = append(result.Files, operatorv1alpha1.InaccessibleFile{ - FileID: f.FileID, ParentFolderID: f.ParentFolderID, RootFolderID: f.RootFolderID, - }) - } - for _, f := range items.ShortcutTargetFolders { - result.ShortcutTargetFolders = append(result.ShortcutTargetFolders, operatorv1alpha1.InaccessibleShortcutFolder{ - ShortcutFileID: f.ShortcutFileID, TargetFolderID: f.TargetFolderID, RootFolderID: f.RootFolderID, - }) - } - for _, f := range items.ShortcutTargetFiles { - result.ShortcutTargetFiles = append(result.ShortcutTargetFiles, operatorv1alpha1.InaccessibleShortcutFile{ - ShortcutFileID: f.ShortcutFileID, TargetFileID: f.TargetFileID, RootFolderID: f.RootFolderID, + capped := items + if len(capped) > maxInaccessibleItems { + capped = capped[:maxInaccessibleItems] + } + apiItems := make([]operatorv1alpha1.InaccessibleItem, 0, len(capped)) + for _, item := range capped { + apiItems = append(apiItems, operatorv1alpha1.InaccessibleItem{ + Kind: item.Kind, + ItemID: item.ItemID, + Name: item.Name, + TargetID: item.TargetID, + RootFolderID: item.RootFolderID, + Reason: item.Reason, }) } - return result, total + return &operatorv1alpha1.InaccessibleSummary{Items: apiItems, Count: total}, total } // findDependents maps a changed pipeline stage back to the SourceCrawlers that depend on it. diff --git a/pkg/unstructured/source.go b/pkg/unstructured/source.go index 327965cc..e8741fd6 100644 --- a/pkg/unstructured/source.go +++ b/pkg/unstructured/source.go @@ -37,9 +37,18 @@ import ( const maxFileSize int64 = 128 << 20 // 128 MB — Snowflake external stage limit +type InaccessibleItem struct { + Kind string + ItemID string + Name string + TargetID string + RootFolderID string + Reason string +} + type SyncResult struct { StoredFiles []RawFileMetadata - InaccessibleItems gdrive.InaccessibleItems + InaccessibleItems []InaccessibleItem } type DataSource interface { @@ -285,19 +294,14 @@ func (g *GDriveSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.F // Merge and filter crawl records to only successful non-folder files var fileRecords []gdrive.CrawlRecord - var mergedInaccessible gdrive.InaccessibleItems + var mergedInaccessible []InaccessibleItem seen := make(map[string]bool) for i, r := range results { if r.err != nil { logger.Error(r.err, "folder crawl failed", "folderID", g.FolderIDs[i]) continue } - mergedInaccessible.Folders = append(mergedInaccessible.Folders, r.result.InaccessibleItems.Folders...) - mergedInaccessible.Files = append(mergedInaccessible.Files, r.result.InaccessibleItems.Files...) - mergedInaccessible.ShortcutTargetFolders = append(mergedInaccessible.ShortcutTargetFolders, - r.result.InaccessibleItems.ShortcutTargetFolders...) - mergedInaccessible.ShortcutTargetFiles = append(mergedInaccessible.ShortcutTargetFiles, - r.result.InaccessibleItems.ShortcutTargetFiles...) + mergedInaccessible = append(mergedInaccessible, flattenInaccessibleItems(r.result.InaccessibleItems)...) for _, record := range r.result.Records { if record.Status != "successful" { continue @@ -502,6 +506,32 @@ func (g *GDriveSource) storeFile( return fileChanged, nil } +func flattenInaccessibleItems(items gdrive.InaccessibleItems) []InaccessibleItem { + total := len(items.Folders) + len(items.Files) + len(items.ShortcutTargetFolders) + len(items.ShortcutTargetFiles) + result := make([]InaccessibleItem, 0, total) + for _, f := range items.Folders { + result = append(result, InaccessibleItem{ + Kind: "folder", ItemID: f.FolderID, Name: f.FolderName, RootFolderID: f.RootFolderID, + }) + } + for _, f := range items.Files { + result = append(result, InaccessibleItem{ + Kind: "file", ItemID: f.FileID, RootFolderID: f.RootFolderID, + }) + } + for _, f := range items.ShortcutTargetFolders { + result = append(result, InaccessibleItem{ + Kind: "shortcutFolder", ItemID: f.ShortcutFileID, TargetID: f.TargetFolderID, RootFolderID: f.RootFolderID, + }) + } + for _, f := range items.ShortcutTargetFiles { + result = append(result, InaccessibleItem{ + Kind: "shortcutFile", ItemID: f.ShortcutFileID, TargetID: f.TargetFileID, RootFolderID: f.RootFolderID, + }) + } + return result +} + // extractFileID extracts the Google Drive file ID from a filestore // path like "/.pdf" or "/.pdf.json". func (g *GDriveSource) extractFileID(filestorePath string) string { From 02cca1540252607888dc0f98db5dd1b107a098da Mon Sep 17 00:00:00 2001 From: Shikhar Gupta Date: Wed, 29 Jul 2026 19:31:35 +0530 Subject: [PATCH 4/4] Surface inaccessible root folders in SourceCrawler status --- api/v1alpha1/sourcecrawler_types.go | 15 ++++++--- api/v1alpha1/zz_generated.deepcopy.go | 20 ++++++++++++ ...r.dataverse.redhat.com_sourcecrawlers.yaml | 14 +++++++++ .../controller/sourcecrawler_controller.go | 31 +++++++++++++++++++ pkg/unstructured/source.go | 15 +++++++++ 5 files changed, 91 insertions(+), 4 deletions(-) diff --git a/api/v1alpha1/sourcecrawler_types.go b/api/v1alpha1/sourcecrawler_types.go index aed12de6..ec5a2d6f 100644 --- a/api/v1alpha1/sourcecrawler_types.go +++ b/api/v1alpha1/sourcecrawler_types.go @@ -75,12 +75,19 @@ type InaccessibleSummary struct { Count int `json:"count"` } +type InaccessibleRootFolder struct { + FolderID string `json:"folderID"` + URL string `json:"url,omitempty"` + Error string `json:"error"` +} + // SourceCrawlerStatus defines the observed state of SourceCrawler. type SourceCrawlerStatus struct { - LastAppliedGeneration int64 `json:"lastAppliedGeneration,omitempty"` - Conditions []metav1.Condition `json:"conditions,omitempty"` - FilesProcessed int64 `json:"filesProcessed,omitempty"` - InaccessibleItems *InaccessibleSummary `json:"inaccessibleItems,omitempty"` + LastAppliedGeneration int64 `json:"lastAppliedGeneration,omitempty"` + Conditions []metav1.Condition `json:"conditions,omitempty"` + FilesProcessed int64 `json:"filesProcessed,omitempty"` + InaccessibleItems *InaccessibleSummary `json:"inaccessibleItems,omitempty"` + InaccessibleRootFolders []InaccessibleRootFolder `json:"inaccessibleRootFolders,omitempty"` } // +kubebuilder:object:root=true diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index 3a1a8911..606ee3fb 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -649,6 +649,21 @@ func (in *InaccessibleItem) DeepCopy() *InaccessibleItem { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *InaccessibleRootFolder) DeepCopyInto(out *InaccessibleRootFolder) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InaccessibleRootFolder. +func (in *InaccessibleRootFolder) DeepCopy() *InaccessibleRootFolder { + if in == nil { + return nil + } + out := new(InaccessibleRootFolder) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *InaccessibleSummary) DeepCopyInto(out *InaccessibleSummary) { *out = *in @@ -1001,6 +1016,11 @@ func (in *SourceCrawlerStatus) DeepCopyInto(out *SourceCrawlerStatus) { *out = new(InaccessibleSummary) (*in).DeepCopyInto(*out) } + if in.InaccessibleRootFolders != nil { + in, out := &in.InaccessibleRootFolders, &out.InaccessibleRootFolders + *out = make([]InaccessibleRootFolder, len(*in)) + copy(*out, *in) + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new SourceCrawlerStatus. diff --git a/config/crd/bases/operator.dataverse.redhat.com_sourcecrawlers.yaml b/config/crd/bases/operator.dataverse.redhat.com_sourcecrawlers.yaml index f5bf26e1..9e9054b3 100644 --- a/config/crd/bases/operator.dataverse.redhat.com_sourcecrawlers.yaml +++ b/config/crd/bases/operator.dataverse.redhat.com_sourcecrawlers.yaml @@ -205,6 +205,20 @@ spec: required: - count type: object + inaccessibleRootFolders: + items: + properties: + error: + type: string + folderID: + type: string + url: + type: string + required: + - error + - folderID + type: object + type: array lastAppliedGeneration: format: int64 type: integer diff --git a/internal/controller/sourcecrawler_controller.go b/internal/controller/sourcecrawler_controller.go index 08ca3dcf..92dce0dc 100644 --- a/internal/controller/sourcecrawler_controller.go +++ b/internal/controller/sourcecrawler_controller.go @@ -139,12 +139,20 @@ func (r *SourceCrawlerReconciler) Reconcile(ctx context.Context, req ctrl.Reques } syncResult, syncErr := source.SyncFilesToFilestore(ctx, r.fileStore) + + // Build inaccessible root folders status from GDriveSource if applicable + var failedRootFolders []operatorv1alpha1.InaccessibleRootFolder + if gds, ok := source.(*unstructured.GDriveSource); ok && len(gds.FailedRootFolders) > 0 { + failedRootFolders = buildInaccessibleRootFolders(gds.FailedRootFolders, sourceCrawlerConfig.GoogleDriveConfig) + } + if syncErr != nil { if syncResult != nil { inaccessible, _ := buildInaccessibleSummary(syncResult.InaccessibleItems) if err := controllerutils.StatusPatch(ctx, r.Client, sourceCrawlerCR, func() { sourceCrawlerCR.Status.FilesProcessed += int64(len(syncResult.StoredFiles)) sourceCrawlerCR.Status.InaccessibleItems = inaccessible + sourceCrawlerCR.Status.InaccessibleRootFolders = failedRootFolders sourceCrawlerCR.UpdateStatus("", syncErr) }); err != nil { logger.Error(err, "failed to update SourceCrawler CR status with partial results") @@ -156,6 +164,9 @@ func (r *SourceCrawlerReconciler) Reconcile(ctx context.Context, req ctrl.Reques logger.Info("successfully stored files to filestore", "count", len(syncResult.StoredFiles)) successMessage := fmt.Sprintf("successfully reconciled source crawler: %s", sourceCrawlerCR.Name) + if len(failedRootFolders) > 0 { + successMessage += fmt.Sprintf(", %d inaccessible root folders", len(failedRootFolders)) + } inaccessible, total := buildInaccessibleSummary(syncResult.InaccessibleItems) if inaccessible != nil { successMessage += fmt.Sprintf(", %d inaccessible items (service account may lack access)", total) @@ -164,6 +175,7 @@ func (r *SourceCrawlerReconciler) Reconcile(ctx context.Context, req ctrl.Reques if err := controllerutils.StatusPatch(ctx, r.Client, sourceCrawlerCR, func() { sourceCrawlerCR.Status.FilesProcessed += int64(len(syncResult.StoredFiles)) sourceCrawlerCR.Status.InaccessibleItems = inaccessible + sourceCrawlerCR.Status.InaccessibleRootFolders = failedRootFolders sourceCrawlerCR.UpdateStatus(successMessage, nil) }); err != nil { logger.Error(err, "failed to update SourceCrawler CR status") @@ -327,6 +339,25 @@ func (r *SourceCrawlerReconciler) handleError(ctx context.Context, sourceCrawler const maxInaccessibleItems = 100 +func buildInaccessibleRootFolders(failed []unstructured.FailedRootFolder, gdriveConfig *operatorv1alpha1.GoogleDriveConfig) []operatorv1alpha1.InaccessibleRootFolder { + folderIDToURL := make(map[string]string, len(gdriveConfig.Folders)) + for _, f := range gdriveConfig.Folders { + id, err := extractGDriveFolderID(f.URL) + if err == nil { + folderIDToURL[id] = f.URL + } + } + result := make([]operatorv1alpha1.InaccessibleRootFolder, 0, len(failed)) + for _, f := range failed { + result = append(result, operatorv1alpha1.InaccessibleRootFolder{ + FolderID: f.FolderID, + URL: folderIDToURL[f.FolderID], + Error: f.Error, + }) + } + return result +} + func buildInaccessibleSummary(items []unstructured.InaccessibleItem) (*operatorv1alpha1.InaccessibleSummary, int) { total := len(items) if total == 0 { diff --git a/pkg/unstructured/source.go b/pkg/unstructured/source.go index e8741fd6..c7d9fc2f 100644 --- a/pkg/unstructured/source.go +++ b/pkg/unstructured/source.go @@ -249,6 +249,11 @@ func (s *S3BucketSource) s3Key(filestorePath string) string { return path.Join(s.Prefix, baseName) } +type FailedRootFolder struct { + FolderID string + Error string +} + // GDriveSource implements DataSource for Google Drive folders. type GDriveSource struct { GDriveClient *gdrive.Client @@ -258,6 +263,7 @@ type GDriveSource struct { ConcurrentFolders int ConcurrentDownloads int OutputDir string + FailedRootFolders []FailedRootFolder } // Close releases resources held by the underlying clients. @@ -299,6 +305,10 @@ func (g *GDriveSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.F 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 } mergedInaccessible = append(mergedInaccessible, flattenInaccessibleItems(r.result.InaccessibleItems)...) @@ -322,6 +332,11 @@ func (g *GDriveSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.F } } + if len(g.FailedRootFolders) == len(g.FolderIDs) { + return &SyncResult{InaccessibleItems: mergedInaccessible}, + errors.New("all configured root folders are inaccessible (service account may lack access)") + } + logger.Info("gdrive crawl complete, starting file download", "discoveredFiles", len(fileRecords), "concurrentDownloads", g.ConcurrentDownloads,