From 429b8d2e985695b5770d0d4be0cee0999de53dc6 Mon Sep 17 00:00:00 2001 From: jesterhazy Date: Tue, 18 Jun 2019 17:14:49 -0700 Subject: [PATCH 1/2] fix: prevent race condition in vpc tests --- tests/integ/{local_mode_utils.py => lock.py} | 17 ++++----- tests/integ/test_local_mode.py | 23 ++++++------ tests/integ/test_source_dirs.py | 4 +-- tests/integ/vpc_test_utils.py | 38 +++++++++++++------- 4 files changed, 48 insertions(+), 34 deletions(-) rename tests/integ/{local_mode_utils.py => lock.py} (64%) diff --git a/tests/integ/local_mode_utils.py b/tests/integ/lock.py similarity index 64% rename from tests/integ/local_mode_utils.py rename to tests/integ/lock.py index 31bf1ed061..bea14350e0 100644 --- a/tests/integ/local_mode_utils.py +++ b/tests/integ/lock.py @@ -15,24 +15,21 @@ import fcntl import os import time +import tempfile from contextlib import contextmanager -import tests.integ - -LOCK_PATH = os.path.join(tests.integ.DATA_DIR, 'local_mode_lock') +DEFAULT_LOCK_PATH = os.path.join(tempfile.gettempdir(), 'sagemaker_test_lock') @contextmanager -def lock(): - # Since Local Mode uses the same port for serving, we need a lock in order - # to allow concurrent test execution. - local_mode_lock_fd = open(LOCK_PATH, 'w') - local_mode_lock = local_mode_lock_fd.fileno() +def lock(path=DEFAULT_LOCK_PATH): + f = open(path, 'w') + fd = f.fileno() - fcntl.lockf(local_mode_lock, fcntl.LOCK_EX) + fcntl.lockf(fd, fcntl.LOCK_EX) try: yield finally: time.sleep(5) - fcntl.lockf(local_mode_lock, fcntl.LOCK_UN) + fcntl.lockf(fd, fcntl.LOCK_UN) diff --git a/tests/integ/test_local_mode.py b/tests/integ/test_local_mode.py index 7c84648e29..78d875e557 100644 --- a/tests/integ/test_local_mode.py +++ b/tests/integ/test_local_mode.py @@ -18,7 +18,9 @@ import boto3 import numpy import pytest -import tests.integ.local_mode_utils as local_mode_utils +import tempfile + +import tests.integ.lock as lock from tests.integ import DATA_DIR, PYTHON_VERSION from tests.integ.timeout import timeout @@ -26,6 +28,7 @@ from sagemaker.mxnet import MXNet from sagemaker.tensorflow import TensorFlow +LOCK_PATH = os.path.join(tempfile.gettempdir(), 'sagemaker_test_local_mode_lock') DATA_PATH = os.path.join(DATA_DIR, 'iris', 'data') DEFAULT_REGION = 'us-west-2' @@ -101,7 +104,7 @@ def test_tf_local_mode(tf_full_version, sagemaker_local_session): print('job succeeded: {}'.format(estimator.latest_training_job.name)) endpoint_name = estimator.latest_training_job.name - with local_mode_utils.lock(): + with lock.lock(LOCK_PATH): try: json_predictor = estimator.deploy(initial_instance_count=1, instance_type='local', @@ -140,7 +143,7 @@ def test_tf_distributed_local_mode(sagemaker_local_session): endpoint_name = estimator.latest_training_job.name - with local_mode_utils.lock(): + with lock.lock(LOCK_PATH): try: json_predictor = estimator.deploy(initial_instance_count=1, instance_type='local', @@ -178,7 +181,7 @@ def test_tf_local_data(sagemaker_local_session): print('job succeeded: {}'.format(estimator.latest_training_job.name)) endpoint_name = estimator.latest_training_job.name - with local_mode_utils.lock(): + with lock.lock(LOCK_PATH): try: json_predictor = estimator.deploy(initial_instance_count=1, instance_type='local', @@ -217,7 +220,7 @@ def test_tf_local_data_local_script(): print('job succeeded: {}'.format(estimator.latest_training_job.name)) endpoint_name = estimator.latest_training_job.name - with local_mode_utils.lock(): + with lock.lock(LOCK_PATH): try: json_predictor = estimator.deploy(initial_instance_count=1, instance_type='local', @@ -241,7 +244,7 @@ def test_local_mode_serving_from_s3_model(sagemaker_local_session, mxnet_model, s3_model.sagemaker_session = sagemaker_local_session predictor = None - with local_mode_utils.lock(): + with lock.lock(LOCK_PATH): try: predictor = s3_model.deploy(initial_instance_count=1, instance_type='local') data = numpy.zeros(shape=(1, 1, 28, 28)) @@ -255,7 +258,7 @@ def test_local_mode_serving_from_s3_model(sagemaker_local_session, mxnet_model, def test_local_mode_serving_from_local_model(tmpdir, sagemaker_local_session, mxnet_model): predictor = None - with local_mode_utils.lock(): + with lock.lock(LOCK_PATH): try: path = 'file://%s' % (str(tmpdir)) model = mxnet_model(path) @@ -285,7 +288,7 @@ def test_mxnet_local_mode(sagemaker_local_session, mxnet_full_version): mx.fit({'train': train_input, 'test': test_input}) endpoint_name = mx.latest_training_job.name - with local_mode_utils.lock(): + with lock.lock(LOCK_PATH): try: predictor = mx.deploy(1, 'local', endpoint_name=endpoint_name) data = numpy.zeros(shape=(1, 1, 28, 28)) @@ -310,7 +313,7 @@ def test_mxnet_local_data_local_script(mxnet_full_version): mx.fit({'train': train_input, 'test': test_input}) endpoint_name = mx.latest_training_job.name - with local_mode_utils.lock(): + with lock.lock(LOCK_PATH): try: predictor = mx.deploy(1, 'local', endpoint_name=endpoint_name) data = numpy.zeros(shape=(1, 1, 28, 28)) @@ -365,7 +368,7 @@ def test_local_transform_mxnet(sagemaker_local_session, tmpdir, mxnet_full_versi transformer = mx.transformer(1, 'local', assemble_with='Line', max_payload=1, strategy='SingleRecord', output_path=output_path) - with local_mode_utils.lock(): + with lock.lock(LOCK_PATH): transformer.transform(transform_input, content_type='text/csv', split_type='Line') transformer.wait() diff --git a/tests/integ/test_source_dirs.py b/tests/integ/test_source_dirs.py index 98410d6fb0..fef75760ba 100644 --- a/tests/integ/test_source_dirs.py +++ b/tests/integ/test_source_dirs.py @@ -16,7 +16,7 @@ import pytest -import tests.integ.local_mode_utils as local_mode_utils +import tests.integ.lock as lock from tests.integ import DATA_DIR, PYTHON_VERSION from sagemaker.pytorch.estimator import PyTorch @@ -37,7 +37,7 @@ def test_source_dirs(tmpdir, sagemaker_local_session): sagemaker_session=sagemaker_local_session) estimator.fit() - with local_mode_utils.lock(): + with lock.lock(): try: predictor = estimator.deploy(initial_instance_count=1, instance_type='local') predict_response = predictor.predict([7]) diff --git a/tests/integ/vpc_test_utils.py b/tests/integ/vpc_test_utils.py index 28783211fb..3cccc4630d 100644 --- a/tests/integ/vpc_test_utils.py +++ b/tests/integ/vpc_test_utils.py @@ -12,7 +12,13 @@ # language governing permissions and limitations under the License. from __future__ import absolute_import +import os +import tempfile + +import tests.integ.lock as lock + VPC_NAME = 'sagemaker-python-sdk-test-vpc' +LOCK_PATH = os.path.join(tempfile.gettempdir(), 'sagemaker_test_vpc_lock') def _get_subnet_ids_by_name(ec2_client, name): @@ -61,12 +67,15 @@ def _create_vpc_with_name(ec2_client, region, name): AvailabilityZone=(region + 'b'))['Subnet']['SubnetId'] print('created subnet: {}'.format(subnet_id_b)) - s3_service = [s for s in ec2_client.describe_vpc_endpoint_services()['ServiceNames'] if s.endswith('s3')][0] + s3_service = \ + [s for s in ec2_client.describe_vpc_endpoint_services()['ServiceNames'] if + s.endswith('s3')][0] ec2_client.create_vpc_endpoint(VpcId=vpc_id, ServiceName=s3_service, RouteTableIds=[_get_route_table_id(ec2_client, vpc_id)]) print('created s3 vpc endpoint') - security_group_id = ec2_client.create_security_group(VpcId=vpc_id, GroupName=name, Description=name)['GroupId'] + security_group_id = \ + ec2_client.create_security_group(VpcId=vpc_id, GroupName=name, Description=name)['GroupId'] print('created security group: {}'.format(security_group_id)) # multi-host vpc jobs require communication among hosts @@ -74,7 +83,8 @@ def _create_vpc_with_name(ec2_client, region, name): IpPermissions=[{'IpProtocol': 'tcp', 'FromPort': 0, 'ToPort': 65535, - 'UserIdGroupPairs': [{'GroupId': security_group_id}]}]) + 'UserIdGroupPairs': [{ + 'GroupId': security_group_id}]}]) ec2_client.create_tags(Resources=[vpc_id, subnet_id_a, subnet_id_b, security_group_id], Tags=[{'Key': 'Name', 'Value': name}]) @@ -83,23 +93,27 @@ def _create_vpc_with_name(ec2_client, region, name): def get_or_create_vpc_resources(ec2_client, region, name=VPC_NAME): - if _vpc_exists(ec2_client, name): - print('using existing vpc: {}'.format(name)) - return _get_subnet_ids_by_name(ec2_client, name), _get_security_id_by_name(ec2_client, name) - else: - print('creating new vpc: {}'.format(name)) - return _create_vpc_with_name(ec2_client, region, name) + with lock.lock(LOCK_PATH): + if _vpc_exists(ec2_client, name): + print('using existing vpc: {}'.format(name)) + return _get_subnet_ids_by_name(ec2_client, name), _get_security_id_by_name(ec2_client, + name) + else: + print('creating new vpc: {}'.format(name)) + return _create_vpc_with_name(ec2_client, region, name) def setup_security_group_for_encryption(ec2_client, security_group_id): sg_desc = ec2_client.describe_security_groups(GroupIds=[security_group_id]) ingress_perms = sg_desc['SecurityGroups'][0]['IpPermissions'] if len(ingress_perms) == 1: - ec2_client.\ + ec2_client. \ authorize_security_group_ingress(GroupId=security_group_id, IpPermissions=[{'IpProtocol': '50', - 'UserIdGroupPairs': [{'GroupId': security_group_id}]}, + 'UserIdGroupPairs': [ + {'GroupId': security_group_id}]}, {'IpProtocol': 'udp', 'FromPort': 500, 'ToPort': 500, - 'UserIdGroupPairs': [{'GroupId': security_group_id}]}]) + 'UserIdGroupPairs': [ + {'GroupId': security_group_id}]}]) From 0bdad3a03693f2b23b250c4239c67514763a8787 Mon Sep 17 00:00:00 2001 From: jesterhazy Date: Tue, 18 Jun 2019 17:41:30 -0700 Subject: [PATCH 2/2] remove extra files --- .gitignore | 2 +- tests/integ/lock.py | 4 ++++ tests/integ/test_local_mode.py | 1 + tests/integ/test_source_dirs.py | 1 + tests/integ/vpc_test_utils.py | 1 + 5 files changed, 8 insertions(+), 1 deletion(-) diff --git a/.gitignore b/.gitignore index 863e0e9595..1d664ef172 100644 --- a/.gitignore +++ b/.gitignore @@ -25,4 +25,4 @@ venv/ *~ .pytest_cache/ *.swp -tests/data/local_mode_lock +.docker/ \ No newline at end of file diff --git a/tests/integ/lock.py b/tests/integ/lock.py index bea14350e0..07d651d2b1 100644 --- a/tests/integ/lock.py +++ b/tests/integ/lock.py @@ -23,6 +23,10 @@ @contextmanager def lock(path=DEFAULT_LOCK_PATH): + """Create a file lock to control concurrent test execution. Certain tests or + test operations need to limit concurrency to work reliably. Examples include + local mode endpoint tests and vpc creation tests. + """ f = open(path, 'w') fd = f.fileno() diff --git a/tests/integ/test_local_mode.py b/tests/integ/test_local_mode.py index 78d875e557..e6b3ef2da8 100644 --- a/tests/integ/test_local_mode.py +++ b/tests/integ/test_local_mode.py @@ -28,6 +28,7 @@ from sagemaker.mxnet import MXNet from sagemaker.tensorflow import TensorFlow +# endpoint tests all use the same port, so we use this lock to prevent concurrent execution LOCK_PATH = os.path.join(tempfile.gettempdir(), 'sagemaker_test_local_mode_lock') DATA_PATH = os.path.join(DATA_DIR, 'iris', 'data') DEFAULT_REGION = 'us-west-2' diff --git a/tests/integ/test_source_dirs.py b/tests/integ/test_source_dirs.py index fef75760ba..c018c6ca9f 100644 --- a/tests/integ/test_source_dirs.py +++ b/tests/integ/test_source_dirs.py @@ -37,6 +37,7 @@ def test_source_dirs(tmpdir, sagemaker_local_session): sagemaker_session=sagemaker_local_session) estimator.fit() + # endpoint tests all use the same port, so we use this lock to prevent concurrent execution with lock.lock(): try: predictor = estimator.deploy(initial_instance_count=1, instance_type='local') diff --git a/tests/integ/vpc_test_utils.py b/tests/integ/vpc_test_utils.py index 3cccc4630d..f381e2b068 100644 --- a/tests/integ/vpc_test_utils.py +++ b/tests/integ/vpc_test_utils.py @@ -93,6 +93,7 @@ def _create_vpc_with_name(ec2_client, region, name): def get_or_create_vpc_resources(ec2_client, region, name=VPC_NAME): + # use lock to prevent race condition when tests are running concurrently with lock.lock(LOCK_PATH): if _vpc_exists(ec2_client, name): print('using existing vpc: {}'.format(name))