diff --git a/api/v1alpha1/sourcecrawler_types.go b/api/v1alpha1/sourcecrawler_types.go index e61df118..060134a5 100644 --- a/api/v1alpha1/sourcecrawler_types.go +++ b/api/v1alpha1/sourcecrawler_types.go @@ -60,11 +60,17 @@ type SourceCrawlerSpec struct { SourceCrawlerConfig SourceCrawlerConfig `json:"sourceCrawlerConfig,omitempty"` } +type GDriveFolderStatus struct { + URL string `json:"url"` + Error string `json:"error,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"` + LastAppliedGeneration int64 `json:"lastAppliedGeneration,omitempty"` + Conditions []metav1.Condition `json:"conditions,omitempty"` + FilesProcessed int64 `json:"filesProcessed,omitempty"` + GDriveStatus []GDriveFolderStatus `json:"gdriveStatus,omitempty"` } // +kubebuilder:object:root=true diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index a055fc17..5637d0b5 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -564,6 +564,21 @@ func (in *DocumentProcessorStatus) DeepCopy() *DocumentProcessorStatus { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *GDriveFolderStatus) DeepCopyInto(out *GDriveFolderStatus) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new GDriveFolderStatus. +func (in *GDriveFolderStatus) DeepCopy() *GDriveFolderStatus { + if in == nil { + return nil + } + out := new(GDriveFolderStatus) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *GeminiEmbedding2Config) DeepCopyInto(out *GeminiEmbedding2Config) { *out = *in @@ -961,6 +976,11 @@ func (in *SourceCrawlerStatus) DeepCopyInto(out *SourceCrawlerStatus) { (*in)[i].DeepCopyInto(&(*out)[i]) } } + if in.GDriveStatus != nil { + in, out := &in.GDriveStatus, &out.GDriveStatus + *out = make([]GDriveFolderStatus, 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..e2446bfe 100644 --- a/config/crd/bases/operator.dataverse.redhat.com_sourcecrawlers.yaml +++ b/config/crd/bases/operator.dataverse.redhat.com_sourcecrawlers.yaml @@ -176,6 +176,17 @@ spec: filesProcessed: format: int64 type: integer + gdriveStatus: + items: + properties: + error: + type: string + url: + type: string + required: + - url + 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..dc3833c9 100644 --- a/internal/controller/sourcecrawler_controller.go +++ b/internal/controller/sourcecrawler_controller.go @@ -139,7 +139,18 @@ func (r *SourceCrawlerReconciler) Reconcile(ctx context.Context, req ctrl.Reques } storedFiles, err := source.SyncFilesToFilestore(ctx, r.fileStore) + + var gdriveStatus []operatorv1alpha1.GDriveFolderStatus + if gds, ok := source.(*unstructured.GDriveSource); ok { + gdriveStatus = buildGDriveStatus(gds, sourceCrawlerConfig.GoogleDriveConfig) + } + if err != nil { + if patchErr := controllerutils.StatusPatch(ctx, r.Client, sourceCrawlerCR, func() { + sourceCrawlerCR.Status.GDriveStatus = gdriveStatus + }); patchErr != nil { + logger.Error(patchErr, "failed to update SourceCrawler CR status with gdrive status") + } 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)) @@ -147,6 +158,7 @@ func (r *SourceCrawlerReconciler) Reconcile(ctx context.Context, req ctrl.Reques successMessage := fmt.Sprintf("successfully reconciled source crawler: %s", sourceCrawlerCR.Name) if err := controllerutils.StatusPatch(ctx, r.Client, sourceCrawlerCR, func() { sourceCrawlerCR.Status.FilesProcessed += int64(len(storedFiles)) + sourceCrawlerCR.Status.GDriveStatus = gdriveStatus sourceCrawlerCR.UpdateStatus(successMessage, nil) }); err != nil { logger.Error(err, "failed to update SourceCrawler CR status") @@ -295,6 +307,25 @@ func extractGDriveFolderID(rawURL string) (string, error) { return "", fmt.Errorf("could not extract folder ID from URL path: %s", u.Path) } +func buildGDriveStatus(gds *unstructured.GDriveSource, gdriveConfig *operatorv1alpha1.GoogleDriveConfig) []operatorv1alpha1.GDriveFolderStatus { + failedMap := make(map[string]string, len(gds.FailedRootFolders)) + for _, f := range gds.FailedRootFolders { + failedMap[f.FolderID] = f.Error + } + result := make([]operatorv1alpha1.GDriveFolderStatus, 0, len(gdriveConfig.Folders)) + for i, f := range gdriveConfig.Folders { + folderID := gds.FolderIDs[i] + status := operatorv1alpha1.GDriveFolderStatus{ + URL: f.URL, + } + if errMsg, failed := failedMap[folderID]; failed { + status.Error = errMsg + } + result = append(result, status) + } + return result +} + func (r *SourceCrawlerReconciler) handleError(ctx context.Context, sourceCrawlerCR *operatorv1alpha1.SourceCrawler, err error) error { logger := log.FromContext(ctx) logger.Error(err, "encountered error") diff --git a/pkg/unstructured/source.go b/pkg/unstructured/source.go index 2d31391e..61e33b50 100644 --- a/pkg/unstructured/source.go +++ b/pkg/unstructured/source.go @@ -236,6 +236,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 +250,7 @@ type GDriveSource struct { ConcurrentFolders int ConcurrentDownloads int OutputDir string + FailedRootFolders []FailedRootFolder } // Close releases resources held by the underlying clients. @@ -285,6 +291,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 } for _, record := range r.result.Records { @@ -307,6 +317,10 @@ func (g *GDriveSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.F } } + if len(g.FailedRootFolders) == len(g.FolderIDs) { + return nil, 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, diff --git a/test/e2e/unstructured_test.go b/test/e2e/unstructured_test.go index 92c3f274..7df19133 100644 --- a/test/e2e/unstructured_test.go +++ b/test/e2e/unstructured_test.go @@ -495,22 +495,35 @@ func TestUnstructuredDataLoad(t *testing.T) { } t.Log("UnstructuredDataPipeline is ready after S3 destination patch") - 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 != expectedCount { - t.Fatalf("expected %d embeddings files in destination bucket, got %d", expectedCount, foundCount) + 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) } t.Logf("Found %d embeddings files in destination bucket", foundCount)