Setup Celery worker with Redis broker #70

jraddaoui · 2018-07-26T01:22:59Z

Add Celery and Redis requirements.
Add Celery config in settings.
Add celery app and load it on init.
Use Redis as Broker and django_celery_results as backend.
Move METS extraction and parsing to a shared task.
Improvements in METS parsing process.
Redirect to collection page on DIP creation.
Reflect when a DIP is being imported:
- Add import_status and ìmport_task_id columns to DIPs model and
  ES doc. Reflect status on related DigitalFiles.
- Set it to PENDING when the METS handling process is launched and
  to FAILURE or SUCCESS in the task after_return call.
- For editors and admins:
  - Show spinning icon instead of links for DIPs and DigitalFiles when
    the import is PENDING.
  - Show error message in view pages.
- For other users:
  - Do not show DIPs or DigitalFiles with an import status of
    FAILURE or PENDING.
- Add flash messages system to show user notifications on import
  launch and on import failure.
Create dips.models.TaskResult proxy model of django_celery_results
to format error message.
Move some functions to helpers and add tests.
Add tests for task and after_return calls
Set CELERY_BROKER_URL env. var. in Tox test environment.
Improve docker-compose environment:
- Remove ES network and use default network.
- Set CELERY_BROKER_URL env. var.
- Add redis and worker services.
- Add redis_data volume.

TODO:

Connects to #15.

codecov · 2018-07-26T01:27:21Z

Codecov Report

Merging #70 into master will increase coverage by 2.04%.
The diff coverage is 55.33%.

@@            Coverage Diff             @@
##           master      #70      +/-   ##
==========================================
+ Coverage   80.17%   82.22%   +2.04%     
==========================================
  Files          27       32       +5     
  Lines         812      945     +133     
==========================================
+ Hits          651      777     +126     
- Misses        161      168       +7

Impacted Files	Coverage Δ
accesspoc/dips/helpers.py	`100% <100%> (ø)`	⬆️
accesspoc/search/documents.py	`100% <100%> (ø)`	⬆️
accesspoc/dips/tasks.py	`100% <100%> (ø)`
...cesspoc/dips/migrations/0011_auto_20180729_1659.py	`100% <100%> (ø)`
accesspoc/dips/migrations/0012_taskresult.py	`100% <100%> (ø)`
accesspoc/accesspoc/settings.py	`100% <100%> (ø)`	⬆️
accesspoc/dips/models.py	`99.3% <100%> (+0.15%)`	⬆️
accesspoc/accesspoc/__init__.py	`100% <100%> (ø)`
accesspoc/dips/parsemets.py	`13.66% <11.66%> (+4.26%)`	⬆️
accesspoc/dips/views.py	`91.59% <70%> (+3.31%)`	⬆️
... and 6 more

sevein

I haven't tried this yet but I have some initial comments.

[UPDATE]: I've just read your TODO checlist :)

sevein · 2018-07-26T02:59:01Z

accesspoc/dips/tasks.py

+    mets = METS(os.path.abspath(metsfile), dip_id)
+    mets.parse_mets()
+    # Delete temporary directoy
+    shutil.rmtree(tmpdir)


I realize this function would look much nicer if you used a context manager, e.g.:

with tempfile.TemporaryDirectory(): zip = zipfile.ZipFile(zip_path) for info in zip.infolist(): if re.match(r'.*METS.[0-9a-f\-]{36}.*$', info.filename): metsfile = zip.extract(info, tmpdir) mets = METS(os.path.abspath(metsfile), dip_id) mets.parse_mets()

One question is what happens if an exception is raised inside this function and it's not controlled?
Have you tested that case? Is it retried? Is it the error recorded somewhere?

sevein · 2018-07-26T03:13:26Z

accesspoc/dips/tasks.py

+            metsfile = zip.extract(info, tmpdir)
+    # Parse METS file
+    mets = METS(os.path.abspath(metsfile), dip_id)
+    mets.parse_mets()


There's a bunch of database action going on here. I'm wondering if Celery wraps all that within a database transaction or you have to do it manually. Have you thought about that at all?

Also, introducing workers may be a reason to move to MySQL. Maybe. We should investigate further. See https://stackoverflow.com/a/10387821. Is WAL enabled for you?

It looks like it has to be done manually, working on that ;)

This is interesting: https://www.sqlite.org/tempfiles.html

