This is an automated email from the ASF dual-hosted git repository.
squakez pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-k.git
The following commit(s) were added to refs/heads/main by this push:
new 0bb1804e4 feat(platform): introducing concurrency vars
0bb1804e4 is described below
commit 0bb1804e404e7972a9bdd2b7b68e519e6ce4057e
Author: Pasquale Congiusti <[email protected]>
AuthorDate: Sat Nov 14 08:43:16 2026 +0100
feat(platform): introducing concurrency vars
* INTEGRATION_MAX_CONCURRENT_RECONCILES
* PIPE_MAX_CONCURRENT_RECONCILES
* CLIENT_QPS
* CLIENT_BURST
---
.../ROOT/pages/installation/advanced/advanced.adoc | 16 +++
pkg/cmd/operator/operator.go | 9 +-
.../integration/integration_controller.go | 4 +
pkg/controller/pipe/pipe_controller.go | 5 +-
pkg/platform/env_platform.go | 56 ++++++++--
pkg/platform/env_platform_test.go | 48 +++++++++
pkg/platform/operator.go | 35 +++++-
pkg/platform/operator_test.go | 120 +++++++++++++++++++++
8 files changed, 273 insertions(+), 20 deletions(-)
diff --git a/docs/modules/ROOT/pages/installation/advanced/advanced.adoc
b/docs/modules/ROOT/pages/installation/advanced/advanced.adoc
index c8503861f..0be19db94 100644
--- a/docs/modules/ROOT/pages/installation/advanced/advanced.adoc
+++ b/docs/modules/ROOT/pages/installation/advanced/advanced.adoc
@@ -72,4 +72,20 @@ The following environment variables can be configured on the
operator Deployment
|`KAMELET_CATALOG_DIR`
|`/tmp/kamelets/`
|Directory where Kamelet catalog files are extracted during installation.
+
+|`INTEGRATION_MAX_CONCURRENT_RECONCILES`
+|`1`
+|The maximum number of concurrent reconciliations executed by the Integration
controller.
+
+|`PIPE_MAX_CONCURRENT_RECONCILES`
+|`1`
+|The maximum number of concurrent reconciliations executed by the Pipe
controller.
+
+|`CLIENT_QPS`
+|`20`
+|Maximum Queries Per Second (QPS) used by the operator REST client for
client-side rate limiting.
+
+|`CLIENT_BURST`
+|`200`
+|Maximum burst rate used by the operator REST client for client-side rate
limiting.
|===
diff --git a/pkg/cmd/operator/operator.go b/pkg/cmd/operator/operator.go
index 490dc5dab..a801fbeea 100644
--- a/pkg/cmd/operator/operator.go
+++ b/pkg/cmd/operator/operator.go
@@ -140,11 +140,10 @@ func Run(healthPort, monitoringPort int32, leaderElection
bool, leaderElectionID
cfg, err := config.GetConfig()
exitOnError(err, "cannot get client config")
- // Increase maximum burst that is used by client-side throttling,
- // to prevent the requests made to apply the bundled Kamelets
- // from being throttled.
- cfg.QPS = 20
- cfg.Burst = 200
+ // Increase maximum burst and QPS used by client-side throttling,
+ // to prevent API requests from being throttled during startup or high
reconciliation loads.
+ cfg.QPS = platform.GetClientQPS()
+ cfg.Burst = platform.GetClientBurst()
bootstrapClient, err := client.NewClientWithConfig(false, cfg)
exitOnError(err, "cannot initialize client")
diff --git a/pkg/controller/integration/integration_controller.go
b/pkg/controller/integration/integration_controller.go
index c45dd3213..e9e1fc200 100644
--- a/pkg/controller/integration/integration_controller.go
+++ b/pkg/controller/integration/integration_controller.go
@@ -33,6 +33,7 @@ import (
"k8s.io/utils/ptr"
"sigs.k8s.io/controller-runtime/pkg/builder"
ctrl "sigs.k8s.io/controller-runtime/pkg/client"
+ ctrlcontroller "sigs.k8s.io/controller-runtime/pkg/controller"
"sigs.k8s.io/controller-runtime/pkg/event"
"sigs.k8s.io/controller-runtime/pkg/handler"
"sigs.k8s.io/controller-runtime/pkg/manager"
@@ -286,6 +287,9 @@ func integrationPlatformEnqueueRequestsFromMapFunc(ctx
context.Context, c client
func add(mgr manager.Manager, c client.Client, r reconcile.Reconciler) error {
b := builder.ControllerManagedBy(mgr).
Named("integration-controller").
+ WithOptions(ctrlcontroller.Options{
+ MaxConcurrentReconciles:
platform.GetIntegrationMaxConcurrentReconciles(),
+ }).
// Watch for changes to primary resource Integration
For(&v1.Integration{}, builder.WithPredicates(
platform.FilteringFuncs[ctrl.Object]{
diff --git a/pkg/controller/pipe/pipe_controller.go
b/pkg/controller/pipe/pipe_controller.go
index b5891c5cf..2dfc76ebe 100644
--- a/pkg/controller/pipe/pipe_controller.go
+++ b/pkg/controller/pipe/pipe_controller.go
@@ -70,7 +70,10 @@ func newReconciler(mgr manager.Manager, c client.Client)
reconcile.Reconciler {
}
func add(mgr manager.Manager, r reconcile.Reconciler, c client.Client) error {
- ctrl, err := controller.New("pipe-controller", mgr,
controller.Options{Reconciler: r})
+ ctrl, err := controller.New("pipe-controller", mgr, controller.Options{
+ Reconciler: r,
+ MaxConcurrentReconciles:
platform.GetPipeMaxConcurrentReconciles(),
+ })
if err != nil {
return err
}
diff --git a/pkg/platform/env_platform.go b/pkg/platform/env_platform.go
index a5e56a49a..26bf0e1f3 100644
--- a/pkg/platform/env_platform.go
+++ b/pkg/platform/env_platform.go
@@ -105,6 +105,50 @@ func GetEnvOrDefault(key string, def string) string {
return def
}
+// GetEnvOrDefaultAsInt returns the integer value of the environment variable
if present and valid (> 0), or def otherwise.
+func GetEnvOrDefaultAsInt(key string, def int) int {
+ valStr := GetEnvOrDefault(key, "")
+ if valStr != "" {
+ val, err := strconv.Atoi(strings.TrimSpace(valStr))
+ if err != nil {
+ log.Errorf(err, "could not parse %s environment
variable, fallback to default value", key)
+
+ return def
+ }
+ if val <= 0 {
+ log.Errorf(nil, "%s environment variable must be
greater than 0, got %d, fallback to default value", key, val)
+
+ return def
+ }
+
+ return val
+ }
+
+ return def
+}
+
+// GetEnvOrDefaultAsFloat32 returns the float32 value of the environment
variable if present and valid (> 0), or def otherwise.
+func GetEnvOrDefaultAsFloat32(key string, def float32) float32 {
+ valStr := GetEnvOrDefault(key, "")
+ if valStr != "" {
+ val, err := strconv.ParseFloat(strings.TrimSpace(valStr), 32)
+ if err != nil {
+ log.Errorf(err, "could not parse %s environment
variable, fallback to default value", key)
+
+ return def
+ }
+ if val <= 0 {
+ log.Errorf(nil, "%s environment variable must be
greater than 0, got %v, fallback to default value", key, val)
+
+ return def
+ }
+
+ return float32(val)
+ }
+
+ return def
+}
+
func getEnvOrDefaultSlice(key string, def string) []string {
env, exists := os.LookupEnv(key)
if !exists {
@@ -115,17 +159,7 @@ func getEnvOrDefaultSlice(key string, def string) []string
{
}
func buildTimeout() time.Duration {
- buildTimeoutSeconds := GetEnvOrDefault("BUILD_TIMEOUT_SECONDS", "")
- if buildTimeoutSeconds != "" {
- seconds, err := strconv.Atoi(buildTimeoutSeconds)
- if err == nil {
- return time.Duration(seconds) * time.Second
- }
-
- log.Error(err, "could not parse BUILD_TIMEOUT_SECONDS
environment variable, fallback to default value")
- }
-
- return DefaultBuildTimeout
+ return time.Duration(GetEnvOrDefaultAsInt("BUILD_TIMEOUT_SECONDS",
int(DefaultBuildTimeout.Seconds()))) * time.Second
}
func buildStrategy() v1.BuildStrategy {
diff --git a/pkg/platform/env_platform_test.go
b/pkg/platform/env_platform_test.go
index a5a0e260d..dcdb8eb24 100644
--- a/pkg/platform/env_platform_test.go
+++ b/pkg/platform/env_platform_test.go
@@ -326,3 +326,51 @@ func TestFromIntegrationProfile_Overrides(t *testing.T) {
got.Maven.Repositories,
)
}
+
+func TestGetEnvOrDefaultAsInt(t *testing.T) {
+ t.Run("not set returns default", func(t *testing.T) {
+ assert.Equal(t, 42, GetEnvOrDefaultAsInt("TEST_UNSET_INT_VAR",
42))
+ })
+
+ t.Run("valid int value", func(t *testing.T) {
+ t.Setenv("TEST_INT_VAR", "10")
+ assert.Equal(t, 10, GetEnvOrDefaultAsInt("TEST_INT_VAR", 42))
+ })
+
+ t.Run("invalid non-numeric returns default", func(t *testing.T) {
+ t.Setenv("TEST_INT_VAR", "abc")
+ assert.Equal(t, 42, GetEnvOrDefaultAsInt("TEST_INT_VAR", 42))
+ })
+
+ t.Run("invalid non-positive returns default", func(t *testing.T) {
+ t.Setenv("TEST_INT_VAR", "0")
+ assert.Equal(t, 42, GetEnvOrDefaultAsInt("TEST_INT_VAR", 42))
+
+ t.Setenv("TEST_INT_VAR", "-5")
+ assert.Equal(t, 42, GetEnvOrDefaultAsInt("TEST_INT_VAR", 42))
+ })
+}
+
+func TestGetEnvOrDefaultAsFloat32(t *testing.T) {
+ t.Run("not set returns default", func(t *testing.T) {
+ assert.Equal(t, float32(3.14),
GetEnvOrDefaultAsFloat32("TEST_UNSET_FLOAT_VAR", 3.14))
+ })
+
+ t.Run("valid float value", func(t *testing.T) {
+ t.Setenv("TEST_FLOAT_VAR", "2.5")
+ assert.Equal(t, float32(2.5),
GetEnvOrDefaultAsFloat32("TEST_FLOAT_VAR", 3.14))
+ })
+
+ t.Run("invalid non-numeric returns default", func(t *testing.T) {
+ t.Setenv("TEST_FLOAT_VAR", "invalid")
+ assert.Equal(t, float32(3.14),
GetEnvOrDefaultAsFloat32("TEST_FLOAT_VAR", 3.14))
+ })
+
+ t.Run("invalid non-positive returns default", func(t *testing.T) {
+ t.Setenv("TEST_FLOAT_VAR", "0")
+ assert.Equal(t, float32(3.14),
GetEnvOrDefaultAsFloat32("TEST_FLOAT_VAR", 3.14))
+
+ t.Setenv("TEST_FLOAT_VAR", "-1.5")
+ assert.Equal(t, float32(3.14),
GetEnvOrDefaultAsFloat32("TEST_FLOAT_VAR", 3.14))
+ })
+}
diff --git a/pkg/platform/operator.go b/pkg/platform/operator.go
index df2ef9c74..e2ae42be6 100644
--- a/pkg/platform/operator.go
+++ b/pkg/platform/operator.go
@@ -35,9 +35,18 @@ import (
)
const (
- OperatorWatchNamespaceEnvVariable = "WATCH_NAMESPACE"
- operatorNamespaceEnvVariable = "NAMESPACE"
- OperatorBuildStrategyEnvVar = "BUILD_STRATEGY"
+ OperatorWatchNamespaceEnvVariable = "WATCH_NAMESPACE"
+ operatorNamespaceEnvVariable = "NAMESPACE"
+ OperatorBuildStrategyEnvVar = "BUILD_STRATEGY"
+ OperatorIntegrationMaxConcurrentReconcilesEnvVar =
"INTEGRATION_MAX_CONCURRENT_RECONCILES"
+ OperatorPipeMaxConcurrentReconcilesEnvVar =
"PIPE_MAX_CONCURRENT_RECONCILES"
+ OperatorClientQPSEnvVar = "CLIENT_QPS"
+ OperatorClientBurstEnvVar = "CLIENT_BURST"
+
+ defaultIntegrationMaxConcurrentReconciles = 1
+ defaultPipeMaxConcurrentReconciles = 1
+ defaultClientQPS = float32(20)
+ defaultClientBurst = 200
)
const OperatorLockName = "camel-k-lock"
@@ -245,4 +254,24 @@ func (f FilteringFuncs[T]) Generic(e
event.TypedGenericEvent[T]) bool {
return true
}
+// GetIntegrationMaxConcurrentReconciles returns the configured max concurrent
reconciles for the integration controller.
+func GetIntegrationMaxConcurrentReconciles() int {
+ return
GetEnvOrDefaultAsInt(OperatorIntegrationMaxConcurrentReconcilesEnvVar,
defaultIntegrationMaxConcurrentReconciles)
+}
+
+// GetPipeMaxConcurrentReconciles returns the configured max concurrent
reconciles for the pipe controller.
+func GetPipeMaxConcurrentReconciles() int {
+ return GetEnvOrDefaultAsInt(OperatorPipeMaxConcurrentReconcilesEnvVar,
defaultPipeMaxConcurrentReconciles)
+}
+
+// GetClientQPS returns the configured QPS for the operator client.
+func GetClientQPS() float32 {
+ return GetEnvOrDefaultAsFloat32(OperatorClientQPSEnvVar,
defaultClientQPS)
+}
+
+// GetClientBurst returns the configured Burst for the operator client.
+func GetClientBurst() int {
+ return GetEnvOrDefaultAsInt(OperatorClientBurstEnvVar,
defaultClientBurst)
+}
+
var _ predicate.Predicate = FilteringFuncs[ctrl.Object]{}
diff --git a/pkg/platform/operator_test.go b/pkg/platform/operator_test.go
new file mode 100644
index 000000000..17ef20a8d
--- /dev/null
+++ b/pkg/platform/operator_test.go
@@ -0,0 +1,120 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package platform
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+)
+
+func TestGetIntegrationMaxConcurrentReconciles(t *testing.T) {
+ t.Run("default value", func(t *testing.T) {
+ assert.Equal(t, 1, GetIntegrationMaxConcurrentReconciles())
+ })
+
+ t.Run("custom valid value", func(t *testing.T) {
+ t.Setenv(OperatorIntegrationMaxConcurrentReconcilesEnvVar, "5")
+ assert.Equal(t, 5, GetIntegrationMaxConcurrentReconciles())
+ })
+
+ t.Run("invalid non-numeric value", func(t *testing.T) {
+ t.Setenv(OperatorIntegrationMaxConcurrentReconcilesEnvVar,
"invalid")
+ assert.Equal(t, 1, GetIntegrationMaxConcurrentReconciles())
+ })
+
+ t.Run("invalid non-positive value", func(t *testing.T) {
+ t.Setenv(OperatorIntegrationMaxConcurrentReconcilesEnvVar, "0")
+ assert.Equal(t, 1, GetIntegrationMaxConcurrentReconciles())
+
+ t.Setenv(OperatorIntegrationMaxConcurrentReconcilesEnvVar, "-5")
+ assert.Equal(t, 1, GetIntegrationMaxConcurrentReconciles())
+ })
+}
+
+func TestGetPipeMaxConcurrentReconciles(t *testing.T) {
+ t.Run("default value", func(t *testing.T) {
+ assert.Equal(t, 1, GetPipeMaxConcurrentReconciles())
+ })
+
+ t.Run("custom valid value", func(t *testing.T) {
+ t.Setenv(OperatorPipeMaxConcurrentReconcilesEnvVar, "4")
+ assert.Equal(t, 4, GetPipeMaxConcurrentReconciles())
+ })
+
+ t.Run("invalid non-numeric value", func(t *testing.T) {
+ t.Setenv(OperatorPipeMaxConcurrentReconcilesEnvVar, "abc")
+ assert.Equal(t, 1, GetPipeMaxConcurrentReconciles())
+ })
+
+ t.Run("invalid non-positive value", func(t *testing.T) {
+ t.Setenv(OperatorPipeMaxConcurrentReconcilesEnvVar, "0")
+ assert.Equal(t, 1, GetPipeMaxConcurrentReconciles())
+
+ t.Setenv(OperatorPipeMaxConcurrentReconcilesEnvVar, "-2")
+ assert.Equal(t, 1, GetPipeMaxConcurrentReconciles())
+ })
+}
+
+func TestGetClientQPS(t *testing.T) {
+ t.Run("default value", func(t *testing.T) {
+ assert.Equal(t, float32(20), GetClientQPS())
+ })
+
+ t.Run("custom valid value", func(t *testing.T) {
+ t.Setenv(OperatorClientQPSEnvVar, "50")
+ assert.Equal(t, float32(50), GetClientQPS())
+ })
+
+ t.Run("invalid non-numeric value", func(t *testing.T) {
+ t.Setenv(OperatorClientQPSEnvVar, "not-a-number")
+ assert.Equal(t, float32(20), GetClientQPS())
+ })
+
+ t.Run("invalid non-positive value", func(t *testing.T) {
+ t.Setenv(OperatorClientQPSEnvVar, "0")
+ assert.Equal(t, float32(20), GetClientQPS())
+
+ t.Setenv(OperatorClientQPSEnvVar, "-10")
+ assert.Equal(t, float32(20), GetClientQPS())
+ })
+}
+
+func TestGetClientBurst(t *testing.T) {
+ t.Run("default value", func(t *testing.T) {
+ assert.Equal(t, 200, GetClientBurst())
+ })
+
+ t.Run("custom valid value", func(t *testing.T) {
+ t.Setenv(OperatorClientBurstEnvVar, "350")
+ assert.Equal(t, 350, GetClientBurst())
+ })
+
+ t.Run("invalid non-numeric value", func(t *testing.T) {
+ t.Setenv(OperatorClientBurstEnvVar, "bad")
+ assert.Equal(t, 200, GetClientBurst())
+ })
+
+ t.Run("invalid non-positive value", func(t *testing.T) {
+ t.Setenv(OperatorClientBurstEnvVar, "0")
+ assert.Equal(t, 200, GetClientBurst())
+
+ t.Setenv(OperatorClientBurstEnvVar, "-100")
+ assert.Equal(t, 200, GetClientBurst())
+ })
+}
