title: Tasks added to DAG during runtime fail to be scheduled
body: <p>My idea is to have a task <code>foo</code> which generates a list of inputs (users, reports, log files, etc), and a task is launched for every element in the input list. The goal is to make use of Airflow's retrying and other logic, instead of reimplementing it.</p>
  
  <p>So, ideally, my DAG should look something like this:
  <a href="https://i.stack.imgur.com/e0NQk.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/e0NQk.png" alt="enter image description here"></a></p>
  
  <p>The only variable here is the number of tasks generated. I want to do some more tasks after all of these are completed, so spinning up a new DAG for every task does not seem appropriate.</p>
  
  <p>This is my code:</p>
  
  <pre><code>default_args = {
      'owner': 'airflow',
      'depends_on_past': False,
      'start_date': datetime(2015, 6, 1)
  }
  
  dag = DAG('dynamic_dag_generator', schedule_interval=None, default_args=default_args)
  
  foo_operator = BashOperator(
      task_id='foo',
      bash_command="echo '%s'" % json.dumps(range(0, random.randint(40,60))),
      xcom_push=True,
      dag=dag)
  
  def gen_nodes(**kwargs):
      ti = kwargs['ti']
      workers = json.loads(ti.xcom_pull(task_ids='foo'))
  
      for wid in workers:
          print("Iterating worker %s" % wid)
          op = PythonOperator(
              task_id='test_op_%s' % wid,
              python_callable=lambda: print("Dynamic task!"),
              dag=dag
          )
  
          op.set_downstream(bar_operator)
          op.set_upstream(dummy_op)
  
  gen_subdag_node_op = PythonOperator(
      task_id='gen_subdag_nodes',
      python_callable=gen_nodes,
      provide_context=True,
      dag=dag
  )
  
  gen_subdag_node_op.set_upstream(foo_operator)
  
  dummy_op = DummyOperator(
      task_id='dummy',
      dag=dag
  )
  
  dummy_op.set_upstream(gen_subdag_node_op)
  
  bar_operator = DummyOperator(
      task_id='bar',
      dag=dag)
  
  bar_operator.set_upstream(dummy_op)
  </code></pre>
  
  <p>In the logs, I can see that <code>gen_nodes</code> is executed correctly (i.e. <code>Iterating worker 5</code>, etc). However, the new tasks are not scheduled and there is no evidence that they were executed.</p>
  
  <p>I found related code samples online, <a href="https://gist.github.com/mtustin-handy/ecd80c1cc9dcad1c4cf7" rel="nofollow noreferrer">such as this</a>, but could not make it work. Am I missing something?</p>
  
  <p>Alternatively, is there a more appropriate approach to this problem (isolating units of work)?</p>