Add UMB broker to docker-compose environment

Signed-off-by: Lukas Holecek <hluk@email.cz>

docker-compose.yml

@@ -1,3 +1,4 @@
+---
 version: '2.1'
 services:
   dev:
@@ -19,6 +20,23 @@ services:
       - resultsdb
       - waiverdb
 
+  resultsdb-consumer:
+    build: .
+    user: ${DEV_USER_ID:-1000}
+    working_dir: /
+    env_file: docker/dev.env
+    command: ["/usr/bin/fedmsg-hub-3"]
+    volumes:
+      - ./:/src:ro,z
+      - ./docker/home:/home/dev:rw,z
+      - ./docker/greenwave-settings.py:/etc/greenwave/settings.py:ro,z
+      - ./conf/policies/:/etc/greenwave/policies/:ro,z
+      - ./conf/subject_types/:/etc/greenwave/subject_types/:ro,z
+      - ./docker/fedmsg.d:/etc/fedmsg.d:ro,z
+    depends_on:
+      - memcached
+      - umb
+
   resultsdb-db:
     image: postgres:9.5.2
     restart: always
@@ -66,14 +84,28 @@ services:
       - ./docker/waiverdb-settings.py:/etc/waiverdb/settings.py:ro,z
       - ./docker/client_secrets.json:/etc/secret/client_secrets.json:ro,z
     ports:
-     - 5004:5004
+      - 5004:5004
     user: '0'
     depends_on:
       - waiverdb-db
 
   memcached:
     image: "quay.io/factory2/memcached"
 
+  umb:
+    image: docker-registry.upshift.redhat.com/c3i/umb:latest
+    restart: unless-stopped
+    command:
+      - "java"
+      - "-jar"
+      - "/usr/share/umb/umb.jar"
+      - "--spring.config.location=/etc/umb.yml"
+    volumes:
+      - ./docker/umb.yml:/etc/umb.yml:ro,z
+    ports:
+      - 5671:5671    # amqp
+      - 61612:61612  # stomp
+
 networks:
   default:
     driver: bridge

docker/dev.env

@@ -4,3 +4,5 @@ GREENWAVE_TEST_URL=http://dev:8080/
 WAIVERDB_TEST_URL=http://waiverdb:5004/
 RESULTSDB_TEST_URL=http://resultsdb:5001/
 PYTEST_ADDOPTS=-o cache_dir=/home/dev/.pytest_cache
+PYTHONPATH=/src
+RESULTSDB_HANDLER=please

docker/fedmsg.d/base.py

@@ -0,0 +1,48 @@
+import os
+
+config = {
+    # fedmsg boilerplate
+    'endpoints': {},
+    'sign_messages': False,
+    'validate_signatures': False,
+
+    # STOMP settings
+    'zmq_enabled': False,
+    'stomp_uri': 'umb:61612',
+    'stomp_heartbeat': 900000,
+    'stomp_ack_mode': 'client-individual',
+
+    # Hacks to make us publish to
+    # /topic/VirtualTopic.eng.greenwave.decision.update
+    'topic_prefix': '/topic/VirtualTopic',
+    'environment': 'eng',
+    'resultsdb_topic_suffix': 'resultsdb.result.new',
+    'waiverdb_topic_suffix': 'waiverdb.waiver.new',
+
+    # Workaround for moksha memory leak and mitigate message loss.
+    # Memory leak is fixed in python-moksha-hub-1.5.7 (https://github.com/mokshaproject/moksha/pull/57).
+    'moksha.blocking_mode': True,
+
+    # moksha-monitor-exporter's point of contact
+    'moksha.monitoring.socket': 'tcp://0.0.0.0:10030',
+}
+
+# Enable one consumer or the other in different deployments.
+if os.environ.get("RESULTSDB_HANDLER") and os.environ.get("WAIVERDB_HANDLER"):
+    raise ValueError("Both RESULTSDB_HANDLER and WAIVERDB_HANDLER may not"
+                     "be specified.  Only one.")
+
+if os.environ.get("RESULTSDB_HANDLER"):
+    config.update({
+        'resultsdb_handler': True,
+        'waiverdb_handler': False,
+        'stomp_queue': '/queue/Consumer.client-greenwave.resultsdb.VirtualTopic.eng.resultsdb.result.new',
+    })
+elif os.environ.get("WAIVERDB_HANDLER"):
+    config.update({
+        'resultsdb_handler': False,
+        'waiverdb_handler': True,
+        'stomp_queue': '/queue/Consumer.client-greenwave.waiverdb.VirtualTopic.eng.waiverdb.waiver.new',
+    })
+
+

