grpc/grpc#4629
grpc/grpc#14561

In Etcd3Client.status() there is the line:

status_response = self.maintenancestub.Status(status_request)

No 'timeout' value is passed in. Therefore, if it can't connect to etcd at this point (maybe the server can't be reached) it will just sit there for a long time. It will probably timeout eventually, but it's too long.

It's easy to honor the timeout value of the etcd client, though, by modifying this line as follows:

status_response = self.maintenancestub.Status(status_request, self.timeout)

This works like a charm and gives a proper ConnectionTimeoutError exception.

{
  "language": "python",
  "python": 3.5,
  "env": "TOXENV=py27",
...
}

Tests are only running against python 3.5.

For some reason the docs are displaying method arguments as (*args, **kwargs).

eg get_prefix(*args, **kwargs) instead of get_prefix(self, key_prefix, sort_order=None, sort_target='key')

Not sure why this is happening, but it makes them really confusing so should be fixed.

The current KV api is nice and simple for getting values:

>>> etcd.get("/key")
b"some-value"

It doesn't allow for retrieving the metadata about keys, however.

We need to figure out a nice api for showing all the data: key, create_revision, mod_revision, version, value, lease.

How other projects do it:

python-etcd

python-etcd returns an object for all reads containing a .value attribute to read the actual value:

client.read('/nodes/n2').value
#recursively read a directory
r = client.read('/nodes', recursive=True, sorted=True)
for child in r.children:
    print("%s: %s" % (child.key,child.value))

kazoo (zookeeper)

kazoo returns two values, the data and the node information:

data, stat = zk.get("/my/favorite")
print("Version: %s, data: %s" % (stat.version, data.decode("utf-8")))

python-consul

python-consul is pretty ugly:

index, data = consul.Consul().kv.get('foo', index=index)
print data['Value']

data is a dict that looks like:

{
    "CreateIndex": 100,
    "ModifyIndex": 200,
    "LockIndex": 200,
    "Key": "foo",
    "Flags": 0,
    "Value": "bar",
    "Session": "adf4238a-882b-9ddc-4a9d-5b6758e4159e"
}

@kragniz , after installed etcd3 version 0.7.0, I can't access data in etcd when user authentication is enabled. The following is the traceback,

In [30]: Etcd3Client(user='root', password='xxxxxx')
Out[30]: <etcd3.client.Etcd3Client at 0x10728a518>

In [31]: Out[30].get('foo')
---------------------------------------------------------------------------
_Rendezvous                               Traceback (most recent call last)
<ipython-input-31-b8d3a98c8e97> in <module>()
----> 1 Out[30].get('foo')

/Users/justdoit/workspace/works/laiye/venv/v3.6/lib/python3.6/site-packages/etcd3/client.py in handler(*args, **kwargs)
     46                 return f(*args, **kwargs)
     47             except grpc.RpcError as exc:
---> 48                 _translate_exception(exc)
     49
     50     return functools.wraps(f)(handler)

/Users/justdoit/workspace/works/laiye/venv/v3.6/lib/python3.6/site-packages/etcd3/client.py in handler(*args, **kwargs)
     44         def handler(*args, **kwargs):
     45             try:
---> 46                 return f(*args, **kwargs)
     47             except grpc.RpcError as exc:
     48                 _translate_exception(exc)

/Users/justdoit/workspace/works/laiye/venv/v3.6/lib/python3.6/site-packages/etcd3/client.py in get(self, key)
    247             range_request,
    248             self.timeout,
--> 249             credentials=self.call_credentials,
    250         )
    251

/Users/justdoit/workspace/works/laiye/venv/v3.6/lib/python3.6/site-packages/grpc/_channel.py in __call__(self, request, timeout, metadata, credentials)
    490         state, call, deadline = self._blocking(request, timeout, metadata,
    491                                                credentials)
--> 492         return _end_unary_response_blocking(state, call, False, deadline)
    493
    494     def with_call(self, request, timeout=None, metadata=None, credentials=None):

