IFC-873 Remove shared storage requirements for git workers (#4867)

Author: gmazoyer

backend/infrahub/git/base.py

@@ -183,7 +183,7 @@ def log(self) -> InfrahubTaskReportLogger:
         raise InitializationError("The repository has not been initialized with a TaskReport")
 
     @property
-    def directory_root(self) -> str:
+    def legacy_directory_root(self) -> str:
         """Return the path to the root directory for this repository."""
         current_dir = os.getcwd()
         repositories_directory = config.SETTINGS.git.repositories_directory
@@ -192,6 +192,16 @@ def directory_root(self) -> str:
 
         return os.path.join(repositories_directory, self.name)
 
+    @property
+    def directory_root(self) -> str:
+        """Return the path to the root directory for this repository."""
+        current_dir = os.getcwd()
+        repositories_directory = config.SETTINGS.git.repositories_directory
+        if not os.path.isabs(repositories_directory):
+            repositories_directory = os.path.join(current_dir, config.SETTINGS.git.repositories_directory)
+
+        return os.path.join(repositories_directory, str(self.id))
+
     @property
     def directory_default(self) -> str:
         """Return the path to the directory of the main branch."""
@@ -242,6 +252,36 @@ def get_git_repo_worktree(self, identifier: str) -> Repo:
             message=f"Unable to find the worktree {identifier}.",
         )
 
+    def relocate_directory_root(self) -> None:
+        """Move an old repository directory based on its name to a directory based on its ID.
+
+        This method will also take care of removing the legacy directory if:
+        1. The regular directory exists
+        2. The regular directory does not exist but will be created after renaming the legacy one
+        """
+        legacy = Path(self.legacy_directory_root)
+        current = Path(self.directory_root)
+
+        if not legacy.exists():
+            return
+
+        if not legacy.is_dir():
+            log.error("A file named after the repository should not exist", repository=self.name)
+            return
+
+        if current.is_dir():
+            log.warning(
+                f"Found legacy directory at {self.legacy_directory_root} but {self.directory_root} exists, deleting legacy directory",
+                repository=self.name,
+            )
+            shutil.rmtree(self.legacy_directory_root)
+        else:
+            log.warning(
+                f"Found legacy directory at {self.legacy_directory_root}, moving it to {self.directory_root}",
+                repository=self.name,
+            )
+            legacy.rename(self.directory_root)
+
     def validate_local_directories(self) -> bool:
         """Check if the local directories structure to ensure that the repository has been properly initialized.
 
@@ -563,6 +603,8 @@ async def fetch(self) -> bool:
 
         log.debug("Fetching the latest updates from remote origin.", repository=self.name)
 
+        self.relocate_directory_root()
+
         repo = self.get_git_repo_main()
         try:
             repo.remotes.origin.fetch()

backend/infrahub/git/repository.py

@@ -30,6 +30,14 @@ async def init(cls, service: Optional[InfrahubServices] = None, **kwargs: Any) -
         log.debug("Initiated the object on an existing directory.", repository=self.name)
         return self
 
+    @classmethod
+    async def new(cls, service: Optional[InfrahubServices] = None, **kwargs: Any) -> InfrahubRepository:
+        service = service or InfrahubServices()
+        self = cls(service=service, **kwargs)
+        await self.create_locally(infrahub_branch_name=self.infrahub_branch_name)
+        log.info("Created new repository locally.", repository=self.name)
+        return self
+
     def get_commit_value(self, branch_name: str, remote: bool = False) -> str:
         branches = {}
         if remote:
@@ -230,14 +238,6 @@ async def rebase(self, branch_name: str, source_branch: str = "main", push_remot
 
         return response
 
-    @classmethod
-    async def new(cls, service: Optional[InfrahubServices] = None, **kwargs: Any) -> InfrahubRepository:
-        service = service or InfrahubServices()
-        self = cls(service=service, **kwargs)
-        await self.create_locally(infrahub_branch_name=self.infrahub_branch_name)
-        log.info("Created the new project locally.", repository=self.name)
-        return self
-
 
 class InfrahubReadOnlyRepository(InfrahubRepositoryIntegrator):
     """
@@ -264,7 +264,7 @@ async def new(cls, service: Optional[InfrahubServices] = None, **kwargs: Any) ->
 
         self = cls(service=service, **kwargs)
         await self.create_locally(checkout_ref=self.ref, infrahub_branch_name=self.infrahub_branch_name)
-        log.info("Created the new project locally.", repository=self.name)
+        log.info("Created new repository locally.", repository=self.name)
         return self
 
     def get_commit_value(self, branch_name: str, remote: bool = False) -> str:
