Adding Kafka Connect Test (#21)

* Updating README to reflect mTLS support

* adding kafkaconnect test components

* fixing bug in internaltest

* removing extra logs from test

Author: mtarng

README.md

@@ -31,7 +31,6 @@ Currently Portworx must be installed manually due to the lack of a Helm chart. P
 ## Future Work
 
 - [ ] Work with Portworx on Helm Chart
-- [ ] Enable mTLS on ingress for Kafka load balancer
 - [ ] Enable volume encryption on Portworx volumes
 - [ ] Get past possible VNET issues in Terraform template
 

test/README.md

@@ -25,12 +25,23 @@ The test results will displayed at the end. The script with exit with code 0 for
 - `kafkacat` needs to be installed and added to $PATH
   - on systems with homebrew installed, you can run `brew install kafkacat`
   - [Kafkacat Installation](https://github.com/edenhill/kafkacat#install)
+- Optionally, Kafka Connect can be tested if there is a backing database for it.
+  - Instructions can be found under the [kafka connect examples](../examples/kafka_connect)
+  - After the test is run, you can check the collection for the test messages
 
 ### Running the test
 
 Run the testing script as configured below:
 - If the External LoadBalancer is enabled with TLS support: `sh ./externaltest.sh -t`
 - If the External LoadBalancer is NOT enabled with TLS support: `sh ./externaltest.sh`
+- If kafkaconnect is enabled:
+  -  export three variables in your shell:
+        ```
+            export MONGODB_CONN_URL='<mongodb-connection-string>'
+            export DATABASE='<database-name>'
+            export COLLECTION'<collection-name>'
+        ```
+  - add the `-k` flag to the arguments: `sh ./externaltest.sh -t -k`
   
 The script will deploy a test topic and connect to the brokers through the external loadbalancer IP, utilizing kafkacat as a producer and consumer.
 The test results will displayed at the end. The script with exit with code 0 for success and 1 for failure.

test/externaltest.sh

@@ -4,8 +4,28 @@
 YELLOW='\033[1;33m'
 RED='\033[1;31m'
 GREEN='\033[0;32m'
+CYAN='\033[0;36m'
 NC='\033[0m' # No Color
 
+MTLS_ON=false
+K_CONNECT_ON=false
+
+#### ARGS AND INPUTS
+while [ ! $# -eq 0 ]
+do
+	case "$1" in
+		-t)
+      echo "${CYAN}mTLS/SSL Support enabled.${NC}"
+      MTLS_ON=true
+			;;
+		-k)
+			echo "${CYAN}Kafka Connect test enabled.${NC}"
+      K_CONNECT_ON=true
+			;;
+	esac
+	shift
+done
+
 #### FUNCTIONS --------------------------------
 create_and_deploy_kafka_test_topic_yaml()
 {
@@ -62,6 +82,57 @@ cleanup_cluster()
     kubectl delete --recursive -f ./temp/${TESTING_TOPIC}
 }
 
+deploy_kafka_connect_connector()
+{
+  # Get MongoDB Credentials from Env Vars.
+
+  # Create connector payload
+  CONNECTOR_FILE="./temp/${TESTING_TOPIC}/create-connector.json"
+  cp ../examples/kafka_connect/connectors/mongoDBSink/CreateMongoSinkConnector.json $CONNECTOR_FILE
+  
+  # 2 "name"
+  KCONNECT_NAME="${TESTING_TOPIC}-connector"
+  echo "Kafka Connect Sink Name: ${KCONNECT_NAME}"
+  sed -i.bak "2s|.*|    \"name\": \"${KCONNECT_NAME}\",|" $CONNECTOR_FILE
+  # 4 "topic"
+  sed -i.bak "4s|.*|        \"topics\": \"${TESTING_TOPIC}\",|" $CONNECTOR_FILE
+  # 12 "connection.uri"
+  sed -i.bak "12s|.*|        \"connection.uri\": \"${MONGODB_CONN_URL}\",|" $CONNECTOR_FILE
+  # 13 "database"
+  sed -i.bak "13s|.*|        \"database\": \"${DATABASE}\",|" $CONNECTOR_FILE
+  # 14 "collection"
+  sed -i.bak "14s|.*|        \"collection\": \"${COLLECTION}\",|" $CONNECTOR_FILE
+
+  # Portforward to kafkaconnect pod & Create Request
+  KCONNECT_POD=`kubectl get pods -n kafka | grep kconnect-cluster-connect | awk '{print $1}'`
+  echo "Kafka Connect Pod: ${KCONNECT_POD}"
+  kubectl -n kafka port-forward $KCONNECT_POD 8083:8083 &
+  KCONNECT_PORT_FORWARD_PID=$!
+
+  sleep 5
+
+  curl -H 'Content-Type: application/json' -X POST -d @$CONNECTOR_FILE http://localhost:8083/connectors
+
+  sleep 5
+  kill $KCONNECT_PORT_FORWARD_PID
+}
+
+remove_kafka_connect_connector()
+{
+  # Portforward to kafkaconnect pod & Create Request
+  KCONNECT_POD=`kubectl get pods -n kafka | grep kconnect-cluster-connect | awk '{print $1}'`
+  echo "Kafka Connect Pod: ${KCONNECT_POD}"
+  kubectl -n kafka port-forward $KCONNECT_POD 8083:8083 &
+  KCONNECT_PORT_FORWARD_PID=$!
+
+  sleep 5
+  
+  curl -X DELETE http://localhost:8083/connectors/$KCONNECT_NAME
+
+  sleep 5
+  kill $KCONNECT_PORT_FORWARD_PID
+}
+
 #### MAIN --------------------------------
 
 # Get Broker LoadBalancer Address
@@ -92,7 +163,7 @@ kubectl apply -f temp/${TESTING_TOPIC}/kafka-test-topic.yaml
 sleep 2
 
 # Create Kafkacat configuration based if TLS/SSL enforcement is enabled
-if [ $1 == "-t" ]; then
+if [ $MTLS_ON == true ]; then
     # TLS Enabled on cluster
     echo "${YELLOW}Configuring Kafkacat with SSL${NC}"
     # Deploy test user with access to test topic
@@ -119,6 +190,11 @@ if [ $1 == "-t" ]; then
     echo "bootstrap.servers=${BROKER_EXTERNAL_ADDRESS}" > temp/${TESTING_TOPIC}/kafkacat.config
 fi
 
+# Deploy Kafka Connect Sink
+if [ $K_CONNECT_ON == true ]; then
+  deploy_kafka_connect_connector
+fi
+
 # Create random test messages
 MESSAGE_INPUT_FILE="./temp/${TESTING_TOPIC}/input-messages.txt"
 
@@ -145,6 +221,11 @@ kill $CONSUMER_PID
 # Delete test topic and user
 cleanup_cluster $TESTING_TOPIC
 
+# Deploy Kafka Connect Sink
+if [ $K_CONNECT_ON == true ]; then
+  remove_kafka_connect_connector
+fi
+
 # Compare contents of input and output
 SORTED_INPUT="./temp/${TESTING_TOPIC}/sorted-input.txt"
 SORTED_OUTPUT="./temp/${TESTING_TOPIC}/sorted-output.txt"

test/internaltest.sh

@@ -21,6 +21,9 @@ echo $UUID
 TESTING_TOPIC="topic-${UUID}"
 echo "${YELLOW}Test Topic: ${TESTING_TOPIC}${NC}"
 
+# Create testing directory
+mkdir temp/${TESTING_TOPIC}
+
 # TODO: Deploy via CRD with kafka-topics.yaml
 # Deploy via kafka broker pod - Alternatively this can be done through the CRD.
 kubectl exec -n kafka -ti kcluster-kafka-0 --container kafka -- bin/kafka-topics.sh --zookeeper localhost:2181 --create --topic $TESTING_TOPIC --partitions 3 --replication-factor 2