Add cluster wide cm for env

.ci/clusters/global_env.yaml

@@ -0,0 +1,7 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: mesh-global-env
+data:
+  global1: globalvalue1
+  shared1: fromglobal

.ci/helm.sh

@@ -596,3 +596,16 @@ function ci::verify_log_topic_with_auth() {
   fi
   return 1
 }
+
+function ci::verify_env() {
+  pod="$1-function-0"
+  key=$2
+  expect=$3
+  result=$(kubectl exec -n ${NAMESPACE} ${pod} -- env | grep "${key}")
+  echo "$result"
+  echo "$expect"
+  if [[ "$result" = "$expect" ]]; then
+    return 0
+  fi
+  return 1
+}

.ci/tests/integration/cases/global-and-namespaced-env/env.yaml

@@ -0,0 +1,8 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: mesh-namespaced-env
+  namespace: default
+data:
+  namespaced1: namespacedvalue1
+  shared1: fromnamespace

.ci/tests/integration/cases/global-and-namespaced-env/manifests.yaml

@@ -0,0 +1,77 @@
+apiVersion: compute.functionmesh.io/v1alpha1
+kind: Function
+metadata:
+  name: function-sample-env
+  namespace: default
+spec:
+  image: streamnative/pulsar-functions-java-sample:2.9.2.23
+  className: org.apache.pulsar.functions.api.examples.ExclamationFunction
+  forwardSourceMessageProperty: true
+  maxPendingAsyncRequests: 1000
+  replicas: 1
+  maxReplicas: 5
+  logTopic: persistent://public/default/logging-function-logs
+  input:
+    topics:
+    - persistent://public/default/input-java-topic
+    typeClassName: java.lang.String
+  output:
+    topic: persistent://public/default/output-java-topic
+    typeClassName: java.lang.String
+  resources:
+    requests:
+      cpu: 50m
+      memory: 1G
+    limits:
+      memory: 1.1G
+  # each secret will be loaded ad an env variable from the `path` secret with the `key` in that secret in the name of `name`
+  secretsMap:
+    "name":
+        path: "test-secret"
+        key: "username"
+    "pwd":
+        path: "test-secret"
+        key: "password"
+  pulsar:
+    pulsarConfig: "test-pulsar"
+    tlsConfig:
+      enabled: false
+      allowInsecure: false
+      hostnameVerification: true
+      certSecretName: sn-platform-tls-broker
+      certSecretKey: ""
+    #authConfig: "test-auth"
+  java:
+    jar: /pulsar/examples/api-examples.jar
+  # to be delete & use admission hook
+  clusterName: test
+  autoAck: true
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-pulsar
+data:
+  webServiceURL: http://sn-platform-pulsar-broker.default.svc.cluster.local:8080
+  brokerServiceURL: pulsar://sn-platform-pulsar-broker.default.svc.cluster.local:6650
+#---
+#apiVersion: v1
+#kind: ConfigMap
+#metadata:
+#  name: test-auth
+#data:
+#  clientAuthenticationPlugin: "abc"
+#  clientAuthenticationParameters: "xyz"
+#  tlsTrustCertsFilePath: "uvw"
+#  useTls: "true"
+#  tlsAllowInsecureConnection: "false"
+#  tlsHostnameVerificationEnable: "true"
+---
+apiVersion: v1
+data:
+  username: YWRtaW4=
+  password: MWYyZDFlMmU2N2Rm
+kind: Secret
+metadata:
+  name: test-secret
+type: Opaque

.ci/tests/integration/cases/global-and-namespaced-env/verify.sh

@@ -0,0 +1,109 @@
+#!/usr/bin/env bash
+#
+# 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.
+#
+
+set -e
+
+E2E_DIR=$(dirname "$0")
+BASE_DIR=$(cd "${E2E_DIR}"/../../../../..;pwd)
+PULSAR_NAMESPACE=${PULSAR_NAMESPACE:-"default"}
+PULSAR_RELEASE_NAME=${PULSAR_RELEASE_NAME:-"sn-platform"}
+E2E_KUBECONFIG=${E2E_KUBECONFIG:-"/tmp/e2e-k8s.config"}
+
+source "${BASE_DIR}"/.ci/helm.sh
+
+if [ ! "$KUBECONFIG" ]; then
+  export KUBECONFIG=${E2E_KUBECONFIG}
+fi
+
+manifests_file="${BASE_DIR}"/.ci/tests/integration/cases/global-and-namespaced-env/manifests.yaml
+env_file="${BASE_DIR}"/.ci/tests/integration/cases/global-and-namespaced-env/env.yaml
+
+kubectl apply -f "${env_file}" > /dev/null 2>&1
+kubectl apply -f "${manifests_file}" > /dev/null 2>&1
+
+verify_fm_result=$(ci::verify_function_mesh function-sample-env 2>&1)
+if [ $? -ne 0 ]; then
+  echo "$verify_fm_result"
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+verify_env_result=$(ci::verify_env "function-sample-env" global1 global1=globalvalue1 2>&1)
+if [ $? -ne 0 ]; then
+  echo "$verify_env_result"
+  kubectl delete -f "${env_file}" > /dev/null 2>&1
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+verify_env_result=$(ci::verify_env "function-sample-env" namespaced1 namespaced1=namespacedvalue1 2>&1)
+if [ $? -ne 0 ]; then
+  echo "$verify_env_result"
+  kubectl delete -f "${env_file}" > /dev/null 2>&1
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+# if global and namespaced env has same key, the value from namespace should be used
+verify_env_result=$(ci::verify_env "function-sample-env" shared1 shared1=fromnamespace 2>&1)
+if [ $? -ne 0 ]; then
+  echo "$verify_env_result"
+  kubectl delete -f "${env_file}" > /dev/null 2>&1
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+
+# delete the namespaced env, the function can start successfully but without namespaced env injected
+kubectl delete -f "${env_file}" > /dev/null 2>&1
+kubectl apply -f "${manifests_file}" > /dev/null 2>&1
+
+verify_fm_result=$(ci::verify_function_mesh function-sample-env 2>&1)
+if [ $? -ne 0 ]; then
+  echo "$verify_fm_result"
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+verify_env_result=$(ci::verify_env "function-sample-env" global1 global1=globalvalue1 2>&1)
+if [ $? -ne 0 ]; then
+  echo "$verify_env_result"
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+verify_env_result=$(ci::verify_env "function-sample-env" shared1 shared1=fromglobal 2>&1)
+if [ $? -ne 0 ]; then
+  echo "$verify_env_result"
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+verify_env_result=$(ci::verify_env "function-sample-env" namespaced1 "" 2>&1)
+if [ $? -eq 0 ]; then
+  echo "e2e-test: ok" | yq eval -
+else
+  echo "$verify_env_result"
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true

.ci/tests/integration/e2e.yaml

@@ -89,14 +89,19 @@ setup:
         bash .ci/upload_function.sh pypip
         bash .ci/upload_function.sh go
 
+    - name: apply global env config map
+      command: |
+        kubectl create ns ${FUNCTION_MESH_NAMESPACE}
+        kubectl create -n ${FUNCTION_MESH_NAMESPACE} -f .ci/clusters/global_env.yaml
+
     - name: install function-mesh operator
       command: |
         make generate
         make helm-crds
         image="function-mesh-operator:latest"
         IMG=${image} make docker-build-skip-test
         kind load docker-image ${image}
-        helm install ${FUNCTION_MESH_RELEASE_NAME} -n ${FUNCTION_MESH_NAMESPACE} --set operatorImage=${image} --create-namespace charts/function-mesh-operator
+        helm install ${FUNCTION_MESH_RELEASE_NAME} -n ${FUNCTION_MESH_NAMESPACE} --set operatorImage=${image} --set controllerManager.globalConfigMap=mesh-global-env --set controllerManager.globalConfigMapNamespace=${FUNCTION_MESH_NAMESPACE} --set controllerManager.namespacedConfigMap=mesh-namespaced-env --create-namespace charts/function-mesh-operator
       wait:
         - namespace: function-mesh
           resource: pod
@@ -168,3 +173,5 @@ verify:
       expected: expected.data.yaml
     - query: timeout 5m bash .ci/tests/integration/cases/python-log-format-json/verify.sh
       expected: expected.data.yaml
+    - query: timeout 5m bash .ci/tests/integration/cases/global-and-namespaced-env/verify.sh
+      expected: expected.data.yaml

charts/function-mesh-operator/templates/controller-manager-deployment.yaml

