-
Notifications
You must be signed in to change notification settings - Fork 14.2k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Fix template_ext processing for Kubernetes Pod Operator #17186
Comments
In my ideal world, passing a path as an argument shouldn't need special escaping, and we should require something like BashOperator(bash_command=Tempalte('/path/to/my-script.sh')) to trigger templating. More realistic though, something like MyOperator(path=Literal('/path/to/my-file.json')) should work? |
The problem is that we already have operators that heavily rely on this behaviour and will replace ANY string ending with the designated extensions with the jinja-processed value of the files they are pointing to. This is convenient, though yeah - super-ambiguous in a number of cases, and I think we need a tactical solution for now to handle confusion it creates. I think the "recursive" behaviour is relatively new and unintended behaviour and we could address it in patchlevel release classifying it as a "bug" rather than breaking feature. Then if we do above, I think we might even not need the "Escaping" but straight away implementing some "explicit" way of marking the template files - a variant of what you proposed @uranusjr and deprecate the usage of implicit extension handling with warning, targeting its removal for Airflow 3. BTW - I thought Literal (https://www.python.org/dev/peps/pep-0586/) is a "type-checking" feature only, not actual class you can instantiate? Am I wrong :D ? |
Oh, of course we shouldn't use |
Airflow shouldn't try to template the following. It's a bug -- for recursive we look for
|
But we have 'env_vars' in K8SPodOperator
(and I find it pretty useful to be honest) |
I think you are confusing 2 things:
airflow/tests/models/test_baseoperator.py Lines 141 to 180 in 4dc622d
airflow/tests/models/test_baseoperator.py Lines 121 to 123 in 4dc622d
Logic - Flow of code: airflow/airflow/models/baseoperator.py Lines 1047 to 1078 in 4dc622d
For nested it should do: airflow/airflow/models/baseoperator.py Lines 1080 to 1091 in 4dc622d
Since env_vars is a list, the items inside that list will be rendered if it is a string but since it is The only reason it is doing that is we override that method for KPO: airflow/airflow/providers/cncf/kubernetes/operators/kubernetes_pod.py Lines 284 to 302 in 4dc622d
So the issue / bug is because of a fix in #14123 - and it does not affect any other operators. |
Ah ok. I see now :) |
I have changed the name to correctly reflect the discussion results. |
This commit effectively revers apache#15942 because of the issues it is causing. Until we find better solution we should revert this change Error: ``` [2021-07-26 20:23:54,109] {taskinstance.py:1108} INFO - Executing <Task(KubernetesPodOperator): write-xcom> on 2021-07-26T20:23:27.058907+00:00 [2021-07-26 20:23:54,113] {standard_task_runner.py:52} INFO - Started process 11 to run task [2021-07-26 20:23:54,297] {standard_task_runner.py:76} INFO - Running: ['airflow', 'tasks', 'run', 'k8_pod_operator_xcom', 'write-xcom', '2021-07-26T20:23:27.058907+00:00', '--job-id', '1757', '--pool', 'default_pool', '--raw', '--subdir', 'DAGS_FOLDER/k8s_xcom_example.py', '--cfg-path', '/tmp/tmp0q94pkhs', '--error-file', '/tmp/tmpoz9qqp2l'] [2021-07-26 20:23:54,298] {standard_task_runner.py:77} INFO - Job 1757: Subtask write-xcom [2021-07-26 20:23:54,511] {logging_mixin.py:104} INFO - Running <TaskInstance: k8_pod_operator_xcom.write-xcom 2021-07-26T20:23:27.058907+00:00 [running]> on host k8podoperatorxcomwritexcom.21384021df914227ad4e4b3a34313710 [2021-07-26 20:23:54,713] {taskinstance.py:1502} ERROR - Task failed with exception Traceback (most recent call last): File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1158, in _run_raw_task self._prepare_and_execute_task_with_callbacks(context, task) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1295, in _prepare_and_execute_task_with_callbacks self.render_templates(context=context) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1796, in render_templates self.task.render_template_fields(context) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 999, in render_template_fields self._do_render_template_fields(self, self.template_fields, context, jinja_env, set()) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1012, in _do_render_template_fields rendered_content = self.render_template(content, context, jinja_env, seen_oids) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in render_template return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in <listcomp> return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1047, in render_template return jinja_env.get_template(content).render(**context) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 883, in get_template return self._load_template(name, self.make_globals(globals)) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 857, in _load_template template = self.loader.load(self, name, globals) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 115, in load source, filename, uptodate = self.get_source(environment, name) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 197, in get_source raise TemplateNotFound(template) jinja2.exceptions.TemplateNotFound: mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json [2021-07-26 20:23:54,797] {taskinstance.py:1552} INFO - Marking task as FAILED. dag_id=k8_pod_operator_xcom, task_id=write-xcom, execution_date=20210726T202327, start_date=20210726T202353, end_date=20210726T202354 [2021-07-26 20:23:54,936] {local_task_job.py:153} INFO - Task exited with return code 1 ``` Dag: ``` from airflow import DAG from airflow.operators.bash import BashOperator from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import ( KubernetesPodOperator, ) from airflow.utils.dates import days_ago from airflow.configuration import conf namespace = conf.get("kubernetes", "NAMESPACE") # This will detect the default namespace locally and read the # environment namespace when deployed to Astronomer. if namespace == "default": config_file = "/usr/local/airflow/include/.kube/config" in_cluster = False else: in_cluster = True config_file = None default_args = { "owner": "airflow", } with DAG( dag_id="k8_pod_operator_xcom", default_args=default_args, schedule_interval=None, start_date=days_ago(2), tags=["k8"], ) as dag: write_xcom = KubernetesPodOperator( namespace=namespace, in_cluster=in_cluster, config_file=config_file, image="ubuntu", cmds=[ "sh", "-c", "mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json", ], name="write-xcom", do_xcom_push=True, is_delete_operator_pod=True, task_id="write-xcom", get_logs=True, ) pod_task_xcom_result = BashOperator( bash_command="echo \"{{ task_instance.xcom_pull('write-xcom')[0] }}\"", task_id="pod_task_xcom_result", ) write_xcom >> pod_task_xcom_result ``` closes apache#17186
This commit effectively revers #15942 because of the issues it is causing. Until we find better solution we should revert this change Error: ``` [2021-07-26 20:23:54,109] {taskinstance.py:1108} INFO - Executing <Task(KubernetesPodOperator): write-xcom> on 2021-07-26T20:23:27.058907+00:00 [2021-07-26 20:23:54,113] {standard_task_runner.py:52} INFO - Started process 11 to run task [2021-07-26 20:23:54,297] {standard_task_runner.py:76} INFO - Running: ['airflow', 'tasks', 'run', 'k8_pod_operator_xcom', 'write-xcom', '2021-07-26T20:23:27.058907+00:00', '--job-id', '1757', '--pool', 'default_pool', '--raw', '--subdir', 'DAGS_FOLDER/k8s_xcom_example.py', '--cfg-path', '/tmp/tmp0q94pkhs', '--error-file', '/tmp/tmpoz9qqp2l'] [2021-07-26 20:23:54,298] {standard_task_runner.py:77} INFO - Job 1757: Subtask write-xcom [2021-07-26 20:23:54,511] {logging_mixin.py:104} INFO - Running <TaskInstance: k8_pod_operator_xcom.write-xcom 2021-07-26T20:23:27.058907+00:00 [running]> on host k8podoperatorxcomwritexcom.21384021df914227ad4e4b3a34313710 [2021-07-26 20:23:54,713] {taskinstance.py:1502} ERROR - Task failed with exception Traceback (most recent call last): File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1158, in _run_raw_task self._prepare_and_execute_task_with_callbacks(context, task) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1295, in _prepare_and_execute_task_with_callbacks self.render_templates(context=context) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1796, in render_templates self.task.render_template_fields(context) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 999, in render_template_fields self._do_render_template_fields(self, self.template_fields, context, jinja_env, set()) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1012, in _do_render_template_fields rendered_content = self.render_template(content, context, jinja_env, seen_oids) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in render_template return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in <listcomp> return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1047, in render_template return jinja_env.get_template(content).render(**context) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 883, in get_template return self._load_template(name, self.make_globals(globals)) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 857, in _load_template template = self.loader.load(self, name, globals) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 115, in load source, filename, uptodate = self.get_source(environment, name) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 197, in get_source raise TemplateNotFound(template) jinja2.exceptions.TemplateNotFound: mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json [2021-07-26 20:23:54,797] {taskinstance.py:1552} INFO - Marking task as FAILED. dag_id=k8_pod_operator_xcom, task_id=write-xcom, execution_date=20210726T202327, start_date=20210726T202353, end_date=20210726T202354 [2021-07-26 20:23:54,936] {local_task_job.py:153} INFO - Task exited with return code 1 ``` Dag: ``` from airflow import DAG from airflow.operators.bash import BashOperator from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import ( KubernetesPodOperator, ) from airflow.utils.dates import days_ago from airflow.configuration import conf namespace = conf.get("kubernetes", "NAMESPACE") # This will detect the default namespace locally and read the # environment namespace when deployed to Astronomer. if namespace == "default": config_file = "/usr/local/airflow/include/.kube/config" in_cluster = False else: in_cluster = True config_file = None default_args = { "owner": "airflow", } with DAG( dag_id="k8_pod_operator_xcom", default_args=default_args, schedule_interval=None, start_date=days_ago(2), tags=["k8"], ) as dag: write_xcom = KubernetesPodOperator( namespace=namespace, in_cluster=in_cluster, config_file=config_file, image="ubuntu", cmds=[ "sh", "-c", "mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json", ], name="write-xcom", do_xcom_push=True, is_delete_operator_pod=True, task_id="write-xcom", get_logs=True, ) pod_task_xcom_result = BashOperator( bash_command="echo \"{{ task_instance.xcom_pull('write-xcom')[0] }}\"", task_id="pod_task_xcom_result", ) write_xcom >> pod_task_xcom_result ``` closes #17186
apache#17186 -- This has made the XCom functionality not work with KubernetesPodOperator, this has been fixed by apache#17760 -- so we should get this out sooner rather than later as recently released Airflow 2.1.3 will pull in latest kubernetes provider when we run `pip install -U apache-airflow[cncf.kubernetes]`
#17186 -- This has made the XCom functionality not work with KubernetesPodOperator, this has been fixed by #17760 -- so we should get this out sooner rather than later as recently released Airflow 2.1.3 will pull in latest kubernetes provider when we run `pip install -U apache-airflow[cncf.kubernetes]`
For people confused by the release 2.0.2 the commit so don't use 2.0.1 ! |
This commit effectively revers apache/airflow#15942 because of the issues it is causing. Until we find better solution we should revert this change Error: ``` [2021-07-26 20:23:54,109] {taskinstance.py:1108} INFO - Executing <Task(KubernetesPodOperator): write-xcom> on 2021-07-26T20:23:27.058907+00:00 [2021-07-26 20:23:54,113] {standard_task_runner.py:52} INFO - Started process 11 to run task [2021-07-26 20:23:54,297] {standard_task_runner.py:76} INFO - Running: ['airflow', 'tasks', 'run', 'k8_pod_operator_xcom', 'write-xcom', '2021-07-26T20:23:27.058907+00:00', '--job-id', '1757', '--pool', 'default_pool', '--raw', '--subdir', 'DAGS_FOLDER/k8s_xcom_example.py', '--cfg-path', '/tmp/tmp0q94pkhs', '--error-file', '/tmp/tmpoz9qqp2l'] [2021-07-26 20:23:54,298] {standard_task_runner.py:77} INFO - Job 1757: Subtask write-xcom [2021-07-26 20:23:54,511] {logging_mixin.py:104} INFO - Running <TaskInstance: k8_pod_operator_xcom.write-xcom 2021-07-26T20:23:27.058907+00:00 [running]> on host k8podoperatorxcomwritexcom.21384021df914227ad4e4b3a34313710 [2021-07-26 20:23:54,713] {taskinstance.py:1502} ERROR - Task failed with exception Traceback (most recent call last): File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1158, in _run_raw_task self._prepare_and_execute_task_with_callbacks(context, task) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1295, in _prepare_and_execute_task_with_callbacks self.render_templates(context=context) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1796, in render_templates self.task.render_template_fields(context) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 999, in render_template_fields self._do_render_template_fields(self, self.template_fields, context, jinja_env, set()) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1012, in _do_render_template_fields rendered_content = self.render_template(content, context, jinja_env, seen_oids) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in render_template return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in <listcomp> return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1047, in render_template return jinja_env.get_template(content).render(**context) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 883, in get_template return self._load_template(name, self.make_globals(globals)) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 857, in _load_template template = self.loader.load(self, name, globals) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 115, in load source, filename, uptodate = self.get_source(environment, name) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 197, in get_source raise TemplateNotFound(template) jinja2.exceptions.TemplateNotFound: mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json [2021-07-26 20:23:54,797] {taskinstance.py:1552} INFO - Marking task as FAILED. dag_id=k8_pod_operator_xcom, task_id=write-xcom, execution_date=20210726T202327, start_date=20210726T202353, end_date=20210726T202354 [2021-07-26 20:23:54,936] {local_task_job.py:153} INFO - Task exited with return code 1 ``` Dag: ``` from airflow import DAG from airflow.operators.bash import BashOperator from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import ( KubernetesPodOperator, ) from airflow.utils.dates import days_ago from airflow.configuration import conf namespace = conf.get("kubernetes", "NAMESPACE") # This will detect the default namespace locally and read the # environment namespace when deployed to Astronomer. if namespace == "default": config_file = "/usr/local/airflow/include/.kube/config" in_cluster = False else: in_cluster = True config_file = None default_args = { "owner": "airflow", } with DAG( dag_id="k8_pod_operator_xcom", default_args=default_args, schedule_interval=None, start_date=days_ago(2), tags=["k8"], ) as dag: write_xcom = KubernetesPodOperator( namespace=namespace, in_cluster=in_cluster, config_file=config_file, image="ubuntu", cmds=[ "sh", "-c", "mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json", ], name="write-xcom", do_xcom_push=True, is_delete_operator_pod=True, task_id="write-xcom", get_logs=True, ) pod_task_xcom_result = BashOperator( bash_command="echo \"{{ task_instance.xcom_pull('write-xcom')[0] }}\"", task_id="pod_task_xcom_result", ) write_xcom >> pod_task_xcom_result ``` closes apache/airflow#17186 GitOrigin-RevId: 73d2b720e0c79323a29741882a07eb8962256762
apache/airflow#17186 -- This has made the XCom functionality not work with KubernetesPodOperator, this has been fixed by apache/airflow#17760 -- so we should get this out sooner rather than later as recently released Airflow 2.1.3 will pull in latest kubernetes provider when we run `pip install -U apache-airflow[cncf.kubernetes]` GitOrigin-RevId: bb5602c652988d0b31ea5e0db8f03725a2f22d34
This commit effectively revers apache/airflow#15942 because of the issues it is causing. Until we find better solution we should revert this change Error: ``` [2021-07-26 20:23:54,109] {taskinstance.py:1108} INFO - Executing <Task(KubernetesPodOperator): write-xcom> on 2021-07-26T20:23:27.058907+00:00 [2021-07-26 20:23:54,113] {standard_task_runner.py:52} INFO - Started process 11 to run task [2021-07-26 20:23:54,297] {standard_task_runner.py:76} INFO - Running: ['airflow', 'tasks', 'run', 'k8_pod_operator_xcom', 'write-xcom', '2021-07-26T20:23:27.058907+00:00', '--job-id', '1757', '--pool', 'default_pool', '--raw', '--subdir', 'DAGS_FOLDER/k8s_xcom_example.py', '--cfg-path', '/tmp/tmp0q94pkhs', '--error-file', '/tmp/tmpoz9qqp2l'] [2021-07-26 20:23:54,298] {standard_task_runner.py:77} INFO - Job 1757: Subtask write-xcom [2021-07-26 20:23:54,511] {logging_mixin.py:104} INFO - Running <TaskInstance: k8_pod_operator_xcom.write-xcom 2021-07-26T20:23:27.058907+00:00 [running]> on host k8podoperatorxcomwritexcom.21384021df914227ad4e4b3a34313710 [2021-07-26 20:23:54,713] {taskinstance.py:1502} ERROR - Task failed with exception Traceback (most recent call last): File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1158, in _run_raw_task self._prepare_and_execute_task_with_callbacks(context, task) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1295, in _prepare_and_execute_task_with_callbacks self.render_templates(context=context) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1796, in render_templates self.task.render_template_fields(context) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 999, in render_template_fields self._do_render_template_fields(self, self.template_fields, context, jinja_env, set()) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1012, in _do_render_template_fields rendered_content = self.render_template(content, context, jinja_env, seen_oids) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in render_template return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in <listcomp> return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1047, in render_template return jinja_env.get_template(content).render(**context) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 883, in get_template return self._load_template(name, self.make_globals(globals)) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 857, in _load_template template = self.loader.load(self, name, globals) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 115, in load source, filename, uptodate = self.get_source(environment, name) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 197, in get_source raise TemplateNotFound(template) jinja2.exceptions.TemplateNotFound: mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json [2021-07-26 20:23:54,797] {taskinstance.py:1552} INFO - Marking task as FAILED. dag_id=k8_pod_operator_xcom, task_id=write-xcom, execution_date=20210726T202327, start_date=20210726T202353, end_date=20210726T202354 [2021-07-26 20:23:54,936] {local_task_job.py:153} INFO - Task exited with return code 1 ``` Dag: ``` from airflow import DAG from airflow.operators.bash import BashOperator from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import ( KubernetesPodOperator, ) from airflow.utils.dates import days_ago from airflow.configuration import conf namespace = conf.get("kubernetes", "NAMESPACE") # This will detect the default namespace locally and read the # environment namespace when deployed to Astronomer. if namespace == "default": config_file = "/usr/local/airflow/include/.kube/config" in_cluster = False else: in_cluster = True config_file = None default_args = { "owner": "airflow", } with DAG( dag_id="k8_pod_operator_xcom", default_args=default_args, schedule_interval=None, start_date=days_ago(2), tags=["k8"], ) as dag: write_xcom = KubernetesPodOperator( namespace=namespace, in_cluster=in_cluster, config_file=config_file, image="ubuntu", cmds=[ "sh", "-c", "mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json", ], name="write-xcom", do_xcom_push=True, is_delete_operator_pod=True, task_id="write-xcom", get_logs=True, ) pod_task_xcom_result = BashOperator( bash_command="echo \"{{ task_instance.xcom_pull('write-xcom')[0] }}\"", task_id="pod_task_xcom_result", ) write_xcom >> pod_task_xcom_result ``` closes apache/airflow#17186 GitOrigin-RevId: 73d2b720e0c79323a29741882a07eb8962256762
apache/airflow#17186 -- This has made the XCom functionality not work with KubernetesPodOperator, this has been fixed by apache/airflow#17760 -- so we should get this out sooner rather than later as recently released Airflow 2.1.3 will pull in latest kubernetes provider when we run `pip install -U apache-airflow[cncf.kubernetes]` GitOrigin-RevId: bb5602c652988d0b31ea5e0db8f03725a2f22d34
This commit effectively revers apache/airflow#15942 because of the issues it is causing. Until we find better solution we should revert this change Error: ``` [2021-07-26 20:23:54,109] {taskinstance.py:1108} INFO - Executing <Task(KubernetesPodOperator): write-xcom> on 2021-07-26T20:23:27.058907+00:00 [2021-07-26 20:23:54,113] {standard_task_runner.py:52} INFO - Started process 11 to run task [2021-07-26 20:23:54,297] {standard_task_runner.py:76} INFO - Running: ['airflow', 'tasks', 'run', 'k8_pod_operator_xcom', 'write-xcom', '2021-07-26T20:23:27.058907+00:00', '--job-id', '1757', '--pool', 'default_pool', '--raw', '--subdir', 'DAGS_FOLDER/k8s_xcom_example.py', '--cfg-path', '/tmp/tmp0q94pkhs', '--error-file', '/tmp/tmpoz9qqp2l'] [2021-07-26 20:23:54,298] {standard_task_runner.py:77} INFO - Job 1757: Subtask write-xcom [2021-07-26 20:23:54,511] {logging_mixin.py:104} INFO - Running <TaskInstance: k8_pod_operator_xcom.write-xcom 2021-07-26T20:23:27.058907+00:00 [running]> on host k8podoperatorxcomwritexcom.21384021df914227ad4e4b3a34313710 [2021-07-26 20:23:54,713] {taskinstance.py:1502} ERROR - Task failed with exception Traceback (most recent call last): File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1158, in _run_raw_task self._prepare_and_execute_task_with_callbacks(context, task) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1295, in _prepare_and_execute_task_with_callbacks self.render_templates(context=context) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1796, in render_templates self.task.render_template_fields(context) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 999, in render_template_fields self._do_render_template_fields(self, self.template_fields, context, jinja_env, set()) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1012, in _do_render_template_fields rendered_content = self.render_template(content, context, jinja_env, seen_oids) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in render_template return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in <listcomp> return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1047, in render_template return jinja_env.get_template(content).render(**context) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 883, in get_template return self._load_template(name, self.make_globals(globals)) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 857, in _load_template template = self.loader.load(self, name, globals) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 115, in load source, filename, uptodate = self.get_source(environment, name) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 197, in get_source raise TemplateNotFound(template) jinja2.exceptions.TemplateNotFound: mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json [2021-07-26 20:23:54,797] {taskinstance.py:1552} INFO - Marking task as FAILED. dag_id=k8_pod_operator_xcom, task_id=write-xcom, execution_date=20210726T202327, start_date=20210726T202353, end_date=20210726T202354 [2021-07-26 20:23:54,936] {local_task_job.py:153} INFO - Task exited with return code 1 ``` Dag: ``` from airflow import DAG from airflow.operators.bash import BashOperator from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import ( KubernetesPodOperator, ) from airflow.utils.dates import days_ago from airflow.configuration import conf namespace = conf.get("kubernetes", "NAMESPACE") # This will detect the default namespace locally and read the # environment namespace when deployed to Astronomer. if namespace == "default": config_file = "/usr/local/airflow/include/.kube/config" in_cluster = False else: in_cluster = True config_file = None default_args = { "owner": "airflow", } with DAG( dag_id="k8_pod_operator_xcom", default_args=default_args, schedule_interval=None, start_date=days_ago(2), tags=["k8"], ) as dag: write_xcom = KubernetesPodOperator( namespace=namespace, in_cluster=in_cluster, config_file=config_file, image="ubuntu", cmds=[ "sh", "-c", "mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json", ], name="write-xcom", do_xcom_push=True, is_delete_operator_pod=True, task_id="write-xcom", get_logs=True, ) pod_task_xcom_result = BashOperator( bash_command="echo \"{{ task_instance.xcom_pull('write-xcom')[0] }}\"", task_id="pod_task_xcom_result", ) write_xcom >> pod_task_xcom_result ``` closes apache/airflow#17186 GitOrigin-RevId: 73d2b720e0c79323a29741882a07eb8962256762
apache/airflow#17186 -- This has made the XCom functionality not work with KubernetesPodOperator, this has been fixed by apache/airflow#17760 -- so we should get this out sooner rather than later as recently released Airflow 2.1.3 will pull in latest kubernetes provider when we run `pip install -U apache-airflow[cncf.kubernetes]` GitOrigin-RevId: bb5602c652988d0b31ea5e0db8f03725a2f22d34
This commit effectively revers apache/airflow#15942 because of the issues it is causing. Until we find better solution we should revert this change Error: ``` [2021-07-26 20:23:54,109] {taskinstance.py:1108} INFO - Executing <Task(KubernetesPodOperator): write-xcom> on 2021-07-26T20:23:27.058907+00:00 [2021-07-26 20:23:54,113] {standard_task_runner.py:52} INFO - Started process 11 to run task [2021-07-26 20:23:54,297] {standard_task_runner.py:76} INFO - Running: ['airflow', 'tasks', 'run', 'k8_pod_operator_xcom', 'write-xcom', '2021-07-26T20:23:27.058907+00:00', '--job-id', '1757', '--pool', 'default_pool', '--raw', '--subdir', 'DAGS_FOLDER/k8s_xcom_example.py', '--cfg-path', '/tmp/tmp0q94pkhs', '--error-file', '/tmp/tmpoz9qqp2l'] [2021-07-26 20:23:54,298] {standard_task_runner.py:77} INFO - Job 1757: Subtask write-xcom [2021-07-26 20:23:54,511] {logging_mixin.py:104} INFO - Running <TaskInstance: k8_pod_operator_xcom.write-xcom 2021-07-26T20:23:27.058907+00:00 [running]> on host k8podoperatorxcomwritexcom.21384021df914227ad4e4b3a34313710 [2021-07-26 20:23:54,713] {taskinstance.py:1502} ERROR - Task failed with exception Traceback (most recent call last): File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1158, in _run_raw_task self._prepare_and_execute_task_with_callbacks(context, task) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1295, in _prepare_and_execute_task_with_callbacks self.render_templates(context=context) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1796, in render_templates self.task.render_template_fields(context) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 999, in render_template_fields self._do_render_template_fields(self, self.template_fields, context, jinja_env, set()) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1012, in _do_render_template_fields rendered_content = self.render_template(content, context, jinja_env, seen_oids) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in render_template return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in <listcomp> return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1047, in render_template return jinja_env.get_template(content).render(**context) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 883, in get_template return self._load_template(name, self.make_globals(globals)) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 857, in _load_template template = self.loader.load(self, name, globals) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 115, in load source, filename, uptodate = self.get_source(environment, name) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 197, in get_source raise TemplateNotFound(template) jinja2.exceptions.TemplateNotFound: mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json [2021-07-26 20:23:54,797] {taskinstance.py:1552} INFO - Marking task as FAILED. dag_id=k8_pod_operator_xcom, task_id=write-xcom, execution_date=20210726T202327, start_date=20210726T202353, end_date=20210726T202354 [2021-07-26 20:23:54,936] {local_task_job.py:153} INFO - Task exited with return code 1 ``` Dag: ``` from airflow import DAG from airflow.operators.bash import BashOperator from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import ( KubernetesPodOperator, ) from airflow.utils.dates import days_ago from airflow.configuration import conf namespace = conf.get("kubernetes", "NAMESPACE") # This will detect the default namespace locally and read the # environment namespace when deployed to Astronomer. if namespace == "default": config_file = "/usr/local/airflow/include/.kube/config" in_cluster = False else: in_cluster = True config_file = None default_args = { "owner": "airflow", } with DAG( dag_id="k8_pod_operator_xcom", default_args=default_args, schedule_interval=None, start_date=days_ago(2), tags=["k8"], ) as dag: write_xcom = KubernetesPodOperator( namespace=namespace, in_cluster=in_cluster, config_file=config_file, image="ubuntu", cmds=[ "sh", "-c", "mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json", ], name="write-xcom", do_xcom_push=True, is_delete_operator_pod=True, task_id="write-xcom", get_logs=True, ) pod_task_xcom_result = BashOperator( bash_command="echo \"{{ task_instance.xcom_pull('write-xcom')[0] }}\"", task_id="pod_task_xcom_result", ) write_xcom >> pod_task_xcom_result ``` closes apache/airflow#17186 GitOrigin-RevId: 73d2b720e0c79323a29741882a07eb8962256762
apache/airflow#17186 -- This has made the XCom functionality not work with KubernetesPodOperator, this has been fixed by apache/airflow#17760 -- so we should get this out sooner rather than later as recently released Airflow 2.1.3 will pull in latest kubernetes provider when we run `pip install -U apache-airflow[cncf.kubernetes]` GitOrigin-RevId: bb5602c652988d0b31ea5e0db8f03725a2f22d34
This commit effectively revers apache/airflow#15942 because of the issues it is causing. Until we find better solution we should revert this change Error: ``` [2021-07-26 20:23:54,109] {taskinstance.py:1108} INFO - Executing <Task(KubernetesPodOperator): write-xcom> on 2021-07-26T20:23:27.058907+00:00 [2021-07-26 20:23:54,113] {standard_task_runner.py:52} INFO - Started process 11 to run task [2021-07-26 20:23:54,297] {standard_task_runner.py:76} INFO - Running: ['airflow', 'tasks', 'run', 'k8_pod_operator_xcom', 'write-xcom', '2021-07-26T20:23:27.058907+00:00', '--job-id', '1757', '--pool', 'default_pool', '--raw', '--subdir', 'DAGS_FOLDER/k8s_xcom_example.py', '--cfg-path', '/tmp/tmp0q94pkhs', '--error-file', '/tmp/tmpoz9qqp2l'] [2021-07-26 20:23:54,298] {standard_task_runner.py:77} INFO - Job 1757: Subtask write-xcom [2021-07-26 20:23:54,511] {logging_mixin.py:104} INFO - Running <TaskInstance: k8_pod_operator_xcom.write-xcom 2021-07-26T20:23:27.058907+00:00 [running]> on host k8podoperatorxcomwritexcom.21384021df914227ad4e4b3a34313710 [2021-07-26 20:23:54,713] {taskinstance.py:1502} ERROR - Task failed with exception Traceback (most recent call last): File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1158, in _run_raw_task self._prepare_and_execute_task_with_callbacks(context, task) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1295, in _prepare_and_execute_task_with_callbacks self.render_templates(context=context) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1796, in render_templates self.task.render_template_fields(context) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 999, in render_template_fields self._do_render_template_fields(self, self.template_fields, context, jinja_env, set()) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1012, in _do_render_template_fields rendered_content = self.render_template(content, context, jinja_env, seen_oids) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in render_template return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in <listcomp> return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1047, in render_template return jinja_env.get_template(content).render(**context) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 883, in get_template return self._load_template(name, self.make_globals(globals)) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 857, in _load_template template = self.loader.load(self, name, globals) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 115, in load source, filename, uptodate = self.get_source(environment, name) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 197, in get_source raise TemplateNotFound(template) jinja2.exceptions.TemplateNotFound: mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json [2021-07-26 20:23:54,797] {taskinstance.py:1552} INFO - Marking task as FAILED. dag_id=k8_pod_operator_xcom, task_id=write-xcom, execution_date=20210726T202327, start_date=20210726T202353, end_date=20210726T202354 [2021-07-26 20:23:54,936] {local_task_job.py:153} INFO - Task exited with return code 1 ``` Dag: ``` from airflow import DAG from airflow.operators.bash import BashOperator from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import ( KubernetesPodOperator, ) from airflow.utils.dates import days_ago from airflow.configuration import conf namespace = conf.get("kubernetes", "NAMESPACE") # This will detect the default namespace locally and read the # environment namespace when deployed to Astronomer. if namespace == "default": config_file = "/usr/local/airflow/include/.kube/config" in_cluster = False else: in_cluster = True config_file = None default_args = { "owner": "airflow", } with DAG( dag_id="k8_pod_operator_xcom", default_args=default_args, schedule_interval=None, start_date=days_ago(2), tags=["k8"], ) as dag: write_xcom = KubernetesPodOperator( namespace=namespace, in_cluster=in_cluster, config_file=config_file, image="ubuntu", cmds=[ "sh", "-c", "mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json", ], name="write-xcom", do_xcom_push=True, is_delete_operator_pod=True, task_id="write-xcom", get_logs=True, ) pod_task_xcom_result = BashOperator( bash_command="echo \"{{ task_instance.xcom_pull('write-xcom')[0] }}\"", task_id="pod_task_xcom_result", ) write_xcom >> pod_task_xcom_result ``` closes apache/airflow#17186 GitOrigin-RevId: 73d2b720e0c79323a29741882a07eb8962256762
apache/airflow#17186 -- This has made the XCom functionality not work with KubernetesPodOperator, this has been fixed by apache/airflow#17760 -- so we should get this out sooner rather than later as recently released Airflow 2.1.3 will pull in latest kubernetes provider when we run `pip install -U apache-airflow[cncf.kubernetes]` GitOrigin-RevId: bb5602c652988d0b31ea5e0db8f03725a2f22d34
This commit effectively revers apache/airflow#15942 because of the issues it is causing. Until we find better solution we should revert this change Error: ``` [2021-07-26 20:23:54,109] {taskinstance.py:1108} INFO - Executing <Task(KubernetesPodOperator): write-xcom> on 2021-07-26T20:23:27.058907+00:00 [2021-07-26 20:23:54,113] {standard_task_runner.py:52} INFO - Started process 11 to run task [2021-07-26 20:23:54,297] {standard_task_runner.py:76} INFO - Running: ['airflow', 'tasks', 'run', 'k8_pod_operator_xcom', 'write-xcom', '2021-07-26T20:23:27.058907+00:00', '--job-id', '1757', '--pool', 'default_pool', '--raw', '--subdir', 'DAGS_FOLDER/k8s_xcom_example.py', '--cfg-path', '/tmp/tmp0q94pkhs', '--error-file', '/tmp/tmpoz9qqp2l'] [2021-07-26 20:23:54,298] {standard_task_runner.py:77} INFO - Job 1757: Subtask write-xcom [2021-07-26 20:23:54,511] {logging_mixin.py:104} INFO - Running <TaskInstance: k8_pod_operator_xcom.write-xcom 2021-07-26T20:23:27.058907+00:00 [running]> on host k8podoperatorxcomwritexcom.21384021df914227ad4e4b3a34313710 [2021-07-26 20:23:54,713] {taskinstance.py:1502} ERROR - Task failed with exception Traceback (most recent call last): File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1158, in _run_raw_task self._prepare_and_execute_task_with_callbacks(context, task) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1295, in _prepare_and_execute_task_with_callbacks self.render_templates(context=context) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1796, in render_templates self.task.render_template_fields(context) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 999, in render_template_fields self._do_render_template_fields(self, self.template_fields, context, jinja_env, set()) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1012, in _do_render_template_fields rendered_content = self.render_template(content, context, jinja_env, seen_oids) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in render_template return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in <listcomp> return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1047, in render_template return jinja_env.get_template(content).render(**context) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 883, in get_template return self._load_template(name, self.make_globals(globals)) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 857, in _load_template template = self.loader.load(self, name, globals) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 115, in load source, filename, uptodate = self.get_source(environment, name) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 197, in get_source raise TemplateNotFound(template) jinja2.exceptions.TemplateNotFound: mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json [2021-07-26 20:23:54,797] {taskinstance.py:1552} INFO - Marking task as FAILED. dag_id=k8_pod_operator_xcom, task_id=write-xcom, execution_date=20210726T202327, start_date=20210726T202353, end_date=20210726T202354 [2021-07-26 20:23:54,936] {local_task_job.py:153} INFO - Task exited with return code 1 ``` Dag: ``` from airflow import DAG from airflow.operators.bash import BashOperator from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import ( KubernetesPodOperator, ) from airflow.utils.dates import days_ago from airflow.configuration import conf namespace = conf.get("kubernetes", "NAMESPACE") # This will detect the default namespace locally and read the # environment namespace when deployed to Astronomer. if namespace == "default": config_file = "/usr/local/airflow/include/.kube/config" in_cluster = False else: in_cluster = True config_file = None default_args = { "owner": "airflow", } with DAG( dag_id="k8_pod_operator_xcom", default_args=default_args, schedule_interval=None, start_date=days_ago(2), tags=["k8"], ) as dag: write_xcom = KubernetesPodOperator( namespace=namespace, in_cluster=in_cluster, config_file=config_file, image="ubuntu", cmds=[ "sh", "-c", "mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json", ], name="write-xcom", do_xcom_push=True, is_delete_operator_pod=True, task_id="write-xcom", get_logs=True, ) pod_task_xcom_result = BashOperator( bash_command="echo \"{{ task_instance.xcom_pull('write-xcom')[0] }}\"", task_id="pod_task_xcom_result", ) write_xcom >> pod_task_xcom_result ``` closes apache/airflow#17186 GitOrigin-RevId: 73d2b720e0c79323a29741882a07eb8962256762
apache/airflow#17186 -- This has made the XCom functionality not work with KubernetesPodOperator, this has been fixed by apache/airflow#17760 -- so we should get this out sooner rather than later as recently released Airflow 2.1.3 will pull in latest kubernetes provider when we run `pip install -U apache-airflow[cncf.kubernetes]` GitOrigin-RevId: bb5602c652988d0b31ea5e0db8f03725a2f22d34
This commit effectively revers apache/airflow#15942 because of the issues it is causing. Until we find better solution we should revert this change Error: ``` [2021-07-26 20:23:54,109] {taskinstance.py:1108} INFO - Executing <Task(KubernetesPodOperator): write-xcom> on 2021-07-26T20:23:27.058907+00:00 [2021-07-26 20:23:54,113] {standard_task_runner.py:52} INFO - Started process 11 to run task [2021-07-26 20:23:54,297] {standard_task_runner.py:76} INFO - Running: ['airflow', 'tasks', 'run', 'k8_pod_operator_xcom', 'write-xcom', '2021-07-26T20:23:27.058907+00:00', '--job-id', '1757', '--pool', 'default_pool', '--raw', '--subdir', 'DAGS_FOLDER/k8s_xcom_example.py', '--cfg-path', '/tmp/tmp0q94pkhs', '--error-file', '/tmp/tmpoz9qqp2l'] [2021-07-26 20:23:54,298] {standard_task_runner.py:77} INFO - Job 1757: Subtask write-xcom [2021-07-26 20:23:54,511] {logging_mixin.py:104} INFO - Running <TaskInstance: k8_pod_operator_xcom.write-xcom 2021-07-26T20:23:27.058907+00:00 [running]> on host k8podoperatorxcomwritexcom.21384021df914227ad4e4b3a34313710 [2021-07-26 20:23:54,713] {taskinstance.py:1502} ERROR - Task failed with exception Traceback (most recent call last): File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1158, in _run_raw_task self._prepare_and_execute_task_with_callbacks(context, task) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1295, in _prepare_and_execute_task_with_callbacks self.render_templates(context=context) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1796, in render_templates self.task.render_template_fields(context) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 999, in render_template_fields self._do_render_template_fields(self, self.template_fields, context, jinja_env, set()) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1012, in _do_render_template_fields rendered_content = self.render_template(content, context, jinja_env, seen_oids) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in render_template return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in <listcomp> return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1047, in render_template return jinja_env.get_template(content).render(**context) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 883, in get_template return self._load_template(name, self.make_globals(globals)) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 857, in _load_template template = self.loader.load(self, name, globals) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 115, in load source, filename, uptodate = self.get_source(environment, name) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 197, in get_source raise TemplateNotFound(template) jinja2.exceptions.TemplateNotFound: mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json [2021-07-26 20:23:54,797] {taskinstance.py:1552} INFO - Marking task as FAILED. dag_id=k8_pod_operator_xcom, task_id=write-xcom, execution_date=20210726T202327, start_date=20210726T202353, end_date=20210726T202354 [2021-07-26 20:23:54,936] {local_task_job.py:153} INFO - Task exited with return code 1 ``` Dag: ``` from airflow import DAG from airflow.operators.bash import BashOperator from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import ( KubernetesPodOperator, ) from airflow.utils.dates import days_ago from airflow.configuration import conf namespace = conf.get("kubernetes", "NAMESPACE") # This will detect the default namespace locally and read the # environment namespace when deployed to Astronomer. if namespace == "default": config_file = "/usr/local/airflow/include/.kube/config" in_cluster = False else: in_cluster = True config_file = None default_args = { "owner": "airflow", } with DAG( dag_id="k8_pod_operator_xcom", default_args=default_args, schedule_interval=None, start_date=days_ago(2), tags=["k8"], ) as dag: write_xcom = KubernetesPodOperator( namespace=namespace, in_cluster=in_cluster, config_file=config_file, image="ubuntu", cmds=[ "sh", "-c", "mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json", ], name="write-xcom", do_xcom_push=True, is_delete_operator_pod=True, task_id="write-xcom", get_logs=True, ) pod_task_xcom_result = BashOperator( bash_command="echo \"{{ task_instance.xcom_pull('write-xcom')[0] }}\"", task_id="pod_task_xcom_result", ) write_xcom >> pod_task_xcom_result ``` closes apache/airflow#17186 GitOrigin-RevId: 73d2b720e0c79323a29741882a07eb8962256762
apache/airflow#17186 -- This has made the XCom functionality not work with KubernetesPodOperator, this has been fixed by apache/airflow#17760 -- so we should get this out sooner rather than later as recently released Airflow 2.1.3 will pull in latest kubernetes provider when we run `pip install -U apache-airflow[cncf.kubernetes]` GitOrigin-RevId: bb5602c652988d0b31ea5e0db8f03725a2f22d34
This commit effectively revers apache/airflow#15942 because of the issues it is causing. Until we find better solution we should revert this change Error: ``` [2021-07-26 20:23:54,109] {taskinstance.py:1108} INFO - Executing <Task(KubernetesPodOperator): write-xcom> on 2021-07-26T20:23:27.058907+00:00 [2021-07-26 20:23:54,113] {standard_task_runner.py:52} INFO - Started process 11 to run task [2021-07-26 20:23:54,297] {standard_task_runner.py:76} INFO - Running: ['airflow', 'tasks', 'run', 'k8_pod_operator_xcom', 'write-xcom', '2021-07-26T20:23:27.058907+00:00', '--job-id', '1757', '--pool', 'default_pool', '--raw', '--subdir', 'DAGS_FOLDER/k8s_xcom_example.py', '--cfg-path', '/tmp/tmp0q94pkhs', '--error-file', '/tmp/tmpoz9qqp2l'] [2021-07-26 20:23:54,298] {standard_task_runner.py:77} INFO - Job 1757: Subtask write-xcom [2021-07-26 20:23:54,511] {logging_mixin.py:104} INFO - Running <TaskInstance: k8_pod_operator_xcom.write-xcom 2021-07-26T20:23:27.058907+00:00 [running]> on host k8podoperatorxcomwritexcom.21384021df914227ad4e4b3a34313710 [2021-07-26 20:23:54,713] {taskinstance.py:1502} ERROR - Task failed with exception Traceback (most recent call last): File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1158, in _run_raw_task self._prepare_and_execute_task_with_callbacks(context, task) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1295, in _prepare_and_execute_task_with_callbacks self.render_templates(context=context) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1796, in render_templates self.task.render_template_fields(context) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 999, in render_template_fields self._do_render_template_fields(self, self.template_fields, context, jinja_env, set()) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1012, in _do_render_template_fields rendered_content = self.render_template(content, context, jinja_env, seen_oids) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in render_template return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in <listcomp> return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1047, in render_template return jinja_env.get_template(content).render(**context) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 883, in get_template return self._load_template(name, self.make_globals(globals)) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 857, in _load_template template = self.loader.load(self, name, globals) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 115, in load source, filename, uptodate = self.get_source(environment, name) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 197, in get_source raise TemplateNotFound(template) jinja2.exceptions.TemplateNotFound: mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json [2021-07-26 20:23:54,797] {taskinstance.py:1552} INFO - Marking task as FAILED. dag_id=k8_pod_operator_xcom, task_id=write-xcom, execution_date=20210726T202327, start_date=20210726T202353, end_date=20210726T202354 [2021-07-26 20:23:54,936] {local_task_job.py:153} INFO - Task exited with return code 1 ``` Dag: ``` from airflow import DAG from airflow.operators.bash import BashOperator from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import ( KubernetesPodOperator, ) from airflow.utils.dates import days_ago from airflow.configuration import conf namespace = conf.get("kubernetes", "NAMESPACE") # This will detect the default namespace locally and read the # environment namespace when deployed to Astronomer. if namespace == "default": config_file = "/usr/local/airflow/include/.kube/config" in_cluster = False else: in_cluster = True config_file = None default_args = { "owner": "airflow", } with DAG( dag_id="k8_pod_operator_xcom", default_args=default_args, schedule_interval=None, start_date=days_ago(2), tags=["k8"], ) as dag: write_xcom = KubernetesPodOperator( namespace=namespace, in_cluster=in_cluster, config_file=config_file, image="ubuntu", cmds=[ "sh", "-c", "mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json", ], name="write-xcom", do_xcom_push=True, is_delete_operator_pod=True, task_id="write-xcom", get_logs=True, ) pod_task_xcom_result = BashOperator( bash_command="echo \"{{ task_instance.xcom_pull('write-xcom')[0] }}\"", task_id="pod_task_xcom_result", ) write_xcom >> pod_task_xcom_result ``` closes apache/airflow#17186 GitOrigin-RevId: 73d2b720e0c79323a29741882a07eb8962256762
apache/airflow#17186 -- This has made the XCom functionality not work with KubernetesPodOperator, this has been fixed by apache/airflow#17760 -- so we should get this out sooner rather than later as recently released Airflow 2.1.3 will pull in latest kubernetes provider when we run `pip install -U apache-airflow[cncf.kubernetes]` GitOrigin-RevId: bb5602c652988d0b31ea5e0db8f03725a2f22d34
This commit effectively revers apache/airflow#15942 because of the issues it is causing. Until we find better solution we should revert this change Error: ``` [2021-07-26 20:23:54,109] {taskinstance.py:1108} INFO - Executing <Task(KubernetesPodOperator): write-xcom> on 2021-07-26T20:23:27.058907+00:00 [2021-07-26 20:23:54,113] {standard_task_runner.py:52} INFO - Started process 11 to run task [2021-07-26 20:23:54,297] {standard_task_runner.py:76} INFO - Running: ['airflow', 'tasks', 'run', 'k8_pod_operator_xcom', 'write-xcom', '2021-07-26T20:23:27.058907+00:00', '--job-id', '1757', '--pool', 'default_pool', '--raw', '--subdir', 'DAGS_FOLDER/k8s_xcom_example.py', '--cfg-path', '/tmp/tmp0q94pkhs', '--error-file', '/tmp/tmpoz9qqp2l'] [2021-07-26 20:23:54,298] {standard_task_runner.py:77} INFO - Job 1757: Subtask write-xcom [2021-07-26 20:23:54,511] {logging_mixin.py:104} INFO - Running <TaskInstance: k8_pod_operator_xcom.write-xcom 2021-07-26T20:23:27.058907+00:00 [running]> on host k8podoperatorxcomwritexcom.21384021df914227ad4e4b3a34313710 [2021-07-26 20:23:54,713] {taskinstance.py:1502} ERROR - Task failed with exception Traceback (most recent call last): File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1158, in _run_raw_task self._prepare_and_execute_task_with_callbacks(context, task) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1295, in _prepare_and_execute_task_with_callbacks self.render_templates(context=context) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1796, in render_templates self.task.render_template_fields(context) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 999, in render_template_fields self._do_render_template_fields(self, self.template_fields, context, jinja_env, set()) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1012, in _do_render_template_fields rendered_content = self.render_template(content, context, jinja_env, seen_oids) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in render_template return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in <listcomp> return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1047, in render_template return jinja_env.get_template(content).render(**context) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 883, in get_template return self._load_template(name, self.make_globals(globals)) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 857, in _load_template template = self.loader.load(self, name, globals) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 115, in load source, filename, uptodate = self.get_source(environment, name) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 197, in get_source raise TemplateNotFound(template) jinja2.exceptions.TemplateNotFound: mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json [2021-07-26 20:23:54,797] {taskinstance.py:1552} INFO - Marking task as FAILED. dag_id=k8_pod_operator_xcom, task_id=write-xcom, execution_date=20210726T202327, start_date=20210726T202353, end_date=20210726T202354 [2021-07-26 20:23:54,936] {local_task_job.py:153} INFO - Task exited with return code 1 ``` Dag: ``` from airflow import DAG from airflow.operators.bash import BashOperator from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import ( KubernetesPodOperator, ) from airflow.utils.dates import days_ago from airflow.configuration import conf namespace = conf.get("kubernetes", "NAMESPACE") # This will detect the default namespace locally and read the # environment namespace when deployed to Astronomer. if namespace == "default": config_file = "/usr/local/airflow/include/.kube/config" in_cluster = False else: in_cluster = True config_file = None default_args = { "owner": "airflow", } with DAG( dag_id="k8_pod_operator_xcom", default_args=default_args, schedule_interval=None, start_date=days_ago(2), tags=["k8"], ) as dag: write_xcom = KubernetesPodOperator( namespace=namespace, in_cluster=in_cluster, config_file=config_file, image="ubuntu", cmds=[ "sh", "-c", "mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json", ], name="write-xcom", do_xcom_push=True, is_delete_operator_pod=True, task_id="write-xcom", get_logs=True, ) pod_task_xcom_result = BashOperator( bash_command="echo \"{{ task_instance.xcom_pull('write-xcom')[0] }}\"", task_id="pod_task_xcom_result", ) write_xcom >> pod_task_xcom_result ``` closes apache/airflow#17186 GitOrigin-RevId: 73d2b720e0c79323a29741882a07eb8962256762
apache/airflow#17186 -- This has made the XCom functionality not work with KubernetesPodOperator, this has been fixed by apache/airflow#17760 -- so we should get this out sooner rather than later as recently released Airflow 2.1.3 will pull in latest kubernetes provider when we run `pip install -U apache-airflow[cncf.kubernetes]` GitOrigin-RevId: bb5602c652988d0b31ea5e0db8f03725a2f22d34
This commit effectively revers apache/airflow#15942 because of the issues it is causing. Until we find better solution we should revert this change Error: ``` [2021-07-26 20:23:54,109] {taskinstance.py:1108} INFO - Executing <Task(KubernetesPodOperator): write-xcom> on 2021-07-26T20:23:27.058907+00:00 [2021-07-26 20:23:54,113] {standard_task_runner.py:52} INFO - Started process 11 to run task [2021-07-26 20:23:54,297] {standard_task_runner.py:76} INFO - Running: ['airflow', 'tasks', 'run', 'k8_pod_operator_xcom', 'write-xcom', '2021-07-26T20:23:27.058907+00:00', '--job-id', '1757', '--pool', 'default_pool', '--raw', '--subdir', 'DAGS_FOLDER/k8s_xcom_example.py', '--cfg-path', '/tmp/tmp0q94pkhs', '--error-file', '/tmp/tmpoz9qqp2l'] [2021-07-26 20:23:54,298] {standard_task_runner.py:77} INFO - Job 1757: Subtask write-xcom [2021-07-26 20:23:54,511] {logging_mixin.py:104} INFO - Running <TaskInstance: k8_pod_operator_xcom.write-xcom 2021-07-26T20:23:27.058907+00:00 [running]> on host k8podoperatorxcomwritexcom.21384021df914227ad4e4b3a34313710 [2021-07-26 20:23:54,713] {taskinstance.py:1502} ERROR - Task failed with exception Traceback (most recent call last): File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1158, in _run_raw_task self._prepare_and_execute_task_with_callbacks(context, task) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1295, in _prepare_and_execute_task_with_callbacks self.render_templates(context=context) File "/usr/local/lib/python3.7/site-packages/airflow/models/taskinstance.py", line 1796, in render_templates self.task.render_template_fields(context) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 999, in render_template_fields self._do_render_template_fields(self, self.template_fields, context, jinja_env, set()) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1012, in _do_render_template_fields rendered_content = self.render_template(content, context, jinja_env, seen_oids) File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in render_template return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1063, in <listcomp> return [self.render_template(element, context, jinja_env) for element in content] File "/usr/local/lib/python3.7/site-packages/airflow/models/baseoperator.py", line 1047, in render_template return jinja_env.get_template(content).render(**context) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 883, in get_template return self._load_template(name, self.make_globals(globals)) File "/usr/local/lib/python3.7/site-packages/jinja2/environment.py", line 857, in _load_template template = self.loader.load(self, name, globals) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 115, in load source, filename, uptodate = self.get_source(environment, name) File "/usr/local/lib/python3.7/site-packages/jinja2/loaders.py", line 197, in get_source raise TemplateNotFound(template) jinja2.exceptions.TemplateNotFound: mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json [2021-07-26 20:23:54,797] {taskinstance.py:1552} INFO - Marking task as FAILED. dag_id=k8_pod_operator_xcom, task_id=write-xcom, execution_date=20210726T202327, start_date=20210726T202353, end_date=20210726T202354 [2021-07-26 20:23:54,936] {local_task_job.py:153} INFO - Task exited with return code 1 ``` Dag: ``` from airflow import DAG from airflow.operators.bash import BashOperator from airflow.providers.cncf.kubernetes.operators.kubernetes_pod import ( KubernetesPodOperator, ) from airflow.utils.dates import days_ago from airflow.configuration import conf namespace = conf.get("kubernetes", "NAMESPACE") # This will detect the default namespace locally and read the # environment namespace when deployed to Astronomer. if namespace == "default": config_file = "/usr/local/airflow/include/.kube/config" in_cluster = False else: in_cluster = True config_file = None default_args = { "owner": "airflow", } with DAG( dag_id="k8_pod_operator_xcom", default_args=default_args, schedule_interval=None, start_date=days_ago(2), tags=["k8"], ) as dag: write_xcom = KubernetesPodOperator( namespace=namespace, in_cluster=in_cluster, config_file=config_file, image="ubuntu", cmds=[ "sh", "-c", "mkdir -p /airflow/xcom/;echo '[1,2,3,4]' > /airflow/xcom/return.json", ], name="write-xcom", do_xcom_push=True, is_delete_operator_pod=True, task_id="write-xcom", get_logs=True, ) pod_task_xcom_result = BashOperator( bash_command="echo \"{{ task_instance.xcom_pull('write-xcom')[0] }}\"", task_id="pod_task_xcom_result", ) write_xcom >> pod_task_xcom_result ``` closes apache/airflow#17186 GitOrigin-RevId: 73d2b720e0c79323a29741882a07eb8962256762
apache/airflow#17186 -- This has made the XCom functionality not work with KubernetesPodOperator, this has been fixed by apache/airflow#17760 -- so we should get this out sooner rather than later as recently released Airflow 2.1.3 will pull in latest kubernetes provider when we run `pip install -U apache-airflow[cncf.kubernetes]` GitOrigin-RevId: bb5602c652988d0b31ea5e0db8f03725a2f22d34
The "template_ext" mechanism is useful for automatically loading and jinja-processing files which are specified in parameters of Operators. However this might lead to certain problems for example (from slack conversation):
I believe the behaviour changed comparing to not-so-long-past. Some of the changes with processing parameters with jinja recursively caused this template behaviour to be applied also nested parameters like above.
There were also several discussions and user confusion with this behaviour: #15942, #16922
There are two ways we could improve the situation:
limit the "template_extension" resolution to only direct string kwargs passed to the operator (I think this is no brainer and we should do it)
propose some "escaping" mechanism, where you could either disable template_extension processing entirely or somehow mark the parameters that should not be treated as templates.
Here I have several proposals:
a) we could add "skip_template_ext_processing" or similar parameter in BaseOperator <- I do not like it as many operators rely on this behaviour for a good reason
b) we could add "template_ext" parameter in the operator that could override the original class-level-field <- I like this one a lot
c) we could add "disable_template_ext_pattern" (str) parameter where we could specify list of regexp's where we could filter out only specific values <- this one will allow to disable template_ext much more "selectively" - only for certain parameters.
UPDATE: It only affects Kubenernetes Pod Operator due to it's recursive behaviour and should be fixed there.
The text was updated successfully, but these errors were encountered: