Add support for pip legacy resolver for versions specified in the agent.package_manager.pip_legacy_resolver configuration option

Add skip existing packages

clearml_agent/backend_api/config/default/agent.conf

@@ -80,6 +80,14 @@
         # additional artifact repositories to use when installing python packages
         # extra_index_url: ["https://allegroai.jfrog.io/clearml/api/pypi/public/simple"]
 
+        # turn on the "--use-deprecated=legacy-resolver" flag for pip, to avoid package dependency version mismatch
+        # is any version restrictions are matched we add the "--use-deprecated=legacy-resolver" flag
+        # example: pip_legacy_resolver = [">=20.3,<24.3", ">99"]
+        # if pip==20.2 or pip==29.0 is installed we do nothing,
+        # if pip==21.1 or pip==101.1 is installed the flag is added
+        # disable the feature by passing an empty list
+        pip_legacy_resolver = [">=20.3,<24.3"]
+
         # control the pytorch wheel resolving algorithm, options are: "pip", "direct", "none"
         # Override with environment variable CLEARML_AGENT_PACKAGE_PYTORCH_RESOLVE
         # "pip" (default): would automatically detect the cuda version, and supply pip with the correct

clearml_agent/commands/worker.py

@@ -123,7 +123,8 @@ from clearml_agent.helper.package.pip_api.system import SystemPip
 from clearml_agent.helper.package.pip_api.venv import VirtualenvPip
 from clearml_agent.helper.package.poetry_api import PoetryConfig, PoetryAPI
 from clearml_agent.helper.package.post_req import PostRequirement
-from clearml_agent.helper.package.priority_req import PriorityPackageRequirement, PackageCollectorRequirement
+from clearml_agent.helper.package.priority_req import PriorityPackageRequirement, PackageCollectorRequirement, \
+    CachedPackageRequirement
 from clearml_agent.helper.package.pytorch import PytorchRequirement
 from clearml_agent.helper.package.requirements import (
     RequirementsManager, )
@@ -2433,7 +2434,7 @@ class Worker(ServiceCommandSection):
             OnlyExternalRequirements.cwd = package_api.cwd = cwd
             package_api.requirements_manager = self._get_requirements_manager(
                 base_interpreter=package_api.requirements_manager.get_interpreter(),
-                requirement_substitutions=[OnlyExternalRequirements],
+                requirement_substitutions=[CachedPackageRequirement, OnlyExternalRequirements],
             )
             # manually update the current state,
             # for the external git reference chance (in the replace callback)
@@ -2852,7 +2853,7 @@ class Worker(ServiceCommandSection):
                     OnlyExternalRequirements.cwd = package_api.cwd = cwd
                     package_api.requirements_manager = self._get_requirements_manager(
                         base_interpreter=package_api.requirements_manager.get_interpreter(),
-                        requirement_substitutions=[OnlyExternalRequirements]
+                        requirement_substitutions=[CachedPackageRequirement, OnlyExternalRequirements]
                     )
                     # manually update the current state,
                     # for the external git reference chance (in the replace callback)

clearml_agent/helper/package/base.py

@@ -29,9 +29,12 @@ class PackageManager(object):
     _config_cache_max_entries = 'agent.venvs_cache.max_entries'
     _config_cache_free_space_threshold = 'agent.venvs_cache.free_space_threshold_gb'
     _config_cache_lock_timeout = 'agent.venvs_cache.lock_timeout'
+    _config_pip_legacy_resolver = 'agent.package_manager.pip_legacy_resolver'
 
     def __init__(self):
         self._cache_manager = None
+        self._existing_packages = []
+        self._base_install_flags = []
 
     @abc.abstractproperty
     def bin(self):
@@ -79,6 +82,23 @@ class PackageManager(object):
         # type: (Iterable[Text]) -> None
         pass
 
+    def add_extra_install_flags(self, extra_flags):  # type: (List[str]) -> None
+        if extra_flags:
+            extra_flags = [
+                e for e in extra_flags if e not in list(self._base_install_flags)
+            ]
+            self._base_install_flags = list(self._base_install_flags) + list(extra_flags)
+
+    def remove_extra_install_flags(self, extra_flags):  # type: (List[str]) -> bool
+        if extra_flags:
+            _base_install_flags = [
+                e for e in self._base_install_flags if e not in list(extra_flags)
+            ]
+            if self._base_install_flags != _base_install_flags:
+                self._base_install_flags = _base_install_flags
+                return True
+        return False
+
     def upgrade_pip(self):
         result = self._install(
             *select_for_platform(
@@ -87,19 +107,58 @@ class PackageManager(object):
             ),
             "--upgrade"
         )
