[AIRFLOW-3892] Create Redis pub sub sensor #4712

mans2singh · 2019-02-14T22:36:12Z

Make sure you have checked all steps below.

Jira

My PR addresses the following Airflow Jira issues and references them in the PR title. For example, "[AIRFLOW-XXX] My Airflow PR"
- https://issues.apache.org/jira/browse/AIRFLOW-3892
- In case you are fixing a typo in the documentation you can prepend your commit with [AIRFLOW-XXX], code changes always need a Jira issue.

Description

Here are some details about my PR, including screenshots of any UI changes: Create sensor for reading messages from Redis.

Tests

My PR adds the following unit tests OR does not need testing for this extremely good reason:
- tests/contrib/sensors/test_redis_pub_sub_sensor.py

Commits

My commits all reference Jira issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "How to write a good git commit message":
1. Subject is separated from body by a blank line
2. Subject is limited to 50 characters (not including Jira issue reference)
3. Subject does not end with a period
4. Subject uses the imperative mood ("add", not "adding")
5. Body wraps at 72 characters
6. Body explains "what" and "why", not "how"

Documentation

In case of new functionality, my PR adds documentation that describes how to use it.
- When adding new operators/hooks/sensors, the autoclass documentation generation needs to be added.
- All the public functions and the classes in the PR contain docstrings that explain what it does

Code Quality

Passes flake8

zhongjiajie

Left some review here.

zhongjiajie · 2019-02-15T01:14:10Z

tests/contrib/sensors/test_redis_pub_sub_sensor.py

+from mock import patch
+from mock import call
+
+from unittest.mock import MagicMock


@mans2singh FYI, CI failed due to from unittest.mock import MagicMock. Maybe you could do

Suggested change

from unittest.mock import MagicMock

from mock import MagicMock

and than optimize imports

zhongjiajie · 2019-02-15T01:18:13Z

airflow/contrib/sensors/redis_pub_sub_sensor.py

+        :param context: the context object
+        :type context: dict
+        :return: ``True`` if message (with type 'message') is available or ``False`` if not
+        :rtype: bool


Maybe we could remove rtype

Suggested change

:rtype: bool

Removed from doc.

zhongjiajie · 2019-02-15T01:27:33Z

airflow/contrib/sensors/redis_pub_sub_sensor.py

+    ui_color = '#f0eee4'
+
+    @apply_defaults
+    def __init__(self, channels, hook, *args, **kwargs):


I think we should use redis_conn_id instand of hook. we already have redis_hook in airflow.contrib.hook.redis_hook we could use it.

Suggested change

def __init__(self, channels, hook, *args, **kwargs):

def __init__(self, channels, redis_conn_id, *args, **kwargs):

than we could create connection using

self.hook = RedisHook(redis_conn_id=self.redis_conn_id)

Hi @zhongjiajie - I was trying the decouple the responsibility of creating the hook from using it by the sensor . This will allow injection of the hook and use mock hooks for unit testing.
Let me know your thoughts.

@mans2singh I think we should use conn str rather than hook as parameter. airflow provide function to create hook by conn str.

@zhongjiajie -
I have seen that pattern of passing a connection id and the sensor/operator creating the hook and using it.
I was trying to use dependency injection pattern to make it easy to inject a mock hook and test the sensor.
Is there any disadvantage of using this pattern ?
Thanks for your advice.

@mans2singh disadvantage is we have to write code to create hook in DAG file. I think not high cohesion

If we pass conn_id str, user could just pass conn_id str to active sensors. For unittest I think we could do like other sensors.

Changed to use redis_conn_id as recommended.

zhongjiajie · 2019-02-15T01:29:51Z

tests/contrib/sensors/test_redis_pub_sub_sensor.py

+        configuration.load_test_config()
+
+        self.log = logging.getLogger()
+        self.log.setLevel(logging.DEBUG)


I think we could do unittest without log

Suggested change

self.log.setLevel(logging.DEBUG)

mans2singh · 2019-02-16T19:30:18Z

@zhongjiajie -
I've updated the code based on your recommendations. Please let me know if you have any additional advice.
Thanks.

codecov-io · 2019-02-16T20:15:59Z

Codecov Report

Merging #4712 into master will increase coverage by 0.01%.
The diff coverage is 100%.

@@            Coverage Diff             @@
##           master    #4712      +/-   ##
==========================================
+ Coverage   74.73%   74.74%   +0.01%     
==========================================
  Files         449      450       +1     
  Lines       28961    28982      +21     
==========================================
+ Hits        21643    21664      +21     
  Misses       7318     7318