And Django only complains about the save points when "autocommit" is turned off: https://docs.djangoproject.com/en/1.11/topics/db/transactions/#savepoints-in-sqlite

sevein · 2018-07-26T03:14:34Z

docker-compose.yml

+      - CELERY_BROKER_URL=redis://redis:6379
+    volumes:
+      - ./accesspoc:/src
+    command: 'celery -A accesspoc worker -l info'


Are you using the result backend too?

sevein · 2018-07-26T03:15:21Z

docker-compose.yml

+  redis:
+    image: redis:4-alpine
+    user: redis
+    command: '--save "" --appendonly no'


You may want to change this so it actually persists to disk. It'd be a shame that you lose all the data because you decided to restart the service.

sevein · 2018-07-26T03:21:16Z

accesspoc/accesspoc/__init__.py

@@ -0,0 +1,5 @@
+# This will make sure the app is always imported when
+# Django starts so that shared_task will use this app.


This should be a proper docstring.

sevein · 2018-07-26T03:27:22Z

accesspoc/dips/tasks.py

+
+
+@shared_task
+def extract_and_parse_mets(dip_id, zip_path):


Can we add a test?
http://docs.celeryproject.org/en/latest/userguide/testing.html

sevein · 2018-07-26T03:37:27Z

accesspoc/dips/models.py

@@ -154,6 +154,7 @@ class DIP(AbstractEsModel):
        verbose_name=_('collection'),
    )
    dc = models.OneToOneField(DublinCore, null=True, on_delete=models.SET_NULL)
+    import_in_progress = models.BooleanField(default=False)


Isn't a new dip always in progress? i.e. defaut=True?

Also, what would happen if the task never completes? Is there a way to report errors to the user? Do we need the result backend?

I set default to true initially and realized that it would require a data migration to set the existing DIPs to false, which made me realize that setting it to true by default is a wrong assumption as the import may not always happen on creation (in the future). I may also remove this field if I find a better solution for this with the results backend.

jraddaoui · 2018-07-26T12:25:32Z

Thanks @sevein, I really appreciate the initial feedback!

jraddaoui · 2018-07-26T12:45:33Z

I have updated the TODO list with some of your feedback ;)

jraddaoui · 2018-07-31T01:29:31Z

Sorry @sevein, I shouldn't have amended the latest commit, since you already reviewed it. I've added your suggestion and a missing migration for the proxy model.

sevein

I haven't had time to test it yet but I've added some comments.

sevein · 2018-07-28T01:38:21Z

accesspoc/dips/parsemets.py

        # Get DIP object
        dip = DIP.objects.get(pk=self.dip_id)

        # Gather info for each file in filegroup "original"
-        for target in mets_root.findall(".//fileGrp[@USE='original']/file"):
+        for file in self.mets_root.findall(".//fileGrp[@USE='original']/file"):


file is a builtin function, you could call it file_.

sevein · 2018-07-28T01:39:29Z

accesspoc/dips/parsemets.py

+
+        return data
+
+    def parse_dc(self):


Is this a private method? If so, the name should start with an underscore to make that clear.

sevein · 2018-07-28T01:40:43Z

accesspoc/dips/tasks.py

+    file during its parsing. This function is meant to be called with
+    `.delay()` to be executed asynchronously by the Celery worker.
+    """
+    with tempfile.TemporaryDirectory() as dir, zipfile.ZipFile(zip_path) as zip:


dir and zip are both builtin functions.

sevein · 2018-07-28T01:42:35Z

accesspoc/dips/tasks.py

+    with tempfile.TemporaryDirectory() as dir, zipfile.ZipFile(zip_path) as zip:
+        # Extract METS file
+        for info in zip.infolist():
+            if re.match(r'.*METS.[0-9a-f\-]{36}.*$', info.filename):


Micro-optimization: use regex = re.compile(r'.*METS.[0-9a-f\-]{36}.*$) before the loop starts.

sevein · 2018-07-28T01:45:34Z

accesspoc/dips/parsemets.py

+        # Parse all DC elements to a dictionary
+        dc_model = dict()
+        for elem in dc_xml:
+            dc_model['%s' % (elem.tag)] = elem.text


Simpler:

dc_model[str(elem.tag)] = elem.text

sevein · 2018-07-28T01:47:49Z

accesspoc/dips/parsemets.py

@@ -19,203 +20,187 @@ def convert_size(size):

 class METS(object):
    """
