diff --git a/api/v1alpha1/sourcecrawler_types.go b/api/v1alpha1/sourcecrawler_types.go index e61df118..ec5a2d6f 100644 --- a/api/v1alpha1/sourcecrawler_types.go +++ b/api/v1alpha1/sourcecrawler_types.go @@ -60,11 +60,34 @@ type SourceCrawlerSpec struct { SourceCrawlerConfig SourceCrawlerConfig `json:"sourceCrawlerConfig,omitempty"` } +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 InaccessibleSummary struct { + // +listType=atomic + Items []InaccessibleItem `json:"items,omitempty"` + 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"` + 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 a055fc17..606ee3fb 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -634,6 +634,56 @@ 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 *InaccessibleItem) DeepCopyInto(out *InaccessibleItem) { + *out = *in +} + +// 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(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 *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 + 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 InaccessibleSummary. +func (in *InaccessibleSummary) DeepCopy() *InaccessibleSummary { + if in == nil { + return nil + } + out := new(InaccessibleSummary) + 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 +1011,16 @@ func (in *SourceCrawlerStatus) DeepCopyInto(out *SourceCrawlerStatus) { (*in)[i].DeepCopyInto(&(*out)[i]) } } + if in.InaccessibleItems != nil { + in, out := &in.InaccessibleItems, &out.InaccessibleItems + *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 1c13ac15..9e9054b3 100644 --- a/config/crd/bases/operator.dataverse.redhat.com_sourcecrawlers.yaml +++ b/config/crd/bases/operator.dataverse.redhat.com_sourcecrawlers.yaml @@ -176,6 +176,49 @@ spec: filesProcessed: format: int64 type: integer + inaccessibleItems: + properties: + count: + type: integer + items: + items: + properties: + itemID: + type: string + kind: + type: string + name: + type: string + reason: + type: string + rootFolderID: + type: string + targetID: + type: string + required: + - itemID + - kind + - rootFolderID + type: object + type: array + x-kubernetes-list-type: atomic + 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 2feeb387..92dce0dc 100644 --- a/internal/controller/sourcecrawler_controller.go +++ b/internal/controller/sourcecrawler_controller.go @@ -138,15 +138,44 @@ 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) - 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) + + // 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") + return ctrl.Result{}, err + } + } + return ctrl.Result{}, r.handleError(ctx, sourceCrawlerCR, syncErr) } - 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) + 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) + } + 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.Status.InaccessibleRootFolders = failedRootFolders sourceCrawlerCR.UpdateStatus(successMessage, nil) }); err != nil { logger.Error(err, "failed to update SourceCrawler CR status") @@ -308,6 +337,50 @@ func (r *SourceCrawlerReconciler) handleError(ctx context.Context, sourceCrawler return reconcileErr } +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 { + return nil, 0 + } + 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 &operatorv1alpha1.InaccessibleSummary{Items: apiItems, Count: total}, total +} + // 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..c7d9fc2f 100644 --- a/pkg/unstructured/source.go +++ b/pkg/unstructured/source.go @@ -37,9 +37,22 @@ 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 []InaccessibleItem +} + 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 +62,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 +136,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 @@ -236,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 @@ -245,6 +263,7 @@ type GDriveSource struct { ConcurrentFolders int ConcurrentDownloads int OutputDir string + FailedRootFolders []FailedRootFolder } // Close releases resources held by the underlying clients. @@ -252,7 +271,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 +300,18 @@ 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 []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]) + g.FailedRootFolders = append(g.FailedRootFolders, FailedRootFolder{ + FolderID: g.FolderIDs[i], + Error: r.err.Error(), + }) continue } + mergedInaccessible = append(mergedInaccessible, flattenInaccessibleItems(r.result.InaccessibleItems)...) for _, record := range r.result.Records { if record.Status != "successful" { continue @@ -307,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, @@ -394,10 +424,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, @@ -491,6 +521,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 {