We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Remove duplicate line of code for configuration copy
This is code improvement PR to remove a duplicate line. The code copies config in either cases of IF statement hence duplicate call. 
Moving that before IF makes it unique and sufficient for the purpose.

I have moved first call above IF and removed ELSE block

---
**^ Add meaningful description above**

Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/main/UPDATING.md).


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `airflow/task/task_runner/base_task_runner.py`
Content:
```
1 #
2 # Licensed to the Apache Software Foundation (ASF) under one
3 # or more contributor license agreements.  See the NOTICE file
4 # distributed with this work for additional information
5 # regarding copyright ownership.  The ASF licenses this file
6 # to you under the Apache License, Version 2.0 (the
7 # "License"); you may not use this file except in compliance
8 # with the License.  You may obtain a copy of the License at
9 #
10 #   http://www.apache.org/licenses/LICENSE-2.0
11 #
12 # Unless required by applicable law or agreed to in writing,
13 # software distributed under the License is distributed on an
14 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15 # KIND, either express or implied.  See the License for the
16 # specific language governing permissions and limitations
17 # under the License.
18 """Base task runner"""
19 import os
20 import subprocess
21 import threading
22 from pwd import getpwnam
23 from tempfile import NamedTemporaryFile
24 from typing import Optional, Union
25 
26 from airflow.configuration import conf
27 from airflow.exceptions import AirflowConfigException
28 from airflow.models.taskinstance import load_error_file
29 from airflow.utils.configuration import tmp_configuration_copy
30 from airflow.utils.log.logging_mixin import LoggingMixin
31 from airflow.utils.net import get_hostname
32 from airflow.utils.platform import getuser
33 
34 PYTHONPATH_VAR = 'PYTHONPATH'
35 
36 
37 class BaseTaskRunner(LoggingMixin):
38     """
39     Runs Airflow task instances by invoking the `airflow tasks run` command with raw
40     mode enabled in a subprocess.
41 
42     :param local_task_job: The local task job associated with running the
43         associated task instance.
44     :type local_task_job: airflow.jobs.local_task_job.LocalTaskJob
45     """
46 
47     def __init__(self, local_task_job):
48         # Pass task instance context into log handlers to setup the logger.
49         super().__init__(local_task_job.task_instance)
50         self._task_instance = local_task_job.task_instance
51 
52         popen_prepend = []
53         if self._task_instance.run_as_user:
54             self.run_as_user = self._task_instance.run_as_user
55         else:
56             try:
57                 self.run_as_user = conf.get('core', 'default_impersonation')
58             except AirflowConfigException:
59                 self.run_as_user = None
60 
61         # Add sudo commands to change user if we need to. Needed to handle SubDagOperator
62         # case using a SequentialExecutor.
63         self.log.debug("Planning to run as the %s user", self.run_as_user)
64         if self.run_as_user and (self.run_as_user != getuser()):
65             # We want to include any environment variables now, as we won't
66             # want to have to specify them in the sudo call - they would show
67             # up in `ps` that way! And run commands now, as the other user
68             # might not be able to run the cmds to get credentials
69             cfg_path = tmp_configuration_copy(chmod=0o600)
70 
71             # Give ownership of file to user; only they can read and write
72             subprocess.call(['sudo', 'chown', self.run_as_user, cfg_path], close_fds=True)
73 
74             # propagate PYTHONPATH environment variable
75             pythonpath_value = os.environ.get(PYTHONPATH_VAR, '')
76             popen_prepend = ['sudo', '-E', '-H', '-u', self.run_as_user]
77 
78             if pythonpath_value:
79                 popen_prepend.append(f'{PYTHONPATH_VAR}={pythonpath_value}')
80 
81         else:
82             # Always provide a copy of the configuration file settings. Since
83             # we are running as the same user, and can pass through environment
84             # variables then we don't need to include those in the config copy
85             # - the runner can read/execute those values as it needs
86             cfg_path = tmp_configuration_copy(chmod=0o600)
87 
88         self._error_file = NamedTemporaryFile(delete=True)
89         if self.run_as_user:
90             try:
91                 os.chown(self._error_file.name, getpwnam(self.run_as_user).pw_uid, -1)
92             except KeyError:
93                 # No user `run_as_user` found
94                 pass
95 
96         self._cfg_path = cfg_path
97         self._command = (
98             popen_prepend
99             + self._task_instance.command_as_list(
100                 raw=True,
101                 pickle_id=local_task_job.pickle_id,
102                 mark_success=local_task_job.mark_success,
103                 job_id=local_task_job.id,
104                 pool=local_task_job.pool,
105                 cfg_path=cfg_path,
106             )
107             + ["--error-file", self._error_file.name]
108         )
109         self.process = None
110 
111     def deserialize_run_error(self) -> Optional[Union[str, Exception]]:
112         """Return task runtime error if its written to provided error file."""
113         return load_error_file(self._error_file)
114 
115     def _read_task_logs(self, stream):
116         while True:
117             line = stream.readline()
118             if isinstance(line, bytes):
119                 line = line.decode('utf-8')
120             if not line:
121                 break
122             self.log.info(
123                 'Job %s: Subtask %s %s',
124                 self._task_instance.job_id,
125                 self._task_instance.task_id,
126                 line.rstrip('\n'),
127             )
128 
129     def run_command(self, run_with=None):
130         """
131         Run the task command.
132 
133         :param run_with: list of tokens to run the task command with e.g. ``['bash', '-c']``
134         :type run_with: list
135         :return: the process that was run
136         :rtype: subprocess.Popen
137         """
138         run_with = run_with or []
139         full_cmd = run_with + self._command
140 
141         self.log.info("Running on host: %s", get_hostname())
142         self.log.info('Running: %s', full_cmd)
143 
144         proc = subprocess.Popen(
145             full_cmd,
146             stdout=subprocess.PIPE,
147             stderr=subprocess.STDOUT,
148             universal_newlines=True,
149             close_fds=True,
150             env=os.environ.copy(),
151             preexec_fn=os.setsid,
152         )
153 
154         # Start daemon thread to read subprocess logging output
155         log_reader = threading.Thread(
156             target=self._read_task_logs,
157             args=(proc.stdout,),
158         )
159         log_reader.daemon = True
160         log_reader.start()
161         return proc
162 
163     def start(self):
164         """Start running the task instance in a subprocess."""
165         raise NotImplementedError()
166 
167     def return_code(self) -> Optional[int]:
168         """
169         :return: The return code associated with running the task instance or
170             None if the task is not yet done.
171         :rtype: int
172         """
173         raise NotImplementedError()
174 
175     def terminate(self) -> None:
176         """Force kill the running task instance."""
177         raise NotImplementedError()
178 
179     def on_finish(self) -> None:
180         """A callback that should be called when this is done running."""
181         if self._cfg_path and os.path.isfile(self._cfg_path):
182             if self.run_as_user:
183                 subprocess.call(['sudo', 'rm', self._cfg_path], close_fds=True)
184             else:
185                 os.remove(self._cfg_path)
186         try:
187             self._error_file.close()
188         except FileNotFoundError:
189             # The subprocess has deleted this file before we do
190             # so we ignore
191             pass
192
```

