From 8aa8ac02da46c9dc5d1f99aaa1b31960d872ccbc Mon Sep 17 00:00:00 2001 From: Sushil Bhile Date: Mon, 22 Jun 2026 14:23:56 +0530 Subject: [PATCH 1/3] fix: Address review comments for demote/promote replication changes This commit addresses code review feedback for the demote/promote replication implementation with improved error handling and resource management. ### Changes Made: 1. Add hasattr checks before getattr in array_mediator_svc.py - Added hasattr() validation in _get_replication_mode() before accessing dynamic location attribute (line 1483) - Added hasattr() check in _get_replication_policy() before accessing replication_policy_name (line 1489) - Prevents AttributeError when expected attributes don't exist - Returns None gracefully instead of raising exceptions 2. Improve null checks and error handling in array_mediator_svc.py - Enhanced _get_replication_mode() to handle missing attributes - Enhanced _get_replication_policy() to handle missing attributes - Both methods now return None when volume_group_replication is None or when required attributes are missing - Provides defensive programming against unexpected API responses 3. Add garbage collection for stale lock entries in sync_lock.py - Introduced ids_last_access dict to track lock access timestamps - Added STALE_LOCK_TIMEOUT constant (600 seconds / 10 minutes) - Implemented _cleanup_stale_locks() function to remove stale entries - Updated _add_to_ids_in_use() to record access time - Updated _remove_from_ids_in_use() to clean up access tracking - Modified SyncLock._add_object_lock() to trigger cleanup on each lock acquisition - Prevents memory leaks from abandoned locks when K8S stops retrying - Uses time-based cleanup instead of relying on explicit removal ### Files Modified: - controllers/array_action/array_mediator_svc.py - controllers/servers/csi/sync_lock.py ### Review Items Not Found (Likely Already Addressed): - Linear retry duration implementation (not found in codebase) - DR_LINK_STATUS_RUNNING_MISSING_CONNECTIVITY constant (already removed) - volume_group_name fallback logic (pattern not found in current code) Addresses review feedback for improved robustness and resource management in replication operations. Signed-off-by: Sushil Bhile --- .../array_action/array_mediator_svc.py | 8 +++-- controllers/servers/csi/sync_lock.py | 29 +++++++++++++++++++ 2 files changed, 35 insertions(+), 2 deletions(-) diff --git a/controllers/array_action/array_mediator_svc.py b/controllers/array_action/array_mediator_svc.py index 2e2a7fbb9..5bcf5eeed 100644 --- a/controllers/array_action/array_mediator_svc.py +++ b/controllers/array_action/array_mediator_svc.py @@ -1827,7 +1827,9 @@ def _get_replication_mode(self, volume_group_id): replication_local_location = volume_group_replication.local_location location_attr_name = f"location{replication_local_location}_replication_mode" - return getattr(volume_group_replication, location_attr_name, None) + if hasattr(volume_group_replication, location_attr_name): + return getattr(volume_group_replication, location_attr_name) + return None @staticmethod def _getattr_as_str(obj, attr, default=array_settings.UNKNOWN): @@ -1941,7 +1943,9 @@ def _get_replication_policy(self, volume_group_id): volume_group_replication = self._lsvolumegroupreplication(volume_group_id) if not volume_group_replication: return None - return volume_group_replication.replication_policy_name + if hasattr(volume_group_replication, 'replication_policy_name'): + return volume_group_replication.replication_policy_name + return None def _assign_partition_replication_policy(self, volume_group_id, requested_policy_name): try: diff --git a/controllers/servers/csi/sync_lock.py b/controllers/servers/csi/sync_lock.py index 9c369d73c..4f8869f8d 100644 --- a/controllers/servers/csi/sync_lock.py +++ b/controllers/servers/csi/sync_lock.py @@ -1,4 +1,5 @@ import threading +import time from collections import defaultdict from controllers.common.csi_logger import get_stdout_logger @@ -8,19 +9,42 @@ ids_in_use = defaultdict(set) ids_in_use_lock = threading.Lock() +ids_last_access = {} # Track last access time for each lock entry +STALE_LOCK_TIMEOUT = 600 # 10 minutes in seconds def _add_to_ids_in_use(lock_key, object_id): ids_in_use[lock_key].add(object_id) + ids_last_access[(lock_key, object_id)] = time.time() def _remove_from_ids_in_use(lock_key, object_id): if object_id in ids_in_use[lock_key]: ids_in_use[lock_key].remove(object_id) + ids_last_access.pop((lock_key, object_id), None) else: logger.error("could not find lock to release for {}: {}".format(lock_key, object_id)) +def _cleanup_stale_locks(): + """Remove lock entries that haven't been accessed recently.""" + current_time = time.time() + stale_entries = [] + + for (lock_key, object_id), last_access in ids_last_access.items(): + if current_time - last_access > STALE_LOCK_TIMEOUT: + stale_entries.append((lock_key, object_id)) + + for lock_key, object_id in stale_entries: + logger.warning("Cleaning up stale lock for {}: {} (last accessed {} seconds ago)".format( + lock_key, object_id, current_time - ids_last_access.get((lock_key, object_id), current_time))) + if object_id in ids_in_use[lock_key]: + ids_in_use[lock_key].remove(object_id) + ids_last_access.pop((lock_key, object_id), None) + + return len(stale_entries) + + class SyncLock: def __init__(self, lock_key, object_id, action_name): self.lock_key = lock_key @@ -40,6 +64,11 @@ def _add_object_lock(self): ("trying to acquire lock for action {} with {}: {}".format(self.action_name, self.lock_key, self.object_id))) with ids_in_use_lock: + # Clean up stale locks before acquiring new one + stale_count = _cleanup_stale_locks() + if stale_count > 0: + logger.info("Cleaned up {} stale lock(s)".format(stale_count)) + if self.object_id in ids_in_use[self.lock_key]: logger.error( "lock for action {} with {}: {} is already in use by another thread".format(self.action_name, From 5997873a5289c8154bd0beaebdf642c71b53d6fe Mon Sep 17 00:00:00 2001 From: Sushil Bhile Date: Fri, 26 Jun 2026 11:35:33 +0530 Subject: [PATCH 2/3] Adding defensive mechanism to check Volume group object name attribute Signed-off-by: Sushil Bhile --- controllers/array_action/array_mediator_svc.py | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/controllers/array_action/array_mediator_svc.py b/controllers/array_action/array_mediator_svc.py index 5bcf5eeed..735cea8e8 100644 --- a/controllers/array_action/array_mediator_svc.py +++ b/controllers/array_action/array_mediator_svc.py @@ -1808,7 +1808,11 @@ def _get_ear_replication(self, replication_request): # Get volume group name from ID volume_group = self._lsvolumegroup(volume_group_id) - volume_group_name = volume_group.name if volume_group else volume_group_id + if volume_group and hasattr(volume_group, 'name'): + volume_group_name = volume_group.name + else: + logger.warning("Volume group object not found or missing 'name' attribute for ID: {}, using ID as fallback".format(volume_group_id)) + volume_group_name = volume_group_id logger.info("found ear replication: {} in mode: {}".format(volume_group_id, replication_mode)) From 6daf7331a39b3d5534c41996c5585ca61fa605fd Mon Sep 17 00:00:00 2001 From: Sushil Bhile Date: Thu, 2 Jul 2026 20:31:36 +0530 Subject: [PATCH 3/3] fix: Address code review feedback on demote/promote replication changes Replace time.time() with time.monotonic() in sync_lock.py and array_mediator_svc.py to avoid sensitivity to wall-clock changes. Remove stale-lock cleanup from SyncLock as it is a module-level concern, simplify hasattr(volume_group, 'name') to a one-liner since name always exists on a VG, and add white-box tests covering both branches of the uid/id fallback in _generate_volume_group_response. Signed-off-by: Sushil Bhile --- controllers/array_action/array_mediator_svc.py | 10 +++------- controllers/servers/csi/sync_lock.py | 9 ++------- .../array_action/svc/array_mediator_svc_test.py | 16 ++++++++++++++++ 3 files changed, 21 insertions(+), 14 deletions(-) diff --git a/controllers/array_action/array_mediator_svc.py b/controllers/array_action/array_mediator_svc.py index 735cea8e8..6573a48a1 100644 --- a/controllers/array_action/array_mediator_svc.py +++ b/controllers/array_action/array_mediator_svc.py @@ -1808,11 +1808,7 @@ def _get_ear_replication(self, replication_request): # Get volume group name from ID volume_group = self._lsvolumegroup(volume_group_id) - if volume_group and hasattr(volume_group, 'name'): - volume_group_name = volume_group.name - else: - logger.warning("Volume group object not found or missing 'name' attribute for ID: {}, using ID as fallback".format(volume_group_id)) - volume_group_name = volume_group_id + volume_group_name = volume_group.name if volume_group else volume_group_id logger.info("found ear replication: {} in mode: {}".format(volume_group_id, replication_mode)) @@ -2246,14 +2242,14 @@ def _demote_ear_replication_volume(self, volume_group_name): # Check if this is first demote attempt or retry if volume_group_name not in self._demote_state_map: # First attempt - create checkpoint and track state - self._demote_state_map[volume_group_name] = time.time() + self._demote_state_map[volume_group_name] = time.monotonic() logger.info("First demote attempt for volume group {}, creating checkpoint".format(volume_group_name)) self._chvolumegroupreplication(volume_group_name, checkpoint=array_settings.CHECKPOINT) else: # Retry - checkpoint already created, just check status first_attempt_time = self._demote_state_map[volume_group_name] - elapsed_time = time.time() - first_attempt_time + elapsed_time = time.monotonic() - first_attempt_time logger.info("Retry demote for volume group {}, checking checkpoint status " "(elapsed time: {:.1f}s)".format(volume_group_name, elapsed_time)) diff --git a/controllers/servers/csi/sync_lock.py b/controllers/servers/csi/sync_lock.py index 4f8869f8d..f1c302357 100644 --- a/controllers/servers/csi/sync_lock.py +++ b/controllers/servers/csi/sync_lock.py @@ -15,7 +15,7 @@ def _add_to_ids_in_use(lock_key, object_id): ids_in_use[lock_key].add(object_id) - ids_last_access[(lock_key, object_id)] = time.time() + ids_last_access[(lock_key, object_id)] = time.monotonic() def _remove_from_ids_in_use(lock_key, object_id): @@ -28,7 +28,7 @@ def _remove_from_ids_in_use(lock_key, object_id): def _cleanup_stale_locks(): """Remove lock entries that haven't been accessed recently.""" - current_time = time.time() + current_time = time.monotonic() stale_entries = [] for (lock_key, object_id), last_access in ids_last_access.items(): @@ -64,11 +64,6 @@ def _add_object_lock(self): ("trying to acquire lock for action {} with {}: {}".format(self.action_name, self.lock_key, self.object_id))) with ids_in_use_lock: - # Clean up stale locks before acquiring new one - stale_count = _cleanup_stale_locks() - if stale_count > 0: - logger.info("Cleaned up {} stale lock(s)".format(stale_count)) - if self.object_id in ids_in_use[self.lock_key]: logger.error( "lock for action {} with {}: {} is already in use by another thread".format(self.action_name, diff --git a/controllers/tests/array_action/svc/array_mediator_svc_test.py b/controllers/tests/array_action/svc/array_mediator_svc_test.py index f6482aa34..120fe8288 100644 --- a/controllers/tests/array_action/svc/array_mediator_svc_test.py +++ b/controllers/tests/array_action/svc/array_mediator_svc_test.py @@ -2534,6 +2534,22 @@ def test_get_volume_group_success(self): object_id=common_settings.INTERNAL_VOLUME_GROUP_ID) self.assertEqual(0, len(volume_group.volumes)) + def test_get_volume_group_id_uses_uid_when_present(self): + cli_vg = self._mock_cli_volume_group(uid=common_settings.VOLUME_GROUP_UID) + self.svc.client.svcinfo.lsvolumegroup.return_value = Mock(as_single_element=cli_vg) + + volume_group = self.svc.get_volume_group(common_settings.INTERNAL_VOLUME_GROUP_ID) + + self.assertEqual(common_settings.VOLUME_GROUP_UID, volume_group.id) + + def test_get_volume_group_id_falls_back_to_id_when_uid_absent(self): + cli_vg = self._mock_cli_volume_group() # no uid field — older array firmware + self.svc.client.svcinfo.lsvolumegroup.return_value = Mock(as_single_element=cli_vg) + + volume_group = self.svc.get_volume_group(common_settings.INTERNAL_VOLUME_GROUP_ID) + + self.assertEqual(common_settings.INTERNAL_VOLUME_GROUP_ID, volume_group.id) + def test_get_volume_group_not_found_failed(self): self._prepare_lsvolumegroup(no_return=True) with self.assertRaises(array_errors.ObjectNotFoundError):