xCAT/confluent
2
0
Fork 0
mirror of https://github.com/xcat2/confluent.git synced 2026-07-21 08:37:29 +00:00
Code Issues Projects Releases Wiki Activity

Await collective address propagation

Browse Source
This commit is contained in:
Markus Hilger
2026-07-13 01:00:17 +02:00
parent 3c6e7d202f
commit 430260becf
2 changed files with 7 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
confluent_server/confluent/collective/manager.py
+2 -2
View File
@@ -656,8 +656,8 @@ async def handle_connection(connection, cert, request, local=False):
retrythread = None
async with leader_init:
cnn = tlvdata.get_socket(connection)
cfm.update_collective_address(request['name'],
cnn.getpeername()[0])
await cfm.update_collective_address(request['name'],
cnn.getpeername()[0])
await tlvdata.send(connection, cfm._dump_keys(None, False))
await tlvdata.send(connection, cfm._cfgstore['collective'])
await tlvdata.send(connection, {'confluent_uuid': cfm.get_global('confluent_uuid')}) # cfm.get_globals())
confluent_server/confluent/config/configmanager.py
+5 -5
View File
@@ -743,7 +743,7 @@ async def relay_slaved_requests(name, listener):
_newquorum = has_quorum()
_hasquorum = _newquorum
if _hasquorum and _pending_collective_updates:
apply_pending_collective_updates()
await apply_pending_collective_updates()
msg = await lh.get_next_msg()
while msg:
if name not in cfgstreams:
@@ -1050,23 +1050,23 @@ def _true_del_collective_member(name, sync=True):
_pending_collective_updates = {}
def update_collective_address(name ,address):
async def update_collective_address(name, address):
fprint = _cfgstore['collective'][name]['fingerprint']
oldaddress = _cfgstore['collective'][name]['address']
if oldaddress == address:
return
try:
check_quorum()
add_collective_member(name, address, fprint)
await add_collective_member(name, address, fprint)
except exc.DegradedCollective:
_pending_collective_updates[name] = address
def apply_pending_collective_updates():
async def apply_pending_collective_updates():
for name in list(_pending_collective_updates):
fprint = _cfgstore['collective'][name]['fingerprint']
address = _pending_collective_updates[name]
add_collective_member(name, address, fprint)
await add_collective_member(name, address, fprint)
del _pending_collective_updates[name]