Merge pull request ceph#62038 from phlogistonjohn/jjm-cephadm-infer-69278

adk3798 · web-flow · commit 3431ac4e55c9 · 2025-04-07T18:30:53.000-04:00
cephadm: update and improve container image infer function

Reviewed-by: Adam King &lt;adking@redhat.com&gt;
diff --git a/src/cephadm/cephadm.py b/src/cephadm/cephadm.py
@@ -201,7 +201,7 @@
     MemUsageStatusUpdater,
     VersionStatusUpdater,
 )
-from cephadmlib.container_lookup import get_container_info
+from cephadmlib.container_lookup import infer_local_ceph_image
 
 
 FuncT = TypeVar('FuncT', bound=Callable)
@@ -465,51 +465,6 @@ def update_default_image(ctx: CephadmContext) -> None:
         ctx.image = _get_default_image(ctx)
 
 
-def infer_local_ceph_image(ctx: CephadmContext, container_path: str) -> Optional[str]:
-    """
-     Infer the local ceph image based on the following priority criteria:
-       1- the image specified by --image arg (if provided).
-       2- the same image as the daemon container specified by --name arg (if provided).
-       3- image used by any ceph container running on the host. In this case we use daemon types.
-       4- if no container is found then we use the most ceph recent image on the host.
-
-     Note: any selected container must have the same fsid inferred previously.
-
-    :return: The most recent local ceph image (already pulled)
-    """
-    # '|' special character is used to separate the output fields into:
-    #  - Repository@digest
-    #  - Image Id
-    #  - Image Tag
-    #  - Image creation date
-    out, _, _ = call_throws(ctx,
-                            [container_path, 'images',
-                             '--filter', 'label=ceph=True',
-                             '--filter', 'dangling=false',
-                             '--format', '{{.Repository}}@{{.Digest}}|{{.ID}}|{{.Tag}}|{{.CreatedAt}}'])
-
-    container_info = None
-    daemon_name = ctx.name if ('name' in ctx and ctx.name and '.' in ctx.name) else None
-    daemons_ls = [daemon_name] if daemon_name is not None else ceph_daemons()  # daemon types: 'mon', 'mgr', etc
-    for daemon in daemons_ls:
-        container_info = get_container_info(ctx, daemon, daemon_name is not None)
-        if container_info is not None:
-            logger.debug(f"Using container info for daemon '{daemon}'")
-            break
-
-    for image in out.splitlines():
-        if image and not image.isspace():
-            (digest, image_id, tag, created_date) = image.lstrip().split('|')
-            if container_info is not None and image_id not in container_info.image_id:
-                continue
-            if digest and not digest.endswith('@'):
-                logger.info(f"Using ceph image with id '{image_id}' and tag '{tag}' created on {created_date}\n{digest}")
-                return digest
-    if container_info is not None:
-        logger.warning(f"Not using image '{container_info.image_id}' as it's not in list of non-dangling images with ceph=True label")
-    return None
-
-
 def get_log_dir(fsid, log_dir):
     # type: (str, str) -> str
     return os.path.join(log_dir, fsid)
diff --git a/src/cephadm/cephadmlib/container_engines.py b/src/cephadm/cephadmlib/container_engines.py
@@ -430,3 +430,111 @@ def normalize_container_id(i: str) -> str:
     if i.startswith(prefix):
         i = i[len(prefix) :]
     return i
+
+
+class ImageInfo:
+    def __init__(
+        self,
+        image_id: str,
+        repository: str,
+        digest: str,
+        tag: str,
+        created: str,
+    ) -> None:
+        self.image_id = image_id
+        self.repository = repository
+        self.digest = digest
+        self.tag = tag
+        self.created = created
+
+    @property
+    def name_with_tag(self) -> str:
+        if self.repository and self.tag:
+            return f'{self.repository}:{self.tag}'
+        return ''
+
+    @property
+    def name_with_digest(self) -> str:
+        if self.repository and self.digest:
+            return f'{self.repository}@{self.digest}'
+        return ''
+
+    @property
+    def name(self) -> str:
+        if self.name_with_digest:
+            return self.name_with_digest
+        if self.name_with_tag:
+            return self.name_with_tag
+        return self.image_id
+
+    def __repr__(self) -> str:
+        return (
+            'ImageInfo('
+            f'image_id={self.image_id!r}, '
+            f'repository={self.repository!r}, '
+            f'digest={self.digest!r}, '
+            f'tag={self.tag!r}, '
+            f'created={self.created!r}'
+            ')'
+        )
+
+
+def _container_image_list(
+    ctx: CephadmContext,
+    filters: Optional[List[str]] = None,
+    *,
+    container_path: str = '',
+) -> Tuple[str, str, int]:
+    """get images with stats from the container engine"""
+    container_path = container_path or ctx.container_engine.path
+    args = [container_path, 'images']
+    for filter_value in filters or []:
+        args.append(f'--filter={filter_value}')
+    fmt = '{{.Repository}}@{{.Digest}}|{{.ID}}|{{.Tag}}|{{.CreatedAt}}'
+    args.append(f'--format={fmt}')
+    return call(ctx, args, verbosity=CallVerbosity.QUIET)
+
+
+def _parse_container_image_list(
+    out: str,
+    err: str,
+    code: int,
+) -> List[ImageInfo]:
+    if code != 0:
+        return []
+    images: List[ImageInfo] = []
+    for line in out.splitlines():
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            digest_part, image_id, tag, created_date = line.split('|')
+        except ValueError:
+            raise ValueError('invalid container image value: {line!r}')
+        try:
+            repository, digest = (digest_part or '@').split('@')
+        except ValueError:
+            raise ValueError('invalid digest part: {digest_part!r}')
+        tag = '' if tag == '<none>' else tag
+        repository = '' if repository == '<none>' else repository
+        image_info = ImageInfo(
+            image_id=image_id,
+            repository=repository,
+            digest=digest,
+            tag=tag,
+            created=created_date,
+        )
+        images.append(image_info)
+    return images
+
+
+def parsed_container_image_list(
+    ctx: CephadmContext,
+    filters: Optional[List[str]] = None,
+    *,
+    container_path: str = '',
+) -> List[ImageInfo]:
+    out, err, code = _container_image_list(
+        ctx, filters=filters, container_path=container_path
+    )
+    return _parse_container_image_list(out, err, code)
diff --git a/src/cephadm/cephadmlib/container_lookup.py b/src/cephadm/cephadmlib/container_lookup.py
@@ -1,13 +1,20 @@
 # container_lookup.py - high-level functions for getting container info
 
-from typing import Optional
+from operator import itemgetter
+from typing import Optional, Tuple
 
 import logging
 
-from .container_engines import ContainerInfo, parsed_container_image_stats
+from .container_engines import (
+    ContainerInfo,
+    ImageInfo,
+    parsed_container_image_list,
+    parsed_container_image_stats,
+)
 from .container_types import get_container_stats
 from .context import CephadmContext
 from .daemon_identity import DaemonIdentity
+from .daemons.ceph import ceph_daemons
 from .listing import daemons_matching
 from .listing_updaters import CoreStatusUpdater
 
@@ -89,3 +96,79 @@ def get_container_info(
         'bad daemon state: no image, not running: %r', matched_deamon
     )
     return None
