Page History

Status

State: Draft

Discussion thread:

JIRA: AIRFLOW-2221

Motivation

Currently Airflow requires dag files to be present on a file system that is accessible to the scheduler, webserver, and worker/s. An increasing number of Airflow implementations are done on distributed cloud frameworks, which adds an increased level of difficult for guaranteeing a file system that is accessible and synced amongst services. By allowing Airflow the ability to fetch dag files from a remote source outside the file system local to the service, this grant a much greater flexibility, eases implementation, and standardizes ways to sync remote sources of dags with Airflow.

Outline

The following outline is based upon the work already done in the PR: https://github.com/apache/incubator-airflow/pull/3138

Dag Manifest:

Currently Airflow assumes valid

DAGs are any Dag object that it finds in the python files in the $AIRFLOW_HOME/dags directory, the system is not the most efficient or the most desirable

Format

The Dag Manifest would be composed of manifest entries an entry would be defined as follows:

entry:
	uri: where dag can be found
	version: which version of the dag to get
	conn_id: connection id to use to interact with remote location

Generation

The manifest would be generated by a callable supplied in the config that when called would generate a list of entries i.e.

[core]
# callable to fetch dag manifest list
dag_list = my_config.get_dag_list

Backwards Compatibility Compatibility

To maintain backwards compatibility to the default for dag_list wold be to simply crawl the dags folder and then produce the manifest that way

Benefits 

• With the manifest people are able to more explicitly note which dags should be looked at for by Airflow
• Airflow no longer has to crawl through a directory importing various files possibly causing problems
• Users are not forced to allow for a way to crawl various remote sources
• Allowing listing the connection id makes it easy to have multiple remote dag locations

DAG URIs:

Dag DAG locations will be given via URI, i.e. s3://my-bucket/dag1.zip, local:////dags/day1.zip

Versioning:

Versions per dag in the dag manifest will increase control and consistency of dag execution. Versions should be implemented in such away that they can not be used by the user such that we maintain backwards compatibility

• • Adding a version attribute to the DagRun model will allow for the same dag to be executed through out the DagRun even if a new version is changed, previously this was an issue with dag update mid dag run
  • The easiest way to implement versioning would be a naming convention within URIs
  • Need to answer the questions of whether Versions are mutable or immutable?
  • How are version conventions forced upon the user?
  • Will versioning be something that is exposed to the end user when it comes to running commands via the cli? Or will we only do this for completing DagRuns tied to an older version
  • Should versions be a part of the URI that is given by the user i.e. s3://my-bucket/dag1-v0.0.0.zip or do you simply specify s3://my-bucket/dag1.zip and v0.0.0 this will be mostly important for how we pass around uris and are able to resolve versions, i.e. if I have a uri for a newer version but I want to get an older version

Caching:

DAG URI is also DAG version.

• • We load the DAG from the same URI throughout the entire DAG run even if the DAG manifest was changed to a new DAG URI. We will add a URI attribute to the DagRun model to persist the URI used for each DAG run.
    
  • Users are free to define their own URI naming convention.
  • Version is immutable.

Caching:

Moving DAGs to a remote location will introduce network overhead so we should cache DAGs and avoid unnecessary fetch. We should only re-fetch a DAG when the URI in the DAG manifest is changed.Moving dags to a remote location will create an overhead when an Airflow service has to fetch a dag, it will be necessary to devise a method to cache dags to avoid fetchs every time

Cache Invalidation

There are various ways to perform cache invalidation:

1. 1. Timestamp comparison: 

• • • Airflow currently does this to see if a file needs to be processed again. In this case we would keep track of the time stamp for the dag object when it was last fetched, and invalidate when the remote location has a newer timestamp
    • This would require that all remote dag locations have a timestamps that accurately reflect the last time the dag object was changed

1. 1. Hash comparions:

• • • This is how image repos tend to work where a hash of the object is created, if the cache hash is kept track of then we invalidate when they differ from the remote hash
    • This requires a way to get the remote hash from the remote location

1. 1. Immutable Versions:

• • • If we implement a versioning systems and then maintain the convention that versions are immutable then we assume that a cached version of a dag is valid until the version of the dag changed on the manifest
    • This leads to very simple checks for invalidating caches, but this puts the onerous of keeping versions immutable on the user uploading to their remote location
    • This also requires we switch to using manifest (I think we should make this switch)

