Feature/more reliable zookeeper

[pvbouwel]

This change will just create 1 KazooClient and keeps it running in the background. The KazooClient itself has logic to recover connectivity.

In order to deal with intermittent failure we also retry operations if they fail due to connectivity related issues. At this time the once that came from test scenario's were kazoo.exceptions.SessionExpiredError and kazoo.exceptions.ConnectionLoss but if other retryable exceptions are identified they could be easily added.

[soxofaan]

(apparently I still had these comments from some time ago)

[soxofaan]

these are all stdlib/thirdparty (based) symbols, so I don't think it's necessary to guard this with if TYPE_CHECKING
(that's typically just a trick to dig you out of circular import holes)

[soxofaan]

as we're still on python 3.8, I don't think we can already use type annotations like tuple[...] and still have to stick to Tuple[...]

[soxofaan]

FYI: I just merged latest master in this feature branch to make sure this PR doesn't get stale

[soxofaan]

I wonder if we can't keep this usage pattern for now:

 with self._zk_client() as zk:
     ...

but return the new wrapper instead there.

That would make the introduced change of this PR smaller (in LoC).
And it would even allow to gradually roll this out (e.g. with a feature flag or config)

[soxofaan]

while reviewing this I got a bit lost in this _wait_100ms_for_zookeeper_connection/_wait_up_to_1minute_for_zookeeper_connection construction.

Can't it be simplified to a single helper like

def _wait_on_connected(self, timeout: float=60, sleep=0.1):
    end = time.time() + timeout
    while time.time() < end:
        if self._zk.connected:
            return
        if self._logger is not None:
            self._logger.debug(f"No active connection for zookeeper client ({self._zkhosts})")
        time.sleep(sleep)
    raise ZookeeperClient.NoActiveConnectionException()

(PS: I would avoid hardcoding implementation details like 100ms and 1minute in the function names to make it easier to fine-tune these when necessary)

[soxofaan]

while reviewing I also started wondering if it wouldn't be easier to just add an option for reuse to

openeo-geopyspark-driver/openeogeotrellis/user_defined_process_repository.py

Lines 83 to 94 in 4f57b9b

	@contextlib.contextmanager
	def _zk_client(self):
	kz_retry = KazooRetry(max_tries=10, delay=0.5, backoff=2)
	zk = KazooClient(hosts=self._hosts,connection_retry=kz_retry,
	command_retry=kz_retry, timeout=3.0)
	zk.start(timeout=15.0)
	try:
	yield zk
	finally:
	zk.stop()
	zk.close()

e.g. PoC:

• PoC to introduce KazooClient reuse in user_defined_process_repository #1037

that approach would also allow to roll client reuse out gradually (e.g. with config or feature flags), instead of a a complete KazooClient replacement as proposed in this PR #992