Support special resource unit for VPA

.ci/helm.sh

@@ -239,7 +239,7 @@ function ci::verify_download_java_function_generic_auth() {
 }
 
 function ci::verify_vpa_java_function() {
-  kubectl wait -l name=function-sample-vpa-function --for=condition=RecommendationProvided --timeout=2m vpa && true
+  kubectl wait -l name=function-sample-vpa --for=condition=RecommendationProvided --timeout=2m vpa
   cpu=`kubectl get vpa function-sample-vpa-function -o jsonpath='{.status.recommendation.containerRecommendations[0].target.cpu}'`
   memory=`kubectl get vpa function-sample-vpa-function -o jsonpath='{.status.recommendation.containerRecommendations[0].target.memory}'`
   resources='{"limits":{"cpu":"'$cpu'","memory":"'$memory'"},"requests":{"cpu":"'$cpu'","memory":"'$memory'"}}'
@@ -269,6 +269,55 @@ function ci::verify_vpa_java_function() {
   ci::verify_exclamation_function "persistent://public/default/input-vpa-java-topic" "persistent://public/default/output-vpa-java-topic" "test-message" "test-message!" 10
 }
 
+function ci::calculate_multiplies() {
+  base=$1
+  value=$2
+  quotient=$(($value / $base))
+  remainder=$(($value % $base))
+
+  # If there's any remainder, we need to round up
+  if [ $remainder -ne 0 ]; then
+      multiple=$(($quotient + 1))
+  else
+      multiple=$quotient
+  fi
+  echo $multiple
+}
+
+function ci::verify_vpa_with_resource_unit() {
+  name=$1
+  kind=$2
+  baseCpu=200 # cpu value of the resource unit
+  baseMemory=`echo $((800*1024*1024))` # memory value of the resource unit
+  vpaName="$1-$2"
+  kubectl wait -l name=$name --for=condition=RecommendationProvided --timeout=2m vpa
+  cpu=`kubectl get vpa $vpaName -o jsonpath='{.status.recommendation.containerRecommendations[0].target.cpu}'`
+  cpu_value=${cpu%m}
+  multiplier=$(ci::calculate_multiplies $baseCpu $cpu_value)
+
+  memory=`kubectl get vpa $vpaName -o jsonpath='{.status.recommendation.containerRecommendations[0].target.memory}'`
+  memory_value=${memory%k}
+  memory_value=$((memory_value*1024))
+  memoryMultiplier=$(ci::calculate_multiplies $baseMemory $memory_value)
+
+  if [ $memoryMultiplier -gt $multiplier ]; then
+    multiplier=$memoryMultiplier
+  fi
+
+  targetCpu=`echo $(($baseCpu* $multiple))m`
+  targetMemory=`echo $(($baseMemory * $multiple))`
+
+  resources='{"limits":{"cpu":"'$targetCpu'","memory":"'$targetMemory'"},"requests":{"cpu":"'$targetCpu'","memory":"'$targetMemory'"}}'
+
+  realResource=`kubectl get $kind $name -o jsonpath='{.spec.resources}'`
+  if [[ "$resources" != "$realResource" ]]; then
+    echo "vpa tests failed for $kind"
+    echo "recommend resource is: ${resources}, actual resource is ${realResource}"
+    exit 1
+  fi
+  echo "vpa tests passed"
+}
+
 function ci::verify_python_function() {
     ci::verify_exclamation_function "persistent://public/default/input-python-topic" "persistent://public/default/output-python-topic" "test-message" "test-message!" 10
 }
@@ -349,7 +398,8 @@ function ci::verify_exclamation_function() {
     timesleep=$5
     kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-client produce -m "${inputmessage}" -n 1 "${inputtopic}"
     sleep "$timesleep"
-    MESSAGE=$(kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-client consume -n 1 -s "sub" --subscription-position Earliest "${outputtopic}")
+    sub=`cat /dev/urandom | tr -dc 'a-z' | head -c 8`
+    MESSAGE=$(kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-client consume -n 1 -s "${sub}" --subscription-position Earliest "${outputtopic}")
     echo "$MESSAGE"
     if [[ "$MESSAGE" == *"$outputmessage"* ]]; then
         return 0
@@ -366,7 +416,8 @@ function ci::verify_exclamation_function_with_auth() {
     command="kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- sh -c 'bin/pulsar-client --auth-plugin \$brokerClientAuthenticationPlugin --auth-params \$brokerClientAuthenticationParameters produce -m \"${inputmessage}\" -n 1 \"${inputtopic}\"'"
     sh -c "$command"
     sleep "$timesleep"
-    consumeCommand="kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- sh -c 'bin/pulsar-client --auth-plugin \$brokerClientAuthenticationPlugin --auth-params \$brokerClientAuthenticationParameters consume -n 1 -s "sub" --subscription-position Earliest \"${outputtopic}\"'"
+    sub=`cat /dev/urandom | tr -dc 'a-z' | head -c 8`
+    consumeCommand="kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- sh -c 'bin/pulsar-client --auth-plugin \$brokerClientAuthenticationPlugin --auth-params \$brokerClientAuthenticationParameters consume -n 1 -s \"${sub}\" --subscription-position Earliest \"${outputtopic}\"'"
     MESSAGE=$(sh -c "$consumeCommand")
     echo "$MESSAGE"
     if [[ "$MESSAGE" == *"$outputmessage"* ]]; then
@@ -383,7 +434,8 @@ function ci::verify_wordcount_function() {
     timesleep=$5
     kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-client produce -m "${inputmessage}" -n 1 "${inputtopic}"
     sleep "$timesleep"
-    MESSAGE=$(kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-client consume -n 3 -s "sub" --subscription-position Earliest "${outputtopic}")
+    sub=`cat /dev/urandom | tr -dc 'a-z' | head -c 8`
+    MESSAGE=$(kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-client consume -n 3 -s "${sub}" --subscription-position Earliest "${outputtopic}")
     echo "$MESSAGE"
     if [[ "$MESSAGE" == *"$outputmessage"* ]]; then
         return 0
@@ -453,12 +505,13 @@ function ci::verify_elasticsearch_sink() {
 
 function ci::verify_mongodb_source() {
     timesleep=$1
-    kubectl exec mongo-dbz-0 -c mongo -- mongo -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory --eval 'db.products.update({"_id":NumberLong(104)},{$set:{weight:1.25}})'
+    kubectl exec mongo-dbz-0 -c mongo -- mongosh -u debezium -p dbz --authenticationDatabase admin localhost:27017/inventory --eval 'db.products.update({"_id":NumberLong(104)},{$set:{weight:1.25}})'
     sleep "$timesleep"
     kubectl logs --tail=-1 -l compute.functionmesh.io/name=source-sample | grep "records sent"
     if [ $? -eq 0 ]; then
         return 0
     fi
+    kubectl logs --tail=-1 -l compute.functionmesh.io/name=source-sample
     return 1
 }
 
@@ -484,7 +537,8 @@ function ci::verify_function_with_encryption() {
 
     kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-client produce -ekn "myapp1" -ekv "data:application/x-pem-file;base64,${correct_pubkey}" -m "${inputmessage}" -n 1 "${inputtopic}"
     sleep "$timesleep"
-    MESSAGE=$(kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-client consume -n 1 -s "sub" --subscription-position Earliest "${outputtopic}")
+    sub=`cat /dev/urandom | tr -dc 'a-z' | head -c 8`
+    MESSAGE=$(kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-client consume -n 1 -s "${sub}" --subscription-position Earliest "${outputtopic}")
     echo "$MESSAGE"
     if [[ "$MESSAGE" == *"$outputmessage"* ]]; then
         return 0
@@ -573,7 +627,8 @@ function ci::verify_log_topic() {
   timesleep=$3
 
   sleep "$timesleep"
-  MESSAGE=$(kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-client consume -n 1 -s "sub" --subscription-position Earliest "${logTopic}")
+  sub=`cat /dev/urandom | tr -dc 'a-z' | head -c 8`
+  MESSAGE=$(kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-client consume -n 1 -s "${sub}" --subscription-position Earliest "${logTopic}")
   echo "$MESSAGE"
   if [[ "$MESSAGE" == *"$message"* ]]; then
     return 0
@@ -588,7 +643,8 @@ function ci::verify_log_topic_with_auth() {
   timesleep=$3
 
   sleep "$timesleep"
-  consumeCommand="kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- sh -c 'bin/pulsar-client --auth-plugin \$brokerClientAuthenticationPlugin --auth-params \$brokerClientAuthenticationParameters consume -n 1 -s "sub" --subscription-position Earliest \"${logTopic}\"'"
+  sub=`cat /dev/urandom | tr -dc 'a-z' | head -c 8`
+  consumeCommand="kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- sh -c 'bin/pulsar-client --auth-plugin \$brokerClientAuthenticationPlugin --auth-params \$brokerClientAuthenticationParameters consume -n 1 -s \"${sub}\" --subscription-position Earliest \"${logTopic}\"'"
   MESSAGE=$(sh -c "$consumeCommand")
   echo "$MESSAGE"
   if [[ "$MESSAGE" == *"$message"* ]]; then

.ci/tests/integration/cases/mongodb-source/manifests.yaml

@@ -23,7 +23,7 @@ spec:
       cpu: "0.1"
       memory: 1G
   sourceConfig:
-    mongodb.hosts: rs0/mongo-dbz-0.mongo.default.svc.cluster.local:27017,rs0/mongo-dbz-1.mongo.default.svc.cluster.local:27017,rs0/mongo-dbz-2.mongo.default.svc.cluster.local:27017
+    mongodb.hosts: rs0/mongo-dbz-0.mongo.default.svc.cluster.local:27017
     mongodb.name: dbserver1
     mongodb.user: debezium
     mongodb.password: dbz

.ci/tests/integration/cases/mongodb-source/mongodb-dbz.yaml

@@ -35,7 +35,7 @@ spec:
     matchLabels:
       role: mongo
   serviceName: "mongo"
-  replicas: 3
+  replicas: 1
   template:
     metadata:
       labels:
@@ -44,7 +44,7 @@ spec:
       terminationGracePeriodSeconds: 10
       containers:
         - name: mongo
-          image: debezium/example-mongodb:0.10
+          image: debezium/example-mongodb:2.6
           env:
             - name: MONGODB_USER
               value: "debezium"

.ci/tests/integration/cases/mongodb-source/verify.sh

@@ -45,7 +45,7 @@ function install_mongodb_server() {
     # install mongodb server
     kubectl apply -f "${mongodb_file}"
     num=0
-    while [[ ${num} -lt 3 ]]; do
+    while [[ ${num} -lt 1 ]]; do
         sleep 5
         kubectl get pods
         num=$(kubectl get pods -l role=mongo | wc -l)

.ci/tests/integration/cases/vpa-with-resource-unit-sink/manifests.yaml

@@ -0,0 +1,57 @@
+apiVersion: compute.functionmesh.io/v1alpha1
+kind: Sink
+metadata:
+  name: vpa-sink-sample
+spec:
+  className: org.apache.pulsar.io.datagenerator.DataGeneratorPrintSink
+  replicas: 1
+  sinkConfig:
+    test: test
+  input:
+    topics:
+      - persistent://public/default/input-vpa-sink-topic
+    typeClassName: org.apache.pulsar.io.datagenerator.Person
+  forwardSourceMessageProperty: true
+  resources:
+    limits:
+      cpu: "0.2"
+      memory: 1.1G
+    requests:
+      cpu: "0.1"
+      memory: 1G
+  pod:
+    vpa:
+      resourceUnit:
+        cpu: 200m
+        memory: 800Mi
+      updatePolicy:
+        updateMode: "Auto"
+      resourcePolicy:
+        containerPolicies:
+          - containerName: "*"
+            minAllowed:
+              cpu: 200m
+              memory: 100Mi
+            maxAllowed:
+              cpu: 1
+              memory: 1000Mi
+  pulsar:
+    pulsarConfig: "test-sink"
+    tlsConfig:
+      enabled: false
+      allowInsecure: false
+      hostnameVerification: true
+      certSecretName: sn-platform-tls-broker
+      certSecretKey: ""
+  image: streamnative/pulsar-io-data-generator:3.2.2.1
+  java:
+    jar: connectors/pulsar-io-data-generator-3.2.2.1.nar
+  clusterName: test-pulsar
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-sink
+data:
+  webServiceURL: http://sn-platform-pulsar-broker.default.svc.cluster.local:8080
+  brokerServiceURL: pulsar://sn-platform-pulsar-broker.default.svc.cluster.local:6650

.ci/tests/integration/cases/vpa-with-resource-unit-sink/verify.sh

@@ -0,0 +1,50 @@
+#!/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
+
+kubectl apply -f "${BASE_DIR}"/.ci/tests/integration/cases/vpa-with-resource-unit-sink/manifests.yaml > /dev/null 2>&1
+
+verify_fm_result=$(ci::verify_function_mesh vpa-sink-sample 2>&1)
+if [ $? -ne 0 ]; then
+  echo "$verify_fm_result"
+  kubectl delete -f "${BASE_DIR}"/.ci/tests/integration/cases/vpa-with-resource-unit-sink/manifests.yaml > /dev/null 2>&1 || true
+  exit 1
+fi
+
+verify_java_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} ci::verify_vpa_with_resource_unit vpa-sink-sample sink 2>&1)
+if [ $? -eq 0 ]; then
+  echo "e2e-test: ok" | yq eval -
+else
+  echo "$verify_java_result"
+fi
+kubectl delete -f "${BASE_DIR}"/.ci/tests/integration/cases/vpa-with-resource-unit-sink/manifests.yaml > /dev/null 2>&1 || true

.ci/tests/integration/cases/vpa-with-resource-unit-source/manifests.yaml

@@ -0,0 +1,60 @@
+apiVersion: compute.functionmesh.io/v1alpha1
+kind: Source
+metadata:
+  name: vpa-source-sample
+spec:
+  className: org.apache.pulsar.io.datagenerator.DataGeneratorSource
+  replicas: 1
+  output:
+    producerConf:
+      maxPendingMessages: 1000
+      maxPendingMessagesAcrossPartitions: 50000
+      useThreadLocalProducers: true
+    topic: persistent://public/default/output-vpa-source-topic
+    typeClassName: org.apache.pulsar.io.datagenerator.Person
+  forwardSourceMessageProperty: true
+  sourceConfig:
+    test: test
+  resources:
+    limits:
+      cpu: "0.2"
+      memory: 1.1G
+    requests:
+      cpu: "0.1"
+      memory: 1G
+  pod:
+    vpa:
+      resourceUnit:
+        cpu: 200m
+        memory: 800Mi
+      updatePolicy:
+        updateMode: "Auto"
+      resourcePolicy:
+        containerPolicies:
+          - containerName: "*"
+            minAllowed:
+              cpu: 200m
+              memory: 100Mi
+            maxAllowed:
+              cpu: 1
+              memory: 1000Mi
+  pulsar:
+    pulsarConfig: "test-source"
+    tlsConfig:
+      enabled: false
+      allowInsecure: false
+      hostnameVerification: true
+      certSecretName: sn-platform-tls-broker
+      certSecretKey: ""
+  image: streamnative/pulsar-io-data-generator:3.2.2.1
+  java:
+    jar: connectors/pulsar-io-data-generator-3.2.2.1.nar
+  clusterName: test-pulsar
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: test-source
+data:
+  webServiceURL: http://sn-platform-pulsar-broker.default.svc.cluster.local:8080
+  brokerServiceURL: pulsar://sn-platform-pulsar-broker.default.svc.cluster.local:6650

.ci/tests/integration/cases/vpa-with-resource-unit-source/verify.sh

@@ -0,0 +1,50 @@
+#!/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
+
+kubectl apply -f "${BASE_DIR}"/.ci/tests/integration/cases/vpa-with-resource-unit-source/manifests.yaml > /dev/null 2>&1
+
+verify_fm_result=$(ci::verify_function_mesh vpa-source-sample 2>&1)
+if [ $? -ne 0 ]; then
+  echo "$verify_fm_result"
+  kubectl delete -f "${BASE_DIR}"/.ci/tests/integration/cases/vpa-with-resource-unit-source/manifests.yaml > /dev/null 2>&1 || true
+  exit 1
+fi
+
+verify_java_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} ci::verify_vpa_with_resource_unit vpa-source-sample source 2>&1)
+if [ $? -eq 0 ]; then
+  echo "e2e-test: ok" | yq eval -
+else
+  echo "$verify_java_result"
+fi
+kubectl delete -f "${BASE_DIR}"/.ci/tests/integration/cases/vpa-with-resource-unit-source/manifests.yaml > /dev/null 2>&1 || true