Cache Location

In order to avoid making a remote fetch every time the dag needs to be run it will be best to keep a local cache of dag files for individual Airflow services to use

• Where should this cache location be? Is this something that can be specified by the user?
• Will the cache be a per service case/ or do we want to look into distributed caching?
• How will cacheing work in a framework like Kubernetes? 
• With the Kubernetes Executor where pods dynamically spawn and then die so they will always miss the cached dags, requiring the user to have a shared file system also removes one of the prime gains of using remote dag fetching

Dag Fetching

DagBag Changes

The main part of the code base that we'll need to be changed is the DagBag, with previous PR, the over architecture was to refactor the DagBag to import a separately defined DagFetcher class that would define a process_file and fetch method. If we assume we will be switching the a dag manifest that is formed by a callable then the fetch method will instead be defined there. As well if we assume that dags will be fetched and stored locally then the process_file method can remain the same as it already handles zip logic, instead we can simply run a wrapper function process_entry which checks cache validity and determines whether the dag at a given URI needs to be fetched, and then calls process_file on the file path. This fetch method would be the only thing that would need to be implemented for each remote source.

def fetch_dag(uri, version, conn_id, local_cache_path):
	"""
	This method given a uri and version will download the dag
	from the remote location to the local_cache_path. Conn_id
	is supplied to fetch credentials
	"""
	
	raise NotImplementedError()

The proccess_entry would look like the following

def process_entry(self, entry, only_if_updated=True, safe_mode=True):
    """
    Given a path to a python module or zip file, this method imports
    the module and look for dag objects within it.
    """
	cache_path = self.get_cache_path(entry.uri, entry.version)
	if not self.dag_in_cache(entry):
		fetch_dag(entry.uri, entry.version, entry.conn_id, cache_path)
	
	self.process_file(cache_path, only_if_update=only_if_update)

The DagBag will also need a way to server a specific version of a dag. Currently I think it would be best of get_dag always serves the latest version (this will depend either on having a consistent naming convention, or the uri in the manifest being versioned), but if we want to have the scheduler be able to get older versions for older dag runs. One way to do this would be to add an optional version argument to get_dag

def get_dag(self, dag_id, version=None):
	if version is not None:
		# serve specific version of dag
	else:
		# serve latest/default version i.e. versions aren't being used

Scheduler Changes

Currently the scheduler checks what dags are on disk by calling list_py_file_paths this will need to be changed to instead look at the manifest as we can no longer crawl the file system, and instead crawl manifest entries

Versioning

If we implement versioning of dags it will require a number of changes to the current scheduler. The biggest issue comes from how the scheduler currently propagates the dag object to it's various function calls for task scheduling. As is the scheduler loads in the dag objects that are found in the filesystem, and these passed along to the resulting functions. In order to implement versions we would need to associate a certain dag version to a DagRun, when previous DagRuns are fetched we'll need to check if they were for an earlier dag version and fetch that version if necessary.

Here is the exact part of _process_task_instances where this check would need to happen

def _process_task_instances(self, dag, queue, session=None):
    """
    This method schedules the tasks for a single DAG by looking at the
    active DAG runs and adding task instances that should run to the
    queue.
    """

    # update the state of the previously active dag runs
    dag_runs = DagRun.find(dag_id=dag.dag_id, state=State.RUNNING, session=session)
    active_dag_runs = []
    for run in dag_runs:
        self.log.info("Examining DAG run %s", run)
        # don't consider runs that are executed in the future
        if run.execution_date > timezone.utcnow():
            self.log.error(
                "Execution date is in future: %s",
                run.execution_date
            )
            continue

        if len(active_dag_runs) >= dag.max_active_runs:
            self.log.info("Active dag runs > max_active_run.")
            continue

        # skip backfill dagruns for now as long as they are not really scheduled
        if run.is_backfill:
            continue

        # todo: run.dag is transient but needs to be set
        run.dag = dag

In this case the dag object will be the object loaded in from the current version listed in the manifest, this last line here should check to pin run.dag to be equal to the dag version of the DagRun

DAG Model

The main thing we'll have to make sure stays consistent about the dag model is the the fileloc attribute points to a file location on disk accessible to the webserver, one way to do this is to set fileloc to be the local cache location, and that when getting a dag from the DagBag we ensure the cache file is available on disk