This is an automated email from the ASF dual-hosted git repository.
squakez pushed a commit to branch v3
in repository https://gitbox.apache.org/repos/asf/camel-k.git
The following commit(s) were added to refs/heads/v3 by this push:
new 48e92fde2 feat(ctrl): simplify monitoring
48e92fde2 is described below
commit 48e92fde27a5fdc635ff47b352825b1640fc1d36
Author: Pasquale Congiusti <[email protected]>
AuthorDate: Fri Oct 23 08:47:08 2026 +0200
feat(ctrl): simplify monitoring
Closes #6850
---
e2e/common/misc/integration_fail_test.go | 12 -
e2e/common/misc/scale_integration_test.go | 12 +-
pkg/controller/integration/health.go | 105 --------
pkg/controller/integration/health_test.go | 101 --------
pkg/controller/integration/monitor.go | 285 +--------------------
pkg/controller/integration/monitor_cronjob.go | 114 ++-------
pkg/controller/integration/monitor_cronjob_test.go | 147 +++++++++++
pkg/controller/integration/monitor_deployment.go | 78 ++----
.../integration/monitor_deployment_test.go | 133 ++++++++++
pkg/controller/integration/monitor_knative.go | 45 +---
pkg/controller/integration/monitor_knative_test.go | 245 +++++-------------
pkg/util/kubernetes/conditions.go | 24 --
12 files changed, 419 insertions(+), 882 deletions(-)
diff --git a/e2e/common/misc/integration_fail_test.go
b/e2e/common/misc/integration_fail_test.go
index d16d7e49f..340b6e86a 100644
--- a/e2e/common/misc/integration_fail_test.go
+++ b/e2e/common/misc/integration_fail_test.go
@@ -28,7 +28,6 @@ import (
"time"
. "github.com/onsi/gomega"
- "github.com/onsi/gomega/gstruct"
corev1 "k8s.io/api/core/v1"
. "github.com/apache/camel-k/v2/e2e/support"
@@ -42,20 +41,9 @@ func TestBadRouteIntegration(t *testing.T) {
name := RandomizedSuffixName("bad-route")
g.Expect(KamelRun(t, ctx, ns, "files/BadRoute.java",
"--name", name, "-t", "health.enabled=false").Execute()).To(Succeed())
g.Eventually(IntegrationPodPhase(t, ctx, ns, name),
TestTimeoutLong).Should(Equal(corev1.PodRunning))
- g.Eventually(IntegrationPhase(t, ctx, ns, name),
TestTimeoutShort).Should(Equal(v1.IntegrationPhaseError))
g.Eventually(IntegrationConditionStatus(t, ctx, ns,
name, v1.IntegrationConditionReady), TestTimeoutShort).
Should(Equal(corev1.ConditionFalse))
- // Make sure the Integration can be scaled
- g.Expect(ScaleIntegration(t, ctx, ns, name,
2)).To(Succeed())
- // Check the scale cascades into the Deployment scale
- g.Eventually(IntegrationPods(t, ctx, ns, name),
TestTimeoutShort).Should(HaveLen(2))
- // Check it also cascades into the Integration scale
subresource Status field
- g.Eventually(IntegrationStatusReplicas(t, ctx, ns,
name), TestTimeoutShort).
- Should(gstruct.PointTo(BeNumerically("==", 2)))
- // Check the Integration stays in error phase
- g.Eventually(IntegrationPhase(t, ctx, ns, name),
TestTimeoutShort).Should(Equal(v1.IntegrationPhaseError))
-
// Kit valid
kitName := IntegrationKitName(t, ctx, ns, name)()
integrationKitNamespace := IntegrationKitNamespace(t,
ctx, ns, name)()
diff --git a/e2e/common/misc/scale_integration_test.go
b/e2e/common/misc/scale_integration_test.go
index 861ff09e7..ae259e3ba 100644
--- a/e2e/common/misc/scale_integration_test.go
+++ b/e2e/common/misc/scale_integration_test.go
@@ -41,18 +41,18 @@ func TestIntegrationScale(t *testing.T) {
WithNewTestNamespace(t, func(ctx context.Context, g *WithT, ns string) {
name := RandomizedSuffixName("java")
g.Expect(KamelRun(t, ctx, ns, "files/Java.java", "--name",
name).Execute()).To(Succeed())
- g.Eventually(IntegrationPodPhase(t, ctx, ns, name),
TestTimeoutLong).Should(Equal(corev1.PodRunning))
- g.Eventually(IntegrationConditionStatus(t, ctx, ns, name,
v1.IntegrationConditionReady),
TestTimeoutShort).Should(Equal(corev1.ConditionTrue))
- g.Eventually(IntegrationLogs(t, ctx, ns, name),
TestTimeoutShort).Should(ContainSubstring("Magicstring!"))
+ g.Eventually(IntegrationConditionStatus(t, ctx, ns, name,
v1.IntegrationConditionReady),
TestTimeoutMedium).Should(Equal(corev1.ConditionTrue))
+ g.Eventually(IntegrationPodPhase(t, ctx, ns,
name)).Should(Equal(corev1.PodRunning))
+ g.Eventually(IntegrationLogs(t, ctx, ns,
name)).Should(ContainSubstring("Magicstring!"))
t.Run("Update integration scale spec", func(t *testing.T) {
g.Expect(ScaleIntegration(t, ctx, ns, name,
3)).To(Succeed())
// Check the readiness condition becomes falsy
- g.Eventually(IntegrationConditionStatus(t, ctx, ns,
name, v1.IntegrationConditionReady),
TestTimeoutShort).Should(Equal(corev1.ConditionFalse))
+ g.Eventually(IntegrationConditionStatus(t, ctx, ns,
name, v1.IntegrationConditionReady)).Should(Equal(corev1.ConditionFalse))
// Check the scale cascades into the Deployment scale
- g.Eventually(IntegrationPods(t, ctx, ns, name),
TestTimeoutShort).Should(HaveLen(3))
+ g.Eventually(IntegrationPods(t, ctx, ns,
name)).Should(HaveLen(3))
// Check it also cascades into the Integration scale
subresource Status field
- g.Eventually(IntegrationStatusReplicas(t, ctx, ns,
name), TestTimeoutShort).
+ g.Eventually(IntegrationStatusReplicas(t, ctx, ns,
name)).
Should(gstruct.PointTo(BeNumerically("==", 3)))
// Finally check the readiness condition becomes truthy
back
g.Eventually(IntegrationConditionStatus(t, ctx, ns,
name, v1.IntegrationConditionReady),
TestTimeoutMedium).Should(Equal(corev1.ConditionTrue))
diff --git a/pkg/controller/integration/health.go
b/pkg/controller/integration/health.go
deleted file mode 100644
index e1a1490ae..000000000
--- a/pkg/controller/integration/health.go
+++ /dev/null
@@ -1,105 +0,0 @@
-/*
-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 integration
-
-import (
- "context"
- "encoding/json"
- "fmt"
- "strconv"
- "strings"
- "time"
-
- v1 "github.com/apache/camel-k/v2/pkg/apis/camel/v1"
-
- corev1 "k8s.io/api/core/v1"
- "k8s.io/apimachinery/pkg/util/intstr"
- "k8s.io/client-go/kubernetes"
-)
-
-const (
- // HealthCheckErrorMessage key used for propagating error details from
Camel health to MicroProfile Health
- // (See CAMEL-17138).
- HealthCheckErrorMessage = "error.message"
-)
-
-func NewHealthCheck(body []byte) (*v1.HealthCheck, error) {
- health := v1.HealthCheck{}
- if err := json.Unmarshal(body, &health); err != nil {
- return nil, err
- }
-
- return &health, nil
-}
-
-func proxyGetHTTPProbe(ctx context.Context, c kubernetes.Interface, p
*corev1.Probe, pod *corev1.Pod, container *corev1.Container) ([]byte, error) {
- if p.HTTPGet == nil {
- return nil, fmt.Errorf("missing probe handler for %s/%s",
pod.Namespace, pod.Name)
- }
-
- // We have to extract the port number from the HTTP probe in case it's
a named port,
- // as Pod proxying via the API Server does not work with named ports.
- port, err := extractPortNumber(p.HTTPGet.Port, container)
- if err != nil {
- return nil, err
- }
-
- probeCtx, cancel := context.WithTimeout(ctx,
time.Duration(p.TimeoutSeconds)*time.Second)
- defer cancel()
-
- params := make(map[string]string)
-
- return c.CoreV1().
- Pods(pod.Namespace).
- ProxyGet(strings.ToLower(string(p.HTTPGet.Scheme)), pod.Name,
strconv.Itoa(port), p.HTTPGet.Path, params).
- DoRaw(probeCtx)
-}
-
-func extractPortNumber(port intstr.IntOrString, container *corev1.Container)
(int, error) {
- number := -1
- var err error
- switch port.Type {
- case intstr.Int:
- number = port.IntValue()
- case intstr.String:
- if number, err = findPortByName(container, port.StrVal); err !=
nil {
- // Last ditch effort - maybe it was an int stored as
string?
- if number, err = strconv.Atoi(port.StrVal); err != nil {
- return number, err
- }
- }
- default:
- return number, fmt.Errorf("intOrString had no kind: %+v", port)
- }
- if number > 0 && number < 65536 {
- return number, nil
- }
-
- return number, fmt.Errorf("invalid port number: %v", number)
-}
-
-// findPortByName is a helper function to look up a port in a container by
name.
-func findPortByName(container *corev1.Container, portName string) (int, error)
{
- for _, port := range container.Ports {
- if port.Name == portName {
- return int(port.ContainerPort), nil
- }
- }
-
- return 0, fmt.Errorf("port %s not found", portName)
-}
diff --git a/pkg/controller/integration/health_test.go
b/pkg/controller/integration/health_test.go
deleted file mode 100644
index 946ae8869..000000000
--- a/pkg/controller/integration/health_test.go
+++ /dev/null
@@ -1,101 +0,0 @@
-/*
-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 integration
-
-import (
- "encoding/json"
- "reflect"
- "testing"
-
- camelv1 "github.com/apache/camel-k/v2/pkg/apis/camel/v1"
-
- "github.com/stretchr/testify/assert"
- "github.com/stretchr/testify/require"
-)
-
-func TestNewHealthCheck(t *testing.T) {
- body := []byte(`
- {
- "status": "DOWN",
- "checks": [
- {
- "name": "camel-routes",
- "status": "DOWN",
- "data": {
- "route.id": "route1",
- "route.context.name": "camel-1",
- "route.status": "Stopped"
- }
- },
- {
- "name": "context",
- "status": "UP",
- "data": {
- "context.name": "camel-1",
- "context.version": "3.16.0",
- "context.status": "Started"
- }
- },
- {
- "name": "camel-consumers",
- "status": "DOWN",
- "data": {
- "route.id": "route1",
- "route.context.name": "camel-1",
- "route.status": "Stopped"
- }
- }
- ]
- }
- `)
- health, err := NewHealthCheck(body)
- require.NoError(t, err)
- assert.Equal(t, camelv1.HealthCheckStatusDown, health.Status)
- assert.Len(t, health.Checks, 3)
- assert.Equal(t, "camel-routes", health.Checks[0].Name)
- assert.Equal(t, camelv1.HealthCheckStatusDown, health.Checks[0].Status)
- assert.True(t, reflect.DeepEqual(raw2map(t, health.Checks[0].Data),
map[string]interface{}{
- "route.id": "route1",
- "route.context.name": "camel-1",
- "route.status": "Stopped",
- }))
- assert.Equal(t, "context", health.Checks[1].Name)
- assert.Equal(t, camelv1.HealthCheckStatusUp, health.Checks[1].Status)
- assert.True(t, reflect.DeepEqual(raw2map(t, health.Checks[1].Data),
map[string]interface{}{
- "context.name": "camel-1",
- "context.version": "3.16.0",
- "context.status": "Started",
- }))
- assert.Equal(t, "camel-consumers", health.Checks[2].Name)
- assert.Equal(t, camelv1.HealthCheckStatusDown, health.Checks[2].Status)
- assert.True(t, reflect.DeepEqual(raw2map(t, health.Checks[2].Data),
map[string]interface{}{
- "route.id": "route1",
- "route.context.name": "camel-1",
- "route.status": "Stopped",
- }))
-}
-
-func raw2map(t *testing.T, in camelv1.RawMessage) map[string]interface{} {
- t.Helper()
- answer := make(map[string]interface{})
-
- err := json.Unmarshal(in, &answer)
- require.NoError(t, err)
-
- return answer
-}
diff --git a/pkg/controller/integration/monitor.go
b/pkg/controller/integration/monitor.go
index 8fb4197fc..024f8e342 100644
--- a/pkg/controller/integration/monitor.go
+++ b/pkg/controller/integration/monitor.go
@@ -19,7 +19,6 @@ package integration
import (
"context"
- "errors"
"fmt"
"reflect"
"strconv"
@@ -27,7 +26,6 @@ import (
appsv1 "k8s.io/api/apps/v1"
batchv1 "k8s.io/api/batch/v1"
corev1 "k8s.io/api/core/v1"
- k8serrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/selection"
"k8s.io/utils/ptr"
@@ -145,31 +143,6 @@ func (action *monitorAction) monitorPods(ctx
context.Context, environment *trait
if err != nil {
return nil, err
}
-
- // In order to simplify the monitoring and have a minor resource
requirement, we will watch only those Pods
- // which are labeled with `camel.apache.org/integration`. This is a
design choice that requires the user to
- // voluntarily add a label to their Pods (via template, possibly) in
order to monitor the non managed Camel applications.
-
- if !controller.hasTemplateIntegrationLabel() {
- // This is happening when the Deployment, CronJob, etc resources
- // miss the Integration label, required to identify sibling
Pods.
- integration.Status.SetConditions(
- v1.IntegrationCondition{
- Type: v1.IntegrationConditionReady,
- Status: corev1.ConditionFalse,
- Reason:
v1.IntegrationConditionMonitoringPodsAvailableReason,
- Message: fmt.Sprintf(
- "Could not find
`camel.apache.org/integration: %s` label in the %s template. "+
- "Make sure to include this
label in the template for Pod monitoring purposes.",
- integration.GetName(),
- controller.getControllerName(),
- ),
- },
- )
-
- return integration, nil
- }
-
// Enforce the scale sub-resource label selector.
// It is used by the HPA that queries the scale sub-resource endpoint,
// to list the pods owned by the integration.
@@ -210,11 +183,8 @@ func (action *monitorAction) monitorPods(ctx
context.Context, environment *trait
if integration.Status.Phase == v1.IntegrationPhaseDeploying {
integration.Status.Phase = v1.IntegrationPhaseRunning
}
- if err = action.updateIntegrationPhaseAndReadyCondition(
- ctx, controller, environment, integration, pendingPods.Items,
runningPods.Items,
- ); err != nil {
- return nil, err
- }
+
+ controller.setReadyCondition(integration)
return integration, nil
}
@@ -322,10 +292,8 @@ func getIntegrationSecretAndConfigmapResourceVersions(ctx
context.Context, clien
}
type controller interface {
- checkReadyCondition(ctx context.Context) (bool, error)
- updateReadyCondition(readyPods int32) bool
- hasTemplateIntegrationLabel() bool
- getControllerName() string
+ // in charge to set the readiness condition, specific to each
controller.
+ setReadyCondition(integration *v1.Integration)
}
func (action *monitorAction) newController(env *trait.Environment, integration
*v1.Integration) (controller, error) {
@@ -339,8 +307,7 @@ func (action *monitorAction) newController(env
*trait.Environment, integration *
return nil, fmt.Errorf("type assertion failed, not a
Deployment: %v", obj)
}
controller = &deploymentController{
- obj: deploy,
- integration: integration,
+ obj: deploy,
}
case
integration.IsConditionTrue(v1.IntegrationConditionKnativeServiceAvailable):
obj = getUpdatedController(env, &servingv1.Service{})
@@ -349,8 +316,7 @@ func (action *monitorAction) newController(env
*trait.Environment, integration *
return nil, fmt.Errorf("type assertion failed, not a
Knative Service: %v", obj)
}
controller = &knativeServiceController{
- obj: svc,
- integration: integration,
+ obj: svc,
}
case
integration.IsConditionTrue(v1.IntegrationConditionCronJobAvailable):
obj = getUpdatedController(env, &batchv1.CronJob{})
@@ -359,9 +325,7 @@ func (action *monitorAction) newController(env
*trait.Environment, integration *
return nil, fmt.Errorf("type assertion failed, not a
CronJob: %v", obj)
}
controller = &cronJobController{
- obj: cj,
- integration: integration,
- client: action.client,
+ obj: cj,
}
default:
return nil, fmt.Errorf("unsupported controller for integration
%s", integration.Name)
@@ -381,230 +345,6 @@ func getUpdatedController(env *trait.Environment, obj
ctrl.Object) ctrl.Object {
})
}
-func (action *monitorAction) updateIntegrationPhaseAndReadyCondition(
- ctx context.Context, controller controller, environment
*trait.Environment, integration *v1.Integration,
- pendingPods []corev1.Pod, runningPods []corev1.Pod,
-) error {
- if done, err := controller.checkReadyCondition(ctx); done || err != nil
{
- // There may be pods that are not ready but still probable for
getting error messages.
- // Ignore returned error from probing as it's expected when the
ctrl obj is not ready.
- _, _, _ = action.probeReadiness(ctx, environment, integration,
runningPods)
-
- return err
- }
- if arePodsFailingStatuses(integration, pendingPods, runningPods) {
- return nil
- }
- readyPods, probeOk, err := action.probeReadiness(ctx, environment,
integration, runningPods)
- if err != nil {
- return err
- }
- if !probeOk {
- integration.Status.Phase = v1.IntegrationPhaseError
-
- return nil
- }
- if done := controller.updateReadyCondition(readyPods); done {
- integration.Status.Phase = v1.IntegrationPhaseRunning
-
- return nil
- }
-
- return nil
-}
-
-func arePodsFailingStatuses(integration *v1.Integration, pendingPods
[]corev1.Pod, runningPods []corev1.Pod) bool {
- // Check Pods statuses
- for _, pod := range pendingPods {
- // Check the scheduled condition
- if scheduled := kubernetes.GetPodCondition(pod,
corev1.PodScheduled); scheduled != nil &&
- scheduled.Status == corev1.ConditionFalse &&
- scheduled.Reason == "Unschedulable" {
- integration.Status.Phase = v1.IntegrationPhaseError
- integration.SetReadyConditionError(scheduled.Message)
-
- return true
- }
- }
- // Check pending container statuses
- for _, pod := range pendingPods {
- containers := make([]corev1.ContainerStatus, 0,
len(pod.Status.InitContainerStatuses)+len(pod.Status.ContainerStatuses))
- containers = append(containers,
pod.Status.InitContainerStatuses...)
- containers = append(containers, pod.Status.ContainerStatuses...)
- for _, container := range containers {
- // Check the images are pulled
- if waiting := container.State.Waiting; waiting != nil
&& waiting.Reason == "ImagePullBackOff" {
- integration.Status.Phase =
v1.IntegrationPhaseError
-
integration.SetReadyConditionError(waiting.Message)
-
- return true
- }
- }
- }
- // Check running container statuses
- for _, pod := range runningPods {
- if pod.DeletionTimestamp != nil {
- continue
- }
- containers := make([]corev1.ContainerStatus, 0,
len(pod.Status.InitContainerStatuses)+len(pod.Status.ContainerStatuses))
- containers = append(containers,
pod.Status.InitContainerStatuses...)
- containers = append(containers, pod.Status.ContainerStatuses...)
- for _, container := range containers {
- // Check the container state
- if waiting := container.State.Waiting; waiting != nil
&& waiting.Reason == "CrashLoopBackOff" {
- integration.Status.Phase =
v1.IntegrationPhaseError
-
integration.SetReadyConditionError(waiting.Message)
-
- return true
- }
- if terminated := container.State.Terminated; terminated
!= nil && terminated.Reason == "Error" {
- integration.Status.Phase =
v1.IntegrationPhaseError
-
integration.SetReadyConditionError(terminated.Message)
-
- return true
- }
- }
- }
-
- return false
-}
-
-// probeReadiness calls the readiness probes of the non-ready Pods directly to
retrieve insights from the Camel runtime.
-// The func return the number of readyPods, the success of the probe and any
error may have happened during its execution.
-//
-//nolint:staticcheck
-func (action *monitorAction) probeReadiness(ctx context.Context, environment
*trait.Environment, integration *v1.Integration, pods []corev1.Pod) (int32,
bool, error) {
- // as a default we assume the Integration is Ready
- readyCondition := v1.IntegrationCondition{
- Type: v1.IntegrationConditionReady,
- Status: corev1.ConditionTrue,
- DeprecatedPods: make([]v1.PodCondition, len(pods)),
- }
-
- readyPods := int32(0)
- unreadyPods := int32(0)
-
- runtimeReady := true
- runtimeFailed := false
- probeReadinessOk := true
-
- for i := range pods {
- pod := &pods[i]
- readyCondition.DeprecatedPods[i].Name = pod.Name
- for p := range pod.Status.Conditions {
- if pod.Status.Conditions[p].Type == corev1.PodReady {
- readyCondition.DeprecatedPods[i].Condition =
pod.Status.Conditions[p]
-
- break
- }
- }
- // If it's in ready status, then we don't care to probe.
- if ready := kubernetes.GetPodCondition(*pod, corev1.PodReady);
ready.Status == corev1.ConditionTrue {
- readyPods++
-
- continue
- }
- unreadyPods++
- container := getIntegrationContainer(environment, pod)
- if container == nil {
- return readyPods, false, fmt.Errorf("integration
container not found in Pod %s/%s", pod.Namespace, pod.Name)
- }
-
- // TODO: this code must be moved to a dedicated function
- //
- //nolint:nestif
- if probe := container.ReadinessProbe; probe != nil &&
probe.HTTPGet != nil {
- body, err := proxyGetHTTPProbe(ctx, action.client,
probe, pod, container)
- // When invoking the HTTP probe, the kubernetes client
exposes a very
- // specific behavior:
- //
- // - if there is no error, that means the pod in not
ready just because
- // the probe has to be called few time as per
configuration, so it means
- // it's not ready, but the probe is OK, and the pod
could become ready
- // at some point
- // - if the error is Service Unavailable (HTTP 503)
then it means the pod
- // is not ready and the probe is failing, in this
case we can use the
- // response to scrape for camel info
- //
- // Here an example of a failed probe (from curl):
- //
- // Trying 127.0.0.1:8080...
- // TCP_NODELAY set
- // Connected to localhost (127.0.0.1) port 8080 (#0)
- // GET /q/health/ready HTTP/1.1
- // Host: localhost:8080
- // User-Agent: curl/7.68.0
- // Accept: */*
- //
- // Mark bundle as not supporting multiuse
- // HTTP/1.1 503 Service Unavailable
- // content-type: application/json; charset=UTF-8
- // content-length: 871
- //
- // {
- // "status": "DOWN",
- // "checks": [ {
- // "name": "camel-routes",
- // "status": "DOWN",
- // "data": {
- // "route.id": "route1",
- // "route.status": "Stopped",
- // "check.kind": "READINESS"
- // }
- // }]
- // }
- if err == nil {
- continue
- }
-
- if errors.Is(err, context.DeadlineExceeded) {
-
readyCondition.DeprecatedPods[i].Condition.Message = fmt.Sprintf("readiness
probe timed out for Pod %s/%s", pod.Namespace, pod.Name)
- runtimeReady = false
-
- continue
- }
- if !k8serrors.IsServiceUnavailable(err) {
-
readyCondition.DeprecatedPods[i].Condition.Message = fmt.Sprintf("readiness
probe failed for Pod %s/%s: %s", pod.Namespace, pod.Name, err.Error())
- runtimeReady = false
-
- continue
- }
-
- health, err := NewHealthCheck(body)
- if err != nil {
- return readyPods, false, err
- }
- for _, check := range health.Checks {
- if check.Status == v1.HealthCheckStatusUp {
- continue
- }
-
- runtimeReady = false
- runtimeFailed = true
-
- readyCondition.DeprecatedPods[i].Health =
append(readyCondition.DeprecatedPods[i].Health, check)
- }
- }
- }
-
- if runtimeFailed {
- probeReadinessOk = false
- readyCondition.Reason = v1.IntegrationConditionErrorReason
- readyCondition.Status = corev1.ConditionFalse
- readyCondition.Message = fmt.Sprintf("%d/%d pods are not
ready", unreadyPods, unreadyPods+readyPods)
- integration.Status.SetConditions(readyCondition)
- }
- if !runtimeReady {
- probeReadinessOk = false
- readyCondition.Reason =
v1.IntegrationConditionRuntimeNotReadyReason
- readyCondition.Status = corev1.ConditionFalse
- readyCondition.Message = fmt.Sprintf("%d/%d pods are not
ready", unreadyPods, unreadyPods+readyPods)
- integration.Status.SetConditions(readyCondition)
- }
-
- return readyPods, probeReadinessOk, nil
-}
-
func findHighestPriorityReadyKit(kits []v1.IntegrationKit)
(*v1.IntegrationKit, error) {
if len(kits) == 0 {
return nil, nil
@@ -627,14 +367,3 @@ func findHighestPriorityReadyKit(kits []v1.IntegrationKit)
(*v1.IntegrationKit,
return kit, nil
}
-
-func getIntegrationContainer(environment *trait.Environment, pod *corev1.Pod)
*corev1.Container {
- name := environment.GetIntegrationContainerName()
- for i, container := range pod.Spec.Containers {
- if container.Name == name {
- return &pod.Spec.Containers[i]
- }
- }
-
- return nil
-}
diff --git a/pkg/controller/integration/monitor_cronjob.go
b/pkg/controller/integration/monitor_cronjob.go
index 692d95f3e..122821746 100644
--- a/pkg/controller/integration/monitor_cronjob.go
+++ b/pkg/controller/integration/monitor_cronjob.go
@@ -18,108 +18,46 @@ limitations under the License.
package integration
import (
- "context"
- "fmt"
-
batchv1 "k8s.io/api/batch/v1"
corev1 "k8s.io/api/core/v1"
- 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/client"
- "github.com/apache/camel-k/v2/pkg/util/kubernetes"
)
type cronJobController struct {
- obj *batchv1.CronJob
- integration *v1.Integration
- client client.Client
- lastCompletedJob *batchv1.Job
+ obj *batchv1.CronJob
}
-var _ controller = &cronJobController{}
-
-func (c *cronJobController) checkReadyCondition(ctx context.Context) (bool,
error) {
- // Check latest job result
- lastScheduleTime := c.obj.Status.LastScheduleTime
-
- if lastScheduleTime == nil || len(c.obj.Status.Active) != 0 {
- return false, nil
- }
+func (c *cronJobController) setReadyCondition(integration *v1.Integration) {
+ cronJob := c.obj
- jobs := batchv1.JobList{}
- if err := c.client.List(ctx, &jobs,
- ctrl.InNamespace(c.integration.Namespace),
- ctrl.MatchingLabels{v1.IntegrationLabel: c.integration.Name},
- ); err != nil {
- return true, err
+ // No scheduled job yet: leave the Ready condition unchanged.
+ if cronJob.Status.LastScheduleTime == nil {
+ return
}
- t := lastScheduleTime.Time
- for i, job := range jobs.Items {
- if job.Status.Active == 0 &&
job.CreationTimestamp.Time.Before(t) {
- continue
- }
- c.lastCompletedJob = &jobs.Items[i]
- t = c.lastCompletedJob.CreationTimestamp.Time
- }
- if c.lastCompletedJob != nil {
- if failed := kubernetes.GetJobCondition(*c.lastCompletedJob,
batchv1.JobFailed); failed != nil &&
- failed.Status == corev1.ConditionTrue {
- c.integration.SetReadyCondition(corev1.ConditionFalse,
- v1.IntegrationConditionLastJobFailedReason,
- fmt.Sprintf("last job %s failed: %s",
c.lastCompletedJob.Name, failed.Message))
- c.integration.Status.Phase = v1.IntegrationPhaseError
- return true, nil
- }
+ // A job is actively running: leave the Ready condition unchanged.
+ if len(cronJob.Status.Active) > 0 {
+ return
}
- return false, nil
-}
-
-func (c *cronJobController) updateReadyCondition(readyPods int32) bool {
- switch {
- case c.obj.Status.LastScheduleTime == nil:
- c.integration.SetReadyCondition(corev1.ConditionTrue,
- v1.IntegrationConditionCronJobCreatedReason, "cronjob
created")
-
- return true
-
- case len(c.obj.Status.Active) > 0:
- c.integration.SetReadyCondition(corev1.ConditionTrue,
- v1.IntegrationConditionCronJobActiveReason, "cronjob
active")
-
- return true
-
- case c.obj.Spec.SuccessfulJobsHistoryLimit != nil &&
*c.obj.Spec.SuccessfulJobsHistoryLimit == 0 &&
- c.obj.Spec.FailedJobsHistoryLimit != nil &&
*c.obj.Spec.FailedJobsHistoryLimit == 0:
- c.integration.SetReadyCondition(corev1.ConditionTrue,
- v1.IntegrationConditionCronJobCreatedReason, "no jobs
history available")
-
- return true
-
- case c.lastCompletedJob != nil:
- if complete := kubernetes.GetJobCondition(*c.lastCompletedJob,
batchv1.JobComplete); complete != nil &&
- complete.Status == corev1.ConditionTrue {
- c.integration.SetReadyCondition(corev1.ConditionTrue,
- v1.IntegrationConditionLastJobSucceededReason,
- fmt.Sprintf("last job %s completed
successfully", c.lastCompletedJob.Name))
-
- return true
- }
-
- default:
- c.integration.SetReadyCondition(corev1.ConditionUnknown, "", "")
+ // The last scheduled job has not succeeded.
+ if cronJob.Status.LastSuccessfulTime == nil ||
+
cronJob.Status.LastSuccessfulTime.Before(cronJob.Status.LastScheduleTime) {
+ integration.Status.Phase = v1.IntegrationPhaseError
+ integration.SetReadyCondition(
+ corev1.ConditionFalse,
+ v1.IntegrationConditionLastJobFailedReason,
+ "The last scheduled job has not succeeded",
+ )
+
+ return
}
- return false
-}
-
-func (c *cronJobController) hasTemplateIntegrationLabel() bool {
- return c.obj.Spec.JobTemplate.Spec.Template.Labels[v1.IntegrationLabel]
!= ""
-}
-
-func (c *cronJobController) getControllerName() string {
- return "CronJob/" + c.obj.Name
+ // The last scheduled job succeeded.
+ integration.SetReadyCondition(
+ corev1.ConditionTrue,
+ v1.IntegrationConditionLastJobSucceededReason,
+ "The last scheduled job succeeded",
+ )
}
diff --git a/pkg/controller/integration/monitor_cronjob_test.go
b/pkg/controller/integration/monitor_cronjob_test.go
new file mode 100644
index 000000000..0e7d3186a
--- /dev/null
+++ b/pkg/controller/integration/monitor_cronjob_test.go
@@ -0,0 +1,147 @@
+/*
+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 integration
+
+import (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ corev1 "k8s.io/api/core/v1"
+ metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+ v1 "github.com/apache/camel-k/v2/pkg/apis/camel/v1"
+ batchv1 "k8s.io/api/batch/v1"
+)
+
+func TestCronJobController_setReadyCondition(t *testing.T) {
+ now := metav1.Now()
+ past := metav1.NewTime(now.Add(-1 * time.Hour))
+
+ tests := []struct {
+ name string
+ lastSchedule *metav1.Time
+ lastSuccessful *metav1.Time
+ activeJobs []corev1.ObjectReference
+ initialPhase v1.IntegrationPhase
+ initialReady *v1.IntegrationCondition
+ expectedPhase v1.IntegrationPhase
+ expectedReady *corev1.ConditionStatus
+ expectedReason string
+ expectedMessage string
+ }{
+ {
+ name: "no scheduled job",
+ initialPhase: v1.IntegrationPhaseNone,
+ expectedPhase: v1.IntegrationPhaseNone,
+ },
+ {
+ name: "active job",
+ lastSchedule: &now,
+ lastSuccessful: &past,
+ activeJobs: []corev1.ObjectReference{{Name:
"active-job"}},
+ initialPhase: v1.IntegrationPhaseNone,
+ expectedPhase: v1.IntegrationPhaseNone,
+ },
+ {
+ name: "last scheduled job has never
succeeded",
+ lastSchedule: &now,
+ initialPhase: v1.IntegrationPhaseNone,
+ expectedPhase: v1.IntegrationPhaseError,
+ expectedReady:
conditionStatusPtr(corev1.ConditionFalse),
+ expectedReason:
v1.IntegrationConditionLastJobFailedReason,
+ expectedMessage: "The last scheduled job has not
succeeded",
+ },
+ {
+ name: "last scheduled job failed",
+ lastSchedule: &now,
+ lastSuccessful: &past,
+ initialPhase: v1.IntegrationPhaseNone,
+ expectedPhase: v1.IntegrationPhaseError,
+ expectedReady:
conditionStatusPtr(corev1.ConditionFalse),
+ expectedReason:
v1.IntegrationConditionLastJobFailedReason,
+ expectedMessage: "The last scheduled job has not
succeeded",
+ },
+ {
+ name: "last scheduled job succeeded",
+ lastSchedule: &past,
+ lastSuccessful: &now,
+ initialPhase: v1.IntegrationPhaseNone,
+ expectedPhase: v1.IntegrationPhaseNone,
+ expectedReady:
conditionStatusPtr(corev1.ConditionTrue),
+ expectedReason:
v1.IntegrationConditionLastJobSucceededReason,
+ expectedMessage: "The last scheduled job succeeded",
+ },
+ {
+ name: "last successful time equals last
schedule time",
+ lastSchedule: &now,
+ lastSuccessful: &now,
+ expectedPhase: v1.IntegrationPhaseNone,
+ expectedReady:
conditionStatusPtr(corev1.ConditionTrue),
+ expectedReason:
v1.IntegrationConditionLastJobSucceededReason,
+ expectedMessage: "The last scheduled job succeeded",
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ cronJob := &batchv1.CronJob{
+ Status: batchv1.CronJobStatus{
+ LastScheduleTime: tt.lastSchedule,
+ LastSuccessfulTime: tt.lastSuccessful,
+ Active: tt.activeJobs,
+ },
+ }
+
+ integration := &v1.Integration{
+ Status: v1.IntegrationStatus{
+ Phase: tt.initialPhase,
+ },
+ }
+
+ if tt.initialReady != nil {
+ integration.SetReadyCondition(
+
corev1.ConditionStatus(tt.initialReady.Status),
+ tt.initialReady.Reason,
+ tt.initialReady.Message,
+ )
+ }
+
+ controller := &cronJobController{obj: cronJob}
+ controller.setReadyCondition(integration)
+ assert.Equal(t, tt.expectedPhase,
integration.Status.Phase)
+
+ ready :=
integration.Status.GetCondition(v1.IntegrationConditionReady)
+ if tt.expectedReady == nil {
+ assert.Nil(t, ready)
+ return
+ }
+
+ require.NotNil(t, ready)
+ assert.Equal(t, *tt.expectedReady,
corev1.ConditionStatus(ready.Status))
+ assert.Equal(t, tt.expectedReason, ready.Reason)
+ assert.Equal(t, tt.expectedMessage, ready.Message)
+ })
+ }
+}
+
+func conditionStatusPtr(status corev1.ConditionStatus) *corev1.ConditionStatus
{
+ return &status
+}
diff --git a/pkg/controller/integration/monitor_deployment.go
b/pkg/controller/integration/monitor_deployment.go
index 637d9478c..43a9f46b3 100644
--- a/pkg/controller/integration/monitor_deployment.go
+++ b/pkg/controller/integration/monitor_deployment.go
@@ -18,9 +18,6 @@ limitations under the License.
package integration
import (
- "context"
- "fmt"
-
appsv1 "k8s.io/api/apps/v1"
corev1 "k8s.io/api/core/v1"
@@ -29,72 +26,41 @@ import (
)
type deploymentController struct {
- obj *appsv1.Deployment
- integration *v1.Integration
+ obj *appsv1.Deployment
}
-var _ controller = &deploymentController{}
-
-func (c *deploymentController) checkReadyCondition(ctx context.Context) (bool,
error) {
- // Check the Deployment progression
+func (c *deploymentController) setReadyCondition(integration *v1.Integration) {
+ // Check the Deployment progression and availability
progressing := kubernetes.GetDeploymentCondition(*c.obj,
appsv1.DeploymentProgressing)
replicaFailure := kubernetes.GetDeploymentCondition(*c.obj,
appsv1.DeploymentReplicaFailure)
+ available := kubernetes.GetDeploymentCondition(*c.obj,
appsv1.DeploymentAvailable)
if replicaFailure != nil && replicaFailure.Status ==
corev1.ConditionTrue {
- c.integration.Status.Phase = v1.IntegrationPhaseError
- c.integration.SetReadyConditionError(replicaFailure.Message)
+ integration.Status.Phase = v1.IntegrationPhaseError
+ integration.SetReadyConditionError(replicaFailure.Message)
- return true, nil
+ return
}
if progressing != nil && progressing.Status == corev1.ConditionFalse &&
progressing.Reason == "ProgressDeadlineExceeded" {
- c.integration.Status.Phase = v1.IntegrationPhaseError
- c.integration.SetReadyConditionError(progressing.Message)
+ integration.Status.Phase = v1.IntegrationPhaseError
+ integration.SetReadyConditionError(progressing.Message)
- return true, nil
+ return
}
-
- return false, nil
-}
-
-func (c *deploymentController) updateReadyCondition(readyPods int32) bool {
- replicas := int32(1)
- if r := c.integration.Spec.Replicas; r != nil {
- replicas = *r
- }
- // The Deployment status reports updated and ready replicas separately,
- // so that the number of ready replicas also accounts for older
versions.
- readyReplicas := readyPods
- switch {
- case readyReplicas >= replicas:
- // The Integration is considered ready when the number of
replicas
- // reported to be ready is larger than or equal to the
specified number
- // of replicas. This avoids reporting a falsy readiness
condition
- // when the Integration is being down-scaled.
- c.integration.SetReadyCondition(corev1.ConditionTrue,
+ if available == nil || available.Status != corev1.ConditionTrue {
+ message := "Deployment does not have minimum availability."
+ if available != nil && available.Message != "" {
+ message = available.Message
+ }
+
+ integration.SetReadyCondition(
+ corev1.ConditionFalse,
v1.IntegrationConditionDeploymentReadyReason,
- fmt.Sprintf("%d/%d ready replicas", readyReplicas,
replicas))
-
- return true
-
- case c.obj.Status.UpdatedReplicas < replicas:
- c.integration.SetReadyCondition(corev1.ConditionFalse,
- v1.IntegrationConditionDeploymentProgressingReason,
- fmt.Sprintf("%d/%d updated replicas",
c.obj.Status.UpdatedReplicas, replicas))
-
- default:
- c.integration.SetReadyCondition(corev1.ConditionFalse,
- v1.IntegrationConditionDeploymentProgressingReason,
- fmt.Sprintf("%d/%d ready replicas", readyReplicas,
replicas))
+ message,
+ )
+ return
}
- return false
-}
-
-func (c *deploymentController) hasTemplateIntegrationLabel() bool {
- return c.obj.Spec.Template.Labels[v1.IntegrationLabel] != ""
-}
-
-func (c *deploymentController) getControllerName() string {
- return "Deployment/" + c.obj.Name
+ integration.SetReadyCondition(corev1.ConditionTrue,
v1.IntegrationConditionDeploymentReadyReason, "Ready")
}
diff --git a/pkg/controller/integration/monitor_deployment_test.go
b/pkg/controller/integration/monitor_deployment_test.go
new file mode 100644
index 000000000..bb2275fa7
--- /dev/null
+++ b/pkg/controller/integration/monitor_deployment_test.go
@@ -0,0 +1,133 @@
+/*
+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 integration
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+ "github.com/stretchr/testify/require"
+
+ corev1 "k8s.io/api/core/v1"
+
+ v1 "github.com/apache/camel-k/v2/pkg/apis/camel/v1"
+ appsv1 "k8s.io/api/apps/v1"
+)
+
+func TestDeploymentController_setReadyCondition(t *testing.T) {
+ tests := []struct {
+ name string
+ conditions []appsv1.DeploymentCondition
+ expectedPhase v1.IntegrationPhase
+ expectedReady corev1.ConditionStatus
+ expectedReason string
+ expectedMessage string
+ }{
+ {
+ name: "replica failure",
+ conditions: []appsv1.DeploymentCondition{
+ {
+ Type:
appsv1.DeploymentReplicaFailure,
+ Status: corev1.ConditionTrue,
+ Reason: "FailedCreate",
+ Message: "Failed to create replicas",
+ },
+ },
+ expectedPhase: v1.IntegrationPhaseError,
+ expectedReady: corev1.ConditionFalse,
+ expectedMessage: "Failed to create replicas",
+ },
+ {
+ name: "progress deadline exceeded",
+ conditions: []appsv1.DeploymentCondition{
+ {
+ Type: appsv1.DeploymentProgressing,
+ Status: corev1.ConditionFalse,
+ Reason: "ProgressDeadlineExceeded",
+ Message: "Deployment exceeded its
progress deadline",
+ },
+ },
+ expectedPhase: v1.IntegrationPhaseError,
+ expectedReady: corev1.ConditionFalse,
+ expectedMessage: "Deployment exceeded its progress
deadline",
+ },
+ {
+ name: "progressing with another reason",
+ conditions: []appsv1.DeploymentCondition{
+ {
+ Type: appsv1.DeploymentProgressing,
+ Status: corev1.ConditionFalse,
+ Reason: "ReplicaSetUpdated",
+ Message: "ReplicaSet is being updated",
+ },
+ },
+ expectedReady: corev1.ConditionTrue,
+ expectedReason:
v1.IntegrationConditionDeploymentReadyReason,
+ expectedMessage: "Ready",
+ },
+ {
+ name: "replica failure condition is false",
+ conditions: []appsv1.DeploymentCondition{
+ {
+ Type: appsv1.DeploymentReplicaFailure,
+ Status: corev1.ConditionFalse,
+ },
+ },
+ expectedReady: corev1.ConditionTrue,
+ expectedReason:
v1.IntegrationConditionDeploymentReadyReason,
+ expectedMessage: "Ready",
+ },
+ {
+ name: "deployment has no conditions",
+ expectedReady: corev1.ConditionTrue,
+ expectedReason:
v1.IntegrationConditionDeploymentReadyReason,
+ expectedMessage: "Ready",
+ },
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ deployment := &appsv1.Deployment{
+ Status: appsv1.DeploymentStatus{
+ Conditions: tt.conditions,
+ },
+ }
+
+ integration := &v1.Integration{}
+
+ controller := &deploymentController{
+ obj: deployment,
+ }
+
+ controller.setReadyCondition(integration)
+ assert.Equal(t, tt.expectedPhase,
integration.Status.Phase)
+
+ ready :=
integration.Status.GetCondition(v1.IntegrationConditionReady)
+ require.NotNil(t, ready)
+ assert.Equal(t, tt.expectedReady,
corev1.ConditionStatus(ready.Status))
+
+ if tt.expectedReason != "" {
+ assert.Equal(t, tt.expectedReason, ready.Reason)
+ }
+
+ if tt.expectedMessage != "" {
+ assert.Equal(t, tt.expectedMessage,
ready.Message)
+ }
+ })
+ }
+}
diff --git a/pkg/controller/integration/monitor_knative.go
b/pkg/controller/integration/monitor_knative.go
index 949969f5b..8ac3c54c9 100644
--- a/pkg/controller/integration/monitor_knative.go
+++ b/pkg/controller/integration/monitor_knative.go
@@ -18,8 +18,6 @@ limitations under the License.
package integration
import (
- "context"
-
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -29,48 +27,19 @@ import (
)
type knativeServiceController struct {
- obj *servingv1.Service
- integration *v1.Integration
+ obj *servingv1.Service
}
-var _ controller = &knativeServiceController{}
-
-func (c *knativeServiceController) checkReadyCondition(ctx context.Context)
(bool, error) {
+func (c *knativeServiceController) setReadyCondition(integration
*v1.Integration) {
// Check the KnativeService conditions
if ready := kubernetes.GetKnativeServiceCondition(*c.obj,
servingv1.ServiceConditionReady); ready != nil &&
ready.Status == metav1.ConditionFalse && ready.Reason ==
"RevisionFailed" {
- c.integration.Status.Phase = v1.IntegrationPhaseError
- c.integration.SetReadyConditionError(ready.Message)
-
- return true, nil
- }
-
- return false, nil
-}
-
-func (c *knativeServiceController) updateReadyCondition(readyPods int32) bool {
- ready := kubernetes.GetKnativeServiceCondition(*c.obj,
servingv1.ServiceConditionReady)
- if ready == nil {
- c.integration.SetReadyCondition(corev1.ConditionFalse, "", "")
+ integration.Status.Phase = v1.IntegrationPhaseError
+ integration.SetReadyConditionError(ready.Message)
- return false
+ return
}
- if ready.Status == metav1.ConditionTrue {
- c.integration.SetReadyCondition(corev1.ConditionTrue,
- v1.IntegrationConditionKnativeServiceReadyReason, "")
-
- return true
- }
- c.integration.SetReadyCondition(corev1.ConditionFalse,
- ready.Reason, ready.Message)
-
- return false
-}
-
-func (c *knativeServiceController) hasTemplateIntegrationLabel() bool {
- return c.obj.Spec.Template.Labels[v1.IntegrationLabel] != ""
-}
-func (c *knativeServiceController) getControllerName() string {
- return "KnativeService/" + c.obj.Name
+ integration.SetReadyCondition(corev1.ConditionTrue,
+ v1.IntegrationConditionKnativeServiceReadyReason, "Ready")
}
diff --git a/pkg/controller/integration/monitor_knative_test.go
b/pkg/controller/integration/monitor_knative_test.go
index 7e87047a9..ed549e55a 100644
--- a/pkg/controller/integration/monitor_knative_test.go
+++ b/pkg/controller/integration/monitor_knative_test.go
@@ -18,8 +18,6 @@ limitations under the License.
package integration
import (
- "context"
- "encoding/json"
"testing"
"github.com/stretchr/testify/assert"
@@ -27,195 +25,94 @@ import (
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
- "k8s.io/utils/ptr"
v1 "github.com/apache/camel-k/v2/pkg/apis/camel/v1"
+ "github.com/apache/camel-k/v2/pkg/apis/duck/knative/apis"
servingv1
"github.com/apache/camel-k/v2/pkg/apis/duck/knative/serving/v1"
)
-func newKnativeServiceController(ready *metav1.Condition)
*knativeServiceController {
- svc := &servingv1.Service{
- ObjectMeta: metav1.ObjectMeta{
- Namespace: "ns",
- Name: "my-ksvc",
+func TestKnativeServiceController_setReadyCondition(t *testing.T) {
+ tests := []struct {
+ name string
+ condition *metav1.Condition
+ expectedPhase v1.IntegrationPhase
+ expectedReady corev1.ConditionStatus
+ expectedReason string
+ expectedMessage string
+ }{
+ {
+ name: "revision failed",
+ condition: &metav1.Condition{
+ Type: servingv1.ServiceConditionReady,
+ Status: metav1.ConditionFalse,
+ Reason: "RevisionFailed",
+ Message: "Revision failed to become ready",
+ },
+ expectedPhase: v1.IntegrationPhaseError,
+ expectedReady: corev1.ConditionFalse,
+ expectedMessage: "Revision failed to become ready",
},
- }
- if ready != nil {
- svc.Status.Conditions = []metav1.Condition{*ready}
- }
-
- return &knativeServiceController{
- obj: svc,
- integration: &v1.Integration{
- ObjectMeta: metav1.ObjectMeta{
- Namespace: "ns",
- Name: "my-it",
+ {
+ name: "service ready",
+ condition: &metav1.Condition{
+ Type: servingv1.ServiceConditionReady,
+ Status: metav1.ConditionTrue,
},
- Status: v1.IntegrationStatus{
- Phase: v1.IntegrationPhaseRunning,
+ expectedReady: corev1.ConditionTrue,
+ expectedReason:
v1.IntegrationConditionKnativeServiceReadyReason,
+ expectedMessage: "Ready",
+ },
+ {
+ name: "service not ready for another reason",
+ condition: &metav1.Condition{
+ Type: servingv1.ServiceConditionReady,
+ Status: metav1.ConditionFalse,
+ Reason: "Deploying",
+ Message: "Revision is deploying",
},
+ expectedReady: corev1.ConditionTrue,
+ expectedReason:
v1.IntegrationConditionKnativeServiceReadyReason,
+ expectedMessage: "Ready",
+ },
+ {
+ name: "no ready condition",
+ expectedReady: corev1.ConditionTrue,
+ expectedReason:
v1.IntegrationConditionKnativeServiceReadyReason,
+ expectedMessage: "Ready",
},
}
-}
-
-func TestKnativeServiceControllerReady(t *testing.T) {
- c := newKnativeServiceController(&metav1.Condition{
- Type: servingv1.ServiceConditionReady,
- Status: metav1.ConditionTrue,
- })
-
- done, err := c.checkReadyCondition(context.TODO())
- require.NoError(t, err)
- assert.False(t, done)
- assert.Equal(t, v1.IntegrationPhaseRunning, c.integration.Status.Phase)
-
- assert.True(t, c.updateReadyCondition(1))
- cond := c.integration.Status.GetCondition(v1.IntegrationConditionReady)
- require.NotNil(t, cond)
- assert.Equal(t, corev1.ConditionTrue, cond.Status)
- assert.Equal(t, v1.IntegrationConditionKnativeServiceReadyReason,
cond.Reason)
-}
-
-func TestKnativeServiceControllerRevisionFailed(t *testing.T) {
- c := newKnativeServiceController(&metav1.Condition{
- Type: servingv1.ServiceConditionReady,
- Status: metav1.ConditionFalse,
- Reason: "RevisionFailed",
- Message: "revision failed",
- })
-
- done, err := c.checkReadyCondition(context.TODO())
- require.NoError(t, err)
- assert.True(t, done)
- assert.Equal(t, v1.IntegrationPhaseError, c.integration.Status.Phase)
- cond := c.integration.Status.GetCondition(v1.IntegrationConditionReady)
- require.NotNil(t, cond)
- assert.Equal(t, corev1.ConditionFalse, cond.Status)
- assert.Equal(t, "revision failed", cond.Message)
-}
-
-func TestKnativeServiceControllerNotReady(t *testing.T) {
- c := newKnativeServiceController(&metav1.Condition{
- Type: servingv1.ServiceConditionReady,
- Status: metav1.ConditionFalse,
- Reason: "Deploying",
- Message: "still deploying",
- })
-
- done, err := c.checkReadyCondition(context.TODO())
- require.NoError(t, err)
- assert.False(t, done)
- assert.Equal(t, v1.IntegrationPhaseRunning, c.integration.Status.Phase)
- assert.False(t, c.updateReadyCondition(0))
- cond := c.integration.Status.GetCondition(v1.IntegrationConditionReady)
- require.NotNil(t, cond)
- assert.Equal(t, corev1.ConditionFalse, cond.Status)
- assert.Equal(t, "Deploying", cond.Reason)
- assert.Equal(t, "still deploying", cond.Message)
-}
-
-func TestKnativeServiceControllerUnknown(t *testing.T) {
- c := newKnativeServiceController(&metav1.Condition{
- Type: servingv1.ServiceConditionReady,
- Status: metav1.ConditionUnknown,
- Reason: "RevisionMissing",
- Message: "waiting for the revision",
- })
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ ksvc := &servingv1.Service{}
- done, err := c.checkReadyCondition(context.TODO())
- require.NoError(t, err)
- assert.False(t, done)
- assert.Equal(t, v1.IntegrationPhaseRunning, c.integration.Status.Phase)
-
- assert.False(t, c.updateReadyCondition(0))
- cond := c.integration.Status.GetCondition(v1.IntegrationConditionReady)
- require.NotNil(t, cond)
- assert.Equal(t, corev1.ConditionFalse, cond.Status)
- assert.Equal(t, "RevisionMissing", cond.Reason)
- assert.Equal(t, "waiting for the revision", cond.Message)
-}
+ if tt.condition != nil {
+ ksvc.Status = servingv1.ServiceStatus{
+ Status: apis.Status{
+ Conditions:
[]metav1.Condition{*tt.condition},
+ },
+ }
+ }
-func TestKnativeServiceControllerMissingCondition(t *testing.T) {
- c := newKnativeServiceController(nil)
+ integration := &v1.Integration{}
+ controller := &knativeServiceController{
+ obj: ksvc,
+ }
+ controller.setReadyCondition(integration)
+ assert.Equal(t, tt.expectedPhase,
integration.Status.Phase)
- done, err := c.checkReadyCondition(context.TODO())
- require.NoError(t, err)
- assert.False(t, done)
- assert.Equal(t, v1.IntegrationPhaseRunning, c.integration.Status.Phase)
+ ready :=
integration.Status.GetCondition(v1.IntegrationConditionReady)
- assert.False(t, c.updateReadyCondition(0))
- cond := c.integration.Status.GetCondition(v1.IntegrationConditionReady)
- require.NotNil(t, cond)
- assert.Equal(t, corev1.ConditionFalse, cond.Status)
-}
+ require.NotNil(t, ready)
+ assert.Equal(t, tt.expectedReady,
corev1.ConditionStatus(ready.Status))
-// knativeServicePayload mimics a Knative Serving Service as returned by the
API server:
-// it carries a number of fields the duck type does not model, which must be
ignored.
-const knativeServicePayload = `{
- "apiVersion": "serving.knative.dev/v1",
- "kind": "Service",
- "metadata": {"name": "my-ksvc", "namespace": "ns", "generation": 2},
- "spec": {
- "template": {
- "metadata": {"labels": {"camel.apache.org/integration":
"my-it"}},
- "spec": {
- "containerConcurrency": 0,
- "containers": [{"name": "integration", "image":
"my-image"}],
- "enableServiceLinks": false,
- "timeoutSeconds": 300
+ if tt.expectedReason != "" {
+ assert.Equal(t, tt.expectedReason, ready.Reason)
}
- },
- "traffic": [{"latestRevision": true, "percent": 100}]
- },
- "status": {
- "address": {"url": "http://my-ksvc.ns.svc.cluster.local"},
- "conditions": [
- {"lastTransitionTime": "2026-01-02T03:04:05Z",
"status": "True", "type": "ConfigurationsReady"},
- {"lastTransitionTime": "2026-01-02T03:04:05Z",
"status": "False", "type": "Ready",
- "reason": "RevisionFailed", "message":
"Revision \"my-ksvc-00001\" failed with message: Container failed."},
- {"lastTransitionTime": "2026-01-02T03:04:05Z",
"status": "False", "type": "RoutesReady",
- "reason": "RevisionMissing", "message":
"Configuration \"my-ksvc\" does not have any ready Revision.", "severity":
"Info"}
- ],
- "latestCreatedRevisionName": "my-ksvc-00001",
- "latestReadyRevisionName": "my-ksvc-00001",
- "observedGeneration": 2,
- "traffic": [{"latestRevision": true, "percent": 100,
"revisionName": "my-ksvc-00001"}],
- "url": "http://my-ksvc.ns.example.com";
- }
-}`
-
-func TestKnativeServiceControllerFromAPIServerPayload(t *testing.T) {
- svc := &servingv1.Service{}
- require.NoError(t, json.Unmarshal([]byte(knativeServicePayload), svc))
- assert.Equal(t, "http://my-ksvc.ns.example.com";,
svc.Status.URL.String())
- require.NotNil(t, svc.Status.Address)
- assert.Equal(t, "my-ksvc.ns.svc.cluster.local",
svc.Status.Address.URL.Host)
- assert.Equal(t, int64(300),
ptr.Deref(svc.Spec.Template.Spec.TimeoutSeconds, 0))
- assert.Len(t, svc.Spec.Template.Spec.Containers, 1)
-
- c := &knativeServiceController{
- obj: svc,
- integration: &v1.Integration{
- ObjectMeta: metav1.ObjectMeta{
- Namespace: "ns",
- Name: "my-it",
- },
- Status: v1.IntegrationStatus{
- Phase: v1.IntegrationPhaseRunning,
- },
- },
+ if tt.expectedMessage != "" {
+ assert.Equal(t, tt.expectedMessage,
ready.Message)
+ }
+ })
}
- assert.True(t, c.hasTemplateIntegrationLabel())
- assert.Equal(t, "KnativeService/my-ksvc", c.getControllerName())
-
- done, err := c.checkReadyCondition(context.TODO())
- require.NoError(t, err)
- assert.True(t, done)
- assert.Equal(t, v1.IntegrationPhaseError, c.integration.Status.Phase)
- cond := c.integration.Status.GetCondition(v1.IntegrationConditionReady)
- require.NotNil(t, cond)
- assert.Equal(t, corev1.ConditionFalse, cond.Status)
- assert.Contains(t, cond.Message, "my-ksvc-00001")
}
diff --git a/pkg/util/kubernetes/conditions.go
b/pkg/util/kubernetes/conditions.go
index f587d63c4..57500809b 100644
--- a/pkg/util/kubernetes/conditions.go
+++ b/pkg/util/kubernetes/conditions.go
@@ -19,24 +19,11 @@ package kubernetes
import (
appsv1 "k8s.io/api/apps/v1"
- batchv1 "k8s.io/api/batch/v1"
- corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
servingv1
"github.com/apache/camel-k/v2/pkg/apis/duck/knative/serving/v1"
)
-func GetPodCondition(pod corev1.Pod, conditionType corev1.PodConditionType)
*corev1.PodCondition {
- for i := range pod.Status.Conditions {
- condition := pod.Status.Conditions[i]
- if condition.Type == conditionType {
- return &condition
- }
- }
-
- return nil
-}
-
func GetDeploymentCondition(deployment appsv1.Deployment, conditionType
appsv1.DeploymentConditionType) *appsv1.DeploymentCondition {
for i := range deployment.Status.Conditions {
condition := deployment.Status.Conditions[i]
@@ -58,14 +45,3 @@ func GetKnativeServiceCondition(service servingv1.Service,
conditionType string)
return nil
}
-
-func GetJobCondition(job batchv1.Job, conditionType batchv1.JobConditionType)
*batchv1.JobCondition {
- for i := range job.Status.Conditions {
- condition := job.Status.Conditions[i]
- if condition.Type == conditionType {
- return &condition
- }
- }
-
- return nil
-}
