WIP - Fix k8s executor when home path different

airflow/cli/commands/task_command.py

@@ -51,7 +51,7 @@
     get_dag_by_file_location,
     get_dag_by_pickle,
     get_dags,
-    suppress_logs_and_warning,
+    suppress_logs_and_warning, process_subdir,
 )
 from airflow.utils.dates import timezone
 from airflow.utils.log.logging_mixin import StreamLogWriter
@@ -243,6 +243,7 @@ def _run_task_by_local_task_job(args, ti):
         ignore_ti_state=args.force,
         pool=args.pool,
         external_executor_id=_extract_external_executor_id(args),
+        subdir=process_subdir(args.subdir),
     )
     try:
         run_job.run()

airflow/jobs/local_task_job.py

@@ -56,6 +56,7 @@ def __init__(
         pickle_id: str | None = None,
         pool: str | None = None,
         external_executor_id: str | None = None,
+        subdir:str | None = None,
         *args,
         **kwargs,
     ):
@@ -69,6 +70,7 @@ def __init__(
         self.pickle_id = pickle_id
         self.mark_success = mark_success
         self.external_executor_id = external_executor_id
+        self.subdir = subdir
 
         # terminating state is used so that a job don't try to
         # terminate multiple times

airflow/models/taskinstance.py

@@ -29,6 +29,7 @@
 from collections import defaultdict
 from datetime import datetime, timedelta
 from functools import partial
+from pathlib import Path
 from types import TracebackType
 from typing import (
     TYPE_CHECKING,
@@ -662,6 +663,7 @@ def command_as_list(
         job_id=None,
         pool=None,
         cfg_path=None,
+        subdir=None,
     ):
         """
         Returns a command that can be executed anywhere where airflow is
@@ -678,7 +680,9 @@ def command_as_list(
         should_pass_filepath = not pickle_id and dag
         path = None
         if should_pass_filepath:
-            if dag.is_subdag:
+            if subdir:
+                path = Path(subdir)
+            elif dag.is_subdag:
                 path = dag.parent_dag.relative_fileloc
             else:
                 path = dag.relative_fileloc

airflow/task/task_runner/base_task_runner.py

@@ -98,6 +98,7 @@ def __init__(self, local_task_job):
             job_id=local_task_job.id,
             pool=local_task_job.pool,
             cfg_path=cfg_path,
+            subdir=local_task_job.subdir,
         )
         self.process = None