fix(knative): Integration in error state when knative SinkBinding mod…

…e=inclusion apache#4119

apache#4119

config/rbac/operator-role-knative.yaml

@@ -71,3 +71,10 @@ rules:
   - list
   - patch
   - update
+- apiGroups:
+  - ""
+  resources:
+  - namespaces
+  verbs:
+  - get
+  - patch

e2e/yaks/common/knative-sinkbinding-http/Rest2Channel.java

@@ -0,0 +1,26 @@
+/*
+ * 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.
+ */
+
+import org.apache.camel.builder.RouteBuilder;
+
+public class Rest2Channel extends RouteBuilder {
+    public void configure() throws Exception {
+        rest("/")
+            .put("/foo/new")
+            .to("knative:channel/messages");
+    }
+}

e2e/yaks/common/knative-sinkbinding-http/messages-channel.yaml

@@ -0,0 +1,21 @@
+# ---------------------------------------------------------------------------
+# 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.
+# ---------------------------------------------------------------------------
+
+apiVersion: messaging.knative.dev/v1
+kind: InMemoryChannel
+metadata:
+  name: messages

e2e/yaks/common/knative-sinkbinding-http/sinkbinding.feature

@@ -0,0 +1,9 @@
+Feature: Camel K can run source HTTP endpoint in sinkbinding mode
+
+  Background:
+    Given Kubernetes resource polling configuration
+      | maxAttempts          | 1   |
+      | delayBetweenAttempts | 500 |
+
+  Scenario: Integration knative-service starts with no errors
+    Given wait for condition=Ready on Kubernetes custom resource integration/rest2channel in integration.camel.apache.org/v1

e2e/yaks/common/knative-sinkbinding-http/yaks-config.yaml

@@ -0,0 +1,35 @@
+# ---------------------------------------------------------------------------
+# 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.
+# ---------------------------------------------------------------------------
+
+config:
+  namespace:
+    temporary: true
+pre:
+- name: installation
+  run: |
+    kubectl -n knative-eventing set env deployments eventing-webhook --containers="eventing-webhook" SINK_BINDING_SELECTION_MODE=inclusion
+    sleep 5s
+
+    kubectl apply -n $YAKS_NAMESPACE -f messages-channel.yaml
+
+    kamel run Rest2Channel.java -w -n $YAKS_NAMESPACE
+
+    kubectl wait integration --all --for=condition=Ready --timeout=10m -n $YAKS_NAMESPACE
+post:
+  - name: print dump
+    if: env:CI=true && failure()
+    run: yaks dump --includes app=camel-k

pkg/apis/camel/v1/trait/knative.go

@@ -57,4 +57,8 @@ type KnativeTrait struct {
 	SinkBinding *bool `property:"sink-binding" json:"sinkBinding,omitempty"`
 	// Enable automatic discovery of all trait properties.
 	Auto *bool `property:"auto" json:"auto,omitempty"`
+	// Enables the camel-k-operator to set the "bindings.knative.dev/include=true" label to the namespace
+	// As Knative requires this label to perform injection of K_SINK URL into the service.
+	// If this is false, the integration pod may start and fail, read the SinkBinding Knative documentation. (default: true)
+	NamespaceLabel *bool `property:"namespace-label" json:"namespaceLabel,omitempty"`
 }

pkg/trait/knative.go

@@ -469,6 +469,16 @@ func (t *knativeTrait) configureSinkBinding(e *Environment, env *knativeapi.Came
 						APIVersion: ref.APIVersion,
 					}
 
+					if pointer.BoolDeref(t.NamespaceLabel, true) {
+						// set the namespace label to allow automatic sinkbinding injection
+						enabled, err := knativeutil.EnableKnativeBindInNamespace(e.Ctx, e.Client, e.Integration.Namespace)
+						if err != nil {
+							t.L.Errorf(err, "Error setting label 'bindings.knative.dev/include=true' in namespace: %s", e.Integration.Namespace)
+						} else if enabled {
+							t.L.Infof("Label 'bindings.knative.dev/include=true' set in namespace: %s", e.Integration.Namespace)
+						}
+					}
+
 					// Add the SinkBinding in first position, to make sure it is created
 					// before the reference source, so that the SinkBinding webhook has
 					// all the information to perform injection.

pkg/util/knative/knative.go

@@ -19,6 +19,7 @@ package knative
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"net/url"
 
@@ -27,6 +28,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
 	"knative.dev/pkg/apis"
 
 	ctrl "sigs.k8s.io/controller-runtime/pkg/client"
@@ -206,3 +208,35 @@ func getSinkURI(ctx context.Context, c client.Client, sink *corev1.ObjectReferen
 	}
 	return addressURL.String(), nil
 }
+
+// EnableKnativeBindInNamespace sets the "bindings.knative.dev/include=true" label to the namespace, only
+// if there aren't any of these labels bindings.knative.dev/include bindings.knative.dev/exclude in the namespace
+// Returns true if the label was set in the namespace
+// https://knative.dev/docs/eventing/custom-event-source/sinkbinding/create-a-sinkbinding
+func EnableKnativeBindInNamespace(ctx context.Context, client client.Client, namespace string) (bool, error) {
+	ns, err := client.CoreV1().Namespaces().Get(ctx, namespace, metav1.GetOptions{})
+	if err != nil {
+		return false, err
+	}
+
+	// if there are sinkbinding labels in the namespace, camel-k-operator respects it and doesn't proceed
+	sinkbindingLabelsExists := ns.Labels["bindings.knative.dev/include"] != "" || ns.Labels["bindings.knative.dev/exclude"] != ""
+	if sinkbindingLabelsExists {
+		return false, nil
+	}
+
+	var jsonLabelPatch = map[string]interface{}{
+		"metadata": map[string]interface{}{
+			"labels": map[string]string{"bindings.knative.dev/include": "true"},
+		},
+	}
+	patch, err := json.Marshal(jsonLabelPatch)
+	if err != nil {
+		return false, err
+	}
+	_, err = client.CoreV1().Namespaces().Patch(ctx, namespace, types.MergePatchType, patch, metav1.PatchOptions{})
+	if err != nil {
+		return false, err
+	}
+	return true, nil
+}