-    Class for METS file parsing methods
+    Class for METS file parsing methods.
    """


I can't comment on convert_size so I'll do it here. If it's private, add the underscore to its name.

sevein · 2018-07-28T01:54:51Z

accesspoc/dips/parsemets.py

+            # Convert from unix to iso8601
+            data['datemodified'] = datetime.fromtimestamp(unixtime).isoformat()
+        else:
+            data['datemodified'] = ''


Would this one be equivalent and a little safer?

try: unixtime = int(data['datemodified']) / 1000 except ValueError: data['datemodified'] = '' else: data['datemodified'] = datetime.fromtimestamp(unixtime).isoformat()

sevein · 2018-07-31T15:43:58Z

accesspoc/dips/views.py

+            'the METS file. After the process finishes and the interface',
+            'is reloaded, a link to the Folder will show up in the',
+            'Folders table at the related Collection page.',
+        ])))


Is that going to work? I would have thought that the join would never happen because the annotated string is extracted beforehand? An alternative could be:

messages.info( request, 'A background process has been launched to extract and parse the' ' METS file. After the process finishes and the interface is' ' reloaded, a link to the Folder will show up in the Folders table' ' at the related Collection page.')

Oh, forgot to change it here from the previous feedback.

sevein · 2018-07-31T15:48:34Z

accesspoc/accesspoc/__init__.py

@@ -0,0 +1,7 @@
+"""
+This will make sure the Celry app is always imported


Typo: s/Celry/Celery.

jraddaoui · 2018-08-09T11:19:18Z

Hi @sevein, this is ready for another round. I'd appreciate if we can get the code right before I start writing the install notes. Thanks!

jraddaoui · 2018-08-09T11:56:58Z

Actually, I'd like to take care of #43 too before writing the install notes, and I'd love to have this merged before that, to avoid a lot of conflicts in the settings files. So, maybe we could add the install notes in a different PR after both things are done ;)

jraddaoui · 2018-08-09T12:12:57Z

Updated the PR description to reflect the latest changes.

sevein

I've tried it and it's really neat, congratulations @jraddaoui.

I've made some more comments but they're mostly nitpicks.

Something that I'd like to ask is to reduce the number of events that we're sending to the logger in tasks.py to just human-actionable events. Most of the places where you're using INFO I'd rather use DEBUG. If something is bad and it requires attention from the operator then use ERROR or WARNING as needed.

As Peter Bourgon wrote in his blog:

[...] services should only log actionable information. That includes serious, panic-level errors that need to be consumed by humans, or structured data that needs to be consumed by machines. An example of the former would be a message signaling that a required database has become completely unavailable. An example of the latter would be a message indicating a media object has been played, recorded so an end-of-day batch process might calculate owed royalties. Logs read by humans should be sparse, ideally silent if nothing is going wrong. Logs read by machines should be well-defined, ideally with a versioned schema.

Thank you, this is really great work.

sevein · 2018-08-13T23:29:32Z

accesspoc/dips/helpers.py

+            instance._meta.get_field(field)
+            setattr(instance, field, value)
+        except FieldDoesNotExist:
+            pass


I think that it'd be clearer if you write:

try: instance._meta.get_field(field) except FieldDoesNotExist: continue setattr(instance, field, value)

sevein · 2018-08-13T23:32:52Z

accesspoc/dips/migrations/0011_auto_20180729_1659.py

@@ -0,0 +1,25 @@
+# -*- coding: utf-8 -*-


I guess it doesn't really matter as long as you're planning to squash the migrations soon, but I feel obliged to say again that we should avoid extra migrations as much as possible, e.g. 0011 and 0012 could become just one migration. Also it'd be good if they have meaningful names instead of the name automatically generated by Django.

Created #78.

sevein · 2018-08-13T23:41:22Z

accesspoc/dips/models.py

+            info = result.get_error_message()
+        except TaskResult.DoesNotExist:
+            info = gettext('A related TaskResult could not be found.')
+        return ' '.join([intro, info, please])


Remember that we shouldn't use string concatenation in i18n because each language has its own syntax.

It'd be better to have a single string with parameters, e.g.:

An error occurred during the process executed to extract and parse the METS file - see error below. Please, contact an administrator. %(error_message)s

