Prior to this commit QPS, threads-per-controller and maximum burst were not configurable when launching the Tekton controller. This meant that Tekton deployments could not be tuned to the environments it was deployed in.

This commit makes DefaultThreadsPerController, QPS, and Burst configurable via flags. This helps tune a deployment depending on expected concurrency in production.

Author: Pierre Tasci

cmd/controller/main.go

@@ -25,6 +25,8 @@ import (
 	"github.com/tektoncd/pipeline/pkg/reconciler/taskrun"
 	"github.com/tektoncd/pipeline/pkg/version"
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/rest"
+	"knative.dev/pkg/controller"
 	"knative.dev/pkg/injection"
 	"knative.dev/pkg/injection/sharedmain"
 	"knative.dev/pkg/signals"
@@ -48,6 +50,12 @@ var (
 	imageDigestExporterImage = flag.String("imagedigest-exporter-image", "", "The container image containing our image digest exporter binary.")
 	namespace                = flag.String("namespace", corev1.NamespaceAll, "Namespace to restrict informer to. Optional, defaults to all namespaces.")
 	versionGiven             = flag.String("version", "devel", "Version of Tekton running")
+	qps                      = flag.Int("kube-api-qps", int(rest.DefaultQPS), "Maximum QPS to the master from this client")
+	burst                    = flag.Int("kube-api-burst", rest.DefaultBurst, "Maximum burst for throttle")
+	threadsPerController     = flag.Int("threads-per-controller", controller.DefaultThreadsPerController, "Threads (goroutines) to create per controller")
+	disableHighAvailability  = flag.Bool("disable-ha", false, "Whether to disable high-availability functionality for this component.  This flag will be deprecated "+
+		"and removed when we have promoted this feature to stable, so do not pass it without filing an "+
+		"issue upstream!")
 )
 
 func main() {
@@ -68,7 +76,18 @@ func main() {
 	if err := images.Validate(); err != nil {
 		log.Fatal(err)
 	}
-	sharedmain.MainWithContext(injection.WithNamespaceScope(signals.NewContext(), *namespace), ControllerLogKey,
+	controller.DefaultThreadsPerController = *threadsPerController
+
+	cfg := sharedmain.ParseAndGetConfigOrDie()
+	// multiply by 2, no of controllers being created
+	cfg.QPS = 2 * float32(*qps)
+	cfg.Burst = 2 * *burst
+
+	ctx := injection.WithNamespaceScope(signals.NewContext(), *namespace)
+	if !*disableHighAvailability {
+		ctx = sharedmain.WithHADisabled(ctx)
+	}
+	sharedmain.MainWithConfig(ctx, ControllerLogKey, cfg,
 		taskrun.NewController(*namespace, images),
 		pipelinerun.NewController(*namespace, images),
 	)

pkg/pod/entrypoint.go

@@ -209,7 +209,7 @@ func StopSidecars(nopImage string, kubeclient kubernetes.Interface, pod corev1.P
 	}
 	if updated {
 		if _, err := kubeclient.CoreV1().Pods(newPod.Namespace).Update(newPod); err != nil {
-			return fmt.Errorf("error adding ready annotation to Pod %q: %w", pod.Name, err)
+			return fmt.Errorf("error stopping sidecars of Pod %q: %w", pod.Name, err)
 		}
 	}
 	return nil

pkg/reconciler/taskrun/taskrun.go

@@ -122,6 +122,7 @@ func (c *Reconciler) ReconcileKind(ctx context.Context, tr *v1beta1.TaskRun) pkg
 		c.timeoutHandler.Release(tr.GetNamespacedName())
 		pod, err := c.KubeClientSet.CoreV1().Pods(tr.Namespace).Get(tr.Status.PodName, metav1.GetOptions{})
 		if err == nil {
+			logger.Debugf("Stopping sidecars for TaskRun %q of Pod %q", tr.Name, tr.Status.PodName)
 			err = podconvert.StopSidecars(c.Images.NopImage, c.KubeClientSet, *pod)
 			if err == nil {
 				// Check if any SidecarStatuses are still shown as Running after stopping
@@ -504,7 +505,7 @@ func (c *Reconciler) handlePodCreationError(ctx context.Context, tr *v1beta1.Tas
 func (c *Reconciler) failTaskRun(ctx context.Context, tr *v1beta1.TaskRun, reason v1beta1.TaskRunReason, message string) error {
 	logger := logging.FromContext(ctx)
 
-	logger.Warn("stopping task run %q because of %q", tr.Name, reason)
+	logger.Warnf("stopping task run %q because of %q", tr.Name, reason)
 	tr.Status.MarkResourceFailed(reason, errors.New(message))
 
 	completionTime := metav1.Time{Time: time.Now()}