You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by pc...@apache.org on 2023/05/08 12:39:51 UTC
[camel-k] 01/07: fix: Limit parallel builds on operator
This is an automated email from the ASF dual-hosted git repository.
pcongiusti pushed a commit to branch release-1.12.x
in repository https://gitbox.apache.org/repos/asf/camel-k.git
commit cfc0ff8f4a6f7750db66e5b754e79dd77b1cad0f
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 | 15 +-
.../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 | 240 +++++++++++++++++++++
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 | 7 +-
pkg/platform/defaults.go | 9 +
pkg/resources/resources.go | 8 +-
20 files changed, 505 insertions(+), 54 deletions(-)
diff --git a/config/crd/bases/camel.apache.org_builds.yaml b/config/crd/bases/camel.apache.org_builds.yaml
index 4b530895f..64c799c2a 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
strategy:
description: The strategy that should be used to perform the Build.
enum:
diff --git a/config/crd/bases/camel.apache.org_integrationplatforms.yaml b/config/crd/bases/camel.apache.org_integrationplatforms.yaml
index f338314b1..412f81c39 100644
--- a/config/crd/bases/camel.apache.org_integrationplatforms.yaml
+++ b/config/crd/bases/camel.apache.org_integrationplatforms.yaml
@@ -238,6 +238,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
@@ -1780,6 +1785,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 ebe490c31..13f9f859c 100644
--- a/docs/modules/ROOT/partials/apis/camel-k-crds.adoc
+++ b/docs/modules/ROOT/partials/apis/camel-k-crds.adoc
@@ -439,6 +439,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
+
|===
@@ -1770,6 +1777,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 4b530895f..64c799c2a 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
strategy:
description: The strategy that should be used to perform the Build.
enum:
diff --git a/helm/camel-k/crds/crd-integration-platform.yaml b/helm/camel-k/crds/crd-integration-platform.yaml
index f338314b1..412f81c39 100644
--- a/helm/camel-k/crds/crd-integration-platform.yaml
+++ b/helm/camel-k/crds/crd-integration-platform.yaml
@@ -238,6 +238,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
@@ -1780,6 +1785,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 de1101864..7bc250df2 100644
--- a/pkg/apis/camel/v1/build_types.go
+++ b/pkg/apis/camel/v1/build_types.go
@@ -37,6 +37,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 0c378f9c7..dea7656ee 100644
--- a/pkg/apis/camel/v1/integrationplatform_types.go
+++ b/pkg/apis/camel/v1/integrationplatform_types.go
@@ -126,6 +126,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 0ab858425..4f0739239 100644
--- a/pkg/client/camel/applyconfiguration/camel/v1/buildspec.go
+++ b/pkg/client/camel/applyconfiguration/camel/v1/buildspec.go
@@ -27,9 +27,10 @@ import (
// BuildSpecApplyConfiguration represents an declarative configuration of the BuildSpec type for use
// with apply.
type BuildSpecApplyConfiguration struct {
- Tasks []TaskApplyConfiguration `json:"tasks,omitempty"`
- Strategy *camelv1.BuildStrategy `json:"strategy,omitempty"`
- Timeout *metav1.Duration `json:"timeout,omitempty"`
+ Tasks []TaskApplyConfiguration `json:"tasks,omitempty"`
+ Strategy *camelv1.BuildStrategy `json:"strategy,omitempty"`
+ Timeout *metav1.Duration `json:"timeout,omitempty"`
+ MaxRunningBuilds *int32 `json:"maxRunningBuilds,omitempty"`
}
// BuildSpecApplyConfiguration constructs an declarative configuration of the BuildSpec type for use with
@@ -66,3 +67,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 554abda90..0e81c9bc8 100644
--- a/pkg/client/camel/applyconfiguration/camel/v1/integrationplatformbuildspec.go
+++ b/pkg/client/camel/applyconfiguration/camel/v1/integrationplatformbuildspec.go
@@ -36,6 +36,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
@@ -121,3 +122,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 13ec31924..262d53b0e 100644
--- a/pkg/cmd/install.go
+++ b/pkg/cmd/install.go
@@ -190,6 +190,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"`
@@ -530,6 +531,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 3bfab58ef..91a8174ec 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..c7d8443d5
--- /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/pkg/apis/camel/v1"
+ "github.com/apache/camel-k/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..cadb37330
--- /dev/null
+++ b/pkg/controller/build/build_monitor_test.go
@@ -0,0 +1,240 @@
+/*
+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/pkg/apis/camel/v1"
+ "github.com/apache/camel-k/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,
+ 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 baaa5800c..0cd4bb859 100644
--- a/pkg/controller/build/monitor_pod.go
+++ b/pkg/controller/build/monitor_pod.go
@@ -87,6 +87,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
}
}
@@ -109,6 +110,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:
@@ -122,6 +129,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)
@@ -168,6 +177,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 2820c8a70..9373e0fa0 100644
--- a/pkg/controller/build/monitor_routine.go
+++ b/pkg/controller/build/monitor_routine.go
@@ -68,6 +68,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}
@@ -85,10 +86,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
}
@@ -174,6 +179,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 12e879d61..0bedfb0ed 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/pkg/apis/camel/v1"
@@ -32,16 +29,18 @@ import (
"github.com/apache/camel-k/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 9ea226fac..8f5524779 100644
--- a/pkg/controller/integrationkit/build.go
+++ b/pkg/controller/integrationkit/build.go
@@ -115,9 +115,10 @@ func (action *buildAction) handleBuildSubmitted(ctx context.Context, kit *v1.Int
Annotations: annotations,
},
Spec: v1.BuildSpec{
- Strategy: env.Platform.Status.Build.BuildStrategy,
- Tasks: env.BuildTasks,
- Timeout: timeout,
+ Strategy: env.Platform.Status.Build.BuildStrategy,
+ Tasks: env.BuildTasks,
+ Timeout: timeout,
+ MaxRunningBuilds: env.Platform.Status.Build.MaxRunningBuilds,
},
}
diff --git a/pkg/platform/defaults.go b/pkg/platform/defaults.go
index f1168c623..823f2bd7d 100644
--- a/pkg/platform/defaults.go
+++ b/pkg/platform/defaults.go
@@ -233,6 +233,15 @@ func setPlatformDefaults(p *v1.IntegrationPlatform, verbose bool) error {
Duration: 5 * time.Minute,
}
}
+
+ 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 23b682afd..3eb87bee9 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: 39199,
+ uncompressedSize: 39418,
- compressedContent: []byte("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\xff\xec\x3d\x6b\x73\x23\x29\x92\xdf\xfd\x2b\x32\xc6\x1f\xda\x1d\x61\x49\x33\xb3\x8f\x9b\xf3\xc5\xc5\x85\xd6\x3d\x33\xeb\xeb\x87\xfb\x5a\x9e\xde\xdd\x6f\x46\x55\x29\x89\x55\x15\xd4\x01\x65\xb5\xf6\xe2\xfe\xfb\x05\x09\x94\x4a\x52\x3d\x28\x3f\xa6\xf7\x76\xc5\x97\x6e\x97\x20\x49\xc8\x24\x5f\x24\x70\x0e\xa3\xe7\x2b\x67\xe7\xf0\x8e\x27\x28\x34\xa6\x60\x24\x98\x15\xc2\xb4\x60\xc9\x0a\x61\x26\x17\x66\xc3\x14\xc2\x4f\xb2\x14\x29\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\xd0\x66\xa9\x5b\xdb\xdf\x04\xce\x34\x49\x98\x33\xc0\x04\xc0\x88\x4b\xa7\xf2\xdf\x53\x68\x00\xc3\x21\x39\x0f\x8c\x1e\x77\x8e\x8f\xf8\xb2\x2b\x0e\xd0\x68\xa0\x1b\xfd\x42\x03\x38\x87\xd1\xf3\x95\xb3\x73\x78\xc7\x13\x14\x1a\x53\x30\x12\xcc\x0a\x61\x5a\xb0\x64\x85\x30\x93\x0b\xb3\x61\x0a\xe1\x47\x59\x8a\x94\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: 171501,
+ uncompressedSize: 171979,
- compressedContent: []byte("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\xff\xec\xbd\x7d\x73\xdb\x36\xb6\x30\xfe\x7f\x3e\x05\xc6\x9d\x3b\x71\x32\x7a\x49\x76\x6f\x77\x7b\x7d\xa7\xf3\x3c\xae\xed\xb6\x6e\xe2\xd8\xd7\x76\xb2\xbb\xd3\x76\x2a\x88\x3c\x92\x50\x93\x00\x2f\x00\xca\x56\x7f\xfb\xfb\xee\xcf\xe0\x00\x20\x29\x89\x04\x29\xc9\x71\xd2\x56\xec\xcc\x6e\x6c\x13\x87\x07\xc0\xc1\x79\xc3\x79\xf9\x82\xf4\x1f\xef\x79\xf6\x05\x79\xcb\x22\xe0\x0a\x62\xa2\x05\xd1\x33\x20\xc7\x19\x8d\x66\x40\x6e\xc4\x44\x [...]
+ compressedContent: []byte("\x1f\x8b\x08\x00\x00\x00\x00\x00\x00\xff\xec\xbd\x7d\x73\xdb\x36\xb6\x30\xfe\x7f\x3e\x05\xc6\x9d\x3b\x71\x32\x7a\x49\x76\x6f\x77\x7b\x7d\xa7\xf3\x3c\xae\xed\xb6\x6e\xe2\xd8\xd7\x76\xb2\xbb\xd3\x76\x2a\x88\x3c\x92\x50\x93\x00\x2f\x00\xca\x56\x7f\xfb\xfb\xee\xcf\xe0\x00\x20\x29\x89\x04\x29\xc9\x71\xd2\x56\xec\xcc\x6e\x6c\x13\x87\x07\xc0\xc1\x79\xc3\x79\xf9\x82\xf4\x1f\xef\x79\xf6\x05\x79\xcb\x22\xe0\x0a\x62\xa2\x05\xd1\x33\x20\xc7\x19\x8d\x66\x40\x6e\xc4\x44\x [...]
},
"/crd/bases/camel.apache.org_integrations.yaml": &vfsgen۰CompressedFileInfo{
name: "camel.apache.org_integrations.yaml",