Impacted Files	Coverage Δ
airflow/contrib/sensors/redis_pub_sub_sensor.py	`100% <100%> (ø)`

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update dda309e...55182eb. Read the comment docs.

zhongjiajie

Please find my two cents

zhongjiajie · 2019-02-17T06:22:10Z

airflow/contrib/sensors/redis_pub_sub_sensor.py

+        super(RedisPubSubSensor, self).__init__(*args, **kwargs)
+        self.channels = channels
+        self.redis_conn_id = redis_conn_id
+        self.pubsub = RedisHook(redis_conn_id=self.redis_conn_id).get_conn().pubsub()


Should we move L47 and L48 into function poke before message = self.pubsub.get_message()? discuss

I placed these lines in the __init__ function so that they are executed once and poke, which can be called multiple times, can just check the message.

@mans2singh Ok, I get your point. maybe we should ask advice from maintainers about it. you can ask their reivew now

zhongjiajie · 2019-02-17T06:22:17Z

airflow/contrib/sensors/redis_pub_sub_sensor.py

+        self.log.info('RedisPubSubSensor checking for message on channels: %s', self.channels)
+
+        message = self.pubsub.get_message()
+        self.log.info('Message %s from channel %s', message, self.channels)


In L66 check is message is None, but log message here. will this get Message None from channel %s or not? And L68 you log message and channel again. Is L63 necessary?

There is some redundancy in the logs but it is easier to review the logs the see that the data/channel when the message is received. I can remove it if required.

@mans2singh Maybe we should discuss with maintainers. You could ask their advice while we finish this review.

@zhongjiajie - I've removed the log as per your recommendation. Thanks for your review.

zhongjiajie · 2019-02-17T06:22:22Z

tests/contrib/sensors/test_redis_pub_sub_sensor.py

+from airflow.utils import timezone
+from mock import patch
+from mock import call
+from mock import MagicMock


should we use single line

Suggested change

from mock import MagicMock

from mock import MagicMock patch call

zhongjiajie · 2019-02-18T01:37:58Z

LGTM, but we still have a discuss in #4712 (comment)

mans2singh · 2019-02-18T02:35:07Z

@feng-tao - Can you please advice who can review this pull request and give me your feedback ?

Fokko · 2019-02-19T15:54:12Z

tests/contrib/sensors/test_redis_pub_sub_sensor.py

+
+class TestRedisPubSubSensor(unittest.TestCase):
+
+    @patch('airflow.contrib.hooks.redis_hook.RedisHook.get_conn')


In the CI actually runs a Redis using docker-compose. Maybe you can do end to end testing instead of mocking. This makes it easier when you want to test against newer versions of Redis. What do you think?

@Fokko -

I see that the redis key sensor uses an integration test and will try your recommendation.

Thanks.

feng-tao · 2019-02-19T20:27:31Z