@@ -309,3 +309,19 @@ async def get_initialized_repo(
         )
 
     raise NotImplementedError(f"The repository kind {repository_kind} has not been implemented")
+
+
+async def initialize_repo(
+    location: str, repository_id: str, name: str, service: InfrahubServices, repository_kind: str
+) -> Union[InfrahubReadOnlyRepository, InfrahubRepository]:
+    if repository_kind == InfrahubKind.REPOSITORY:
+        return await InfrahubRepository.new(
+            location=location, id=repository_id, name=name, client=service._client, service=service
+        )
+
+    if repository_kind == InfrahubKind.READONLYREPOSITORY:
+        return await InfrahubReadOnlyRepository.new(
+            location=location, id=repository_id, name=name, client=service._client, service=service
+        )
+
+    raise NotImplementedError(f"The repository kind {repository_kind} has not been implemented")

backend/infrahub/git/tasks.py

@@ -6,9 +6,11 @@
 from infrahub.core.protocols import CoreRepository
 from infrahub.core.registry import registry
 from infrahub.exceptions import RepositoryError
+from infrahub.message_bus import Meta, messages
 from infrahub.services import services
+from infrahub.worker import WORKER_IDENTITY
 
-from ..log import get_logger
+from ..log import get_log_data, get_logger
 from ..tasks.artifact import define_artifact
 from ..workflows.catalogue import REQUEST_ARTIFACT_DEFINITION_GENERATE, REQUEST_ARTIFACT_GENERATE
 from ..workflows.utils import add_branch_tag
@@ -54,6 +56,17 @@ async def add_git_repository(model: GitRepositoryAdd) -> None:
             if model.internal_status == RepositoryInternalStatus.ACTIVE.value:
                 await repo.sync()
 
