Merge pull request ceph#33116 from batrick/i43137

nautilus: pybind/mgr/volumes: idle connection drop is not working

Reviewed-by: Ramana Raja <rraja@redhat.com>
Reviewed-by: Sage Weil <sage@redhat.com>

qa/tasks/cephfs/cephfs_test_case.py

@@ -227,6 +227,15 @@ def get_session(self, client_id, session_ls=None):
     def _session_by_id(self, session_ls):
         return dict([(s['id'], s) for s in session_ls])
 
+    def wait_until_evicted(self, client_id, timeout=30):
+        def is_client_evicted():
+            ls = self._session_list()
+            for s in ls:
+                if s['id'] == client_id:
+                    return False
+            return True
+        self.wait_until_true(is_client_evicted, timeout)
+
     def wait_for_daemon_start(self, daemon_ids=None):
         """
         Wait until all the daemons appear in the FSMap, either assigned

qa/tasks/cephfs/test_volumes.py

@@ -96,6 +96,21 @@ def tearDown(self):
             self._delete_test_volume()
         super(TestVolumes, self).tearDown()
 
+    def test_connection_expiration(self):
+        # unmount any cephfs mounts
+        self.mount_a.umount_wait()
+        sessions = self._session_list()
+        self.assertLessEqual(len(sessions), 1) # maybe mgr is already mounted
+
+        # Get the mgr to definitely mount cephfs
+        subvolume = self._generate_random_subvolume_name()
+        self._fs_cmd("subvolume", "create", self.volname, subvolume)
+        sessions = self._session_list()
+        self.assertEqual(len(sessions), 1)
+
+        # Now wait for the mgr to expire the connection:
+        self.wait_until_evicted(sessions[0]['id'], timeout=90)
+
     def test_volume_rm(self):
         try:
             self._fs_cmd("volume", "rm", self.volname)

src/pybind/mgr/volumes/fs/volume.py

@@ -95,10 +95,14 @@ class RTimer(Timer):
         recurring timer variant of Timer
         """
         def run(self):
-            while not self.finished.is_set():
-                self.finished.wait(self.interval)
-                self.function(*self.args, **self.kwargs)
-            self.finished.set()
+            try:
+                while not self.finished.is_set():
+                    self.finished.wait(self.interval)
+                    self.function(*self.args, **self.kwargs)
+                self.finished.set()
+            except Exception as e:
+                log.error("ConnectionPool.RTimer: %s", e)
+                raise
 
     # TODO: make this configurable
     TIMER_TASK_RUN_INTERVAL = 30.0  # seconds
@@ -115,7 +119,7 @@ def __init__(self, mgr):
     def cleanup_connections(self):
         with self.lock:
             log.info("scanning for idle connections..")
-            idle_fs = [fs_name for fs_name,conn in self.connections.iteritems()
+            idle_fs = [fs_name for fs_name,conn in self.connections.items()
                        if conn.is_connection_idle(ConnectionPool.CONNECTION_IDLE_INTERVAL)]
             for fs_name in idle_fs:
                 log.info("cleaning up connection for '{}'".format(fs_name))