docker/fedmsg.d/logging.py

@@ -0,0 +1,38 @@
+config = dict(logging={
+    "version": 1,
+    "disable_existing_loggers": False,
+    "formatters": {
+        "bare": {
+            "datefmt": "%Y-%m-%d %H:%M:%S",
+            "format": "[%(asctime)s][%(name)10s %(levelname)7s] %(message)s"
+        }
+    },
+    "loggers": {
+        "greenwave": {
+            "handlers": ["console"], "propagate": True, "level": "DEBUG"
+        },
+        "moksha": {
+            "handlers": ["console"], "propagate": False, "level": "DEBUG"
+        },
+        "requests": {
+            "handlers": ["console"], "propagate": False, "level": "DEBUG"
+        },
+        "resultsdb_handler": {
+            "handlers": ["console"], "propagate": False, "level": "DEBUG"
+        },
+        "waiverdb_handler": {
+            "handlers": ["console"], "propagate": False, "level": "DEBUG"
+        },
+        "dogpile": {
+            "handlers": ["console"], "propagate": False, "level": "DEBUG"
+        },
+    },
+    "handlers": {
+        "console": {
+            "formatter": "bare",
+            "class": "logging.StreamHandler",
+            "stream": "ext://sys.stdout",
+            "level": "DEBUG"
+        }
+    },
+})

docker/resultsdb-settings.py

@@ -6,7 +6,14 @@
 STREAM_LOGGING = True
 RUN_HOST = '0.0.0.0'
 RUN_PORT = 5001
-MESSAGE_BUS_PUBLISH = False
-MESSAGE_BUS_PLUGIN = 'fedmsg'
-MESSAGE_BUS_KWARGS = {'modname': 'resultsdb'}
+MESSAGE_BUS_PUBLISH = True
+MESSAGE_BUS_PLUGIN = 'stomp'
+MESSAGE_BUS_KWARGS = {
+    'modname': 'resultsdb',
+    'destination': '/topic/VirtualTopic.eng.resultsdb.result.new',
+    'connection': {
+        'host_and_ports': [('localhost', 61612)],
+        'use_ssl': False,
+    },
+}
 ADDITIONAL_RESULT_OUTCOMES = ('RUNNING', 'QUEUED', 'ERROR')

docker/scripts/consume.py

@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+
+import json
+import os
+import itertools
+
+from pprint import pprint
+
+from rhmsg.activemq.consumer import AMQConsumer
+
+ADDRESS = "Consumer.mister.queue.VirtualTopic.eng.>"
+SUBSCRIPTION_NAME = "Greenwave"
+URLS = ["amqp://localhost:5671"]
+
+counter = itertools.count(1)
+
+
+class InsecureAMQConsumer(AMQConsumer):
+    ssl_domain = None
+
+    def __init__(self, urls):
+        self.urls = urls
+
+
+def message_handler(message, data):
+    num = next(counter)
+
+    body = message.body
+    if isinstance(body, str):
+        body = body.encode("utf-8", "backslashreplace")
+    if data["dump"]:
+        print("------------- ({0}) {1} --------------".format(num, message.id))
+        print("address:", message.address)
+        print("subject:", message.subject)
+        print("properties:", message.properties)
+        print("durable:", message.durable)
+        print("content_type:", message.content_type)
+        print("content_encoding:", message.content_encoding)
+        print("delivery_count:", message.delivery_count)
+        print("reply_to:", message.reply_to)
+        print("priority:", message.priority)
+        if data["pp"]:
+            print("body:")
+            pprint(json.loads(body))
+        else:
+            print("body:", body)
+    else:
+        if data["pp"]:
+            print("Got [%02d]:" % num)
+            pprint(json.loads(body))
+        else:
+            print("Got [%02d]:" % num, body)
+
+    return data["one_message_only"], not data["manual_ack"]
+
+
+def main():
+    os.environ['PN_TRACE_FRM'] = '1'
+    consumer = InsecureAMQConsumer(urls=URLS)
+    consumer.consume(
+        ADDRESS,
+        selector=None,
+        callback=message_handler,
+        auto_accept=False,
+        subscription_name=SUBSCRIPTION_NAME,
+        data={
+            "dump": False,
+            "pp": False,
+            "one_message_only": False,
+            "manual_ack": False,
+        },
+    )
+
+
+if __name__ == '__main__':
+    main()

