2853: Merge branch 'master' into 2853-rendezvous
[arvados.git] / sdk / python / tests / test_keep_client.py
index b12ae77fe47da507d843f1f276478cc1681b3981..bb0022eb432628dbf0c21b9b9fc2bd20a73f1fb1 100644 (file)
@@ -1,5 +1,7 @@
+import hashlib
 import mock
 import os
+import re
 import socket
 import unittest
 import urlparse
@@ -9,6 +11,84 @@ import arvados.retry
 import arvados_testutil as tutil
 import run_test_server
 
+class KeepRendezvousWeightTestCase(unittest.TestCase):
+    def setUp(self):
+        self.keep_client = arvados.KeepClient(
+            api_client=mock.MagicMock(name='api_client'),
+            proxy='', local_store='')
+        self.keep_client._keep_services = []
+        self.n_services = 0
+
+    def addServices(self, n):
+        for x in range(n):
+            uuid = "zzzzz-bi6l4-{:015x}".format(self.n_services)
+            uri = "https://keep0x{:x}.zzzzz.arvadosapi.com:25107/".format(self.n_services)
+            self.keep_client._keep_services.append(
+                {'uuid': uuid, '_service_root': uri})
+            self.n_services += 1
+
+    def test_ProbeOrderReferenceSet(self):
+        # expected_order[i] is the probe order for
+        # hash=md5(sprintf("%064x",i)) where there are 16 services
+        # with uuid sprintf("anything-%015x",j) with j in 0..15. E.g.,
+        # the first probe for the block consisting of 64 "0"
+        # characters is the service whose uuid is
+        # "zzzzz-bi6l4-000000000000003", so expected_order[0][0]=='3'.
+        expected_order = [
+            list('3eab2d5fc9681074'),
+            list('097dba52e648f1c3'),
+            list('c5b4e023f8a7d691'),
+            list('9d81c02e76a3bf54'),
+            ]
+        hashes = [
+            hashlib.md5("{:064x}".format(x)).hexdigest()
+            for x in range(len(expected_order))]
+        self.addServices(16)
+        for i, hash in enumerate(hashes):
+            roots = self.keep_client.weighted_service_roots(hash)
+            got_order = [
+                re.search(r'//keep0x([0-9a-f]+)', root).group(1)
+                for root in roots]
+            self.assertEqual(got_order, expected_order[i])
+
+    def test_ProbeWasteAddingOneServer(self):
+        hashes = [
+            hashlib.md5("{:064x}".format(x)).hexdigest() for x in range(100)]
+        initial_services = 12
+        self.addServices(initial_services)
+        probes_before = [
+            self.keep_client.weighted_service_roots(hash) for hash in hashes]
+        for added_services in range(1, 12):
+            self.addServices(1)
+            total_penalty = 0
+            for hash_index in range(len(hashes)):
+                probe_after = self.keep_client.weighted_service_roots(
+                    hashes[hash_index])
+                penalty = probe_after.index(probes_before[hash_index][0])
+                self.assertLessEqual(penalty, added_services)
+                total_penalty += penalty
+            # Average penalty per block should not exceed
+            # N(added)/N(orig) by more than 20%, and should get closer
+            # to the ideal as we add data points.
+            expect_penalty = (
+                added_services *
+                len(hashes) / initial_services)
+            max_penalty = (
+                expect_penalty *
+                (120 - added_services)/100)
+            min_penalty = (
+                expect_penalty * 8/10)
+            self.assertTrue(
+                min_penalty <= total_penalty <= max_penalty,
+                "With {}+{} services, {} blocks, penalty {} but expected {}..{}".format(
+                    initial_services,
+                    added_services,
+                    len(hashes),
+                    total_penalty,
+                    min_penalty,
+                    max_penalty))
+
+
 class KeepTestCase(run_test_server.TestCaseWithServers):
     MAIN_SERVER = {}
     KEEP_SERVER = {}
@@ -66,6 +146,13 @@ class KeepTestCase(run_test_server.TestCaseWithServers):
                          blob_str,
                          'wrong content from Keep.get(md5(<binarydata>))')
 
+    def test_KeepEmptyCollectionTest(self):
+        blob_locator = self.keep_client.put('', copies=1)
+        self.assertRegexpMatches(
+            blob_locator,
+            '^d41d8cd98f00b204e9800998ecf8427e\+0',
+            ('wrong locator from Keep.put(""): ' + blob_locator))
+
 
 class KeepPermissionTestCase(run_test_server.TestCaseWithServers):
     MAIN_SERVER = {}
@@ -242,7 +329,7 @@ class KeepClientServiceTestCase(unittest.TestCase):
     def get_service_roots(self, *services):
         api_client = self.mock_keep_services(*services)
         keep_client = arvados.KeepClient(api_client=api_client)
-        services = keep_client.shuffled_service_roots('000000')
+        services = keep_client.weighted_service_roots('000000')
         return [urlparse.urlparse(url) for url in sorted(services)]
 
     def test_ssl_flag_respected_in_roots(self):
