xCAT/confluent
2
0
Fork 0
mirror of https://github.com/xcat2/confluent.git synced 2025-01-13 11:17:49 +00:00
Code Issues Projects Releases Wiki Activity

Error on loss of manager in flight

Browse Source
This commit is contained in:
Jarrod Johnson 2018-07-19 14:36:23 -04:00
parent 200569e7af
commit b053d41cd8
1 changed files with 35 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
confluent_server/confluent/core.py
View File

@ -831,6 +831,7 @@ def dispatch_request(nodes, manager, element, configmanager, inputdata,
a = configmanager.get_collective_member(manager)
try:
remote = socket.create_connection((a['address'], 13001))
remote.settimeout(90)
remote = ssl.wrap_socket(remote, cert_reqs=ssl.CERT_NONE,
keyfile='/etc/confluent/privkey.pem',
certfile='/etc/confluent/srvcert.pem')
@ -851,20 +852,48 @@ def dispatch_request(nodes, manager, element, configmanager, inputdata,
tlvdata.send(remote, {'dispatch': {'name': myname, 'length': len(dreq)}})
remote.sendall(dreq)
while True:
rlen = remote.recv(8)
try:
rlen = remote.recv(8)
except Exception:
for node in nodes:
yield msg.ConfluentResourceUnavailable(
node, 'Collective member {0} went unreachable'.format(
a['name']))
return
while len(rlen) < 8:
nlen = remote.recv(8 - len(rlen))
try:
nlen = remote.recv(8 - len(rlen))
except Exception:
nlen = 0
if not nlen:
raise Exception('Error receiving data')
for node in nodes:
yield msg.ConfluentResourceUnavailable(
node, 'Collective member {0} went unreachable'.format(
a['name']))
return
rlen += nlen
rlen = struct.unpack('!Q', rlen)[0]
if rlen == 0:
break
rsp = remote.recv(rlen)
try:
rsp = remote.recv(rlen)
except Exception:
for node in nodes:
yield msg.ConfluentResourceUnavailable(
node, 'Collective member {0} went unreachable'.format(
a['name']))
return
while len(rsp) < rlen:
nrsp = remote.recv(rlen - len(rsp))
try:
nrsp = remote.recv(rlen - len(rsp))
except Exception:
nrsp = 0
if not nrsp:
raise Exception('Error receving data')
for node in nodes:
yield msg.ConfluentResourceUnavailable(
node, 'Collective member {0} went unreachable'.format(
a['name']))
return
rsp += nrsp
rsp = pickle.loads(rsp)
if isinstance(rsp, Exception):