diff --git a/cmd/main.go b/cmd/main.go index 5c9dcbcf..ce7092ec 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -63,6 +63,7 @@ func main() { var probeAddr string var secureMetrics bool var watchNamespace string + var maxConcurrentReconciles int var enableHTTP2 bool var tlsOpts []func(*tls.Config) flag.StringVar(&metricsAddr, "metrics-bind-address", "0", "The address the metrics endpoint binds to. "+ @@ -84,6 +85,9 @@ func main() { "If set, HTTP/2 will be enabled for the metrics and webhook servers") flag.StringVar(&watchNamespace, "watch-namespace", "", "The namespace to watch for changes. If not specified, all namespaces will be watched.") + flag.IntVar(&maxConcurrentReconciles, "max-concurrent-reconciles", 10, + "Maximum number of concurrent reconciles per controller. "+ + "Does not apply to the ControllerConfig controller which always uses 1.") opts := zap.Options{ Development: true, } @@ -221,7 +225,7 @@ func main() { if err := (&controller.UnstructuredDataPipelineReconciler{ Client: mgr.GetClient(), Scheme: mgr.GetScheme(), - }).SetupWithManager(mgr); err != nil { + }).SetupWithManager(mgr, maxConcurrentReconciles); err != nil { setupLog.Error(err, "unable to create controller", "controller", "UnstructuredDataPipeline") os.Exit(1) } @@ -229,7 +233,7 @@ func main() { if err := (&controller.DocumentProcessorReconciler{ Client: mgr.GetClient(), Scheme: mgr.GetScheme(), - }).SetupWithManager(mgr); err != nil { + }).SetupWithManager(mgr, maxConcurrentReconciles); err != nil { setupLog.Error(err, "unable to create controller", "controller", "DocumentProcessor") os.Exit(1) } @@ -237,28 +241,28 @@ func main() { if err := (&controller.ChunksGeneratorReconciler{ Client: mgr.GetClient(), Scheme: mgr.GetScheme(), - }).SetupWithManager(mgr); err != nil { + }).SetupWithManager(mgr, maxConcurrentReconciles); err != nil { setupLog.Error(err, "unable to create controller", "controller", "ChunksGenerator") os.Exit(1) } if err := (&controller.VectorEmbeddingsGeneratorReconciler{ Client: mgr.GetClient(), Scheme: mgr.GetScheme(), - }).SetupWithManager(mgr); err != nil { + }).SetupWithManager(mgr, maxConcurrentReconciles); err != nil { setupLog.Error(err, "unable to create controller", "controller", "VectorEmbeddingsGenerator") os.Exit(1) } if err := (&controller.SourceCrawlerReconciler{ Client: mgr.GetClient(), Scheme: mgr.GetScheme(), - }).SetupWithManager(mgr); err != nil { + }).SetupWithManager(mgr, maxConcurrentReconciles); err != nil { setupLog.Error(err, "unable to create controller", "controller", "SourceCrawler") os.Exit(1) } if err := (&controller.DestinationSyncerReconciler{ Client: mgr.GetClient(), Scheme: mgr.GetScheme(), - }).SetupWithManager(mgr); err != nil { + }).SetupWithManager(mgr, maxConcurrentReconciles); err != nil { setupLog.Error(err, "unable to create controller", "controller", "DestinationSyncer") os.Exit(1) } diff --git a/internal/controller/chunksgenerator_controller.go b/internal/controller/chunksgenerator_controller.go index 503b1a1a..070ac6de 100644 --- a/internal/controller/chunksgenerator_controller.go +++ b/internal/controller/chunksgenerator_controller.go @@ -28,6 +28,7 @@ import ( ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" "sigs.k8s.io/controller-runtime/pkg/handler" "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/predicate" @@ -49,10 +50,6 @@ const ( type ChunksGeneratorReconciler struct { client.Client Scheme *runtime.Scheme - - fileStore *filestore.FileStore - inputPath string - outputPath string } // +kubebuilder:rbac:groups=operator.dataverse.redhat.com,namespace=unstructured-controller-namespace,resources=chunksgenerators,verbs=get;list;watch;create;update;patch;delete @@ -109,15 +106,13 @@ func (r *ChunksGeneratorReconciler) Reconcile(ctx context.Context, req ctrl.Requ logger.Error(err, "failed to create filestore") return r.handleError(ctx, chunksGeneratorCR, err) } - r.fileStore = fs - pipelineName, err := controllerutils.ParentPipelineNameFromOwnerReference(chunksGeneratorCR) if err != nil { return r.handleError(ctx, chunksGeneratorCR, err) } - r.inputPath = unstructured.StagePath(pipelineName, chunksGeneratorCR.Spec.DependsOn[0].Name) - r.outputPath = unstructured.StagePath(pipelineName, chunksGeneratorCR.Spec.StageName) - filePaths, err := r.fileStore.ListFilesInPath(ctx, r.inputPath) + inputPath := unstructured.StagePath(pipelineName, chunksGeneratorCR.Spec.DependsOn[0].Name) + outputPath := unstructured.StagePath(pipelineName, chunksGeneratorCR.Spec.StageName) + filePaths, err := fs.ListFilesInPath(ctx, inputPath) if err != nil { logger.Error(err, "failed to list files in path") return r.handleError(ctx, chunksGeneratorCR, err) @@ -129,7 +124,7 @@ func (r *ChunksGeneratorReconciler) Reconcile(ctx context.Context, req ctrl.Requ for _, convertedFilePath := range filePaths { logger.Info("processing converted file", "file", convertedFilePath) - processed, err := r.processConvertedFile(ctx, convertedFilePath, chunksGeneratorCR) + processed, err := r.processConvertedFile(ctx, fs, inputPath, outputPath, convertedFilePath, chunksGeneratorCR) if err != nil { if strings.Contains(err.Error(), langchain.SemaphoreAcquireError) { logger.Error(err, "failed to process converted file, semaphore acquire error, will try again later", "file", convertedFilePath) @@ -172,14 +167,14 @@ func (r *ChunksGeneratorReconciler) Reconcile(ctx context.Context, req ctrl.Requ return ctrl.Result{}, nil } -func (r *ChunksGeneratorReconciler) processConvertedFile(ctx context.Context, convertedFilePath string, chunksGeneratorCR *operatorv1alpha1.ChunksGenerator) (bool, error) { +func (r *ChunksGeneratorReconciler) processConvertedFile(ctx context.Context, fs *filestore.FileStore, inputPath, outputPath, convertedFilePath string, chunksGeneratorCR *operatorv1alpha1.ChunksGenerator) (bool, error) { logger := log.FromContext(ctx) logger.Info("processing converted file", "file", convertedFilePath) - chunksFilePath := unstructured.RemapToOutputDir(convertedFilePath, r.inputPath, r.outputPath) + chunksFilePath := unstructured.RemapToOutputDir(convertedFilePath, inputPath, outputPath) // figure out if the file is already chunked - needsChunking, err := r.needsChunking(ctx, convertedFilePath, chunksGeneratorCR) + needsChunking, err := r.needsChunking(ctx, fs, inputPath, outputPath, convertedFilePath, chunksGeneratorCR) if err != nil { logger.Error(err, "failed to check if file needs chunking") return false, err @@ -190,7 +185,7 @@ func (r *ChunksGeneratorReconciler) processConvertedFile(ctx context.Context, co } // chunk the file - chunksFile, err := r.chunkFile(ctx, convertedFilePath, chunksGeneratorCR) + chunksFile, err := r.chunkFile(ctx, fs, convertedFilePath, chunksGeneratorCR) if err != nil { logger.Error(err, "failed to chunk file") return false, err @@ -202,7 +197,7 @@ func (r *ChunksGeneratorReconciler) processConvertedFile(ctx context.Context, co logger.Error(err, "failed to marshal chunks file") return false, err } - if err := r.fileStore.Store(ctx, chunksFilePath, chunksFileBytes); err != nil { + if err := fs.Store(ctx, chunksFilePath, chunksFileBytes); err != nil { logger.Error(err, "failed to store chunks file") return false, err } @@ -210,15 +205,15 @@ func (r *ChunksGeneratorReconciler) processConvertedFile(ctx context.Context, co return true, nil } -func (r *ChunksGeneratorReconciler) needsChunking(ctx context.Context, convertedFilePath string, chunksGeneratorCR *operatorv1alpha1.ChunksGenerator) (bool, error) { +func (r *ChunksGeneratorReconciler) needsChunking(ctx context.Context, fs *filestore.FileStore, inputPath, outputPath, convertedFilePath string, chunksGeneratorCR *operatorv1alpha1.ChunksGenerator) (bool, error) { logger := log.FromContext(ctx) logger.Info("checking if file needs chunking", "file", convertedFilePath) - chunksFilePath := unstructured.RemapToOutputDir(convertedFilePath, r.inputPath, r.outputPath) + chunksFilePath := unstructured.RemapToOutputDir(convertedFilePath, inputPath, outputPath) // fetch the converted file from the filestore // this will also make sure that the converted file exists in the filestore - convertedFileRaw, err := r.fileStore.Retrieve(ctx, convertedFilePath) + convertedFileRaw, err := fs.Retrieve(ctx, convertedFilePath) if err != nil { return false, err } @@ -232,7 +227,7 @@ func (r *ChunksGeneratorReconciler) needsChunking(ctx context.Context, converted convertedFileMetadata := convertedFile.ConvertedDocument.Metadata // check if the chunked file does not exist in the filestore then return true - chunksFileExists, err := r.fileStore.Exists(ctx, chunksFilePath) + chunksFileExists, err := fs.Exists(ctx, chunksFilePath) if err != nil { return false, err } @@ -241,7 +236,7 @@ func (r *ChunksGeneratorReconciler) needsChunking(ctx context.Context, converted } // if the chunked file exists, then we need to check if it has the same configuration - chunksFileRaw, err := r.fileStore.Retrieve(ctx, chunksFilePath) + chunksFileRaw, err := fs.Retrieve(ctx, chunksFilePath) if err != nil { return false, err } @@ -266,12 +261,12 @@ func (r *ChunksGeneratorReconciler) needsChunking(ctx context.Context, converted return false, nil } -func (r *ChunksGeneratorReconciler) chunkFile(ctx context.Context, convertedFilePath string, chunksGeneratorCR *operatorv1alpha1.ChunksGenerator) (*unstructured.ChunksFile, error) { +func (r *ChunksGeneratorReconciler) chunkFile(ctx context.Context, fs *filestore.FileStore, convertedFilePath string, chunksGeneratorCR *operatorv1alpha1.ChunksGenerator) (*unstructured.ChunksFile, error) { logger := log.FromContext(ctx) logger.Info("chunking file", "file", convertedFilePath) // read the converted file from the filestore - convertedFileRaw, err := r.fileStore.Retrieve(ctx, convertedFilePath) + convertedFileRaw, err := fs.Retrieve(ctx, convertedFilePath) if err != nil { return nil, err } @@ -367,8 +362,9 @@ func (r *ChunksGeneratorReconciler) findDependents(ctx context.Context, obj clie } // SetupWithManager sets up the controller with the Manager. -func (r *ChunksGeneratorReconciler) SetupWithManager(mgr ctrl.Manager) error { +func (r *ChunksGeneratorReconciler) SetupWithManager(mgr ctrl.Manager, maxConcurrentReconciles int) error { return ctrl.NewControllerManagedBy(mgr). + WithOptions(controller.Options{MaxConcurrentReconciles: maxConcurrentReconciles}). For(&operatorv1alpha1.ChunksGenerator{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). Watches(&operatorv1alpha1.SourceCrawler{}, handler.EnqueueRequestsFromMapFunc(r.findDependents), builder.WithPredicates(controllerutils.FilesProcessedChangedPredicate{})). Watches(&operatorv1alpha1.DocumentProcessor{}, handler.EnqueueRequestsFromMapFunc(r.findDependents), builder.WithPredicates(controllerutils.FilesProcessedChangedPredicate{})). diff --git a/internal/controller/destinationsyncer_controller.go b/internal/controller/destinationsyncer_controller.go index b9baf7db..82a46add 100644 --- a/internal/controller/destinationsyncer_controller.go +++ b/internal/controller/destinationsyncer_controller.go @@ -27,6 +27,7 @@ import ( ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" "sigs.k8s.io/controller-runtime/pkg/handler" "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/predicate" @@ -46,8 +47,7 @@ const ( // DestinationSyncerReconciler reconciles a DestinationSyncer object type DestinationSyncerReconciler struct { client.Client - Scheme *runtime.Scheme - fileStore *filestore.FileStore + Scheme *runtime.Scheme } // +kubebuilder:rbac:groups=operator.dataverse.redhat.com,namespace=unstructured-controller-namespace,resources=destinationsyncers,verbs=get;list;watch;create;update;patch;delete @@ -92,7 +92,6 @@ func (r *DestinationSyncerReconciler) Reconcile(ctx context.Context, req ctrl.Re logger.Error(err, "failed to create filestore") return r.handleError(ctx, destinationSyncCR, err) } - r.fileStore = fs pipelineName, err := controllerutils.ParentPipelineNameFromOwnerReference(destinationSyncCR) if err != nil { return r.handleError(ctx, destinationSyncCR, err) @@ -120,14 +119,14 @@ func (r *DestinationSyncerReconciler) Reconcile(ctx context.Context, req ctrl.Re } inputPath := unstructured.StagePath(pipelineName, dep.Name) - filePaths, err := r.fileStore.ListFilesInPath(ctx, inputPath) + filePaths, err := fs.ListFilesInPath(ctx, inputPath) if err != nil { logger.Error(err, "failed to list files in path", "stage", dep.Name) return r.handleError(ctx, destinationSyncCR, err) } logger.Info("files to ingest to destination", "stage", dep.Name, "count", len(filePaths)) - if err := destination.SyncFilesToDestination(ctx, r.fileStore, filePaths); err != nil { + if err := destination.SyncFilesToDestination(ctx, fs, filePaths); err != nil { logger.Error(err, "failed to ingest files to destination", "stage", dep.Name) return r.handleError(ctx, destinationSyncCR, err) } @@ -202,8 +201,9 @@ func (r *DestinationSyncerReconciler) findSecretDependents(ctx context.Context, } // SetupWithManager sets up the controller with the Manager. -func (r *DestinationSyncerReconciler) SetupWithManager(mgr ctrl.Manager) error { +func (r *DestinationSyncerReconciler) SetupWithManager(mgr ctrl.Manager, maxConcurrentReconciles int) error { return ctrl.NewControllerManagedBy(mgr). + WithOptions(controller.Options{MaxConcurrentReconciles: maxConcurrentReconciles}). For(&operatorv1alpha1.DestinationSyncer{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). Watches(&operatorv1alpha1.SourceCrawler{}, handler.EnqueueRequestsFromMapFunc(r.findDependents), builder.WithPredicates(controllerutils.FilesProcessedChangedPredicate{})). Watches(&operatorv1alpha1.DocumentProcessor{}, handler.EnqueueRequestsFromMapFunc(r.findDependents), builder.WithPredicates(controllerutils.FilesProcessedChangedPredicate{})). diff --git a/internal/controller/documentprocessor_controller.go b/internal/controller/documentprocessor_controller.go index 0b82f534..557aa3ba 100644 --- a/internal/controller/documentprocessor_controller.go +++ b/internal/controller/documentprocessor_controller.go @@ -29,6 +29,7 @@ import ( ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" "sigs.k8s.io/controller-runtime/pkg/handler" "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/predicate" @@ -54,11 +55,6 @@ var ( type DocumentProcessorReconciler struct { client.Client Scheme *runtime.Scheme - - doclingConfig *docling.DoclingConfig - fileStore *filestore.FileStore - inputPath string - outputPath string } // +kubebuilder:rbac:groups=operator.dataverse.redhat.com,namespace=unstructured-controller-namespace,resources=documentprocessors,verbs=get;list;watch;create;update;patch;delete @@ -95,7 +91,7 @@ func (r *DocumentProcessorReconciler) Reconcile(ctx context.Context, req ctrl.Re return ctrl.Result{}, err } - r.doclingConfig = &docling.DoclingConfig{ + doclingCfg := &docling.DoclingConfig{ FromFormats: documentProcessorCR.Spec.DocumentProcessorConfig.DoclingConfig.FromFormats, ImageExportMode: documentProcessorCR.Spec.DocumentProcessorConfig.DoclingConfig.ImageExportMode, DoOCR: documentProcessorCR.Spec.DocumentProcessorConfig.DoclingConfig.DoOCR, @@ -115,26 +111,26 @@ func (r *DocumentProcessorReconciler) Reconcile(ctx context.Context, req ctrl.Re logger.Error(err, "failed to create filestore") return r.handleError(ctx, documentProcessorCR, err) } - r.fileStore = fs + + // read from upstream stage directory, write to own stage directory + pipelineName, err := controllerutils.ParentPipelineNameFromOwnerReference(documentProcessorCR) + if err != nil { + return r.handleError(ctx, documentProcessorCR, err) + } + inputPath := unstructured.StagePath(pipelineName, documentProcessorCR.Spec.DependsOn[0].Name) + outputPath := unstructured.StagePath(pipelineName, documentProcessorCR.Spec.StageName) // first, let's figure out the jobs that are currently running jobProcessingErrors := []error{} for _, job := range documentProcessorCR.Status.Jobs { logger.Info("reconciling job", "job", job) - if err := r.reconcileJob(ctx, job, documentProcessorCR); err != nil { + if err := r.reconcileJob(ctx, fs, inputPath, outputPath, job, documentProcessorCR); err != nil { jobProcessingErrors = append(jobProcessingErrors, err) logger.Error(err, "failed to process job", "job", job.FilePath) } } - // read from upstream stage directory, write to own stage directory - pipelineName, err := controllerutils.ParentPipelineNameFromOwnerReference(documentProcessorCR) - if err != nil { - return r.handleError(ctx, documentProcessorCR, err) - } - r.inputPath = unstructured.StagePath(pipelineName, documentProcessorCR.Spec.DependsOn[0].Name) - r.outputPath = unstructured.StagePath(pipelineName, documentProcessorCR.Spec.StageName) - filePaths, err := r.fileStore.ListFilesInPath(ctx, r.inputPath) + filePaths, err := fs.ListFilesInPath(ctx, inputPath) if err != nil { logger.Error(err, "failed to list files in path") return r.handleError(ctx, documentProcessorCR, err) @@ -144,7 +140,7 @@ func (r *DocumentProcessorReconciler) Reconcile(ctx context.Context, req ctrl.Re rawFilePaths := unstructured.FilterRawFilePaths(filePaths) for _, rawFilePath := range rawFilePaths { logger.Info("processing document", "document", rawFilePath) - if err := r.processDocument(ctx, rawFilePath, documentProcessorCR); err != nil { + if err := r.processDocument(ctx, fs, inputPath, outputPath, doclingCfg, rawFilePath, documentProcessorCR); err != nil { documentProcessingErrors = append(documentProcessingErrors, err) logger.Error(err, "failed to process document", "document", rawFilePath) } @@ -175,7 +171,7 @@ func (r *DocumentProcessorReconciler) Reconcile(ctx context.Context, req ctrl.Re return ctrl.Result{}, nil } -func (r *DocumentProcessorReconciler) reconcileJob(ctx context.Context, job operatorv1alpha1.Job, documentProcessorCR *operatorv1alpha1.DocumentProcessor) (err error) { +func (r *DocumentProcessorReconciler) reconcileJob(ctx context.Context, fs *filestore.FileStore, inputPath, outputPath string, job operatorv1alpha1.Job, documentProcessorCR *operatorv1alpha1.DocumentProcessor) (err error) { logger := log.FromContext(ctx) // recover from panic semaphore panic defer func() { @@ -228,8 +224,8 @@ func (r *DocumentProcessorReconciler) reconcileJob(ctx context.Context, job oper if err != nil { return err } - convertedFilePath := unstructured.RemapToOutputDir(job.FilePath+".json", r.inputPath, r.outputPath) - if err := r.fileStore.Store(ctx, convertedFilePath, convertedFileBytes); err != nil { + convertedFilePath := unstructured.RemapToOutputDir(job.FilePath+".json", inputPath, outputPath) + if err := fs.Store(ctx, convertedFilePath, convertedFileBytes); err != nil { return err } logger.Info("successfully stored the converted file in the filestore", "filePath", convertedFilePath) @@ -271,12 +267,12 @@ func (r *DocumentProcessorReconciler) reconcileJob(ctx context.Context, job oper return nil } -func (r *DocumentProcessorReconciler) processDocument(ctx context.Context, rawFilePath string, documentProcessorCR *operatorv1alpha1.DocumentProcessor) error { +func (r *DocumentProcessorReconciler) processDocument(ctx context.Context, fs *filestore.FileStore, inputPath, outputPath string, doclingCfg *docling.DoclingConfig, rawFilePath string, documentProcessorCR *operatorv1alpha1.DocumentProcessor) error { logger := log.FromContext(ctx) logger.Info("processing document", "rawFilePath", rawFilePath) // check if the document needs to be converted - needsConversion, err := r.needsConversion(ctx, rawFilePath, documentProcessorCR) + needsConversion, err := r.needsConversion(ctx, fs, inputPath, outputPath, rawFilePath, documentProcessorCR) if err != nil { logger.Error(err, "failed to check if document needs conversion") return err @@ -287,19 +283,19 @@ func (r *DocumentProcessorReconciler) processDocument(ctx context.Context, rawFi } // fetch the File UID - fileUID, err := r.getFileUID(ctx, rawFilePath) + fileUID, err := r.getFileUID(ctx, fs, rawFilePath) if err != nil { logger.Error(err, "Failed to get file UID") return err } // create a new job for the document - fileURL, err := r.fileStore.GetFileURL(ctx, rawFilePath) + fileURL, err := fs.GetFileURL(ctx, rawFilePath) if err != nil { logger.Error(err, "failed to get file URL") return err } - response, err := doclingClient.ConvertFile(ctx, fileURL, *r.doclingConfig) + response, err := doclingClient.ConvertFile(ctx, fileURL, *doclingCfg) if err != nil { logger.Error(err, "failed to convert file") if strings.Contains(err.Error(), docling.SemaphoreAcquireError) { @@ -337,13 +333,13 @@ func (r *DocumentProcessorReconciler) processDocument(ctx context.Context, rawFi return nil } -func (r *DocumentProcessorReconciler) needsConversion(ctx context.Context, rawFilePath string, documentProcessorCR *operatorv1alpha1.DocumentProcessor) (bool, error) { +func (r *DocumentProcessorReconciler) needsConversion(ctx context.Context, fs *filestore.FileStore, inputPath, outputPath, rawFilePath string, documentProcessorCR *operatorv1alpha1.DocumentProcessor) (bool, error) { logger := log.FromContext(ctx) logger.Info("checking if document should be converted", "filePath", rawFilePath) // does the raw file even exist? this is unlikely to happen, but just in case - rawFileExists, err := r.fileStore.Exists(ctx, rawFilePath) + rawFileExists, err := fs.Exists(ctx, rawFilePath) if err != nil { return false, err } @@ -354,7 +350,7 @@ func (r *DocumentProcessorReconciler) needsConversion(ctx context.Context, rawFi } // check whether metadata file exists - rawMetadataFileExists, err := r.fileStore.Exists(ctx, unstructured.MetadataPath(rawFilePath)) + rawMetadataFileExists, err := fs.Exists(ctx, unstructured.MetadataPath(rawFilePath)) if err != nil { return false, err } @@ -364,21 +360,21 @@ func (r *DocumentProcessorReconciler) needsConversion(ctx context.Context, rawFi } // fetch the File UID - fileUID, err := r.getFileUID(ctx, rawFilePath) + fileUID, err := r.getFileUID(ctx, fs, rawFilePath) if err != nil { logger.Error(err, "Failed to get file UID") return false, err } // does the converted file exist in the output directory? - convertedFilePath := unstructured.RemapToOutputDir(rawFilePath+".json", r.inputPath, r.outputPath) - convertedFileExists, err := r.fileStore.Exists(ctx, convertedFilePath) + convertedFilePath := unstructured.RemapToOutputDir(rawFilePath+".json", inputPath, outputPath) + convertedFileExists, err := fs.Exists(ctx, convertedFilePath) if err != nil { return false, err } if convertedFileExists { // if the converted file exists, then we need to check if it has the same configuration - convertedFileRaw, err := r.fileStore.Retrieve(ctx, convertedFilePath) + convertedFileRaw, err := fs.Retrieve(ctx, convertedFilePath) if err != nil { return false, err } @@ -458,9 +454,9 @@ func (r *DocumentProcessorReconciler) needsConversion(ctx context.Context, rawFi return true, nil } -func (r *DocumentProcessorReconciler) getFileUID(ctx context.Context, rawFilePath string) (string, error) { +func (r *DocumentProcessorReconciler) getFileUID(ctx context.Context, fs *filestore.FileStore, rawFilePath string) (string, error) { logger := log.FromContext(ctx) - metaDataFileRaw, err := r.fileStore.Retrieve(ctx, unstructured.MetadataPath(rawFilePath)) + metaDataFileRaw, err := fs.Retrieve(ctx, unstructured.MetadataPath(rawFilePath)) if err != nil { logger.Error(err, "Failed to retrieve metadata file") return "", err @@ -503,8 +499,9 @@ func (r *DocumentProcessorReconciler) findDependents(ctx context.Context, obj cl // SetupWithManager sets up the controller with the Manager. // SetupWithManager registers this controller. Spec changes trigger via GenerationChangedPredicate. // Watches on other stage types trigger reconcile when an upstream dependency's status changes. -func (r *DocumentProcessorReconciler) SetupWithManager(mgr ctrl.Manager) error { +func (r *DocumentProcessorReconciler) SetupWithManager(mgr ctrl.Manager, maxConcurrentReconciles int) error { return ctrl.NewControllerManagedBy(mgr). + WithOptions(controller.Options{MaxConcurrentReconciles: maxConcurrentReconciles}). For(&operatorv1alpha1.DocumentProcessor{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). Watches(&operatorv1alpha1.SourceCrawler{}, handler.EnqueueRequestsFromMapFunc(r.findDependents), builder.WithPredicates(controllerutils.FilesProcessedChangedPredicate{})). Watches(&operatorv1alpha1.ChunksGenerator{}, handler.EnqueueRequestsFromMapFunc(r.findDependents), builder.WithPredicates(controllerutils.FilesProcessedChangedPredicate{})). diff --git a/internal/controller/sourcecrawler_controller.go b/internal/controller/sourcecrawler_controller.go index 6c6496f9..1343fcc5 100644 --- a/internal/controller/sourcecrawler_controller.go +++ b/internal/controller/sourcecrawler_controller.go @@ -30,6 +30,7 @@ import ( ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" "sigs.k8s.io/controller-runtime/pkg/handler" "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/predicate" @@ -52,8 +53,7 @@ const ( // SourceCrawlerReconciler reconciles a SourceCrawler object type SourceCrawlerReconciler struct { client.Client - Scheme *runtime.Scheme - fileStore *filestore.FileStore + Scheme *runtime.Scheme } // +kubebuilder:rbac:groups=operator.dataverse.redhat.com,namespace=unstructured-controller-namespace,resources=sourcecrawlers,verbs=get;list;watch;create;update;patch;delete @@ -96,8 +96,6 @@ func (r *SourceCrawlerReconciler) Reconcile(ctx context.Context, req ctrl.Reques } return ctrl.Result{}, r.handleError(ctx, sourceCrawlerCR, fmt.Errorf("failed to create filestore: %w", err)) } - r.fileStore = fs - parentPipeline, err := controllerutils.ParentPipelineNameFromOwnerReference(sourceCrawlerCR) if err != nil { return ctrl.Result{}, r.handleError(ctx, sourceCrawlerCR, err) @@ -138,7 +136,7 @@ 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) + storedFiles, err := source.SyncFilesToFilestore(ctx, fs) if err != nil { return ctrl.Result{}, r.handleError(ctx, sourceCrawlerCR, fmt.Errorf("failed to store files to filestore: %w", err)) } @@ -360,8 +358,9 @@ func (r *SourceCrawlerReconciler) findSecretDependents(ctx context.Context, obj // SetupWithManager registers watches on all downstream pipeline stages and secrets so that // changes to any dependency trigger a reconcile of the owning SourceCrawler. -func (r *SourceCrawlerReconciler) SetupWithManager(mgr ctrl.Manager) error { +func (r *SourceCrawlerReconciler) SetupWithManager(mgr ctrl.Manager, maxConcurrentReconciles int) error { return ctrl.NewControllerManagedBy(mgr). + WithOptions(controller.Options{MaxConcurrentReconciles: maxConcurrentReconciles}). For(&operatorv1alpha1.SourceCrawler{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). Watches(&operatorv1alpha1.DocumentProcessor{}, handler.EnqueueRequestsFromMapFunc(r.findDependents), builder.WithPredicates(controllerutils.FilesProcessedChangedPredicate{})). Watches(&operatorv1alpha1.ChunksGenerator{}, handler.EnqueueRequestsFromMapFunc(r.findDependents), builder.WithPredicates(controllerutils.FilesProcessedChangedPredicate{})). diff --git a/internal/controller/unstructureddatapipeline_controller.go b/internal/controller/unstructureddatapipeline_controller.go index d1b80439..a2ad261a 100644 --- a/internal/controller/unstructureddatapipeline_controller.go +++ b/internal/controller/unstructureddatapipeline_controller.go @@ -29,6 +29,7 @@ import ( ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/predicate" @@ -342,8 +343,9 @@ func (r *UnstructuredDataPipelineReconciler) ensureChildCR(ctx context.Context, } // SetupWithManager sets up the controller with the Manager. -func (r *UnstructuredDataPipelineReconciler) SetupWithManager(mgr ctrl.Manager) error { +func (r *UnstructuredDataPipelineReconciler) SetupWithManager(mgr ctrl.Manager, maxConcurrentReconciles int) error { return ctrl.NewControllerManagedBy(mgr). + WithOptions(controller.Options{MaxConcurrentReconciles: maxConcurrentReconciles}). For(&operatorv1alpha1.UnstructuredDataPipeline{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). Owns(&operatorv1alpha1.SourceCrawler{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). Owns(&operatorv1alpha1.DocumentProcessor{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). diff --git a/internal/controller/vectorembeddingsgenerator_controller.go b/internal/controller/vectorembeddingsgenerator_controller.go index fecbeee3..4b8c6108 100644 --- a/internal/controller/vectorembeddingsgenerator_controller.go +++ b/internal/controller/vectorembeddingsgenerator_controller.go @@ -29,6 +29,7 @@ import ( ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/builder" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" "sigs.k8s.io/controller-runtime/pkg/handler" "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/predicate" @@ -47,10 +48,7 @@ const ( type VectorEmbeddingsGeneratorReconciler struct { client.Client - Scheme *runtime.Scheme - fileStore *filestore.FileStore - inputPath string - outputPath string + Scheme *runtime.Scheme } // +kubebuilder:rbac:groups=operator.dataverse.redhat.com,namespace=unstructured-controller-namespace,resources=vectorembeddingsgenerators,verbs=get;list;watch;create;update;patch;delete @@ -100,15 +98,14 @@ func (r *VectorEmbeddingsGeneratorReconciler) Reconcile(ctx context.Context, req logger.Error(err, "failed to create the filestore client") return r.handleError(ctx, vectorEmbeddingsGeneratorCR, err) } - r.fileStore = fs pipelineName, err := controllerutils.ParentPipelineNameFromOwnerReference(vectorEmbeddingsGeneratorCR) if err != nil { return r.handleError(ctx, vectorEmbeddingsGeneratorCR, err) } - r.inputPath = unstructured.StagePath(pipelineName, vectorEmbeddingsGeneratorCR.Spec.DependsOn[0].Name) - r.outputPath = unstructured.StagePath(pipelineName, vectorEmbeddingsGeneratorCR.Spec.StageName) - filePaths, err := r.fileStore.ListFilesInPath(ctx, r.inputPath) + inputPath := unstructured.StagePath(pipelineName, vectorEmbeddingsGeneratorCR.Spec.DependsOn[0].Name) + outputPath := unstructured.StagePath(pipelineName, vectorEmbeddingsGeneratorCR.Spec.StageName) + filePaths, err := fs.ListFilesInPath(ctx, inputPath) logger.Info("files in path", "count", len(filePaths)) if err != nil { logger.Error(err, "failed to list files in path") @@ -119,7 +116,7 @@ func (r *VectorEmbeddingsGeneratorReconciler) Reconcile(ctx context.Context, req var filesProcessed int64 for _, chunksFilePath := range filePaths { logger.Info("processing chunked file for embedding", "file", chunksFilePath) - processed, err := r.processChunkedFile(ctx, chunksFilePath, vectorEmbeddingsGeneratorCR) + processed, err := r.processChunkedFile(ctx, fs, inputPath, outputPath, chunksFilePath, vectorEmbeddingsGeneratorCR) if err != nil { embeddingErrors = append(embeddingErrors, err) logger.Error(err, "failed to process chunked file", "file", chunksFilePath) @@ -149,11 +146,11 @@ func (r *VectorEmbeddingsGeneratorReconciler) Reconcile(ctx context.Context, req return ctrl.Result{}, nil } -func (r *VectorEmbeddingsGeneratorReconciler) processChunkedFile(ctx context.Context, chunksFilePath string, vectorEmbeddingsGeneratorCR *operatorv1alpha1.VectorEmbeddingsGenerator) (bool, error) { +func (r *VectorEmbeddingsGeneratorReconciler) processChunkedFile(ctx context.Context, fs *filestore.FileStore, inputPath, outputPath, chunksFilePath string, vectorEmbeddingsGeneratorCR *operatorv1alpha1.VectorEmbeddingsGenerator) (bool, error) { logger := log.FromContext(ctx) logger.Info("processing chunked file", "chunksFilePath", chunksFilePath) - needsEmbedding, err := r.needsEmbedding(ctx, chunksFilePath, vectorEmbeddingsGeneratorCR) + needsEmbedding, err := r.needsEmbedding(ctx, fs, inputPath, outputPath, chunksFilePath, vectorEmbeddingsGeneratorCR) if err != nil { logger.Error(err, "failed to check if file needs embedding") return false, err @@ -164,7 +161,7 @@ func (r *VectorEmbeddingsGeneratorReconciler) processChunkedFile(ctx context.Con } logger.Info("retrieving chunked file from filestore", "file", chunksFilePath) - chunkedFileRaw, err := r.fileStore.Retrieve(ctx, chunksFilePath) + chunkedFileRaw, err := fs.Retrieve(ctx, chunksFilePath) if err != nil { logger.Error(err, "failed to retrieve chunked file") return false, err @@ -266,9 +263,9 @@ func (r *VectorEmbeddingsGeneratorReconciler) processChunkedFile(ctx context.Con return false, err } - embeddingsFilePath := unstructured.RemapToOutputDir(chunksFilePath, r.inputPath, r.outputPath) + embeddingsFilePath := unstructured.RemapToOutputDir(chunksFilePath, inputPath, outputPath) logger.Info("storing embedded file", "embeddingsFilePath", embeddingsFilePath) - if err := r.fileStore.Store(ctx, embeddingsFilePath, embeddingsFileBytes); err != nil { + if err := fs.Store(ctx, embeddingsFilePath, embeddingsFileBytes); err != nil { logger.Error(err, "failed to store embedded file") return false, err } @@ -277,11 +274,11 @@ func (r *VectorEmbeddingsGeneratorReconciler) processChunkedFile(ctx context.Con return true, nil } -func (r *VectorEmbeddingsGeneratorReconciler) needsEmbedding(ctx context.Context, chunksFilePath string, vectorEmbeddingsGeneratorCR *operatorv1alpha1.VectorEmbeddingsGenerator) (bool, error) { +func (r *VectorEmbeddingsGeneratorReconciler) needsEmbedding(ctx context.Context, fs *filestore.FileStore, inputPath, outputPath, chunksFilePath string, vectorEmbeddingsGeneratorCR *operatorv1alpha1.VectorEmbeddingsGenerator) (bool, error) { logger := log.FromContext(ctx) logger.Info("checking if file needs embedding", "file", chunksFilePath) - chunksFileExists, err := r.fileStore.Exists(ctx, chunksFilePath) + chunksFileExists, err := fs.Exists(ctx, chunksFilePath) if err != nil { return false, err } @@ -291,7 +288,7 @@ func (r *VectorEmbeddingsGeneratorReconciler) needsEmbedding(ctx context.Context return false, err } - chunkedFileRaw, err := r.fileStore.Retrieve(ctx, chunksFilePath) + chunkedFileRaw, err := fs.Retrieve(ctx, chunksFilePath) if err != nil { return false, err } @@ -301,15 +298,15 @@ func (r *VectorEmbeddingsGeneratorReconciler) needsEmbedding(ctx context.Context return false, err } - embeddingsFilePath := unstructured.RemapToOutputDir(chunksFilePath, r.inputPath, r.outputPath) + embeddingsFilePath := unstructured.RemapToOutputDir(chunksFilePath, inputPath, outputPath) logger.Info("embeddings file path", "embeddingsFilePath", embeddingsFilePath) - embeddingsFileExists, err := r.fileStore.Exists(ctx, embeddingsFilePath) + embeddingsFileExists, err := fs.Exists(ctx, embeddingsFilePath) if err != nil { return false, err } if embeddingsFileExists { - embeddingsFileRaw, err := r.fileStore.Retrieve(ctx, embeddingsFilePath) + embeddingsFileRaw, err := fs.Retrieve(ctx, embeddingsFilePath) if err != nil { return false, err } @@ -383,8 +380,9 @@ func (r *VectorEmbeddingsGeneratorReconciler) findDependents(ctx context.Context } // SetupWithManager sets up the controller with the Manager. -func (r *VectorEmbeddingsGeneratorReconciler) SetupWithManager(mgr ctrl.Manager) error { +func (r *VectorEmbeddingsGeneratorReconciler) SetupWithManager(mgr ctrl.Manager, maxConcurrentReconciles int) error { return ctrl.NewControllerManagedBy(mgr). + WithOptions(controller.Options{MaxConcurrentReconciles: maxConcurrentReconciles}). For(&operatorv1alpha1.VectorEmbeddingsGenerator{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). Watches(&operatorv1alpha1.SourceCrawler{}, handler.EnqueueRequestsFromMapFunc(r.findDependents), builder.WithPredicates(controllerutils.FilesProcessedChangedPredicate{})). Watches(&operatorv1alpha1.DocumentProcessor{}, handler.EnqueueRequestsFromMapFunc(r.findDependents), builder.WithPredicates(controllerutils.FilesProcessedChangedPredicate{})).