@@ -258,6 +345,64 @@ class KeepClientServiceTestCase(unittest.TestCase):
         self.assertEqual('100::1', service.hostname)
         self.assertEqual(10, service.port)
 
+    # test_get_timeout and test_put_timeout test that
+    # KeepClient.get and KeepClient.put use the appropriate timeouts
+    # when connected directly to a Keep server (i.e. non-proxy timeout)
+
+    def test_get_timeout(self):
+        api_client = self.mock_keep_services(('keep', 10, False, 'disk'))
+        keep_client = arvados.KeepClient(api_client=api_client)
+        force_timeout = [socket.timeout("timed out")]
+        with mock.patch('requests.get', side_effect=force_timeout) as mock_request:
+            with self.assertRaises(arvados.errors.KeepReadError):
+                keep_client.get('ffffffffffffffffffffffffffffffff')
+            self.assertTrue(mock_request.called)
+            self.assertEqual(
+                arvados.KeepClient.DEFAULT_TIMEOUT,
+                mock_request.call_args[1]['timeout'])
+
+    def test_put_timeout(self):
+        api_client = self.mock_keep_services(('keep', 10, False, 'disk'))
+        keep_client = arvados.KeepClient(api_client=api_client)
+        force_timeout = [socket.timeout("timed out")]
+        with mock.patch('requests.put', side_effect=force_timeout) as mock_request:
+            with self.assertRaises(arvados.errors.KeepWriteError):
+                keep_client.put('foo')
+            self.assertTrue(mock_request.called)
+            self.assertEqual(
+                arvados.KeepClient.DEFAULT_TIMEOUT,
+                mock_request.call_args[1]['timeout'])
+
+    def test_proxy_get_timeout(self):
+        # Force a timeout, verifying that the requests.get or
+        # requests.put method was called with the proxy_timeout
+        # setting rather than the default timeout.
+        api_client = self.mock_keep_services(('keep', 10, False, 'proxy'))
+        keep_client = arvados.KeepClient(api_client=api_client)
+        force_timeout = [socket.timeout("timed out")]
+        with mock.patch('requests.get', side_effect=force_timeout) as mock_request:
+            with self.assertRaises(arvados.errors.KeepReadError):
+                keep_client.get('ffffffffffffffffffffffffffffffff')
+            self.assertTrue(mock_request.called)
+            self.assertEqual(
+                arvados.KeepClient.DEFAULT_PROXY_TIMEOUT,
+                mock_request.call_args[1]['timeout'])
+
+    def test_proxy_put_timeout(self):
+        # Force a timeout, verifying that the requests.get or
+        # requests.put method was called with the proxy_timeout
+        # setting rather than the default timeout.
+        api_client = self.mock_keep_services(('keep', 10, False, 'proxy'))
+        keep_client = arvados.KeepClient(api_client=api_client)
+        force_timeout = [socket.timeout("timed out")]
+        with mock.patch('requests.put', side_effect=force_timeout) as mock_request:
+            with self.assertRaises(arvados.errors.KeepWriteError):
+                keep_client.put('foo')
+            self.assertTrue(mock_request.called)
+            self.assertEqual(
+                arvados.KeepClient.DEFAULT_PROXY_TIMEOUT,
+                mock_request.call_args[1]['timeout'])
+
 
 class KeepClientRetryTestMixin(object):
     # Testing with a local Keep store won't exercise the retry behavior.
@@ -270,6 +415,10 @@ class KeepClientRetryTestMixin(object):
     # supporting servers, and prevents side effects in case something hiccups.
     # To use this mixin, define DEFAULT_EXPECT, DEFAULT_EXCEPTION, and
     # run_method().
+    #
+    # Test classes must define TEST_PATCHER to a method that mocks
+    # out appropriate methods in the client.
+
     PROXY_ADDR = 'http://[%s]:65535/' % (tutil.TEST_HOST,)
     TEST_DATA = 'testdata'
     TEST_LOCATOR = 'ef654c40ab4f1747fc699915d4f70902+8'
@@ -296,28 +445,28 @@ class KeepClientRetryTestMixin(object):
         self.assertRaises(error_class, self.run_method, *args, **kwargs)
 
     def test_immediate_success(self):
-        with tutil.mock_responses(self.DEFAULT_EXPECT, 200):
+        with self.TEST_PATCHER(self.DEFAULT_EXPECT, 200):
             self.check_success()
 
     def test_retry_then_success(self):
-        with tutil.mock_responses(self.DEFAULT_EXPECT, 500, 200):
+        with self.TEST_PATCHER(self.DEFAULT_EXPECT, 500, 200):
             self.check_success(num_retries=3)
 
     def test_no_default_retry(self):
-        with tutil.mock_responses(self.DEFAULT_EXPECT, 500, 200):
+        with self.TEST_PATCHER(self.DEFAULT_EXPECT, 500, 200):
             self.check_exception()
 
     def test_no_retry_after_permanent_error(self):