/Users/justdoit/workspace/works/laiye/venv/v3.6/lib/python3.6/site-packages/grpc/_channel.py in _end_unary_response_blocking(state, call, with_call, deadline)
    438             return state.response
    439     else:
--> 440         raise _Rendezvous(state, None, None, deadline)
    441
    442

_Rendezvous: <_Rendezvous of RPC that terminated with (StatusCode.INVALID_ARGUMENT, etcdserver: user name is empty)>

This was tested on my Mac, with python version 3.6, and detailed as the followed,

Python 3.6.0 (default, Mar 24 2017, 18:00:20)
[GCC 4.2.1 Compatible Apple LLVM 8.0.0 (clang-800.0.42.1)] on darwin

The Mac version is Darwin yusenbindeMacBook-Pro.local 17.2.0 Darwin Kernel Version 17.2.0: Fri Sep 29 18:27:05 PDT 2017; root:xnu-4570.20.62~3/RELEASE_X86_64 x86_64.

However, I could access with etcdctl command line tool,

yusenbindeMacBook-Pro:etcd justdoit$ export ETCDCTL_API=3
yusenbindeMacBook-Pro:etcd justdoit$ etcdctl --user=root get foo
Password:
foo
bar_1

The authentication process had already succeeded, and the message was logged.

2017-11-13 17:06:16.295082 D | auth: authorized root, token is dAhzbXDmVRQsFJYA.72`

It's the same as what etcdctl has did. Hence it seems that the gRPC's call credential check has failed.

So, guys, how does this occurred?

gRPC is nice, but is awkward to deploy in some situations (using eventlet, running on musl libc, and other problems with the C library). The HTTP proxy API appears to be in a better state these days, so a backend could be created to only communicate using that.

This would expose the same python API.

The client can throw out quite confusing grpc exceptions which could be neatly wrapped.

Traceback (most recent call last):
  File "/usr/local/lib/python3.5/site-packages/flask/app.py", line 1982, in wsgi_app
    response = self.full_dispatch_request()
  File "/usr/local/lib/python3.5/site-packages/flask/app.py", line 1614, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File "/usr/local/lib/python3.5/site-packages/flask/app.py", line 1517, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File "/usr/local/lib/python3.5/site-packages/flask/_compat.py", line 33, in reraise
    raise value
  File "/usr/local/lib/python3.5/site-packages/flask/app.py", line 1612, in full_dispatch_request
    rv = self.dispatch_request()
  File "/usr/local/lib/python3.5/site-packages/flask/app.py", line 1598, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  ...
  File "/usr/local/lib/python3.5/site-packages/etcd3/client.py", line 525, in add_member
    self.timeout)
  File "/usr/local/lib/python3.5/site-packages/grpc/_channel.py", line 481, in __call__
    return _end_unary_response_blocking(state, False, deadline)
  File "/usr/local/lib/python3.5/site-packages/grpc/_channel.py", line 432, in _end_unary_response_blocking
    raise _Rendezvous(state, None, None, deadline)

Deadline exceeded is a non critical error I would like to catch:

for i in range(4):
    try:
        etcd.add_member(["http://172.1.1.1:2380"])
        return True   
    except DeadlineExceeded:
        print("hit deadline")
     return False

Another obvious one I have seen: StatusUnavailable

Traceback (most recent call last):
  File "/usr/local/lib/python3.5/site-packages/etcd3/client.py", line 564, in members
    self.timeout)
  File "/usr/local/lib/python3.5/site-packages/grpc/_channel.py", line 481, in __call__
    return _end_unary_response_blocking(state, False, deadline)
  File "/usr/local/lib/python3.5/site-packages/grpc/_channel.py", line 432, in _end_unary_response_blocking
    raise _Rendezvous(state, None, None, deadline)
grpc._channel._Rendezvous: <_Rendezvous of RPC that terminated with (StatusCode.UNAVAILABLE, )>

In application I can't detect when etcd return compact_revision in WatchResponse. I expect, what in this case exception will raised and i can react on it (for example run full synchronizaion)

can i set etcd cluster address in etcd3.client(host="192.168.0.9,192.168.0.10“)

Why do you prefix the lock keys with "/locks/"?
Why is this not mentioned in the API documentation?
Why is this not configurable?

lock_prefix = '/locks/'  
[...]  
self.key = lock_prefix + self.name  

Doesn't look like there's anyway to provide user credentials to the etcd.client object.

Hi,

When we try, to add a watch but etcd is not running, there is no exception raise or status failed return.

In the next example, I try to connect the with the wrong port. No issue for client creation, but when I create my watch, etcd3 never return.

>>> import etcd3
>>> c=etcd3.client(host='127.0.0.1', port='2378', timeout=3)
>>> c.watch_once('1111')

In the watch.py file (add_callback method), the watch queue is not receiving the watcher timeout.

I think at this line:
return self._watch_id_queue.get()
should send the timeout value:
return self._watch_id_queue.get(timeout=self.timeout)

In this case, the add_callback can catch the Empty exception sent by queue.Queue.

Best regards,

Is etcd3 client thread safe?

https://github.com/kragniz/python-etcd3/blob/master/etcd3/client.py#L108

As of right now if no certs are provided it's assumed it's an insecure connection. We should be setting the default grpc channel credentials in the event https is present without certs.

Example:
https://github.com/davissp14/etcdv3-ruby/blob/master/lib/etcdv3/connection.rb#L41-L47

To reproduce, run make a watcher.py:

import etcd3

etcd = etcd3.client()

etcd.put('/doot/watch', 'dfdd')

for event, cancel in etcd.watch_prefix('/doot/watch'):
    print(event)

I'd expect Ctrl+c to exit the process, but that doesn't work:

(py27) ~/g/python-etcd3 (feature-watch) $ python watcher.py 
header {
  cluster_id: 14841639068965178418
  member_id: 10276657743932975437
  revision: 5057
  raft_term: 4
}
created: true

^C^C^C^C^C^C^C^C^C^C^C^C^C^C^C^C

Sending a SIGTERM makes it exit.

Firstly, thanks for taking the time to make this module. lots of people have been waiting for a v3 implementation.

Using Ubuntu 12.04 LTS with python 2.7.6 and pip 1.5.4. After installing via pip install etcd3 and getting many, many warning messages (but successful installation):

$ cat test3.py
#!/usr/bin/python

import etcd3
conn = etcd3.client(host='127.0.0.1')
conn.get('/test')

$ ./test3.py
Traceback (most recent call last):
  File "./test3.py", line 3, in <module>
    import etcd3
  File "/usr/local/lib/python2.7/dist-packages/etcd3/__init__.py", line 3, in <module>
    from etcd3.client import Etcd3Client
  File "/usr/local/lib/python2.7/dist-packages/etcd3/client.py", line 5, in <module>
    import etcd3.transactions as transactions
  File "/usr/local/lib/python2.7/dist-packages/etcd3/transactions.py", line 1, in <module>
    from etcd3.etcdrpc import rpc_pb2 as etcdrpc
ImportError: No module named etcdrpc

$ ls /usr/local/lib/python2.7/dist-packages/etcd3
client.py  client.pyc  exceptions.py  exceptions.pyc  __init__.py  __init__.pyc  members.py  members.pyc  transactions.py  transactions.pyc  utils.py  utils.pyc

In get/put/del/watch response, can't access to ResponseHeader (in particular, revision field)

The _handle_errors generator ought to work with regular functions, but generators like get_prefix and get_all fails outside of the return. It should be handled specifically.

mport etcd3
import json
import time

etcd = etcd3.client(host='127.0.0.1', port=2379)

while True:
try:
etcd.put('test', '123456')
time.sleep(1)
print '===='
except Exception as error:
print 'error'

if kill the etcd server, then the cpu of the client is too high, maybe 80%

The python docker image version is not decalring a specific version, so the latest one is downloaded.
Currently the latest version is 3.6.1. This causes this step of the image build to fail:

RUN tox -epy35 --notest

since python3.5 is not installed.

One solution would be to explicitly declare the desired python version in the FROM directive.

Hi 👊

This is my first visit to this fine repo, but it seems you have been working hard to keep all dependencies updated so far.

Once you have closed this issue, I'll create seperate pull requests for every update as soon as I find one.

That's it for now!

Happy merging! 🤖

Hi, now in Etcd3Client's initialize function, the watcher member is a self-started thread, while watcher is not for all.

Initialize Etcd3Client while initializing watcher

self.watcher = watch.Watcher(etcdrpc.WatchStub(self.channel),
                                     timeout=self.timeout)

Watcher is a auto-start daemon thread.

class Watcher(threading.Thread):

    def __init__(self, watchstub, timeout=None):
        threading.Thread.__init__(self)
        self.timeout = timeout
        self._watch_id_callbacks = {}
        self._watch_id_queue = queue.Queue()
        self._watch_id_lock = threading.Lock()
        self._watch_requests_queue = queue.Queue()
        self._watch_response_iterator = \
            watchstub.Watch(self._requests_iterator)
        self._callback = None
        self.daemon = True
        self.start()

Is there any consideration about lazy watcher thread start?

We're seeing this error message intermittently in CI, causing most jobs to fail:

subprocess.CalledProcessError: Command '['etcdctl', '-w', 'json', 'get', '/doot/put_1']' returned non-zero exit status -11

The get_all() API returns the values of all the keys, but not the keys themselves.

After putting the following keys and values, what is the API to get a list of just the keys /foo, /bar, /baz and /qux?

>>> etcd.put("/foo", "a")
>>> etcd.put("/bar", "b")
>>> etcd.put("/baz", "c")
>>> etcd.put("/qux", "d")

I was not able to find this in the documentation in https://python-etcd3.readthedocs.io/en/latest/usage.html.

It is possible to get just the keys in etcd3 using etcdctl the following way:

ETCDCTL_API=3 etcdctl get --prefix=true "" | grep '/'
/bar
/baz
/foo
/qux

But, I wasn't able to find the python API to do the above. Let me know if I'm missing anything.

Nothing major, just a few broken links in the install docs that mention github.com/kragniz/etcd3 instead of github.com/kragniz/python-etcd3.

Hello,

Thanks for working on this. I've just started using etcd v3 so maybe I'm missing something, but I find that the default behavior of the client's 'get' method when requesting a non-existing key is inappropriate. I think we should take advantage of a behavior similar to the Python dictionary, i.e. be able to get at least 'None' instead of catching an exception.

This is especially true when we want to test if a key already exist. Moreover, it would be more consistent with the 'etcdctl' CLI tool that does not display anything when a non-existing key is requested, an return a 0 exit code.

tenacity is not compatible with python 2.7 after version 4.4.0.

Related issue: jd/tenacity#88

EDIT:

It's a problem of building a RPM package and can be resolved by setting the RPM spec:

%global _python_bytecompile_errors_terminate_build 0

Hi,

While testing how a lease works i noticed that upon a lease expiration this unhandled exception is raised:

grpc._channel._Rendezvous: <_Rendezvous of RPC that terminated with (StatusCode.NOT_FOUND, etcdserver: requested lease not found)>

I used the python debugger and the error handling seems to fail in the _handle_errors decorator here

pdb output:

except grpc.RpcError as exc:
(Pdb) n
/usr/lib/python2.7/site-packages/etcd3/client.py(33)handler()
-> code = exc.code()
(Pdb) n
/usr/lib/python2.7/site-packages/etcd3/client.py(34)handler()
-> exception = _EXCEPTIONS_BY_CODE.get(code)
(Pdb) print code
StatusCode.NOT_FOUND
(Pdb) print _EXCEPTIONS_BY_CODE.get(code)
None
(Pdb) n
/usr/lib/python2.7/site-packages/etcd3/client.py(35)handler()
-> if exception is None:
(Pdb) n
/usr/lib/python2.7/site-packages/etcd3/client.py(36)handler()
-> raise

The code i used to test it:

import etcd3
from etcd3 import exceptions
import time

etcd_con = etcd3.client()

lease = etcd_con.lease(ttl=1)

time.sleep(3)

try:
    etcd_con.put('/my_key', '1', lease.id)
except exceptions.Etcd3Exception:
    pass

It is also strange that if the process sleeps for 2 seconds instead of 3 the lease does not expire as expected since the ttl is set to 1 sec.

Best Regards

The following gRPC methods should have comparative functionality in the python-etcd3 API. This issue is to keep track of features that still require some implementation.

Auth

• AuthEnable
• AuthDisable
• Authenticate
• UserAdd
• UserGet
• UserList
• UserDelete
• UserChangePassword
• UserGrantRole
• UserRevokeRole
• RoleAdd
• RoleGet
• RoleList
• RoleDelete
• RoleGrantPermission
• RoleRevokePermission

Cluster

• MemberAdd
• MemberRemove
• MemberUpdate
• MemberList

KV

• Range
• Put
• DeleteRange
• Txn
• Compact

Lease

• LeaseGrant
• LeaseRevoke
• LeaseKeepAlive
• LeaseTimeToLive

Maintenance

• Alarm
• Status
• Defragment
• Hash
• Snapshot

Watch

• Watch

The current implementation of method Etcd3Client.add_watch_callback will trigger deadlock (block indefinitely) if it is invoked in a watch callback function.

For example:

import time
import etcd3

etcd = etcd3.client(host='localhost')

def callback(event):
    print 'Callback begins.'
    etcd.add_watch_callback('/123', callback=callback)
    print 'Callback done.'

callback(None)  # Add the callback

print "Putting new key..."
etcd.put('/123', '123')  # Trigger the added callback
print "New key was uploaded."

while True:
    time.sleep(1)  # Prevent the daemon watcher from being stopped

The second "Callback done." can never be printed if you run this code.

This problem is applicable to other methods which invoke add_watch_callback, including watch, watch_once, watch_prefix and watch_prefix_once.

When the algorithm uses ecdsa, the GRPC default cipher suite cannot connect to etcd。

os.environ["GRPC_SSL_CIPHER_SUITES"] = "ECDHE-ECDSA-AES256-GCM-SHA384"

At the moment, python-etcd3 code for transactions doesn't allow no value to be defined in compare operations (which would be, in SQL terms, the NULL term).

This means that it's impossible to write even simple constructs like "Create this key if it doesn't exist".

c = etcd3.client()
res = c.transaction(compare=[c.transactions.create('test') == 0], success=[c.transactions.put('test', 'testvalue')], failure=[]) # This returns false
res = c.transaction(compare=[c.transactions.create('test') == None], success=[c.transactions.put('test', 'testvalue')], failure=[]) # This raises an exception

We want to check the whole etcd hasn't changed, for reasons... Would it be cool to return them instead of None, something like the following:

diff --git a/etcd3/client.py b/etcd3/client.py
index 662d125..55a85df 100644
--- a/etcd3/client.py
+++ b/etcd3/client.py
@@ -346,12 +346,13 @@ class Etcd3Client(object):
         :type lease: either :class:`.Lease`, or int (ID of lease)
         """
         put_request = self._build_put_request(key, value, lease=lease)
-        self.kvstub.Put(
+        meta_data = self.kvstub.Put(
             put_request,
             self.timeout,
             credentials=self.call_credentials,
             metadata=self.metadata
         )
+        return (True, meta_data)
 
     @_handle_errors
     def replace(self, key, initial_value, new_value):
@@ -644,7 +645,7 @@ class Etcd3Client(object):
         for response in txn_response.responses:
             response_type = response.WhichOneof('response')
             if response_type == 'response_put':
-                responses.append(None)
+                responses.append(response)
 
             elif response_type == 'response_range':
                 range_kvs = []

I was browsing through this client while working on a version for Node.js. I noticed that your locking is implemented on a straight k/v without using a lease. I suggest you consider using a lease for this, so that if something happens to the server or application which makes the lock, the resource isn't abandoned indefinitely. The downside is that it's possible that a lock is 'lost' while a process thinks it has it, but the set of situations in which that can occur and cause issues is a strict subset of the situations (only one I can think of: a partial network partition) in which the lock would otherwise be held indefinitely.

My implementation and little more reasoning is here, if you're interested: microsoft/etcd3#5

When 10 threads wait to acquire the etcd3 lock, I see this exception sometimes:

  with etcd_client.lock("etcd_lock"):
  File "/usr/local/lib/python2.7/dist-packages/etcd3/locks.py", line 136, in __enter__
    self.acquire()
  File "/usr/local/lib/python2.7/dist-packages/etcd3/locks.py", line 103, in acquire
    return _acquire()
  File "/usr/local/lib/python2.7/dist-packages/tenacity/__init__.py", line 231, in wrapped_f
    return self.call(f, *args, **kw)
  File "/usr/local/lib/python2.7/dist-packages/tenacity/__init__.py", line 313, in call
    start_time=start_time)
  File "/usr/local/lib/python2.7/dist-packages/tenacity/__init__.py", line 269, in iter
    return fut.result()
  File "/usr/local/lib/python2.7/dist-packages/concurrent/futures/_base.py", line 455, in result
    return self.__get_result()
  File "/usr/local/lib/python2.7/dist-packages/tenacity/__init__.py", line 316, in call
    result = fn(*args, **kwargs)
  File "/usr/local/lib/python2.7/dist-packages/etcd3/locks.py", line 83, in _acquire
    self.lease = self.etcd_client.lease(self.ttl)
  File "/usr/local/lib/python2.7/dist-packages/etcd3/client.py", line 48, in handler
    _translate_exception(exc)
  File "/usr/local/lib/python2.7/dist-packages/etcd3/client.py", line 32, in _translate_exception
    raise exception
  ConnectionFailedError

It'd be nicer if the grpc code was generated at build time rather than kept in tree.

Something wrong when create etcd client,

    Traceback (most recent call last):
      File "grpc.py", line 1, in <module>
         import etcd3
      File "/home/vagrant/.pyenv/versions/promotion/lib/python3.6/site-packages/etcd3/__init__.py", l 
   ine 4, in <module>
        from etcd3.client import Etcd3Client
      File "/home/vagrant/.pyenv/versions/promotion/lib/python3.6/site-packages/etcd3/client.py", line 
    6, in <module>
        import grpc._channel
    ModuleNotFoundError: No module named 'grpc._channel'; 'grpc' is not a package

my python version is 3.6.5, anyone can give me some suggestions

Repeatedly calling etcd3.client(...) in a loop will leak lots of memory over time. The main reasons seems to be that this creates a watcher thread every time. This thread is never stopped or deleted thus causing memory to grow without bounds.

It would be nice to support using the same DNS SRV record structure as etcdctl and etcd can use to discover endpoints.

When running etcd3 as a cluster, it'd be useful to be able to specify multiple endpoints to try to connect to. It seems the Go client allows that, for example.

Hi,

I am one of the authors of python-etcd and I have written a very crude client for etcd3 as well, which I was going to publish when I got some of the parts there to work better.

Would you like me to submit pull requests with parts of my code that are (IMHO) more complete to your current implementation? If so, what I have to offer is the following:

• Cluster discovery via DNS or a list of endpoints can be passed, then the client autodiscovers the cluster and will reconnect to another server in case of failure, pretty much how python-etcd does
• Some transaction-related syntactic sugar
• Some of the methods implementation that lack some features in your code

What I, unluckily, was unable to make work are two things that are pretty important:

• Auth (I am able to authenticate, but then I can't figure out how to use the resulting token in subsequent requests)
• Watch (watch is simply broken even with py-grpc 1.0.0 and I didn't have time to look at it thoroughly)

Let me know if you're interested!

G

The docs say they return keys.
My experience says they return values.
etcdctl returns both keys and values for get_prefix

Am I crazy?

My python code is simple from etcd3 readme.md.

import etcd3
etcd = etcd3.client()
etcd.get('foo')

However, when the script run, etcd3.exceptions.ConnectionFailedError has occurred.

Because I can it by using curl + etcdctl, so, I think it is not etcd problem.

$ curl -L http://localhost:4001/v2/keys/mykey -XPUT -d value="Hello"
{"action":"set","node":{"key":"/mykey","value":"Hello","modifiedIndex":52,"createdIndex":52},"prevNode":{"key":"/mykey","value":"Hello World","modifiedIndex":51,"createdIndex":51}}
$ 
$ etcdctl get /mykey
Hello

Is it etcd3's bug? How can I solve this problem?

I'm currently using etcd3 0.7.0.
I've tried even the old version(0.4.0 ~ 0.6.0), but I still have problems.

My stack trace is below,

(scaling_python_kor) hyun@hyun-VirtualBox:~/work/scaling_python_kor/Chapter07$ python 08_etcd-lock.py 
Traceback (most recent call last):
  File "/home/hyun/work/scaling_python_kor/lib/python3.5/site-packages/etcd3/client.py", line 46, in handler
    return f(*args, **kwargs)
  File "/home/hyun/work/scaling_python_kor/lib/python3.5/site-packages/etcd3/client.py", line 249, in get
    credentials=self.call_credentials,
  File "/home/hyun/work/scaling_python_kor/lib/python3.5/site-packages/grpc/_channel.py", line 487, in __call__
    return _end_unary_response_blocking(state, call, False, deadline)
  File "/home/hyun/work/scaling_python_kor/lib/python3.5/site-packages/grpc/_channel.py", line 437, in _end_unary_response_blocking
    raise _Rendezvous(state, None, None, deadline)
grpc._channel._Rendezvous: <_Rendezvous of RPC that terminated with (StatusCode.UNAVAILABLE, Trying to connect an http1.x server)>

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "08_etcd-lock.py", line 5, in <module>
    etcd.get('foo')
  File "/home/hyun/work/scaling_python_kor/lib/python3.5/site-packages/etcd3/client.py", line 48, in handler
    _translate_exception(exc)
  File "/home/hyun/work/scaling_python_kor/lib/python3.5/site-packages/etcd3/client.py", line 32, in _translate_exception
    raise exception
etcd3.exceptions.ConnectionFailedError

support py3.6?

python-etcd3 version 0.5.2
client.py

336 for m in self.members:
337 if m.id == status_response.leader:
338 leader = m
339 else:
340 # raise exception?
341 leader = None

leader maybe return None in most cases

centos7.2, python2.7 etcd3 0.5.2

I put 20w key to etcd3, try to test python-etcd3 watch performance.

def watch_node():
    host = random.choice(ETCD_SERVER)
    etcd = etcd3.client(host=host[0], port=host[1])
    watch_count = 0
    events_iterator, cancel = etcd.watch_prefix('/')
    f = open('etcdout', 'w')
    for event in events_iterator:
        # print(watch_count, event)
        f.write(str(watch_count))
        f.write("\n")
        f.flush()
        # f.write(str(event))
        watch_count += 1
        if watch_count > 1000000:
            cancel()

package main
import "fmt"
import "time"
import (
	"log"
        "math/rand"
        "os"
	"golang.org/x/net/context"
	"github.com/coreos/etcd/clientv3"
)



func worker(id int, jobs <-chan int, results chan<- int) {
    num := rand.Int31n(10)
    time.Sleep(time.Duration(num) * time.Second)

    cfg := clientv3.Config{
		Endpoints:               []string{"http://10.10.3.10:2379", "http://10.1.106.39:2379", "http://10.15.12.253:2379"},
		DialTimeout: 5 * time.Second,
	}
	cli, err := clientv3.New(cfg)
	if err != nil {
		log.Fatal(err)
	}
    for j := range jobs {
        
        // ctx, cancel := context.WithTimeout(context.Background(), 3)
        resp, err := cli.Put(context.Background(), fmt.Sprintf("/computed/host_vm/%i/%i/state", id, j), "RUNNING")
        // defer cancel()
        log.Println("xxxxxxxxxxxxxxxxxxxxxx")
	if err != nil {
		log.Println(err)
                cli, err = clientv3.New(cfg)
                if err != nil {
                    log.Println(err)
                }
	} else {
		// print common key info
                _ = resp
		log.Printf("Set is done. Metadata is %q\n", resp)
	}
        time.Sleep(10 * time.Second)
        results <- j
    }
    // for j := range jobs {
    //     fmt.Println("worker", id, "started  job", j)
    //     time.Sleep(time.Second)
    //     fmt.Println("worker", id, "finished job", j)
    //     results <- j * 2
    // }
}

func main() {
    f, err := os.OpenFile("testlogfile", os.O_RDWR | os.O_CREATE | os.O_APPEND, 0666)
    if err != nil {
        fmt.Println("error opening file: %v", err)
    }
    defer f.Close()
    
    log.SetOutput(f)

    jobs := make(chan int, 500000)
    results := make(chan int, 500000)

    for w := 1; w <= 30000; w++ {
        go worker(w, jobs, results)
    }

    for j := 1; j <= 500000; j++ {
        jobs <- j
    }
    close(jobs)

    for a := 1; a <= 500000; a++ {
        <-results
    }
}

(gdb) bt
#0  0x00007f3ab2d4e790 in sem_wait () from /lib64/libpthread.so.0
#1  0x00007f3ab30697b5 in PyThread_acquire_lock () from /lib64/libpython2.7.so.1.0
#2  0x00007f3ab306d442 in lock_PyThread_acquire_lock () from /lib64/libpython2.7.so.1.0
#3  0x00007f3ab303daa4 in PyEval_EvalFrameEx () from /lib64/libpython2.7.so.1.0
#4  0x00007f3ab303f0bd in PyEval_EvalCodeEx () from /lib64/libpython2.7.so.1.0
#5  0x00007f3ab303d76f in PyEval_EvalFrameEx () from /lib64/libpython2.7.so.1.0
#6  0x00007f3ab303f0bd in PyEval_EvalCodeEx () from /lib64/libpython2.7.so.1.0
#7  0x00007f3ab303d76f in PyEval_EvalFrameEx () from /lib64/libpython2.7.so.1.0
#8  0x00007f3ab2fc1908 in gen_send_ex.isra.0 () from /lib64/libpython2.7.so.1.0
#9  0x00007f3ab303b4bd in PyEval_EvalFrameEx () from /lib64/libpython2.7.so.1.0

Hi,

It looks like there is currently no way to add a callback to watch a prefix. Client's add_watch_callback method works perfectly when watching a key, but the only way to watch a prefix is to do it in a blocking way as shown in the README:

watch_count = 0
events_iterator, cancel = etcd.watch_prefix("/doot/watch/prefix/")
for event in events_iterator:
    print(event)
    watch_count += 1
    if watch_count > 10:
        cancel()

When doing this in a dedicated thread, the execution is blocked as long as the events_iterator generator does not yield any value, and I'm not able to interrupt the thread. Am I missing something, or should there be an add_watch_prefix_callback dedicated method?