sevein · 2018-08-13T23:49:13Z

accesspoc/dips/tasks.py

+import os
+import re
+import tempfile
+import zipfile


It'd be nice at some point if you introduce in CI the use of https://github.com/PyCQA/flake8-import-order. It supports several styles: https://github.com/PyCQA/flake8-import-order#styles. Yours is different to all of them I believe. I think that google is my fav - or one of its derivatives.

Added to #67.

sevein · 2018-08-13T23:50:41Z

accesspoc/dips/tasks.py

+        """
+        if status not in states.READY_STATES:
+            return
+        dip = DIP.objects.get(pk=args[0])


What'd happen if it raises a DIP.DoesNotExist?

Good question, will check.

From the worker log:

worker_1 | [2018-08-14 17:33:13,237: INFO/ForkPoolWorker-8] Task dips.tasks.extract_and_parse_mets[c79c61e0-4441-4cd5-9ebc-fc5732e514dd] succeeded in 0.781619158002286s: None worker_1 | [2018-08-14 17:33:13,238: WARNING/ForkPoolWorker-8] /usr/local/lib/python3.6/site-packages/celery/app/trace.py:561: RuntimeWarning: Exception raised outside body: Exception('Manually raised.',): worker_1 | Traceback (most recent call last): worker_1 | File "/usr/local/lib/python3.6/site-packages/celery/app/trace.py", line 465, in trace_task worker_1 | state, retval, uuid, args, kwargs, None, worker_1 | File "/src/dips/tasks.py", line 27, in after_return worker_1 | raise Exception('Manually raised.') worker_1 | Exception: Manually raised. worker_1 | worker_1 | exc, exc_info.traceback))) worker_1 | [2018-08-14 17:33:13,244: ERROR/ForkPoolWorker-8] Task dips.tasks.extract_and_parse_mets[c79c61e0-4441-4cd5-9ebc-fc5732e514dd] raised unexpected: Exception('Manually raised.',) worker_1 | Traceback (most recent call last): worker_1 | File "/usr/local/lib/python3.6/site-packages/celery/app/trace.py", line 465, in trace_task worker_1 | state, retval, uuid, args, kwargs, None, worker_1 | File "/src/dips/tasks.py", line 27, in after_return worker_1 | raise Exception('Manually raised.') worker_1 | Exception: Manually raised.

The worker is still running and, amazingly, the error is recorded in the TaskResult. Great job Celery!

sevein · 2018-08-13T23:59:52Z

accesspoc/dips/views.py

+    if (digitalfile.dip.import_status == DIP.IMPORT_PENDING or (
+        not request.user.is_editor() and
+        digitalfile.dip.import_status == DIP.IMPORT_FAILURE
+    )):


Nitpick: as you've already done this twice, why not to write a small method in the model that takes the request.user and returns what you're looking for, e.g. digitalfile.dip.is_visitable(request.user).

def is_visitable(self, user): if self.import_status == DIP.IMPORT_PENDING: return False if self.import_status == DIP.IMPORT_FAILURE \ and user.is_editor(): return False return True

... ^ quick draft to make sure I'm clear on what I'm asking.

sevein · 2018-08-14T00:02:45Z

accesspoc/dips/views.py

+        # TODO: Avoid this save from updating the related ES
+        # documents, as a later save when the async_result id
+        # is added, will repeat the process. It may require
+        # stop using signals and move to a custom save method.


It'd be good to file a bug once it's merged! Unless you've already done it :)

Created #79. I'll add more info after this is merged.

jraddaoui · 2018-08-14T15:15:43Z

Thanks @sevein!! Great feedback in all the PR, I really appreciate that kind of involvement.

About the log level, the reason I used INFO is because elasticsearch-py logs each request with that level and I don't like them appearing by themselves. I think it's okay to leave them like that and recommend to run the Celery worker with a level of WARNING in the install notes. Please, let me know if you still don't like that idea ;)

sevein · 2018-08-14T15:29:08Z

Makes sense!

- Add Celery and Redis requirements. - Add Celery config in settings. - Add celery app and load it on init. - Use Redis as Broker and `django_celery_results` as backend. - Move METS extraction and parsing to a shared task. - Improvements in METS parsing process. - Redirect to collection page on DIP creation. - Reflect when a DIP is being imported: - Add `import_status` and `ìmport_task_id` columns to DIPs model and ES doc. Reflect status on related DigitalFiles. - Set it to `PENDING` when the METS handling process is launched and to `FAILURE` or `SUCCESS` in the task `after_return` call. - For editors and admins: - Show spinning icon instead of links for DIPs and DigitalFiles when the import is `PENDING`. - Show error message in view pages. - For other users: - Do not show DIPs or DigitalFiles with an import status of `FAILURE` or `PENDING`. - Add flash messages system to show user notifications on import launch and on import failure. - Create `dips.models.TaskResult` proxy model of `django_celery_results` to format error message. - Move some functions to helpers and add tests. - Add tests for task and `after_return` calls - Set `CELERY_BROKER_URL` env. var. in Tox test environment. - Improve docker-compose environment: - Remove ES network and use default network. - Set `CELERY_BROKER_URL` env. var. - Add redis and worker services. - Add redis_data volume.

jraddaoui added the in progress label Jul 26, 2018

jraddaoui added this to the beta milestone Jul 26, 2018

jraddaoui self-assigned this Jul 26, 2018

sevein requested changes Jul 26, 2018

View reviewed changes

jraddaoui force-pushed the dev/issue-15-celery branch from 4f0f944 to 2ac2419 Compare July 26, 2018 19:11

This was referenced Jul 28, 2018

Add tests for METS file parsing process #73

Closed

Reduce mismatch possibility between database and Elasticsearch indexes #74

Closed

jraddaoui force-pushed the dev/issue-15-celery branch from 1f3e237 to 7ccc1e3 Compare July 29, 2018 06:29

jraddaoui mentioned this pull request Jul 29, 2018

Consider the use of a different database service #29

Closed

jraddaoui force-pushed the dev/issue-15-celery branch from a19ac75 to 834c921 Compare July 30, 2018 00:07

jraddaoui mentioned this pull request Jul 30, 2018

Move session data from database to cache #75

Open

jraddaoui force-pushed the dev/issue-15-celery branch from b612889 to adc9fa6 Compare July 31, 2018 01:21

sevein reviewed Jul 31, 2018

View reviewed changes

CCA-Public deleted a comment from sevein Jul 31, 2018

jraddaoui force-pushed the dev/issue-15-celery branch 2 times, most recently from 083c8f0 to 8a9a073 Compare August 7, 2018 20:13

jraddaoui mentioned this pull request Aug 9, 2018

Technical requirements documentation #48

Closed

jraddaoui force-pushed the dev/issue-15-celery branch from 6f100e6 to f7174e2 Compare August 9, 2018 17:17

sevein reviewed Aug 14, 2018

View reviewed changes

jraddaoui mentioned this pull request Aug 14, 2018

Improve CI process #67

Closed

sevein approved these changes Aug 14, 2018

View reviewed changes

jraddaoui force-pushed the dev/issue-15-celery branch from c50485d to 46b8b42 Compare August 14, 2018 18:19

jraddaoui merged commit 46b8b42 into master Aug 14, 2018

jraddaoui removed the in progress label Aug 14, 2018

jraddaoui deleted the dev/issue-15-celery branch August 14, 2018 18:26

		@@ -0,0 +1,5 @@
		# This will make sure the app is always imported when
		# Django starts so that shared_task will use this app.

		@@ -0,0 +1,7 @@
		"""
		This will make sure the Celry app is always imported

Setup Celery worker with Redis broker #70

Setup Celery worker with Redis broker #70

Conversation

jraddaoui commented Jul 26, 2018 • edited Loading

codecov bot commented Jul 26, 2018 • edited Loading

Codecov Report

sevein left a comment • edited Loading

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

Choose a reason for hiding this comment

jraddaoui commented Jul 26, 2018

jraddaoui commented Jul 26, 2018

jraddaoui commented Jul 31, 2018

sevein 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

Choose a reason for hiding this comment

jraddaoui commented Aug 9, 2018

jraddaoui commented Aug 9, 2018

jraddaoui commented Aug 9, 2018

sevein 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

sevein Aug 13, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jraddaoui commented Aug 14, 2018

sevein commented Aug 14, 2018

jraddaoui commented Jul 26, 2018 •

edited

Loading

codecov bot commented Jul 26, 2018 •

edited

Loading

sevein left a comment •

edited

Loading

sevein Aug 13, 2018 •

edited

Loading