-        packages = self.run_with_env(('list',), output=True).splitlines()
-        # p.split is ('pip', 'x.y.z')
-        pip = [p.split() for p in packages if len(p.split()) == 2 and p.split()[0] == 'pip']
-        if pip:
-            # noinspection PyBroadException
+
+        packages = (self.freeze(freeze_full_environment=True) or dict()).get("pip")
+        if packages:
+            from clearml_agent.helper.package.requirements import RequirementsManager
+            from .requirements import MarkerRequirement, SimpleVersion
+
+            # store existing packages so that we can check if we can skip preinstalled packages
+            # we will only check "@ file" "@ vcs" for exact match
+            self._existing_packages = RequirementsManager.parse_requirements_section_to_marker_requirements(
+                packages, skip_local_file_validation=True)
+
             try:
-                from .requirements import MarkerRequirement
-                pip = pip[0][1].split('.')
-                MarkerRequirement.pip_new_version = bool(int(pip[0]) >= 20)
-            except Exception:
-                pass
+                pip_pkg = next(p for p in self._existing_packages if p.name == "pip")
+            except StopIteration:
+                pip_pkg = None
+
+            # check if we need to list the pip version as well
+            if pip_pkg:
+                MarkerRequirement.pip_new_version = SimpleVersion.compare_versions(pip_pkg.version, ">=", "20")
+
+            # add --use-deprecated=legacy-resolver to pip install to avoid mismatched packages issues
+            self._add_legacy_resolver_flag(pip_pkg.version)
+
         return result
 
+    def _add_legacy_resolver_flag(self, pip_pkg_version):
+        if not self.session.config.get(self._config_pip_legacy_resolver, None):
+            return
+
+        from .requirements import SimpleVersion
+
+        match_versions = self.session.config.get(self._config_pip_legacy_resolver)
+        matched = False
+        for rule in match_versions:
+            matched = False
+            # make sure we match all the parts of the rule
+            for a_version in rule.split(","):
+                o, v = SimpleVersion.split_op_version(a_version.strip())
+                matched = SimpleVersion.compare_versions(pip_pkg_version, o, v)
+                if not matched:
+                    break
+            # if the rule is fully matched we have a match
+            if matched:
+                break
+
+        legacy_resolver_flags = ["--use-deprecated=legacy-resolver"]
+        if matched:
+            print("INFO: Using legacy resolver for PIP to avoid inconsistency with package versions!")
+            self.add_extra_install_flags(legacy_resolver_flags)
+        elif self.remove_extra_install_flags(legacy_resolver_flags):
+            print("INFO: removing pip legacy resolver!")
+
     def get_python_command(self, extra=()):
         # type: (...) -> Executable
         return Argv(self.bin, *extra)
@@ -149,6 +208,18 @@ class PackageManager(object):
                     return False
             except Exception:
                 return False
+
+            try:
+                from .requirements import Requirement, MarkerRequirement
+                req = MarkerRequirement(Requirement.parse(package_name))
+
+                # if pip was part of the requirements, make sure we update the flags
+                # add --use-deprecated=legacy-resolver to pip install to avoid mismatched packages issues
+                if req.name == "pip" and req.version:
+                    PackageManager._selected_manager._add_legacy_resolver_flag(req.version)
+            except Exception as e:
+                print("WARNING: Error while parsing pip version legacy [{}]".format(e))
+
         return True
 
     @classmethod

clearml_agent/helper/package/pip_api/system.py

@@ -97,7 +97,7 @@ class SystemPip(PackageManager):
         return Argv(self.bin, '-m', 'pip', '--disable-pip-version-check', *command)
 
     def install_flags(self):
