Merge pull request #249 from splunk/dev/test-upgrade-INGEST-17735

mwang2016 · web-flow · commit a9761a0bcf3e · 2020-09-03T16:54:41.000-07:00
added test update connectors after upgrade
diff --git a/.circleci/config.yml b/.circleci/config.yml
@@ -1,4 +1,16 @@
-version: 2
+version: 2.1
+
+workflows:
+  version: 2.1
+  test:
+    jobs:
+      - build_test:
+          filters:
+            branches:
+              ignore:
+                - /^release\/.*/
+                - master
+
 jobs:
   build_test:
     resource_class: large
@@ -47,26 +59,39 @@ jobs:
       - run:
           name: Create file for checksum
           command: echo $CI_SPLUNK_VERSION > /tmp/splunk_version.txt
+      - install_splunk
+      - install_kafka
+      - test_kafka_connect_upgrade
+      - run_functional_tests
+
+commands:
+  install_splunk:
+    description: "Install Splunk"
+    steps:
       - run:
           name: Install Splunk
           command: |
             cd /tmp && wget -O $CI_SPLUNK_FILENAME 'https://www.splunk.com/bin/splunk/DownloadActivityServlet?architecture=x86_64&platform=linux&version='$CI_SPLUNK_VERSION'&product=splunk&filename='$CI_SPLUNK_FILENAME'&wget=true'
             sudo dpkg -i $CI_SPLUNK_FILENAME
             # Set user seed
             hashed_pwd=$(sudo /opt/splunk/bin/splunk hash-passwd $CI_SPLUNK_PASSWORD)
-            sudo tee /opt/splunk/etc/system/local/user-seed.conf > /dev/null << EOF
+            sudo tee /opt/splunk/etc/system/local/user-seed.conf > /dev/null \<< EOF
             [user_info]
             USERNAME = $CI_SPLUNK_USERNAME
             HASHED_PASSWORD = $hashed_pwd
             EOF
             # Add delete capability to admin role
-            sudo tee -a /opt/splunk/etc/system/local/authorize.conf > /dev/null << EOF
+            sudo tee -a /opt/splunk/etc/system/local/authorize.conf > /dev/null \<< EOF
             [role_admin]
             delete_by_keyword = enabled
             EOF
-            # start Splunk
+      - run:
+          name: Start Splunk
+          command: |
             sudo /opt/splunk/bin/splunk start --accept-license --answer-yes --no-prompt
-            # Enable HEC services
+      - run:
+          name: Enable HEC services
+          command: |
             curl -X POST -u $CI_SPLUNK_USERNAME:$CI_SPLUNK_PASSWORD -k https://$CI_SPLUNK_HOST:$CI_SPLUNK_PORT/servicesNS/nobody/splunk_httpinput/data/inputs/http/http/enable
             # Create new HEC token
             curl -X POST -u $CI_SPLUNK_USERNAME:$CI_SPLUNK_PASSWORD -k -d "name=splunk_hec_token&token=$CI_SPLUNK_HEC_TOKEN" https://$CI_SPLUNK_HOST:$CI_SPLUNK_PORT/servicesNS/nobody/splunk_httpinput/data/inputs/http
@@ -78,18 +103,24 @@ jobs:
             sudo /opt/splunk/bin/splunk http-event-collector enable -name splunk_hec_token_ack -uri https://$CI_SPLUNK_HOST:$CI_SPLUNK_PORT -auth $CI_SPLUNK_USERNAME:$CI_SPLUNK_PASSWORD
             # Disable SSL for HEC new-token
             #sudo /opt/splunk/bin/splunk http-event-collector update -uri https://$CI_SPLUNK_HOST:$CI_SPLUNK_PORT -enable-ssl 0 -auth $CI_SPLUNK_USERNAME:$CI_SPLUNK_PASSWORD
