Skip to content

Commit 039d445

Browse files
fasaxcfasaxc
committed
Add logging, clean up some exception handling.
1 parent 72c8f0e commit 039d445

3 files changed

Lines changed: 93 additions & 16 deletions

File tree

‎src/etcd/__init__.py‎

Lines changed: 22 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,21 @@
1-
import collections
1+
import logging
22
from .client import Client
33
from .lock import Lock
44
from .election import LeaderElection
55

6+
_log = logging.getLogger(__name__)
7+
8+
# Prevent "no handler" warnings to stderr in projects that do not configure
9+
# logging.
10+
try:
11+
from logging import NullHandler
12+
except ImportError:
13+
# Python <2.7, just define it.
14+
class NullHandler(logging.Handler):
15+
def emit(self, record):
16+
pass
17+
_log.addHandler(NullHandler())
18+
619

720
class EtcdResult(object):
821
_node_props = {
@@ -146,6 +159,14 @@ class EtcdEventIndexCleared(EtcdException):
146159
"""
147160
pass
148161

162+
163+
class EtcdConnectionFailed(EtcdException):
164+
"""
165+
Connection to etcd failed.
166+
"""
167+
pass
168+
169+
149170
class EtcdError(object):
150171
# See https://github.com/coreos/etcd/blob/master/Documentation/errorcode.md
151172
error_exceptions = {

‎src/etcd/client.py‎

Lines changed: 67 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,9 @@
66
77
88
"""
9+
import logging
10+
import httplib
11+
import socket
912
import urllib3
1013
import json
1114
import ssl
@@ -16,6 +19,10 @@
1619
except ImportError:
1720
from urllib.parse import urlparse
1821

22+
23+
_log = logging.getLogger(__name__)
24+
25+
1926
class Client(object):
2027

2128
"""
@@ -73,6 +80,8 @@ def __init__(
7380
use_proxies (bool): we are using a list of proxies to which we connect,
7481
and don't want to connect to the original etcd cluster.
7582
"""
83+
_log.info("New etcd client created for %s:%s%s",
84+
host, port, version_prefix)
7685
self._protocol = protocol
7786

7887
def uri(protocol, host, port):
@@ -83,6 +92,7 @@ def uri(protocol, host, port):
8392
self._base_uri = uri(self._protocol, host, port)
8493
else:
8594
if not allow_reconnect:
95+
_log.error("List of hosts incompatible with allow_reconnect.")
8696
raise etcd.EtcdException("A list of hosts to connect to was given, but reconnection not allowed?")
8797
self._machines_cache = [uri(self._protocol, *conn) for conn in host]
8898
self._base_uri = self._machines_cache.pop(0)
@@ -104,6 +114,7 @@ def uri(protocol, host, port):
104114
if protocol == 'https':
105115
# If we don't allow TLSv1, clients using older version of OpenSSL
106116
# (<1.0) won't be able to connect.
117+
_log.debug("HTTPS enabled.")
107118
kw['ssl_version'] = ssl.PROTOCOL_TLSv1
108119

109120
if cert:
@@ -135,10 +146,12 @@ def uri(protocol, host, port):
135146
# extend the list given to the client with what we get
136147
# from self.machines
137148
if not self._use_proxies:
138-
self._machines_cache = list(set(self._machines_cache).union(set(self.machines)))
149+
self._machines_cache = list(set(self._machines_cache) |
150+
set(self.machines))
139151
if self._base_uri in self._machines_cache:
140152
self._machines_cache.remove(self._base_uri)
141-
153+
_log.debug("Machines cache initialised to %s",
154+
self._machines_cache)
142155

143156
@property
144157
def base_uri(self):
@@ -190,18 +203,28 @@ def machines(self):
190203
timeout=self.read_timeout,
191204
redirect=self.allow_redirect)
192205

