You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ac...@apache.org on 2023/04/26 07:08:37 UTC
[camel-k] 01/03: fix: Limit parallel builds on operator
This is an automated email from the ASF dual-hosted git repository.
acosentino pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-k.git
commit 76ad0fdc5426296a3e5699d36509b1385e80dab5
Author: Christoph Deppisch <cd...@redhat.com>
AuthorDate: Wed Apr 5 21:11:33 2023 +0200
fix: Limit parallel builds on operator
- Avoid many parallel integration builds
- Monitor all builds started by the operator instance and limit max number of running builds according to given setting
- By default use max running builds limit = 3 for build strategy routine
- By default use max running builds limit = 10 for build strategy pod
- Add max running builds setting to IntegrationPlatform
- Add some documentation on build strategy and build queues
---
config/crd/bases/camel.apache.org_builds.yaml | 5 +
.../camel.apache.org_integrationplatforms.yaml | 10 +
docs/modules/ROOT/pages/architecture/cr/build.adoc | 31 +++
docs/modules/ROOT/partials/apis/camel-k-crds.adoc | 14 ++
helm/camel-k/crds/crd-build.yaml | 5 +
helm/camel-k/crds/crd-integration-platform.yaml | 10 +
pkg/apis/camel/v1/build_types.go | 2 +
pkg/apis/camel/v1/integrationplatform_types.go | 2 +
.../camel/applyconfiguration/camel/v1/buildspec.go | 9 +
.../camel/v1/integrationplatformbuildspec.go | 9 +
pkg/cmd/install.go | 5 +
pkg/controller/build/build_controller.go | 8 +-
pkg/controller/build/build_monitor.go | 107 +++++++++
pkg/controller/build/build_monitor_test.go | 242 +++++++++++++++++++++
pkg/controller/build/monitor_pod.go | 11 +
pkg/controller/build/monitor_routine.go | 7 +
pkg/controller/build/schedule.go | 54 +----
pkg/controller/integrationkit/build.go | 1 +
pkg/platform/defaults.go | 8 +
pkg/resources/resources.go | 8 +-
20 files changed, 500 insertions(+), 48 deletions(-)
diff --git a/config/crd/bases/camel.apache.org_builds.yaml b/config/crd/bases/camel.apache.org_builds.yaml
index 033392e7d..58c050154 100644
--- a/config/crd/bases/camel.apache.org_builds.yaml
+++ b/config/crd/bases/camel.apache.org_builds.yaml
@@ -79,6 +79,11 @@ spec:
spec:
description: BuildSpec defines the Build operation to be executed
properties:
+ maxRunningBuilds:
+ description: the maximum amount of parallel running builds started
+ by this operator instance
+ format: int32
+ type: integer
operatorNamespace:
description: The namespace where to run the builder Pod (must be the
same of the operator in charge of this Build reconciliation).
diff --git a/config/crd/bases/camel.apache.org_integrationplatforms.yaml b/config/crd/bases/camel.apache.org_integrationplatforms.yaml
index 5589fbc0f..036d4dd22 100644
--- a/config/crd/bases/camel.apache.org_integrationplatforms.yaml
+++ b/config/crd/bases/camel.apache.org_integrationplatforms.yaml
@@ -242,6 +242,11 @@ spec:
type: object
type: object
type: object
+ maxRunningBuilds:
+ description: the maximum amount of parallel running builds started
+ by this operator instance
+ format: int32
+ type: integer
publishStrategy:
description: the strategy to adopt for publishing an Integration
base image
@@ -1792,6 +1797,11 @@ spec:
type: object
type: object
type: object
+ maxRunningBuilds:
+ description: the maximum amount of parallel running builds started
+ by this operator instance
+ format: int32
+ type: integer
publishStrategy:
description: the strategy to adopt for publishing an Integration
base image
diff --git a/docs/modules/ROOT/pages/architecture/cr/build.adoc b/docs/modules/ROOT/pages/architecture/cr/build.adoc
index 60cc9cae2..e7502616a 100644
--- a/docs/modules/ROOT/pages/architecture/cr/build.adoc
+++ b/docs/modules/ROOT/pages/architecture/cr/build.adoc
@@ -3,6 +3,8 @@
A *Build* resource, describes the process of assembling a container image that copes with the requirement of an xref:architecture/cr/integration.adoc[Integration] or xref:architecture/cr/integration-kit.adoc[IntegrationKit].
+The result of a build is an xref:architecture/cr/integration-kit.adoc[IntegrationKit] that can and should be reused for multiple xref:architecture/cr/integration.adoc[Integrations].
+
[source,go]
----
type Build struct {
@@ -25,3 +27,32 @@ the full go definition can be found https://github.com/apache/camel-k/blob/main/
image::architecture/camel-k-state-machine-build.png[life cycle]
+[[build-strategy]]
+= Build strategy
+
+You can choose from different build strategies. The build strategy defines how a build should be executed.
+At the moment the available strategies are:
+
+- buildStrategy: pod (each build is run in a separate pod, the operator monitors the pod state)
+- buildStrategy: routine (each build is run as a go routine inside the operator pod)
+
+[[build-queue]]
+= Build queues
+
+IntegrationKits and its base images should be reused for multiple Integrations in order to
+accomplish an efficient resource management and to optimize build and startup times for Camel K Integrations.
+
+In order to reuse images the operator is going to queue builds in sequential order.
+This way the operator is able to use efficient image layering for Integrations.
+
+By default, builds are queued sequentially based on their layout (e.g. native, fast-jar) and the build namespace.
+
+To avoid having many builds running in parallel the operator uses a maximum number of running builds setting that limits the
+amount of builds running.
+
+You can set this limit in the xref:architecture/cr/integration-platform.adoc[IntegrationPlatform] settings.
+
+The default values for this limitation is based on the build strategy.
+
+- buildStrategy: pod (MaxRunningBuilds=10)
+- buildStrategy: routine (MaxRunningBuilds=3)
diff --git a/docs/modules/ROOT/partials/apis/camel-k-crds.adoc b/docs/modules/ROOT/partials/apis/camel-k-crds.adoc
index 5a093cd9a..c98fce367 100644
--- a/docs/modules/ROOT/partials/apis/camel-k-crds.adoc
+++ b/docs/modules/ROOT/partials/apis/camel-k-crds.adoc
@@ -453,6 +453,13 @@ The Build deadline is set to the Build start time plus the Timeout duration.
If the Build deadline is exceeded, the Build context is canceled,
and its phase set to BuildPhaseFailed.
+|`maxRunningBuilds` +
+int32
+|
+
+
+the maximum amount of parallel running builds started by this operator instance
+
|===
@@ -1905,6 +1912,13 @@ map[string]string
Generic options that can used by each publish strategy
+|`maxRunningBuilds` +
+int32
+|
+
+
+the maximum amount of parallel running builds started by this operator instance
+
|===
diff --git a/helm/camel-k/crds/crd-build.yaml b/helm/camel-k/crds/crd-build.yaml
index 033392e7d..58c050154 100644
--- a/helm/camel-k/crds/crd-build.yaml
+++ b/helm/camel-k/crds/crd-build.yaml
@@ -79,6 +79,11 @@ spec:
spec:
description: BuildSpec defines the Build operation to be executed
properties:
+ maxRunningBuilds:
+ description: the maximum amount of parallel running builds started
+ by this operator instance
+ format: int32
+ type: integer
operatorNamespace:
description: The namespace where to run the builder Pod (must be the
same of the operator in charge of this Build reconciliation).
diff --git a/helm/camel-k/crds/crd-integration-platform.yaml b/helm/camel-k/crds/crd-integration-platform.yaml
index 5589fbc0f..036d4dd22 100644
--- a/helm/camel-k/crds/crd-integration-platform.yaml
+++ b/helm/camel-k/crds/crd-integration-platform.yaml
@@ -242,6 +242,11 @@ spec:
type: object
type: object
type: object
+ maxRunningBuilds:
+ description: the maximum amount of parallel running builds started
+ by this operator instance
+ format: int32
+ type: integer
publishStrategy:
description: the strategy to adopt for publishing an Integration
base image
@@ -1792,6 +1797,11 @@ spec:
type: object
type: object
type: object
+ maxRunningBuilds:
+ description: the maximum amount of parallel running builds started
+ by this operator instance
+ format: int32
+ type: integer
publishStrategy:
description: the strategy to adopt for publishing an Integration
base image
diff --git a/pkg/apis/camel/v1/build_types.go b/pkg/apis/camel/v1/build_types.go
index 32210306e..2618c4d64 100644
--- a/pkg/apis/camel/v1/build_types.go
+++ b/pkg/apis/camel/v1/build_types.go
@@ -41,6 +41,8 @@ type BuildSpec struct {
// and its phase set to BuildPhaseFailed.
// +kubebuilder:validation:Format=duration
Timeout metav1.Duration `json:"timeout,omitempty"`
+ // the maximum amount of parallel running builds started by this operator instance
+ MaxRunningBuilds int32 `json:"maxRunningBuilds,omitempty"`
}
// Task represents the abstract task. Only one of the task should be configured to represent the specific task chosen.
diff --git a/pkg/apis/camel/v1/integrationplatform_types.go b/pkg/apis/camel/v1/integrationplatform_types.go
index f63163094..2219f8c63 100644
--- a/pkg/apis/camel/v1/integrationplatform_types.go
+++ b/pkg/apis/camel/v1/integrationplatform_types.go
@@ -128,6 +128,8 @@ type IntegrationPlatformBuildSpec struct {
Maven MavenSpec `json:"maven,omitempty"`
// Generic options that can used by each publish strategy
PublishStrategyOptions map[string]string `json:"PublishStrategyOptions,omitempty"`
+ // the maximum amount of parallel running builds started by this operator instance
+ MaxRunningBuilds int32 `json:"maxRunningBuilds,omitempty"`
}
// IntegrationPlatformKameletSpec define the behavior for all the Kamelets controller by the IntegrationPlatform
diff --git a/pkg/client/camel/applyconfiguration/camel/v1/buildspec.go b/pkg/client/camel/applyconfiguration/camel/v1/buildspec.go
index 9b0c1f464..6ebf7927c 100644
--- a/pkg/client/camel/applyconfiguration/camel/v1/buildspec.go
+++ b/pkg/client/camel/applyconfiguration/camel/v1/buildspec.go
@@ -32,6 +32,7 @@ type BuildSpecApplyConfiguration struct {
ToolImage *string `json:"toolImage,omitempty"`
BuilderPodNamespace *string `json:"operatorNamespace,omitempty"`
Timeout *metav1.Duration `json:"timeout,omitempty"`
+ MaxRunningBuilds *int32 `json:"maxRunningBuilds,omitempty"`
}
// BuildSpecApplyConfiguration constructs an declarative configuration of the BuildSpec type for use with
@@ -84,3 +85,11 @@ func (b *BuildSpecApplyConfiguration) WithTimeout(value metav1.Duration) *BuildS
b.Timeout = &value
return b
}
+
+// WithMaxRunningBuilds sets the MaxRunningBuilds field in the declarative configuration to the given value
+// and returns the receiver, so that objects can be built by chaining "With" function invocations.
+// If called multiple times, the MaxRunningBuilds field is set to the value of the last call.
+func (b *BuildSpecApplyConfiguration) WithMaxRunningBuilds(value int32) *BuildSpecApplyConfiguration {
+ b.MaxRunningBuilds = &value
+ return b
+}
diff --git a/pkg/client/camel/applyconfiguration/camel/v1/integrationplatformbuildspec.go b/pkg/client/camel/applyconfiguration/camel/v1/integrationplatformbuildspec.go
index d884675af..2775d220a 100644
--- a/pkg/client/camel/applyconfiguration/camel/v1/integrationplatformbuildspec.go
+++ b/pkg/client/camel/applyconfiguration/camel/v1/integrationplatformbuildspec.go
@@ -37,6 +37,7 @@ type IntegrationPlatformBuildSpecApplyConfiguration struct {
Timeout *metav1.Duration `json:"timeout,omitempty"`
Maven *MavenSpecApplyConfiguration `json:"maven,omitempty"`
PublishStrategyOptions map[string]string `json:"PublishStrategyOptions,omitempty"`
+ MaxRunningBuilds *int32 `json:"maxRunningBuilds,omitempty"`
}
// IntegrationPlatformBuildSpecApplyConfiguration constructs an declarative configuration of the IntegrationPlatformBuildSpec type for use with
@@ -130,3 +131,11 @@ func (b *IntegrationPlatformBuildSpecApplyConfiguration) WithPublishStrategyOpti
}
return b
}
+
+// WithMaxRunningBuilds sets the MaxRunningBuilds field in the declarative configuration to the given value
+// and returns the receiver, so that objects can be built by chaining "With" function invocations.
+// If called multiple times, the MaxRunningBuilds field is set to the value of the last call.
+func (b *IntegrationPlatformBuildSpecApplyConfiguration) WithMaxRunningBuilds(value int32) *IntegrationPlatformBuildSpecApplyConfiguration {
+ b.MaxRunningBuilds = &value
+ return b
+}
diff --git a/pkg/cmd/install.go b/pkg/cmd/install.go
index 2f6f8bd16..5303c1945 100644
--- a/pkg/cmd/install.go
+++ b/pkg/cmd/install.go
@@ -197,6 +197,7 @@ type installCmdOptions struct {
MavenCASecret string `mapstructure:"maven-ca-secret"`
MavenCLIOptions []string `mapstructure:"maven-cli-options"`
HealthPort int32 `mapstructure:"health-port"`
+ MaxRunningBuilds int32 `mapstructure:"max-running-builds"`
Monitoring bool `mapstructure:"monitoring"`
MonitoringPort int32 `mapstructure:"monitoring-port"`
TraitProfile string `mapstructure:"trait-profile"`
@@ -539,6 +540,10 @@ func (o *installCmdOptions) setupIntegrationPlatform(
Duration: d,
}
}
+ if o.MaxRunningBuilds > 0 {
+ platform.Spec.Build.MaxRunningBuilds = o.MaxRunningBuilds
+ }
+
if o.TraitProfile != "" {
platform.Spec.Profile = v1.TraitProfileByName(o.TraitProfile)
}
diff --git a/pkg/controller/build/build_controller.go b/pkg/controller/build/build_controller.go
index e0172ab20..ba294e6e2 100644
--- a/pkg/controller/build/build_controller.go
+++ b/pkg/controller/build/build_controller.go
@@ -142,11 +142,15 @@ func (r *reconcileBuild) Reconcile(ctx context.Context, request reconcile.Reques
var actions []Action
+ buildMonitor := Monitor{
+ maxRunningBuilds: instance.Spec.MaxRunningBuilds,
+ }
+
switch instance.Spec.Strategy {
case v1.BuildStrategyPod:
actions = []Action{
newInitializePodAction(r.reader),
- newScheduleAction(r.reader),
+ newScheduleAction(r.reader, buildMonitor),
newMonitorPodAction(r.reader),
newErrorRecoveryAction(),
newErrorAction(),
@@ -154,7 +158,7 @@ func (r *reconcileBuild) Reconcile(ctx context.Context, request reconcile.Reques
case v1.BuildStrategyRoutine:
actions = []Action{
newInitializeRoutineAction(),
- newScheduleAction(r.reader),
+ newScheduleAction(r.reader, buildMonitor),
newMonitorRoutineAction(),
newErrorRecoveryAction(),
newErrorAction(),
diff --git a/pkg/controller/build/build_monitor.go b/pkg/controller/build/build_monitor.go
new file mode 100644
index 000000000..521dd6b86
--- /dev/null
+++ b/pkg/controller/build/build_monitor.go
@@ -0,0 +1,107 @@
+/*
+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 build
+
+import (
+ "context"
+ "sync"
+
+ "k8s.io/apimachinery/pkg/labels"
+ "k8s.io/apimachinery/pkg/selection"
+ "k8s.io/apimachinery/pkg/types"
+ ctrl "sigs.k8s.io/controller-runtime/pkg/client"
+
+ v1 "github.com/apache/camel-k/v2/pkg/apis/camel/v1"
+ "github.com/apache/camel-k/v2/pkg/util/kubernetes"
+)
+
+var runningBuilds sync.Map
+
+type Monitor struct {
+ maxRunningBuilds int32
+}
+
+func (bm *Monitor) canSchedule(ctx context.Context, c ctrl.Reader, build *v1.Build) (bool, error) {
+ var runningBuildsTotal int32
+ runningBuilds.Range(func(_, v interface{}) bool {
+ runningBuildsTotal++
+ return true
+ })
+
+ if runningBuildsTotal >= bm.maxRunningBuilds {
+ requestName := build.Name
+ requestNamespace := build.Namespace
+ buildCreator := kubernetes.GetCamelCreator(build)
+ if buildCreator != nil {
+ requestName = buildCreator.Name
+ requestNamespace = buildCreator.Namespace
+ }
+
+ Log.WithValues("request-namespace", requestNamespace, "request-name", requestName, "max-running-builds-limit", runningBuildsTotal).
+ ForBuild(build).Infof("Maximum number of running builds (%d) exceeded - the build gets enqueued", runningBuildsTotal)
+
+ // max number of running builds limit exceeded
+ return false, nil
+ }
+
+ layout := build.Labels[v1.IntegrationKitLayoutLabel]
+
+ // Native builds can be run in parallel, as incremental images is not applicable.
+ if layout == v1.IntegrationKitLayoutNative {
+ return true, nil
+ }
+
+ // We assume incremental images is only applicable across images whose layout is identical
+ withCompatibleLayout, err := labels.NewRequirement(v1.IntegrationKitLayoutLabel, selection.Equals, []string{layout})
+ if err != nil {
+ return false, err
+ }
+
+ builds := &v1.BuildList{}
+ // We use the non-caching client as informers cache is not invalidated nor updated
+ // atomically by write operations
+ err = c.List(ctx, builds,
+ ctrl.InNamespace(build.Namespace),
+ ctrl.MatchingLabelsSelector{
+ Selector: labels.NewSelector().Add(*withCompatibleLayout),
+ })
+ if err != nil {
+ return false, err
+ }
+
+ // Emulate a serialized working queue to only allow one build to run at a given time.
+ // This is currently necessary for the incremental build to work as expected.
+ // We may want to explicitly manage build priority as opposed to relying on
+ // the reconciliation loop to handle the queuing.
+ for _, b := range builds.Items {
+ if b.Status.Phase == v1.BuildPhasePending || b.Status.Phase == v1.BuildPhaseRunning {
+ // Let's requeue the build in case one is already running
+ return false, nil
+ }
+ }
+
+ return true, nil
+}
+
+func monitorRunningBuild(build *v1.Build) {
+ runningBuilds.Store(types.NamespacedName{Namespace: build.Namespace, Name: build.Name}.String(), true)
+}
+
+func monitorFinishedBuild(build *v1.Build) {
+ runningBuilds.Delete(types.NamespacedName{Namespace: build.Namespace, Name: build.Name}.String())
+}
diff --git a/pkg/controller/build/build_monitor_test.go b/pkg/controller/build/build_monitor_test.go
new file mode 100644
index 000000000..aace7b647
--- /dev/null
+++ b/pkg/controller/build/build_monitor_test.go
@@ -0,0 +1,242 @@
+/*
+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 build
+
+import (
+ "context"
+ "testing"
+ "time"
+
+ v1 "github.com/apache/camel-k/v2/pkg/apis/camel/v1"
+ "github.com/apache/camel-k/v2/pkg/util/test"
+
+ "github.com/stretchr/testify/assert"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+ "k8s.io/apimachinery/pkg/runtime"
+)
+
+func TestMonitorBuilds(t *testing.T) {
+ testcases := []struct {
+ name string
+ running []*v1.Build
+ finished []*v1.Build
+ build *v1.Build
+ allowed bool
+ }{
+ {
+ name: "allowNewBuild",
+ running: []*v1.Build{},
+ finished: []*v1.Build{},
+ build: newBuild("ns", "my-build"),
+ allowed: true,
+ },
+ {
+ name: "allowNewNativeBuild",
+ running: []*v1.Build{},
+ finished: []*v1.Build{},
+ build: newNativeBuild("ns", "my-build"),
+ allowed: true,
+ },
+ {
+ name: "allowNewBuildWhenOthersFinished",
+ running: []*v1.Build{},
+ finished: []*v1.Build{
+ newBuildInPhase("ns", "my-build-x", v1.BuildPhaseSucceeded),
+ newBuildInPhase("ns", "my-build-failed", v1.BuildPhaseFailed),
+ },
+ build: newBuild("ns", "my-build"),
+ allowed: true,
+ },
+ {
+ name: "allowNewNativeBuildWhenOthersFinished",
+ running: []*v1.Build{},
+ finished: []*v1.Build{
+ newNativeBuildInPhase("ns", "my-build-x", v1.BuildPhaseSucceeded),
+ newNativeBuildInPhase("ns", "my-build-failed", v1.BuildPhaseFailed),
+ },
+ build: newNativeBuild("ns", "my-build"),
+ allowed: true,
+ },
+ {
+ name: "limitMaxRunningBuilds",
+ running: []*v1.Build{
+ newBuild("some-ns", "my-build-1"),
+ newBuild("other-ns", "my-build-2"),
+ newBuild("another-ns", "my-build-3"),
+ },
+ finished: []*v1.Build{
+ newBuildInPhase("ns", "my-build-x", v1.BuildPhaseSucceeded),
+ },
+ build: newBuild("ns", "my-build"),
+ allowed: false,
+ },
+ {
+ name: "limitMaxRunningNativeBuilds",
+ running: []*v1.Build{
+ newBuildInPhase("some-ns", "my-build-1", v1.BuildPhaseRunning),
+ newNativeBuildInPhase("other-ns", "my-build-2", v1.BuildPhaseRunning),
+ newNativeBuildInPhase("another-ns", "my-build-3", v1.BuildPhaseRunning),
+ },
+ finished: []*v1.Build{
+ newNativeBuildInPhase("ns", "my-build-x", v1.BuildPhaseSucceeded),
+ },
+ build: newNativeBuildInPhase("ns", "my-build", v1.BuildPhaseInitialization),
+ allowed: false,
+ },
+ {
+ name: "allowParallelBuildsWithDifferentLayout",
+ running: []*v1.Build{
+ newNativeBuildInPhase("ns", "my-build-1", v1.BuildPhaseRunning),
+ },
+ build: newBuild("ns", "my-build"),
+ allowed: true,
+ },
+ {
+ name: "queueBuildsInSameNamespaceWithSameLayout",
+ running: []*v1.Build{
+ newBuild("ns", "my-build-1"),
+ newBuild("other-ns", "my-build-2"),
+ },
+ finished: []*v1.Build{
+ newBuildInPhase("ns", "my-build-x", v1.BuildPhaseSucceeded),
+ },
+ build: newBuild("ns", "my-build"),
+ allowed: false,
+ },
+ {
+ name: "allowBuildsInNewNamespace",
+ running: []*v1.Build{
+ newBuild("some-ns", "my-build-1"),
+ newBuild("other-ns", "my-build-2"),
+ },
+ finished: []*v1.Build{
+ newBuildInPhase("ns", "my-build-x", v1.BuildPhaseSucceeded),
+ },
+ build: newBuild("ns", "my-build"),
+ allowed: true,
+ },
+ }
+
+ for _, tc := range testcases {
+ t.Run(tc.name, func(t *testing.T) {
+ var initObjs []runtime.Object
+ for _, build := range append(tc.running, tc.finished...) {
+ initObjs = append(initObjs, build)
+ }
+
+ c, err := test.NewFakeClient(initObjs...)
+
+ assert.Nil(t, err)
+
+ bm := Monitor{
+ maxRunningBuilds: 3,
+ }
+
+ // reset running builds in memory cache
+ cleanRunningBuildsMonitor()
+ for _, build := range tc.running {
+ monitorRunningBuild(build)
+ }
+
+ allowed, err := bm.canSchedule(context.TODO(), c, tc.build)
+
+ assert.Nil(t, err)
+ assert.Equal(t, tc.allowed, allowed)
+ })
+ }
+}
+
+func TestAllowBuildRequeue(t *testing.T) {
+ c, err := test.NewFakeClient()
+
+ assert.Nil(t, err)
+
+ bm := Monitor{
+ maxRunningBuilds: 3,
+ }
+
+ runningBuild := newBuild("some-ns", "my-build-1")
+ // reset running builds in memory cache
+ cleanRunningBuildsMonitor()
+ monitorRunningBuild(runningBuild)
+ monitorRunningBuild(newBuild("other-ns", "my-build-2"))
+ monitorRunningBuild(newBuild("another-ns", "my-build-3"))
+
+ build := newBuild("ns", "my-build")
+ allowed, err := bm.canSchedule(context.TODO(), c, build)
+
+ assert.Nil(t, err)
+ assert.False(t, allowed)
+
+ monitorFinishedBuild(runningBuild)
+
+ allowed, err = bm.canSchedule(context.TODO(), c, build)
+
+ assert.Nil(t, err)
+ assert.True(t, allowed)
+}
+
+func cleanRunningBuildsMonitor() {
+ runningBuilds.Range(func(key interface{}, v interface{}) bool {
+ runningBuilds.Delete(key)
+ return true
+ })
+}
+
+func newBuild(namespace string, name string) *v1.Build {
+ return newBuildWithLayoutInPhase(namespace, name, v1.IntegrationKitLayoutFastJar, v1.BuildPhasePending)
+}
+
+func newNativeBuild(namespace string, name string) *v1.Build {
+ return newBuildWithLayoutInPhase(namespace, name, v1.IntegrationKitLayoutNative, v1.BuildPhasePending)
+}
+
+func newBuildInPhase(namespace string, name string, phase v1.BuildPhase) *v1.Build {
+ return newBuildWithLayoutInPhase(namespace, name, v1.IntegrationKitLayoutFastJar, phase)
+}
+
+func newNativeBuildInPhase(namespace string, name string, phase v1.BuildPhase) *v1.Build {
+ return newBuildWithLayoutInPhase(namespace, name, v1.IntegrationKitLayoutNative, phase)
+}
+
+func newBuildWithLayoutInPhase(namespace string, name string, layout string, phase v1.BuildPhase) *v1.Build {
+ return &v1.Build{
+ TypeMeta: metav1.TypeMeta{
+ APIVersion: v1.SchemeGroupVersion.String(),
+ Kind: v1.BuildKind,
+ },
+ ObjectMeta: metav1.ObjectMeta{
+ Namespace: namespace,
+ Name: name,
+ Labels: map[string]string{
+ v1.IntegrationKitLayoutLabel: layout,
+ },
+ },
+ Spec: v1.BuildSpec{
+ Strategy: v1.BuildStrategyRoutine,
+ ToolImage: "camel:latest",
+ BuilderPodNamespace: "ns",
+ Tasks: []v1.Task{},
+ Timeout: metav1.Duration{Duration: 5 * time.Minute},
+ MaxRunningBuilds: 3,
+ },
+ Status: v1.BuildStatus{
+ Phase: phase,
+ },
+ }
+}
diff --git a/pkg/controller/build/monitor_pod.go b/pkg/controller/build/monitor_pod.go
index d75353dba..c9b7be754 100644
--- a/pkg/controller/build/monitor_pod.go
+++ b/pkg/controller/build/monitor_pod.go
@@ -99,6 +99,7 @@ func (action *monitorPodAction) Handle(ctx context.Context, build *v1.Build) (*v
// Emulate context cancellation
build.Status.Phase = v1.BuildPhaseInterrupted
build.Status.Error = "Pod deleted"
+ monitorFinishedBuild(build)
return build, nil
}
}
@@ -121,6 +122,12 @@ func (action *monitorPodAction) Handle(ctx context.Context, build *v1.Build) (*v
// Requeue
return nil, err
}
+
+ monitorFinishedBuild(build)
+ } else {
+ // Monitor running state of the build - this may have been done already by the schedule action but the build monitor is idempotent
+ // We do this here to potentially restore the running build state in the monitor in case of an operator restart
+ monitorRunningBuild(build)
}
case corev1.PodSucceeded:
@@ -134,6 +141,8 @@ func (action *monitorPodAction) Handle(ctx context.Context, build *v1.Build) (*v
duration := finishedAt.Sub(build.Status.StartedAt.Time)
build.Status.Duration = duration.String()
+ monitorFinishedBuild(build)
+
buildCreator := kubernetes.GetCamelCreator(build)
// Account for the Build metrics
observeBuildResult(build, build.Status.Phase, buildCreator, duration)
@@ -180,6 +189,8 @@ func (action *monitorPodAction) Handle(ctx context.Context, build *v1.Build) (*v
duration := finishedAt.Sub(build.Status.StartedAt.Time)
build.Status.Duration = duration.String()
+ monitorFinishedBuild(build)
+
buildCreator := kubernetes.GetCamelCreator(build)
// Account for the Build metrics
observeBuildResult(build, build.Status.Phase, buildCreator, duration)
diff --git a/pkg/controller/build/monitor_routine.go b/pkg/controller/build/monitor_routine.go
index 0a08919fc..6aa9ed696 100644
--- a/pkg/controller/build/monitor_routine.go
+++ b/pkg/controller/build/monitor_routine.go
@@ -67,6 +67,7 @@ func (action *monitorRoutineAction) Handle(ctx context.Context, build *v1.Build)
routines.Delete(build.Name)
build.Status.Phase = v1.BuildPhaseFailed
build.Status.Error = "Build routine exists"
+ monitorFinishedBuild(build)
return build, nil
}
status := v1.BuildStatus{Phase: v1.BuildPhaseRunning}
@@ -84,10 +85,14 @@ func (action *monitorRoutineAction) Handle(ctx context.Context, build *v1.Build)
// stops abruptly and restarts or the build status update fails.
build.Status.Phase = v1.BuildPhaseFailed
build.Status.Error = "Build routine not running"
+ monitorFinishedBuild(build)
return build, nil
}
}
+ // Monitor running state of the build - this may have been done already by the schedule action but the monitor action is idempotent
+ // We do this here to recover the running build state in the monitor in case of an operator restart
+ monitorRunningBuild(build)
return nil, nil
}
@@ -173,6 +178,8 @@ tasks:
duration := metav1.Now().Sub(build.Status.StartedAt.Time)
status.Duration = duration.String()
+ monitorFinishedBuild(build)
+
buildCreator := kubernetes.GetCamelCreator(build)
// Account for the Build metrics
observeBuildResult(build, status.Phase, buildCreator, duration)
diff --git a/pkg/controller/build/schedule.go b/pkg/controller/build/schedule.go
index 746db319d..5b67b86a9 100644
--- a/pkg/controller/build/schedule.go
+++ b/pkg/controller/build/schedule.go
@@ -22,9 +22,6 @@ import (
"sync"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
- "k8s.io/apimachinery/pkg/labels"
- "k8s.io/apimachinery/pkg/selection"
-
ctrl "sigs.k8s.io/controller-runtime/pkg/client"
v1 "github.com/apache/camel-k/v2/pkg/apis/camel/v1"
@@ -32,16 +29,18 @@ import (
"github.com/apache/camel-k/v2/pkg/util/kubernetes"
)
-func newScheduleAction(reader ctrl.Reader) Action {
+func newScheduleAction(reader ctrl.Reader, buildMonitor Monitor) Action {
return &scheduleAction{
- reader: reader,
+ reader: reader,
+ buildMonitor: buildMonitor,
}
}
type scheduleAction struct {
baseAction
- lock sync.Mutex
- reader ctrl.Reader
+ lock sync.Mutex
+ reader ctrl.Reader
+ buildMonitor Monitor
}
// Name returns a common name of the action.
@@ -60,42 +59,11 @@ func (action *scheduleAction) Handle(ctx context.Context, build *v1.Build) (*v1.
action.lock.Lock()
defer action.lock.Unlock()
- layout := build.Labels[v1.IntegrationKitLayoutLabel]
-
- // Native builds can be run in parallel, as incremental images is not applicable.
- if layout == v1.IntegrationKitLayoutNative {
- // Reset the Build status, and transition it to pending phase.
- // This must be done in the critical section, rather than delegated to the controller.
- return nil, action.toPendingPhase(ctx, build)
- }
-
- // We assume incremental images is only applicable across images whose layout is identical
- withCompatibleLayout, err := labels.NewRequirement(v1.IntegrationKitLayoutLabel, selection.Equals, []string{layout})
- if err != nil {
+ if allowed, err := action.buildMonitor.canSchedule(ctx, action.reader, build); err != nil {
return nil, err
- }
-
- builds := &v1.BuildList{}
- // We use the non-caching client as informers cache is not invalidated nor updated
- // atomically by write operations
- err = action.reader.List(ctx, builds,
- ctrl.InNamespace(build.Namespace),
- ctrl.MatchingLabelsSelector{
- Selector: labels.NewSelector().Add(*withCompatibleLayout),
- })
- if err != nil {
- return nil, err
- }
-
- // Emulate a serialized working queue to only allow one build to run at a given time.
- // This is currently necessary for the incremental build to work as expected.
- // We may want to explicitly manage build priority as opposed to relying on
- // the reconciliation loop to handle the queuing.
- for _, b := range builds.Items {
- if b.Status.Phase == v1.BuildPhasePending || b.Status.Phase == v1.BuildPhaseRunning {
- // Let's requeue the build in case one is already running
- return nil, nil
- }
+ } else if !allowed {
+ // Build not allowed at this state (probably max running builds limit exceeded) - let's requeue the build
+ return nil, nil
}
// Reset the Build status, and transition it to pending phase.
@@ -117,6 +85,8 @@ func (action *scheduleAction) toPendingPhase(ctx context.Context, build *v1.Buil
return err
}
+ monitorRunningBuild(build)
+
buildCreator := kubernetes.GetCamelCreator(build)
// Report the duration the Build has been waiting in the build queue
observeBuildQueueDuration(build, buildCreator)
diff --git a/pkg/controller/integrationkit/build.go b/pkg/controller/integrationkit/build.go
index c98007dfb..c33deb920 100644
--- a/pkg/controller/integrationkit/build.go
+++ b/pkg/controller/integrationkit/build.go
@@ -146,6 +146,7 @@ func (action *buildAction) handleBuildSubmitted(ctx context.Context, kit *v1.Int
BuilderPodNamespace: builderPodNamespace,
Tasks: env.BuildTasks,
Timeout: timeout,
+ MaxRunningBuilds: env.Platform.Status.Build.MaxRunningBuilds,
},
}
diff --git a/pkg/platform/defaults.go b/pkg/platform/defaults.go
index fa24f4fa9..4f7da9f57 100644
--- a/pkg/platform/defaults.go
+++ b/pkg/platform/defaults.go
@@ -246,6 +246,14 @@ func setPlatformDefaults(p *v1.IntegrationPlatform, verbose bool) error {
}
}
+ if p.Status.Build.MaxRunningBuilds <= 0 {
+ if p.Status.Build.BuildStrategy == v1.BuildStrategyRoutine {
+ p.Status.Build.MaxRunningBuilds = 3
+ } else if p.Status.Build.BuildStrategy == v1.BuildStrategyPod {
+ p.Status.Build.MaxRunningBuilds = 10
+ }
+ }
+
_, cacheEnabled := p.Status.Build.PublishStrategyOptions[builder.KanikoBuildCacheEnabled]
if p.Status.Build.PublishStrategy == v1.IntegrationPlatformBuildPublishStrategyKaniko && !cacheEnabled {
// Default to disabling Kaniko cache warmer
diff --git a/pkg/resources/resources.go b/pkg/resources/resources.go
index b6b803d0e..3149ebade 100644
--- a/pkg/resources/resources.go
+++ b/pkg/resources/resources.go
@@ -117,9 +117,9 @@ var assets = func() http.FileSystem {
"/crd/bases/camel.apache.org_builds.yaml": &vfsgen۰CompressedFileInfo{
name: "camel.apache.org_builds.yaml",
modTime: time.Time{},
- uncompressedSize: 39558,
+ uncompressedSize: 39777,
- compressedContent: []byte("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\xff\xec\x3d\x6b\x73\x23\x39\x6e\xdf\xfd\x2b\x50\xeb\x0f\xe3\xa9\xb2\xa4\xdd\xbd\x47\x36\x4e\xa5\x52\x3a\xcf\xee\x9e\x33\x0f\x4f\x46\xde\xb9\xbb\x6f\xa6\xba\x21\x89\xa7\x6e\xb2\x43\xb2\xad\xd1\xa5\xf2\xdf\x53\x04\xc9\x56\x4b\xea\x07\xdb\x8f\x9d\xcb\x9d\xf8\x65\xc6\x2d\x3e\x00\x10\x04\x40\x10\x04\xcf\x61\xf4\x7c\xe5\xec\x1c\xde\xf1\x04\x85\xc6\x14\x8c\x04\xb3\x42\x98\x16\x2c\x59\x21\xcc\xe4\xc2\x6c\x98\x42\xf8\x49\x96\x22\x [...]
+ compressedContent: []byte("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\xff\xec\x3d\x6b\x73\x23\xb9\x71\xdf\xf5\x2b\xba\x4e\x1f\x56\x5b\x25\x92\xf7\xb0\x9d\x8b\x52\xa9\x14\xad\xbd\x3b\x2b\xfb\x90\xb2\xd4\xad\xed\x6f\x02\x67\x9a\x24\xcc\x19\x60\x02\x60\x24\xd1\xa9\xfc\xf7\x14\x1a\xc0\x70\x48\xce\x03\xa3\xc7\x9d\xe3\x23\xbe\xec\x6a\x08\x34\xba\x81\x46\xbf\xd0\x00\x4e\x61\xf4\x72\xe5\xe4\x14\x3e\xf0\x04\x85\xc6\x14\x8c\x04\xb3\x42\x98\x16\x2c\x59\x21\xcc\xe4\xc2\x3c\x30\x85\xf0\xa3\x2c\x45\xca\x [...]
},
"/crd/bases/camel.apache.org_camelcatalogs.yaml": &vfsgen۰CompressedFileInfo{
name: "camel.apache.org_camelcatalogs.yaml",
@@ -138,9 +138,9 @@ var assets = func() http.FileSystem {
"/crd/bases/camel.apache.org_integrationplatforms.yaml": &vfsgen۰CompressedFileInfo{
name: "camel.apache.org_integrationplatforms.yaml",
modTime: time.Time{},
- uncompressedSize: 172313,
+ uncompressedSize: 172791,
- compressedContent: []byte("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\xff\xec\xbd\x7b\x73\xe3\x36\xb2\x28\xfe\xff\x7c\x0a\x94\x53\xa7\xc6\x33\x65\xc9\x33\xbb\x27\xbb\x39\x3e\x95\xba\xd7\xf1\x4c\x12\x67\x1e\xf6\xb1\x3d\xb3\xbb\x95\xa4\x22\x88\x6c\x49\x88\x41\x80\x07\x00\x65\x2b\xbf\xfd\x7d\xf7\x5b\x68\x00\x24\x25\x91\x20\x25\x79\x1e\x49\xc4\x54\xed\x8e\x6d\xa2\xd9\x00\x1a\xfd\x42\x3f\xbe\x20\x83\x87\x7b\x1e\x7d\x41\x5e\xb3\x04\x84\x86\x94\x18\x49\xcc\x0c\xc8\x69\x4e\x93\x19\x90\x6b\x39\x31\x [...]
+ compressedContent: []byte("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\xff\xec\xbd\x7b\x73\xe3\x36\xb2\x28\xfe\xff\x7c\x0a\x94\x53\xa7\xc6\x33\x65\xc9\x33\xbb\x27\xbb\x39\x3e\x95\xba\xd7\xf1\x4c\x12\x67\x1e\xf6\xb1\x3d\xb3\xbb\x95\xa4\x22\x88\x6c\x49\x88\x41\x80\x07\x00\x65\x2b\xbf\xfd\x7d\xf7\x5b\x68\x00\x24\x25\x91\x20\x25\x79\x1e\x49\xc4\x54\xed\x8e\x6d\xa2\xd9\x00\x1a\xfd\x42\x3f\xbe\x20\x83\x87\x7b\x1e\x7d\x41\x5e\xb3\x04\x84\x86\x94\x18\x49\xcc\x0c\xc8\x69\x4e\x93\x19\x90\x6b\x39\x31\x [...]
},
"/crd/bases/camel.apache.org_integrations.yaml": &vfsgen۰CompressedFileInfo{
name: "camel.apache.org_integrations.yaml",