Use concurrent informer for iam-folder-reconciler (#110987)

apps/iam/cmd/operator/config.go

@@ -31,7 +31,8 @@ type WebhookServerConfig struct {
 }
 
 type FolderReconcilerConfig struct {
-	Namespace string
+	Namespace            string
+	MaxConcurrentWorkers uint64
 }
 
 func LoadConfigFromEnv() (*Config, error) {
@@ -115,6 +116,16 @@ func LoadConfigFromEnv() (*Config, error) {
 	cfg.ZanzanaClient.ServerCertFile = os.Getenv("ZANZANA_SERVER_CERT_FILE")
 
 	cfg.FolderReconciler.Namespace = os.Getenv("FOLDER_RECONCILER_NAMESPACE")
+	maxConcurrentWorkersStr := os.Getenv("FOLDER_RECONCILER_MAX_CONCURRENT_WORKERS")
+	if maxConcurrentWorkersStr == "" {
+		cfg.FolderReconciler.MaxConcurrentWorkers = 20
+	} else {
+		maxConcurrentWorkers, err := strconv.ParseUint(maxConcurrentWorkersStr, 10, 64)
+		if err != nil {
+			return nil, fmt.Errorf("invalid FOLDER_RECONCILER_MAX_CONCURRENT_WORKERS '%s': %w", maxConcurrentWorkersStr, err)
+		}
+		cfg.FolderReconciler.MaxConcurrentWorkers = maxConcurrentWorkers
+	}
 
 	return &cfg, nil
 }

apps/iam/cmd/operator/main.go

@@ -69,6 +69,9 @@ func main() {
 	appCfg := app.AppConfig{
 		ZanzanaClientCfg:          cfg.ZanzanaClient,
 		FolderReconcilerNamespace: cfg.FolderReconciler.Namespace,
+		InformerConfig: app.InformerConfig{
+			MaxConcurrentWorkers: cfg.FolderReconciler.MaxConcurrentWorkers,
+		},
 	}
 
 	// Run

apps/iam/pkg/app/app.go

@@ -6,6 +6,8 @@ import (
 
 	"github.com/grafana/grafana-app-sdk/app"
 	"github.com/grafana/grafana-app-sdk/logging"
+	"github.com/grafana/grafana-app-sdk/operator"
+	"github.com/grafana/grafana-app-sdk/resource"
 	"github.com/grafana/grafana-app-sdk/simple"
 	foldersKind "github.com/grafana/grafana/apps/folder/pkg/apis/folder/v1beta1"
 	"github.com/grafana/grafana/apps/iam/pkg/reconcilers"
@@ -17,8 +19,13 @@ var appManifestData = app.ManifestData{
 	Group:   "iam.grafana.app",
 }
 
+type InformerConfig struct {
+	MaxConcurrentWorkers uint64
+}
+
 type AppConfig struct {
 	ZanzanaClientCfg          authz.ZanzanaClientConfig
+	InformerConfig            InformerConfig
 	FolderReconcilerNamespace string
 }
 
@@ -26,6 +33,35 @@ func Provider(appCfg app.SpecificConfig) app.Provider {
 	return simple.NewAppProvider(app.NewEmbeddedManifest(appManifestData), appCfg, New)
 }
 
+func generateInformerSupplier(informerConfig InformerConfig) simple.InformerSupplier {
+	return func(kind resource.Kind, clients resource.ClientGenerator, options operator.ListWatchOptions) (operator.Informer, error) {
+		client, err := clients.ClientFor(kind)
+		if err != nil {
+			return nil, err
+		}
+
+		informer, err := operator.NewKubernetesBasedInformer(
+			kind, client,
+			operator.KubernetesBasedInformerOptions{
+				ListWatchOptions: options,
+			},
+		)
+		if err != nil {
+			return nil, err
+		}
+
+		return operator.NewConcurrentInformer(
+			informer,
+			operator.ConcurrentInformerOptions{
+				MaxConcurrentWorkers: informerConfig.MaxConcurrentWorkers,
+				ErrorHandler: func(ctx context.Context, err error) {
+					logging.FromContext(ctx).With("error", err).Error("ConcurrentInformer processing error")
+				},
+			},
+		)
+	}
+}
+
 func New(cfg app.Config) (app.App, error) {
 	appSpecificConfig, ok := cfg.SpecificConfig.(AppConfig)
 	if !ok {
@@ -53,8 +89,8 @@ func New(cfg app.Config) (app.App, error) {
 		Name:       cfg.ManifestData.AppName,
 		KubeConfig: cfg.KubeConfig,
 		InformerConfig: simple.AppInformerConfig{
+			InformerSupplier: generateInformerSupplier(appSpecificConfig.InformerConfig),
 			ErrorHandler: func(ctx context.Context, err error) {
-				// FIXME: add your own error handling here
 				logging.FromContext(ctx).With("error", err).Error("Informer processing error")
 			},
 		},

pkg/operators/iam/zanzana_folder_reconciler.go

@@ -106,6 +106,8 @@ func buildIAMConfigFromSettings(cfg *setting.Cfg) (*iamConfig, error) {
 	}
 	iamCfg.AppConfig.ZanzanaClientCfg.URL = zanzanaURL
 
+	iamCfg.AppConfig.InformerConfig.MaxConcurrentWorkers = operatorSec.Key("max_concurrent_workers").MustUint64(20)
+
 	folderAppURL := operatorSec.Key("folder_app_url").MustString("")
 	if folderAppURL == "" {
 		return nil, fmt.Errorf("folder_app_url is required in [operator] section")