193-
return [
206+
machines = [
194207
node.strip() for node in
195208
self._handle_server_response(response).data.decode('utf-8').split(',')
196209
]
197-
except:
198-
# We can't get the list of machines, if one server is in the machines cache, try on it
210+
_log.debug("Retrieved list of machines: %s", machines)
211+
return machines
212+
except (urllib3.exceptions.HTTPError,
213+
httplib.HTTPException,
214+
socket.error) as e:
215+
# We can't get the list of machines, if one server is in the
216+
# machines cache, try on it
217+
_log.error("Failed to get list of machines from %s%s: %r",
218+
self._base_uri, self.version_prefix, e)
199219
if self._machines_cache:
200220
self._base_uri = self._machines_cache.pop(0)
221+
_log.info("Retrying on %s", self._base_uri)
201222
# Call myself
202223
return self.machines
203224
else:
204-
raise etcd.EtcdException("Could not get the list of servers, maybe you provided the wrong host(s) to connect to?")
225+
raise etcd.EtcdException("Could not get the list of servers, "
226+
"maybe you provided the wrong "
227+
"host(s) to connect to?")
205228

206229
@property
207230
def leader(self):
@@ -273,6 +296,8 @@ def write(self, key, value, ttl=None, dir=False, append=False, **kwdargs):
273296
'newValue'
274297
275298
"""
299+
_log.info("Writing %s to key %s ttl=%s dir=%s append=%s",
300+
value, key, ttl, dir, append)
276301
key = self._sanitize_key(key)
277302
params = {}
278303
if value is not None:
@@ -316,6 +341,7 @@ def update(self, obj):
316341
obj (etcd.EtcdResult): The object that needs updating.
317342
318343
"""
344+
_log.info("Updating %s to %s.", obj.key, obj.value)
319345
kwdargs = {
320346
'dir': obj.dir,
321347
'ttl': obj.ttl,
@@ -362,6 +388,7 @@ def read(self, key, **kwdargs):
362388
'value'
363389
364390
"""
391+
_log.info("Issuing read for key %s with args %s", key, kwdargs)
365392
key = self._sanitize_key(key)
366393

367394
params = {}
@@ -407,6 +434,8 @@ def delete(self, key, recursive=None, dir=None, **kwdargs):
407434
'/key'
408435
409436
"""
437+
_log.info("Deleting %s recursive=%s dir=%s extra args=%s",
438+
key, recursive, dir, kwdargs)
410439
key = self._sanitize_key(key)
411440

412441
kwds = {}
@@ -418,6 +447,7 @@ def delete(self, key, recursive=None, dir=None, **kwdargs):
418447
for k in self._del_conditions:
419448
if k in kwdargs:
420449
kwds[k] = kwdargs[k]
450+
_log.debug("Calculated params = %s", kwds)
421451

422452
response = self.api_execute(
423453
self.key_endpoint + key, self._MDELETE, params=kwds)
@@ -509,6 +539,7 @@ def watch(self, key, index=None, timeout=None, recursive=None):
509539
'value'
510540
511541
"""
542+
_log.debug("About to wait on key %s, index %s", key, index)
512543
if index:
513544
return self.read(key, wait=True, waitIndex=index, timeout=timeout,
514545
recursive=recursive)
@@ -564,10 +595,16 @@ def _result_from_response(self, response):
564595

565596
def _next_server(self):
566597
""" Selects the next server in the list, refreshes the server list. """
598+
_log.debug("Selection next machine in cache. Available machines: %s",
599+
self._machines_cache)
567600
try:
568-
return self._machines_cache.pop()
601+
mach = self._machines_cache.pop()
569602
except IndexError:
570-
raise etcd.EtcdException('No more machines in the cluster')
603+
_log.error("Machines cache is empty, no machines to try.")
604+
raise etcd.EtcdConnectionFailed('No more machines in the cluster')
605+
else:
606+
_log.info("Selected new etcd server %s", mach)
607+
return mach
571608

572609
def api_execute(self, path, method, params=None, timeout=None):
573610
""" Executes the query. """
@@ -608,13 +645,31 @@ def api_execute(self, path, method, params=None, timeout=None):
608645
raise etcd.EtcdException(
609646
'HTTP method {} not supported'.format(method))
610647

611-
except urllib3.exceptions.MaxRetryError:
612-
self._base_uri = self._next_server()
613-
some_request_failed = True
648+
# urllib3 doesn't wrap all httplib exceptions and earlier versions
649+
# don't wrap socket errors either.
650+
except (urllib3.exceptions.HTTPError,
651+
httplib.HTTPException,
652+
socket.error) as e:
653+
_log.error("Request to server %s failed: %r",
654+
self._base_uri, e)
655+
if self._allow_reconnect:
656+
_log.info("Reconnection allowed, looking for another "
657+
"server.")
658+
# _next_server() raises EtcdException if there are no
659+
# machines left to try, breaking out of the loop.
660+
self._base_uri = self._next_server()
661+
some_request_failed = True
662+
else:
663+
_log.info("Reconnection disabled, giving up.")
664+
raise etcd.EtcdConnectionFailed(
665+
"Connection to etcd failed due to %r" % e)
666+
except:
667+
_log.exception("Unexpected request failure, re-raising.")
668+
raise
614669

615670
if some_request_failed:
616671
if not self._use_proxies:
617-
# The cluster may have changed since last invokation
672+
# The cluster may have changed since last invocation
618673
self._machines_cache = self.machines
619674
self._machines_cache.remove(self._base_uri)
620675
return self._handle_server_response(response)

‎src/etcd/tests/integration/test_simple.py‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -217,7 +217,8 @@ def test_reconnect_not_allowed(self):
217217
self.processHelper.run(number=3)
218218
self.client = etcd.Client(port=6001, allow_reconnect=False)
219219
self.processHelper.kill_one(0)
220-
self.assertRaises(etcd.EtcdException, self.client.get, '/test_set')
220+
self.assertRaises(etcd.EtcdConnectionFailed, self.client.get,
221+
'/test_set')
221222

222223
def test_reconnet_fails(self):
223224
""" INTEGRATION: fails to reconnect if no available machines """
@@ -487,8 +488,8 @@ def test_get_set_unauthenticated_with_ca(self):
487488
client = etcd.Client(
488489
protocol='https', port=6001, ca_cert=self.ca2_cert_path)
489490

490-
self.assertRaises(urllib3.exceptions.SSLError, client.set, '/test-set', 'test-key')
491-
self.assertRaises(urllib3.exceptions.SSLError, client.get, '/test-set')
491+
self.assertRaises(etcd.EtcdConnectionFailed, client.set, '/test-set', 'test-key')
492+
self.assertRaises(etcd.EtcdConnectionFailed, client.get, '/test-set')
492493

493494
def test_get_set_authenticated(self):
494495
""" INTEGRATION: set/get a new value authenticated """

0 commit comments

Comments
 (0)