| #!/usr/bin/env python |
| # -*- encoding: utf-8 -*- |
| |
| # 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. |
| |
| """ test_template.py """ |
| from future.standard_library import install_aliases |
| install_aliases() |
| |
| import json |
| import logging |
| import os |
| import time |
| from urllib.request import urlopen |
| import signal |
| import subprocess |
| from collections import namedtuple |
| |
| from ..common import status |
| |
| # Test input. Please set each variable as it's own line, ended with \n, otherwise the value of lines |
| # passed into the topology will be incorrect, and the test will fail. |
| TEST_INPUT = ["1\n", "2\n", "3\n", "4\n", "5\n", "6\n", "7\n", "8\n", |
| "9\n", "10\n", "11\n", "12\n"] |
| |
| # Retry variables in case the output is different from the input |
| RETRY_COUNT = 5 |
| RETRY_INTERVAL = 10 |
| # Topology shard definitions |
| NON_TMASTER_SHARD = 1 |
| # Topology process name definitions |
| STMGR = 'stmgr' |
| HERON_BIN = "bin" |
| HERON_CORE = "heron-core" |
| HERON_METRICSMGR = 'metricsmgr' |
| HERON_SANDBOX_HOME = "." |
| HERON_STMGR = "heron-stmgr" |
| HERON_STMGR_CMD = os.path.join(HERON_SANDBOX_HOME, HERON_CORE, HERON_BIN, HERON_STMGR) |
| ProcessTuple = namedtuple('ProcessTuple', 'pid cmd') |
| |
| class TestTemplate(object): |
| """ Class that encapsulates the template used for integration tests. Intended to be abstract and |
| subclassed for specific tests. """ |
| |
| def __init__(self, testname, params): |
| self.testname = testname |
| self.params = params |
| |
| # pylint: disable=too-many-return-statements, too-many-branches, |
| # pylint: disable=too-many-statements |
| def run_test(self): |
| """ Runs the test template. Must either return TestSuccess or raise TestFailure""" |
| topology_submitted = False |
| try: |
| # prepare test data, start the topology and block until it's running |
| self._prepare_test_data() |
| self.submit_topology() |
| topology_submitted = True |
| _block_until_stmgr_running(self.get_expected_container_count()) |
| |
| self._block_until_topology_running(self.get_expected_min_instance_count()) |
| |
| # Execute the specific test logic and block until topology is running again |
| self.execute_test_case() |
| |
| _block_until_stmgr_running(self.get_expected_container_count()) |
| physical_plan_json =\ |
| self._block_until_topology_running(self.get_expected_min_instance_count()) |
| |
| # trigger the test data to flow and invoke the pre_check_results hook |
| self._inject_test_data() |
| self.pre_check_results(physical_plan_json) |
| |
| # finally verify the expected results |
| result = self._check_results() |
| return result |
| |
| except status.TestFailure as e: |
| raise e |
| except Exception as e: |
| raise status.TestFailure("Exception thrown during test", e) |
| finally: |
| if topology_submitted: |
| self.cleanup_test() |
| |
| def submit_topology(self): |
| _submit_topology( |
| self.params['cliPath'], |
| self.params['cluster'], |
| self.params['testJarPath'], |
| self.params['topologyClassPath'], |
| self.params['topologyName'], |
| self.params['readFile'], |
| self.params['outputFile'] |
| ) |
| |
| # pylint: disable=no-self-use |
| def get_expected_container_count(self): |
| return 1 |
| |
| # pylint: disable=no-self-use |
| def get_expected_min_instance_count(self): |
| return 1 |
| |
| def execute_test_case(self): |
| pass |
| |
| # pylint: disable=no-self-use,unused-argument |
| def pre_check_results(self, physical_plan_json): |
| return True |
| |
| def cleanup_test(self): |
| try: |
| _kill_topology(self.params['cliPath'], self.params['cluster'], self.params['topologyName']) |
| except Exception as e: |
| logging.error("Failed to kill %s topology: %s", self.params['topologyName'], str(e)) |
| finally: |
| self._delete_test_data_files() |
| |
| def _delete_test_data_files(self): |
| _safe_delete_file(self.params['readFile']) |
| _safe_delete_file(self.params['outputFile']) |
| |
| def _prepare_test_data(self): |
| self._delete_test_data_files() |
| |
| # insert lines into temp file and then move to read file |
| try: |
| with open('temp.txt', 'w') as f: |
| for line in TEST_INPUT: |
| f.write(line) |
| except Exception as e: |
| logging.error("Failed to write to temp.txt file: %s", str(e)) |
| return False |
| |
| def _inject_test_data(self): |
| # move to read file. This guarantees contents will be put into the file the |
| # spout is reading from atomically |
| # which increases the determinism |
| os.rename('temp.txt', self.params['readFile']) |
| |
| def _check_results(self): |
| """ get actual and expected result. |
| retry if results are not equal a predesignated amount of times |
| """ |
| expected_result = "" |
| actual_result = "" |
| retries_left = RETRY_COUNT |
| _sleep("before trying to check results for test %s" % self.testname, RETRY_INTERVAL) |
| while retries_left > 0: |
| retries_left -= 1 |
| try: |
| with open(self.params['readFile'], 'r') as f: |
| expected_result = f.read() |
| with open(self.params['outputFile'], 'r') as g: |
| actual_result = g.read() |
| except Exception as e: |
| message =\ |
| "Failed to read expected or actual results from file for test %s: %s" % self.testname |
| if retries_left == 0: |
| raise status.TestFailure(message, e) |
| logging.error(message, e) |
| # if we get expected result, no need to retry |
| expected_sorted = sorted(expected_result.split('\n')) |
| actual_sorted = sorted(actual_result.split('\n')) |
| if expected_sorted == actual_sorted: |
| break |
| if retries_left > 0: |
| expected_result = "" |
| actual_result = "" |
| expected_sorted = [] |
| actual_sorted = [] |
| logging.info("Failed to get expected results for test %s (attempt %s/%s), "\ |
| + "retrying after %s seconds", |
| self.testname, RETRY_COUNT - retries_left, RETRY_COUNT, RETRY_INTERVAL) |
| time.sleep(RETRY_INTERVAL) |
| |
| # Compare the actual and expected result |
| if actual_sorted == expected_sorted: |
| success = status.TestSuccess( |
| "Actual result matched expected result for test %s" % self.testname) |
| logging.info("Actual result ---------- \n%s", actual_sorted) |
| logging.info("Expected result ---------- \n%s", expected_sorted) |
| return success |
| else: |
| failure = status.TestFailure( |
| "Actual result did not match expected result for test %s" % self.testname) |
| logging.info("Actual result ---------- \n%s", actual_sorted) |
| logging.info("Expected result ---------- \n%s", expected_sorted) |
| raise failure |
| |
| # pylint: disable=no-self-use |
| def get_pid(self, process_name, heron_working_directory): |
| """ |
| opens .pid file of process and reads the first and only line, which should be the process pid |
| if fail, return -1 |
| """ |
| process_pid_file = os.path.join(heron_working_directory, process_name + '.pid') |
| try: |
| with open(process_pid_file, 'r') as f: |
| pid = f.readline() |
| return pid |
| except Exception: |
| logging.error("Unable to open file %s", process_pid_file) |
| return -1 |
| |
| # pylint: disable=no-self-use |
| def kill_process(self, process_number): |
| """ kills process by running unix command kill """ |
| if process_number < 1: |
| raise RuntimeError( |
| "Not attempting to kill process id < 1 passed to kill_process: %d" % process_number) |
| |
| logging.info("Killing process number %s", process_number) |
| |
| try: |
| os.kill(int(process_number), signal.SIGTERM) |
| except OSError as ex: |
| if "No such process" in str(ex): # killing a non-existing process condsidered as success |
| logging.info(str(ex)) |
| else: |
| raise RuntimeError("Unable to kill process %s" % process_number) |
| except Exception: |
| raise RuntimeError("Unable to kill process %s" % process_number) |
| |
| logging.info("Killed process number %s", process_number) |
| |
| def kill_strmgr(self): |
| logging.info("Executing kill stream manager") |
| stmgr_pid = self.get_pid('%s-%d' % (STMGR, NON_TMASTER_SHARD), self.params['workingDirectory']) |
| self.kill_process(stmgr_pid) |
| |
| def kill_metricsmgr(self): |
| logging.info("Executing kill metrics manager") |
| metricsmgr_pid = self.get_pid( |
| '%s-%d' % (HERON_METRICSMGR, NON_TMASTER_SHARD), self.params['workingDirectory']) |
| self.kill_process(metricsmgr_pid) |
| |
| def _get_tracker_pplan(self): |
| url = 'http://localhost:%s/topologies/physicalplan?' % self.params['trackerPort']\ |
| + 'cluster=local&environ=default&topology=IntegrationTest_LocalReadWriteTopology' |
| logging.debug("Fetching physical plan from %s", url) |
| response = urlopen(url) |
| physical_plan_json = json.loads(response.read()) |
| |
| if 'result' not in physical_plan_json: |
| raise status.TestFailure( |
| "Could not find result json in physical plan request to tracker: %s" % url) |
| |
| return physical_plan_json['result'] |
| |
| def _block_until_topology_running(self, min_instances): |
| retries_left = RETRY_COUNT |
| _sleep("before trying to fetch pplan for test %s" % self.testname, RETRY_INTERVAL) |
| while retries_left > 0: |
| retries_left -= 1 |
| packing_plan = self._get_tracker_pplan() |
| if packing_plan: |
| instances_found = len(packing_plan['instances']) |
| if instances_found >= min_instances: |
| logging.info("Successfully fetched pplan from tracker for test %s after %s attempts.", |
| self.testname, RETRY_COUNT - retries_left) |
| return packing_plan |
| elif retries_left == 0: |
| raise status.TestFailure( |
| "Got pplan from tracker for test %s but the number of " % self.testname + |
| "instances found (%d) was less than min expected (%s)." % |
| (instances_found, min_instances)) |
| |
| if retries_left > 0: |
| _sleep("before trying again to fetch pplan for test %s (attempt %s/%s)" % |
| (self.testname, RETRY_COUNT - retries_left, RETRY_COUNT), RETRY_INTERVAL) |
| else: |
| raise status.TestFailure("Failed to get pplan from tracker for test %s after %s attempts." |
| % (self.testname, RETRY_COUNT)) |
| |
| def _block_until_stmgr_running(expected_stmgrs): |
| # block until ./heron-stmgr exists |
| process_list = _get_processes() |
| while not _processes_exists(process_list, HERON_STMGR_CMD, expected_stmgrs): |
| process_list = _get_processes() |
| time.sleep(1) |
| |
| def _submit_topology(heron_cli_path, test_cluster, test_jar_path, topology_class_path, |
| topology_name, input_file, output_file): |
| """ Submit topology using heron-cli """ |
| # unicode string messes up subprocess.call quotations, must change into string type |
| splitcmd = [ |
| heron_cli_path, 'submit', '--verbose', '--', test_cluster, test_jar_path, |
| topology_class_path, topology_name, input_file, output_file, str(len(TEST_INPUT)) |
| ] |
| logging.info("Submitting topology: %s", splitcmd) |
| p = subprocess.Popen(splitcmd) |
| p.wait() |
| if p.returncode != 0: |
| raise status.TestFailure("Failed to submit topology %s" % topology_name) |
| |
| logging.info("Submitted topology %s", topology_name) |
| |
| def _kill_topology(heron_cli_path, test_cluster, topology_name): |
| """ Kill a topology using heron-cli """ |
| splitcmd = [heron_cli_path, 'kill', test_cluster, topology_name] |
| logging.info("Killing topology: %s", ' '.join(splitcmd)) |
| # this call can be blocking, no need for subprocess |
| if subprocess.call(splitcmd) != 0: |
| raise RuntimeError("Unable to kill the topology: %s" % topology_name) |
| |
| def _get_processes(): |
| """ |
| returns a list of process tuples (pid, cmd) |
| This only applies only for local scheduler as it uses the ps command |
| and assumes the topology will be running on different processes on same machine |
| """ |
| # pylint: disable=fixme |
| # TODO: if the submit fails before we get here (e.g., Topology already exists), this hangs |
| processes = subprocess.check_output(['ps', '-o', 'pid,args']) |
| processes = processes.split('\n') |
| processes = processes[1:] # remove first line, which is name of columns |
| process_list = [] |
| for process in processes: |
| # remove empty lines |
| if process == '': |
| continue |
| pretuple = process.split(' ', 1) |
| process_list.append(ProcessTuple(pretuple[0], pretuple[1])) |
| return process_list |
| |
| def _sleep(message, seconds): |
| logging.info("Sleeping for %d seconds %s", seconds, message) |
| time.sleep(seconds) |
| |
| def _processes_exists(process_list, process_cmd, min_processes): |
| """ check if a process is running """ |
| proccess_count = 0 |
| for process in process_list: |
| if process_cmd in process.cmd: |
| proccess_count += 1 |
| |
| return proccess_count >= min_processes |
| |
| def _safe_delete_file(file_name): |
| if os.path.isfile(file_name) and os.path.exists(file_name): |
| try: |
| os.remove(file_name) |
| except Exception as e: |
| logging.error("Failed to delete file: %s: %s", file_name, e) |
| return False |
