PostgreSQL database high availability Patroni source code learning - etcd class

Etcd Class inherits from AbstractEtcd abstract class ,AbstractEtcd Abstract classes inherit from AbstractDCS（AbstractEtcd The constructor of will pass in EtcdClient class ）.Etcd The constructor of the class will first call the constructor of the parent class , And then __do_not_watch Members are set to False.

class Etcd(AbstractEtcd):
    def __init__(self, config):
        super(Etcd, self).__init__(config, EtcdClient, (etcd.EtcdLeaderElectionInProgress, EtcdRaftInternal))
        self.__do_not_watch = False

Function name	purpose	Code
touch_member(self, data, permanent=False)		self._client.set(self.member_path, json.dumps(data, separators=(’,’, ‘:’)), None if permanent else self._ttl)
take_leader(self)		self.retry(self._client.write, self.leader_path, self._name, ttl=self._ttl)
set_failover_value(self, value, index=None)		self._client.write(self.failover_path, value, prevIndex=index or 0)
set_config_value(self, value, index=None)		self._client.write(self.config_path, value, prevIndex=index or 0)
_write_leader_optime(self, last_lsn)		self._client.set(self.leader_optime_path, last_lsn)
_write_status(self, value)		self._client.set(self.status_path, value)
_update_leader(self)		self.retry(self._client.write, self.leader_path, self._name, prevValue=self._name, ttl=self._ttl)
_delete_leader(self)		self._client.delete(self.leader_path, prevValue=self._name)
initialize(self, create_new=True, sysid="")		self.retry(self._client.write, self.initialize_path, sysid, prevExist=(not create_new))
cancel_initialization(self)		self.retry(self._client.delete, self.initialize_path)
delete_cluster(self)		self.retry(self._client.delete, self.client_path(’’), recursive=True)
set_history_value(self, value)		self._client.write(self.history_path, value)
set_sync_state_value(self, value, index=None)		self.retry(self._client.write, self.sync_path, value, prevIndex=index or 0)
delete_sync_state(self, index=None)		self.retry(self._client.delete, self.sync_path, prevIndex=index or 0)

    def attempt_to_acquire_leader(self, permanent=False):
        try:
            return bool(self.retry(self._client.write, self.leader_path, self._name, ttl=None if permanent else self._ttl, prevExist=False))
        except etcd.EtcdAlreadyExist:
            logger.info('Could not take out TTL lock')
        except (RetryFailedError, etcd.EtcdException):
            pass
        return False

watch Member functions

watch Member functions

    def watch(self, leader_index, timeout):
        if self.__do_not_watch:
            self.__do_not_watch = False
            return True
        if leader_index:
            end_time = time.time() + timeout
            while timeout >= 1:  # when timeout is too small urllib3 doesn't have enough time to connect
                try:
                    result = self._client.watch(self.leader_path, index=leader_index, timeout=timeout + 0.5)
                    self._has_failed = False
                    if result.action == 'compareAndSwap':
                        time.sleep(0.01)
                    # Synchronous work of all cluster members with etcd is less expensive
                    # than reestablishing http connection every time from every replica.
                    return True
                except etcd.EtcdWatchTimedOut:
                    self._has_failed = False
                    return False
                except (etcd.EtcdEventIndexCleared, etcd.EtcdWatcherCleared):  # Watch failed
                    self._has_failed = False
                    return True  # leave the loop, because watch with the same parameters will fail anyway
                except etcd.EtcdException as e:
                    self._handle_exception(e, 'watch', True)
                timeout = end_time - time.time()
        try:
            return super(Etcd, self).watch(None, timeout)
        finally:
            self.event.clear()