refine test cases of the Sink/Source/Crypto function

.ci/helm.sh

@@ -137,7 +137,7 @@ function ci::verify_function_mesh() {
         sleep 5s
         kubectl get pods -l name="${FUNCTION_NAME}"
         kubectl logs -l name="${FUNCTION_NAME}" --all-containers=true --tail=50 || true
-        num=$(kubectl logs -lname="${FUNCTION_NAME}" --all-containers=true --tail=-1 | grep "Created producer\|Created consumer" | wc -l)
+        num=$(kubectl logs -lname="${FUNCTION_NAME}" --all-containers=true --tail=-1 | grep "Created producer\|Created consumer\|Subscribed to topic" | wc -l)
     done
 }
 
@@ -206,46 +206,58 @@ function ci::test_function_runners() {
 }
 
 function ci::verify_go_function() {
-    ci:verify_exclamation_function "persistent://public/default/input-go-topic" "persistent://public/default/output-go-topic" "test-message" "test-message!" 10
+    ci::verify_exclamation_function "persistent://public/default/input-go-topic" "persistent://public/default/output-go-topic" "test-message" "test-message!" 10
 }
 
 function ci::verify_download_go_function() {
-    ci:verify_exclamation_function "persistent://public/default/input-download-go-topic" "persistent://public/default/output-download-go-topic" "test-message" "test-message!" 10
+    ci::verify_exclamation_function "persistent://public/default/input-download-go-topic" "persistent://public/default/output-download-go-topic" "test-message" "test-message!" 10
 }
 
 function ci::verify_java_function() {
-    ci:verify_exclamation_function "persistent://public/default/input-java-topic" "persistent://public/default/output-java-topic" "test-message" "test-message!" 10
+    ci::verify_exclamation_function "persistent://public/default/input-java-topic" "persistent://public/default/output-java-topic" "test-message" "test-message!" 10
 }
 
 function ci::verify_download_java_function() {
-    ci:verify_exclamation_function "persistent://public/default/input-download-java-topic" "persistent://public/default/output-download-java-topic" "test-message" "test-message!" 10
+    ci::verify_exclamation_function "persistent://public/default/input-download-java-topic" "persistent://public/default/output-download-java-topic" "test-message" "test-message!" 10
 }
 
 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
+    ci::verify_exclamation_function "persistent://public/default/input-python-topic" "persistent://public/default/output-python-topic" "test-message" "test-message!" 10
 }
 
 function ci::verify_download_python_function() {
-    ci:verify_exclamation_function "persistent://public/default/input-download-python-topic" "persistent://public/default/output-download-python-topic" "test-message" "test-message!" 10
+    ci::verify_exclamation_function "persistent://public/default/input-download-python-topic" "persistent://public/default/output-download-python-topic" "test-message" "test-message!" 10
 }
 
 function ci::verify_download_python_zip_function() {
-    ci:verify_exclamation_function "persistent://public/default/input-download-python-zip-topic" "persistent://public/default/output-download-python-zip-topic" "test-message" "test-message!" 10
+    ci::verify_exclamation_function "persistent://public/default/input-download-python-zip-topic" "persistent://public/default/output-download-python-zip-topic" "test-message" "test-message!" 10
 }
 
 function ci::verify_download_python_pip_function() {
-    ci:verify_exclamation_function "persistent://public/default/input-download-python-pip-topic" "persistent://public/default/output-download-python-pip-topic" "test-message" "test-message!" 10
+    ci::verify_exclamation_function "persistent://public/default/input-download-python-pip-topic" "persistent://public/default/output-download-python-pip-topic" "test-message" "test-message!" 10
 }
 
 function ci::verify_stateful_function() {
-    ci:verify_wordcount_function "persistent://public/default/python-function-stateful-input-topic" "persistent://public/default/logging-stateful-function-logs" "apple apple apple" "The value is 3" 10
+    ci::verify_wordcount_function "persistent://public/default/python-function-stateful-input-topic" "persistent://public/default/logging-stateful-function-logs" "apple apple apple" "The value is 3" 10
 }
 
 function ci::verify_mesh_function() {
-    ci:verify_exclamation_function "persistent://public/default/functionmesh-input-topic" "persistent://public/default/functionmesh-python-topic" "test-message" "test-message!!!" 10
+    ci::verify_exclamation_function "persistent://public/default/functionmesh-input-topic" "persistent://public/default/functionmesh-python-topic" "test-message" "test-message!!!" 10
 }
 
-function ci:verify_exclamation_function() {
+function ci::verify_sink() {
+    ci::verify_elasticsearch_sink "persistent://public/default/input-sink-topic" "{\"a\":1}" 10
+}
+
+function ci::verify_source() {
+    ci::verify_mongodb_source 30
+}
+
+function ci::verify_crypto_function() {
+    ci::verify_function_with_encryption "persistent://public/default/java-function-crypto-input-topic" "persistent://public/default/java-function-crypto-output-topic" "test-message" "test-message!" 10
+}
+
+function ci::verify_exclamation_function() {
     inputtopic=$1
     outputtopic=$2
     inputmessage=$3
@@ -261,7 +273,7 @@ function ci:verify_exclamation_function() {
     return 1
 }
 
-function ci:verify_wordcount_function() {
+function ci::verify_wordcount_function() {
     inputtopic=$1
     outputtopic=$2
     inputmessage=$3
@@ -323,3 +335,57 @@ function ci::verify_functionmesh_reconciliation() {
         fi
     done
 }
+
+function ci::verify_elasticsearch_sink() {
+    inputtopic=$1
+    inputmessage=$2
+    timesleep=$3
+    kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-client produce -m "${inputmessage}" -n 1 "${inputtopic}"
+    sleep "$timesleep"
+    kubectl logs --all-containers=true --tail=-1 quickstart-es-default-0 | grep "creating index"
+    if [ $? -eq 0 ]; then
+        return 0
+    fi
+    return 1
+}
+
+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}})'
+    sleep "$timesleep"
+    kubectl logs --all-containers=true --tail=-1 -l name=source-sample | grep "records sent"
+    if [ $? -eq 0 ]; then
+        return 0
+    fi
+    return 1
+}
+
+function ci::verify_function_with_encryption() {
+    inputtopic=$1
+    outputtopic=$2
+    inputmessage=$3
+    outputmessage=$4
+    timesleep=$5
+
+    # correct pubkey
+    correct_pubkey="LS0tLS1CRUdJTiBQVUJMSUMgS0VZLS0tLS0KTUZZd0VBWUhLb1pJemowQ0FRWUZLNEVFQUFvRFFnQUUvZ0cxbko0SHBHVnB0WWR2YjRUWUVCUVRpS3kwSmF1TApqa0FXalpqTE5WVW5JaEtCUkttV1M3cjA1MWU1VHRwdFRvOWZEVDR3L29zMmVTTUhpWVl5dEE9PQotLS0tLUVORCBQVUJMSUMgS0VZLS0tLS0K"
+    # incorrect pubkey
+    incorrect_pubkey="LS0tLS1CRUdJTiBQVUJMSUMgS0VZLS0tLS0KTUlJQ1hUQ0NBZEFHQnlxR1NNNDlBZ0V3Z2dIREFnRUJNRTBHQnlxR1NNNDlBUUVDUWdILy8vLy8vLy8vLy8vLwovLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vCi8vLy8vLy8vL3pDQm53UkNBZi8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8KLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vOEJFSUFVWlUrdVdHT0hKb2ZrcG9ob0xhRgpRTzZpMm5KYm1iTVY4N2kwaVpHTzhRbmhWaGs1VWV4K2szc1dVc0M5TzdHL0J6VnozNGc5TERUeDcwVWYxR3RRClB3QURGUURRbm9nQUtSeTRVNWJNWnhjNU1vU3FvTnBrdWdTQmhRUUF4b1dPQnJjRUJPbk5uajdMWmlPVnRFS2MKWklFNUJUKzFJZmdvcjJCclRUMjZvVXRlZCsvbldTaitIY0Vub3YrbzNqTklzOEdGYWtLYitYNStNY0xsdldZQgpHRGtwYW5pYU84QUVYSXBmdEN4OUc5bVk5VVJKVjV0RWFCZXZ2UmNuUG1Zc2wrNXltVjcwSmtERlVMa0JQNjBICllUVThjSWFpY3NKQWlMNlVkcC9SWmxBQ1FnSC8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8vLy8KLy8vLy8vcFJob2VEdnkrV2EzL01BVWozQ2FYUU83WEp1SW1jUjY2N2I3Y2VrVGhrQ1FJQkFRT0JoZ0FFQWF3QwpXb2NQMTBndWJsb0hkYnJDNnVlSEpzM1VDNVRvbUZWanlCdWIvZHBjRVZ3VGZpOW54R1Jsa3lPZkZvM0NTZnVWCjVidE5wVXZveXhSMG1VZ0FDTWZrQWVzS2JkSkdyWHR5Tk1VTHVKeWtYNDBoSllmNDZRbzc2VlV1UUFCaXRrei8KMjhWNHlTbGcvZHZUbmVkMkIydUtyNHUrUjZzbHVNbWJYYi8xZ0lkUENZcDIKLS0tLS1FTkQgUFVCTElDIEtFWS0tLS0tCg=="
+
+    # incorrect pubkey test
+    kubectl exec -n ${NAMESPACE} ${CLUSTER}-pulsar-broker-0 -- bin/pulsar-client produce -ekn "myapp1" -ekv "data:application/x-pem-file;base64,${incorrect_pubkey}" -m "${inputmessage}" -n 1 "${inputtopic}"
+    sleep "$timesleep"
+    kubectl logs --all-containers=true --tail=-1 -l name=java-function-crypto-sample | grep "Message delivery failed since unable to decrypt incoming message"
+    if [ $? -ne 0 ]; then
+        return 1
+    fi
+
+    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}")
+    echo "$MESSAGE"
+    if [[ "$MESSAGE" == *"$outputmessage"* ]]; then
+        return 0
+    fi
+    return 1
+}

