PR#3921: Add support for the MQTT message bus - pagure

pagure

#3921 Add support for the MQTT message bus

Merged 5 years ago by pingou. Opened 5 years ago by jingjing.

jingjing/pagure mqtt-support into master

Add support for the MQTT message bus

jingjing • 5 years ago

edd8fd9

pagure/default_config.py

file modified

+13

		`@@ -413,6 +413,19 @@`
		`# easy denial of service to the system if enabled.`
		`ALLOW_PROJECT_DOWAIT = False`

		`+ # Settings for MQTT message sending`
		`+ MQTT_NOTIFICATIONS = False`
		`+ MQTT_HOST = None`
		`+ MQTT_PORT = None`
		`+ MQTT_USERNAME = None`
		`+ MQTT_PASSWORD = None`
		`+ MQTT_CA_CERTS = None`
		`+ MQTT_CERTFILE = None`
		`+ MQTT_KEYFILE = None`
		`+ MQTT_CERT_REQS = None`
		`+ MQTT_TLS_VERSION = None`
		`+ MQTT_CIPHERS = None`
		`+`
		`# Settings for Stomp message sending`
		`STOMP_NOTIFICATIONS = False`
		`STOMP_BROKERS = []`

pagure/lib/model.py

file modified

		`@@ -627,6 +627,7 @@`
		`"issues_default_to_private": False,`
		`"fedmsg_notifications": True,`
		`"stomp_notifications": True,`
		`+ "mqtt_notifications": True,`
		`"pull_request_access_only": False,`
		`"notify_on_pull-request_flag": False,`
		`"notify_on_commit_flag": False,`

pagure/lib/notify.py

file modified

+33 -1

		`@@ -98,12 +98,44 @@`
		`except Exception:`
		`_log.exception("Error sending stomp message")`

		`-`
		`def blinker_publish(topic, message):`
		`_log.info("Sending blinker signal to: pagure - topic: %s", topic)`
		`ready = blinker.signal("pagure")`
		`ready.send("pagure", topic=topic, message=message)`

		`+ def mqtt_publish(topic, message):`
		`+ """ Try to publish a message on a MQTT message bus. """`
		`+ if not pagure_config.get("MQTT_NOTIFICATIONS", True):`
		`+ return`
		`+ # We catch Exception if we want :-p`
		`+ # pylint: disable=broad-except`
		`+ # Ignore message about mqtt import`
		`+ # pylint: disable=import-error`
		`+ try:`
		`+ import paho.mqtt.client as mqtt`
		`+ import os`
		`+`
		`+ mqtt_host = pagure_config.get("MQTT_HOST")`
		`+ mqtt_port = pagure_config.get("MQTT_PORT")`
		`+ mqtt_username = pagure_config.get("MQTT_USERNAME")`
		`+ mqtt_pass = pagure_config.get("MQTT_PASSWORD")`
		`+ mqtt_ca_certs = pagure_config.get("MQTT_CA_CERTS")`
		`+ mqtt_certfile = pagure_config.get("MQTT_CERTFILE")`
		`+ mqtt_keyfile = pagure_config.get("MQTT_KEYFILE")`
		`+ mqtt_cert_reqs = pagure_config.get("MQTT_CERT_REQS", "ssl.CERT_REQUIRED")`
		`+ mqtt_tls_version = pagure_config.get("MQTT_TLS_VERSION","ssl.PROTOCOL_TLS")`
		`+ mqtt_ciphers = pagure_config.get("MQTT_CIPHERS")`
		`+`
		`+ client = mqtt.Client(os.uname()[1])`
		`+ client.tls_set(ca_certs=mqtt_ca_certs, certfile=mqtt_certfile, keyfile=mqtt_keyfile, cert_reqs=mqtt_cert_reqs, tls_version=mqtt_tls_version, cliphers=mqtt_ciphers)`
		`+ client.username_pw_set(mqtt_username, mqtt_pass)`
		`+ client.connect(mqtt_host, mqtt_port)`
		`+ client.publish(topic, json.dumps(message))`
		`+ client.disconnect()`
		`+`
		`+ except Exception:`
		`+ _log.exception("Error sending mqtt message")`
		`+`

		`def log(project, topic, msg, webhook=True):`
		`""" This is the place where we send notifications to user about actions`

