Mailing-List: contact dev-help@airflow.incubator.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@airflow.incubator.apache.org
Date: Mon, 17 Oct 2016 10:24:45 +0000 (UTC)
From: Michal K <postrational@yahoo.com.INVALID>
Reply-To: Michal K <postrational@yahoo.com>
To: "dev@airflow.incubator.apache.org" <dev@airflow.incubator.apache.org>
Message-ID: <467686170.1052419.1476699885895@mail.yahoo.com>
Subject: Suspend and resume Sensors
MIME-Version: 1.0
Content-Type: multipart/alternative;
	boundary="----=_Part_1052418_1557389063.1476699885890"
References: <467686170.1052419.1476699885895.ref@mail.yahoo.com>
archived-at: Mon, 17 Oct 2016 10:30:01 -0000

------=_Part_1052418_1557389063.1476699885890
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: quoted-printable

Hello Everyone,
our company is currently considering Airflow for the engine of a pretty big=
 workflow system.We use Python 3 and have some HA requirements, so we plan =
to use Celery with multiple worker and scheduler nodes and RabbitMQ as mess=
age bus. We are also using Docker containers for deployment.
We have a specific case, which we need to handle in our workflows. Some tas=
ks, which we trigger are very long lived (many hours long). These long task=
s are carried out by external systems, independent of the Airflow nodes. In=
 our workflow need to start, then monitor these external tasks and trigger =
downstream operators based on their results.
This seems like a perfect job for Airflow sensors. Here however, we run int=
o a small problem.=C2=A0In our testing it seems not possible to stop an Air=
flow worker while a sensor is running. The sensor continues to run during a=
 Celery warm shutdown and it's background process remains active even after=
 a cold shutdown. This prevents us from being able to safely destroy the wo=
rker's Docker container.
Since at least one of the external tasks we need to monitor is practically =
always running, we would not be able to stop our Airflow workers in order t=
o deploy code changes to our DAGs or the libraries they depend on.
It would be perfect for our case to have the ability to suspend an Airflow =
sensor while shutting down the worker and resume it when the worker restart=
s.
I was wondering how this could be implemented in Airflow and I came up with=
 this initial idea:https://github.com/postrational/incubator-airflow/commit=
/74ac6f7290d6838362ff437e228465bb49fe198f
The code adds a signal handler to the BaseSensorOperator, which raises an e=
xception if it detects that the worker is shutting down (SIGINT detected).
Later on it handles the exception in the 'run' method of the TaskInstance. =
The idea is to put the sensor in a state which would cause it to be cleanly=
 resumed after Airflow comes back up after a restart.
So far, my draft code works some of the time, but not always. Sometimes the=
 sensor resumes correctly, but sometimes it doesn't trigger its downstream =
operator and the whole DAG run is marked as "failed".
I noticed the following line in the sensor's logs:[2016-10-13 19:15:17,033]=
 {jobs.py:1976} WARNING - State of this instance has been externally set to=
 None. Taking the poison pill. So long.
I would like to ask you how to properly mark an operator to be restarted cl=
eanly.
Before I proceed any further I would also like to ask for your opinion of t=
his approach.=C2=A0How do you handle long running sensors?=C2=A0Do you use =
Docker with Airflow?
Any advice would be greatly appreciated. Thanks,
Michal
------=_Part_1052418_1557389063.1476699885890--