Surface gdrive folder status in SourceCrawler CR status

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

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

internal/controller/sourcecrawler_controller.go

@@ -139,14 +139,26 @@ 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))
 
 	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")

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,

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)