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
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
|
import time
from yaml import dump
from twisted.internet.defer import inlineCallbacks, succeed, fail
from txaws.ec2.model import Instance
from txaws.ec2.exception import EC2Error
import zookeeper
from txzookeeper import ZookeeperClient
from txzookeeper.tests.utils import deleteTree
from ensemble.lib.testing import TestCase
from ensemble.errors import (
EnvironmentNotFound, EnvironmentPending, NoConnection)
from ensemble.state.sshclient import SSHClient
from ensemble.providers.ec2.tests.common import EC2TestMixin
from ensemble.providers.ec2.connect import EC2Connect
from ensemble.errors import ProviderInteractionError
class EC2ConnectTest(EC2TestMixin, TestCase):
def test_connect_no_state(self):
"""
When loading saved state from S3, the provider connect gracefully
handles the scenario where there is no saved state.
"""
self.s3.get_object(self.env_name, "provider-state")
self.mocker.result(succeed(dump([])))
self.mocker.replay()
provider = self.get_provider()
connect = EC2Connect(provider)
d = connect.run()
self.failUnlessFailure(d, EnvironmentNotFound)
return d
def test_connect_state_no_hosts(self):
"""
If the saved state from s3 exists, but has no zookeeper hosts,
the provider connect correctly detects this and raises
EnvironmentNotFound.
"""
self.s3.get_object(self.env_name, "provider-state")
self.mocker.result(succeed(dump({"abc": 123})))
self.mocker.replay()
provider = self.get_provider()
connect = EC2Connect(provider)
d = connect.run()
self.failUnlessFailure(d, EnvironmentNotFound)
return d
def test_connect_not_running(self):
"""
The bootstrap load of the saved instance state, will attempt to
validate and filter the saved state to only return running instances.
If there are no running instances it returns False.
"""
self.s3.get_object(self.env_name, "provider-state")
self.mocker.result(succeed(dump(
{"zookeeper-instances": ["i-x"]})))
self.ec2.describe_instances(*["i-x"])
self.mocker.result(succeed([]))
self.mocker.replay()
provider = self.get_provider()
connect = EC2Connect(provider)
d = connect.run()
self.failUnlessFailure(d, EnvironmentNotFound)
return d
def test_provider_connect(self):
"""
Test connecting to a running ensemble zookeeper server within ec2.
"""
server1 = Instance("i-foobar", "pending", dns_name="x1.example.com")
server2 = Instance("i-foobaz", "running", dns_name="x2.example.com")
self.s3.get_object(self.env_name, "provider-state")
self.mocker.result(succeed(dump(
{"zookeeper-instances": [server1.instance_id,
server2.instance_id]})))
# Zk instances are checked individually to handle invalid ids correctly
self.ec2.describe_instances(server1.instance_id)
self.mocker.result(succeed([server1]))
self.ec2.describe_instances(server2.instance_id)
self.mocker.result(succeed([server2]))
client = self.mocker.patch(SSHClient)
connected_client = self.mocker.mock(type=SSHClient)
client.connect(server2.dns_name + ":2181", timeout=30)
self.mocker.result(succeed(connected_client))
# We'll test the wait on initialization separately.
connected_client.exists_and_watch("/initialized")
self.mocker.result((succeed(True), None))
self.mocker.replay()
provider = self.get_provider()
d = provider.connect()
def verify_result(result):
self.assertIdentical(connected_client, result)
d.addCallback(verify_result)
return d
def test_provider_connect_with_invalid_instance_id(self):
"""
Using an invalid instance id, still gets the appropriate ensemble error
"""
server1 = Instance("i-foobar", "pending", dns_name="x1.example.com")
self.s3.get_object(self.env_name, "provider-state")
self.mocker.result(succeed(dump(
{"zookeeper-instances": [server1.instance_id]})))
invalid_id_error = EC2Error("<error/>", 400)
invalid_id_error.errors = [{"Code": "InvalidInstanceID.NotFound"}]
self.ec2.describe_instances(server1.instance_id)
self.mocker.result(fail(invalid_id_error))
self.mocker.replay()
provider = self.get_provider()
d = provider.connect()
return self.assertFailure(d, EnvironmentNotFound)
@inlineCallbacks
def test_provider_connect_waits_on_initialization(self):
"""
Test connecting to a running ensemble zookeeper server within ec2.
"""
server = Instance("i-foobar", "running", dns_name="x1.example.com")
self.s3.get_object(self.env_name, "provider-state")
self.mocker.result(succeed(dump(
{"zookeeper-instances": [server.instance_id]})))
self.ec2.describe_instances(server.instance_id)
self.mocker.result(succeed([server]))
# Hand back a real connected client to test the wait on initialization.
connected_client = ZookeeperClient()
client = self.mocker.patch(SSHClient)
client.connect(server.dns_name + ":2181", timeout=30)
self.mocker.result(succeed(connected_client))
self.mocker.replay()
zookeeper.set_debug_level(0)
yield connected_client.connect("localhost:2181")
client_result = []
provider = self.get_provider()
client_deferred = provider.connect()
client_deferred.addCallback(client_result.append)
# Give it some time to do it incorrectly.
from twisted.internet.defer import Deferred
from twisted.internet import reactor
sleep_deferred = Deferred()
reactor.callLater(0.1, sleep_deferred.callback, None)
yield sleep_deferred
# XXX Replace the logic above with self.sleep(0.1) when it's merged.
try:
self.assertEquals(client_result, [])
yield connected_client.create("/initialized")
yield client_deferred
self.assertTrue(client_result, client_result)
self.assertIdentical(client_result[0], connected_client)
finally:
deleteTree("/", connected_client.handle)
connected_client.close()
def test_provider_connect_with_pending(self):
"""
If there are only pending machines, it should provide a reasonable
error message in return.
"""
server = Instance("i-foobar", "pending", dns_name="x1.example.com")
self.s3.get_object(self.env_name, "provider-state")
self.mocker.result(succeed(dump(
{"zookeeper-instances": [server.instance_id]})))
self.ec2.describe_instances(*[server.instance_id])
self.mocker.result(succeed([server]))
self.mocker.replay()
provider = self.get_provider()
connect = EC2Connect(provider)
d = connect.run()
self.assertFailure(d, EnvironmentPending)
def check_error(error):
self.assertEquals(str(error),
"Started machine is still pending.")
d.addCallback(check_error)
return d
def test_provider_connection_fails_with_recent_machine(self):
"""
If we can't connect to a recently bootstrapped machine,
we show an error indicating the machine might still be
initializing.
"""
launched_minutes_ago = 1
server = Instance("i-foobaz", "running", dns_name="x2.example.com",
launch_time=time.time()-launched_minutes_ago*60-1)
self.s3.get_object(self.env_name, "provider-state")
self.mocker.result(succeed(dump(
{"zookeeper-instances": [server.instance_id]})))
self.ec2.describe_instances(server.instance_id)
self.mocker.result(succeed([server]))
client = self.mocker.patch(SSHClient)
client.connect(server.dns_name + ":2181", timeout=30)
self.mocker.result(fail(NoConnection("BADABOOM!")))
self.mocker.replay()
provider = self.get_provider()
d = provider.connect()
def check_error(error):
self.assertEqual(
str(error), "Can't yet connect to started machine: BADABOOM!")
self.assertFailure(d, NoConnection)
d.addCallback(check_error)
return d
def test_provider_connection_fails_with_old_machine(self):
"""
If the machine has started for a while, though, we won't
tweak the error message in any way.
"""
launched_minutes_ago = 5
server = Instance("i-foobaz", "running", dns_name="x2.example.com",
launch_time=time.time()-launched_minutes_ago*60-1)
self.s3.get_object(self.env_name, "provider-state")
self.mocker.result(succeed(dump(
{"zookeeper-instances": [server.instance_id]})))
self.ec2.describe_instances(server.instance_id)
self.mocker.result(succeed([server]))
client = self.mocker.patch(SSHClient)
client.connect(server.dns_name + ":2181", timeout=30)
self.mocker.result(fail(NoConnection("BADABOOM!")))
self.mocker.replay()
provider = self.get_provider()
d = provider.connect()
def check_error(error):
self.assertEqual(str(error), "BADABOOM!")
self.assertFailure(d, NoConnection)
d.addCallback(check_error)
return d
def test_recent_machine_error_tweaking_works_with_no_conn_only(self):
"""
If the error raised is not NoConnection, then it should not
be tweaked even if it is a recent machine.
"""
launched_minutes_ago = 1
server = Instance("i-foobaz", "running", dns_name="x2.example.com",
launch_time=time.time()-launched_minutes_ago*60-1)
self.s3.get_object(self.env_name, "provider-state")
self.mocker.result(succeed(dump(
{"zookeeper-instances": [server.instance_id]})))
self.ec2.describe_instances(server.instance_id)
self.mocker.result(succeed([server]))
client = self.mocker.patch(SSHClient)
client.connect(server.dns_name + ":2181", timeout=30)
self.mocker.result(fail(ValueError("BADAPOFT!")))
self.mocker.replay()
provider = self.get_provider()
d = provider.connect()
def check_error(error):
self.assertEqual(str(error.error), "BADAPOFT!")
self.assertFailure(d, ProviderInteractionError)
d.addCallback(check_error)
return d
|