<p dir="auto"><strong>Description</strong></p>
<p dir="auto">When using decorators to define dags, e.g. dag_1.py:</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="from airflow.decorators import dag, task
from airflow.utils.dates import days_ago

DEFAULT_ARGS = {
    &quot;owner&quot;: &quot;airflow&quot;,
}


@task
def some_task():
    pass


@dag(
    default_args=DEFAULT_ARGS,
    schedule_interval=None,
    start_date=days_ago(2),
)
def my_dag():
    some_task()


DAG_1 = my_dag()
"><pre class="notranslate"><span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">decorators</span> <span class="pl-k">import</span> <span class="pl-s1">dag</span>, <span class="pl-s1">task</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">utils</span>.<span class="pl-s1">dates</span> <span class="pl-k">import</span> <span class="pl-s1">days_ago</span>

<span class="pl-v">DEFAULT_ARGS</span> <span class="pl-c1">=</span> {
    <span class="pl-s">"owner"</span>: <span class="pl-s">"airflow"</span>,
}


<span class="pl-en">@<span class="pl-s1">task</span></span>
<span class="pl-k">def</span> <span class="pl-en">some_task</span>():
    <span class="pl-k">pass</span>


<span class="pl-en">@<span class="pl-en">dag</span>(</span>
<span class="pl-en">    <span class="pl-s1">default_args</span><span class="pl-c1">=</span><span class="pl-v">DEFAULT_ARGS</span>,</span>
<span class="pl-en">    <span class="pl-s1">schedule_interval</span><span class="pl-c1">=</span><span class="pl-c1">None</span>,</span>
<span class="pl-en">    <span class="pl-s1">start_date</span><span class="pl-c1">=</span><span class="pl-en">days_ago</span>(<span class="pl-c1">2</span>),</span>
<span class="pl-en">)</span>
<span class="pl-k">def</span> <span class="pl-en">my_dag</span>():
    <span class="pl-en">some_task</span>()


<span class="pl-v">DAG_1</span> <span class="pl-c1">=</span> <span class="pl-en">my_dag</span>()</pre></div>
<p dir="auto">and</p>
<p dir="auto">dag_2.py:</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="from airflow.decorators import dag, task
from airflow.utils.dates import days_ago

DEFAULT_ARGS = {
    &quot;owner&quot;: &quot;airflow&quot;,
}


@task
def some_other_task():
    pass


@dag(
    default_args=DEFAULT_ARGS,
    schedule_interval=None,
    start_date=days_ago(2),
)
def my_dag():
    some_other_task()


DAG_2 = my_dag()
"><pre class="notranslate"><span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">decorators</span> <span class="pl-k">import</span> <span class="pl-s1">dag</span>, <span class="pl-s1">task</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">utils</span>.<span class="pl-s1">dates</span> <span class="pl-k">import</span> <span class="pl-s1">days_ago</span>

<span class="pl-v">DEFAULT_ARGS</span> <span class="pl-c1">=</span> {
    <span class="pl-s">"owner"</span>: <span class="pl-s">"airflow"</span>,
}


<span class="pl-en">@<span class="pl-s1">task</span></span>
<span class="pl-k">def</span> <span class="pl-en">some_other_task</span>():
    <span class="pl-k">pass</span>


<span class="pl-en">@<span class="pl-en">dag</span>(</span>
<span class="pl-en">    <span class="pl-s1">default_args</span><span class="pl-c1">=</span><span class="pl-v">DEFAULT_ARGS</span>,</span>
<span class="pl-en">    <span class="pl-s1">schedule_interval</span><span class="pl-c1">=</span><span class="pl-c1">None</span>,</span>
<span class="pl-en">    <span class="pl-s1">start_date</span><span class="pl-c1">=</span><span class="pl-en">days_ago</span>(<span class="pl-c1">2</span>),</span>
<span class="pl-en">)</span>
<span class="pl-k">def</span> <span class="pl-en">my_dag</span>():
    <span class="pl-en">some_other_task</span>()


