1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
|
# The MIT License (MIT)
# Copyright (c) Microsoft Corporation. All rights reserved.
import os
import unittest
import uuid
import pytest
from azure.core.exceptions import ServiceResponseError
import test_config
from azure.cosmos import _partition_health_tracker, _location_cache
from azure.cosmos import CosmosClient
from azure.cosmos.exceptions import CosmosHttpResponseError
from _fault_injection_transport import FaultInjectionTransport
from test_per_partition_circuit_breaker_mm import perform_write_operation, perform_read_operation
from test_per_partition_circuit_breaker_mm_async import create_doc, PK_VALUE, write_operations_and_errors, \
operations, REGION_2, REGION_1
from test_per_partition_circuit_breaker_sm_mrr_async import validate_unhealthy_partitions
COLLECTION = "created_collection"
@pytest.mark.cosmosCircuitBreakerMultiRegion
class TestPerPartitionCircuitBreakerSmMrr:
host = test_config.TestConfig.host
master_key = test_config.TestConfig.masterKey
connectionPolicy = test_config.TestConfig.connectionPolicy
TEST_DATABASE_ID = test_config.TestConfig.TEST_DATABASE_ID
TEST_CONTAINER_SINGLE_PARTITION_ID = test_config.TestConfig.TEST_MULTI_PARTITION_CONTAINER_ID
def setup_method_with_custom_transport(self, custom_transport, default_endpoint=host, **kwargs):
client = CosmosClient(default_endpoint, self.master_key, consistency_level="Session",
preferred_locations=[REGION_1, REGION_2],
transport=custom_transport, **kwargs)
db = client.get_database_client(self.TEST_DATABASE_ID)
container = db.get_container_client(self.TEST_CONTAINER_SINGLE_PARTITION_ID)
return {"client": client, "db": db, "col": container}
def setup_info(self, error):
expected_uri = _location_cache.LocationCache.GetLocationalEndpoint(self.host, REGION_2)
uri_down = _location_cache.LocationCache.GetLocationalEndpoint(self.host, REGION_1)
custom_transport = FaultInjectionTransport()
# two documents targeted to same partition, one will always fail and the other will succeed
doc = create_doc()
predicate = lambda r: (FaultInjectionTransport.predicate_is_document_operation(r) and
FaultInjectionTransport.predicate_targets_region(r, uri_down))
custom_transport.add_fault(predicate,
error)
custom_setup = self.setup_method_with_custom_transport(custom_transport, default_endpoint=self.host)
setup = self.setup_method_with_custom_transport(None, default_endpoint=self.host)
return setup, doc, expected_uri, uri_down, custom_setup, custom_transport, predicate
@pytest.mark.parametrize("write_operation, error", write_operations_and_errors())
def test_write_consecutive_failure_threshold(self, write_operation, error):
error_lambda = lambda r: FaultInjectionTransport.error_after_delay(0, error)
setup, doc, expected_uri, uri_down, custom_setup, custom_transport, predicate = self.setup_info(error_lambda)
container = setup['col']
fault_injection_container = custom_setup['col']
global_endpoint_manager = container.client_connection._global_endpoint_manager
# writes should fail in sm mrr with circuit breaker and should not mark unavailable a partition
for i in range(6):
validate_unhealthy_partitions(global_endpoint_manager, 0)
with pytest.raises((CosmosHttpResponseError, ServiceResponseError)):
perform_write_operation(
write_operation,
container,
fault_injection_container,
str(uuid.uuid4()),
PK_VALUE,
expected_uri,
)
validate_unhealthy_partitions(global_endpoint_manager, 0)
@pytest.mark.parametrize("write_operation, error", write_operations_and_errors())
def test_write_failure_rate_threshold(self, write_operation, error):
error_lambda = lambda r: FaultInjectionTransport.error_after_delay(0, error)
setup, doc, expected_uri, uri_down, custom_setup, custom_transport, predicate = self.setup_info(error_lambda)
container = setup['col']
fault_injection_container = custom_setup['col']
global_endpoint_manager = fault_injection_container.client_connection._global_endpoint_manager
# lower minimum requests for testing
_partition_health_tracker.MINIMUM_REQUESTS_FOR_FAILURE_RATE = 10
os.environ["AZURE_COSMOS_FAILURE_PERCENTAGE_TOLERATED"] = "80"
try:
# writes should fail but still be tracked and mark unavailable a partition after crossing threshold
for i in range(10):
validate_unhealthy_partitions(global_endpoint_manager, 0)
if i == 4 or i == 8:
# perform some successful creates to reset consecutive counter
# remove faults and perform a write
custom_transport.faults = []
fault_injection_container.upsert_item(body=doc)
custom_transport.add_fault(predicate,
lambda r: FaultInjectionTransport.error_after_delay(
0,
error
))
else:
with pytest.raises((CosmosHttpResponseError, ServiceResponseError)) as exc_info:
perform_write_operation(write_operation,
container,
fault_injection_container,
str(uuid.uuid4()),
PK_VALUE,
expected_uri)
assert exc_info.value == error
validate_unhealthy_partitions(global_endpoint_manager, 0)
finally:
os.environ["AZURE_COSMOS_FAILURE_PERCENTAGE_TOLERATED"] = "90"
# restore minimum requests
_partition_health_tracker.MINIMUM_REQUESTS_FOR_FAILURE_RATE = 100
@pytest.mark.parametrize("read_operation, write_operation", operations())
def test_service_request_error(self, read_operation, write_operation):
# the region should be tried 4 times before failing over and mark the partition as unavailable
# the region should not be marked as unavailable
error_lambda = lambda r: FaultInjectionTransport.error_region_down()
setup, doc, expected_uri, uri_down, custom_setup, custom_transport, predicate = self.setup_info(error_lambda)
container = setup['col']
fault_injection_container = custom_setup['col']
container.upsert_item(body=doc)
perform_read_operation(read_operation,
fault_injection_container,
doc['id'],
PK_VALUE,
expected_uri)
global_endpoint_manager = fault_injection_container.client_connection._global_endpoint_manager
# there shouldn't be partition marked as unavailable
validate_unhealthy_partitions(global_endpoint_manager, 0)
assert len(global_endpoint_manager.location_cache.location_unavailability_info_by_endpoint) == 1
# recover partition
# remove faults and reduce initial recover time and perform a write
original_unavailable_time = _partition_health_tracker.INITIAL_UNAVAILABLE_TIME
_partition_health_tracker.INITIAL_UNAVAILABLE_TIME = 1
custom_transport.faults = []
try:
perform_read_operation(read_operation,
fault_injection_container,
doc['id'],
PK_VALUE,
expected_uri)
finally:
_partition_health_tracker.INITIAL_UNAVAILABLE_TIME = original_unavailable_time
validate_unhealthy_partitions(global_endpoint_manager, 0)
custom_transport.add_fault(predicate,
lambda r: FaultInjectionTransport.error_region_down())
# The global endpoint would be used for the write operation
expected_uri = self.host
perform_write_operation(write_operation,
container,
fault_injection_container,
str(uuid.uuid4()),
PK_VALUE,
expected_uri)
global_endpoint_manager = fault_injection_container.client_connection._global_endpoint_manager
validate_unhealthy_partitions(global_endpoint_manager, 0)
# there shouldn't be region marked as unavailable
assert len(global_endpoint_manager.location_cache.location_unavailability_info_by_endpoint) == 1
# test cosmos client timeout
if __name__ == '__main__':
unittest.main()
|