.ci/tests/integration/cases/crypto-function/manifests.yaml

@@ -0,0 +1,85 @@
+apiVersion: compute.functionmesh.io/v1alpha1
+kind: Function
+metadata:
+  name: java-function-crypto-sample
+  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/java-function-crypto-input-topic
+    typeClassName: java.lang.String
+    sourceSpecs:
+      "persistent://public/default/java-function-crypto-input-topic":
+        cryptoConfig:
+          cryptoKeyReaderClassName: "org.apache.pulsar.functions.api.examples.RawFileKeyReader"
+          cryptoKeyReaderConfig:
+            PUBLIC: "/keys/pubkey/test_ecdsa_pubkey.pem"
+            PRIVATE: "/keys/privkey/test_ecdsa_privkey.pem"
+          producerCryptoFailureAction: "FAIL"
+          encryptionKeys:
+            - "myapp1"
+          cryptoSecrets:
+            - secretName: "java-function-crypto-sample-crypto-secret"
+              secretKey: "test_ecdsa_privkey.pem"
+              asVolume: "/keys/privkey"
+            - secretName: "java-function-crypto-sample-crypto-secret"
+              secretKey: "test_ecdsa_pubkey.pem"
+              asVolume: "/keys/pubkey"
+  output:
+    topic: persistent://public/default/java-function-crypto-output-topic
+    typeClassName: java.lang.String
+  resources:
+    requests:
+      cpu: "0.1"
+      memory: 1G
+    limits:
+      cpu: "0.2"
+      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"
+  java:
+    jar: /pulsar/examples/api-examples.jar
+  # to be delete & use admission hook
+  clusterName: test-pulsar
+  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
+data:
+  username: YWRtaW4=
+  password: MWYyZDFlMmU2N2Rm
+kind: Secret
+metadata:
+  name: test-secret
+type: Opaque
+---
+apiVersion: v1
+data:
+  "test_ecdsa_privkey.pem": LS0tLS1CRUdJTiBFQyBQQVJBTUVURVJTLS0tLS0KQmdVcmdRUUFDZz09Ci0tLS0tRU5EIEVDIFBBUkFNRVRFUlMtLS0tLQotLS0tLUJFR0lOIEVDIFBSSVZBVEUgS0VZLS0tLS0KTUhRQ0FRRUVJRnpKSlpIRTEvMkpPR0VPejFSLzA2NEEzd0hBZmJxTFMvMzB5SW53UTBKVm9BY0dCU3VCQkFBSwpvVVFEUWdBRS9nRzFuSjRIcEdWcHRZZHZiNFRZRUJRVGlLeTBKYXVMamtBV2paakxOVlVuSWhLQlJLbVdTN3IwCjUxZTVUdHB0VG85ZkRUNHcvb3MyZVNNSGlZWXl0QT09Ci0tLS0tRU5EIEVDIFBSSVZBVEUgS0VZLS0tLS0K
+  "test_ecdsa_pubkey.pem": LS0tLS1CRUdJTiBQVUJMSUMgS0VZLS0tLS0KTUZZd0VBWUhLb1pJemowQ0FRWUZLNEVFQUFvRFFnQUUvZ0cxbko0SHBHVnB0WWR2YjRUWUVCUVRpS3kwSmF1TApqa0FXalpqTE5WVW5JaEtCUkttV1M3cjA1MWU1VHRwdFRvOWZEVDR3L29zMmVTTUhpWVl5dEE9PQotLS0tLUVORCBQVUJMSUMgS0VZLS0tLS0K
+kind: Secret
+metadata:
+  name: java-function-crypto-sample-crypto-secret
+type: Opaque

.ci/tests/integration/cases/crypto-function/verify.sh

@@ -0,0 +1,51 @@
+#!/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/crypto-function/manifests.yaml
+kubectl apply -f ${manifests_file} > /dev/null 2>&1
+
+verify_fm_result=$(ci::verify_function_mesh java-function-crypto-sample 2>&1)
+if [ $? -ne 0 ]; then
+  echo "$verify_fm_result"
+  kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true
+  exit 1
+fi
+
+verify_crypto_result=$(NAMESPACE=${PULSAR_NAMESPACE} CLUSTER=${PULSAR_RELEASE_NAME} ci::verify_crypto_function 2>&1)
+if [ $? -eq 0 ]; then
+  echo "e2e-test: ok" | yq eval -
+else
+  echo "$verify_crypto_result"
+fi
+kubectl delete -f "${manifests_file}" > /dev/null 2>&1 || true

.ci/tests/integration/cases/elasticsearch-sink/elasticsearch.yaml

@@ -0,0 +1,18 @@
+apiVersion: elasticsearch.k8s.elastic.co/v1
+kind: Elasticsearch
+metadata:
+  name: quickstart
+spec:
+  version: 7.9.2
+  http:
+    tls:
+      selfSignedCertificate:
+        disabled: true
+  nodeSets:
+    - name: default
+      count: 1
+      config:
+        node.master: true
+        node.data: true
+        node.ingest: true
+        node.store.allow_mmap: false

.ci/tests/integration/cases/elasticsearch-sink/manifests.yaml

@@ -0,0 +1,39 @@
+apiVersion: compute.functionmesh.io/v1alpha1
+kind: Sink
+metadata:
+  name: sink-sample
+spec:
+  className: org.apache.pulsar.io.elasticsearch.ElasticSearchSink
+  replicas: 1
+  maxReplicas: 1
+  input:
+    topics:
+      - persistent://public/default/input-sink-topic
+    typeClassName: "org.apache.pulsar.client.api.schema.GenericObject"
+  sinkConfig:
+    elasticSearchUrl: "http://quickstart-es-http.default.svc.cluster.local:9200"
+    username: "elastic"
+    password: "QqB1OtT6m79vfP7H9H0q2a82"
+  pulsar:
+    pulsarConfig: "test-sink"
+  resources:
+    limits:
+      cpu: "0.2"
+      memory: 1.1G
+    requests:
+      cpu: "0.1"
+      memory: 1G
+  image: streamnative/pulsar-io-elastic-search:2.9.2.23
+  java:
+    jar: connectors/pulsar-io-elastic-search-2.9.2.23.nar
+    jarLocation: "" # use pulsar provided connectors
+  clusterName: test-pulsar
+  autoAck: true
+---
+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