connectors-ci: improve modified connectors detection

Author: alafanechere

airbyte-ci/connectors/connector_ops/connector_ops/utils.py

@@ -319,7 +319,7 @@ def __repr__(self) -> str:
 
     @functools.lru_cache(maxsize=2)
     def get_local_dependency_paths(self, with_test_dependencies: bool = True) -> Set[Path]:
-        dependencies_paths = [self.code_directory]
+        dependencies_paths = []
         if self.language == ConnectorLanguage.JAVA:
             dependencies_paths += get_all_gradle_dependencies(
                 self.code_directory / "build.gradle", with_test_dependencies=with_test_dependencies
@@ -352,8 +352,11 @@ def get_changed_connectors(
 
 
 def get_all_released_connectors() -> Set:
+    repo = git.Repo(search_parent_directories=True)
+    repo_path = repo.working_tree_dir
+
     return {
         Connector(Path(metadata_file).parent.name)
-        for metadata_file in glob("airbyte-integrations/connectors/**/metadata.yaml", recursive=True)
+        for metadata_file in glob(f"{repo_path}/airbyte-integrations/connectors/**/metadata.yaml", recursive=True)
         if SCAFFOLD_CONNECTOR_GLOB not in metadata_file
     }

airbyte-ci/connectors/connector_ops/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "connector_ops"
-version = "0.2.1"
+version = "0.2.2"
 description = "Packaged maintained by the connector operations team to perform CI for connectors"
 authors = ["Airbyte <[email protected]>"]
 

airbyte-ci/connectors/pipelines/pipelines/commands/airbyte_ci.py

@@ -7,6 +7,7 @@
 from typing import List
 
 import click
+from github import PullRequest
 from pipelines import github, main_logger
 from pipelines.bases import CIContext
 from pipelines.utils import (
@@ -17,7 +18,6 @@
     get_modified_files_in_commit,
     get_modified_files_in_pull_request,
 )
-from github import PullRequest
 
 from .groups.connectors import connectors
 from .groups.metadata import metadata
@@ -133,6 +133,7 @@ def airbyte_ci(
         main_logger.info(f"Pipeline Start Timestamp: {pipeline_start_timestamp}")
         main_logger.info(f"Modified Files: {ctx.obj['modified_files']}")
 
+
 airbyte_ci.add_command(connectors)
 airbyte_ci.add_command(metadata)
 

airbyte-ci/connectors/pipelines/pipelines/commands/groups/connectors.py

@@ -7,10 +7,11 @@
 import os
 import sys
 from pathlib import Path
-from typing import Any, Dict, Tuple
+from typing import Set, Tuple, Union
 
 import anyio
 import click
+from connector_ops.utils import Connector, ConnectorLanguage, console, get_all_released_connectors
 from pipelines import main_logger
 from pipelines.builds import run_connector_build_pipeline
 from pipelines.contexts import ConnectorContext, ContextState, PublishConnectorContext
@@ -19,8 +20,7 @@
 from pipelines.pipelines.connectors import run_connectors_pipelines
 from pipelines.publish import reorder_contexts, run_connector_publish_pipeline
 from pipelines.tests import run_connector_test_pipeline
-from pipelines.utils import DaggerPipelineCommand, get_modified_connectors
-from connector_ops.utils import ConnectorLanguage, console, get_all_released_connectors
+from pipelines.utils import DaggerPipelineCommand, get_connector_modified_files, get_modified_connectors_and_files
 from rich.table import Table
 from rich.text import Text
 
@@ -87,45 +87,26 @@ def connectors(
 
     ctx.ensure_object(dict)
     ctx.obj["use_remote_secrets"] = use_remote_secrets
-    ctx.obj["connector_names"] = names
-    ctx.obj["connector_languages"] = languages
-    ctx.obj["release_states"] = release_stages
-    ctx.obj["modified"] = modified
+    ctx.obj["selected_connectors_names"] = set(names) if names else set()
+    ctx.obj["selected_connectors_languages"] = set(languages) if languages else set()
+    ctx.obj["selected_connectors_release_stages"] = set(release_stages) if release_stages else set()
+    ctx.obj["select_modified_connectors"] = modified
     ctx.obj["concurrency"] = concurrency
     ctx.obj["execute_timeout"] = execute_timeout
-
-    all_connectors = get_all_released_connectors()
-
-    # We get the modified connectors and downstream connector deps, and files
-    modified_connectors_and_files = get_modified_connectors(ctx.obj["modified_files"])
-
-    # We select all connectors by default
-    # and attach modified files to them
-    selected_connectors_and_files = {connector: modified_connectors_and_files.get(connector, []) for connector in all_connectors}
-
-    if modified:
-        selected_connectors_and_files = modified_connectors_and_files
-    if names:
-        selected_connectors_and_files = {
-            connector: selected_connectors_and_files[connector]
-            for connector in selected_connectors_and_files
-            if connector.technical_name in names
-        }
-    if languages:
-        selected_connectors_and_files = {
-            connector: selected_connectors_and_files[connector]
-            for connector in selected_connectors_and_files
-            if connector.language in languages
-        }
-    if release_stages:
-        selected_connectors_and_files = {
-            connector: selected_connectors_and_files[connector]
-            for connector in selected_connectors_and_files
-            if connector.release_stage in release_stages
-        }
-
-    ctx.obj["selected_connectors_and_files"] = selected_connectors_and_files
-    ctx.obj["selected_connectors_names"] = [c.technical_name for c in selected_connectors_and_files.keys()]
+    ctx.obj["all_connectors"] = get_all_released_connectors()
+    ctx.obj["selected_connectors_by_names"] = {Connector(technical_name=name) for name in names}
+    ctx.obj["selected_connectors_by_languages"] = {connector for connector in ctx.obj["all_connectors"] if connector.language in languages}
+    ctx.obj["selected_connectors_by_release_stages"] = {
+        connector for connector in ctx.obj["all_connectors"] if connector.release_stage in release_stages
+    }
+    ctx.obj["selected_connectors"] = (
+        ctx.obj["selected_connectors_by_names"]
+        | ctx.obj["selected_connectors_by_languages"]
+        | ctx.obj["selected_connectors_by_release_stages"]
+    )
+    ctx.obj["selected_connectors_and_files"] = {
+        connector: get_connector_modified_files(connector, ctx.obj["modified_files"]) for connector in ctx.obj["selected_connectors"]
+    }
 
 
 @connectors.command(cls=DaggerPipelineCommand, help="Test all the selected connectors.")
@@ -142,7 +123,13 @@ def test(
         main_logger.info("Skipping connectors tests for draft pull request.")
         sys.exit(0)
 
-    main_logger.info(f"Will run the test pipeline for the following connectors: {', '.join(ctx.obj['selected_connectors_names'])}")
+    if ctx.obj["select_modified_connectors"]:
+        ctx.obj["selected_connectors_and_files"] = {
+            **ctx.obj["selected_connectors_and_files"],
+            **get_modified_connectors_and_files(ctx.obj["modified_files"]),
+        }
+    main_logger.info(ctx.obj)
+    log_selected_connectors(ctx.obj["selected_connectors_and_files"])
     if ctx.obj["selected_connectors_and_files"]:
         update_global_commit_status_check_for_tests(ctx.obj, "pending")
     else:
@@ -198,7 +185,14 @@ def send_commit_status_check() -> None:
 @connectors.command(cls=DaggerPipelineCommand, help="Build all images for the selected connectors.")
 @click.pass_context
 def build(ctx: click.Context) -> bool:
-    main_logger.info(f"Will build the following connectors: {', '.join(ctx.obj['selected_connectors_names'])}.")
+    """Runs a build pipeline for the selected connectors."""
+    if ctx.obj["select_modified_connectors"]:
+        ctx.obj["selected_connectors_and_files"] = {
+            **ctx.obj["selected_connectors_and_files"],
+            **get_modified_connectors_and_files(ctx.obj["modified_files"]),
+        }
+
+    log_selected_connectors(ctx.obj["selected_connectors_and_files"])
     connectors_contexts = [
         ConnectorContext(
             pipeline_name=f"Build connector {connector.technical_name}",
@@ -305,17 +299,19 @@ def publish(
     ctx.obj["spec_cache_bucket_name"] = spec_cache_bucket_name
     ctx.obj["metadata_service_bucket_name"] = metadata_service_bucket_name
     ctx.obj["metadata_service_gcs_credentials"] = metadata_service_gcs_credentials
-
     if ctx.obj["is_local"]:
         click.confirm(
-            "Publishing from a local environment is not recommend and requires to be logged in Airbyte's DockerHub registry, do you want to continue?",
+            "Publishing from a local environment is not recommended and requires to be logged in Airbyte's DockerHub registry, do you want to continue?",
             abort=True,
         )
 
-    selected_connectors_and_files = ctx.obj["selected_connectors_and_files"]
-    selected_connectors_names = ctx.obj["selected_connectors_names"]
+    if ctx.obj["select_modified_connectors"]:
+        ctx.obj["selected_connectors_and_files"] = {
+            **ctx.obj["selected_connectors_and_files"],
+            **get_modified_connectors_and_files(ctx.obj["modified_files"], metadata_modification_only=True, dependency_scanning=False),
+        }
 
-    main_logger.info(f"Will publish the following connectors: {', '.join(selected_connectors_names)}")
+    log_selected_connectors(ctx.obj["selected_connectors_and_files"])
     publish_connector_contexts = reorder_contexts(
         [
             PublishConnectorContext(
@@ -342,7 +338,7 @@ def publish(
                 ci_gcs_credentials=ctx.obj["ci_gcs_credentials"],
                 pull_request=ctx.obj.get("pull_request"),
             )
-            for connector, modified_files in selected_connectors_and_files.items()
+            for connector, modified_files in ctx.obj["selected_connectors_and_files"].items()
         ]
     )
 
@@ -394,23 +390,18 @@ def list(
     return True
 
 
-@connectors.command(cls=DaggerPipelineCommand, help="Autoformat connector code.")
+@connectors.command(name="format", cls=DaggerPipelineCommand, help="Autoformat connector code.")
 @click.pass_context
-def format(ctx: click.Context) -> bool:
-    if ctx.obj["modified"]:
-        # We only want to format the connector that with modified files on the current branch.
-        connectors_and_files_to_format = [
-            (connector, modified_files) for connector, modified_files in ctx.obj["selected_connectors_and_files"].items() if modified_files
-        ]
-    else:
-        # We explicitly want to format specific connectors
-        connectors_and_files_to_format = [
-            (connector, modified_files) for connector, modified_files in ctx.obj["selected_connectors_and_files"].items()
-        ]
+def format_code(ctx: click.Context) -> bool:
+    if ctx.obj["select_modified_connectors"]:
+        ctx.obj["selected_connectors_and_files"] = {
+            **ctx.obj["selected_connectors_and_files"],
+            **get_modified_connectors_and_files(ctx.obj["modified_files"]),
+        }
 
-    if connectors_and_files_to_format:
+    if ctx.obj["selected_connectors_and_files"]:
         main_logger.info(
-            f"Will format the following connectors: {', '.join([connector.technical_name for connector, _ in connectors_and_files_to_format])}."
+            f"Will format the following connectors: {', '.join([connector.technical_name for connector, _ in ctx.obj['selected_connectors_and_files'].keys()])}."
         )
     else:
         main_logger.info("No connectors to format.")
@@ -435,7 +426,7 @@ def format(ctx: click.Context) -> bool:
             pull_request=ctx.obj.get("pull_request"),
             should_save_report=False,
         )
-        for connector, modified_files in connectors_and_files_to_format
+        for connector, modified_files in ctx.obj["selected_connectors_and_files"].items()
     ]
 
     anyio.run(
@@ -449,3 +440,11 @@ def format(ctx: click.Context) -> bool:
     )
 
     return True
+
+
+def log_selected_connectors(selected_connectors_and_files: dict[Connector, Set[Union[str, Path]]]) -> None:
+    if selected_connectors_and_files:
+        selected_connectors_names = [c.technical_name for c in selected_connectors_and_files.keys()]
+        main_logger.info(f"Will run on the following connectors: {', '.join(selected_connectors_names)}.")
+    else:
+        main_logger.info("No connectors to run.")

airbyte-ci/connectors/pipelines/pipelines/utils.py

@@ -12,34 +12,35 @@
 import re
 import sys
 import unicodedata
-
 from glob import glob
+from io import TextIOWrapper
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Callable, List, Optional, Set, Tuple, Union
-from io import TextIOWrapper
 
 import anyio
 import asyncer
 import click
 import git
-from pipelines import consts, main_logger, sentry_utils
-from pipelines.consts import GCS_PUBLIC_DOMAIN
 from connector_ops.utils import get_all_released_connectors, get_changed_connectors
 from dagger import Client, Config, Connection, Container, DaggerError, ExecError, File, ImageLayerCompression, QueryError, Secret
 from google.cloud import storage
 from google.oauth2 import service_account
 from more_itertools import chunked
+from pipelines import consts, main_logger, sentry_utils
+from pipelines.consts import GCS_PUBLIC_DOMAIN
 
 if TYPE_CHECKING:
-    from pipelines.contexts import ConnectorContext
+    from connector_ops.utils import Connector
     from github import PullRequest
+    from pipelines.contexts import ConnectorContext
 
 DAGGER_CONFIG = Config(log_output=sys.stderr)
 AIRBYTE_REPO_URL = "https://github.com/airbytehq/airbyte.git"
 METADATA_FILE_NAME = "metadata.yaml"
 METADATA_ICON_FILE_NAME = "icon.svg"
 DIFF_FILTER = "MADRT"  # Modified, Added, Deleted, Renamed, Type changed
 IGNORED_FILE_EXTENSIONS = [".md"]
+ALL_CONNECTOR_DEPENDENCIES = [(connector, connector.get_local_dependency_paths()) for connector in get_all_released_connectors()]
 
 
 # This utils will probably be redundant once https://github.com/dagger/dagger/issues/3764 is implemented
@@ -323,47 +324,54 @@ def _file_path_starts_with(given_file_path: Path, starts_with_path: Path) -> boo
     return given_file_path_parts[: len(starts_with_path_parts)] == starts_with_path_parts
 
 
-def _find_modified_connectors(file: Union[str, Path], all_dependencies: list) -> dict:
-    """Find all connectors whose dependencies were modified."""
+def _find_modified_connectors(file_path: Union[str, Path], dependency_scanning: bool = True) -> dict:
+    """Find all connectors impacted by the file change."""
     modified_connectors = {}
-    for connector, connector_dependencies in all_dependencies:
-        for connector_dependency in connector_dependencies:
-            file_path = Path(file)
-
-            if _file_path_starts_with(file_path, connector_dependency):
-                # Add the connector to the modified connectors
-                modified_connectors.setdefault(connector, [])
-                connector_directory_path = Path(connector.code_directory)
-
-                # If the file is in the connector directory, add it to the modified files
-                if _file_path_starts_with(file_path, connector_directory_path):
-                    modified_connectors[connector].append(file)
-                else:
-                    main_logger.info(f"Adding connector '{connector}' due to dependency modification: '{file}'.")
+    for connector, connector_dependencies in ALL_CONNECTOR_DEPENDENCIES:
+        if Path(file_path).is_relative_to(Path(connector.code_directory)):
+            main_logger.info(f"Adding connector '{connector}' due to connector file modification: {file_path}.")
+            modified_connectors.setdefault(connector, [])
+            modified_connectors[connector].append(file_path)
+
+        if dependency_scanning:
+            for connector_dependency in connector_dependencies:
+                if Path(file_path).is_relative_to(Path(connector_dependency)):
+                    # Add the connector to the modified connectors
+                    modified_connectors.setdefault(connector, [])
+                    main_logger.info(f"Adding connector '{connector}' due to dependency modification: '{file_path}'.")
 
     return modified_connectors
 
 
-def get_modified_connectors(modified_files: Set[Union[str, Path]]) -> dict:
+def get_modified_connectors_and_files(
+    modified_files: Set[Union[str, Path]], metadata_modification_only: bool = False, dependency_scanning: bool = True
+) -> dict[Connector, Set[Union[str, Path]]]:
     """Create a mapping of modified connectors (key) and modified files (value).
     As we call connector.get_local_dependencies_paths() any modification to a dependency will trigger connector pipeline for all connectors that depend on it.
     The get_local_dependencies_paths function currently computes dependencies for Java connectors only.
     It's especially useful to trigger tests of strict-encrypt variant when a change is made to the base connector.
     Or to tests all jdbc connectors when a change is made to source-jdbc or base-java.
     We'll consider extending the dependency resolution to Python connectors once we confirm that it's needed and feasible in term of scale.
     """
-    all_connector_dependencies = [(connector, connector.get_local_dependency_paths()) for connector in get_all_released_connectors()]
 
     # Ignore files with certain extensions
     modified_files = [file for file in modified_files if not _is_ignored_file(file)]
-
+    if metadata_modification_only:
+        modified_files = get_modified_metadata_files(modified_files)
+        main_logger.info(f"Modified metadata files: {modified_files}")
     modified_connectors = {}
     for modified_file in modified_files:
-        modified_connectors.update(_find_modified_connectors(modified_file, all_connector_dependencies))
-
+        modified_connectors.update(_find_modified_connectors(modified_file, dependency_scanning))
     return modified_connectors
 
 
+def get_connector_modified_files(connector: Connector, all_modified_files: Set[Union[str, Path]]) -> Set[Union[str, Path]]:
+    for modified_file in all_modified_files:
+        modified_file_path = Path(modified_file)
+        if modified_file_path.is_relative_to(connector.code_directory):
+            yield modified_file
+
+
 def get_modified_metadata_files(modified_files: Set[Union[str, Path]]) -> Set[Path]:
     return {
         Path(str(f))

airbyte-ci/connectors/pipelines/poetry.lock

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "pipelines"
-version = "0.1.0"
+version = "0.1.1"
 description = "Packaged maintained by the connector operations team to perform CI for connectors' pipelines"
 authors = ["Airbyte <[email protected]>"]