configure worker count

Signed-off-by: Frank Jogeleit <frank.jogeleit@web.de>

CHANGELOG.md

@@ -1,5 +1,13 @@
 # Changelog
 
+# 2.17.0
+* Policy Reporter
+    * Use metaclient to reduce informer memory usage
+    * Use workerqueue to control concurrent processing of PolicyReports
+    * Remove internal PolicyReport structures
+    * Make sqlite volume configurable [[#255](https://github.com/kyverno/policy-reporter/pull/255) by [monotek](https://github.com/monotek)]
+    * use defer to unlock when possible [[#259](https://github.com/kyverno/policy-reporter/pull/259) by [eddycharly](https://github.com/eddycharly)]
+
 # 2.16.0
 * Add `nameOverride` to all charts [[#254](https://github.com/kyverno/policy-reporter/pull/254) by [mjnagel](https://github.com/mjnagel)]
 

charts/policy-reporter/Chart.yaml

@@ -5,8 +5,8 @@ description: |
   It creates Prometheus Metrics and can send rule validation events to different targets like Loki, Elasticsearch, Slack or Discord
 
 type: application
-version: 2.17.0
-appVersion: 2.12.0
+version: 2.18.0
+appVersion: 2.13.0
 
 icon: https://github.com/kyverno/kyverno/raw/main/img/logo.png
 home: https://kyverno.github.io/policy-reporter

charts/policy-reporter/config.yaml

@@ -213,6 +213,8 @@ priorityMap:
   {{- toYaml . | nindent 2 }}
 {{- end }}
 
+worker: {{ .Values.worker }}
+
 metrics:
   mode: {{ .Values.metrics.mode }}
   {{- with .Values.metrics.filter }}

charts/policy-reporter/values.yaml

@@ -5,7 +5,7 @@ image:
   registry: ghcr.io
   repository: kyverno/policy-reporter
   pullPolicy: IfNotPresent
-  tag: 2.12.0
+  tag: 2.13.0
 
 imagePullSecrets: []
 
@@ -114,6 +114,9 @@ metrics:
 profiling:
   enabled: false
 
+# amount of queue workers for PolicyReport resource processing
+worker: 5
+
 # Filter PolicyReport resources to process
 reportFilter:
   namespaces:

cmd/run.go

@@ -106,8 +106,9 @@ func newRunCMD() *cobra.Command {
 			g.Go(func() error {
 				stop := make(chan struct{})
 				defer close(stop)
+				log.Printf("[INFO] start client with %d workers", c.WorkerCount)
 
-				return client.Run(stop)
+				return client.Run(c.WorkerCount, stop)
 			})
 
 			return g.Wait()
@@ -123,6 +124,7 @@ func newRunCMD() *cobra.Command {
 	cmd.PersistentFlags().BoolP("rest-enabled", "r", false, "Enable Policy Reporter's REST API")
 	cmd.PersistentFlags().Bool("profile", false, "Enable application profiling with pprof")
 	cmd.PersistentFlags().String("lease-name", "policy-reporter", "name of the LeaseLock")
+	cmd.PersistentFlags().Int("worker", 5, "amount of queue worker")
 
 	flag.Parse()
 

pkg/config/config.go

@@ -269,6 +269,7 @@ type Config struct {
 	Webhook        Webhook        `mapstructure:"webhook"`
 	API            API            `mapstructure:"api"`
 	Kubeconfig     string         `mapstructure:"kubeconfig"`
+	WorkerCount    int            `mapstructure:"worker"`
 	DBFile         string         `mapstructure:"dbfile"`
 	Metrics        Metrics        `mapstructure:"metrics"`
 	REST           REST           `mapstructure:"rest"`

pkg/config/load.go

@@ -41,6 +41,10 @@ func Load(cmd *cobra.Command) (*Config, error) {
 		v.BindPFlag("kubeconfig", flag)
 	}
 
+	if flag := cmd.Flags().Lookup("worker"); flag != nil {
+		v.BindPFlag("worker", flag)
+	}
+
 	if flag := cmd.Flags().Lookup("port"); flag != nil {
 		v.BindPFlag("api.port", flag)
 	}

pkg/kubernetes/policy_report_client.go

@@ -54,12 +54,12 @@ func (k *k8sPolicyReportClient) Sync(stopper chan struct{}) error {
 	return nil
 }
 
-func (k *k8sPolicyReportClient) Run(stopper chan struct{}) error {
+func (k *k8sPolicyReportClient) Run(worker int, stopper chan struct{}) error {
 	if err := k.Sync(stopper); err != nil {
 		return err
 	}
 
-	k.queue.Run(5, stopper)
+	k.queue.Run(worker, stopper)
 
 	return nil
 }

pkg/kubernetes/policy_report_client_test.go

@@ -44,7 +44,7 @@ func Test_PolicyReportWatcher(t *testing.T) {
 	client := kubernetes.NewPolicyReportClient(kclient, filter, queue)
 
 	go func() {
-		err := client.Run(stop)
+		err := client.Run(1, stop)
 		if err != nil {
 			t.Error(err)
 		}
@@ -94,7 +94,7 @@ func Test_ClusterPolicyReportWatcher(t *testing.T) {
 	client := kubernetes.NewPolicyReportClient(kclient, filter, queue)
 
 	go func() {
-		err := client.Run(stop)
+		err := client.Run(1, stop)
 		if err != nil {
 			t.Error(err)
 		}

pkg/report/client.go

@@ -11,7 +11,7 @@ type PolicyReportResultListener = func(v1alpha2.ReportInterface, v1alpha2.Policy
 // PolicyReportClient watches for PolicyReport Events and executes registered callback
 type PolicyReportClient interface {
 	// Run starts the informer and workerqueue
-	Run(stopper chan struct{}) error
+	Run(worker int, stopper chan struct{}) error
 	// Sync Report Informer and start watching for events
 	Sync(stopper chan struct{}) error
 	// HasSynced the configured PolicyReport