-            # Setup Indexes
+      - run:
+          name: Setup Indexes
+          command: |
             curl -X POST -u $CI_SPLUNK_USERNAME:$CI_SPLUNK_PASSWORD -k -d "name=$CI_INDEX_EVENTS&datatype=event" https://$CI_SPLUNK_HOST:$CI_SPLUNK_PORT/servicesNS/-/search/data/indexes
             curl -X POST -u $CI_SPLUNK_USERNAME:$CI_SPLUNK_PASSWORD -k -d "name=$CI_KAFKA_HEADER_INDEX&datatype=event" https://$CI_SPLUNK_HOST:$CI_SPLUNK_PORT/servicesNS/-/search/data/indexes
             sudo /opt/splunk/bin/splunk restart --accept-license --answer-yes --no-prompt
+
+  install_kafka:
+    description: "Install Kafka"
+    steps:
       - run:
           name: Install Kafka
           command: |
             cd /tmp && wget http://www-us.apache.org/dist/kafka/$CI_KAFKA_VERSION/$CI_KAFKA_FILENAME
             sudo tar xzf $CI_KAFKA_FILENAME
             rm $CI_KAFKA_FILENAME
             sudo mv kafka_* /usr/local/kafka
-            sudo tee /etc/systemd/system/zookeeper.service > /dev/null << EOF
+            sudo tee /etc/systemd/system/zookeeper.service > /dev/null \<< EOF
             [Unit]
             Description=Apache Zookeeper server
             Documentation=http://zookeeper.apache.org
@@ -104,7 +135,7 @@ jobs:
             WantedBy=multi-user.target
             EOF
 
-            sudo tee /etc/systemd/system/kafka.service > /dev/null << EOF
+            sudo tee /etc/systemd/system/kafka.service > /dev/null \<< EOF
             [Unit]
             Description=Apache Kafka Server
             Documentation=http://kafka.apache.org/documentation.html
@@ -117,49 +148,46 @@ jobs:
             [Install]
             WantedBy=multi-user.target
             EOF
+      - run:
+          name: Start Kafka Server
+          command: |
             sudo systemctl daemon-reload
             sudo systemctl start zookeeper
             sudo systemctl start kafka
+  test_kafka_connect_upgrade:
+    description: "Test kafka connect upgrade"
+    steps:
       - run:
-          name: Setup kafka connect
+          name: Test kafka connect upgrade
           command: |
+            echo "Download kafka connect "$CI_OLD_CONNECTOR_VERSION
             sudo mkdir -p /usr/local/share/kafka/plugins/
             wget https://github.com/splunk/kafka-connect-splunk/releases/download/$CI_OLD_CONNECTOR_VERSION/splunk-kafka-connect-$CI_OLD_CONNECTOR_VERSION.jar
             sudo cp splunk-kafka-connect-$CI_OLD_CONNECTOR_VERSION.jar /usr/local/share/kafka/plugins/
             sed -i 's/plugin\.path\=connectors\//plugin\.path\=\/usr\/local\/share\/kafka\/plugins\//' /home/circleci/repo/config/connect-distributed-quickstart.properties
             sed -i 's/key\.converter\=org\.apache\.kafka\.connect\.storage\.StringConverter/key\.converter\=org\.apache\.kafka\.connect\.json\.JsonConverter/' /home/circleci/repo/config/connect-distributed-quickstart.properties
             sed -i 's/value\.converter\=org\.apache\.kafka\.connect\.storage\.StringConverter/value\.converter\=org\.apache\.kafka\.connect\.json\.JsonConverter/' /home/circleci/repo/config/connect-distributed-quickstart.properties
-      - run:
-          name: Test kafka connect upgrade
-          command: |
             pyenv global 3.6.5
             pip install --upgrade pip
             pip install -r test/requirements.txt
             export PYTHONWARNINGS="ignore:Unverified HTTPS request"
             echo "Test kafka connect upgrade ..."
             python test/lib/connector_upgrade.py
+
+  run_functional_tests:
+    description: "Run functional tests on current kafka connect version"
+    steps:
       - run:
-          name: Start kafka connect
+          name: Install kafka connect
           command: |
+            sudo rm -f /usr/local/share/kafka/plugins/splunk-kafka-connect*.jar
+            sudo cp /home/circleci/repo/target/splunk-kafka-connect*.jar /usr/local/share/kafka/plugins/
+            test -f /usr/local/share/kafka/plugins/splunk-kafka-connect*.jar && echo /usr/local/share/kafka/plugins/splunk-kafka-connect*.jar
             sudo /usr/local/kafka/bin/connect-distributed.sh /home/circleci/repo/config/connect-distributed-quickstart.properties
           background: true
       - run:
           name: Run Functional tests
           command: |
             sleep 5