-        with tutil.mock_responses(self.DEFAULT_EXPECT, 403, 200):
+        with self.TEST_PATCHER(self.DEFAULT_EXPECT, 403, 200):
             self.check_exception(num_retries=3)
 
     def test_error_after_retries_exhausted(self):
-        with tutil.mock_responses(self.DEFAULT_EXPECT, 500, 500, 200):
+        with self.TEST_PATCHER(self.DEFAULT_EXPECT, 500, 500, 200):
             self.check_exception(num_retries=1)
 
     def test_num_retries_instance_fallback(self):
         self.client_kwargs['num_retries'] = 3
-        with tutil.mock_responses(self.DEFAULT_EXPECT, 500, 200):
+        with self.TEST_PATCHER(self.DEFAULT_EXPECT, 500, 200):
             self.check_success()
 
 
@@ -326,13 +475,14 @@ class KeepClientRetryGetTestCase(KeepClientRetryTestMixin, unittest.TestCase):
     DEFAULT_EXPECT = KeepClientRetryTestMixin.TEST_DATA
     DEFAULT_EXCEPTION = arvados.errors.KeepReadError
     HINTED_LOCATOR = KeepClientRetryTestMixin.TEST_LOCATOR + '+K@xyzzy'
+    TEST_PATCHER = staticmethod(tutil.mock_get_responses)
 
     def run_method(self, locator=KeepClientRetryTestMixin.TEST_LOCATOR,
                    *args, **kwargs):
         return self.new_client().get(locator, *args, **kwargs)
 
     def test_specific_exception_when_not_found(self):
-        with tutil.mock_responses(self.DEFAULT_EXPECT, 404, 200):
+        with tutil.mock_get_responses(self.DEFAULT_EXPECT, 404, 200):
             self.check_exception(arvados.errors.NotFoundError, num_retries=3)
 
     def test_general_exception_with_mixed_errors(self):
@@ -341,7 +491,7 @@ class KeepClientRetryGetTestCase(KeepClientRetryTestMixin, unittest.TestCase):
         # This test rigs up 50/50 disagreement between two servers, and
         # checks that it does not become a NotFoundError.
         client = self.new_client()
-        with tutil.mock_responses(self.DEFAULT_EXPECT, 404, 500):
+        with tutil.mock_get_responses(self.DEFAULT_EXPECT, 404, 500):
             with self.assertRaises(arvados.errors.KeepReadError) as exc_check:
                 client.get(self.HINTED_LOCATOR)
             self.assertNotIsInstance(
@@ -349,21 +499,21 @@ class KeepClientRetryGetTestCase(KeepClientRetryTestMixin, unittest.TestCase):
                 "mixed errors raised NotFoundError")
 
     def test_hint_server_can_succeed_without_retries(self):
-        with tutil.mock_responses(self.DEFAULT_EXPECT, 404, 200, 500):
+        with tutil.mock_get_responses(self.DEFAULT_EXPECT, 404, 200, 500):
             self.check_success(locator=self.HINTED_LOCATOR)
 
     def test_try_next_server_after_timeout(self):
         side_effects = [
             socket.timeout("timed out"),
-            (tutil.fake_httplib2_response(200), self.DEFAULT_EXPECT)]
-        with mock.patch('httplib2.Http.request',
+            tutil.fake_requests_response(200, self.DEFAULT_EXPECT)]
+        with mock.patch('requests.get',
                         side_effect=iter(side_effects)):
             self.check_success(locator=self.HINTED_LOCATOR)
 
     def test_retry_data_with_wrong_checksum(self):
-        side_effects = ((tutil.fake_httplib2_response(200), s)
+        side_effects = (tutil.fake_requests_response(200, s)
                         for s in ['baddata', self.TEST_DATA])
-        with mock.patch('httplib2.Http.request', side_effect=side_effects):
+        with mock.patch('requests.get', side_effect=side_effects):
             self.check_success(locator=self.HINTED_LOCATOR)
 
 
@@ -371,11 +521,12 @@ class KeepClientRetryGetTestCase(KeepClientRetryTestMixin, unittest.TestCase):
 class KeepClientRetryPutTestCase(KeepClientRetryTestMixin, unittest.TestCase):
     DEFAULT_EXPECT = KeepClientRetryTestMixin.TEST_LOCATOR
     DEFAULT_EXCEPTION = arvados.errors.KeepWriteError
+    TEST_PATCHER = staticmethod(tutil.mock_put_responses)
 
     def run_method(self, data=KeepClientRetryTestMixin.TEST_DATA,
                    copies=1, *args, **kwargs):
         return self.new_client().put(data, copies, *args, **kwargs)
 
     def test_do_not_send_multiple_copies_to_same_server(self):
-        with tutil.mock_responses(self.DEFAULT_EXPECT, 200):
+        with tutil.mock_put_responses(self.DEFAULT_EXPECT, 200):
             self.check_exception(copies=2, num_retries=3)