Publish messages via fedora_messaging

This is the bare minimum to publish messages using fedora_messaging.
It includes support to use the fedmsg publisher as a config option in
case things go wrong during deployment. This can be removed later.

Signed-off-by: Jeremy Cline <jcline@redhat.com>

anitya/admin.py

@@ -43,7 +43,7 @@ def add_distro():
 
         utilities.log(
             Session,
-            distro=distro,
+            distro=distro.__json__(),
             topic='distro.add',
             message=dict(
                 agent=flask.g.user.username,
@@ -88,7 +88,7 @@ def edit_distro(distro_name):
         if name != distro.name:
             utilities.log(
                 Session,
-                distro=distro,
+                distro=distro.__json__(),
                 topic='distro.edit',
                 message=dict(
                     agent=flask.g.user.username,
@@ -131,7 +131,7 @@ def delete_distro(distro_name):
     if form.validate_on_submit():
         utilities.log(
             Session,
-            distro=distro,
+            distro=distro.__json__(),
             topic='distro.remove',
             message=dict(
                 agent=flask.g.user.username,
@@ -171,7 +171,7 @@ def delete_project(project_id):
         if confirm:
             utilities.log(
                 Session,
-                project=project,
+                project=project.__json__(),
                 topic='project.remove',
                 message=dict(
                     agent=flask.g.user.username,
@@ -226,7 +226,7 @@ def delete_project_mapping(project_id, distro_name, pkg_name):
         if confirm:
             utilities.log(
                 Session,
-                project=project,
+                project=project.__json__(),
                 topic='project.map.remove',
                 message=dict(
                     agent=flask.g.user.username,
@@ -281,7 +281,7 @@ def delete_project_version(project_id, version):
         if confirm:
             utilities.log(
                 Session,
-                project=project,
+                project=project.__json__(),
                 topic='project.version.remove',
                 message=dict(
                     agent=flask.g.user.username,

anitya/config.py

@@ -97,6 +97,7 @@
                   '(?:minsrc|src|source|asc|release))?\.(?:tar|t[bglx]z|tbz2|zip)',
     # Token for GitHub API
     GITHUB_ACCESS_TOKEN=None,
+    LEGACY_MESSAGING=False,  # If True, publish with fedmsg instead of fedora_messaging
 )
 
 # Start with a basic logging configuration, which will be replaced by any user-

anitya/lib/utilities.py

@@ -22,13 +22,15 @@
 import logging
 
 import sqlalchemy as sa
+from fedora_messaging import api, message, exceptions as fm_exceptions
 from sqlalchemy import create_engine
 from sqlalchemy.exc import SQLAlchemyError, IntegrityError
 from sqlalchemy.orm import sessionmaker
 from sqlalchemy.orm import scoped_session
 from sqlalchemy.orm.exc import NoResultFound
 
 from . import plugins, exceptions
+from anitya import config
 from anitya.db import models, Base
 
 
@@ -37,19 +39,29 @@
 
 def fedmsg_publish(*args, **kwargs):  # pragma: no cover
     ''' Try to publish a message on the fedmsg bus. '''
-    # We catch Exception if we want :-p
-    # pylint: disable=W0703
-    # Ignore message about fedmsg import
-    # pylint: disable=F0401
-    kwargs['modname'] = 'anitya'
-    kwargs['cert_prefix'] = 'anitya'
-    kwargs['name'] = 'relay_inbound'
-    kwargs['active'] = True
-    try:
-        import fedmsg
-        fedmsg.publish(*args, **kwargs)
-    except Exception as err:
-        _log.error(str(err))
+    if config.config["LEGACY_MESSAGING"]:
+        # We catch Exception if we want :-p
+        # pylint: disable=W0703
+        # Ignore message about fedmsg import
+        # pylint: disable=F0401
+        kwargs['modname'] = 'anitya'
+        kwargs['cert_prefix'] = 'anitya'
+        kwargs['name'] = 'relay_inbound'
+        kwargs['active'] = True
+        try:
+            import fedmsg
+            fedmsg.publish(*args, **kwargs)
+        except Exception as err:
+            _log.error(str(err))
+    else:
+        try:
+            message_class = message.get_class("anitya." + kwargs["topic"])
+            api.publish(message_class(body=kwargs["msg"]))
+        except (fm_exceptions.ConnectionException, fm_exceptions.PublishException) as err:
+            # For now, continue just logging the error. Once the messaging has
+            # been untangled into SQLAlchemy events, it should probably result
+            # in an exception and the client should try again later.
+            _log.error(str(err))
 
 
 def check_project_release(project, session, test=False):
@@ -108,7 +120,7 @@ def check_project_release(project, session, test=False):
     if publish:
         log(
             session,
-            project=project,
+            project=project.__json__(),
             topic="project.version.update",
             message=dict(
                 project=project.__json__(),
@@ -181,13 +193,6 @@ def log(session, project=None, distro=None, topic=None, message=None):
         message=message,
     ))
 
-    models.Log.insert(
-        session,
-        user=message['agent'],
-        project=project,
-        distro=distro,
-        description=final_msg)
-
     return final_msg
 
 
@@ -270,7 +275,7 @@ def create_project(
 
     log(
         session,
-        project=project,
+        project=project.__json__(),
         topic='project.add',
         message=dict(
             agent=user_id,
@@ -332,7 +337,7 @@ def edit_project(
         if changes:
             log(
                 session,
-                project=project,
+                project=project.__json__(),
                 topic='project.edit',
                 message=dict(
                     agent=user_id,
@@ -380,7 +385,7 @@ def map_project(
         distro_obj = models.Distro(name=distribution)
         log(
             session,
-            distro=distro_obj,
+            distro=distro_obj.__json__(),
             topic='distro.add',
             message=dict(
                 agent=user_id,
@@ -458,8 +463,8 @@ def map_project(
 
     log(
         session,
-        project=project,
-        distro=distro_obj,
+        project=project.__json__(),
+        distro=distro_obj.__json__(),
         topic=topic,
         message=message,
     )
@@ -489,7 +494,7 @@ def flag_project(session, project, reason, user_email, user_id):
 
     log(
         session,
-        project=project,
+        project=project.__json__(),
         topic='project.flag',
         message=dict(
             agent=user_id,

ansible/roles/anitya-dev/defaults/main.yml

@@ -1 +1,2 @@
 import_production_database: true
+rabbitmq_cluster_file_limit: 500000

ansible/roles/anitya-dev/handlers/main.yml

@@ -0,0 +1,5 @@
+- name: reload rabbitmq
+  systemd:
+    name: rabbitmq-server
+    state: restarted
+    daemon_reload: yes

ansible/roles/anitya-dev/tasks/main.yml

@@ -72,9 +72,12 @@
 - name: Install Anitya into the virtualenv
   become_user: "{{ ansible_env.SUDO_USER }}"
   pip:
-    name: "file:///home/{{ ansible_env.SUDO_USER }}/devel"
+    name: "{{ item }}"
     extra_args: "-e"
     virtualenv: /home/{{ ansible_env.SUDO_USER }}/.virtualenvs/anitya/
+  with_items:
+    - "file:///home/{{ ansible_env.SUDO_USER }}/devel"
+    - "file:///home/{{ ansible_env.SUDO_USER }}/devel/anitya_schema"
 
 - name: Create user systemd directory
   become_user: "{{ ansible_env.SUDO_USER }}"
@@ -102,3 +105,4 @@
 
 - import_tasks: db.yml
 - import_tasks: librariesio.yml
+- import_tasks: rabbitmq.yml

ansible/roles/anitya-dev/tasks/rabbitmq.yml

@@ -0,0 +1,26 @@
+---
+- name: Install RabbitMQ packages
+  package:
+      name: "{{ item }}"
+      state: present
+  with_items:
+      - rabbitmq-server
+
+- name: Create RabbitMQ systemd override directory
+  file:
+    path: /etc/systemd/system/rabbitmq-server.service.d/
+    state: directory
+
+- name: Override file limit on rabbitmq
+  copy:
+    content: "[Service]\nLimitNOFILE={{rabbitmq_cluster_file_limit}}\n"
+    dest: /etc/systemd/system/rabbitmq-server.service.d/override.conf
+  notify:
+    - reload rabbitmq
+
+- name: Enables the rabbitmq management and SSL authentication plugins
+  rabbitmq_plugin:
+    names: rabbitmq_management,rabbitmq_auth_mechanism_ssl
+
+- name: start rabbitmq
+  service: name=rabbitmq-server state=started enabled=yes