-        indices_args = tuple(
+        base_args = tuple(self._base_install_flags or []) + tuple(
             chain.from_iterable(('--extra-index-url', x) for x in PIP_EXTRA_INDICES)
         )
 
@@ -105,7 +105,7 @@ class SystemPip(PackageManager):
             ENV_PIP_EXTRA_INSTALL_FLAGS.get() or \
             self.session.config.get("agent.package_manager.extra_pip_install_flags", None)
 
-        return (indices_args + tuple(extra_pip_flags)) if extra_pip_flags else indices_args
+        return (base_args + tuple(extra_pip_flags)) if extra_pip_flags else base_args
 
     def download_flags(self):
         indices_args = tuple(

clearml_agent/helper/package/pip_api/venv.py

@@ -37,7 +37,9 @@ class VirtualenvPip(SystemPip, PackageManager):
 
     def load_requirements(self, requirements):
         if isinstance(requirements, dict) and requirements.get("pip"):
-            requirements["pip"] = self.requirements_manager.replace(requirements["pip"])
+            requirements["pip"] = self.requirements_manager.replace(
+                requirements["pip"], existing_packages=self._existing_packages
+            )
         super(VirtualenvPip, self).load_requirements(requirements)
         self.requirements_manager.post_install(self.session, package_manager=self)
 

clearml_agent/helper/package/priority_req.py

@@ -53,12 +53,18 @@ class PriorityPackageRequirement(SimpleSubstitution):
         if not self._replaced_packages:
             return list_of_requirements
 
+        # we assume that both pip & setup tools are not in list_of_requirements, and we need to add them
+
         if "pip" in self._replaced_packages:
             full_freeze = PackageManager.out_of_scope_freeze(freeze_full_environment=True)
-            # now let's look for pip
-            pips = [line for line in full_freeze.get("pip", []) if line.split("==")[0] == "pip"]
-            if pips and "pip" in list_of_requirements:
-                list_of_requirements["pip"] = [pips[0]] + list_of_requirements["pip"]
+            if not full_freeze:
+                if "pip" in list_of_requirements:
+                    list_of_requirements["pip"] = [self._replaced_packages["pip"]] + list_of_requirements["pip"]
+            else:
+                # now let's look for pip
+                pips = [line for line in full_freeze.get("pip", []) if str(line.split("==")[0]).strip() == "pip"]
+                if pips and "pip" in list_of_requirements:
+                    list_of_requirements["pip"] = [pips[0]] + list_of_requirements["pip"]
 
         if "setuptools" in self._replaced_packages:
             try:
@@ -87,6 +93,20 @@ class PriorityPackageRequirement(SimpleSubstitution):
         return list_of_requirements
 
 
+class CachedPackageRequirement(PriorityPackageRequirement):
+
+    name = ("setuptools", "pip", )
+    optional_package_names = tuple()
+
+    def replace(self, req):
+        """
+        Put the requirement in the list for later conversion
+        :raises: ValueError if version is pre-release
+        """
+        self._replaced_packages[req.name] = req.line
+        return Text(req)
+
+
 class PackageCollectorRequirement(SimpleSubstitution):
     """
     This RequirementSubstitution class will allow you to have multiple instances of the same

clearml_agent/helper/package/requirements.py

@@ -19,7 +19,7 @@ import logging
 from clearml_agent.definitions import PIP_EXTRA_INDICES
 from clearml_agent.helper.base import (
     warning, is_conda, which, join_lines, is_windows_platform,
-    convert_cuda_version_to_int_10_base_str, )
+    convert_cuda_version_to_int_10_base_str, dump_yaml, )
 from clearml_agent.helper.process import Argv, PathLike
 from clearml_agent.helper.gpu.gpustat import get_driver_cuda_version
 from clearml_agent.session import Session, normalize_cuda_version
@@ -94,6 +94,12 @@ class MarkerRequirement(object):
     def __repr__(self):
         return '{self.__class__.__name__}[{self}]'.format(self=self)
 
+    def __eq__(self, other):
+        return isinstance(other, MarkerRequirement) and str(self) == str(other)
+
+    def __hash__(self):
+        return str(self).__hash__()
+
     def format_specs(self, num_parts=None, max_num_parts=None):
         max_num_parts = max_num_parts or num_parts
         if max_num_parts is None or not self.specs:
@@ -116,6 +122,10 @@ class MarkerRequirement(object):
     def specs(self):  # type: () -> List[Tuple[Text, Text]]
         return self.req.specs
 
+    @property
+    def version(self):  # type: () -> Text
+        return self.specs[0][1] if self.specs else ""
+
     @specs.setter
     def specs(self, value):  # type: (List[Tuple[Text, Text]]) -> None
         self.req.specs = value
@@ -143,6 +153,8 @@ class MarkerRequirement(object):
         If the requested version is 1.2 the self.spec should be 1.2*
         etc.
 
+        usage: it returns the value of the following comparison: requested_version "op" self.version
+
         :param str requested_version:
         :param str op: '==', '>', '>=', '<=', '<', '~='
         :param int num_parts: number of parts to compare
@@ -152,7 +164,7 @@ class MarkerRequirement(object):
         if not self.specs:
             return True
 
-        version = self.specs[0][1]
+        version = self.version
         op = (op or self.specs[0][0]).strip()
 
         return SimpleVersion.compare_versions(
@@ -170,11 +182,21 @@ class MarkerRequirement(object):
         self.req.local_file = False
         return True
 
-    def validate_local_file_ref(self):
+    def is_local_package_ref(self):
         # if local file does not exist, remove the reference to it
         if self.vcs or self.editable or self.path or not self.local_file or not self.name or \
                 not self.uri or not self.uri.startswith("file://"):
+            return False
+        return True
+
+    def is_vcs_ref(self):
+        return bool(self.vcs)
+
+    def validate_local_file_ref(self):
+        # if local file does not exist, remove the reference to it
+        if not self.is_local_package_ref():
             return
+
         local_path = Path(self.uri[len("file://"):])
         if not local_path.exists():
             local_path = Path(unquote(self.uri)[len("file://"):])
@@ -221,6 +243,19 @@ class SimpleVersion:
     _local_version_separators = re.compile(r"[\._-]")
     _regex = re.compile(r"^\s*" + VERSION_PATTERN + r"\s*$", re.VERBOSE | re.IGNORECASE)
 
+    @classmethod
+    def split_op_version(cls, line):
+        """
+        Split a string in the form of ">=1.2.3" into a (op, version), i.e. (">=", "1.2.3")
+        Notice is calling with only a version string (e.g. "1.2.3") default operator is "=="
+        which means you get ("==", "1.2.3")
+        :param line: string examples: "<=0.1.2"
+        :return: tuple of (op, version) example ("<=", "0.1.2")
+        """
+        match = r"\s*([>=<~!]*)\s*(\S*)\s*"
+        groups = re.match(match, line).groups()
+        return groups[0] or "==", groups[1]
+
     @classmethod
     def compare_versions(cls, version_a, op, version_b, ignore_sub_versions=True, num_parts=3):
         """
@@ -624,14 +659,54 @@ class RequirementsManager(object):
                 return handler.replace(req)
         return None
 
-    def replace(self, requirements):  # type: (Text) -> Text
+    def replace(
+            self,
+            requirements,  # type: Text
+            existing_packages=None,  # type: List[MarkerRequirement]
+            pkg_skip_existing_local=True,  # type: bool
+            pkg_skip_existing_vcs=True,  # type: bool
+            pkg_skip_existing=True,  # type: bool
+    ):  # type: (...) -> Text
         parsed_requirements = self.parse_requirements_section_to_marker_requirements(
-            requirements=requirements, cwd=self._cwd)
+            requirements=requirements, cwd=self._cwd, skip_local_file_validation=True)
 
+        if parsed_requirements and existing_packages:
+            skipped_packages = None
+            if pkg_skip_existing:
+                skipped_packages = set(parsed_requirements) & set(existing_packages)
+            elif pkg_skip_existing_local or pkg_skip_existing_vcs:
+                existing_packages = [
+                    p for p in existing_packages if (
+                        (pkg_skip_existing_local and p.is_local_package_ref()) or
+                        (pkg_skip_existing_vcs and p.is_vcs_ref())
+                    )
+                ]
+                skipped_packages = set(parsed_requirements) & set(existing_packages)
+
+            if skipped_packages:
+                # maintain order
+                num_skipped_packages = len(parsed_requirements)
+                parsed_requirements = [p for p in parsed_requirements if p not in skipped_packages]
+                num_skipped_packages -= len(parsed_requirements)
+                print("Skipping {} pre-installed packages:\n{}Remaining {} additional packages to install".format(
+                    num_skipped_packages,
+                    dump_yaml(sorted([str(p) for p in skipped_packages])),
+                    len(parsed_requirements)
+                ))
+
+                # nothing to install!
+                if not parsed_requirements:
+                    return ""
+
+        # sanity check
         if not parsed_requirements:
             # return the original requirements just in case
             return requirements
 
+        # remove local file reference that do not exist
+        for p in parsed_requirements:
+            p.validate_local_file_ref()
+
         def replace_one(i, req):
             # type: (int, MarkerRequirement) -> Optional[Text]
             try:
@@ -805,7 +880,7 @@ class RequirementsManager(object):
                 normalize_cuda_version(cudnn_version or 0))
 
     @staticmethod
-    def parse_requirements_section_to_marker_requirements(requirements, cwd=None):
+    def parse_requirements_section_to_marker_requirements(requirements, cwd=None, skip_local_file_validation=False):
         def safe_parse(req_str):
             # noinspection PyBroadException
             try:
@@ -815,7 +890,8 @@ class RequirementsManager(object):
 
         def create_req(x):
             r = MarkerRequirement(x)
-            r.validate_local_file_ref()
+            if not skip_local_file_validation:
+                r.validate_local_file_ref()
             return r
 
         if not requirements: