From 670fc87e1d007db28b624b3d8830343f46218493 Mon Sep 17 00:00:00 2001 From: Jarrod Johnson Date: Mon, 5 Apr 2021 16:39:41 -0400 Subject: [PATCH] Address several collective issues When a stream has been deleted from cfgstreams, contiue exception handling since the desired result. For connections to a manager, institute a 15 second socket level timeout. This avoids an abandoned server conversation from locking a colleective member startup. When scheduling the failover check, first block any redundant attempts to schedule. Wrap the collective startup in an exception handler, so that a retry is more well guaranteed. --- .../confluent/collective/manager.py | 68 ++++++++++--------- .../confluent/config/configmanager.py | 5 +- 2 files changed, 39 insertions(+), 34 deletions(-) diff --git a/confluent_server/confluent/collective/manager.py b/confluent_server/confluent/collective/manager.py index f8a46de9..3a6d7bb1 100644 --- a/confluent_server/confluent/collective/manager.py +++ b/confluent_server/confluent/collective/manager.py @@ -181,7 +181,7 @@ def follow_leader(remote, leader): def connect_to_collective(cert, member): - remote = socket.create_connection((member, 13001)) + remote = socket.create_connection((member, 13001), 15) # TLS cert validation is custom and will not pass normal CA vetting # to override completely in the right place requires enormous effort, so just defer until after connect remote = ssl.wrap_socket(remote, cert_reqs=ssl.CERT_NONE, keyfile='/etc/confluent/privkey.pem', @@ -231,7 +231,7 @@ def handle_connection(connection, cert, request, local=False): return if follower: linfo = cfm.get_collective_member_by_address(currentleader) - remote = socket.create_connection((currentleader, 13001)) + remote = socket.create_connection((currentleader, 13001), 15) remote = ssl.wrap_socket(remote, cert_reqs=ssl.CERT_NONE, keyfile='/etc/confluent/privkey.pem', certfile='/etc/confluent/srvcert.pem') @@ -307,7 +307,7 @@ def handle_connection(connection, cert, request, local=False): return host = request['server'] try: - remote = socket.create_connection((host, 13001)) + remote = socket.create_connection((host, 13001), 15) # This isn't what it looks like. We do CERT_NONE to disable # openssl verification, but then use the invitation as a # shared secret to validate the certs as part of the join @@ -483,7 +483,7 @@ def handle_connection(connection, cert, request, local=False): if not cfm.relay_slaved_requests(drone, connection): if not retrythread: # start a recovery if everyone else seems # to have disappeared - retrythread = eventlet.spawn_after(30 + random.random(), + retrythread = eventlet.spawn_after(5 + random.random(), start_collective) # ok, we have a connecting member whose certificate checks out # He needs to bootstrap his configuration and subscribe it to updates @@ -629,39 +629,41 @@ def check_managers(): def schedule_rebalance(): global failovercheck if not failovercheck: + failovercheck = True failovercheck = eventlet.spawn_after(10, check_managers) def start_collective(): - cfm.membership_callback = schedule_rebalance global follower global retrythread - if follower: - follower.kill() - cfm.stop_following() - follower = None try: - if cfm.cfgstreams: - cfm.check_quorum() - # Do not start if we have quorum and are leader + cfm.membership_callback = schedule_rebalance + if follower: + follower.kill() + cfm.stop_following() + follower = None + try: + if cfm.cfgstreams: + cfm.check_quorum() + # Do not start if we have quorum and are leader + return + except exc.DegradedCollective: + pass + if leader_init.active: # do not start trying to connect if we are + # xmitting data to a follower return - except exc.DegradedCollective: - pass - if leader_init.active: # do not start trying to connect if we are - # xmitting data to a follower - return - myname = get_myname() - for member in sorted(list(cfm.list_collective())): - if member == myname: - continue - if cfm.cfgleader is None: - cfm.stop_following(True) - ldrcandidate = cfm.get_collective_member(member)['address'] - log.log({'info': 'Performing startup attempt to {0}'.format( - ldrcandidate), 'subsystem': 'collective'}) - if connect_to_leader(name=myname, leader=ldrcandidate): - break - else: - retrythread = eventlet.spawn_after(30 + random.random(), - start_collective) - - + myname = get_myname() + for member in sorted(list(cfm.list_collective())): + if member == myname: + continue + if cfm.cfgleader is None: + cfm.stop_following(True) + ldrcandidate = cfm.get_collective_member(member)['address'] + log.log({'info': 'Performing startup attempt to {0}'.format( + ldrcandidate), 'subsystem': 'collective'}) + if connect_to_leader(name=myname, leader=ldrcandidate): + break + else: + retrythread = eventlet.spawn_after(5 + random.random(), + start_collective) + except Exception as e: + retrythread = eventlet.spawn_after(random.random(), start_collective) diff --git a/confluent_server/confluent/config/configmanager.py b/confluent_server/confluent/config/configmanager.py index 06b29e26..c8056402 100644 --- a/confluent_server/confluent/config/configmanager.py +++ b/confluent_server/confluent/config/configmanager.py @@ -409,7 +409,10 @@ def _push_rpc(stream, payload): return True except Exception: logException() - del cfgstreams[stream] + try: + del cfgstreams[stream] + except KeyError: + pass if membership_callback: membership_callback() stream.close()