[ML][Pipelines] support ignore files in additional includes (Azure#28515)

* fix: support ignore file in additional includes

* fix: hello.py/hello.py => hello.py

* doc: docstring for util func

* fix: roll back unexpected snapshot id change

* feat: update ignore logic in additional includes

* fix: avoid get ignore file for file copying

Author: elliotzh

sdk/ml/azure-ai-ml/azure/ai/ml/_internal/entities/_additional_includes.py

@@ -7,11 +7,11 @@
 import tempfile
 import zipfile
 from pathlib import Path
-from typing import Optional, Union
+from typing import Union
 
 import yaml
 
-from azure.ai.ml._utils._asset_utils import IgnoreFile, get_local_paths
+from azure.ai.ml._utils._asset_utils import IgnoreFile, get_local_paths, get_ignore_file
 from azure.ai.ml.entities._util import _general_copy
 from azure.ai.ml.entities._validation import MutableValidationResult, _ValidationResultBuilder
 
@@ -29,25 +29,22 @@ def __init__(
         self,
         code_path: Union[None, str],
         yaml_path: str,
-        ignore_file: Optional[InternalComponentIgnoreFile] = None,
     ):
         self.__yaml_path = yaml_path
         self.__code_path = code_path
-        self._ignore_file = ignore_file
 
         self._tmp_code_path = None
         self.__includes = None
-        self._is_artifact_includes = False
-        self._artifact_validate_result = _ValidationResultBuilder.success()
+        # artifact validation is done on loading now, so need a private variable to store the result
+        self.__artifact_validate_result = None
 
     @property
     def _includes(self):
         if not self._additional_includes_file_path.is_file():
             return []
         if self.__includes is None:
-            self._is_artifact_includes = self._is_yaml_format_additional_includes()
             if self._is_artifact_includes:
-                self.__includes = self._load_yaml_format_additional_includes()
+                self.__includes = self._load_artifact_additional_includes()
             else:
                 with open(self._additional_includes_file_path, "r") as f:
                     lines = f.readlines()
@@ -84,24 +81,27 @@ def _additional_includes_file_path(self) -> Path:
     def code(self) -> Path:
         return self._tmp_code_path if self._tmp_code_path else self._code_path
 
-    def _copy(self, src: Path, dst: Path, ignore_file=None) -> None:
+    @staticmethod
+    def _copy(src: Path, dst: Path, *, ignore_file=None) -> None:
         if src.is_file():
+            if not dst.parent.is_dir():
+                dst.parent.mkdir(parents=True)
             _general_copy(src, dst)
         else:
             # use os.walk to replace shutil.copytree, which may raise
             # FileExistsError for same folder, the expected behavior
             # is merging ignore will be also applied during this process
+            # TODO: inner ignore file is not supported with current implementation
+            # TODO: empty folder will be ignored with current implementation
             local_paths, _ = get_local_paths(
                 source_path=str(src),
-                ignore_file=(ignore_file or self._ignore_file)
+                ignore_file=ignore_file or IgnoreFile(),
             )
-            for path in local_paths:
-                dst_root = Path(dst) / Path(path).relative_to(src)
-                dst_root_mkdir_flag = dst_root.is_dir()
-                # if there is nothing to copy under current dst_root, no need to create this folder
-                if dst_root_mkdir_flag is False:
-                    dst_root.mkdir(parents=True)
-                _general_copy(path, dst_root / Path(path).name)
+            # local_paths contains and only contains all file paths, so no need to apply ignore-file
+            for src_path in local_paths:
+                src_path = Path(src_path)
+                dst_path = Path(dst) / src_path.relative_to(src)
+                _AdditionalIncludes._copy(src_path, dst_path)
 
     @staticmethod
     def _is_folder_to_compress(path: Path) -> bool:
@@ -163,7 +163,7 @@ def _validate(self) -> MutableValidationResult:
                 validation_result.append_error(message=error_msg)
         return validation_result
 
-    def resolve(self) -> None:
+    def resolve(self, ignore_file: IgnoreFile) -> None:
         """Resolve code and potential additional includes.
         If no additional includes is specified, just return and use
         original real code path; otherwise, create a tmp folder and copy
@@ -176,34 +176,59 @@ def resolve(self) -> None:
         if Path(self._code_path).is_file():
             self._copy(Path(self._code_path), tmp_folder_path / Path(self._code_path).name)
         else:
-            for path in os.listdir(self._code_path):
-                src_path = (Path(self._code_path) / str(path)).resolve()
-                if src_path.suffix == ADDITIONAL_INCLUDES_SUFFIX:
-                    continue
-                dst_path = tmp_folder_path / str(path)
-                self._copy(src_path, dst_path)
+            self._copy(self._code_path, tmp_folder_path, ignore_file=ignore_file)
         # additional includes
         base_path = self._additional_includes_file_path.parent
+        # additional includes from artifact will be downloaded to a temp local path on calling
+        # self._includes, so no need to add specific logic for artifact
+
+        # TODO: skip ignored files defined in code when copying additional includes
+        # copy additional includes disregarding ignore files as current ignore file implementation
+        # is based on absolute path, which is not suitable for additional includes
         for additional_include in self._includes:
             src_path = Path(additional_include)
             if not src_path.is_absolute():
                 src_path = (base_path / additional_include).resolve()
+            dst_path = (tmp_folder_path / src_path.name).resolve()
+
             if self._is_folder_to_compress(src_path):
-                self._resolve_folder_to_compress(additional_include, Path(tmp_folder_path))
+                self._resolve_folder_to_compress(
+                    additional_include,
+                    Path(tmp_folder_path),
+                    # TODO: seems it won't work as current ignore file implementation is based on absolute path
+                    ignore_file=ignore_file,
+                )
+            elif src_path.is_dir():
+                # support ignore file in additional includes
+                self._copy(src_path, dst_path, ignore_file=get_ignore_file(src_path))
             else:
-                dst_path = (tmp_folder_path / src_path.name).resolve()
-                self._copy(src_path, dst_path, ignore_file=IgnoreFile() if self._is_artifact_includes else None)
+                # do not apply ignore file for files
+                self._copy(src_path, dst_path)
+
+        # Remove ignored files copied from additional includes
+        rebased_ignore_file = InternalComponentIgnoreFile(
+            directory_path=tmp_folder_path,
+            additional_include_file_name=self._additional_includes_file_path.name,
+        )
+        for base, dirs, files in os.walk(tmp_folder_path):
+            for name in files + dirs:
+                path = os.path.join(base, name)
+                if rebased_ignore_file.is_file_excluded(path):
+                    if os.path.isdir(path):
+                        shutil.rmtree(path)
+                    if os.path.isfile(path):
+                        os.remove(path)
         self._tmp_code_path = tmp_folder_path  # point code path to tmp folder
         return
 
-    def _resolve_folder_to_compress(self, include: str, dst_path: Path) -> None:
+    def _resolve_folder_to_compress(self, include: str, dst_path: Path, ignore_file: IgnoreFile) -> None:
         """resolve the zip additional include, need to compress corresponding folder."""
         zip_additional_include = (self._additional_includes_file_path.parent / include).resolve()
         folder_to_zip = zip_additional_include.parent / zip_additional_include.stem
         zip_file = dst_path / zip_additional_include.name
         with zipfile.ZipFile(zip_file, "w") as zf:
             zf.write(folder_to_zip, os.path.relpath(folder_to_zip, folder_to_zip.parent))  # write root in zip
-            local_paths, _ = get_local_paths(source_path=str(folder_to_zip), ignore_file=self._ignore_file)
+            local_paths, _ = get_local_paths(source_path=str(folder_to_zip), ignore_file=ignore_file)
             for path in local_paths:
                 zf.write(path, os.path.relpath(path, folder_to_zip.parent))
 
@@ -216,7 +241,8 @@ def cleanup(self) -> None:
             shutil.rmtree(self._tmp_code_path)
         self._tmp_code_path = None  # point code path back to real path
 
-    def _is_yaml_format_additional_includes(self):
+    @property
+    def _is_artifact_includes(self):
         try:
             with open(self._additional_includes_file_path) as f:
                 additional_includes_configs = yaml.safe_load(f)
@@ -227,7 +253,16 @@ def _is_yaml_format_additional_includes(self):
         except Exception:  # pylint: disable=broad-except
             return False
 
-    def _load_yaml_format_additional_includes(self):
+    @property
+    def _artifact_validate_result(self):
+        if not self._is_artifact_includes:
+            return _ValidationResultBuilder.success()
+        if self.__artifact_validate_result is None:
+            # artifact validation is done on loading now, so trigger it here
+            self._load_artifact_additional_includes()
+        return self.__artifact_validate_result
+
+    def _load_artifact_additional_includes(self):
         """
         Load the additional includes by yaml format.
         Addition includes is a list of include files, such as local paths and Azure Devops Artifacts.
@@ -247,7 +282,7 @@ def _load_yaml_format_additional_includes(self):
         :rtype additional_includes: List[str]
         """
         additional_includes, conflict_files = [], {}
-        self._artifact_validate_result = _ValidationResultBuilder.success()
+        self.__artifact_validate_result = _ValidationResultBuilder.success()
 
         def merge_local_path_to_additional_includes(local_path, config_info):
             additional_includes.append(local_path)

sdk/ml/azure-ai-ml/azure/ai/ml/_internal/entities/code.py

@@ -5,19 +5,41 @@
 from pathlib import Path
 from typing import List, Optional, Union
 
+from ..._utils._asset_utils import IgnoreFile
 from ...entities._assets import Code
 from ...entities._component.code import ComponentIgnoreFile
 
 
 class InternalComponentIgnoreFile(ComponentIgnoreFile):
-    _INTERNAL_COMPONENT_CODE_IGNORES = ["*.additional_includes"]
-
-    def __init__(self, directory_path: Union[str, Path]):
+    def __init__(self, directory_path: Union[str, Path], additional_include_file_name: Optional[str]):
         super(InternalComponentIgnoreFile, self).__init__(directory_path=directory_path)
+        # only the additional include file in root directory is ignored
+        # additional include files in subdirectories are not processed so keep them
+        self._additional_include_file_name = additional_include_file_name
+        self._other_ignores = []
 
     def _get_ignore_list(self) -> List[str]:
         """Override to add custom ignores for internal component."""
-        return super(InternalComponentIgnoreFile, self)._get_ignore_list() + self._INTERNAL_COMPONENT_CODE_IGNORES
+        if self._additional_include_file_name is None:
+            return super(InternalComponentIgnoreFile, self)._get_ignore_list()
+        return super(InternalComponentIgnoreFile, self)._get_ignore_list() + [self._additional_include_file_name]
+
+    def merge(self, other: IgnoreFile):
+        """Merge other ignore file with this one and create a new IgnoreFile for it.
+        """
+        ignore_file = InternalComponentIgnoreFile(self._base_path, self._additional_include_file_name)
+        ignore_file._other_ignores.append(other)  # pylint: disable=protected-access
+        return ignore_file
+
+    def is_file_excluded(self, file_path: Union[str, Path]) -> bool:
+        """Override to check if file is excluded in other ignore files."""
+        # TODO: current design of ignore file can't distinguish between files and directories of the same name
+        if super(InternalComponentIgnoreFile, self).is_file_excluded(file_path):
+            return True
+        for other in self._other_ignores:
+            if other.is_file_excluded(file_path):
+                return True
+        return False
 
 
 class InternalCode(Code):

sdk/ml/azure-ai-ml/azure/ai/ml/_internal/entities/component.py

@@ -20,10 +20,11 @@
 
 from ... import Input, Output
 from ..._utils._arm_id_utils import parse_name_label
+from ..._utils._asset_utils import IgnoreFile
 from ...entities._assets import Code
 from ...entities._job.distribution import DistributionConfiguration
 from .._schema.component import InternalComponentSchema
-from ._additional_includes import _AdditionalIncludes
+from ._additional_includes import _AdditionalIncludes, ADDITIONAL_INCLUDES_SUFFIX
 from ._input_outputs import InternalInput, InternalOutput
 from ._merkle_tree import create_merkletree
 from .code import InternalCode, InternalComponentIgnoreFile
@@ -124,6 +125,7 @@ def __init__(
         self.environment = InternalEnvironment(**environment) if isinstance(environment, dict) else environment
         self.environment_variables = environment_variables
         self.__additional_includes = None
+        self.__ignore_file = None
         # TODO: remove these to keep it a general component class
         self.command = command
         self.scope = scope
@@ -152,9 +154,6 @@ def _additional_includes(self):
             self.__additional_includes = _AdditionalIncludes(
                 code_path=self.code,
                 yaml_path=self._source_path,
-                ignore_file=InternalComponentIgnoreFile(
-                    self.code if self.code is not None else Path(self._source_path).parent,
-                ),  # use YAML's parent as code when self.code is None
             )
         return self.__additional_includes
 
@@ -208,15 +207,15 @@ def _to_rest_object(self) -> ComponentVersionData:
     def _get_snapshot_id(
         cls,
         code_path: Union[str, PathLike],
-        ignore_file: Optional[InternalComponentIgnoreFile],
+        ignore_file: IgnoreFile,
     ) -> str:
         """Get the snapshot id of a component with specific working directory in ml-components.
         Use this as the name of code asset to reuse steps in a pipeline job from ml-components runs.
 
         :param code_path: The path of the working directory.
         :type code_path: str
         :param ignore_file: The ignore file of the snapshot.
-        :type ignore_file: InternalComponentIgnoreFile
+        :type ignore_file: IgnoreFile
         :return: The snapshot id of a component in ml-components with code_path as its working directory.
         """
         curr_root = create_merkletree(code_path, ignore_file.is_file_excluded)
@@ -242,7 +241,26 @@ def _resolve_local_code(self) -> Optional[Code]:
             yield code
             return
 
-        self._additional_includes.resolve()
+        def get_additional_include_file_name():
+            if self._source_path is not None:
+                return Path(self._source_path).with_suffix(ADDITIONAL_INCLUDES_SUFFIX).name
+            return None
+
+        def get_ignore_file() -> InternalComponentIgnoreFile:
+            if self.code is None and self._source_path is None:
+                # no code and no yaml, ignore file is not needed; return ignore file on cwd to avoid error
+                return InternalComponentIgnoreFile(
+                    self.base_path,
+                    additional_include_file_name=get_additional_include_file_name(),
+                )
+
+            return InternalComponentIgnoreFile(
+                self.code if self.code is not None else Path(self._source_path).parent,
+                additional_include_file_name=get_additional_include_file_name(),
+            )
+        ignore_file = get_ignore_file()
+
+        self._additional_includes.resolve(ignore_file=ignore_file)
 
         # file dependency in code will be read during internal environment resolution
         # for example, docker file of the environment may be in additional includes
@@ -252,18 +270,26 @@ def _resolve_local_code(self) -> Optional[Code]:
             self.environment.resolve(self._additional_includes.code)
         # use absolute path in case temp folder & work dir are in different drive
         tmp_code_dir = self._additional_includes.code.absolute()
-        ignore_file = InternalComponentIgnoreFile(tmp_code_dir)
+        rebased_ignore_file = InternalComponentIgnoreFile(
+            tmp_code_dir,
+            additional_include_file_name=get_additional_include_file_name(),
+        )
         # Use the snapshot id in ml-components as code name to enable anonymous
         # component reuse from ml-component runs.
         # calculate snapshot id here instead of inside InternalCode to ensure that
         # snapshot id is calculated based on the resolved code path
         yield InternalCode(
-            name=self._get_snapshot_id(tmp_code_dir, ignore_file),
+            name=self._get_snapshot_id(
+                # use absolute path in case temp folder & work dir are in different drive
+                self._additional_includes.code.absolute(),
+                # this ignore-file should be rebased to the resolved code path
+                rebased_ignore_file,
+            ),
             version="1",
             base_path=self._base_path,
             path=tmp_code_dir,
             is_anonymous=True,
-            ignore_file=ignore_file,
+            ignore_file=rebased_ignore_file,
         )
 
         self._additional_includes.cleanup()

sdk/ml/azure-ai-ml/azure/ai/ml/entities/_component/code.py

@@ -13,10 +13,10 @@ class ComponentIgnoreFile(IgnoreFile):
     _COMPONENT_CODE_IGNORES = ["__pycache__"]
 
     def __init__(self, directory_path: Union[str, Path]):
+        self._base_path = Path(directory_path)
         # note: the parameter changes to directory path in this class, rather than file path
         file_path = get_ignore_file(directory_path).path
         super(ComponentIgnoreFile, self).__init__(file_path=file_path)
-        self._base_path = Path(directory_path)
 
     def exists(self) -> bool:
         """Override to always return True as we do have default ignores."""

sdk/ml/azure-ai-ml/tests/component/unittests/test_command_component_entity.py

@@ -10,7 +10,7 @@
 import pytest
 
 from conftest import normalized_arm_id_in_object
-from test_utilities.utils import verify_entity_load_and_dump
+from test_utilities.utils import verify_entity_load_and_dump, build_temp_folder
 
 from azure.ai.ml import Input, MpiDistribution, Output, TensorFlowDistribution, command, load_component
 from azure.ai.ml._utils.utils import load_yaml
@@ -503,12 +503,13 @@ def test_invalid_component_outputs(self) -> None:
     def test_component_code_asset_ignoring_pycache(self) -> None:
         component_yaml = "./tests/test_configs/components/helloworld_component.yml"
         component = load_component(component_yaml)
-        with tempfile.TemporaryDirectory() as temp_dir:
-            # create some files/folders expected to ignore
-            pycache = Path(temp_dir) / "__pycache__"
-            pycache.mkdir()
-            expected_exclude = pycache / "a.pyc"
-            expected_exclude.touch()
+        with build_temp_folder(
+            source_base_dir="./tests/test_configs/components",
+            relative_files_to_copy=["helloworld_component.yml"],
+            extra_files_to_create={
+                "__pycache__/a.pyc": None
+            }
+        ) as temp_dir:
             # resolve and test for ignore_file's is_file_excluded
             component.code = temp_dir
             with component._resolve_local_code() as code:
@@ -518,7 +519,7 @@ def test_component_code_asset_ignoring_pycache(self) -> None:
                         path = os.path.join(root, name)
                         if code._ignore_file.is_file_excluded(path):
                             excluded.append(path)
-                assert excluded == [str(expected_exclude.absolute())]
+                assert excluded == [str((Path(temp_dir) / "__pycache__/a.pyc").absolute())]
 
     def test_normalized_arm_id_in_component_dict(self):
         component_dict = {