You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "dheerajturaga (via GitHub)" <gi...@apache.org> on 2025/09/29 11:59:54 UTC
[PR] Fix corrupted bare Git repository recovery in DAG bundles [airflow]
dheerajturaga opened a new pull request, #56206:
URL: https://github.com/apache/airflow/pull/56206
When using git DAG bundles, corrupted bare repositories can cause all tasks
landing on a host to fail with InvalidGitRepositoryError. This adds retry
logic that detects corrupted bare repositories, cleans them up, and attempts
to re-clone them once before failing.
Changes:
- Add InvalidGitRepositoryError handling in _clone_bare_repo_if_required()
- Implement cleanup and retry logic with shutil.rmtree()
- Add comprehensive tests for both successful retry and retry failure scenarios
- Ensure all existing tests continue to pass
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix corrupted bare Git repository recovery in DAG bundles [airflow]
Posted by "dheerajturaga (via GitHub)" <gi...@apache.org>.
dheerajturaga commented on PR #56206:
URL: https://github.com/apache/airflow/pull/56206#issuecomment-3403852499
@potiuk this looks like an unrelated failure?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix corrupted bare Git repository recovery in DAG bundles [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk merged PR #56206:
URL: https://github.com/apache/airflow/pull/56206
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix corrupted bare Git repository recovery in DAG bundles [airflow]
Posted by "jscheffl (via GitHub)" <gi...@apache.org>.
jscheffl commented on code in PR #56206:
URL: https://github.com/apache/airflow/pull/56206#discussion_r2433918400
##########
providers/git/src/airflow/providers/git/bundles/git.py:
##########
@@ -109,48 +120,77 @@ def _initialize(self):
def initialize(self) -> None:
if not self.repo_url:
- raise AirflowException(f"Connection {self.git_conn_id} doesn't have a host url")
+ raise RuntimeError(f"Connection {self.git_conn_id} doesn't have a host url")
self._initialize()
super().initialize()
+ @retry(
+ retry=retry_if_exception_type((InvalidGitRepositoryError, GitCommandError)),
+ stop=stop_after_attempt(2),
+ reraise=True,
+ )
def _clone_repo_if_required(self) -> None:
- if not os.path.exists(self.repo_path):
- self._log.info("Cloning repository", repo_path=self.repo_path, bare_repo_path=self.bare_repo_path)
- try:
+ try:
+ if not os.path.exists(self.repo_path):
+ self._log.info(
+ "Cloning repository", repo_path=self.repo_path, bare_repo_path=self.bare_repo_path
+ )
Repo.clone_from(
url=self.bare_repo_path,
to_path=self.repo_path,
)
- except NoSuchPathError as e:
- # Protection should the bare repo be removed manually
- raise AirflowException("Repository path: %s not found", self.bare_repo_path) from e
- else:
- self._log.debug("repo exists", repo_path=self.repo_path)
- self.repo = Repo(self.repo_path)
-
+ else:
+ self._log.debug("repo exists", repo_path=self.repo_path)
+ self.repo = Repo(self.repo_path)
+ except NoSuchPathError as e:
+ # Protection should the bare repo be removed manually
+ raise RuntimeError("Repository path: %s not found", self.bare_repo_path) from e
+ except (InvalidGitRepositoryError, GitCommandError) as e:
+ self._log.warning(
+ "Repository clone/open failed, cleaning up and retrying",
+ repo_path=self.repo_path,
+ exc=e,
+ )
+ if os.path.exists(self.repo_path):
+ shutil.rmtree(self.repo_path)
+ raise
+
+ @retry(
+ retry=retry_if_exception_type((InvalidGitRepositoryError, GitCommandError)),
+ stop=stop_after_attempt(2),
+ reraise=True,
+ )
def _clone_bare_repo_if_required(self) -> None:
if not self.repo_url:
- raise AirflowException(f"Connection {self.git_conn_id} doesn't have a host url")
- if not os.path.exists(self.bare_repo_path):
- self._log.info("Cloning bare repository", bare_repo_path=self.bare_repo_path)
- try:
+ raise RuntimeError(f"Connection {self.git_conn_id} doesn't have a host url")
+
+ try:
+ if not os.path.exists(self.bare_repo_path):
+ self._log.info("Cloning bare repository", bare_repo_path=self.bare_repo_path)
Repo.clone_from(
url=self.repo_url,
to_path=self.bare_repo_path,
bare=True,
env=self.hook.env if self.hook else None,
)
- except GitCommandError as e:
- raise AirflowException("Error cloning repository") from e
- self.bare_repo = Repo(self.bare_repo_path)
+ self.bare_repo = Repo(self.bare_repo_path)
+ except (InvalidGitRepositoryError, GitCommandError) as e:
+ self._log.warning(
+ "Bare repository clone/open failed, cleaning up and retrying",
+ bare_repo_path=self.bare_repo_path,
+ exc=e,
+ )
+ if os.path.exists(self.bare_repo_path):
+ shutil.rmtree(self.bare_repo_path)
+ raise
def _ensure_version_in_bare_repo(self) -> None:
if not self.version:
return
if not self._has_version(self.bare_repo, self.version):
self._fetch_bare_repo()
if not self._has_version(self.bare_repo, self.version):
- raise AirflowException(f"Version {self.version} not found in the repository")
+ raise RuntimeError(f"Version {self.version} not found in the repository")
Review Comment:
Git provider is still 0.* and not a 1.* - so I think it is okay to clean the future legacy with this PR as we want to get rid of `AirflowExceptions` in general.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix corrupted bare Git repository recovery in DAG bundles [airflow]
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on code in PR #56206:
URL: https://github.com/apache/airflow/pull/56206#discussion_r2428421988
##########
providers/git/src/airflow/providers/git/bundles/git.py:
##########
@@ -91,11 +93,21 @@ def _initialize(self):
with self.lock():
cm = self.hook.configure_hook_env() if self.hook else nullcontext()
with cm:
- self._clone_bare_repo_if_required()
+ try:
+ self._clone_bare_repo_if_required()
+ except GitCommandError as e:
+ raise AirflowException("Error cloning repository") from e
+ except InvalidGitRepositoryError as e:
+ raise AirflowException(f"Invalid git repository at {self.bare_repo_path}") from e
self._ensure_version_in_bare_repo()
self.bare_repo.close()
- self._clone_repo_if_required()
+ try:
+ self._clone_repo_if_required()
+ except GitCommandError as e:
+ raise AirflowException("Error cloning repository") from e
Review Comment:
Since this is new, let's use RuntimeError exception for all the AirflowExceptions raised in this PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix corrupted bare Git repository recovery in DAG bundles [airflow]
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on code in PR #56206:
URL: https://github.com/apache/airflow/pull/56206#discussion_r2431748184
##########
providers/git/src/airflow/providers/git/bundles/git.py:
##########
@@ -109,48 +120,77 @@ def _initialize(self):
def initialize(self) -> None:
if not self.repo_url:
- raise AirflowException(f"Connection {self.git_conn_id} doesn't have a host url")
+ raise RuntimeError(f"Connection {self.git_conn_id} doesn't have a host url")
self._initialize()
super().initialize()
+ @retry(
+ retry=retry_if_exception_type((InvalidGitRepositoryError, GitCommandError)),
+ stop=stop_after_attempt(2),
+ reraise=True,
+ )
def _clone_repo_if_required(self) -> None:
- if not os.path.exists(self.repo_path):
- self._log.info("Cloning repository", repo_path=self.repo_path, bare_repo_path=self.bare_repo_path)
- try:
+ try:
+ if not os.path.exists(self.repo_path):
+ self._log.info(
+ "Cloning repository", repo_path=self.repo_path, bare_repo_path=self.bare_repo_path
+ )
Repo.clone_from(
url=self.bare_repo_path,
to_path=self.repo_path,
)
- except NoSuchPathError as e:
- # Protection should the bare repo be removed manually
- raise AirflowException("Repository path: %s not found", self.bare_repo_path) from e
- else:
- self._log.debug("repo exists", repo_path=self.repo_path)
- self.repo = Repo(self.repo_path)
-
+ else:
+ self._log.debug("repo exists", repo_path=self.repo_path)
+ self.repo = Repo(self.repo_path)
+ except NoSuchPathError as e:
+ # Protection should the bare repo be removed manually
+ raise RuntimeError("Repository path: %s not found", self.bare_repo_path) from e
+ except (InvalidGitRepositoryError, GitCommandError) as e:
+ self._log.warning(
+ "Repository clone/open failed, cleaning up and retrying",
+ repo_path=self.repo_path,
+ exc=e,
+ )
+ if os.path.exists(self.repo_path):
+ shutil.rmtree(self.repo_path)
+ raise
+
+ @retry(
+ retry=retry_if_exception_type((InvalidGitRepositoryError, GitCommandError)),
+ stop=stop_after_attempt(2),
+ reraise=True,
+ )
def _clone_bare_repo_if_required(self) -> None:
if not self.repo_url:
- raise AirflowException(f"Connection {self.git_conn_id} doesn't have a host url")
- if not os.path.exists(self.bare_repo_path):
- self._log.info("Cloning bare repository", bare_repo_path=self.bare_repo_path)
- try:
+ raise RuntimeError(f"Connection {self.git_conn_id} doesn't have a host url")
+
+ try:
+ if not os.path.exists(self.bare_repo_path):
+ self._log.info("Cloning bare repository", bare_repo_path=self.bare_repo_path)
Repo.clone_from(
url=self.repo_url,
to_path=self.bare_repo_path,
bare=True,
env=self.hook.env if self.hook else None,
)
- except GitCommandError as e:
- raise AirflowException("Error cloning repository") from e
- self.bare_repo = Repo(self.bare_repo_path)
+ self.bare_repo = Repo(self.bare_repo_path)
+ except (InvalidGitRepositoryError, GitCommandError) as e:
+ self._log.warning(
+ "Bare repository clone/open failed, cleaning up and retrying",
+ bare_repo_path=self.bare_repo_path,
+ exc=e,
+ )
+ if os.path.exists(self.bare_repo_path):
+ shutil.rmtree(self.bare_repo_path)
+ raise
def _ensure_version_in_bare_repo(self) -> None:
if not self.version:
return
if not self._has_version(self.bare_repo, self.version):
self._fetch_bare_repo()
if not self._has_version(self.bare_repo, self.version):
- raise AirflowException(f"Version {self.version} not found in the repository")
+ raise RuntimeError(f"Version {self.version} not found in the repository")
Review Comment:
Nope. I didn't ask you to change all the AirflowException to RuntimeError. I only asked you to change the AirflowException to RuntimeError for the new ones you raised. Please it's a breaking change to change all the exceptions
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix corrupted bare Git repository recovery in DAG bundles [airflow]
Posted by "dheerajturaga (via GitHub)" <gi...@apache.org>.
dheerajturaga commented on PR #56206:
URL: https://github.com/apache/airflow/pull/56206#issuecomment-3393383309
cc: @jedcunningham @ephraimbuddy @kaxil @jscheffl
When using gitbundles with edge workers (EdgeExectutor) it could happen that the git connection could be unstable causing git clone/ git bare clone to fail. If the bare clone is broken, all subsequent tasks on the worker fail unless you manually ssh onto the machine to tidy up the bare repo. This is an attempt to self heal
Im hoping to get this in the next wave of provider release. cc: @eladkal
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix corrupted bare Git repository recovery in DAG bundles [airflow]
Posted by "dheerajturaga (via GitHub)" <gi...@apache.org>.
dheerajturaga commented on PR #56206:
URL: https://github.com/apache/airflow/pull/56206#issuecomment-3393383631
> Not required, but just a thought, we could use tenacity, here instead of a custom retry loop. It might make the retry/backoff logic easier to read/maintain. For example:
>
> ```python
> @retry(stop=stop_after_attempt(3), wait=wait_exponential(), reraise=True)
> def clone_bare_repo(url, path, env=None):
> return Repo.clone_from(url, path, bare=True, env=env)
> ```
good idea!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix corrupted bare Git repository recovery in DAG bundles [airflow]
Posted by "ephraimbuddy (via GitHub)" <gi...@apache.org>.
ephraimbuddy commented on code in PR #56206:
URL: https://github.com/apache/airflow/pull/56206#discussion_r2435628436
##########
providers/git/src/airflow/providers/git/bundles/git.py:
##########
@@ -109,48 +120,77 @@ def _initialize(self):
def initialize(self) -> None:
if not self.repo_url:
- raise AirflowException(f"Connection {self.git_conn_id} doesn't have a host url")
+ raise RuntimeError(f"Connection {self.git_conn_id} doesn't have a host url")
self._initialize()
super().initialize()
+ @retry(
+ retry=retry_if_exception_type((InvalidGitRepositoryError, GitCommandError)),
+ stop=stop_after_attempt(2),
+ reraise=True,
+ )
def _clone_repo_if_required(self) -> None:
- if not os.path.exists(self.repo_path):
- self._log.info("Cloning repository", repo_path=self.repo_path, bare_repo_path=self.bare_repo_path)
- try:
+ try:
+ if not os.path.exists(self.repo_path):
+ self._log.info(
+ "Cloning repository", repo_path=self.repo_path, bare_repo_path=self.bare_repo_path
+ )
Repo.clone_from(
url=self.bare_repo_path,
to_path=self.repo_path,
)
- except NoSuchPathError as e:
- # Protection should the bare repo be removed manually
- raise AirflowException("Repository path: %s not found", self.bare_repo_path) from e
- else:
- self._log.debug("repo exists", repo_path=self.repo_path)
- self.repo = Repo(self.repo_path)
-
+ else:
+ self._log.debug("repo exists", repo_path=self.repo_path)
+ self.repo = Repo(self.repo_path)
+ except NoSuchPathError as e:
+ # Protection should the bare repo be removed manually
+ raise RuntimeError("Repository path: %s not found", self.bare_repo_path) from e
+ except (InvalidGitRepositoryError, GitCommandError) as e:
+ self._log.warning(
+ "Repository clone/open failed, cleaning up and retrying",
+ repo_path=self.repo_path,
+ exc=e,
+ )
+ if os.path.exists(self.repo_path):
+ shutil.rmtree(self.repo_path)
+ raise
+
+ @retry(
+ retry=retry_if_exception_type((InvalidGitRepositoryError, GitCommandError)),
+ stop=stop_after_attempt(2),
+ reraise=True,
+ )
def _clone_bare_repo_if_required(self) -> None:
if not self.repo_url:
- raise AirflowException(f"Connection {self.git_conn_id} doesn't have a host url")
- if not os.path.exists(self.bare_repo_path):
- self._log.info("Cloning bare repository", bare_repo_path=self.bare_repo_path)
- try:
+ raise RuntimeError(f"Connection {self.git_conn_id} doesn't have a host url")
+
+ try:
+ if not os.path.exists(self.bare_repo_path):
+ self._log.info("Cloning bare repository", bare_repo_path=self.bare_repo_path)
Repo.clone_from(
url=self.repo_url,
to_path=self.bare_repo_path,
bare=True,
env=self.hook.env if self.hook else None,
)
- except GitCommandError as e:
- raise AirflowException("Error cloning repository") from e
- self.bare_repo = Repo(self.bare_repo_path)
+ self.bare_repo = Repo(self.bare_repo_path)
+ except (InvalidGitRepositoryError, GitCommandError) as e:
+ self._log.warning(
+ "Bare repository clone/open failed, cleaning up and retrying",
+ bare_repo_path=self.bare_repo_path,
+ exc=e,
+ )
+ if os.path.exists(self.bare_repo_path):
+ shutil.rmtree(self.bare_repo_path)
+ raise
def _ensure_version_in_bare_repo(self) -> None:
if not self.version:
return
if not self._has_version(self.bare_repo, self.version):
self._fetch_bare_repo()
if not self._has_version(self.bare_repo, self.version):
- raise AirflowException(f"Version {self.version} not found in the repository")
+ raise RuntimeError(f"Version {self.version} not found in the repository")
Review Comment:
@jscheffl , yeah, makes sense. We could do that in a separate PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix corrupted bare Git repository recovery in DAG bundles [airflow]
Posted by "dheerajturaga (via GitHub)" <gi...@apache.org>.
dheerajturaga commented on code in PR #56206:
URL: https://github.com/apache/airflow/pull/56206#discussion_r2430310312
##########
providers/git/src/airflow/providers/git/bundles/git.py:
##########
@@ -91,11 +93,21 @@ def _initialize(self):
with self.lock():
cm = self.hook.configure_hook_env() if self.hook else nullcontext()
with cm:
- self._clone_bare_repo_if_required()
+ try:
+ self._clone_bare_repo_if_required()
+ except GitCommandError as e:
+ raise AirflowException("Error cloning repository") from e
+ except InvalidGitRepositoryError as e:
+ raise AirflowException(f"Invalid git repository at {self.bare_repo_path}") from e
self._ensure_version_in_bare_repo()
self.bare_repo.close()
- self._clone_repo_if_required()
+ try:
+ self._clone_repo_if_required()
+ except GitCommandError as e:
+ raise AirflowException("Error cloning repository") from e
Review Comment:
Done! Please let me know if this can be merged
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix corrupted bare Git repository recovery in DAG bundles [airflow]
Posted by "dheerajturaga (via GitHub)" <gi...@apache.org>.
dheerajturaga commented on code in PR #56206:
URL: https://github.com/apache/airflow/pull/56206#discussion_r2434093881
##########
providers/git/src/airflow/providers/git/bundles/git.py:
##########
@@ -109,48 +120,77 @@ def _initialize(self):
def initialize(self) -> None:
if not self.repo_url:
- raise AirflowException(f"Connection {self.git_conn_id} doesn't have a host url")
+ raise RuntimeError(f"Connection {self.git_conn_id} doesn't have a host url")
self._initialize()
super().initialize()
+ @retry(
+ retry=retry_if_exception_type((InvalidGitRepositoryError, GitCommandError)),
+ stop=stop_after_attempt(2),
+ reraise=True,
+ )
def _clone_repo_if_required(self) -> None:
- if not os.path.exists(self.repo_path):
- self._log.info("Cloning repository", repo_path=self.repo_path, bare_repo_path=self.bare_repo_path)
- try:
+ try:
+ if not os.path.exists(self.repo_path):
+ self._log.info(
+ "Cloning repository", repo_path=self.repo_path, bare_repo_path=self.bare_repo_path
+ )
Repo.clone_from(
url=self.bare_repo_path,
to_path=self.repo_path,
)
- except NoSuchPathError as e:
- # Protection should the bare repo be removed manually
- raise AirflowException("Repository path: %s not found", self.bare_repo_path) from e
- else:
- self._log.debug("repo exists", repo_path=self.repo_path)
- self.repo = Repo(self.repo_path)
-
+ else:
+ self._log.debug("repo exists", repo_path=self.repo_path)
+ self.repo = Repo(self.repo_path)
+ except NoSuchPathError as e:
+ # Protection should the bare repo be removed manually
+ raise RuntimeError("Repository path: %s not found", self.bare_repo_path) from e
+ except (InvalidGitRepositoryError, GitCommandError) as e:
+ self._log.warning(
+ "Repository clone/open failed, cleaning up and retrying",
+ repo_path=self.repo_path,
+ exc=e,
+ )
+ if os.path.exists(self.repo_path):
+ shutil.rmtree(self.repo_path)
+ raise
+
+ @retry(
+ retry=retry_if_exception_type((InvalidGitRepositoryError, GitCommandError)),
+ stop=stop_after_attempt(2),
+ reraise=True,
+ )
def _clone_bare_repo_if_required(self) -> None:
if not self.repo_url:
- raise AirflowException(f"Connection {self.git_conn_id} doesn't have a host url")
- if not os.path.exists(self.bare_repo_path):
- self._log.info("Cloning bare repository", bare_repo_path=self.bare_repo_path)
- try:
+ raise RuntimeError(f"Connection {self.git_conn_id} doesn't have a host url")
+
+ try:
+ if not os.path.exists(self.bare_repo_path):
+ self._log.info("Cloning bare repository", bare_repo_path=self.bare_repo_path)
Repo.clone_from(
url=self.repo_url,
to_path=self.bare_repo_path,
bare=True,
env=self.hook.env if self.hook else None,
)
- except GitCommandError as e:
- raise AirflowException("Error cloning repository") from e
- self.bare_repo = Repo(self.bare_repo_path)
+ self.bare_repo = Repo(self.bare_repo_path)
+ except (InvalidGitRepositoryError, GitCommandError) as e:
+ self._log.warning(
+ "Bare repository clone/open failed, cleaning up and retrying",
+ bare_repo_path=self.bare_repo_path,
+ exc=e,
+ )
+ if os.path.exists(self.bare_repo_path):
+ shutil.rmtree(self.bare_repo_path)
+ raise
def _ensure_version_in_bare_repo(self) -> None:
if not self.version:
return
if not self._has_version(self.bare_repo, self.version):
self._fetch_bare_repo()
if not self._has_version(self.bare_repo, self.version):
- raise AirflowException(f"Version {self.version} not found in the repository")
+ raise RuntimeError(f"Version {self.version} not found in the repository")
Review Comment:
@ephraimbuddy , I updated my PR to use RuntimeError instead in just the parts I have touched. Hopefully all CI will turn green. If this doesn't pass CI I would prefer to revert this commit and go back to my earlier implementation and tackle the cleanup in another PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org