Path: `airflow/utils/configuration.py`
Content:
```
1 #
2 # Licensed to the Apache Software Foundation (ASF) under one
3 # or more contributor license agreements.  See the NOTICE file
4 # distributed with this work for additional information
5 # regarding copyright ownership.  The ASF licenses this file
6 # to you under the Apache License, Version 2.0 (the
7 # "License"); you may not use this file except in compliance
8 # with the License.  You may obtain a copy of the License at
9 #
10 #   http://www.apache.org/licenses/LICENSE-2.0
11 #
12 # Unless required by applicable law or agreed to in writing,
13 # software distributed under the License is distributed on an
14 # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15 # KIND, either express or implied.  See the License for the
16 # specific language governing permissions and limitations
17 # under the License.
18 
19 import json
20 import os
21 from tempfile import mkstemp
22 
23 from airflow.configuration import conf
24 
25 
26 def tmp_configuration_copy(chmod=0o600):
27     """
28     Returns a path for a temporary file including a full copy of the configuration
29     settings.
30     :return: a path to a temporary file
31     """
32     cfg_dict = conf.as_dict(display_sensitive=True, raw=True)
33     temp_fd, cfg_path = mkstemp()
34 
35     with os.fdopen(temp_fd, 'w') as temp_file:
36         # Set the permissions before we write anything to it.
37         if chmod is not None:
38             os.fchmod(temp_fd, chmod)
39         json.dump(cfg_dict, temp_file)
40 
41     return cfg_path
42
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/airflow/task/task_runner/base_task_runner.py b/airflow/task/task_runner/base_task_runner.py
--- a/airflow/task/task_runner/base_task_runner.py
+++ b/airflow/task/task_runner/base_task_runner.py
@@ -66,7 +66,7 @@
             # want to have to specify them in the sudo call - they would show
             # up in `ps` that way! And run commands now, as the other user
             # might not be able to run the cmds to get credentials
-            cfg_path = tmp_configuration_copy(chmod=0o600)
+            cfg_path = tmp_configuration_copy(chmod=0o600, include_env=True, include_cmds=True)
 
             # Give ownership of file to user; only they can read and write
             subprocess.call(['sudo', 'chown', self.run_as_user, cfg_path], close_fds=True)
@@ -83,7 +83,7 @@
             # we are running as the same user, and can pass through environment
             # variables then we don't need to include those in the config copy
             # - the runner can read/execute those values as it needs
-            cfg_path = tmp_configuration_copy(chmod=0o600)
+            cfg_path = tmp_configuration_copy(chmod=0o600, include_env=False, include_cmds=False)
 
         self._error_file = NamedTemporaryFile(delete=True)
         if self.run_as_user:
diff --git a/airflow/utils/configuration.py b/airflow/utils/configuration.py
--- a/airflow/utils/configuration.py
+++ b/airflow/utils/configuration.py
@@ -23,13 +23,23 @@
 from airflow.configuration import conf
 
 
-def tmp_configuration_copy(chmod=0o600):
+def tmp_configuration_copy(chmod=0o600, include_env=True, include_cmds=True):
     """
     Returns a path for a temporary file including a full copy of the configuration
     settings.
+
+    :param include_env: Should the value of configuration from ``AIRFLOW__``
+        environment variables be included or not
+    :type include_env: bool
+    :param include_cmds: Should the result of calling any *_cmd config be
+        set (True, default), or should the _cmd options be left as the
+        command to run (False)
+    :type include_cmds: bool
     :return: a path to a temporary file
     """
-    cfg_dict = conf.as_dict(display_sensitive=True, raw=True)
+    cfg_dict = conf.as_dict(
+        display_sensitive=True, raw=True, include_cmds=include_cmds, include_env=include_env
+    )
     temp_fd, cfg_path = mkstemp()
 
     with os.fdopen(temp_fd, 'w') as temp_file:

{"golden_diff": "diff --git a/airflow/task/task_runner/base_task_runner.py b/airflow/task/task_runner/base_task_runner.py\n--- a/airflow/task/task_runner/base_task_runner.py\n+++ b/airflow/task/task_runner/base_task_runner.py\n@@ -66,7 +66,7 @@\n             # want to have to specify them in the sudo call - they would show\n             # up in `ps` that way! And run commands now, as the other user\n             # might not be able to run the cmds to get credentials\n-            cfg_path = tmp_configuration_copy(chmod=0o600)\n+            cfg_path = tmp_configuration_copy(chmod=0o600, include_env=True, include_cmds=True)\n \n             # Give ownership of file to user; only they can read and write\n             subprocess.call(['sudo', 'chown', self.run_as_user, cfg_path], close_fds=True)\n@@ -83,7 +83,7 @@\n             # we are running as the same user, and can pass through environment\n             # variables then we don't need to include those in the config copy\n             # - the runner can read/execute those values as it needs\n-            cfg_path = tmp_configuration_copy(chmod=0o600)\n+            cfg_path = tmp_configuration_copy(chmod=0o600, include_env=False, include_cmds=False)\n \n         self._error_file = NamedTemporaryFile(delete=True)\n         if self.run_as_user:\ndiff --git a/airflow/utils/configuration.py b/airflow/utils/configuration.py\n--- a/airflow/utils/configuration.py\n+++ b/airflow/utils/configuration.py\n@@ -23,13 +23,23 @@\n from airflow.configuration import conf\n \n \n-def tmp_configuration_copy(chmod=0o600):\n+def tmp_configuration_copy(chmod=0o600, include_env=True, include_cmds=True):\n     \"\"\"\n     Returns a path for a temporary file including a full copy of the configuration\n     settings.\n+\n+    :param include_env: Should the value of configuration from ``AIRFLOW__``\n+        environment variables be included or not\n+    :type include_env: bool\n+    :param include_cmds: Should the result of calling any *_cmd config be\n+        set (True, default), or should the _cmd options be left as the\n+        command to run (False)\n+    :type include_cmds: bool\n     :return: a path to a temporary file\n     \"\"\"\n-    cfg_dict = conf.as_dict(display_sensitive=True, raw=True)\n+    cfg_dict = conf.as_dict(\n+        display_sensitive=True, raw=True, include_cmds=include_cmds, include_env=include_env\n+    )\n     temp_fd, cfg_path = mkstemp()\n \n     with os.fdopen(temp_fd, 'w') as temp_file:\n", "issue": "Remove duplicate line of code for configuration copy\nThis is code improvement PR to remove a duplicate line. The code copies config in either cases of IF statement hence duplicate call. \r\nMoving that before IF makes it unique and sufficient for the purpose.\r\n\r\nI have moved first call above IF and removed ELSE block\r\n\r\n---\r\n**^ Add meaningful description above**\r\n\r\nRead the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst#pull-request-guidelines)** for more information.\r\nIn case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.\r\nIn case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).\r\nIn case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/main/UPDATING.md).\r\n\n", "before_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\"\"\"Base task runner\"\"\"\nimport os\nimport subprocess\nimport threading\nfrom pwd import getpwnam\nfrom tempfile import NamedTemporaryFile\nfrom typing import Optional, Union\n\nfrom airflow.configuration import conf\nfrom airflow.exceptions import AirflowConfigException\nfrom airflow.models.taskinstance import load_error_file\nfrom airflow.utils.configuration import tmp_configuration_copy\nfrom airflow.utils.log.logging_mixin import LoggingMixin\nfrom airflow.utils.net import get_hostname\nfrom airflow.utils.platform import getuser\n\nPYTHONPATH_VAR = 'PYTHONPATH'\n\n\nclass BaseTaskRunner(LoggingMixin):\n    \"\"\"\n    Runs Airflow task instances by invoking the `airflow tasks run` command with raw\n    mode enabled in a subprocess.\n\n    :param local_task_job: The local task job associated with running the\n        associated task instance.\n    :type local_task_job: airflow.jobs.local_task_job.LocalTaskJob\n    \"\"\"\n\n    def __init__(self, local_task_job):\n        # Pass task instance context into log handlers to setup the logger.\n        super().__init__(local_task_job.task_instance)\n        self._task_instance = local_task_job.task_instance\n\n        popen_prepend = []\n        if self._task_instance.run_as_user:\n            self.run_as_user = self._task_instance.run_as_user\n        else:\n            try:\n                self.run_as_user = conf.get('core', 'default_impersonation')\n            except AirflowConfigException:\n                self.run_as_user = None\n\n        # Add sudo commands to change user if we need to. Needed to handle SubDagOperator\n        # case using a SequentialExecutor.\n        self.log.debug(\"Planning to run as the %s user\", self.run_as_user)\n        if self.run_as_user and (self.run_as_user != getuser()):\n            # We want to include any environment variables now, as we won't\n            # want to have to specify them in the sudo call - they would show\n            # up in `ps` that way! And run commands now, as the other user\n            # might not be able to run the cmds to get credentials\n            cfg_path = tmp_configuration_copy(chmod=0o600)\n\n            # Give ownership of file to user; only they can read and write\n            subprocess.call(['sudo', 'chown', self.run_as_user, cfg_path], close_fds=True)\n\n            # propagate PYTHONPATH environment variable\n            pythonpath_value = os.environ.get(PYTHONPATH_VAR, '')\n            popen_prepend = ['sudo', '-E', '-H', '-u', self.run_as_user]\n\n            if pythonpath_value:\n                popen_prepend.append(f'{PYTHONPATH_VAR}={pythonpath_value}')\n\n        else:\n            # Always provide a copy of the configuration file settings. Since\n            # we are running as the same user, and can pass through environment\n            # variables then we don't need to include those in the config copy\n            # - the runner can read/execute those values as it needs\n            cfg_path = tmp_configuration_copy(chmod=0o600)\n\n        self._error_file = NamedTemporaryFile(delete=True)\n        if self.run_as_user:\n            try:\n                os.chown(self._error_file.name, getpwnam(self.run_as_user).pw_uid, -1)\n            except KeyError:\n                # No user `run_as_user` found\n                pass\n\n        self._cfg_path = cfg_path\n        self._command = (\n            popen_prepend\n            + self._task_instance.command_as_list(\n                raw=True,\n                pickle_id=local_task_job.pickle_id,\n                mark_success=local_task_job.mark_success,\n                job_id=local_task_job.id,\n                pool=local_task_job.pool,\n                cfg_path=cfg_path,\n            )\n            + [\"--error-file\", self._error_file.name]\n        )\n        self.process = None\n\n    def deserialize_run_error(self) -> Optional[Union[str, Exception]]:\n        \"\"\"Return task runtime error if its written to provided error file.\"\"\"\n        return load_error_file(self._error_file)\n\n    def _read_task_logs(self, stream):\n        while True:\n            line = stream.readline()\n            if isinstance(line, bytes):\n                line = line.decode('utf-8')\n            if not line:\n                break\n            self.log.info(\n                'Job %s: Subtask %s %s',\n                self._task_instance.job_id,\n                self._task_instance.task_id,\n                line.rstrip('\\n'),\n            )\n\n    def run_command(self, run_with=None):\n        \"\"\"\n        Run the task command.\n\n        :param run_with: list of tokens to run the task command with e.g. ``['bash', '-c']``\n        :type run_with: list\n        :return: the process that was run\n        :rtype: subprocess.Popen\n        \"\"\"\n        run_with = run_with or []\n        full_cmd = run_with + self._command\n\n        self.log.info(\"Running on host: %s\", get_hostname())\n        self.log.info('Running: %s', full_cmd)\n\n        proc = subprocess.Popen(\n            full_cmd,\n            stdout=subprocess.PIPE,\n            stderr=subprocess.STDOUT,\n            universal_newlines=True,\n            close_fds=True,\n            env=os.environ.copy(),\n            preexec_fn=os.setsid,\n        )\n\n        # Start daemon thread to read subprocess logging output\n        log_reader = threading.Thread(\n            target=self._read_task_logs,\n            args=(proc.stdout,),\n        )\n        log_reader.daemon = True\n        log_reader.start()\n        return proc\n\n    def start(self):\n        \"\"\"Start running the task instance in a subprocess.\"\"\"\n        raise NotImplementedError()\n\n    def return_code(self) -> Optional[int]:\n        \"\"\"\n        :return: The return code associated with running the task instance or\n            None if the task is not yet done.\n        :rtype: int\n        \"\"\"\n        raise NotImplementedError()\n\n    def terminate(self) -> None:\n        \"\"\"Force kill the running task instance.\"\"\"\n        raise NotImplementedError()\n\n    def on_finish(self) -> None:\n        \"\"\"A callback that should be called when this is done running.\"\"\"\n        if self._cfg_path and os.path.isfile(self._cfg_path):\n            if self.run_as_user:\n                subprocess.call(['sudo', 'rm', self._cfg_path], close_fds=True)\n            else:\n                os.remove(self._cfg_path)\n        try:\n            self._error_file.close()\n        except FileNotFoundError:\n            # The subprocess has deleted this file before we do\n            # so we ignore\n            pass\n", "path": "airflow/task/task_runner/base_task_runner.py"}, {"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\nimport json\nimport os\nfrom tempfile import mkstemp\n\nfrom airflow.configuration import conf\n\n\ndef tmp_configuration_copy(chmod=0o600):\n    \"\"\"\n    Returns a path for a temporary file including a full copy of the configuration\n    settings.\n    :return: a path to a temporary file\n    \"\"\"\n    cfg_dict = conf.as_dict(display_sensitive=True, raw=True)\n    temp_fd, cfg_path = mkstemp()\n\n    with os.fdopen(temp_fd, 'w') as temp_file:\n        # Set the permissions before we write anything to it.\n        if chmod is not None:\n            os.fchmod(temp_fd, chmod)\n        json.dump(cfg_dict, temp_file)\n\n    return cfg_path\n", "path": "airflow/utils/configuration.py"}], "after_files": [{"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\"\"\"Base task runner\"\"\"\nimport os\nimport subprocess\nimport threading\nfrom pwd import getpwnam\nfrom tempfile import NamedTemporaryFile\nfrom typing import Optional, Union\n\nfrom airflow.configuration import conf\nfrom airflow.exceptions import AirflowConfigException\nfrom airflow.models.taskinstance import load_error_file\nfrom airflow.utils.configuration import tmp_configuration_copy\nfrom airflow.utils.log.logging_mixin import LoggingMixin\nfrom airflow.utils.net import get_hostname\nfrom airflow.utils.platform import getuser\n\nPYTHONPATH_VAR = 'PYTHONPATH'\n\n\nclass BaseTaskRunner(LoggingMixin):\n    \"\"\"\n    Runs Airflow task instances by invoking the `airflow tasks run` command with raw\n    mode enabled in a subprocess.\n\n    :param local_task_job: The local task job associated with running the\n        associated task instance.\n    :type local_task_job: airflow.jobs.local_task_job.LocalTaskJob\n    \"\"\"\n\n    def __init__(self, local_task_job):\n        # Pass task instance context into log handlers to setup the logger.\n        super().__init__(local_task_job.task_instance)\n        self._task_instance = local_task_job.task_instance\n\n        popen_prepend = []\n        if self._task_instance.run_as_user:\n            self.run_as_user = self._task_instance.run_as_user\n        else:\n            try:\n                self.run_as_user = conf.get('core', 'default_impersonation')\n            except AirflowConfigException:\n                self.run_as_user = None\n\n        # Add sudo commands to change user if we need to. Needed to handle SubDagOperator\n        # case using a SequentialExecutor.\n        self.log.debug(\"Planning to run as the %s user\", self.run_as_user)\n        if self.run_as_user and (self.run_as_user != getuser()):\n            # We want to include any environment variables now, as we won't\n            # want to have to specify them in the sudo call - they would show\n            # up in `ps` that way! And run commands now, as the other user\n            # might not be able to run the cmds to get credentials\n            cfg_path = tmp_configuration_copy(chmod=0o600, include_env=True, include_cmds=True)\n\n            # Give ownership of file to user; only they can read and write\n            subprocess.call(['sudo', 'chown', self.run_as_user, cfg_path], close_fds=True)\n\n            # propagate PYTHONPATH environment variable\n            pythonpath_value = os.environ.get(PYTHONPATH_VAR, '')\n            popen_prepend = ['sudo', '-E', '-H', '-u', self.run_as_user]\n\n            if pythonpath_value:\n                popen_prepend.append(f'{PYTHONPATH_VAR}={pythonpath_value}')\n\n        else:\n            # Always provide a copy of the configuration file settings. Since\n            # we are running as the same user, and can pass through environment\n            # variables then we don't need to include those in the config copy\n            # - the runner can read/execute those values as it needs\n            cfg_path = tmp_configuration_copy(chmod=0o600, include_env=False, include_cmds=False)\n\n        self._error_file = NamedTemporaryFile(delete=True)\n        if self.run_as_user:\n            try:\n                os.chown(self._error_file.name, getpwnam(self.run_as_user).pw_uid, -1)\n            except KeyError:\n                # No user `run_as_user` found\n                pass\n\n        self._cfg_path = cfg_path\n        self._command = (\n            popen_prepend\n            + self._task_instance.command_as_list(\n                raw=True,\n                pickle_id=local_task_job.pickle_id,\n                mark_success=local_task_job.mark_success,\n                job_id=local_task_job.id,\n                pool=local_task_job.pool,\n                cfg_path=cfg_path,\n            )\n            + [\"--error-file\", self._error_file.name]\n        )\n        self.process = None\n\n    def deserialize_run_error(self) -> Optional[Union[str, Exception]]:\n        \"\"\"Return task runtime error if its written to provided error file.\"\"\"\n        return load_error_file(self._error_file)\n\n    def _read_task_logs(self, stream):\n        while True:\n            line = stream.readline()\n            if isinstance(line, bytes):\n                line = line.decode('utf-8')\n            if not line:\n                break\n            self.log.info(\n                'Job %s: Subtask %s %s',\n                self._task_instance.job_id,\n                self._task_instance.task_id,\n                line.rstrip('\\n'),\n            )\n\n    def run_command(self, run_with=None):\n        \"\"\"\n        Run the task command.\n\n        :param run_with: list of tokens to run the task command with e.g. ``['bash', '-c']``\n        :type run_with: list\n        :return: the process that was run\n        :rtype: subprocess.Popen\n        \"\"\"\n        run_with = run_with or []\n        full_cmd = run_with + self._command\n\n        self.log.info(\"Running on host: %s\", get_hostname())\n        self.log.info('Running: %s', full_cmd)\n\n        proc = subprocess.Popen(\n            full_cmd,\n            stdout=subprocess.PIPE,\n            stderr=subprocess.STDOUT,\n            universal_newlines=True,\n            close_fds=True,\n            env=os.environ.copy(),\n            preexec_fn=os.setsid,\n        )\n\n        # Start daemon thread to read subprocess logging output\n        log_reader = threading.Thread(\n            target=self._read_task_logs,\n            args=(proc.stdout,),\n        )\n        log_reader.daemon = True\n        log_reader.start()\n        return proc\n\n    def start(self):\n        \"\"\"Start running the task instance in a subprocess.\"\"\"\n        raise NotImplementedError()\n\n    def return_code(self) -> Optional[int]:\n        \"\"\"\n        :return: The return code associated with running the task instance or\n            None if the task is not yet done.\n        :rtype: int\n        \"\"\"\n        raise NotImplementedError()\n\n    def terminate(self) -> None:\n        \"\"\"Force kill the running task instance.\"\"\"\n        raise NotImplementedError()\n\n    def on_finish(self) -> None:\n        \"\"\"A callback that should be called when this is done running.\"\"\"\n        if self._cfg_path and os.path.isfile(self._cfg_path):\n            if self.run_as_user:\n                subprocess.call(['sudo', 'rm', self._cfg_path], close_fds=True)\n            else:\n                os.remove(self._cfg_path)\n        try:\n            self._error_file.close()\n        except FileNotFoundError:\n            # The subprocess has deleted this file before we do\n            # so we ignore\n            pass\n", "path": "airflow/task/task_runner/base_task_runner.py"}, {"content": "#\n# Licensed to the Apache Software Foundation (ASF) under one\n# or more contributor license agreements.  See the NOTICE file\n# distributed with this work for additional information\n# regarding copyright ownership.  The ASF licenses this file\n# to you under the Apache License, Version 2.0 (the\n# \"License\"); you may not use this file except in compliance\n# with the License.  You may obtain a copy of the License at\n#\n#   http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing,\n# software distributed under the License is distributed on an\n# \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY\n# KIND, either express or implied.  See the License for the\n# specific language governing permissions and limitations\n# under the License.\n\nimport json\nimport os\nfrom tempfile import mkstemp\n\nfrom airflow.configuration import conf\n\n\ndef tmp_configuration_copy(chmod=0o600, include_env=True, include_cmds=True):\n    \"\"\"\n    Returns a path for a temporary file including a full copy of the configuration\n    settings.\n\n    :param include_env: Should the value of configuration from ``AIRFLOW__``\n        environment variables be included or not\n    :type include_env: bool\n    :param include_cmds: Should the result of calling any *_cmd config be\n        set (True, default), or should the _cmd options be left as the\n        command to run (False)\n    :type include_cmds: bool\n    :return: a path to a temporary file\n    \"\"\"\n    cfg_dict = conf.as_dict(\n        display_sensitive=True, raw=True, include_cmds=include_cmds, include_env=include_env\n    )\n    temp_fd, cfg_path = mkstemp()\n\n    with os.fdopen(temp_fd, 'w') as temp_file:\n        # Set the permissions before we write anything to it.\n        if chmod is not None:\n            os.fchmod(temp_fd, chmod)\n        json.dump(cfg_dict, temp_file)\n\n    return cfg_path\n", "path": "airflow/utils/configuration.py"}]}