Merge pull request #69691 from guits/fix-staggered-upgrade

cephadm: fix staggered upgrade

src/pybind/mgr/cephadm/tests/test_upgrade.py

@@ -1052,3 +1052,121 @@ def test_staggered_upgrade_validation(
     else:
         cephadm_module.upgrade._validate_upgrade_filters(
             'new_image_name', daemon_types, hosts, services)
+
+
+@mock.patch("cephadm.module.HostCache.get_daemons")
+def test_filtered_scope_up_to_date(
+    get_daemons: mock.MagicMock,
+    cephadm_module: CephadmOrchestrator,
+) -> None:
+    target_digest = 'new_image@repo_digest'
+    old_digest = 'old_image@repo_digest'
+
+    cephadm_module.upgrade.upgrade_state = UpgradeState(
+        'target_image',
+        '0',
+        target_digests=[target_digest],
+        daemon_types=['mon'],
+        hosts=['trial031'],
+    )
+
+    get_daemons.return_value = [
+        DaemonDescription(
+            daemon_type='mon',
+            daemon_id='a',
+            hostname='trial031',
+            container_image_digests=[target_digest],
+        ),
+        DaemonDescription(
+            daemon_type='mon',
+            daemon_id='c',
+            hostname='trial031',
+            container_image_digests=[old_digest],
+        ),
+    ]
+
+    assert not cephadm_module.upgrade._filtered_scope_up_to_date(
+        [target_digest], 'target_image',
+    )
+
+    get_daemons.return_value[1].container_image_digests = [target_digest]
+    assert cephadm_module.upgrade._filtered_scope_up_to_date(
+        [target_digest], 'target_image',
+    )
+
+
+@mock.patch.object(CephadmUpgrade, '_mark_upgrade_complete')
+@mock.patch.object(CephadmUpgrade, '_filtered_scope_up_to_date', return_value=False)
+@mock.patch.object(CephadmUpgrade, '_handle_need_upgrade_self')
+@mock.patch.object(CephadmUpgrade, '_to_upgrade', return_value=(True, []))
+@mock.patch.object(
+    CephadmUpgrade, '_detect_need_upgrade', return_value=(False, [], [], 0),
+)
+@mock.patch.object(CephadmUpgrade, '_set_container_images')
+@mock.patch.object(CephadmUpgrade, '_complete_osd_upgrade')
+@mock.patch.object(CephadmUpgrade, '_complete_mds_upgrade')
+@mock.patch.object(CephadmUpgrade, '_update_upgrade_progress')
+@mock.patch.object(CephadmUpgrade, 'get_distinct_container_image_settings', return_value={})
+@mock.patch("cephadm.module.CephadmOrchestrator.lookup_release_name", return_value='tentacle')
+@mock.patch("cephadm.module.CephadmOrchestrator.check_mon_command", return_value=(0, '{}', ''))
+@mock.patch("cephadm.module.CephadmOrchestrator.set_container_image")
+@mock.patch("cephadm.module.CephadmOrchestrator.get_active_mgr_digests")
+@mock.patch("cephadm.module.CephadmOrchestrator.get", return_value={
+    'min_mon_release': 19,
+    'require_osd_release': 'tentacle',
+    'have_local_config_map': True,
+})
+@mock.patch(
+    "cephadm.module.CephadmOrchestrator.version",
+    new_callable=mock.PropertyMock,
+    return_value='ceph version 19.3.0-0 (hash)',
+)
+@mock.patch("cephadm.module.HostCache.get_daemons")
+def test_do_upgrade_limit_exhausted_marks_complete_without_scope_check(
+    get_daemons: mock.MagicMock,
+    _version: mock.MagicMock,
+    _get: mock.MagicMock,
+    get_active_mgr_digests: mock.MagicMock,
+    _set_container_image: mock.MagicMock,
+    _check_mon_command: mock.MagicMock,
+    _lookup_release_name: mock.MagicMock,
+    _get_distinct_container_image_settings: mock.MagicMock,
+    _update_upgrade_progress: mock.MagicMock,
+    _complete_mds_upgrade: mock.MagicMock,
+    _complete_osd_upgrade: mock.MagicMock,
+    _set_container_images: mock.MagicMock,
+    _detect_need_upgrade: mock.MagicMock,
+    _to_upgrade: mock.MagicMock,
+    _handle_need_upgrade_self: mock.MagicMock,
+    filtered_scope: mock.MagicMock,
+    mark_complete: mock.MagicMock,
+    cephadm_module: CephadmOrchestrator,
+) -> None:
+    target_digest = 'new_image@repo_digest'
+    old_digest = 'old_image@repo_digest'
+    get_active_mgr_digests.return_value = [target_digest]
+    get_daemons.return_value = [
+        DaemonDescription(
+            daemon_type='osd',
+            daemon_id=str(i),
+            hostname='host1',
+            container_image_digests=[target_digest] if i < 2 else [old_digest],
+        )
+        for i in range(8)
+    ]
+
+    cephadm_module.upgrade.upgrade_state = UpgradeState(
+        'target_image',
+        '0',
+        target_id='image_id',
+        target_digests=[target_digest],
+        target_version='19.3.0-0',
+        daemon_types=['osd'],
+        total_count=2,
+        remaining_count=0,
+    )
+
+    cephadm_module.upgrade._do_upgrade()
+
+    mark_complete.assert_called_once()
+    filtered_scope.assert_not_called()

src/pybind/mgr/cephadm/upgrade.py

@@ -1692,6 +1692,46 @@ class CephadmUpgrade:
             self.upgrade_state.fs_original_allow_standby_replay = {}
             self._save_upgrade_state()
 
+    def _filtered_scope_up_to_date(
+        self,
+        target_digests: Optional[List[str]],
+        target_name: str,
+    ) -> bool:
+        assert self.upgrade_state is not None
+        if target_digests is None:
+            target_digests = []
+
+        if self.mgr.use_agent:
+            hosts: Set[str] = set()
+            if self.upgrade_state.hosts is not None:
+                hosts.update(self.upgrade_state.hosts)
+            for d in self._get_filtered_daemons():
+                if d.hostname is not None:
+                    hosts.add(d.hostname)
+            for hostname in hosts:
+                if not self.mgr.cache.host_metadata_up_to_date(hostname):
+                    logger.info(
+                        'Upgrade: Waiting for host %s metadata before completing',
+                        hostname,
+                    )
+                    self.mgr.agent_helpers._request_ack_all_not_up_to_date()
+                    return False
+
+        for d in self._get_filtered_daemons():
+            if d.daemon_type not in CEPH_IMAGE_TYPES:
+                continue
+            if (
+                (self.mgr.use_repo_digest and d.matches_digests(target_digests))
+                or (not self.mgr.use_repo_digest and d.matches_image_name(target_name))
+            ):
+                continue
+            logger.info(
+                'Upgrade: Waiting for %s to match target image before completing',
+                d.name(),
+            )
+            return False
+        return True
+
     def _mark_upgrade_complete(self) -> None:
         if not self.upgrade_state:
             logger.debug('_mark_upgrade_complete upgrade already marked complete, exiting')
@@ -1948,5 +1988,17 @@ class CephadmUpgrade:
                 'who': name_to_config_section(daemon_type),
             })
 
+        # Limited (--limit) upgrades end when the batch quota is exhausted,
+        # even if other daemons in the filter still need the target image.
+        if (
+            self.upgrade_state.remaining_count is not None
+            and self.upgrade_state.remaining_count <= 0
+        ):
+            self._mark_upgrade_complete()
+            return
+        if not self._filtered_scope_up_to_date(
+            target_digests, self.upgrade_state._target_name,
+        ):
+            return
         self._mark_upgrade_complete()
         return