+
+
+def infer_local_ceph_image(
+    ctx: CephadmContext, container_path: str = ''
+) -> Optional[str]:
+    """Infer the best ceph image to use based on the following criteria:
+    Out of all images labeled as ceph that are non-dangling, prefer
+    1. the same image as the daemon container specified by -name arg (if provided).
+    2. the image used by any ceph container running on the host
+    3. the most ceph recent image on the host
+
+    :return: An image name or none
+    """
+    # enumerate ceph images on the system
+    images = parsed_container_image_list(
+        ctx,
+        filters=['dangling=false', 'label=ceph=True'],
+        container_path=container_path,
+    )
+    if not images:
+        logger.warning('No non-dangling ceph images found')
+        return None  # no images at all cached on host
+
+    # find running ceph daemons
+    _daemons = ceph_daemons()
+    daemon_name = getattr(ctx, 'name', '')
+    _cinfo_key = '_container_info'
+    _updater = CoreStatusUpdater(keep_container_info=_cinfo_key)
+    matching_daemons = [
+        itemgetter(_cinfo_key, 'name')(_updater.expand(ctx, entry))
+        for entry in daemons_matching(
+            ctx, fsid=ctx.fsid, daemon_type_predicate=lambda t: t in _daemons
+        )
+    ]
+    # collect the running ceph daemon image ids
+    images_in_use_by_daemon = set(
+        d.image_id for d, n in matching_daemons if n == daemon_name
+    )
+    images_in_use = set(d.image_id for d, _ in matching_daemons)
+
+    # prioritize images
+    def _keyfunc(image: ImageInfo) -> Tuple[bool, bool, str]:
+        return (
+            bool(
+                image.digest
+                and any(
+                    v.startswith(image.image_id)
+                    for v in images_in_use_by_daemon
+                )
+            ),
+            bool(
+                image.digest
+                and any(v.startswith(image.image_id) for v in images_in_use)
+            ),
+            image.created,
+        )
+
+    images.sort(key=_keyfunc, reverse=True)
+    best_image = images[0]
+    name_match, ceph_match, _ = _keyfunc(best_image)
+    reason = 'not in the list of non-dangling images with ceph=True label'
+    if images_in_use_by_daemon and not name_match:
+        expected = list(images_in_use_by_daemon)[0]
+        logger.warning(
+            'Not using image %r of named daemon: %s',
+            expected,
+            reason,
+        )
+    if images_in_use and not ceph_match:
+        expected = list(images_in_use)[0]
+        logger.warning(
+            'Not using image %r of ceph daemon: %s',
+            expected,
+            reason,
+        )
+    return best_image.name
diff --git a/src/cephadm/cephadmlib/listing.py b/src/cephadm/cephadmlib/listing.py
@@ -50,7 +50,17 @@
 import os
 import logging
 
-from typing import TypedDict, Union, Optional, Iterator, List, Any, Dict, cast
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    Iterator,
+    List,
+    Optional,
+    TypedDict,
+    Union,
+    cast,
+)
 
 from .context import CephadmContext
 from .daemon_identity import DaemonIdentity
@@ -191,6 +201,7 @@ def daemons_matching(
     daemon_name: Optional[str] = None,
     daemon_type: Optional[str] = None,
     fsid: Optional[str] = None,
+    daemon_type_predicate: Optional[Callable[[str], bool]] = None,
 ) -> Iterator[Union[LegacyDaemonEntry, DaemonEntry]]:
     """Iterate over the daemons configured on the current node, matching daemon
     name or daemon type if supplied.
@@ -201,6 +212,11 @@ def daemons_matching(
                 continue
             if daemon_type is not None and daemon_type != entry.daemon_type:
                 continue
+            if (
+                daemon_type_predicate is not None
+                and not daemon_type_predicate(entry.daemon_type)
+            ):
+                continue
         elif isinstance(entry, DaemonEntry):
             if fsid is not None and fsid != entry.identity.fsid:
                 continue
@@ -214,6 +230,11 @@ def daemons_matching(
                 and daemon_type != entry.identity.daemon_type
             ):
                 continue
+            if (
+                daemon_type_predicate is not None
+                and not daemon_type_predicate(entry.identity.daemon_type)
+            ):
+                continue
         else:
             raise ValueError(f'unexpected entry type: {entry}')
         yield entry
diff --git a/src/cephadm/tests/fixtures.py b/src/cephadm/tests/fixtures.py
@@ -169,7 +169,7 @@ def with_cephadm_ctx(
     with contextlib.ExitStack() as stack:
         stack.enter_context(mock.patch('cephadmlib.net_utils.attempt_bind'))
         stack.enter_context(mock.patch('cephadmlib.exe_utils.find_executable', return_value='foo'))
-        stack.enter_context(mock.patch('cephadm.get_container_info', return_value=None))
+        stack.enter_context(mock.patch('cephadmlib.container_lookup.get_container_info', return_value=None))
         stack.enter_context(mock.patch('cephadm.is_available', return_value=True))
         stack.enter_context(mock.patch('cephadm.json_loads_retry', return_value={'epoch' : 1}))
         stack.enter_context(mock.patch('cephadm.logger'))
diff --git a/src/cephadm/tests/test_cephadm.py b/src/cephadm/tests/test_cephadm.py