+                # Notify other workers they need to clone the repository
+                notification = messages.RefreshGitFetch(
+                    meta=Meta(initiator_id=WORKER_IDENTITY, request_id=get_log_data().get("request_id", "")),
+                    location=model.location,
+                    repository_id=model.repository_id,
+                    repository_name=model.repository_name,
+                    repository_kind=InfrahubKind.REPOSITORY,
+                    infrahub_branch_name=model.infrahub_branch_name,
+                )
+                await service.send(message=notification)
+
 
 @flow(
     name="git-repository-add-read-only",
@@ -81,6 +94,17 @@ async def add_git_repository_read_only(model: GitRepositoryAddReadOnly) -> None:
             if model.internal_status == RepositoryInternalStatus.ACTIVE.value:
                 await repo.sync_from_remote()
 
+                # Notify other workers they need to clone the repository
+                notification = messages.RefreshGitFetch(
+                    meta=Meta(initiator_id=WORKER_IDENTITY, request_id=get_log_data().get("request_id", "")),
+                    location=model.location,
+                    repository_id=model.repository_id,
+                    repository_name=model.repository_name,
+                    repository_kind=InfrahubKind.REPOSITORY,
+                    infrahub_branch_name=model.infrahub_branch_name,
+                )
+                await service.send(message=notification)
+
 
 @flow(name="git_repositories_create_branch")
 async def create_branch(branch: str, branch_id: str) -> None:
@@ -166,6 +190,16 @@ async def sync_remote_repositories() -> None:
 
                 try:
                     await repo.sync(staging_branch=staging_branch)
+                    # Tell workers to fetch to stay in sync
+                    message = messages.RefreshGitFetch(
+                        meta=Meta(initiator_id=WORKER_IDENTITY, request_id=get_log_data().get("request_id", "")),
+                        location=repository_data.repository.location.value,
+                        repository_id=repository_data.repository.id,
+                        repository_name=repository_data.repository.name.value,
+                        repository_kind=repository_data.repository.get_kind(),
+                        infrahub_branch_name=infrahub_branch,
+                    )
+                    await service.send(message=message)
                 except RepositoryError as exc:
                     error = exc
 
@@ -178,9 +212,22 @@ async def sync_remote_repositories() -> None:
 async def git_branch_create(
     client: InfrahubClient, branch: str, branch_id: str, repository_id: str, repository_name: str
 ) -> None:
+    service = services.service
+
     repo = await InfrahubRepository.init(id=repository_id, name=repository_name, client=client)
     async with lock.registry.get(name=repository_name, namespace="repository"):
         await repo.create_branch_in_git(branch_name=branch, branch_id=branch_id)
+        if repo.location:
+            # New branch has been pushed remotely, tell workers to fetch it
+            message = messages.RefreshGitFetch(
+                meta=Meta(initiator_id=WORKER_IDENTITY, request_id=get_log_data().get("request_id", "")),
+                location=repo.location,
+                repository_id=str(repo.id),
+                repository_name=repo.name,
+                repository_kind=InfrahubKind.REPOSITORY,
+                infrahub_branch_name=branch,
+            )
+            await service.send(message=message)
 
 
 @flow(name="artifact-definition-generate")
@@ -340,6 +387,17 @@ async def pull_read_only(model: GitRepositoryPullReadOnly) -> None:
             await repo.import_objects_from_files(infrahub_branch_name=model.infrahub_branch_name, commit=model.commit)
             await repo.sync_from_remote(commit=model.commit)
 
+            # Tell workers to fetch to stay in sync
+            message = messages.RefreshGitFetch(
+                meta=Meta(initiator_id=WORKER_IDENTITY, request_id=get_log_data().get("request_id", "")),
+                location=model.location,
+                repository_id=model.repository_id,
+                repository_name=model.repository_name,
+                repository_kind=InfrahubKind.READONLYREPOSITORY,
+                infrahub_branch_name=model.infrahub_branch_name,
+            )
+            await service.send(message=message)
+
 
 @flow(name="git-repository-merge")
 async def merge_git_repository(model: GitRepositoryMerge) -> None:
@@ -371,7 +429,17 @@ async def merge_git_repository(model: GitRepositoryMerge) -> None:
         repo_main.commit.value = commit
 
         await repo_main.save()
-
     else:
         async with lock.registry.get(name=model.repository_name, namespace="repository"):
             await repo.merge(source_branch=model.source_branch, dest_branch=model.destination_branch)
+            if repo.location:
+                # Destination branch has changed and pushed remotely, tell workers to re-fetch
+                message = messages.RefreshGitFetch(
+                    meta=Meta(initiator_id=WORKER_IDENTITY, request_id=get_log_data().get("request_id", "")),
+                    location=repo.location,
+                    repository_id=str(repo.id),
+                    repository_name=repo.name,
+                    repository_kind=InfrahubKind.REPOSITORY,
+                    infrahub_branch_name=model.destination_branch,
+                )
+                await service.send(message=message)

backend/infrahub/message_bus/messages/__init__.py

@@ -22,6 +22,7 @@
 from .proposed_change.request_proposedchange_rungenerators import RequestProposedChangeRunGenerators
 from .proposed_change.request_proposedchange_runtests import RequestProposedChangeRunTests
 from .proposed_change.request_proposedchange_schemaintegrity import RequestProposedChangeSchemaIntegrity
+from .refresh_git_fetch import RefreshGitFetch
 from .refresh_registry_branches import RefreshRegistryBranches
 from .refresh_registry_rebasedbranch import RefreshRegistryRebasedBranch
 from .refresh_webhook_configuration import RefreshWebhookConfiguration
@@ -55,6 +56,7 @@
     "git.repository.import_objects": GitRepositoryImportObjects,
     "schema.migration.path": SchemaMigrationPath,
     "schema.validator.path": SchemaValidatorPath,
+    "refresh.git.fetch": RefreshGitFetch,
     "refresh.registry.branches": RefreshRegistryBranches,
     "refresh.registry.rebased_branch": RefreshRegistryRebasedBranch,
     "refresh.webhook.configuration": RefreshWebhookConfiguration,

backend/infrahub/message_bus/messages/refresh_git_fetch.py

@@ -0,0 +1,13 @@
+from pydantic import Field
+
+from infrahub.message_bus import InfrahubMessage
+
+
+class RefreshGitFetch(InfrahubMessage):
+    """Fetch a repository remote changes."""
+
+    location: str = Field(..., description="The external URL of the repository")
+    repository_id: str = Field(..., description="The unique ID of the repository")
+    repository_name: str = Field(..., description="The name of the repository")
+    repository_kind: str = Field(..., description="The type of repository")
+    infrahub_branch_name: str = Field(..., description="Infrahub branch on which to sync the remote repository")

backend/infrahub/message_bus/operations/__init__.py

@@ -36,6 +36,7 @@
     "git.file.get": git.file.get,
     "git.repository.connectivity": git.repository.connectivity,
     "git.repository.import_objects": git.repository.import_objects,
+    "refresh.git.fetch": git.repository.fetch,
     "refresh.registry.branches": refresh.registry.branches,
     "refresh.registry.rebased_branch": refresh.registry.rebased_branch,
     "refresh.webhook.configuration": refresh.webhook.configuration,

backend/infrahub/message_bus/operations/git/repository.py

@@ -1,14 +1,15 @@
 from prefect import flow
 
 from infrahub.exceptions import RepositoryError
-from infrahub.git.repository import InfrahubRepository, get_initialized_repo
+from infrahub.git.repository import InfrahubRepository, get_initialized_repo, initialize_repo
 from infrahub.log import get_logger
 from infrahub.message_bus import messages
 from infrahub.message_bus.messages.git_repository_connectivity import (
     GitRepositoryConnectivityResponse,
     GitRepositoryConnectivityResponseData,
 )
 from infrahub.services import InfrahubServices
+from infrahub.worker import WORKER_IDENTITY
 
 log = get_logger()
 
@@ -44,3 +45,29 @@ async def import_objects(message: messages.GitRepositoryImportObjects, service:
         )
         repo.task_report = git_report
         await repo.import_objects_from_files(infrahub_branch_name=message.infrahub_branch_name, commit=message.commit)
+
+
+@flow(name="refresh-git-fetch", flow_run_name="Fetch git repository {message.repository_name} on " + WORKER_IDENTITY)
+async def fetch(message: messages.RefreshGitFetch, service: InfrahubServices) -> None:
+    if message.meta and message.meta.initiator_id == WORKER_IDENTITY:
+        log.info("Ignoring git fetch request originating from self", worker=WORKER_IDENTITY)
+        return
+
+    try:
+        repo = await get_initialized_repo(
+            repository_id=message.repository_id,
+            name=message.repository_name,
+            service=service,
+            repository_kind=message.repository_kind,
+        )
+    except RepositoryError:
+        repo = await initialize_repo(
+            location=message.location,
+            repository_id=message.repository_id,
+            name=message.repository_name,
+            service=service,
+            repository_kind=message.repository_kind,
+        )
+
+    await repo.fetch()
+    await repo.pull(branch_name=message.infrahub_branch_name)

backend/infrahub/services/adapters/message_bus/__init__.py

@@ -25,6 +25,7 @@ class InfrahubMessageBus:
         "trigger.*.*",
     ]
     event_bindings: list[str] = ["refresh.registry.*"]
+    broadcasted_event_bindings: list[str] = ["refresh.git.*"]
 
     async def initialize(self, service: InfrahubServices) -> None:
         """Initialize the Message bus"""

backend/infrahub/services/adapters/message_bus/nats.py

@@ -167,7 +167,8 @@ async def _initialize_api_server(self) -> None:
         self.message_enrichers.append(_add_request_id)
 
     async def _initialize_git_worker(self) -> None:
-        await self._subscribe_events(self.event_bindings, f"git-worker-{WORKER_IDENTITY}")
+        bindings = self.event_bindings + self.broadcasted_event_bindings
+        await self._subscribe_events(bindings, f"git-worker-{WORKER_IDENTITY}")
 
         consumer_config = nats.js.api.ConsumerConfig(
             ack_policy=nats.js.api.AckPolicy.EXPLICIT,
@@ -177,7 +178,7 @@ async def _initialize_git_worker(self) -> None:
             # max_ack_pending=self.settings.maximum_concurrent_messages,
             # flow_control=True,
             # idle_heartbeat=5.0,  # default value
-            filter_subjects=self.worker_bindings,
+            filter_subjects=bindings,
             durable_name="git-workers",
             deliver_group="git-workers",
             deliver_subject=self.connection.new_inbox(),
@@ -189,7 +190,7 @@ async def _initialize_git_worker(self) -> None:
             if exc.err_code != 10013:  # consumer name already in use
                 raise
 
-        for subject in self.worker_bindings:
+        for subject in bindings:
             await self.jetstream.subscribe(
                 subject=subject,
                 queue="git-workers",

backend/infrahub/services/adapters/message_bus/rabbitmq.py

@@ -173,13 +173,14 @@ async def _initialize_api_server(self) -> None:
         self.message_enrichers.append(_add_request_id)
 
     async def _initialize_git_worker(self) -> None:
+        bindings = self.event_bindings + self.broadcasted_event_bindings
         events_queue = await self.channel.declare_queue(name=f"worker-events-{WORKER_IDENTITY}", exclusive=True)
 
         self.exchange = await self.channel.declare_exchange(
             f"{self.settings.namespace}.events", type="topic", durable=True
         )
 
-        for routing_key in self.event_bindings:
+        for routing_key in bindings:
             await events_queue.bind(self.exchange, routing_key=routing_key)
         self.delayed_exchange = await self.channel.get_exchange(name=f"{self.settings.namespace}.delayed")