-            sudo apt-get install jq
-            curl localhost:8083/connector-plugins | jq
             echo "Running functional tests....."
             python -m pytest -p no:warnings -s
-
-workflows:
-  version: 2
-  test:
-    jobs:
-      - build_test:
-          filters:
-            branches:
-              ignore: 
-                - /^release\/.*/
-                - master
-
diff --git a/test/lib/connector_upgrade.py b/test/lib/connector_upgrade.py
@@ -22,6 +22,9 @@
     config = yaml.load(yaml_file)
 now = datetime.now()
 _time_stamp = str(datetime.timestamp(now))
+_topic = 'kafka_connect_upgrade'
+_connector = 'kafka_connect'
+_connector_ack = 'kafka_connect_ack'
 
 
 def start_old_connector():
@@ -41,47 +44,53 @@ def start_old_connector():
 
 def generate_kafka_events(num):
     # Generate message data
-    topics = ["kafka_data_gen"]
+    topics = [_topic]
     connector_content = {
-        "name": "kafka_connect",
+        "name": _connector,
         "config": {
             "connector.class": "com.splunk.kafka.connect.SplunkSinkConnector",
             "tasks.max": "1",
             "splunk.indexes": config["splunk_index"],
-            "topics": "kafka_data_gen",
+            "topics": _topic,
             "splunk.hec.ack.enabled": "false",
             "splunk.hec.uri": config["splunk_hec_url"],
             "splunk.hec.ssl.validate.certs": "false",
-            "splunk.hec.token": config["splunk_token"]
+            "splunk.hec.token": config["splunk_token"],
+            "splunk.sources": _connector
         }
     }
     create_kafka_connector(config, connector_content)
     connector_content_ack = {
-        "name": "kafka_connect_ack",
+        "name": _connector_ack,
         "config": {
             "connector.class": "com.splunk.kafka.connect.SplunkSinkConnector",
             "tasks.max": "1",
             "splunk.indexes": config["splunk_index"],
-            "topics": "kafka_data_gen",
+            "topics": _topic,
             "splunk.hec.ack.enabled": "true",
             "splunk.hec.uri": config["splunk_hec_url"],
             "splunk.hec.ssl.validate.certs": "false",
-            "splunk.hec.token": config["splunk_token_ack"]
+            "splunk.hec.token": config["splunk_token_ack"],
+            "splunk.sources": _connector_ack
         }
     }
     create_kafka_connector(config, connector_content_ack)
-    create_kafka_topics(config, topics)
+    client = KafkaAdminClient(bootstrap_servers=config["kafka_broker_url"], client_id='test')
+    broker_topics = client.list_topics()
+    logger.info(broker_topics)
+    if _topic not in broker_topics:
+        create_kafka_topics(config, topics)
     producer = KafkaProducer(bootstrap_servers=config["kafka_broker_url"],
                              value_serializer=lambda v: json.dumps(v).encode('utf-8'))
 
     for _ in range(num):
         msg = {"timestamp": _time_stamp}
-        producer.send("kafka_data_gen", msg)
+        producer.send(_topic, msg)
         time.sleep(0.05)
         producer.flush()
 
 