<span class="pl-v">DAG_2</span> <span class="pl-c1">=</span> <span class="pl-en">my_dag</span>()</pre></div>
<p dir="auto">We have two different dags which have been written in isolation, but by sheer bad luck both define <code class="notranslate">my_dag()</code>. This seems fine for each file in isolation, but on the airflow UI, we only end up seeing one entry for <code class="notranslate">my_dag</code>, where it has picked up <code class="notranslate">dag_1.py</code> and ignored <code class="notranslate">dag_2.py</code>.</p>
<p dir="auto"><strong>Use case / motivation</strong></p>
<p dir="auto">We currently end up with only one DAG showing up on the UI, and no indication as to why the other one hasn't appeared.</p>
<p dir="auto">Suggestion: popup similar to 'DAG import error' to highlight what needs changing in one of the DAG files in order for both to show up ("DAG import error: duplicate dag names found - please review {duplicate files} and ensure all dag definitions are unique"?)</p>
<p dir="auto"><strong>Are you willing to submit a PR?</strong></p>
<p dir="auto">No time to spare on this at present</p>
<p dir="auto"><strong>Related Issues</strong></p>
<p dir="auto">I haven't found any related issues with the search function.</p>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.1.4</p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Debian</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto">apache-airflow==2.1.4<br>
apache-airflow-providers-amazon==2.2.0<br>
apache-airflow-providers-celery==2.0.0<br>
apache-airflow-providers-cncf-kubernetes==2.0.2<br>
apache-airflow-providers-docker==2.1.1<br>
apache-airflow-providers-elasticsearch==2.0.3<br>
apache-airflow-providers-ftp==2.0.1<br>
apache-airflow-providers-google==5.1.0<br>
apache-airflow-providers-grpc==2.0.1<br>
apache-airflow-providers-hashicorp==2.1.0<br>
apache-airflow-providers-http==2.0.1<br>
apache-airflow-providers-imap==2.0.1<br>
apache-airflow-providers-microsoft-azure==3.1.1<br>
apache-airflow-providers-mysql==2.1.1<br>
apache-airflow-providers-postgres==2.2.0<br>
apache-airflow-providers-redis==2.0.1<br>
apache-airflow-providers-sendgrid==2.0.1<br>
apache-airflow-providers-sftp==2.1.1<br>
apache-airflow-providers-slack==4.0.1<br>
apache-airflow-providers-sqlite==2.0.1<br>
apache-airflow-providers-ssh==2.1.1</p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Official Apache Airflow Helm Chart</p>
<h3 dir="auto">Deployment details</h3>
<ul dir="auto">
<li>Kubernetes 1.17 (managed)</li>
<li>kubectl 1.19</li>
<li>Airflow 2.1.4</li>
<li>Airflow Helm chart 8.5.2 (Community)</li>
<li>Custom registry to pull <code class="notranslate">alpine</code> from, configured through <code class="notranslate">airflow_local_settings.py</code> provided as k8s Secret which is configured as <code class="notranslate">extraVolumeMounts</code>in <code class="notranslate">values.yaml</code> for helm chart</li>
<li>KubernetesExecutor with <code class="notranslate">AIRFLOW__KUBERNETES__WORKER_CONTAINER_REPOSITORY</code> set</li>
</ul>
<h3 dir="auto">What happened</h3>
<p dir="auto">In the DAG the tasks are using <code class="notranslate">KubernetesPodOperator</code> like the following:</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="test1 = KubernetesPodOperator(
    task_id = 'test1',
    name = 'tester_test1',
    namespace = 'airflow',
    image = 'custom path to image',
    image_pull_policy = 'IfNotPresent',
    image_pull_secrets = [k8s.V1LocalObjectReference('...')],
    do_xcom_push = True,
    get_logs = True,
    in_cluster = True,
    is_delete_operator_pod = True,
    init_containers = [init_container],
    dag = dag
)"><pre class="notranslate"><span class="pl-s1">test1</span> <span class="pl-c1">=</span> <span class="pl-v">KubernetesPodOperator</span>(
    <span class="pl-s1">task_id</span> <span class="pl-c1">=</span> <span class="pl-s">'test1'</span>,
    <span class="pl-s1">name</span> <span class="pl-c1">=</span> <span class="pl-s">'tester_test1'</span>,
    <span class="pl-s1">namespace</span> <span class="pl-c1">=</span> <span class="pl-s">'airflow'</span>,
    <span class="pl-s1">image</span> <span class="pl-c1">=</span> <span class="pl-s">'custom path to image'</span>,
    <span class="pl-s1">image_pull_policy</span> <span class="pl-c1">=</span> <span class="pl-s">'IfNotPresent'</span>,
    <span class="pl-s1">image_pull_secrets</span> <span class="pl-c1">=</span> [<span class="pl-s1">k8s</span>.<span class="pl-v">V1LocalObjectReference</span>(<span class="pl-s">'...'</span>)],
    <span class="pl-s1">do_xcom_push</span> <span class="pl-c1">=</span> <span class="pl-c1">True</span>,
    <span class="pl-s1">get_logs</span> <span class="pl-c1">=</span> <span class="pl-c1">True</span>,
    <span class="pl-s1">in_cluster</span> <span class="pl-c1">=</span> <span class="pl-c1">True</span>,
    <span class="pl-s1">is_delete_operator_pod</span> <span class="pl-c1">=</span> <span class="pl-c1">True</span>,
    <span class="pl-s1">init_containers</span> <span class="pl-c1">=</span> [<span class="pl-s1">init_container</span>],
    <span class="pl-s1">dag</span> <span class="pl-c1">=</span> <span class="pl-s1">dag</span>
)</pre></div>
<p dir="auto">When triggering the DAG and starting a task the output of <code class="notranslate">kubectl describe pod &lt;worker pod&gt; -n airflow</code> is like:</p>
<blockquote>
<p dir="auto">Failed to pull image "alpine": rpc error: code = Unknown desc = Error response from daemon: toomanyrequests: You have reached your pull rate limit. You may increase the limit by authenticating and upgrading: <a href="https://www.docker.com/increase-rate-limit" rel="nofollow">https://www.docker.com/increase-rate-limit</a></p>
</blockquote>
<h3 dir="auto">What you expected to happen</h3>
<p dir="auto">When triggering the DAG and starting a task the expected output of <code class="notranslate">kubectl describe pod &lt;worker pod&gt; -n airflow</code> is like:</p>
<blockquote>
<p dir="auto">Successfully pulled image "alpine"</p>
</blockquote>
<p dir="auto">This problem is well known and common solution is setting <code class="notranslate">airflow_local_settings.py</code> as described in the <a href="https://artifacthub.io/packages/helm/airflow-helm/airflow#how-to-set-airflow-configs" rel="nofollow">Airflow Helm chart (Community) FAQ</a>.</p>
<p dir="auto">As far as I understand, this is related to the <code class="notranslate">KubernetesExecutor</code>, because it is not possible to configure the executor to inject a configuration into the used image for the worker Pod.</p>
<p dir="auto">It is possible to configure the <code class="notranslate">KubernetesExecutor</code> to use a custom image from a custom registry by setting <code class="notranslate">AIRFLOW__KUBERNETES__WORKER_CONTAINER_REPOSITORY</code> and <code class="notranslate">AIRFLOW__KUBERNETES__WORKER_CONTAINER_TAG</code>. But it is not possible to build a custom Airflow 2.1.4 image which is already configured to use the custom registry for pulling <code class="notranslate">alpine</code> from.</p>
<h3 dir="auto">How to reproduce</h3>
<ol dir="auto">
<li>Setup Airflow 2.1.4 on Kubernetes using Helm chart 8.5.2 (community)</li>
<li>Configure <code class="notranslate">airflow_local_settings.py</code> as described in Helm chart 8.5.2 (community) FAQ</li>
<li>Use <code class="notranslate">KubernetesPodOperator</code> with <code class="notranslate">do_xcom_push=True</code></li>
<li>Trigger DAG that launch the task</li>
</ol>
<h3 dir="auto">Anything else</h3>
<p dir="auto">This problem occures every time the Docker pull limit is hit.</p>
<p dir="auto"><code class="notranslate">kubectl describe pod tester-test1 -n airflow</code> returns at end of output:</p>
<p dir="auto"><strong>Events:</strong></p>
<table role="table">
<thead>
<tr>
<th>Type</th>
<th>Reason</th>
<th>Age</th>
<th>From</th>
<th>Message</th>
</tr>
</thead>
<tbody>
<tr>
<td>Normal</td>
<td>Scheduled</td>
<td>21s</td>
<td>default-scheduler</td>
<td>Successfully assigned tester-test1</td>
</tr>
<tr>
<td>Normal</td>
<td>Pulled</td>
<td>20s</td>
<td>kubelet</td>
<td>Container image "custom/repo/alpine:3.14.0" already present on machine</td>
</tr>
<tr>
<td>Normal</td>
<td>Created</td>
<td>20s</td>
<td>kubelet</td>
<td>Created container init-container</td>
</tr>
<tr>
<td>Normal</td>
<td>Started</td>
<td>19s</td>
<td>kubelet</td>
<td>Started container init-container</td>
</tr>
<tr>
<td>Normal</td>
<td>Pulled</td>
<td>19s</td>
<td>kubelet</td>
<td>Container image "custom/repo/xcom_test:latest" already present on machine</td>
</tr>
<tr>
<td>Normal</td>
<td>Created</td>
<td>19s</td>
<td>kubelet</td>
<td>Created container base</td>
</tr>
<tr>
<td>Normal</td>
<td>Started</td>
<td>19s</td>
<td>kubelet</td>
<td>Started container base</td>
</tr>
<tr>
<td>Normal</td>
<td>BackOff</td>
<td>16s (x2 over 17s)</td>
<td>kubelet</td>
<td>Back-off pulling image "alpine"</td>
</tr>
<tr>
<td>Warning</td>
<td>Failed</td>
<td>16s (x2 over 17s)</td>
<td>kubelet</td>
<td>Error: ImagePullBackOff</td>
</tr>
<tr>
<td>Normal</td>
<td>Pulling</td>
<td>4s (x2 over 19s)</td>
<td>kubelet</td>
<td>Pulling image "alpine"</td>
</tr>
<tr>
<td>Warning</td>
<td>Failed</td>
<td>3s (x2 over 17s)</td>
<td>kubelet</td>
<td>Failed to pull image "alpine": rpc error: code = Unknown desc = Error response from daemon: toomanyrequests: You have reached your pull rate limit. You may increase the limit by authenticating and upgrading: <a href="https://www.docker.com/increase-rate-limit" rel="nofollow">https://www.docker.com/increase-rate-limit</a></td>
</tr>
<tr>
<td>Warning</td>
<td>Failed</td>
<td>3s (x2 over 17s)</td>
<td>kubelet</td>
<td>Error: ErrImagePull</td>
</tr>
</tbody>
</table>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>