Welcome to the VerneMQ documentation! This is a reference guide for most of the available features and options of VerneMQ. The Getting Started guide might be a good entry point.

For a more general overview on VerneMQ and MQTT, you might want to start with the introduction.

For downloading VerneMQ see Downloads.

Install VerneMQ

Once you have downloaded the binary package, execute the following command to install VerneMQ:

sudo dpkg -i vernemq-<VERSION>.bionic.x86_64.deb

Verify your installation

You can verify that VerneMQ is successfully installed by running:

dpkg -s vernemq | grep Status

If VerneMQ has been installed successfully Status: install ok installed is returned.

Activate VerneMQ node

Once you've installed VerneMQ, start it on your node:

service vernemq start

Default Directories and Paths

The whereis vernemq command will give you a couple of directories:

whereis vernemq
vernemq: /usr/sbin/vernemq /usr/lib/vernemq /etc/vernemq /usr/share/vernemq

Next Steps

Now that you've installed VerneMQ, check out How to configure VerneMQ.

Installing VerneMQ

VerneMQ is a high-performance, distributed MQTT message broker. It scales horizontally and vertically on commodity hardware to support a high number of concurrent publishers and consumers while maintaining low latency and fault tolerance. To use it, all you need to do is install the VerneMQ package.

Choose your OS and follow the instructions:

• CentOS/RHEL

• Debian/Ubuntu

It is also possible to run VerneMQ using our Docker image:

• Docker

Starting VerneMQ

To start a VerneMQ broker, use the vernemq start command in your Shell:

vernemq start

A successful start will return no output. If there is a problem starting the broker, an error message is printed to STDERR.

To run VerneMQ with an attached interactive Erlang console:

vernemq console

A VerneMQ broker is typically started in console mode for debugging or troubleshooting purposes. Note that if you start VerneMQ in this manner, it is running as a foreground process that will exit when the console is closed.

You can close the console by issuing this command at the Erlang prompt:

q().

Once your broker has started, you can initially check that it is running with the vernemq ping command:

vernemq ping

The command will respond with pong if the broker is running or Node <NodeName> not responding to pings in case it’s not.

Every VerneMQ node has to be configured. Depending on the installation method and chosen platform the configuration file vernemq.conf resides at different locations. If VerneMQ was installed through a Linux package the default location for the configuration file is /etc/vernemq/vernemq.conf.

File Format

• A single setting is handled on one line.

• Lines are structured Key = Value

• Any line starting with # is a comment, and will be ignored

Minimal Quickstart configuration

You certainly want to try out VerneMQ right away. For that you could disable authentication like so:

• Set allow_anonymous = on

By default the vmq_acl authorization plugin is enabled and configured to allow publishing and subscribing to any topic, see here for more information.

Start a VerneMQ cluster node

Somtimes you need to configure a forwarding for ports (on a Mac for example):

This starts a new node that listens on 1883 for MQTT connections and on 8080 for MQTT over websocket connections. However, at this moment the broker won't be able to authenticate the connecting clients. To allow anonymous clients use the DOCKER_VERNEMQ_ALLOW_ANONYMOUS=on environment variable.

Autojoining a VerneMQ cluster

This allows a newly started container to automatically join a VerneMQ cluster. Assuming you started your first node like the example above you could autojoin the cluster (which currently consists of a single container 'vernemq1') like the following:

(Note, you can find the IP of a docker container using docker inspect <CONTAINER_NAME> | grep \"IPAddress\").

Checking cluster status

To check if the above containers have successfully clustered you can issue the vmq-admin command:

VerneMQ supports the WebSocket protocol out of the box. To be able to open a WebSocket connection to VerneMQ, you have to configure a WebSocket listener or Secure WebSocket listener in the vernemq.conf file first:

listener.ws.default = 127.0.0.1:9001

listener.wss.default = 127.0.0.1:9002

Keep in mind that you'll use MQTT-over-WebSocket, so you will need a Javascript library that implements the MQTT client behaviour. We have used the Eclipse Paho client as well as MQTT.js

You won't be able to open WebSocket connections on a base URL, always add the /mqtt path.

Maximum Client Id Size

Set the maximum size for client ids, MQTT v3.1 specifies a limit of 23 characters.

max_client_id_size = 23

This option default to 23.

Persistent Client Expiration

This option allows persistent clients (those with clean_session set to false) to be removed if they do not reconnect within a certain time frame.

The expiration period should be an integer followed by one of h, d, w, m, y for hour, day, week, month, and year; or never:

persistent_client_expiration = 1w

This option defaults to never.

Message Size Limit

Limit the maximum publish payload size in bytes that VerneMQ allows. Messages that exceed this size won't be accepted.

max_message_size = 0

Defaults to 0, which means that all valid messages are accepted. MQTT specification imposes a maximum payload size of 268435455 bytes.

Install VerneMQ

Once you have downloaded the binary package, execute the following command to install VerneMQ:

sudo yum install vernemq-<VERSION>.centos7.x86_64.rpm

or:

sudo rpm -Uvh vernemq-<VERSION>.centos7.x86_64.rpm

Activate VerneMQ node

Once you've installed VerneMQ, start it on your node:

service vernemq start

Verify your installation

You can verify that VerneMQ is successfully installed by running:

rpm -qa | grep vernemq

If VerneMQ has been installed successfully vernemq is returned.

Next Steps

Now that you've installed VerneMQ, check out How to configure VerneMQ.

Introduction and general setup

VerneMQ supports authentication and authorization using a number of popular databases and the below sections describe how to configure the different databases.

The database drivers are handled using the vmq_diversity plugin and it therefore needs to be enabled:

plugins.vmq_diversity = on

When using database based authentication/authorization the enabled-by-default file based authentication and authorization are most likely not needed and should be disabled:

plugins.vmq_passwd = off
plugins.vmq_acl = off

In order to use a database for authentication and authorization the database must be properly configured and the auth-data (username, clientid, password, acls) to be present. The following sections show some sample requests that can be used to insert such data.

While the handling of authentication differs among the different databases, the handling of ACLs is roughly identical and make use of a JSON array containing one or many ACL objects per configured client.

A minimal publish & subscribe ACL JSON object takes the following form:

General ACL

{
    "pattern": "a/+/c"
}

The pattern is a MQTT topic string that can contain MQTT wildcards, but also the template variables %m (mountpoint), %u (username), and %c (client id) which are automatically substituted with the auth data provided.

Publish ACL

The publish ACL makes it possible to control the maximum QoS and payload size that is allowed, and if the message is allowed to be retained.

{
    "pattern": "a/+/c",
    "max_qos": 2,
    "max_payload_size": 128,
    "allowed_retain": true
}

Moreover, the publish ACL makes it possible to modify the properties of a published message through specifying one or multiple modifiers. Please note that the modified message isn't re-validated by the ACL.

{
    "pattern": "a/+/c",
    "max_qos": 2,
    "max_payload_size": 128,
    "allowed_retain": true,
    "modifiers": {
        "topic": "new/topic",
        "payload": "new payload",
        "qos": 2,
        "retain": true,
        "mountpoint": "other-mountpoint"
    }
}

Subscribe ACL

The subscribe ACL makes it possible to control the maxium QoS a client is allowed to subscribe to.

{
    "pattern": "a/+/c",
    "max_qos": 2
}

Like the publish ACL, the subscribe ACL makes it possible to change the current subscription request by returning a custom set of topic/qos pairs. Please note that the modified subscription isn't re-validated by the ACL.

{
    "pattern": "a/+/c",
    "max_qos": 2,
    "modifiers": [
        ["new/topic/1", 1],
        ["new/topic/2", 1]
    ]
}

Password verification and hashing methods

When deciding on which database to use one has to consider which kind of password hashing and key derivation functions are available and required. Different databases provide different mechanisms, for example PostgreSQL provides the pgcrypto module which supports verifying hashed and salted passwords, while Redis has no such features. VerneMQ therefore also provides client-side password verification mechanisms such as bcrypt.

There is a trade-off between verifying passwords on the client-side versus on the server-side. Verifying passwords client-side of course means doing the computations on the VerneMQ broker and this takes away resources from other tasks such as routing messages. With hashing functions such as bcrypt which are designed specifically to be slow (proportional to the number of rounds) in order to make brute-force attacks infeasible, this can become a problem. For example, if verifying a password with bcrypt takes 0.5 seconds then on a single threaded core 2 verifications/second are possible and using 4 single threaded cores 8 verifications/second. So, the number of rounds/security paramenters have a direct impact on the max number of verifications/second and hence also the maximum arrival rate of new clients per second.

For each database it is specified which password verification mechanisms are available and if they are client-side or server-side.

PostgreSQL

To enable PostgreSQL authentication and authorization the following need to be configured in the vernemq.conf file:

vmq_diversity.auth_postgres.enabled = on
vmq_diversity.postgres.host = 127.0.0.1
vmq_diversity.postgres.port = 5432
vmq_diversity.postgres.user = vernemq
vmq_diversity.postgres.password = vernemq
vmq_diversity.postgres.database = vernemq_db
vmq_diversity.postgres.password_hash_method = crypt

PostgreSQL hashing methods:

The following SQL DDL must be applied, the pgcrypto extension is required if using the server-side crypt hashing method:

CREATE EXTENSION pgcrypto;
CREATE TABLE vmq_auth_acl
 (
   mountpoint character varying(10) NOT NULL,
   client_id character varying(128) NOT NULL,
   username character varying(128) NOT NULL,
   password character varying(128),
   publish_acl json,
   subscribe_acl json,
   CONSTRAINT vmq_auth_acl_primary_key PRIMARY KEY (mountpoint, client_id, username)
 );

To enter new ACL entries use a query similar to the following:

WITH x AS (
    SELECT
        ''::text AS mountpoint,
           'test-client'::text AS client_id,
           'test-user'::text AS username,
           '123'::text AS password,
           gen_salt('bf')::text AS salt,
           '[{"pattern": "a/b/c"}, {"pattern": "c/b/#"}]'::json AS publish_acl,
           '[{"pattern": "a/b/c"}, {"pattern": "c/b/#"}]'::json AS subscribe_acl
    ) 
INSERT INTO vmq_auth_acl (mountpoint, client_id, username, password, publish_acl, subscribe_acl)
    SELECT 
        x.mountpoint,
        x.client_id,
        x.username,
        crypt(x.password, x.salt),
        publish_acl,
        subscribe_acl
    FROM x;

CockroachDB

To enable PostgreSQL authentication and authorization the following need to be configured in the vernemq.conf file:

vmq_diversity.auth_cockroachdb.enabled = on
vmq_diversity.cockroachdb.host = 127.0.0.1
vmq_diversity.cockroachdb.port = 26257
vmq_diversity.cockroachdb.user = vernemq
vmq_diversity.cockroachdb.password = vernemq
vmq_diversity.cockroachdb.database = vernemq_db
vmq_diversity.cockroachdb.ssl = on
vmq_diversity.cockroachdb.password_hash_method = bcrypt

Notice that if the CockroachDB installation is secure, then TLS is required. If using an insecure installation without TLS, then vmq_diversity.cockroachdb.ssl can be set to off.

CockroachDB hashing methods:

The following SQL DDL must be applied:

CREATE TABLE vmq_auth_acl
 (
   mountpoint character varying(10) NOT NULL,
   client_id character varying(128) NOT NULL,
   username character varying(128) NOT NULL,
   password character varying(128),
   publish_acl json,
   subscribe_acl json,
   CONSTRAINT vmq_auth_acl_primary_key PRIMARY KEY (mountpoint, client_id, username)
 );

To enter new ACL entries use a query similar to the following, the example is for the bcrypt hashing method:

WITH x AS (
    SELECT
        ''::text AS mountpoint,
           'test-client1'::text AS client_id,
           'test-user1'::text AS username,
           '$2a$12$97PlnSsouvCV7HaxDPV80.EXfsKM4Fg7DAwWhSbGJ6O5CpNep20n2'::text AS hash,
           '[{"pattern": "a/b/c"}, {"pattern": "c/b/#"}]'::json AS publish_acl,
           '[{"pattern": "a/b/c"}, {"pattern": "c/b/#"}]'::json AS subscribe_acl
    )
INSERT INTO vmq_auth_acl (mountpoint, client_id, username, password, publish_acl, subscribe_acl)
    SELECT
        x.mountpoint,
        x.client_id,
        x.username,
        x.hash,
        publish_acl,
        subscribe_acl
    FROM x;

MySQL

For MySQL authentication and authorization configure the following in vernemq.conf:

vmq_diversity.auth_mysql.enabled = on
vmq_diversity.mysql.host = 127.0.0.1
vmq_diversity.mysql.port = 3306
vmq_diversity.mysql.user = vernemq
vmq_diversity.mysql.password = vernemq
vmq_diversity.mysql.database = vernemq_db
vmq_diversity.mysql.password_hash_method = password

MySQL hashing methods:

It should be noted that all the above options stores unsalted passwords which are vulnerable to rainbow table attacks, so the threat-model should be considered carefully when using these. Also note the methods marked with * are no longer considered secure hashes.

The following SQL DDL must be applied:

CREATE TABLE vmq_auth_acl
(
  mountpoint VARCHAR(10) NOT NULL,
  client_id VARCHAR(128) NOT NULL,
  username VARCHAR(128) NOT NULL,
  password VARCHAR(128),
  publish_acl TEXT,
  subscribe_acl TEXT,
  CONSTRAINT vmq_auth_acl_primary_key PRIMARY KEY (mountpoint, client_id, username)
)

To enter new ACL entries use a query similar to the following, the example uses PASSWWORD to for password hashing:

INSERT INTO vmq_auth_acl 
    (mountpoint, client_id, username, 
     password, publish_acl, subscribe_acl)
VALUES 
    ('', 'test-client', 'test-user', PASSWORD('123'), 
     '[{"pattern":"a/b/c"},{"pattern":"c/b/#"}]', 
     '[{"pattern":"a/b/c"},{"pattern":"c/b/#"}]');

MongoDB

For MongoDB authentication and authorization configure the following in vernemq.conf:

vmq_diversity.auth_mongodb.enabled = on
vmq_diversity.mongodb.host = 127.0.0.1
vmq_diversity.mongodb.port = 27017
# vmq_diversity.mongodb.login =
# vmq_diversity.mongodb.password = 
# vmq_diversity.mongodb.database =

MongoDB hashing methods:

Insert the ACL using the mongo shell or any software library. The passhash property contains the bcrypt hash of the clients password.

db.vmq_acl_auth.insert({
    mountpoint: '', 
    client_id: 'test-client', 
    username: 'test-user', 
    passhash: '$2a$12$WDzmynWSMRVzfszQkB2MsOWYQK9qGtfjVpO8iBdimTOjCK/u6CzJK', 
    publish_acl: [
        {pattern: 'a/b/c'}, 
        {pattern: 'a/+/d'}
    ], 
    subscribe_acl: [
        {pattern: 'a/#'}
    ]
})

Redis

For Redis authentication and authorization configure the following in vernemq.conf:

vmq_diversity.auth_redis.enabled = on
vmq_diversity.redis.host = 127.0.0.1
vmq_diversity.redis.port = 6379
# vmq_diversity.redis.password = 
# vmq_divserity.redis.database = 0

Redis hashing methods:

Insert the ACL using the redis-cli shell or any software library. The passhash property contains the bcrypt hash of the clients password. The key is an encoded JSON array containing the mountpoint, username, and client id. Note that no spaces are allowed between the array items.

SET "[\"\",\"test-client\",\"test-user\"]" "{\"passhash\":\"$2a$12$WDzmynWSMRVzfszQkB2MsOWYQK9qGtfjVpO8iBdimTOjCK/u6CzJK\",\"subscribe_acl\":[{\"pattern\":\"a/+/c\"}]}"

Note, currently bcrypt version 2a (prefix $2a$) is supported.

The VerneMQ HTTP listener is used to serve various VerneMQ subsystems such as:

By default it runs on port 8888. To disable the HTTP listener or change the port, adapt the configuration in vernemq.conf:

Listeners specify on which IP address and port VerneMQ should accept new incoming connections. Depending on the chosen transport (TCP, SSL, WebSocket) different configuration parameters have to be provided. VerneMQ allows to write the listener configurations in a hierarchical manner, enabling very flexible setups. VerneMQ applies reasonable defaults on the top level, which can be of course overridden if needed.

These are the only default parameters that are applied for all transports, and the only one that are of interest for plain TCP and WebSocket listeners.

These global defaults can be overridden for a specific transport protocol listener.tcp.CONFIG = VAL, or even for a specific listener listener.tcp.LISTENER.CONFIG = VAL. The placeholder LISTENER is freely chosen and is only used as a reference for further configuring this particular listener.

Mountpoints

Normally, an MQTT broker hosts one single topic tree. This means that all topics are accessible to all publishers and subscribers (limited by the ACLs you configured, of course). Mountpoints are a way to host multiple topic trees in a single broker. They are completely separated and clients with different topic trees cannot publish messages to each other. This could be useful if you provide MQTT services to multiple separated use cases/verticals or clients, with a single broker. Note that mountpoints are configured via different listeners. As a consequence, the MQTT clients will have to connect to a specific port to connect to a specific topic space (mountpoint).

The mountpoints can be configured on the protocol level or configurred or overridden on the specific listener level.

Allowed protocol versions

Since VerneMQ 1.5.0 it is possible to configure which MQTT protocol versions as listener will accept.

VerneMQ supports MQTT 3.1, 3.1.1, and 5.0 (since VerneMQ 1.6.0). To allow these protocol versions, set:

Here 3,4,5 are the protocol level versions corresponding to MQTT 3.1, 3.1.1 and 5.0 respectively. The default value is 3,4 thus allowing MQTT 3.1 and 3.1.1, while MQTT 5.0 is disabled.

Sample Config

Listen on TCP port 1883 and for WebSocket Connections on port 8888:

An additional listener can be added by using a different name. In the example above the name equals to default and can be used for further configuring this particular listener. The following example demonstrates how an additional listener is defined as well as how the maximum number of connections can be limited for this listener:

PROXY protocol

VerneMQ listeners can be configured to accept connections from a proxy server that supports the PROXY protocol. This enables VerneMQ to retrieve peer information such as source IP/Port but also PROXY Version 2 protocol TLS client certificate details if the proxy was used to terminate TLS.

To enable the PROXY protocol for tcp listeners use listener.tcp.proxy_protocol = on or for a specific listener use listener.tcp.LISTENER.proxy_protocol = on.

If client certificates are used you can set listener.tcp.proxy_protocol_use_cn_as_username = on which will overwrite the MQTT username set by the client with the common name from the client certificate before authentication and authorization is performed.

Sample SSL Config

Accepting SSL connections on port 8883:

If you want to use client certificates to authenticate your clients you have to set the following option:

If you use client certificates and want to use the certificates CN value as a username you can set:

Both options require_certificate and use_identity_as_username default to off.

The same configuration options can be used for securing WebSocket connections, just use wss as the protocol identifier e.g. listener.wss.require_certificate.

Retry Interval

Set the time in seconds after a QoS=1 or QoS=2 message has been sent that VerneMQ will wait before retrying when no response is received.

This option default to 20 seconds.

Inflight Messages

This option defines the maximum number of QoS 1 or 2 messages that can be in the process of being transmitted simultaneously.

Defaults to 20 messages, use 0 for no limit. The inflight window serves as a protection for sessions, on the incoming side.

Load Shedding

The maximum number of messages to hold in the queue above those messages that are currently in flight. Defaults to 1000. Set to -1 for no limit. This option protects a client session from overload by dropping messages (of any QoS).

Defaults to 1000 messages, use -1 for no limit. This parameter was named max_queued_messages in 0.10.*. Note that 0 will totally block message delivery from any queue!

This option specifies the maximum number of QoS 1 and 2 messages to hold in the offline queue.

Defaults to 1000 messages, use -1 for no limit, use 0 if no messages should be stored.

In contrast to the session based inflight window, max_online_messages and max_offline_messages serves as a protection of queues, on the outgoing side.

Authentication

VerneMQ comes with a simple file-based password authentication mechanism which is enabled by default. If you don't need this it can be disabled by setting:

Per default VerneMQ doesn't accept any client that hasn't been configured using vmq-passwd. If you want to change this and accept any client connection you can set:

In a production setup we recommend to use the provided password based authentication mechanism or implement your own authentication plugins.

VerneMQ periodically checks the specified password file.

The check interval defaults to 10 seconds and can also be defined in the vernemq.conf.

Setting the password_reload_interval = 0 disables automatic reloading.

Manage Password Files for VerneMQ

vmq-passwd is a tool for managing password files for the VerneMQ broker. Usernames must not contain ":", passwords are stored in similar format to .

How to use vmq-passwd

Options

-c

Creates a new password file. If the file already exists, it will be overwritten.

-D

Deletes the specified user from the password file.

-U

This option can be used to upgrade/convert a password file with plain text passwords into one using hashed passwords. It will modify the specified file. It does not detect whether passwords are already hashed, so using it on a password file that already contains hashed passwords will generate new hashes based on the old hashes and render the password file unusable. Note, with this option neither usernames or passwords may contain ":".

passwordfile

The password file to modify.

username

The username to add/update/delete.

Examples

Add a user to a new password file: (you can choose an arbitrary name for the password file, it only has to match the configuration in the VerneMQ configuration file).

Delete a user from a password file

Acknowledgements

The original version of vmq-passwd was developed by Roger Light (roger@atchoo.org).

vmq-passwd includes :

• software developed by the [OpenSSL
• cryptographic software written by Eric Young

  (eay@cryptsoft.com)

• software written by Tim Hudson (tjh@cryptsoft.com)

Authorization

VerneMQ comes with a simple ACL based authorization mechanism which is enabled by default. If you don't need this it can be disabled by setting:

VerneMQ periodically checks the specified ACL file.

The check interval defaults to 10 seconds and can also be defined in the vernemq.conf.

Setting the acl_reload_interval = 0 disables automatic reloading.

Managing the ACL entries

Topic access is added with lines of the format:

The access type is controlled using read or write. If not provided then read an write access is granted for the topic. The topic can use the MQTT subscription wildcards + or #.

The first set of topics are applied to all anonymous clients (assuming allow_anonymous = on). User specific ACLs are added after a user line as follows (this is the username not the client id):

It is also possible to define ACLs based on pattern substitution within the the topic. The form is the same as for the topic keyword, but using pattern as the keyword.

The patterns available for substitution are:

• %c to match the client id of the client

• %u to match the username of the client

The substitution pattern must be the only text for that level of hierarchy. Pattern ACLs apply to all users even if the user keyword has previously been given.

Example:

Simple ACL Example

Anonymous users are allowed to

• publish & subscribe to topic bar.

• publish to topic foo.

• subscribe to topic all.

User john is allowed to

• publish & subscribe to topic foo.

• subscribe to topic baz.

• publish to topic all.

Many aspects of VerneMQ can be extended using plugins. The standard VerneMQ package comes with several official plugins. You can show the enabled & running plugins via:

The command above displays all the enabled plugins together with the hooks they implement:

Enable a plugin

This enables the ACL plugin. Because the vmq_acl plugin is already started the above command won't succeed. In case the plugin sits in an external directory you must also to provide the --path=PathToPlugin.

Disable a plugin

Persisting plugins

To make a plugin start when VerneMQ starts they need to be configured in the main vernemq.conf file.

The general syntax to enable a plugin is to add a line like plugins.pluginname = on, using the vmq_passwd plugin as an example:

And if the plugin is external the path can be specified like this:

Plugin specific settings can be configured via myplugin.somesetting = value, like:

See the vernemq.conf file for details.

Console Logging

Where should VerneMQ emit the default console log messages (which are typically at info severity):

VerneMQ defaults to log the console messages to a file, which can specified by:

This option defaults to /var/log/vernemq/console.log for Ubuntu, Debian, RHEL and Docker installs.

The default console logging level info could be setting one of the following:

Error Logging

VerneMQ log error messages by default. One can change the default behaviour by setting:

VerneMQ defaults to log the error messages to a file, which can specified by:

This option defaults to /var/log/vernemq/error.log for Ubuntu, Debian, RHEL and Docker installs.

Crash Logging

VerneMQ log crash messages by default. One can change the default behaviour by setting:

VerneMQ defaults to log the crash messages to a file, which can specified by:

This option defaults to /var/log/vernemq/crash.log for Ubuntu, Debian, RHEL and Docker installs.

The maximum sizes in bytes of inidividual messages in the crash log defaults to 64KB but can be specified by:

VerneMQ rotate crash logs. By default, the crash log file is rotated at midnight or when the size exceeds 10MGB. This behaviour can be changed by setting:

The default number of rotated log files is 5 and can be set with the option:

SysLog

VerneMQ supports logging to SysLog, enable it by setting:

Logging to SysLog is disabled by default.

A shared subscription is a mechanism for distributing messages to a set of subscribers to shared subscription topic, such that each message is received by only one subscriber. This contrasts with normal subscriptions where each subscriber will receive a copy of the published message.

A shared subscription is on the form $share/sharename/topic and subscribers to this topic will receive messages published to the topic topic. The messages will be distributed according to the defined distribution policy.

Configuration

Currently three message distribution policies for shared subscriptions are supported: prefer_local, random and local_only. Under the random policy messages will be published to a random member of the shared subscription, if any exist. Under the prefer_local policy messages will be delivered to a random node-local member of the shared subscription, if none exist, the message will be delivered to a random member of the shared subscription on a remote cluster node. Under the local_only policy message will be delivered to a random node-local member of the shared subscription.

When a messages is being delivered to subscribers of a shared subscription, the message will be delivered to an online subscriber if possible, otherwise the message will be delivered to an offline subscriber.

Examples

Subscriptions Note: When subscribing to a shared topic, make sure to escape the $

So, for dash or bash shells

Publishing Note: When publishing to a shared topic, do not include the prefix $share/group/ as part of the publish topic name

Sometimes consumers get overwhelmed by the number of messages they receive. VerneMQ can load balance between multiple consumer instances subscribed to the same topic with the same ClientId.

Enabling Session Balancing

To enable session balancing, activate the following two settings in vernemq.conf

VerneMQ uses the Erlang distribution mechanism for most inter-node communication. VerneMQ identifies other machines in the cluster using Erlang identifiers (e.g. VerneMQ@10.9.8.7). Erlang resolves these node identifiers to a TCP port on a given machine via the Erlang Port Mapper daemon (epmd) running on each cluster node.

By default, epmd binds to TCP port 4369 and listens on the wildcard interface. For inter-node communication, Erlang uses an unpredictable port by default; it binds to port 0, which means the first available port.

For ease of firewall configuration, VerneMQ can be configured to instruct the Erlang interpreter to use a limited range of ports. For example, to restrict the range of ports that Erlang will use for inter-Erlang node communication to 6000-7999, add the following lines to vernemq.conf on each VerneMQ node:

erlang.distribution.port_range.minimum = 6000
erlang.distribution.port_range.maximum = 7999

The settings above are only used for distributing subscription updates and maintenance messages. For distributing the 'real' MQTT messages the proper vmq listener must be configured in the vernemq.conf.

listener.vmq.clustering = 0.0.0.0:44053

Attributions:

This section, "VerneMQ Inter-node Communication", is a derivative of Security and Firewalls by Riak, used under Creative Commons Attribution 3.0 Unported License.

There are a couple of hidden options you can set in the vernemq.conf file. Hidden means that you have to add and set the value explicitly. Hidden options still have default values. Changing them should be considered advanced, possibly with the exception of setting a max_message_rate.

Queue Deliver mode

Specify how the queue should deliver messages when multiple sessions are allowed. In case of fanout all the attached sessions will receive the message, in case of balance an attached session is choosen randomly.

queue_deliver_mode = balance

Queue Type

Specify how queues should process messages, either the fifo or lifo way. Default is fifo.

queue_type = fifo

Max Message Rate

Specifies the maximum incoming publish rate per session per second. Depending on the underlying network buffers this rate isn't enforced. Defaults to 0, which means no rate limits apply. Setting to a value of 2 limits any publisher to 2 messages per second, for instance.

max_message_rate = 2

Max Drain Time

Due to the eventual consistent nature of the subscriber store it is possible that during queue migration messages still arrive on the old cluster node. This parameter enables to compensate this by keeping the queue around for some time (in seconds) after it was migrated to the other cluster node.

max_drain_time = 20

Max Msgs per Drain Step

Specifies the number of messages that are delivered to the remote node per drain step. A large value will provide a faster migration of a queue, but increases the waste of bandwidth in case the migration fails.

max_msgs_per_drain_step = 1000

Default Reg View

Allows to select a new default reg_view. A reg_view is a pre-defined way to route messages. Multiple views can be loaded and used, but one has to be selected as a default. The default routing is vmq_reg_trie, i.e. routing via the built-in trie data structure.

vmq_reg_view = "vmq_reg_trie"

Reg Views

A list of views that are started during startup. It's only used in plugins that want to choose dynamically between routing reg_views.

reg_views = "[vmq_reg_trie]"

Outgoing Clustering Buffer Size

An integer specifying how many bytes are buffered in case the remote node is not available. Default is 10000

outgoing_clustering_buffer_size = 15000

This section elaborates how a VerneMQ cluster deals with network partitions (aka. netsplit or split brain situation). A netsplit is mostly the result of a failure of one or more network devices resulting in a cluster where nodes can no longer reach each other.

VerneMQ is able to detect a network partition, and by default it will stop serving CONNECT, PUBLISH, SUBSCRIBE, and UNSUBSCRIBE requests. A properly implemented client will always resend unacked commands and messages are therefore not lost (QoS 0 publishes will be lost). However, the time window between the network partition and the time VerneMQ detects the partition much can happen. Moreover, this time frame will be different on every participating cluster node. In this guide we're referring to this time frame as the Window of Uncertainty.

Possible Scenario for Message Loss:

VerneMQ follows an eventually consistent model for storing and replicating the subscription data. This also includes retained messages.

Due to the eventually consistent data model it is possible that during the Window of Uncertainty a publish won't take into account a subscription made on a remote node (in another partition). Obviously, VerneMQ can't deliver the message in this case. The same holds for delivering retained messages to remote subscribers.

last will messages that are triggered during the Window of Uncertainty will be delivered to the reachable subscribers. Currently during a netsplit, but after the Window of Uncertainty last will messages will be lost.

Possible Scenario for Duplicate Clients:

Normally, client registration is synchronized using an elected leader node for the given client id. Such a synchronization removes the race condition between multiple clients trying to connect with the same client id on different nodes. However, during the Window of Uncertainty it is currently possible that VerneMQ fails to disconnect a client connected to a different node. Although this scenario sounds like artificially crafted it is possible to end up with duplicate clients connected to the cluster.

Recovering from a Netsplit

As soon as the partition is healed, and connectivity reestablished, the VerneMQ nodes replicate the latest changes made to the subscription data. This includes all the changes 'accidentally' made during the Window of Uncertainty. Using Dotted Version Vectors VerneMQ ensures that convergence regarding subscription data and retained messages is eventually reached.

On every VerneMQ node you'll find the vmq-admin command line tool in the release's bin directory. It has different sub-commands that let you check for status, start and stop listeners, re-configure values and a couple of other administrative tasks.

Inspecting sessions

VerneMQ comes with powerful tools for inspecting the state of MQTT sessions. To list current MQTT sessions simply invoke vmq-admin session show:

$ vmq-admin session show
+---------+---------+----------+---------+---------+---------+
|client_id|is_online|mountpoint|peer_host|peer_port|  user   |
+---------+---------+----------+---------+---------+---------+
| client2 |  true   |          |127.0.0.1|  37098  |undefined|
| client1 |  true   |          |127.0.0.1|  37094  |undefined|
+---------+---------+----------+---------+---------+---------+

To see detailed information about the command see vmq-admin session show --help.

The command is able to show a lot of different information about a client, for example the client id, the peer host and port if the client is online or offline and much more, see vmq-admin session show --help for details. Further the information can also be used to filter information which is very helpful when wanting to narrow down the information to a single client.

A sample query which lists only the node where the client session exists and if the client is online would look like the following:

$ vmq-admin session show --node --is_online --client_id=client1
+---------+--------------+
|is_online|     node     |
+---------+--------------+
|  true   |dev2@127.0.0.1|
+---------+--------------+

More examples

Listing the clients and the subscriptions one can do the following:

$ vmq-admin session show --topic --client_id
+---------+----------------+
|client_id|     topic      |
+---------+----------------+
| client2 |some/other/topic|
| client1 |some/other/topic|
| client1 |   some/topic   |
+---------+----------------+

And to list only the clients subscribed to the topic some/topic:

$ vmq-admin session show --topic --client_id --topic=some/topic
+---------+----------+
|client_id|  topic   |
+---------+----------+
| client1 |some/topic|
+---------+----------+

To figure out when the queue for a persisted session (clean_session=false) was created and when the client last connected one can use the --queue_started_at and --session_started_at to list the POSIX timestamps (in microseconds):

$ vmq-admin session show --client_id=client1 --queue_started_at --session_started_at
+----------------+------------------+
|queue_started_at|session_started_at|
+----------------+------------------+
| 1549379963575  |  1549379974905   |
+----------------+------------------+

Besides the examples above it is also possible to inspect the number of online or offline messages as well as their payloads and much more. See vmq-admin session show --help for an exhaustive list of all the available options.

Managing sessions

VerneMQ also supports disconnecting clients and reauthorizing client subscriptions. To disconnect a client and cleanup store messages and remove subscriptions one can invoke:

$ vmq-admin session disconnect client-id=client1 --cleanup

See vmq-admin session disconnect --help for more options and details.

To reauthorize subscriptions for a client issue the following command:

$ vmq-admin session reauthorize username=username client-id=client1
Unchanged

This works by reapplying the logic in any installed auth_on_subscribe or auth_on_subscribe_m5 plugin hooks to check the validity of the existing topics and removing those that are no longer allowed. In the example above the reauthorization of the client subscriptions resulted in no changes.

Bridges are a non-standard way, although kind of a de-facto standard among MQTT broker implementations, to connect two different MQTT brokers to eachother. This allows for example that a topic tree of a remote broker becomes part of the topic tree on the local broker. VerneMQ supports plain TCP connections as well as SSL connections.

Enabling the bridge functionality

in VerneMQ the bridge is distributed with VerneMQ as a plugin and is not enabled by default. After configuring the bridge as described below, make sure to enable the plugin by setting:

plugins.vmq_bridge = on

See Managing plugins for more information on working with plugins.

When the plugin is enabled a simple status interface is available:

$ vmq-admin bridge show
+-----------------+-----------+----------+-------------------+
|   endpoint      |buffer size|buffer max|buffer dropped msgs|
+-----------------+-----------+----------+-------------------+
|192.168.1.10:1883|     0     |    0     |         0         |
+-----------------+-----------+----------+-------------------+

Sample MQTT Bridge

Setup a bridge to a remote broker:

vmq_bridge.tcp.br0 = 192.168.1.12:1883

Different connection parameters can be set:

# use a clean session (defaults to 'off')
vmq_bridge.tcp.br0.cleansession = off | on

# set the client id (defaults to 'auto', which generates one)
vmq_bridge.tcp.br0.client_id = auto | my_bridge_client_id

# set keepalive interval (defaults to 60 seconds)
vmq_bridge.tcp.br0.keepalive_interval = 60

# set the username and password for the bridge connection
vmq_bridge.tcp.br0.username = my_bridge_user
vmq_bridge.tcp.br0.password = my_bridge_pwd

# set the restart timeout (defaults to 10 seconds)
vmq_bridge.tcp.br0.restart_timeout = 10

# VerneMQ indicates other brokers that the connection
# is established by a bridge instead of a normal client.
# This can be turned off if needed:
vmq_bridge.tcp.br0.try_private = off

# Set the maximum number of outgoing messages the bridge will buffer
# while not connected to the remote broker. Messages published while
# the buffer is full are dropped. A value of 0 means buffering is
# disabled.
vmq_bridge.tcp.br0.max_outgoing_buffered_messages = 100

Define the topics the bridge should incorporate in its local topic tree (by subscribing to the remote), or the topics it should export to the remote broker (by publishing to the remote). We share a similar configuration syntax to that used by the Mosquitto broker:

topic [[[ out | in | both ] qos-level] local-prefix remote-prefix]

topic defines a topic pattern that is shared between the two brokers. Any topics matching the pattern (which may include wildcards) are shared. The second parameter defines the direction that the messages will be shared in, so it is possible to import messages from a remote broker using in, export messages to a remote broker using out or share messages in both directions. If this parameter is not defined, VerneMQ defaults to out. The QoS level defines the publish/subscribe QoS level used for this topic and defaults to 0. (Source: mosquitto.conf)

The local-prefix and remote-prefix can be used to prefix incoming or outgoing publish messages.

A simple example:

# share messages in both directions and use QoS 1
vmq_bridge.tcp.br0.topic.1 = /demo/+ both 1

# import the $SYS tree of the remote broker and
# prefix it with the string 'remote'
vmq_bridge.tcp.br0.topic.2 = $SYS/* in remote

Sample MQTT Bridge that uses SSL/TLS

SSL bridges support the same configuration parameters as TCP bridges, but need further instructions for handling the SSL specifics:

# define the CA certificate file or the path to the
# installed CA certificates
vmq_bridge.ssl.br0.cafile = cafile.crt
#or
vmq_bridge.ssl.br0.capath = /path/to/cacerts

# if the remote broker requires client certificate authentication
vmq_bridge.ssl.br0.certfile = /path/to/certfile.pem
# and the keyfile
vmq_bridge.ssl.br0.keyfile = /path/to/keyfile

# disable the verification of the remote certificate (defaults to 'off')
vmq_bridge.ssl.br0.insecure = off

# set the used tls version (defaults to 'tlsv1.2')
vmq_bridge.ssl.br0.tls_version = tlsv1.2

VerneMQ uses Google's LevelDB as a fast storage backend for messages and subscriber information. Each VerneMQ node runs its own embedded LevelDB store.

Configuration of LevelDB memory

There's not much you need to know about LevelDB and VerneMQ. One really important thing to note is that LevelDB manages its own memory. This means that VerneMQ will not allocate and free memory for LevelDB. Instead you'll have to configure a configuration value in vernemq.conf that tells LevelDB how much memory it can use up.

Configuring LevelDB memory:

leveldb.maximum_memory.percent = 20

To list the retained messages simply invoke vmq-admin retain show:

$ vmq-admin retain show
+------------------+----------------+
|     payload      |     topic      |
+------------------+----------------+
| a-third-message  | a/third/topic  |
|some-other-message|some/other/topic|
|    a-message     |   some/topic   |
|    a-message     | another/topic  |
+------------------+----------------+

Besides listing the retained messages it is also possible to filter them:

$ vmq-admin retain show --payload --topic=some/topic
+---------+
| payload |
+---------+
|a-message|
+---------+

In the above example we list only the payload for the topic some/topic.

Another example where all topics are list with retained messages with a specific payload:

$ vmq-admin retain show --payload a-message --topic
+-------------+
|    topic    |
+-------------+
| some/topic  |
|another/topic|
+-------------+

See the full set of options and documentation by invoking vmq-admin retain show --help.

You can dynamically re-configure most of VerneMQ's settings on a running node by using the vmq-admin set command.

The following config values can be handled dynamically:

allow_anonymous
topic_alias_max_broker
receive_max_broker
vmq_acl.acl_file
graphite_host
vmq_acl.acl_reload_interval
graphite_enabled
queue_type
suppress_lwt_on_session_takeover
max_message_size
vmq_passwd.password_file
graphite_port
max_client_id_size
upgrade_outgoing_qos
max_message_rate
graphite_interval
allow_multiple_sessions
systree_enabled
max_last_will_delay
retry_interval
receive_max_client
max_offline_messages
max_online_messages
max_inflight_messages
allow_register_during_netsplit
vmq_passwd.password_reload_interval
topic_alias_max_client
systree_interval
allow_publish_during_netsplit
coordinate_registrations
remote_enqueue_timeout
persistent_client_expiration
allow_unsubscribe_during_netsplit
graphite_include_labels
shared_subscription_policy
queue_deliver_mode
allow_subscribe_during_netsplit

Setting a value for the local node

Let's change the max_client_id_size as an example. (We might have noticed that some clients can't login because their client ID is too long, but we don't want to restart the broker for that). Note that you can also set multiple values with the same command.

vmq-admin set max_client_id_size=45

Setting a value for an arbitrary cluster node

vmq-admin set max_client_id_size=45 --node=VerneMQ20@192.168.1.20

Setting a value for all cluster nodes

vmq-admin set max_client_id_size=45 --all

Show current VerneMQ config values

For the local node

You can show one or multiple values in a simple table:

vmq-admin show max_client_id_size retry_interval

+----------------------+------------------+--------------+
|         node         |max_client_id_size|retry_interval|
+----------------------+------------------+--------------+
|VerneMQ20@192.168.1.50|        28        |      20      |
+----------------------+------------------+--------------+

`

For an arbitrary node

vmq-admin show max_client_id_size retry_interval --node VerneMQ20@192.168.1.20

For all cluster nodes

vmq-admin show max_client_id_size retry_interval --all

+----------------------+------------------+--------------+
|         node         |max_client_id_size|retry_interval|
+----------------------+------------------+--------------+
|VerneMQ30@192.168.1.30|        33        |      20      |
|VerneMQ40@192.168.1.40|        33        |      20      |
|VerneMQ10@192.168.1.10|        33        |      20      |
|VerneMQ50@192.168.1.50|        33        |      20      |
|VerneMQ20@192.168.1.20|        28        |      20      |
+----------------------+------------------+--------------+

You can configure as many listeners as you wish in the vernemq.conf file. In addition to this, the vmq-admin listener command let's you configure, start, stop and delete listeners on the fly. Those can be MQTT, WebSocket or Cluster listeners, in the command line output they will be tagged mqtt, ws or vmq accordingly.

Status of all listeners

vmq-admin listener show

    +----+-------+------------+-----+----------+---------+
    |type|status |     ip     |port |mountpoint|max_conns|
    +----+-------+------------+-----+----------+---------+
    |vmq |running|192.168.1.50|44053|          |  30000  |
    |mqtt|running|192.168.1.50|1883 |          |  30000  |
    +----+-------+------------+-----+----------+---------+
`

Starting a new listener

vmq-admin listener start address=192.168.1.50 port=1884 --mountpoint /test --nr_of_acceptors=10 --max_connections=1000

This will start an MQTT listener on port 1884 and IP address 192.168.1.50. If you want to start a WebSocket listener, just tell VerneMQ by adding the --websocket flag. There are more options, mainly for configuring SSL (use vmq-admin listener start --help).

You can isolate client connections accepted by a certain listener from other clients by setting a mountpoint.

To start an MQTT listener using defaults, just set the port and IP address as a minimum.

Stopping a listener

vmq-admin listener stop address=192.168.1.50 port=1884

You can add the -k or --kill_sessions switch to that command. This will disconnect all client connections setup by that listener. In combination with a mountpoint, this can be useful for terminating clients for a specific application, or to force re-connects to another cluster node (to prepare for a cluster leave for your node).

Restarting a stopped listener

vmq-admin listener restart address=192.168.1.50 port=1884

Deleting a stopped listener

vmq-admin listener delete address=192.168.1.50 port=1884

The VerneMQ HTTP API is enabled by default and installs an HTTP handler on http://localhost:8888/api/v1. To read more about configuring the HTTP listener, see HTTP Listener Configuration. You can configure a HTTP listener, or a HTTPS listener to serve the HTTP API v1.

Managing API keys

The VerneMQ HTTP API uses basic authentication where an API key is passed as the username and the password is left empty. So the first step to us the HTTP API is to create an API key:

$ vmq-admin api-key create
JxctXkZ1OTVnlwvguSCE9KtujacMkOLF

The key is persisted and available on all cluster nodes.

To list existing keys do:

$ vmq-admin api-key show
+--------------------------------+
|              Key               |
+--------------------------------+
|JxctXkZ1OTVnlwvguSCE9KtujacMkOLF|
+--------------------------------+

To add an API key of your own choosing, do:

vmq-admin api-key add key=mykey

To delete an API key do:

vmq-admin api-key delete key=JxctXkZ1OTVnlwvguSCE9KtujacMkOLF

API usage

The VerneMQ HTTP API is a wrapper over the vmq-admin CLI tool, and anything that can be done using vmq-admin can be done using the HTTP API. Note that the HTTP API is therefore subject to any changes made to the vmq-admin tools and their flags & options structure. All requests are performed doing a HTTP GET and if no errors occurred an HTTP 200 OK code is returned with a possible non-empty JSON payload.

The API is using basic auth where the API key is passed as the username. An example using curl would look like this:

curl "http://JxctXkZ1OTVnlwvguSCE9KtujacMkOLF@localhost:8888/api/v1/session/show"

The mapping between vmq-admin and the HTTP API is straightforward, and if one is already familiar with how the vmq-admin tool works, working with the API should be easy. The mapping works such that the command part of a vmq-admin invocation is turned into a path, and the options and flags are turned into the query string.

A mandatory parameter like the client-id in the vmq-admin session disconnect client-id=myclient command should be translated as: ?client-id=myclient.

An optional flag like --cleanup in the vmq-admin session disconnect client-id=myclient --cleanup command should be translated as: &--cleanup

Let's look at the cluster join command as an example, which looks like this:

vmq-admin cluster join discovery-node=NodeB@10.0.0.2

This turns into a GET request:

GET /api/v1/cluster/join?discovery-node=NodeB@10.0.0.2

To test, run it with curl:

curl "http://JxctXkZ1OTVnlwvguSCE9KtujacMkOLF@localhost:8888/api/v1/cluster/join?discovery-node=NodeB@10.0.0.2"

And the returned response would look like:

{
    "text": "Done",
    "type": "text"
}

Below are some other examples.

Get cluster status information

Request:

GET /api/v1/cluster/show

Curl:

curl "http://JxctXkZ1OTVnlwvguSCE9KtujacMkOLF@localhost:8888/api/v1/cluster/show"

Response:

{
   "type" : "table",
   "table" : [
      {
         "Running" : true,
         "Node" : "VerneMQ@127.0.0.1"
      }
   ]
}

Retrieve session information

Request:

GET /api/v1/session/show

Curl:

curl "http://JxctXkZ1OTVnlwvguSCE9KtujacMkOLF@localhost:8888/api/v1/session/show"

Response:

{
   "type" : "table",
   "table" : [
      {
         "user" : null,
         "client_id" : "mosqsub/11690-algol",
         "peer_host" : "localhost",
         "pid" : "<0.551.0>",
         "peer_port" : 50964
      },
      {
         "peer_host" : "localhost",
         "pid" : "<0.557.0>",
         "peer_port" : 50996,
         "user" : null,
         "client_id" : "mosqsub/11767-algol"
      }
   ]
}

List all installed listeners

Request:

GET /api/v1/listener/show

Curl:

curl "http://JxctXkZ1OTVnlwvguSCE9KtujacMkOLF@localhost:8888/api/v1/listener/show"

Response:

{
   "type" : "table",
   "table" : [
      {
         "max_conns" : 10000,
         "port" : "8888",
         "mountpoint" : "",
         "ip" : "127.0.0.1",
         "type" : "http",
         "status" : "running"
      },
      {
         "status" : "running",
         "max_conns" : 10000,
         "port" : "44053",
         "mountpoint" : "",
         "ip" : "0.0.0.0",
         "type" : "vmq"
      },
      {
         "max_conns" : 10000,
         "port" : "1883",
         "mountpoint" : "",
         "ip" : "127.0.0.1",
         "type" : "mqtt",
         "status" : "running"
      }
   ]
}

Retrieve plugin information

Request:

GET /api/v1/plugin/show

Curl:

curl "http://JxctXkZ1OTVnlwvguSCE9KtujacMkOLF@localhost:8888/api/v1/plugin/show"

Response:

    {
   "type" : "table",
   "table" : [
      {
         "Hook(s)" : "auth_on_register\n",
         "Plugin" : "vmq_passwd",
         "M:F/A" : "vmq_passwd:auth_on_register/5\n",
         "Type" : "application"
      },
      {
         "Type" : "application",
         "M:F/A" : "vmq_acl:auth_on_publish/6\nvmq_acl:auth_on_subscribe/3\n",
         "Plugin" : "vmq_acl",
         "Hook(s)" : "auth_on_publish\nauth_on_subscribe\n"
      }
   ]
}

Set configuration values

Request:

GET /api/v1/set?allow_publish_during_netsplit=on

Curl:

curl "http://JxctXkZ1OTVnlwvguSCE9KtujacMkOLF@localhost:8888/api/v1/set?allow_publish_during_netsplit=on"

Response:

[]

Disconnect a client

Request:

GET /api/v1/session/disconnect?client-id=myclient&--cleanup

Curl:

curl "http://JxctXkZ1OTVnlwvguSCE9KtujacMkOLF@localhost:8888/api/v1/session/disconnect?client-id=myclient&--cleanup"

Response:

[]

Introduction

When working with a system like VerneMQ sometimes when troubleshooting it would be nice to know what a client is actually sending and receiving and what VerneMQ is doing with this information. For this purpose VerneMQ has a built-in tracing mechanism which is safe to use in production settings as there is very little overhead in running the tracer and has built-in protection mechanisms to stop traces that produce too much information.

Tracing clients

To trace a client the following command is available:

vmq-admin trace client client-id=<client-id>

See the available flags by calling vmq-admin trace client --help.

A typical trace could look like the following:

$ vmq-admin trace client client-id=client
No sessions found for client "client"
New session with PID <7616.3443.1> found for client "client"
<7616.3443.1> MQTT RECV: CID: "client" CONNECT(c: client, v: 4, u: username, p: password, cs: 1, ka: 30)
<7616.3443.1> Calling auth_on_register({{172,17,0,1},34274},{[],<<"client">>},username,password,true) 
<7616.3443.1> Hook returned "ok"
<7616.3443.1> MQTT SEND: CID: "client" CONNACK(sp: 0, rc: 0)
<7616.3443.1> MQTT RECV: CID: "client" SUBSCRIBE(m1) with topics:
    q:0, t: "topic"
<7616.3443.1> Calling auth_on_subscribe(username,{[],<<"client">>}) with topics:
    q:0, t: "topic"
<7616.3443.1> Hook returned "ok"
<7616.3443.1> MQTT SEND: CID: "client" SUBACK(m1, qt[0])
<7616.3443.1> Trace session for client stopped

In this particular trace a trace was started for the client with client-id client. At first no clients are connected to the node where the trace has been started, but a little later the client connects and we see the trace come alive. The strange identifier <7616.3443.1> is called a process identifier and is the identifier of the process in which the trace happened - this isn't relevant unless one wants to correlate the trace with log entries where process identifiers are also logged. Besides the process identifier there are some lines with MQTT SEND and MQTT RECV which are to be understood from the perspective of the broker. In the above trace this means that first the broker receives a CONNECT frame and replies with a CONNACK frame. Each MQTT event is annotated with the data from the MQTT frame to give as much detail and insight as possible.

Notice the auth_on_register call between CONNECT and CONNACK which is the authentication plugin hook being called to authenticate the client. In this case the hook returned ok which means the client was successfully authenticated.

Likewise notice the auth_on_subscribe call between the SUBSCRIBE and SUBACK frames which is plugin hook used to authorize if this particular subscription should be allowed or not. In this case the subscription was authorized.

VerneMQ can be monitored in several ways. We implemented native support for , , and .

The metrics are also available via the command line tool:

Or with:

Which will output the metrics together with a short description describing what the metric is about. An example looks like:

Notice that the metrics:

Are no longer used (always 0) and will be removed in the future. They were replaced with mqtt_connack_sent using the return_code label. For MQTT 5.0 the reason_code label is used instead.

The output on the command line are aggregated by default, but details for a label can be shown as well, for example all metrics with the not_authorized label:

All available labels can be show using vmq-admin metrics show --help.

The graphite exporter reports the broker metrics at a fixed interval (defined in milliseconds) to a graphite server. The necessary configuration is done inside the vernemq.conf.

You can further tune the connection to the Graphite server:

The Prometheus exporter is enabled by default and installs an HTTP handler on http://localhost:8888/metrics. To read more about configuring the HTTP listener, see .

Example Scrape Config

Add the following configuration to the scrape_configs section inside prometheus.yml of your Prometheus server.

This tells Prometheus to scrape the VerneMQ metrics endpoint every 5 seconds.

Please follow the documentation on the website to properly configure the metrics scraping as well as how to access those metrics and configure alarms and graphs.

VerneMQ comes with a built-in status page which by default is enabled and is available on http://localhost:8888/status, see .

The status page is a simple overview of the cluster and the individual nodes in the cluster as seen below:

VerneMQ provides multiple hooks throughout the lifetime of a session. The most important ones are the auth_on_register and auth_on_register_m5 hooks which act as an application level firewall granting or rejecting new clients.

auth_on_register and auth_on_register_m5

Every plugin that implements the auth_on_register or auth_on_register_m5 hooks are part of a conditional plugin chain. For this reason we allow the hook to return different values depending on how the plugin grants or rejects this client. In case the plugin doesn't know the client it is best to return next as this would allow subsequent plugins in the chain to validate this client. If no plugin is able to validate the client it gets automatically rejected.

on_auth_m5

on_register and on_register_m5

on_client_wakeup

on_client_offline

on_client_gone

VerneMQ is implemented in Erlang OTP and therefore runs on top of the Erlang VM. For this reason plugins have to be developed in a programming language that runs on the Erlang VM. The most popular choice is obviously the Erlang programming language itself, but Elixir or Lisp flavoured Erlang LFE could be used too.

This guide explains the different flows that expose different hooks to be used for custom plugins. It also describes the code structure a plugin must comply to in order to be successfully loaded and started by the VerneMQ plugin mechanism.

All the hooks that are currently exposed fall into one of three categories.

1. Hooks that allow you to change the protocol flow. An example could be to authenticate a client using the auth_on_register hook.

2. Hooks that inform you about a certain action, that could be used for example to implement a custom logging or audit plugin.

3. Hooks that are called given a certain condition

Notice that some hooks come in two variants, for example the auth_on_register and then auth_on_register_m5 hooks. The _m5 postfix refers to the fact that this hook is only invoked in an MQTT 5.0 session context whereas the other is invoked in a MQTT 3.1/3.1.1 session context.

Before going into the details, let's give a quick intro to the VerneMQ plugin system.

Plugin System

The VerneMQ plugin system allows you to load, unload, start and stop plugins during runtime, and you can even upgrade a plugin during runtime. To make this work it is required that the plugin is an OTP application and strictly follows the rules of implementing the Erlang OTP application behaviour. It is recommended to use the rebar3 toolchain to compile the plugin. VerneMQ comes with built-in support for the directory structure used by rebar3.

Every plugin has to describe the hooks it is implementing as part of its application environment file. The vmq_acl plugin for instance comes with the application environment file below:

Lines 6 to 10 instruct the plugin system to ensure that those dependent applications are loaded and started. If you're using third party dependencies make sure that they are available in compiled form and part of the plugin load path. Lines 16 to 20 allow the plugin system to compile the plugin rules. Yes, you've heard correctly. The rules are compiled into Erlang VM code to make sure the lookup and execution of plugin code is as fast as possible. Some hooks exist which are used internally such as the change_config/1, we'll describe those at some other point.

The environment value for vmq_plugin_hooks is a list of hooks. A hook is specified by {Module, Function, Arity, Options}.

To streamline the plugin development we provide a different Erlang behaviour for every hook a plugin implements. Those behaviours are part of the vernemq_dev library application, which you should add as a dependency to your plugin. vernemq_dev also comes with a header file that contains all the type definitions used by the hooks.

Chaining

It is possible to have multiple plugins serving the same hook. Depending on the hook the plugin chain is used differently. The most elaborate chains can be found for the hooks that deal with authentication and authorization flows. We also call them conditional chains as a plugin can give control away to the next plugin in the chain. The image show a sample plugin chain for the auth_on_register hook.

Most hooks don't require conditions and are mainly used as event handlers. In this case all plugins in a chain are called. An example for such a hook would be the on_register hook.

A rather specific case is the need to call only one plugin instead of iterating through the whole chain. VerneMQ uses such hooks for it's pluggable message storage system.

Unless you're implementing your custom message storage backend, you probably won't need this style of hook.

Startup

The plugin mechanism uses the application environment file to infer the applications that it has to load and start prior to starting the plugin itself. It internally uses the application:ensure_all_started/1 function call to start the plugin. If your setup is more complex you could override this behaviour by implementing a custom start/0 function inside a module that's named after your plugin.

Teardown

The plugin mechanism uses application:stop/1 to stop and unload the plugin. This won't stop the dependent application started at startup. If you rely on third party applications that aren't started as part of the VerneMQ release, e.g. a database driver, you can implement a custom stop/0 function inside a module that's named after your plugin and properly stop the driver there.

Public Type Specs

The vmq_types.hrl exposes all the type specs used by the hooks. The following types are used by the plugin system:

In this section the subscription flow is described. VerneMQ provides several hooks to intercept the subscription flow. The most important ones are the auth_on_subscribe and auth_on_subscribe_m5 hooks which act as an application level firewall granting or rejecting subscribe requests.

auth_on_subscribe and auth_on_subscribe_m5

on_subscribe and on_subscribe_m5

on_unsubscribe and on_unsubscribe_m5

A simple way to gauge the health of a VerneMQ cluster is to query the /health path on the .

The health check will return 200 when VerneMQ is accepting connections and is joined with the cluster (for clustered setups). 503 will be returned in case any of those two conditions are not met.

The systree functionality is enabled by default and reports the broker metrics at a fixed interval defined in the vernemq.conf. The metrics defined are transformed to MQTT topics e.g. mqtt_publish_received is transformed to $SYS/<nodename>/mqtt/publish/received. <nodename> is your node's name, as configured in the vernemq.conf. To find it, you can grep the file for it: grep nodename vernemq.conf

The complete list of metrics can be found

This option defaults to 20000 milliseconds.

If the systree feature is not required it can be disabled in vernemq.conf

Examples:

Intro

Kubernetes (K8s) is possibly the most mature technology for deploying Docker containers at scale. While running a single Docker container is supposed to be easy, running a Kubernetes cluster definitely isn't. That's why we recommended to work with a certified Kubernetes partner such as Amazon AWS EKS, Google Cloud GKE, Microsoft Azure AKS, or DigitalOcean.

If your applications already live in Docker containers and are deployed on Kubernetes it can be beneficial to also run VerneMQ on Kubernetes. This guide covers how to successfully deploy a VerneMQ cluster on Kubernetes. Multiple options exist to deploy a VerneMQ cluster at this point. This guide describes how to use the official Helm chart as well as the still experimental Kubernetes Operator.

For the sake of clarity, this guide defines the following terms:

• Kubernetes Node: A single virtual or physical machine in a Kubernetes cluster.

• Kubernetes Cluster: A group of nodes firewalled from the internet, that are the primary compute resources managed by Kubernetes.

• Edge router: A router that enforces the firewall policy for your cluster. This could be a gateway managed by a cloud provider or a physical piece of hardware.

• Cluster network: A set of links, logical or physical, that facilitate communication within a cluster according to the Kubernetes networking model.

• Service: A Kubernetes Service that identifies a set of pods using label selectors. Unless mentioned otherwise, Services are assumed to have virtual IPs only routable within the cluster network

• VerneMQ Cluster: A group of VerneMQ containers that are connected via the Erlang Distribution as well as the VerneMQ clustering mechanism.

Deploy VerneMQ with Helm

Helm calls itself the package manager for Kubernetes. In Helm a package is called a chart. VerneMQ comes with such a Helm chart simplifying the initial setup tremendously. If you don't have setup Helm yet, please navigate through their quickstart guide.

Once Helm is properly setup just run the following command in your shell.

helm install vernemq/vernemq

This will deploy a single node VerneMQ cluster. Have a look at the possible configuration here.

Deploy VerneMQ using the Kubernetes Operator

A Kubernetes Operator is a method of packaging, deploying and managing a Kubernetes application. The VerneMQ Operator is basically just a Pod with the task to deploy a VerneMQ cluster given a so called Custom Resource Definition (CRD). The VerneMQ CRD aims that all required configuration can be made through the CRD and no further configuration should be required. The following command installs the operator along a two node VerneMQ cluster into the namespace messaging

curl -L https://codeload.github.com/vernemq/vmq-operator/zip/master --output repo.zip; \
unzip -j repo.zip '*/examples/only_vernemq/*' -d only_vernemq; \
kubectl apply -f only_vernemq

This will result in the following Pods:

kubectl get pods --namespace messaging
NAME                                      READY   STATUS        RESTARTS   AGE
vernemq-k8s-0                             1/1     Running       0          53m
vernemq-k8s-1                             1/1     Running       0          4m14s
vernemq-k8s-deployment-59f5684549-s7jd4   1/1     Running       0          2d17h
vmq-operator-76f5f78f96-2jbwt             1/1     Running       0          4m28s

And the following cluster status:

kubectl exec vernemq-k8s-0 vmq-admin cluster show --namespace messaging
+-----------------------------------------------------------------+-------+
|                              Node                               |Running|
+-----------------------------------------------------------------+-------+
|vmq@vernemq-k8s-0.vernemq-k8s-service.messaging.svc.cluster.local| true  |
|vmq@vernemq-k8s-1.vernemq-k8s-service.messaging.svc.cluster.local| true  |
+-----------------------------------------------------------------+-------+

Load Balancing in Kubernetes

In a VerneMQ cluster it doesn't matter to which node a MQTT client connects, subscribes or publishes. A VerneMQ cluster looks like one big MQTT broker to the outside. While this is the main idea of VerneMQ it comes with a cost, namely the data replication/synchronization overhead when 'persistent' clients hop from one pod to the other. As a consequence, we recommend to intelligently choose how to load balance your MQTT clients.

Load balancing in Kubernetes is configured via the Service object. Multiple service types exist:

ClusterIP

The ClusterIP type is the default and only permits access from within the Kubernetes cluster. Other pods in the Kubernetes cluster can access VerneMQ via ClusterIP:Port . The underlying balancing strategy is based on the settings of kube-proxy. Also this type requires that one terminates TLS either in VerneMQ directly or via a different Pod e.g. HAproxy.

NodePort

The NodePort type uses ClusterIP under the hood but allocates a Port on every Kubernetes node and routes incoming traffic from NodeIP:NodePort to the ClusterIP:Port . Like with ClusterIP this type requires that one terminates TLS either in VerneMQ directly or via a different Pod e.g. HAproxy.

Loadbalancer

The Loadbalancer type uses an external load balancer provided by the cloud provider. In fact this Service type only provides the glue code required to interact with the Loadbalancing services from different cloud providers. If you're running a bare-metal Kubernetes cluster you won't be able to use this Service type, unless you deploy a Kubernetes aware network loadbalancer yourself. Check out MetalLB, which provides a network loadbalancer for bare-metal Kubernetes clusters.

Ingress Controllers

Ingress controllers provide another way to do load balancing and TLS termination in a Kubernetes cluster. However the officially supported ingress controllers nginx and GCE focus on balancing HTTP requests instead of plain TCP connections. Therefore their support for TLS termination is also limited to HTTPS.

Multiple third-party ingress controllers exist, however most of them focus on handling HTTP requests. One of the exceptions is Voyager by AppsCode an ingress controller based on HAProxy, which also efficiently terminates TLS.

General recommendations for large scale deployments

1. Use an external loadbalancer provided by the cloud provider that is capable of terminating TLS and apply a load balancing strategy that provides session affinity e.g. via source hashing.

2. Terminate TLS outside VerneMQ.

3. Configure the Pod NodeAffinity correctly to ensure that only one VerneMQ pod is scheduled on any Kubernetes cluster node.

4. It's preferred to have a smaller number of Pods that are very powerful in terms of available CPU and RAM than the opposite.

We recommend to use the rebar3 toolchain to generate the basic Erlang OTP application boilerplate and start from there.

rebar3 new app name="myplugin" desc="this is my first VerneMQ plugin"
===> Writing myplugin/src/myplugin_app.erl
===> Writing myplugin/src/myplugin_sup.erl
===> Writing myplugin/src/myplugin.app.src
===> Writing myplugin/rebar.config
===> Writing myplugin/.gitignore
===> Writing myplugin/LICENSE
===> Writing myplugin/README.md

Change the rebar.config file to include the vernemq_dev dependency:

{erl_opts, [debug_info]}.
{deps, [{vernemq_dev,
    {git, "git://github.com/vernemq/vernemq_dev.git", {branch, "master"}}}
]}.

Compile the application, this will automatically fetch vernemq_dev.

rebar3 compile                             
===> Verifying dependencies...
===> Fetching vmq_commons ({git,
                                      "git://github.com/vernemq/vernemq_dev.git",
                                      {branch,"master"}})
===> Compiling vernemq_dev
===> Compiling myplugin

Now you're ready to implement the hooks. Don't forget to add the proper vmq_plugin_hooks entries to your src/myplugin.app.src file.

For a complete example, see the vernemq_demo_plugin.

The VerneMQ Webhooks plugin provides an easy and flexible way to build powerful plugins for VerneMQ using web hooks. With VerneMQ Webhooks you are free to select the implementation language to match your technical requirements or the language in which you feel comfortable and productive in. You can use any modern language such as Python, Go, C#/.Net and indeed any language in which you can build something that can handle HTTP requests.

The idea of VerneMQ Webhooks very simple: you can register an HTTP endpoint with a VerneMQ plugin hook and whenever the hook (such as auth_on_register) is called, the VerneMQ Webhooks plugin dispatches a HTTP post request to the registered endpoint. The HTTP post request contains a HTTP header like vernemq-hook: auth_on_register and a JSON encoded payload. The endpoint then responds with code 200 on success and with a JSON encoded payload informing the VerneMQ Webhooks plugin which action to take (if any).

Configuring webhooks

To enable webhooks make sure to set:

plugins.vmq_webhooks = on

And then each webhook can be configured like this:

vmq_webhooks.mywebhook1.hook = auth_on_register
vmq_webhooks.mywebhook1.endpoint = http://127.0.0.1/myendpoints

It is also possible to dynamically register webhooks at run-time:

$ vmq-admin webhooks register hook=auth_on_register endpoint="http://localhost"

See which endpoints are registered:

$ vmq-admin webhooks show

And finally deregistering an endpoint:

$ vmq-admin webhooks deregister hook=auth_on_register endpoint="http://localhost"

Connection pool configuration

Each registered hook uses by default a connection pool containing maximally 100 connections. This can be changed by setting vmq_webhooks.pool_max_connections to a different value. Similarly the vmq_webhooks.pool_timeout configuration (value is in milliseconds) can be set to control how long an unused connection should stay in the connection pool before being closed and removed. The default value is 60000 (60 seconds).

These options are available in VerneMQ 1.4.0.

Caching

VerneMQ webhooks support caching of the auth_on_register, auth_on_publish and auth_on_subscribe hooks.

This can be used to speed up authentication and authorization tremendously. All data passed to these hooks is used to look if the call is in the cache, except in the case of the auth_on_publish where the payload is omitted.

To enable caching for an endpoint simply return the cache-control: max-age=AgeInSeconds in the response headers to one of the mentioned hooks. If the call was successful (authentication granted), the request will be cached together with any modifiers, except for the payload modifier in the auth_on_publish hook.

Whenever a non-expired entry is looked up in the cache the endpoint will not be called and the modifiers of the cached entry will be returned, if any.

It is possible to inspect the cache using:

$ vmq-admin webhooks cache show

Webhook specs

All webhooks are called with method POST. All hooks need to be answered with the HTTP code 200 to be considered successful. Any hook called that does not return the 200 code will be logged as an error as will any hook with an unparseable payload.

All hooks are called with the header vernemq-hook which contains the name of the hook in question.

For detailed information about the hooks and when they are called, see the sections Session Lifecycle, Subscribe Flow and Publish Flow.

auth_on_register

Header: vernemq-hook: auth_on_register

Webhook example payload:

{
    "peer_addr": "127.0.0.1",
    "peer_port": 8888,
    "username": "username",
    "password": "password",
    "mountpoint": "",
    "client_id": "clientid",
    "clean_session": false
}

A minimal response indicating the authentication was successful looks like:

{
    "result": "ok"
}

It is also possible to override various client specific settings by returning an array of modifiers:

{
    "result": "ok",
    "modifiers": {
        "max_message_size": 65535,
        "max_inflight_messages": 10000,
        "retry_interval": 20000
    }
}

Note, the retry_interval is in milli-seconds. It is possible to override many more settings, see the Session Lifecycle for more information.

Other possible responses:

{
  "result": "next"
}

{
  "result": {
    "error": "not_allowed"
  }
}

auth_on_subscribe

Header: vernemq-hook: auth_on_subscribe

Webhook example payload:

{
    "client_id": "clientid",
    "mountpoint": "",
    "username": "username",
    "topics":
        [{"topic": "a/b",
          "qos": 1},
         {"topic": "c/d",
          "qos": 2}]
}

A minimal response indicating the subscription authorization was successful looks like:

{
    "result": "ok"
}

Another example where where the topics to subscribe have been rewritten looks like:

{
    "result": "ok",
    "topics":
        [{"topic": "rewritten/topic",
          "qos": 0}]
}

Note, you can also pass a qos with value 128 which means it was either not possible or the client was not allowed to subscribe to that specific question.

Other possible responses:

{
    "result": "next"
}

{
    "result": { "error": "some error message" }
}

auth_on_publish

Header: vernemq-hook: auth_on_publish

Note, in the example below the payload is not base64 encoded which is not the default.

Webhook example payload:

{
    "username": "username",
    "client_id": "clientid",
    "mountpoint": "",
    "qos": 1,
    "topic": "a/b",
    "payload": "hello",
    "retain": false
}

A minimal response indicating the publish was authorized looks like:

{
    "result": "ok"
}

A more complex example where the publish topic, qos, payload and retain flag is rewritten looks like:

{
    "result": "ok",
    "modifiers": {
        "topic": "rewritten/topic",
        "qos": 2,
        "payload": "rewritten payload",
        "retain": true
    }
}

Other possible responses:

{
    "result": "next"
}

{
    "result": { "error": "some error message" }
}

on_register

Header: vernemq-hook: on_register

Webhook example payload:

{
    "peer_addr": "127.0.0.1",
    "peer_port": 8888,
    "username": "username",
    "mountpoint": "",
    "client_id": "clientid"
}

The response should be an empty json object {}.

on_publish

Header: vernemq-hook: on_publish

Note, in the example below the payload is not base64 encoded which is not the default.

Webhook example payload:

{
    "username": "username",
    "client_id": "clientid",
    "mountpoint": "",
    "qos": 1,
    "topic": "a/b",
    "payload": "hello",
    "retain": false
}

The response should be an empty json object {}.

on_subscribe

Header: vernemq-hook: on_subscribe

Webhook example payload:

{
    "client_id": "clientid",
    "mountpoint": "",
    "username": "username",
    "topics":
        [{"topic": "a/b",
          "qos": 1},
         {"topic": "c/d",
          "qos": 2}]
}

The response should be an empty json object {}.

on_unsubscribe

Header: vernemq-hook: on_unsubscribe

Webhook example payload:

{
    "username": "username",
    "client_id": "clientid",
    "mountpoint": "",
    "topics":
        ["a/b", "c/d"]
}

Example response:

{
    "result": "ok",
    "topics":
        ["rewritten/topic"]
}

Other possible responses:

{
    "result": "next"
}

{
    "result": { "error": "some error message" }
}

on_deliver

Header: vernemq-hook: on_deliver

Note, in the example below the payload is not base64 encoded which is not the default.

Webhook example payload:

{
    "username": "username",
    "client_id": "clientid",
    "mountpoint": "",
    "topic": "a/b",
    "payload": "hello"
}

Example response:

{
  "result": "ok",
  "modifiers":
  {
        "topic": "rewritten/topic",
        "payload": "rewritten payload"
    }
}

Other possible responses:

{
    "result": "next"
}

on_offline_message

Header: vernemq-hook: on_offline_message

Note, in the example below the payload is not base64 encoded which is not the default.

Webhook example payload:

{
    "client_id": "clientid",
    "mountpoint": "",
    "qos": "1",
    "topic": "sometopic",
    "payload": "payload",
    "retain": false
}

The response should be an empty json object {}.

on_client_wakeup

Header: vernemq-hook: on_client_wakeup

Webhook example payload:

{
    "client_id": "clientid",
    "mountpoint": ""
}

The response should be an empty json object {}.

on_client_offline

Header: vernemq-hook: on_client_offline

Webhook example payload:

{
    "client_id": "clientid",
    "mountpoint": ""
}

The response should be an empty json object {}.

on_client_gone

Header: vernemq-hook: on_client_gone

Webhook example payload:

{
    "client_id": "clientid",
    "mountpoint": ""
}

The response should be an empty json object {}.

auth_on_register_m5

Header: vernemq-hook: auth_on_register_m5

Webhook example payload:

{
    "peer_addr": "127.0.0.1",
    "peer_port": 8888,
    "mountpoint": "",
    "client_id": "client-id",
    "username": "username",
    "password": "password",
    "clean_start": true,
    "properties": {}
}

A minimal response indicating the authentication was successful looks like:

{
    "result": "ok"
}

It is also possible to override various client specific settings by returning an array of modifiers:

{
    "result": "ok",
    "modifiers": {
        "max_message_size": 65535,
        "max_inflight_messages": 10000
    }
}

Note, the retry_interval is in milli-seconds. It is possible to override many more settings, see the Session Lifecycle for more information.

Other possible responses:

{
    "result": "next"
}

{
    "result": {
      "error": "not_allowed"
  }
}

on_auth_m5

Header vernemq-hook: on_auth_m5

Webhook example payload:

{
    "username": "username",
    "mountpoint": "",
    "client_id": "client-id",
    "properties": {
      "p_authentication_data": "QVVUSF9EQVRBMA==",
      "p_authentication_method": "AUTH_METHOD"
    }
}

Note, as the authentication data is binary data it is base64 encoded.

A minimal response indicating the authentication was successful looks like:

  "modifiers": {
    "properties": {
      "p_authentication_data": "QVVUSF9EQVRBMQ==",
      "p_authentication_method": "AUTH_METHOD"
    }
    "reason_code": 0
  },
  "result": "ok"
}

If authentication were to continue for another round a reason code with value 24 (Continue Authentication) should be returned instead. See also the relevant section in the MQTT 5.0 specification.

auth_on_subscribe_m5

Header: vernemq-hook: auth_on_subscribe_m5

Webhook example payload:

{
    "username": "username",
    "mountpoint": "",
    "client_id": "client-id",
    "topics": [
      {
        "topic": "test/topic",
        "qos": 1
      }
    ],
    "properties": {}
  }