docker/scripts/produce.py

@@ -0,0 +1,69 @@
+#!/usr/bin/env python3
+
+import json
+import proton
+import os
+import sys
+
+from rhmsg.activemq.producer import AMQProducer
+
+TOPIC = "VirtualTopic.eng.resultsdb.result.new"
+URLS = ["amqp://localhost:5671"]
+SUBJECT = f"test_message_{sys.argv[1]}"
+MESSAGE = {
+    "submit_time": "2019-08-27T13:57:53.490376",
+    "testcase": {
+        "name": "example_test"
+    },
+    "data": {
+        "type": ["brew-build"],
+        "item": ["example-container"]
+    }
+}
+MESSAGE = {
+    "data": {
+        "category": ["validation"],
+        "ci_email": ["exd-guild-gating@redhat.com"],
+        "ci_irc": ["not available"],
+        "ci_name": ["example-jenkins"],
+        "ci_team": ["PnT DevOps"],
+        "ci_url": ["https://jenkins.example.com"],
+        "component": ["nethack-prod"],
+        "item": ["nethack-prod-3.5.202110051331.w9756"],
+        "log": ["https://jenkins.example.com/job/x/build/y/console"],
+        "publisher_id": ["msg-greenwave-segment-test"],
+        "type": ["koji_build"],
+        "version": ["3.5.202110051331.w9756"]
+    },
+    "groups": ["52c6b84b-b617-4b79-af47-8975d11bb635"],
+    "href": "http://resultsdb/api/v2.0/results/123",
+    "id": 123,
+    "note": "",
+    "outcome": "PASSED",
+    "ref_url": "https://jenkins.example.com/job/x/build/y",
+    "submit_time": "2021-10-05T13:35:29.721850",
+    "testcase": {
+        "href": "http://resultsdb/api/v2.0/testcases/dist.abicheck",
+        "name": "dist.abicheck",
+        "ref_url": "https://jenkins.example.com/job/x/build/y"
+    }
+}
+
+
+def main():
+    os.environ['PN_TRACE_FRM'] = '1'
+
+    with AMQProducer(urls=URLS) as producer:
+        # Disable SSL
+        del producer.conf["cert"]
+
+        producer.through_topic(TOPIC)
+        body = json.dumps(MESSAGE)
+        message = proton.Message(
+            subject=SUBJECT,
+            body=body)
+        producer.send(message)
+
+
+if __name__ == '__main__':
+    main()

docker/umb.yml

@@ -0,0 +1,35 @@
+---
+server:
+  port: 8081
+management:
+  security.enabled: false
+  endpoints:
+    enabled-by-default: false
+    web:
+      base-path: /
+      exposure:
+        include: "*"
+  health:
+    jms.enabled: false
+    amq.enabled: true
+    ldap.enabled: false
+umb:
+  ldap:
+    embedded:
+      enabled: false
+  amq:
+    broker:
+      plugins:
+        enableLdapBackedAuthentication: false
+        enableLdapBackedAuthorization: false
+      transportConnectors:
+        - amqp://0.0.0.0:5671?transport.maximumConnections=2500&transport.needClientAuth=true&transport.transformer=jms&wireFormat.allowNonSaslConnections=true&transport.closeAsync=false&transport.daemon=true&wireFormat.maxInactivityDurationInitalDelay=60000&wireFormat.maxInactivityDuration=60000
+        - stomp://0.0.0.0:61612?transport.maximumConnections=2500&transport.needClientAuth=true&transport.closeAsync=false&transport.daemon=true&wireFormat.maxInactivityDurationInitalDelay=60000&wireFormat.maxInactivityDuration=60000
+logging:
+  pattern:
+    dateformat: "yyyy-MM-dd'T'HH:mm:ss.SSSZ"
+  path: /var/log/umb
+  file.max-size: 10MB
+  file.max-history: 50
+  server.asyncappender.queue.size: 20000
+  dlq.asyncappender.queue.size: 5000