Skip to content

Increase controller concurrency for parallel reconciliation #273

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
concaf wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

Increase controller concurrency for parallel reconciliation #273

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 10 additions & 6 deletions cmd/main.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@ func main() {
var probeAddr string
var secureMetrics bool
var watchNamespace string
var maxConcurrentReconciles int

@concaf concaf Jul 4, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

figure out exponential backoff for all services

var enableHTTP2 bool
var tlsOpts []func(*tls.Config)
flag.StringVar(&metricsAddr, "metrics-bind-address", "0", "The address the metrics endpoint binds to. "+
Expand All @@ -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,
}
Expand Down Expand Up @@ -221,44 +225,44 @@ 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)
}

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)
}

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)
}
Expand Down
42 changes: 19 additions & 23 deletions internal/controller/chunksgenerator_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
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"
Expand All @@ -49,10 +50,6 @@
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
Expand Down Expand Up @@ -109,15 +106,13 @@
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)

@gemini-code-assist gemini-code-assist Bot Jul 3, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Instead of implementing manual validation in the controller to check if DependsOn is empty, rely on Kubernetes API validation markers (such as +kubebuilder:validation:MinItems=1 or CEL validation markers) at the CRD level. This allows the API server to reject invalid configurations before they reach the controller, preventing potential out-of-range panics when accessing DependsOn[0].

References
  1. Rely on Kubernetes API validation markers to enforce constraints on configuration fields, rather than implementing redundant manual checks within the controller logic.
  2. Use CEL validation markers at the CRD level to enforce type-configuration pairings and prevent nil pointer dereferences.

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)
Expand All @@ -129,7 +124,7 @@

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)
Expand Down Expand Up @@ -172,14 +167,14 @@
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
Expand All @@ -190,7 +185,7 @@
}

// 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
Expand All @@ -202,23 +197,23 @@
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
}

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) {

Check failure on line 208 in internal/controller/chunksgenerator_controller.go

View workflow job for this annotation

GitHub Actions / GolangCILint and YAMLLint 

unused-receiver: method receiver 'r' is not referenced in method's body, consider removing or renaming it as _ (revive)
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
}
Expand All @@ -232,7 +227,7 @@
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
}
Expand All @@ -241,7 +236,7 @@
}

// 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
}
Expand All @@ -266,12 +261,12 @@
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) {

Check failure on line 264 in internal/controller/chunksgenerator_controller.go

View workflow job for this annotation

GitHub Actions / GolangCILint and YAMLLint 

unused-receiver: method receiver 'r' is not referenced in method's body, consider removing or renaming it as _ (revive)
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
}
Expand Down Expand Up @@ -367,8 +362,9 @@
}

// 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{})).
Expand Down
12 changes: 6 additions & 6 deletions internal/controller/destinationsyncer_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand All @@ -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
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
}
Expand Down Expand Up @@ -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{})).
Expand Down
Loading
Loading