fix: increase concurrency limits for 500 pods fault injection (#1279)

- exec/model/parallelizer.go: change maxWorkers from const to configurable variable
- pkg/runtime/runtime.go:
  - add --max-workers flag (default 64)
  - update --max-concurrent-reconciles default from 20 to 50
  - increase QPS from 20 to 100
- pkg/controller/chaosblade/daemonset.go:
  - add resource requests (CPU 100m, Memory 128Mi)
  - add resource limits (CPU 1000m, Memory 512Mi)

Fixes: #1279

Change-Id: I66695d4661f6399bf7f95bd1e4f028bd4b103340

Author: xcaspar

exec/model/parallelizer.go

@@ -18,14 +18,15 @@ package model
 
 import "sync"
 
-const (
-	maxWorkers = 64 // magic number
+var (
+	// MaxWorkers can be configured via environment variable or flag
+	MaxWorkers = 64
 )
 
 type DoWorkFunc func(workID int)
 
 func ParallelizeExec(workCount int, doWork DoWorkFunc) {
-	workers := maxWorkers
+	workers := MaxWorkers
 	toExec := make(chan int, workCount)
 
 	for i := 0; i < workCount; i++ {

pkg/controller/chaosblade/daemonset.go

@@ -25,6 +25,7 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
@@ -176,5 +177,15 @@ func createContainer() corev1.Container {
 			{Name: "hosts", MountPath: "/etc/hosts"},
 		},
 		SecurityContext: &corev1.SecurityContext{Privileged: &trueVar},
+		Resources: corev1.ResourceRequirements{
+			Requests: corev1.ResourceList{
+				corev1.ResourceCPU:    resource.MustParse("100m"),
+				corev1.ResourceMemory: resource.MustParse("128Mi"),
+			},
+			Limits: corev1.ResourceList{
+				corev1.ResourceCPU:    resource.MustParse("1000m"),
+				corev1.ResourceMemory: resource.MustParse("512Mi"),
+			},
+		},
 	}
 }

pkg/runtime/runtime.go

@@ -30,13 +30,15 @@ var (
 	LogLevel                string
 	MaxConcurrentReconciles int
 	QPS                     float32
+	MaxWorkers              int
 )
 
 func init() {
 	flagSet = pflag.NewFlagSet("operator", pflag.ExitOnError)
 	flagSet.StringVar(&LogLevel, "log-level", "info", "Log level, such as panic|fatal|error|warn|info|debug|trace")
-	flagSet.IntVar(&MaxConcurrentReconciles, "reconcile-count", 20, "Max concurrent reconciles count, default value is 20")
-	flagSet.Float32Var(&QPS, "qps", 20, "qps of kubernetes client")
+	flagSet.IntVar(&MaxConcurrentReconciles, "max-concurrent-reconciles", 50, "Max concurrent reconciles count, default value is 50")
+	flagSet.Float32Var(&QPS, "qps", 100, "qps of kubernetes client, increased from 20 to 100 for better performance")
+	flagSet.IntVar(&MaxWorkers, "max-workers", 64, "Max workers for parallel execution, default value is 64")
 
 	flagSet.AddFlagSet(aliyun.FlagSet())
 	flagSet.AddFlagSet(chaosblade.FlagSet())