zuul/zuul
Code Issues Proposed changes

Merge "Delete LockableZKObject lock paths"

Browse Source
This commit is contained in:
Zuul
2024-12-02 20:56:55 +00:00
committed by Gerrit Code Review
parent 0d09c87690 50e1338c32
commit 41845f9632
7 changed files with 165 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_launcher.py
+1 -1
View File
@@ -773,7 +773,7 @@ class TestMinReadyLauncher(LauncherBaseTestCase):
node.acquireLock(ctx)
node.updateAttributes(ctx, state_time=0)
finally:
node.releaseLock()
node.releaseLock(ctx)
for _ in iterate_timeout(60, "node to be cleaned up"):
nodes = self.launcher.api.nodes_cache.getItems()
tests/unit/test_zk.py
+6 -6
View File
@@ -2659,7 +2659,7 @@ class TestLauncherApi(ZooKeeperBaseTestCase):
self.assertIsNotNone(request._zstat)
self.assertIsNotNone(request.getPath())
self.assertIsNotNone(request.acquireLock(self.zk_client))
self.assertIsNotNone(request.acquireLock(context))
self.assertTrue(request.hasLock())
for _ in iterate_timeout(10, "request to be locked"):
if request.is_locked:
@@ -2687,7 +2687,7 @@ class TestLauncherApi(ZooKeeperBaseTestCase):
# "Fulfill" requested provider nodes
for node in self.api.getMatchingProviderNodes():
self.assertIsNotNone(node.acquireLock(self.zk_client))
self.assertIsNotNone(node.acquireLock(context))
self.assertTrue(node.hasLock())
for _ in iterate_timeout(10, "node to be locked"):
if node.is_locked:
@@ -2696,7 +2696,7 @@ class TestLauncherApi(ZooKeeperBaseTestCase):
context,
state=model.ProviderNode.State.READY
)
node.releaseLock()
node.releaseLock(context)
self.assertFalse(node.hasLock())
for _ in iterate_timeout(10, "node to be unlocked"):
if not node.is_locked:
@@ -2717,7 +2717,7 @@ class TestLauncherApi(ZooKeeperBaseTestCase):
context,
state=model.NodesetRequest.State.FULFILLED,
)
request.releaseLock()
request.releaseLock(context)
self.assertFalse(request.hasLock())
for _ in iterate_timeout(10, "request to be unlocked"):
if not request.is_locked:
@@ -2742,7 +2742,7 @@ class TestLauncherApi(ZooKeeperBaseTestCase):
# Mark nodes as used
for node in provider_nodes:
self.assertIsNotNone(node.acquireLock(self.zk_client))
self.assertIsNotNone(node.acquireLock(context))
for _ in iterate_timeout(10, "wait for lock to show up"):
if node.is_locked:
break
@@ -2750,7 +2750,7 @@ class TestLauncherApi(ZooKeeperBaseTestCase):
context,
state=model.ProviderNode.State.USED,
)
node.releaseLock()
node.releaseLock(context)
# Cleanup used nodes and wait for them to be removed from the cache
for _ in iterate_timeout(10, "nodes to be removed"):
zuul/launcher/client.py
+10 -10
View File
@@ -134,19 +134,19 @@ class LauncherClient:
provider_node = getattr(node, "_provider_node", None)
if not provider_node:
continue
try:
node.state = STATE_USED
with self.createZKContext(provider_node._lock, log) as ctx:
with self.createZKContext(provider_node._lock, log) as ctx:
try:
node.state = STATE_USED
provider_node.updateAttributes(
ctx, state=ProviderNode.State.USED)
log.debug("Released %s", provider_node)
except Exception:
log.exception("Unable to return node %s", provider_node)
finally:
try:
provider_node.releaseLock()
log.debug("Released %s", provider_node)
except Exception:
log.exception("Error unlocking node %s", provider_node)
log.exception("Unable to return node %s", provider_node)
finally:
try:
provider_node.releaseLock(ctx)
except Exception:
log.exception("Error unlocking node %s", provider_node)
def createZKContext(self, lock, log):
return ZKContext(self.zk_client, lock, self.stop_event, log)
zuul/launcher/server.py
+59 -52
View File
@@ -103,48 +103,50 @@ class UploadJob:
# and skip download
acquired = []
path = None
try:
with self.launcher.createZKContext(None, self.log) as ctx:
try:
with self.image_build_artifact.locked(
self.launcher.zk_client, blocking=False):
for upload in self.uploads:
if upload.acquireLock(
self.launcher.zk_client, blocking=False):
acquired.append(upload)
self.log.debug("Acquired upload lock for %s",
try:
with self.image_build_artifact.locked(ctx, blocking=False):
for upload in self.uploads:
if upload.acquireLock(ctx, blocking=False):
acquired.append(upload)
self.log.debug("Acquired upload lock for %s",
upload)
except LockException:
return
if not acquired:
return
path = self.launcher.downloadArtifact(
self.image_build_artifact)
futures = []
for upload in acquired:
future = self.launcher.endpoint_upload_executor.submit(
EndpointUploadJob(
self.launcher, self.image_build_artifact,
upload, path).run)
futures.append((upload, future))
for upload, future in futures:
try:
future.result()
self.log.info("Finished upload %s", upload)
except Exception:
self.log.exception("Unable to upload image %s", upload)
finally:
for upload in acquired:
try:
upload.releaseLock(ctx)
self.log.debug("Released upload lock for %s", upload)
except Exception:
self.log.exception("Unable to release lock for %s",
upload)
except LockException:
return
if not acquired:
return
path = self.launcher.downloadArtifact(self.image_build_artifact)
futures = []
for upload in acquired:
future = self.launcher.endpoint_upload_executor.submit(
EndpointUploadJob(self.launcher, self.image_build_artifact,
upload, path).run)
futures.append((upload, future))
for upload, future in futures:
try:
future.result()
self.log.info("Finished upload %s", upload)
except Exception:
self.log.exception("Unable to upload image %s", upload)
finally:
for upload in acquired:
try:
upload.releaseLock()
self.log.debug("Released upload lock for %s", upload)
except Exception:
self.log.exception("Unable to release lock for %s", upload)
if path:
try:
os.unlink(path)
self.log.info("Deleted %s", path)
except Exception:
self.log.exception("Unable to delete %s", path)
if path:
try:
os.unlink(path)
self.log.info("Deleted %s", path)
except Exception:
self.log.exception("Unable to delete %s", path)
class EndpointUploadJob:
@@ -324,9 +326,11 @@ class Launcher:
# Nothing to do here
continue
log.debug("Got request %s", request)
if not request.acquireLock(self.zk_client, blocking=False):
log.debug("Failed to lock matching request %s", request)
continue
with self.createZKContext(None, log) as ctx:
if not request.acquireLock(ctx, blocking=False):
log.debug("Failed to lock matching request %s",
request)
continue
if not self._cachesReadyForRequest(request):
self.log.debug("Caches are not up-to-date for %s", request)
@@ -350,7 +354,8 @@ class Launcher:
except Exception:
log.exception("Error processing request %s", request)
if request.state in request.FINAL_STATES:
request.releaseLock()
with self.createZKContext(None, log) as ctx:
request.releaseLock(ctx)
def _cachesReadyForRequest(self, request):
# Make sure we have all associated provider nodes in the cache
@@ -384,7 +389,7 @@ class Launcher:
else:
continue
if not node.acquireLock(self.zk_client, blocking=False):
if not node.acquireLock(ctx, blocking=False):
log.debug("Failed to lock matching ready node %s",
node)
continue
@@ -406,7 +411,7 @@ class Launcher:
log.exception("Faild to assign ready node %s", node)
continue
finally:
node.releaseLock()
node.releaseLock(ctx)
else:
node = self._requestNode(
label, request, provider, log, ctx)
@@ -525,9 +530,10 @@ class Launcher:
if not self._isNodeActionable(node):
continue
if not node.acquireLock(self.zk_client, blocking=False):
log.debug("Failed to lock matching node %s", node)
continue
with self.createZKContext(None, log) as ctx:
if not node.acquireLock(ctx, blocking=False):
log.debug("Failed to lock matching node %s", node)
continue
request = self.api.getNodesetRequest(node.request_id)
if ((request or node.request_id is None)
@@ -583,7 +589,8 @@ class Launcher:
self.wake_event.set()
if node.state == model.ProviderNode.State.READY:
node.releaseLock()
with self.createZKContext(None, self.log) as ctx:
node.releaseLock(ctx)
def _isNodeActionable(self, node):
if node.is_locked:
@@ -634,7 +641,7 @@ class Launcher:
node.setState(node.State.READY)
self.wake_event.set()
log.debug("Marking node %s as %s", node, node.state)
node.releaseLock()
node.releaseLock(ctx)
def _cleanupNode(self, node, log):
with self.createZKContext(node._lock, self.log) as ctx:
@@ -671,7 +678,7 @@ class Launcher:
if not self.api.getNodesetRequest(node.request_id):
log.debug("Removing provider node %s", node)
node.delete(ctx)
node.releaseLock()
node.releaseLock(ctx)
def _processMinReady(self):
if not self.api.nodes_cache.waitForSync(
zuul/zk/launcher.py
+9 -8
View File
@@ -245,12 +245,13 @@ class LauncherApi:
continue
if self.getNodesetRequest(node.request_id):
continue
if lock := node.acquireLock(self.zk_client):
try:
with self.createZKContext(None) as outer_ctx:
if lock := node.acquireLock(outer_ctx):
with self.createZKContext(lock) as ctx:
node.delete(ctx)
except NoNodeError:
# Node is already deleted
pass
finally:
node.releaseLock()
try:
node.delete(ctx)
except NoNodeError:
# Node is already deleted
pass
finally:
node.releaseLock(ctx)
zuul/zk/locks.py
+8 -1
View File
@@ -28,15 +28,22 @@ CONNECTION_LOCK_ROOT = f"{LOCK_ROOT}/connection"
class SessionAwareMixin:
def __init__(self, client, path, identifier=None, extra_lock_patterns=()):
def __init__(self, client, path, identifier=None, extra_lock_patterns=(),
ensure_path=True):
self._zuul_ephemeral = None
self._zuul_session_expired = False
self._zuul_watching_session = False
self._zuul_seen_contenders = set()
self._zuul_seen_contender_names = set()
self._zuul_contender_watch = None
self._zuul_ensure_path = ensure_path
super().__init__(client, path, identifier, extra_lock_patterns)
def _ensure_path(self):
# Override
if self._zuul_ensure_path:
return super()._ensure_path()
def acquire(self, blocking=True, timeout=None, ephemeral=True):
ret = super().acquire(blocking, timeout, ephemeral)
self._zuul_session_expired = False
zuul/zk/zkobject.py
+72 -9
View File
@@ -19,11 +19,17 @@ import contextlib
import json
import logging
import sys
import time
import types
import zlib
import collections
from kazoo.exceptions import LockTimeout, NodeExistsError, NoNodeError
from kazoo.exceptions import (
LockTimeout,
NoNodeError,
NodeExistsError,
NotEmptyError,
)
from kazoo.retry import KazooRetry
from zuul.zk import sharding
@@ -141,6 +147,7 @@ class ZKObject:
_retry_interval = 5
_zkobject_compressed_size = 0
_zkobject_uncompressed_size = 0
_deleted = False
io_reader_class = sharding.RawZKIO
io_writer_class = sharding.RawZKIO
truncate_on_create = False
@@ -293,6 +300,7 @@ class ZKObject:
context.log.error(
"Exception deleting ZKObject %s at %s", self, path)
raise
self._set(_deleted=True)
def estimateDataSize(self, seen=None):
"""Attempt to find all ZKObjects below this one and sum their
@@ -513,32 +521,80 @@ class LockableZKObject(ZKObject):
"""
raise NotImplementedError()
@classmethod
def new(klass, context, **kw):
"""Create a new instance and save it in ZooKeeper"""
obj = klass()
obj._set(**kw)
# Create the lock path first. In the future, if we want to
# support creating locked objects, we can acquire it here.
obj._createLockPath(context)
data = obj._trySerialize(context)
obj._save(context, data, create=True)
return obj
def _createLockPath(self, context):
if isinstance(context, LocalZKContext):
return
path = self.getLockPath()
if context.sessionIsInvalid():
raise Exception("ZooKeeper session or lock not valid")
try:
self._retry(context, context.client.ensure_path, path)
except Exception:
context.log.error(
"Exception creating ZKObject %s lock path at %s", self, path)
raise
def _deleteLockPath(self, client):
path = self.getLockPath()
# Give other actors 30 seconds to realize this object
# doesn't exist anymore and they should stop trying to
# lock it.
for x in range(30):
# This handles connection-related retries, but not
# conflicts. We ignore the session here because we're
# trying to delete a lock; we want to do that even if we
# lose whatever lock the context might have.
kazoo_retry = KazooRetry(max_tries=-1,
delay=self._retry_interval,
backoff=0)
try:
return kazoo_retry(client.delete, path, recursive=True)
except NotEmptyError:
time.sleep(1)
@contextmanager
def locked(self, zk_client, blocking=True, timeout=None):
if not (lock := self.acquireLock(zk_client, blocking=blocking,
def locked(self, context, blocking=True, timeout=None):
if not (lock := self.acquireLock(context, blocking=blocking,
timeout=timeout)):
raise LockException(f"Failed to acquire lock on {self}")
try:
yield lock
finally:
try:
self.releaseLock()
self.releaseLock(context)
except Exception:
self.log.exception("Failed to release lock on %s", self)
context.log.exception("Failed to release lock on %s", self)
def acquireLock(self, zk_client, blocking=True, timeout=None):
def acquireLock(self, context, blocking=True, timeout=None):
have_lock = False
lock = None
path = self.getLockPath()
try:
lock = SessionAwareLock(zk_client.client, path)
# We create the lock path when we create the object in ZK,
# so there is no need to ensure the path on lock. This
# lets us avoid re-creating the lock if the object was
# deleted behind our back.
lock = SessionAwareLock(context.client, path,
ensure_path=False)
have_lock = lock.acquire(blocking, timeout)
except NoNodeError:
# Request disappeared
have_lock = False
except LockTimeout:
have_lock = False
self.log.error("Timeout trying to acquire lock: %s", path)
context.log.error("Timeout trying to acquire lock: %s", path)
# If we aren't blocking, it's possible we didn't get the lock
# because someone else has it.
@@ -548,11 +604,18 @@ class LockableZKObject(ZKObject):
self._set(_lock=lock)
return lock
def releaseLock(self):
def releaseLock(self, context):
if self._lock is None:
return
self._lock.release()
self._set(_lock=None)
if self._deleted:
# If we are releasing the lock after deleting the object,
# also cleanup the lock path.
try:
self._deleteLockPath(context.client)
except Exception:
context.log.error("Unable to delete lock path for %s", self)
def hasLock(self):
if self._lock is None: