1 # -*- coding: utf-8 -*- 2 # Copyright 2013 Google Inc. All Rights Reserved. 3 # 4 # Licensed under the Apache License, Version 2.0 (the "License"); 5 # you may not use this file except in compliance with the License. 6 # You may obtain a copy of the License at 7 # 8 # http://www.apache.org/licenses/LICENSE-2.0 9 # 10 # Unless required by applicable law or agreed to in writing, software 11 # distributed under the License is distributed on an "AS IS" BASIS, 12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 13 # See the License for the specific language governing permissions and 14 # limitations under the License. 15 """Integration tests for perfdiag command.""" 16 17 from __future__ import absolute_import 18 19 import os 20 import socket 21 22 import boto 23 24 import gslib.tests.testcase as testcase 25 from gslib.tests.util import ObjectToURI as suri 26 from gslib.tests.util import RUN_S3_TESTS 27 from gslib.tests.util import unittest 28 from gslib.util import IS_WINDOWS 29 30 31 class TestPerfDiag(testcase.GsUtilIntegrationTestCase): 32 """Integration tests for perfdiag command.""" 33 34 # We want to test that perfdiag works both when connecting to the standard gs 35 # endpoint, and when connecting to a specific IP or host while setting the 36 # host header. For the 2nd case we resolve gs_host (normally 37 # storage.googleapis.com) to a specific IP and connect to that explicitly. 38 _gs_host = boto.config.get( 39 'Credentials', 'gs_host', boto.gs.connection.GSConnection.DefaultHost) 40 _gs_ip = socket.gethostbyname(_gs_host) 41 _custom_endpoint_flags = [ 42 '-o', 'Credentials:gs_host=' + _gs_ip, 43 '-o', 'Credentials:gs_host_header=' + _gs_host, 44 # TODO: gsutil-beta: Add host header support for JSON 45 '-o', 'Boto:https_validate_certificates=False'] 46 47 def _should_run_with_custom_endpoints(self): 48 # Host headers are only supported for XML, and not when 49 # using environment variables for proxies. 50 return (self.test_api == 'XML' and not RUN_S3_TESTS and not 51 (os.environ.get('http_proxy') or os.environ.get('https_proxy') or 52 os.environ.get('HTTPS_PROXY'))) 53 54 def test_latency(self): 55 bucket_uri = self.CreateBucket() 56 cmd = ['perfdiag', '-n', '1', '-t', 'lat', suri(bucket_uri)] 57 self.RunGsUtil(cmd) 58 if self._should_run_with_custom_endpoints(): 59 self.RunGsUtil(self._custom_endpoint_flags + cmd) 60 self.AssertNObjectsInBucket(bucket_uri, 0, versioned=True) 61 62 def _run_throughput_test(self, test_name, num_processes, num_threads, 63 parallelism_strategy=None): 64 bucket_uri = self.CreateBucket() 65 66 cmd = ['perfdiag', '-n', str(num_processes * num_threads), 67 '-s', '1024', '-c', str(num_processes), '-k', str(num_threads), 68 '-t', test_name] 69 if parallelism_strategy: 70 cmd += ['-p', parallelism_strategy] 71 cmd += [suri(bucket_uri)] 72 73 self.RunGsUtil(cmd) 74 if self._should_run_with_custom_endpoints(): 75 self.RunGsUtil(self._custom_endpoint_flags + cmd) 76 self.AssertNObjectsInBucket(bucket_uri, 0, versioned=True) 77 78 def _run_each_parallel_throughput_test(self, test_name, num_processes, 79 num_threads): 80 self._run_throughput_test(test_name, num_processes, num_threads, 'fan') 81 if not RUN_S3_TESTS: 82 self._run_throughput_test(test_name, num_processes, num_threads, 'slice') 83 self._run_throughput_test(test_name, num_processes, num_threads, 'both') 84 85 def test_write_throughput_single_process_single_thread(self): 86 self._run_throughput_test('wthru', 1, 1) 87 self._run_throughput_test('wthru_file', 1, 1) 88 89 def test_write_throughput_single_process_multi_thread(self): 90 self._run_each_parallel_throughput_test('wthru', 1, 2) 91 self._run_each_parallel_throughput_test('wthru_file', 1, 2) 92 93 @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') 94 def test_write_throughput_multi_process_single_thread(self): 95 self._run_each_parallel_throughput_test('wthru', 2, 1) 96 self._run_each_parallel_throughput_test('wthru_file', 2, 1) 97 98 @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') 99 def test_write_throughput_multi_process_multi_thread(self): 100 self._run_each_parallel_throughput_test('wthru', 2, 2) 101 self._run_each_parallel_throughput_test('wthru_file', 2, 2) 102 103 def test_read_throughput_single_process_single_thread(self): 104 self._run_throughput_test('rthru', 1, 1) 105 self._run_throughput_test('rthru_file', 1, 1) 106 107 def test_read_throughput_single_process_multi_thread(self): 108 self._run_each_parallel_throughput_test('rthru', 1, 2) 109 self._run_each_parallel_throughput_test('rthru_file', 1, 2) 110 111 @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') 112 def test_read_throughput_multi_process_single_thread(self): 113 self._run_each_parallel_throughput_test('rthru', 2, 1) 114 self._run_each_parallel_throughput_test('rthru_file', 2, 1) 115 116 @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') 117 def test_read_throughput_multi_process_multi_thread(self): 118 self._run_each_parallel_throughput_test('rthru', 2, 2) 119 self._run_each_parallel_throughput_test('rthru_file', 2, 2) 120 121 @unittest.skipIf(IS_WINDOWS, 'Multiprocessing is not supported on Windows') 122 def test_read_and_write_file_ordering(self): 123 """Tests that rthru_file and wthru_file work when run together.""" 124 self._run_throughput_test('rthru_file,wthru_file', 1, 1) 125 self._run_throughput_test('rthru_file,wthru_file', 2, 2, 'fan') 126 if not RUN_S3_TESTS: 127 self._run_throughput_test('rthru_file,wthru_file', 2, 2, 'slice') 128 self._run_throughput_test('rthru_file,wthru_file', 2, 2, 'both') 129 130 def test_input_output(self): 131 outpath = self.CreateTempFile() 132 bucket_uri = self.CreateBucket() 133 self.RunGsUtil(['perfdiag', '-o', outpath, '-n', '1', '-t', 'lat', 134 suri(bucket_uri)]) 135 self.RunGsUtil(['perfdiag', '-i', outpath]) 136 137 def test_invalid_size(self): 138 stderr = self.RunGsUtil( 139 ['perfdiag', '-n', '1', '-s', 'foo', '-t', 'wthru', 'gs://foobar'], 140 expected_status=1, return_stderr=True) 141 self.assertIn('Invalid -s', stderr) 142 143 def test_toobig_size(self): 144 stderr = self.RunGsUtil( 145 ['perfdiag', '-n', '1', '-s', '3pb', '-t', 'wthru', 'gs://foobar'], 146 expected_status=1, return_stderr=True) 147 self.assertIn('in-memory tests maximum file size', stderr) 148 149 def test_listing(self): 150 bucket_uri = self.CreateBucket() 151 stdout = self.RunGsUtil( 152 ['perfdiag', '-n', '1', '-t', 'list', suri(bucket_uri)], 153 return_stdout=True) 154 self.assertIn('Number of listing calls made:', stdout) 155 self.AssertNObjectsInBucket(bucket_uri, 0, versioned=True) 156
