You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by cr...@apache.org on 2017/04/10 21:28:16 UTC
[1/4] incubator-airflow git commit: [AIRFLOW-111] Include queued
tasks in scheduler concurrency check
Repository: incubator-airflow
Updated Branches:
refs/heads/v1-8-test 4199cd3d2 -> a9e0894ba
[AIRFLOW-111] Include queued tasks in scheduler concurrency check
The concurrency argument in dags appears to not be obeyed because the
scheduler does not check the concurrency properly when checking tasks.
The tasks do not run, but this leads to a lot of scheduler churn.
(cherry picked from commit 31fce01251957812b6aa392dcd70bb4519305e2a)
Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/aec97708
Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/aec97708
Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/aec97708
Branch: refs/heads/v1-8-test
Commit: aec9770831d69ca04569c9af8da51942261be4ca
Parents: 4199cd3
Author: Alex Guziel <al...@airbnb.com>
Authored: Mon Apr 3 14:10:21 2017 -0700
Committer: Chris Riccomini <cr...@apache.org>
Committed: Mon Apr 10 14:12:48 2017 -0700
----------------------------------------------------------------------
----------------------------------------------------------------------
[2/4] incubator-airflow git commit: [AIRFLOW-1001] Fix landing times
if there is no following schedule
Posted by cr...@apache.org.
[AIRFLOW-1001] Fix landing times if there is no following schedule
@once does not have a following schedule. This was
not checked for
and therefore the landing times page could bork.
Closes #2213 from bolkedebruin/AIRFLOW-1001
(cherry picked from commit 0371df4f1bd78e220e591d5cb23630d6a062f109)
Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/c94b3a02
Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/c94b3a02
Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/c94b3a02
Branch: refs/heads/v1-8-test
Commit: c94b3a02f430f1a5a86c83d5f7286dcdac31492b
Parents: aec9770
Author: Bolke de Bruin <bo...@xs4all.nl>
Authored: Wed Apr 5 09:57:55 2017 +0200
Committer: Chris Riccomini <cr...@apache.org>
Committed: Mon Apr 10 14:19:53 2017 -0700
----------------------------------------------------------------------
airflow/www/views.py | 2 +-
tests/core.py | 16 ++++++++++++++--
2 files changed, 15 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/c94b3a02/airflow/www/views.py
----------------------------------------------------------------------
diff --git a/airflow/www/views.py b/airflow/www/views.py
index 962c1f0..fec4779 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -1553,7 +1553,7 @@ class Airflow(BaseView):
for ti in task.get_task_instances(session, start_date=min_date,
end_date=base_date):
ts = ti.execution_date
- if dag.schedule_interval:
+ if dag.schedule_interval and dag.following_schedule(ts):
ts = dag.following_schedule(ts)
if ti.end_date:
dttm = wwwutils.epoch(ti.execution_date)
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/c94b3a02/tests/core.py
----------------------------------------------------------------------
diff --git a/tests/core.py b/tests/core.py
index 870a0cb..c55b1e2 100644
--- a/tests/core.py
+++ b/tests/core.py
@@ -1413,6 +1413,7 @@ class WebUiTests(unittest.TestCase):
self.dag_bash2 = self.dagbag.dags['test_example_bash_operator']
self.sub_dag = self.dagbag.dags['example_subdag_operator']
self.runme_0 = self.dag_bash.get_task('runme_0')
+ self.example_xcom = self.dagbag.dags['example_xcom']
self.dag_bash2.create_dagrun(
run_id="test_{}".format(models.DagRun.id_for_date(datetime.now())),
@@ -1428,6 +1429,13 @@ class WebUiTests(unittest.TestCase):
state=State.RUNNING
)
+ self.example_xcom.create_dagrun(
+ run_id="test_{}".format(models.DagRun.id_for_date(datetime.now())),
+ execution_date=DEFAULT_DATE,
+ start_date=datetime.now(),
+ state=State.RUNNING
+ )
+
def test_index(self):
response = self.app.get('/', follow_redirects=True)
assert "DAGs" in response.data.decode('utf-8')
@@ -1473,8 +1481,12 @@ class WebUiTests(unittest.TestCase):
assert "example_bash_operator" in response.data.decode('utf-8')
response = self.app.get(
'/admin/airflow/landing_times?'
- 'days=30&dag_id=example_bash_operator')
- assert "example_bash_operator" in response.data.decode('utf-8')
+ 'days=30&dag_id=test_example_bash_operator')
+ assert "test_example_bash_operator" in response.data.decode('utf-8')
+ response = self.app.get(
+ '/admin/airflow/landing_times?'
+ 'days=30&dag_id=example_xcom')
+ assert "example_xcom" in response.data.decode('utf-8')
response = self.app.get(
'/admin/airflow/gantt?dag_id=example_bash_operator')
assert "example_bash_operator" in response.data.decode('utf-8')
[3/4] incubator-airflow git commit: [AIRFLOW-1004][AIRFLOW-276] Fix
`airflow webserver -D` to run in background
Posted by cr...@apache.org.
[AIRFLOW-1004][AIRFLOW-276] Fix `airflow webserver -D` to run in background
AIRFLOW-276 introduced a monitor process for gunicorn
to find new files in the dag folder, but it also changed
`airflow webserver -D`'s behavior to run in foreground.
This PR fixes that by running the monitor as a daemon
process.
Closes #2208 from sekikn/AIRFLOW-1004
(cherry picked from commit a9b20a04b052e9479dbb79fd46124293085610e9)
Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/0a5fb785
Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/0a5fb785
Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/0a5fb785
Branch: refs/heads/v1-8-test
Commit: 0a5fb7856b545073516210fcfc369d2072823ae9
Parents: c94b3a0
Author: Kengo Seki <se...@apache.org>
Authored: Tue Apr 4 08:32:44 2017 +0200
Committer: Chris Riccomini <cr...@apache.org>
Committed: Mon Apr 10 14:24:31 2017 -0700
----------------------------------------------------------------------
airflow/bin/cli.py | 64 ++++++++++++++++++++++++++++++++++++++++---------
tests/core.py | 54 +++++++++++++++++++++++++++++++++++++++++
2 files changed, 107 insertions(+), 11 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/0a5fb785/airflow/bin/cli.py
----------------------------------------------------------------------
diff --git a/airflow/bin/cli.py b/airflow/bin/cli.py
index e9c54e6..e4755c7 100755
--- a/airflow/bin/cli.py
+++ b/airflow/bin/cli.py
@@ -753,7 +753,12 @@ def webserver(args):
app.run(debug=True, port=args.port, host=args.hostname,
ssl_context=(ssl_cert, ssl_key) if ssl_cert and ssl_key else None)
else:
- pid, stdout, stderr, log_file = setup_locations("webserver", pid=args.pid)
+ pid, stdout, stderr, log_file = setup_locations("webserver", args.pid, args.stdout, args.stderr, args.log_file)
+ if args.daemon:
+ handle = setup_logging(log_file)
+ stdout = open(stdout, 'w+')
+ stderr = open(stderr, 'w+')
+
print(
textwrap.dedent('''\
Running the Gunicorn Server with:
@@ -771,7 +776,6 @@ def webserver(args):
'-t', str(worker_timeout),
'-b', args.hostname + ':' + str(args.port),
'-n', 'airflow-webserver',
- '-p', str(pid),
'-c', 'airflow.www.gunicorn_config'
]
@@ -782,28 +786,66 @@ def webserver(args):
run_args += ['--error-logfile', str(args.error_logfile)]
if args.daemon:
- run_args += ["-D"]
+ run_args += ['-D', '-p', str(pid)]
+
if ssl_cert:
run_args += ['--certfile', ssl_cert, '--keyfile', ssl_key]
run_args += ["airflow.www.app:cached_app()"]
- gunicorn_master_proc = subprocess.Popen(run_args)
+ gunicorn_master_proc = None
def kill_proc(dummy_signum, dummy_frame):
gunicorn_master_proc.terminate()
gunicorn_master_proc.wait()
sys.exit(0)
- signal.signal(signal.SIGINT, kill_proc)
- signal.signal(signal.SIGTERM, kill_proc)
+ def monitor_gunicorn(gunicorn_master_proc):
+ # These run forever until SIG{INT, TERM, KILL, ...} signal is sent
+ if conf.getint('webserver', 'worker_refresh_interval') > 0:
+ restart_workers(gunicorn_master_proc, num_workers)
+ else:
+ while True:
+ time.sleep(1)
- # These run forever until SIG{INT, TERM, KILL, ...} signal is sent
- if conf.getint('webserver', 'worker_refresh_interval') > 0:
- restart_workers(gunicorn_master_proc, num_workers)
+ if args.daemon:
+ base, ext = os.path.splitext(pid)
+ ctx = daemon.DaemonContext(
+ pidfile=TimeoutPIDLockFile(base + "-monitor" + ext, -1),
+ files_preserve=[handle],
+ stdout=stdout,
+ stderr=stderr,
+ signal_map={
+ signal.SIGINT: kill_proc,
+ signal.SIGTERM: kill_proc
+ },
+ )
+ with ctx:
+ subprocess.Popen(run_args)
+
+ # Reading pid file directly, since Popen#pid doesn't
+ # seem to return the right value with DaemonContext.
+ while True:
+ try:
+ with open(pid) as f:
+ gunicorn_master_proc_pid = int(f.read())
+ break
+ except IOError:
+ logging.debug("Waiting for gunicorn's pid file to be created.")
+ time.sleep(0.1)
+
+ gunicorn_master_proc = psutil.Process(gunicorn_master_proc_pid)
+ monitor_gunicorn(gunicorn_master_proc)
+
+ stdout.close()
+ stderr.close()
else:
- while True:
- time.sleep(1)
+ gunicorn_master_proc = subprocess.Popen(run_args)
+
+ signal.signal(signal.SIGINT, kill_proc)
+ signal.signal(signal.SIGTERM, kill_proc)
+
+ monitor_gunicorn(gunicorn_master_proc)
def scheduler(args):
http://git-wip-us.apache.org/repos/asf/incubator-airflow/blob/0a5fb785/tests/core.py
----------------------------------------------------------------------
diff --git a/tests/core.py b/tests/core.py
index c55b1e2..4fd2f08 100644
--- a/tests/core.py
+++ b/tests/core.py
@@ -1398,6 +1398,60 @@ class CliTests(unittest.TestCase):
os.remove('variables1.json')
os.remove('variables2.json')
+ def test_cli_webserver_foreground(self):
+ import subprocess
+
+ # Confirm that webserver hasn't been launched.
+ # pgrep returns exit status 1 if no process matched.
+ self.assertEqual(1, subprocess.Popen(["pgrep", "-c", "airflow"]).wait())
+ self.assertEqual(1, subprocess.Popen(["pgrep", "-c", "gunicorn"]).wait())
+
+ # Run webserver in foreground and terminate it.
+ p = subprocess.Popen(["airflow", "webserver"])
+ p.terminate()
+ p.wait()
+
+ # Assert that no process remains.
+ self.assertEqual(1, subprocess.Popen(["pgrep", "-c", "airflow"]).wait())
+ self.assertEqual(1, subprocess.Popen(["pgrep", "-c", "gunicorn"]).wait())
+
+ @unittest.skipIf("TRAVIS" in os.environ and bool(os.environ["TRAVIS"]),
+ "Skipping test due to lack of required file permission")
+ def test_cli_webserver_background(self):
+ import subprocess
+ import psutil
+
+ def wait_pidfile(pidfile):
+ while True:
+ try:
+ with open(pidfile) as f:
+ return int(f.read())
+ except IOError:
+ sleep(1)
+
+ # Confirm that webserver hasn't been launched.
+ self.assertEqual(1, subprocess.Popen(["pgrep", "-c", "airflow"]).wait())
+ self.assertEqual(1, subprocess.Popen(["pgrep", "-c", "gunicorn"]).wait())
+
+ # Run webserver in background.
+ subprocess.Popen(["airflow", "webserver", "-D"])
+ pidfile = cli.setup_locations("webserver")[0]
+ wait_pidfile(pidfile)
+
+ # Assert that gunicorn and its monitor are launched.
+ self.assertEqual(0, subprocess.Popen(["pgrep", "-c", "airflow"]).wait())
+ self.assertEqual(0, subprocess.Popen(["pgrep", "-c", "gunicorn"]).wait())
+
+ # Terminate monitor process.
+ pidfile = cli.setup_locations("webserver-monitor")[0]
+ pid = wait_pidfile(pidfile)
+ p = psutil.Process(pid)
+ p.terminate()
+ p.wait()
+
+ # Assert that no process remains.
+ self.assertEqual(1, subprocess.Popen(["pgrep", "-c", "airflow"]).wait())
+ self.assertEqual(1, subprocess.Popen(["pgrep", "-c", "gunicorn"]).wait())
class WebUiTests(unittest.TestCase):
def setUp(self):
[4/4] incubator-airflow git commit: [AIRFLOW-1030][AIRFLOW-1] Fix
hook import for HttpSensor
Posted by cr...@apache.org.
[AIRFLOW-1030][AIRFLOW-1] Fix hook import for HttpSensor
Closes #2180 from
pdambrauskas/fix/http_hook_import
(cherry picked from commit f2dae7d15623e2534e7c0dab3b5a7e02d4cff81d)
Project: http://git-wip-us.apache.org/repos/asf/incubator-airflow/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-airflow/commit/a9e0894b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-airflow/tree/a9e0894b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-airflow/diff/a9e0894b
Branch: refs/heads/v1-8-test
Commit: a9e0894ba0113cf62c7e9006fb0b42085bc5e9f9
Parents: 0a5fb78
Author: pdambrauskas <p....@gmail.com>
Authored: Tue Apr 4 08:39:54 2017 +0200
Committer: Chris Riccomini <cr...@apache.org>
Committed: Mon Apr 10 14:25:43 2017 -0700
----------------------------------------------------------------------
----------------------------------------------------------------------