we have a celery queue sensor(https://github.com/apache/airflow/blob/master/airflow/contrib/sensors/celery_queue_sensor.py)which could be used to wait wait for celery queue back by Redis. Could you share more on your sensor use case?

mans2singh · 2019-02-20T23:17:22Z

Hi @feng-tao

The use case I have is to drive EMR workflow and its configuration based on events.

In my workflow the time at which data is available and data size vary. Using static cluster config results in over/under provisioning of EMR cluster. Scheduling the workflow late in day when I am sure the data is available, leads to higher latency in processing. By using the information in the event, I can trigger the workflow with the configuration of cluster based on contents of event. Airflow already has Redis hook and a key sensor. This sensor adds events based triggering/configuration capabilities to it.

The celery_queue_sensor, from my understanding, is for task scheduling, but let me know if I have misunderstood it's functionality.

Please let me know your thoughts.

mans2singh · 2019-02-23T21:55:11Z

@feng-tao @Fokko -

I've added integration tests based on your advice. Let me know if you have any other recommendations.

Thanks

mans2singh · 2019-03-02T03:08:59Z

@feng-tao

Please let me know if you have advice for me. I can also create a simple example dag to print the events to demonstrate its usage.

Thanks

Mans

feng-tao · 2019-03-05T21:34:02Z

@mans2singh sorry for the late review. Overall lgtm. We recently upgrade redis client(#4834). Could you rebase your pr with master and do a sanity test and see if it still works as expected?

Use redis_conn_id, removed logging from test, corrected unittest import, removed rtype from poke doc

mans2singh · 2019-03-06T00:53:20Z

Hi @feng-tao -

I've rebased the code and the integration unit/tests for redis sensor are passing locally.
It looks like two kubernetes tests are failing for python 3.5 on travis-ci.

Can you please let me know how to resolve this and if there is anything else is required ?

Thanks for your advice and help.

Mans

zhongjiajie · 2019-03-06T01:07:22Z

@mans2singh The CI currently flaky, If you could not find detail error, maybe you could submit and restart CI test again.

mans2singh · 2019-03-06T04:50:50Z

@zhongjiajie -
I kicked off another build and it is passing but this page still shows failed build. Please let me know what I could be missing.
Thanks

feng-tao · 2019-03-06T05:04:08Z

@mans2singh , I think we are fine. But could we remove that example dag as I don't think we need that given the sensor is not complicated? Thanks. Once you remove, I think we are good to go.

mans2singh · 2019-03-06T11:43:53Z

@feng-tao -

I've removed the example per your recommendation. Please let me know if there is anything else required.

Thanks for your comments/advice.

Mans

feng-tao · 2019-03-06T17:47:17Z

thanks @mans2singh , merged

* [AIRFLOW-3892] Create Redis pub sub sensor * [AIRFLOW-3892] - Updated based on review comments (#4712) Use redis_conn_id, removed logging from test, corrected unittest import, removed rtype from poke doc * [AIRFLOW-3892] Combined import based on review comments * [AIRFLOW-3892] Removed extra log based on review comments * [AIRFLOW-3892] - Added integration tests based on review comments * [AIRFLOW-3892] - Added example dag to kick off the build * [AIRFLOW-3892] - Removed example dag based on review comments

* [AIRFLOW-3892] Create Redis pub sub sensor * [AIRFLOW-3892] - Updated based on review comments (apache#4712) Use redis_conn_id, removed logging from test, corrected unittest import, removed rtype from poke doc * [AIRFLOW-3892] Combined import based on review comments * [AIRFLOW-3892] Removed extra log based on review comments * [AIRFLOW-3892] - Added integration tests based on review comments * [AIRFLOW-3892] - Added example dag to kick off the build * [AIRFLOW-3892] - Removed example dag based on review comments

zhongjiajie reviewed Feb 15, 2019

View reviewed changes

zhongjiajie reviewed Feb 17, 2019

View reviewed changes

Fokko reviewed Feb 19, 2019

View reviewed changes

mans2singh added 5 commits March 5, 2019 18:49

[AIRFLOW-3892] Create Redis pub sub sensor

499ce2f

[AIRFLOW-3892] - Updated based on review comments (#4712)

56b903b

Use redis_conn_id, removed logging from test, corrected unittest import, removed rtype from poke doc

[AIRFLOW-3892] Combined import based on review comments

d51e98a

[AIRFLOW-3892] Removed extra log based on review comments

98f96bb

[AIRFLOW-3892] - Added integration tests based on review comments

bdc9636

[AIRFLOW-3892] - Added example dag to kick off the build

6aff292

[AIRFLOW-3892] - Removed example dag based on review comments

55182eb

feng-tao merged commit afa06ae into apache:master Mar 6, 2019

	from unittest.mock import MagicMock
	from mock import MagicMock

	def __init__(self, channels, hook, args, *kwargs):
	def __init__(self, channels, redis_conn_id, args, *kwargs):

	from mock import MagicMock
	from mock import MagicMock patch call


		class TestRedisPubSubSensor(unittest.TestCase):

		@patch('airflow.contrib.hooks.redis_hook.RedisHook.get_conn')

[AIRFLOW-3892] Create Redis pub sub sensor #4712

[AIRFLOW-3892] Create Redis pub sub sensor #4712

Conversation

mans2singh commented Feb 14, 2019

Jira

Description

Tests

Commits

Documentation

Code Quality

zhongjiajie left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mans2singh Feb 16, 2019 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mans2singh commented Feb 16, 2019

codecov-io commented Feb 16, 2019 • edited

Codecov Report

zhongjiajie left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zhongjiajie commented Feb 18, 2019

mans2singh commented Feb 18, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

feng-tao commented Feb 19, 2019

mans2singh commented Feb 20, 2019 • edited

mans2singh commented Feb 23, 2019

mans2singh commented Mar 2, 2019

feng-tao commented Mar 5, 2019

mans2singh commented Mar 6, 2019

zhongjiajie commented Mar 6, 2019

mans2singh commented Mar 6, 2019

feng-tao commented Mar 6, 2019

mans2singh commented Mar 6, 2019

feng-tao commented Mar 6, 2019

mans2singh Feb 16, 2019 •

edited

codecov-io commented Feb 16, 2019 •

edited

mans2singh commented Feb 20, 2019 •

edited