koji / fedora-messaging-plugin: for now pull plugin in so we can modify it

Longer term we need to figure out how to manage it best.

Signed-off-by: Kevin Fenzi <kevin@scrye.com>

roles/koji_hub/tasks/main.yml

@@ -144,12 +144,18 @@
   - koji_hub
   - fedora-messaging
 
+# discuss how best to manage this plugin
+#- name: koji fedora-messaging plugin - installed as fedmsg-koji-plugin
+#  get_url:
+#      url: https://pagure.io/koji-fedmsg-plugin/raw/master/f/fedmsg-koji-plugin.py
+#      dest: /usr/lib/koji-hub-plugins/fedmsg-koji-plugin.py
+#      mode: 0644
+#      checksum: sha256:a20cd840cc2f0aa6f9720e9fe985777b614931b112e975e11aeaca2a3d185405
 - name: koji fedora-messaging plugin - installed as fedmsg-koji-plugin
-  get_url:
-      url: https://pagure.io/koji-fedmsg-plugin/raw/master/f/fedmsg-koji-plugin.py
-      dest: /usr/lib/koji-hub-plugins/fedmsg-koji-plugin.py
-      mode: 0644
-      checksum: sha256:a20cd840cc2f0aa6f9720e9fe985777b614931b112e975e11aeaca2a3d185405
+  template:
+        src: fedmsg-koji-plugin.py
+        dest: /usr/lib/koji-hub-plugins/fedmsg-koji-plugin.py
+        mode: 644
   notify:
   - reload httpd
   tags:

roles/koji_hub/templates/fedmsg-koji-plugin.py

@@ -1,25 +1,27 @@
-# Koji callback for sending notifications about events to the fedmsg messagebus
-# Copyright (c) 2009-2012 Red Hat, Inc.
+# Koji callback for sending notifications about events to the fedmsg message bus
+# Copyright (c) 2009-2019 Red Hat, Inc.
+#
+# Source: https://pagure.io/koji-fedmsg-plugin/
 #
 # Authors:
 #     Ralph Bean <rbean@redhat.com>
 #     Mike Bonnet <mikeb@redhat.com>
 
+import logging
+import re
+import time
+
 from koji.context import context
 from koji.plugin import callbacks
 from koji.plugin import callback
 from koji.plugin import ignore_error
-
-import fedmsg
+import fedora_messaging.api
+import fedora_messaging.exceptions
 import kojihub
-import re
 
-import pprint
-
-# Talk to the fedmsg-relay
-fedmsg.init(name='relay_inbound', cert_prefix='koji', active=True)
 
 MAX_KEY_LENGTH = 255
+log = logging.getLogger(__name__)
 
 
 def camel_to_dots(name):
@@ -27,6 +29,13 @@ def camel_to_dots(name):
     return re.sub('([a-z0-9])([A-Z])', r'\1.\2', s1).lower()
 
 
+def serialize_datetime_in_task(task):
+    for date_key in ("completion_time", "create_time", "start_time"):
+        if task[date_key] is None:
+            continue
+        task[date_key] = time.mktime(task[date_key].timetuple())
+
+
 def get_message_body(topic, *args, **kws):
     msg = {}
 
@@ -44,10 +53,15 @@ def get_message_body(topic, *args, **kws):
         msg['update'] = kws.get('update', None)
     elif topic == 'task.state.change':
         info = kws['info']
+        serialize_datetime_in_task(info)
 
         # Stuff in information about descendant tasks
         task = kojihub.Task(info['id'])
-        info['children'] = task.getChildren()
+        info['children'] = []
+        for child_orig in task.getChildren():
+            child = child_orig.copy()
+            serialize_datetime_in_task(child)
+            info['children'].append(child)
 
         # Send the whole info dict along because it might have useful info.
         # For instance, it contains the mention of what format createAppliance
@@ -142,7 +156,7 @@ def get_message_body(topic, *args, **kws):
 @callback(*[
     c for c in callbacks.keys()
     if c.startswith('post') and c not in [
-        'postImport', # This is kind of useless; also noisy.
+        'postImport',  # This is kind of useless; also noisy.
         # This one is special, and is called every time, so ignore it.
         # Added here https://pagure.io/koji/pull-request/148
         'postCommit',
@@ -174,7 +188,7 @@ def queue_message(cbtype, *args, **kws):
 
     # We need this to distinguish between messages from primary koji
     # and the secondary hubs off for s390 and ppc.
-    body['instance'] = '{{ fedmsg_koji_instance }}'
+    body['instance'] = 'primary'
 
     # Don't publish these uninformative rpm.sign messages if there's no actual
     # sigkey present.  Koji apparently adds a dummy sig value when rpms are
@@ -191,12 +205,15 @@ def queue_message(cbtype, *args, **kws):
     # These fields are floating points which get json-encoded differently on
     # rhel and fedora.
     problem_fields = ['weight', 'start_ts', 'create_ts', 'completion_ts']
+
     def scrub(obj):
         if isinstance(obj, list):
             return [scrub(item) for item in obj]
         if isinstance(obj, dict):
             return dict([
-                (k, scrub(v)) for k, v in obj.items() if k not in problem_fields
+                (k, scrub(v))
+                for k, v in obj.items()
+                if k not in problem_fields
             ])
         return obj
 
@@ -205,7 +222,7 @@ def queue_message(cbtype, *args, **kws):
     # Queue the message for later.
     # It will only get sent after postCommit is called.
     messages = getattr(context, 'fedmsg_plugin_messages', [])
-    messages.append(dict(topic=topic, msg=body, modname='buildsys'))
+    messages.append(dict(topic=topic, msg=body))
     context.fedmsg_plugin_messages = messages
 
 
@@ -214,5 +231,19 @@ def queue_message(cbtype, *args, **kws):
 @ignore_error
 def send_messages(cbtype, *args, **kws):
     messages = getattr(context, 'fedmsg_plugin_messages', [])
+
     for message in messages:
-        fedmsg.publish(**message)
+        try:
+            msg = fedora_messaging.api.Message(
+                topic="buildsys.{}".format(message['topic']),
+                body=message['msg']
+            )
+            fedora_messaging.api.publish(msg)
+        except fedora_messaging.exceptions.PublishReturned as e:
+            log.warning(
+                "Fedora Messaging broker rejected message %s: %s", msg.id, e
+            )
+        except fedora_messaging.exceptions.ConnectionException as e:
+            log.warning("Error sending message %s: %s", msg.id, e)
+        except Exception:
+            log.exception("Un-expected error sending fedora-messaging message")