Upload kafka alert notice after sending

This MR will upload the alert dictionary as a json file to the superevent after kafka alerts have been sent. It's uploaded after because the kafka alert worker currently uses a solo pool, so adding this upload task to the kafka alert send group before the send tasks would delay the alerts (see #450).

Oct 6 edit: The task that uploads the notice json to gracedb now is next in the chain after the tasks that send the kafka notices, instead of being in the same group.

Oct 11 edit: The task now adds an upload task to the queue instead of performing the upload itself, and uploads the serialized messages (i.e. a .avro file for the scimma notice and a .json file for the gcn notice).

This MR depends on !944 (merged) because the unit tests fail due to the skymap being included in the alert dictionary when uploading after sending the alerts. !944 (merged) makes it so the skymap is never added to the alert_dict variable that is shared between tasks.

gwcelery/tasks/alerts.py

+from base64 import b64encode
 import json
+
 from astropy import time
 from celery import group
 from celery.utils.log import get_logger
 from hop.models import AvroBlob
 
 from ..import app
+from .core import identity, get_last
 from . import gracedb
 
 log = get_logger(__name__)
@@ -80,6 +83,22 @@ def _add_external_coinc_to_alert(alert_dict, superevent):
     return alert_dict
 
 
+@gracedb.task(shared=False, ignore_result=True)
+def upload_alert_dict(alert_dict, skymap):
+    if skymap is not None:
+        alert_dict['event']['skymap'] = b64encode(skymap).decode('utf-8')
+
+    alert_dict_json = json.dumps(alert_dict)
+    filename = 'notice-dict-{}-{}.json'.format(
+        alert_dict['superevent_id'],
+        alert_dict['alert_type'].lower()
+    )
+    message = 'Dictionary used to generate Kafka alert notices (without ' + \
+              'skymap)'
+    gracedb.upload(alert_dict_json, filename, alert_dict['superevent_id'],
+                   message)
+
+
 @app.task(bind=True, ignore_result=True, queue='kafka', shared=False)
 def _send(self, alert_dict, skymap, brokerhost):
     """Write the alert to the Kafka topic"""
@@ -140,14 +159,26 @@ def send(self, skymap_and_classification, superevent, alert_type,
             _add_external_coinc_to_alert.s(alert_dict, superevent)
             |
             group(
-                _send.s(skymap, brokerhost) for brokerhost in
-                self.app.conf['kafka_streams'].keys()
+                *(
+                    _send.s(skymap, brokerhost) for brokerhost in
+                    self.app.conf['kafka_streams'].keys()
+                ),
+
+                identity.s()
             )
+            |
+            get_last.s()
+            |
+            upload_alert_dict.s(skymap)
         )
     else:
-        canvas = group(
-            _send.s(alert_dict, skymap, brokerhost) for brokerhost in
-            self.app.conf['kafka_streams'].keys()
+        canvas = (
+            group(
+                _send.s(alert_dict, skymap, brokerhost) for brokerhost in
+                self.app.conf['kafka_streams'].keys()
+            )
+            |
+            upload_alert_dict.si(alert_dict, skymap)
         )
 
     canvas.apply_async()