Skip unsupported file types during source crawl

internal/controller/sourcecrawler_controller.go

@@ -145,6 +145,10 @@ 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 := skippedUnsupportedFromSource(source); len(skipped) > 0 {
+		successMessage += fmt.Sprintf("; skipped %d unsupported file(s): %s",
+			len(skipped), formatSkippedFiles(skipped, 10))
+	}
 	if err := controllerutils.StatusPatch(ctx, r.Client, sourceCrawlerCR, func() {
 		sourceCrawlerCR.Status.FilesProcessed += int64(len(storedFiles))
 		sourceCrawlerCR.UpdateStatus(successMessage, nil)
@@ -308,6 +312,24 @@ func (r *SourceCrawlerReconciler) handleError(ctx context.Context, sourceCrawler
 	return reconcileErr
 }
 
+func skippedUnsupportedFromSource(source unstructured.DataSource) []string {
+	switch s := source.(type) {
+	case *unstructured.S3BucketSource:
+		return s.SkippedUnsupported
+	case *unstructured.GDriveSource:
+		return s.SkippedUnsupported
+	default:
+		return nil
+	}
+}
+
+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:

pkg/unstructured/file_types.go

@@ -0,0 +1,50 @@
+/*
+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 defaultFromFormats, excluding image.
+// Supported: docx, pptx, html, pdf, asciidoc, md, csv, xlsx.
+var SupportedFileExtensions = map[string]bool{
+	".docx":     true,
+	".pptx":     true,
+	".html":     true,
+	".htm":      true,
+	".pdf":      true,
+	".adoc":     true,
+	".asciidoc": true,
+	".md":       true,
+	".markdown": true,
+	".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 "(none)"
+	}
+	return ext
+}

pkg/unstructured/source.go

@@ -43,10 +43,11 @@ type DataSource interface {
 }
 
 type S3BucketSource struct {
-	S3Client  *s3.Client
-	Bucket    string
-	Prefix    string
-	OutputDir string
+	S3Client           *s3.Client
+	Bucket             string
+	Prefix             string
+	OutputDir          string
+	SkippedUnsupported []string
 }
 
 func (s *S3BucketSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.FileStore) ([]RawFileMetadata, error) {
@@ -57,6 +58,7 @@ func (s *S3BucketSource) SyncFilesToFilestore(ctx context.Context, fs *filestore
 		return nil, err
 	}
 
+	s.SkippedUnsupported = nil
 	storedFiles := []RawFileMetadata{}
 	errorList := map[string]error{}
 	sourceFileMap := map[string]bool{}
@@ -72,6 +74,16 @@ func (s *S3BucketSource) SyncFilesToFilestore(ctx context.Context, fs *filestore
 				"key", *object.Key, "sizeMB", *object.Size/(1<<20))
 			continue
 		}
+
+		if !IsSupportedFileType(*object.Key) {
+			logger.Info("skipping unsupported file type",
+				"file", *object.Key,
+				"extension", FileExtension(*object.Key),
+			)
+			s.SkippedUnsupported = append(s.SkippedUnsupported, *object.Key)
+			continue
+		}
+
 		file := RawFileMetadata{
 			FilePath: s.filestorePath(*object.Key),
 			UID:      *object.ETag,
@@ -245,6 +257,7 @@ type GDriveSource struct {
 	ConcurrentFolders   int
 	ConcurrentDownloads int
 	OutputDir           string
+	SkippedUnsupported  []string
 }
 
 // Close releases resources held by the underlying clients.
@@ -313,6 +326,7 @@ func (g *GDriveSource) SyncFilesToFilestore(ctx context.Context, fs *filestore.F
 	)
 
 	// Phase 2: Download files, fetch permissions, store to filestore
+	g.SkippedUnsupported = nil
 	var mu sync.Mutex
 	var storedFiles []RawFileMetadata
 	errorList := map[string]error{}
@@ -327,6 +341,18 @@ 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.SkippedUnsupported = append(g.SkippedUnsupported, record.FileName)
+			continue
+		}
+
 		currentFiles[record.FileID] = ext
 		dlGroup.Go(func() error {
 			filestorePath := path.Join(g.OutputDir, record.FileID+ext)