jingjing commented 5 years ago

Fixes #3869

jingjing commented 5 years ago

@pingou Please help to review my changes.

pingou commented 5 years ago

From a quick look this seems fine, I'll need to test it and we'll need to fix jenkins :)

jingjing commented 5 years ago

@pingou
I test this function with the free online broker: iot.eclipse.org and it works.
If I want to add unit test or help to fix jenkins, what am I supposed to do ?

arrfab commented 5 years ago

That sounds awesome .. I can give it a try on a centos mqtt broker soon on our test instance. Wondering about the "format" used in "message" : is that already a json payload ?
And what about the topic to publish to ? I see "topic" but it's not coming from pagure conf ?

jingjing commented 5 years ago

@arrfab
I just wrote the function following the function format of the stomp, the topic and message should be a string.

arrfab commented 5 years ago

Yes, but then how is topic defined and where ?
WRT the message payload itself, wondering if we wouldn't have to add an option to let user decide if he wants to receive json at the mqtt level, and so a simple json.dumps(message) .
But that assumes that message is actually a dict, which I admit I don't know, nor had a look at :)
I put this on my TOTEST list, but worth also adding eventually a TLS_CERT option also , so that pagure can use a TLS cert for auth, instead of just username/pass.

Let me also have a look at this

pingou commented 5 years ago

Yes, but then how is topic defined and where ?

Each message "type" has a different topic, they are the same as for fedmsg: https://fedora-fedmsg.readthedocs.io/en/latest/topics.html#id571

But that assumes that message is actually a dict, which I admit I don't know, nor had a look at :)

It's currently a dict indeed and if mqtt doesn't convert it to JSON then I think we should do it, +1

arrfab commented 5 years ago

Ah, so something we can't configure then. So what about still having a MQTT_TOPIC variable in pagure.cfg that would let user create a parent topic in mqtt, and pagure would then be allowed in acl to write to all sub-topics .. example if parent topic in mqtt is set to git-prod, if someone pushes to pagure, the hooks would push to mqtt as /git-prod/pagure.git.receive (assuming that I understood the automatic topic, based on previous url for fedmsg)

arrfab commented 5 years ago

Apart from the mqtt parent topic suggestion, it would be probably be good to have some other options, like for broker using TLS auth, so then client.tls_set(mqtt_cacert, tls_version=2) should be adapted.
https://www.eclipse.org/paho/clients/python/docs (and tls_set() has the doc for certfile,keyfile)

pingou commented 5 years ago

@jingjing are you still working on this PR? Did you see the suggestions from @arrfab ?

rebased onto 1ef945f8f29d4b45f0c5953e12fc241d2ccd2319

5 years ago

jingjing commented 5 years ago

Hi @arrfab @pingou

Sorry that I missed the comment which arrfab added and have updated the PR according to comment, please help to review again, thank you!

pingou commented on line 32 of pagure/lib/notify.py 5 years ago

Where is message being defined? In the function definition it's msg.

Also, is it a dict or JSON? I believe the function gets passed a dict and from the discussion with @arrfab I thought the idea was to convert it to JSON.

pingou commented 5 years ago

@jingjing did you see my last comment?

pingou commented 5 years ago

@jingjing ping?

rebased onto edd8fd9

5 years ago

jingjing commented 5 years ago

@pingou
I've been busy recently, please help to review the updated PR.

pingou commented 5 years ago

@jingjing since you're busy, I'll take over the PR. To get jenkins to pass on this we need a few style changes.
Many thanks for your help on this, it's greatly appreciated, you did the lion's share of the work all I'll do is polishing :)

pingou commented 5 years ago

This is the follow up PR: https://pagure.io/pagure/pull-request/4132

jingjing commented 5 years ago

@pingou Thank you. It's my pleasure working on this project:-)

pingou commented 5 years ago

Commit 4eda2c4 fixes this pull-request