@@ -61,6 +61,9 @@ spec:
           - --pprof-addr=:{{ .Values.controllerManager.pprof.port }}
           - --config-file={{ .Values.controllerManager.configFile }}
           - --enable-init-containers={{ .Values.controllerManager.enableInitContainers }}
+          - --global-config-map={{ .Values.controllerManager.globalConfigMap }}
+          - --global-config-map-namespace={{ .Values.controllerManager.globalConfigMapNamespace }}
+          - --namespaced-config-map={{ .Values.controllerManager.namespacedConfigMap }}
         env:
           - name: NAMESPACE
             valueFrom:

controllers/function.go

@@ -69,7 +69,7 @@ func (r *FunctionReconciler) ObserveFunctionStatefulSet(ctx context.Context, fun
 	}
 	function.Status.Selector = selector.String()
 
-	if r.checkIfStatefulSetNeedUpdate(statefulSet, function) {
+	if r.checkIfStatefulSetNeedUpdate(ctx, statefulSet, function) {
 		condition.Status = metav1.ConditionFalse
 		condition.Action = v1alpha1.Update
 		function.Status.Conditions[v1alpha1.StatefulSet] = condition
@@ -93,7 +93,7 @@ func (r *FunctionReconciler) ApplyFunctionStatefulSet(ctx context.Context, funct
 	if condition.Status == metav1.ConditionTrue && !newGeneration {
 		return nil
 	}
-	desiredStatefulSet := spec.MakeFunctionStatefulSet(function)
+	desiredStatefulSet := spec.MakeFunctionStatefulSet(ctx, r, function)
 	keepStatefulSetUnchangeableFields(ctx, r, r.Log, desiredStatefulSet)
 	desiredStatefulSetSpec := desiredStatefulSet.Spec
 	if _, err := ctrl.CreateOrUpdate(ctx, r.Client, desiredStatefulSet, func() error {
@@ -403,9 +403,9 @@ func (r *FunctionReconciler) ApplyFunctionCleanUpJob(ctx context.Context, functi
 	return nil
 }
 
-func (r *FunctionReconciler) checkIfStatefulSetNeedUpdate(statefulSet *appsv1.StatefulSet,
+func (r *FunctionReconciler) checkIfStatefulSetNeedUpdate(ctx context.Context, statefulSet *appsv1.StatefulSet,
 	function *v1alpha1.Function) bool {
-	return !spec.CheckIfStatefulSetSpecIsEqual(&statefulSet.Spec, &spec.MakeFunctionStatefulSet(function).Spec)
+	return !spec.CheckIfStatefulSetSpecIsEqual(&statefulSet.Spec, &spec.MakeFunctionStatefulSet(ctx, r, function).Spec)
 }
 
 func (r *FunctionReconciler) checkIfHPANeedUpdate(hpa *autov2.HorizontalPodAutoscaler,

controllers/sink.go

@@ -69,7 +69,7 @@ func (r *SinkReconciler) ObserveSinkStatefulSet(ctx context.Context, sink *v1alp
 	}
 	sink.Status.Selector = selector.String()
 
-	if r.checkIfStatefulSetNeedUpdate(statefulSet, sink) {
+	if r.checkIfStatefulSetNeedUpdate(ctx, statefulSet, sink) {
 		condition.Status = metav1.ConditionFalse
 		condition.Action = v1alpha1.Update
 		sink.Status.Conditions[v1alpha1.StatefulSet] = condition
@@ -92,7 +92,7 @@ func (r *SinkReconciler) ApplySinkStatefulSet(ctx context.Context, sink *v1alpha
 	if condition.Status == metav1.ConditionTrue && !newGeneration {
 		return nil
 	}
-	desiredStatefulSet := spec.MakeSinkStatefulSet(sink)
+	desiredStatefulSet := spec.MakeSinkStatefulSet(ctx, r, sink)
 	keepStatefulSetUnchangeableFields(ctx, r, r.Log, desiredStatefulSet)
 	desiredStatefulSetSpec := desiredStatefulSet.Spec
 	if _, err := ctrl.CreateOrUpdate(ctx, r.Client, desiredStatefulSet, func() error {
@@ -400,8 +400,8 @@ func (r *SinkReconciler) ApplySinkCleanUpJob(ctx context.Context, sink *v1alpha1
 	return nil
 }
 
-func (r *SinkReconciler) checkIfStatefulSetNeedUpdate(statefulSet *appsv1.StatefulSet, sink *v1alpha1.Sink) bool {
-	return !spec.CheckIfStatefulSetSpecIsEqual(&statefulSet.Spec, &spec.MakeSinkStatefulSet(sink).Spec)
+func (r *SinkReconciler) checkIfStatefulSetNeedUpdate(ctx context.Context, statefulSet *appsv1.StatefulSet, sink *v1alpha1.Sink) bool {
+	return !spec.CheckIfStatefulSetSpecIsEqual(&statefulSet.Spec, &spec.MakeSinkStatefulSet(ctx, r, sink).Spec)
 }
 
 func (r *SinkReconciler) checkIfHPANeedUpdate(hpa *autov2.HorizontalPodAutoscaler, sink *v1alpha1.Sink) bool {

controllers/sink_controller_test.go

@@ -41,7 +41,7 @@ var _ = Describe("Sink Controller", func() {
 		if sink.Status.Conditions == nil {
 			sink.Status.Conditions = make(map[v1alpha1.Component]v1alpha1.ResourceCondition)
 		}
-		statefulSet := spec.MakeSinkStatefulSet(sink)
+		statefulSet := spec.MakeSinkStatefulSet(context.Background(), k8sClient, sink)
 
 		It("Should create pulsar configmap successfully", func() {
 			Expect(k8sClient.Create(context.Background(), pulsarConfig)).Should(Succeed())

controllers/source.go

@@ -69,7 +69,7 @@ func (r *SourceReconciler) ObserveSourceStatefulSet(ctx context.Context, source
 	}
 	source.Status.Selector = selector.String()
 
-	if r.checkIfStatefulSetNeedUpdate(statefulSet, source) {
+	if r.checkIfStatefulSetNeedUpdate(ctx, statefulSet, source) {
 		condition.Status = metav1.ConditionFalse
 		condition.Action = v1alpha1.Update
 		source.Status.Conditions[v1alpha1.StatefulSet] = condition
@@ -93,7 +93,7 @@ func (r *SourceReconciler) ApplySourceStatefulSet(ctx context.Context, source *v
 	if condition.Status == metav1.ConditionTrue && !newGeneration {
 		return nil
 	}
-	desiredStatefulSet := spec.MakeSourceStatefulSet(source)
+	desiredStatefulSet := spec.MakeSourceStatefulSet(ctx, r, source)
 	keepStatefulSetUnchangeableFields(ctx, r, r.Log, desiredStatefulSet)
 	desiredStatefulSetSpec := desiredStatefulSet.Spec
 	if _, err := ctrl.CreateOrUpdate(ctx, r.Client, desiredStatefulSet, func() error {
@@ -402,8 +402,8 @@ func (r *SourceReconciler) ApplySourceCleanUpJob(ctx context.Context, source *v1
 	return nil
 }
 
-func (r *SourceReconciler) checkIfStatefulSetNeedUpdate(statefulSet *appsv1.StatefulSet, source *v1alpha1.Source) bool {
-	return !spec.CheckIfStatefulSetSpecIsEqual(&statefulSet.Spec, &spec.MakeSourceStatefulSet(source).Spec)
+func (r *SourceReconciler) checkIfStatefulSetNeedUpdate(ctx context.Context, statefulSet *appsv1.StatefulSet, source *v1alpha1.Source) bool {
+	return !spec.CheckIfStatefulSetSpecIsEqual(&statefulSet.Spec, &spec.MakeSourceStatefulSet(ctx, r, source).Spec)
 }
 
 func (r *SourceReconciler) checkIfHPANeedUpdate(hpa *autov2.HorizontalPodAutoscaler, source *v1alpha1.Source) bool {

controllers/source_controller_test.go

@@ -40,7 +40,7 @@ var _ = Describe("Source Controller", func() {
 		if source.Status.Conditions == nil {
 			source.Status.Conditions = make(map[v1alpha1.Component]v1alpha1.ResourceCondition)
 		}
-		statefulSet := spec.MakeSourceStatefulSet(source)
+		statefulSet := spec.MakeSourceStatefulSet(context.Background(), k8sClient, source)
 
 		It("Should create pulsar configmap successfully", func() {
 			Expect(k8sClient.Create(context.Background(), pulsarConfig)).Should(Succeed())