External triggers / cron support /

[mistercrunch]

@ctrebing, thanks for getting this started, I'm adding on to the work you've done as this will be a large changeset:

• externally triggered DAGs
• cron expression support in schedule_interval as in 0 0 0 * *
• allow for multiple concurrent scheduler (might be another PR, we'll see)

[mahrz24]

I wonder whether it would be helpful to create a DagRun with state QUEUED (or a new state SCHEDULED) and only set it to running once the first scheduler created/queued the first TaskInstance. For example, if no scheduler is running, the DagRun will be displayed as running in the UI and one has to see if the corresponding task instances are actually created to find the problem why nothing happens.

[mistercrunch]

closes #570

[mistercrunch]

closes #586

[pedrorjbr]

I registered a new dag. This dag will be external triggered, so I dont need to schedule it. How can I make the scheduler ignore it (nto schedule it and just run when i trigger it).

[mistercrunch]

@pedrorjbr : schedule_interval=None
http://pythonhosted.org/airflow/scheduler.html#dag-runs

[pedrorjbr]

What is the correct way (if exists one) to trigger a dag when a file is uploaded on AWS S3, for example?
Is it better to create a scheduled job that makes pooling on S3?

I could start a dag with a S3Sensor, and makes the task wait_for_downstream=true.
So the task will be scheduled again just when the downstream terminates. So I will process the file just once.

[brycedrennan]

@mistercrunch
I think this code doesn't work. Shouldn't it be:

    if isinstance(delta, six.string_types):
        cron = croniter(cron_frequency, dt + timedelta(minutes=1))
        return cron.get_prev(datetime)

I used this doctest to test for this case:

    >>> round_time(datetime(2015, 9, 12, 6, 0), '0 0 * * *', datetime(2015, 9, 2, 0, 0))
    datetime.datetime(2015, 9, 12, 0, 0)

I'm unclear what the intent was behind that if statement at the end.