-def upgrade_connector():
+def upgrade_connector_plugin():
     cmds = ["sudo kill $(sudo lsof -t -i:8083) && sleep 2",
             "sudo rm {}/{} && sleep 2".format(config["connector_path"], config["old_connector_name"]),
             "sudo cp {0}/splunk-kafka-connect*.jar {1} && sleep 2".format(config["connector_build_target"],
@@ -96,10 +105,50 @@ def upgrade_connector():
                                 stderr=subprocess.STDOUT)
         output, error = proc.communicate()
         logger.info(output)
+        time.sleep(2)
+        update_kafka_connectors()
     except OSError as e:
         logger.error(e)
 
 
+def update_kafka_connectors():
+    logger.info("Update kafka connectors ...")
+    connector_content = {
+        "name": _connector,
+        "config": {
+            "connector.class": "com.splunk.kafka.connect.SplunkSinkConnector",
+            "tasks.max": "1",
+            "splunk.indexes": config["splunk_index"],
+            "topics": _topic,
+            "splunk.hec.ack.enabled": "false",
+            "splunk.hec.uri": config["splunk_hec_url"],
+            "splunk.hec.ssl.validate.certs": "false",
+            "splunk.hec.token": config["splunk_token"],
+            "splunk.sources": _connector,
+            "splunk.hec.json.event.formatted": "true",
+            "splunk.hec.raw": True
+        }
+    }
+    create_kafka_connector(config, connector_content)
+    connector_content_ack = {
+        "name": _connector_ack,
+        "config": {
+            "connector.class": "com.splunk.kafka.connect.SplunkSinkConnector",
+            "tasks.max": "1",
+            "splunk.indexes": config["splunk_index"],
+            "topics": _topic,
+            "splunk.hec.ack.enabled": "true",
+            "splunk.hec.uri": config["splunk_hec_url"],
+            "splunk.hec.ssl.validate.certs": "false",
+            "splunk.hec.token": config["splunk_token_ack"],
+            "splunk.sources": _connector_ack,
+            "splunk.hec.json.event.formatted": "true",
+            "splunk.hec.raw": True
+        }
+    }
+    create_kafka_connector(config, connector_content_ack)
+
+
 if __name__ == '__main__':
     logger.info("Start old Kafka connector ...")
     thread_old_connect = threading.Thread(target=start_old_connector, daemon=True)
@@ -110,15 +159,26 @@ def upgrade_connector():
     thread_gen.start()
     time.sleep(50)
     logger.info("Upgrade Kafka connector ...")
-    thread_upgrade = threading.Thread(target=upgrade_connector, daemon=True)
+    thread_upgrade = threading.Thread(target=upgrade_connector_plugin, daemon=True)
     thread_upgrade.start()
     time.sleep(100)
-    search_query = "index={0} | search timestamp=\"{1}\"".format(config['splunk_index'], _time_stamp)
-    logger.info(search_query)
-    events = check_events_from_splunk(start_time="-15m@m",
+    search_query_1 = "index={0} | search timestamp=\"{1}\" source::{2}".format(config['splunk_index'], _time_stamp,
+                                                                               _connector)
+    logger.info(search_query_1)
+    events_1 = check_events_from_splunk(start_time="-15m@m",
                                       url=config["splunkd_url"],
                                       user=config["splunk_user"],
-                                      query=["search {}".format(search_query)],
+                                      query=["search {}".format(search_query_1)],
                                       password=config["splunk_password"])
-    logger.info("Splunk received %s events in the last 15m", len(events))
-    assert len(events) == 4000
+    logger.info("Splunk received %s events in the last 15m", len(events_1))
+    assert len(events_1) == 2000
+    search_query_2 = "index={0} | search timestamp=\"{1}\" source::{2}".format(config['splunk_index'], _time_stamp,
+                                                                               _connector_ack)
+    logger.info(search_query_2)
+    events_2 = check_events_from_splunk(start_time="-15m@m",
+                                        url=config["splunkd_url"],
+                                        user=config["splunk_user"],
+                                        query=["search {}".format(search_query_2)],
+                                        password=config["splunk_password"])
+    logger.info("Splunk received %s events in the last 15m", len(events_2))
+    assert len(events_2) == 2000
diff --git a/test/testcases/test_configurations.py b/test/testcases/test_configurations.py
@@ -141,9 +141,9 @@ def test_header_support_false_event_data(self, setup, test_scenario, test_input,
     def test_header_support_true_event_data(self, setup, test_scenario, test_input, expected):
         logger.info("testing {0} input={1} expected={2} event(s)".format(test_scenario, test_input, expected))
 
-        search_query = "index={0} | search timestamp=\"{1}\" {2}".format(setup['kafka_header_index'],
-                                                                         setup["timestamp"],
-                                                                         test_input)
+        search_query = "index={0} | search \"{1}\" {2}".format(setup['kafka_header_index'],
+                                                               setup["timestamp"],
+                                                               test_input)
         logger.info(search_query)
         events = check_events_from_splunk(start_time="-15m@m",
                                           url=setup["splunkd_url"],
