comqtt Documentation

The Helm chart deploys a StatefulSet and a headless Service to provide stable DNS names for each pod (e.g., release-comqtt-0...).

Bootstrapping and Scaling

• Seed Members: Computed automatically from replicaCount and the headless Service FQDN. You can scale up by updating replicaCount (must remain odd) without a chart upgrade.
• Raft Bootstrap: The chart only sets Raft bootstrap to true if the pod hostname ends in -0 (the genesis pod) AND the Raft data directory is empty. This makes helm rollout restart safe, as existing pods will simply re-join the cluster.
• Durability: Each replica uses a volumeClaimTemplate for Raft log durability. persistence.enabled=false is not allowed in cluster mode.

Reliability

• Quorum Protection: A PodDisruptionBudget is configured with minAvailable = ⌈(replicas+1)/2⌉ to prevent voluntary disruptions from breaking quorum.
• Anti-Affinity: Soft pod anti-affinity is enabled by default. Set cluster.hardAntiAffinity=true to force pods onto distinct hostnames.

Comqtt provides a wide range of event hooks to interact with the broker and client lifecycle. These hooks allow you to monitor events, modify packets, or implement custom logic for authentication and authorization.

Key Hook Categories

• Packet Manipulation (Most Flexible): Use OnPacketRead, OnPacketEncode, and OnPacketSent to control and modify all incoming and outgoing packets.
• Authentication & Authorization: To implement custom security, you MUST use OnConnectAuthenticate (to allow/deny access) and OnACLCheck (to control publish/subscribe permissions).
• Client Lifecycle: Monitor connections and disconnections using OnConnect, OnSessionEstablished, OnDisconnect, and OnClientExpired.
• Message Lifecycle: Track messages using OnPublish, OnPublished, OnSubscribe, OnUnsubscribe, and OnRetainMessage.
• Persistence: Implement custom storage by using hooks like StoredClients, StoredSubscriptions, StoredInflightMessages, and StoredRetainedMessages.

For full function signatures and the mqtt.Hook interface, refer to mqtt/hooks.go in the source code.

| Function               | Usage |
|------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| OnStarted              | Called when the server has successfully started.                                                                                                                                                                                                                                                           |
| OnStopped              | Called when the server has successfully stopped.                                                                                                                                                                                                                                                             |
| OnConnectAuthenticate  | Called when a user attempts to authenticate with the server. An implementation of this method MUST be used to allow or deny access to the server (see hooks/auth/allow_all or basic). It can be used in custom hooks to check connecting users against an existing user database. Returns true if allowed. |
| OnACLCheck             | Called when a user attempts to publish or subscribe to a topic filter. As above.                                                                                                                                                                                                                                                          |
| OnSysInfoTick          | Called when the $SYS topic values are published out.                                                                                                                                                                                                                                                                     |
| OnConnect              | Called when a new client connects, may return an error or packet code to halt the client connection process.                                                                                                                                                                                                                              |
| OnSessionEstablish     | Called immediately after a new client connects and authenticates and immediately before the session is established and CONNACK is sent.
| OnSessionEstablished   | Called when a new client successfully establishes a session (after OnConnect)                                                                                                                                                                                                                              |
| OnDisconnect           | Called when a client is disconnected for any reason.                                                                                                                                                                                                                                                                     |
| OnAuthPacket           | Called when an auth packet is received. It is intended to allow developers to create their own mqtt v5 Auth Packet handling mechanisms. Allows packet modification.                                                                                                                                                                                          |
| OnPacketRead           | Called when a packet is received from a client. Allows packet modification.                                                                                                                                                                                                                                                               |
| OnPacketEncode         | Called immediately before a packet is encoded to be sent to a client. Allows packet modification.                                                                                                                                                                                                                                      |
| OnPacketSent           | Called when a packet has been sent to a client.                                                                                                                                                                                                                                                                           |
| OnPacketProcessed      | Called when a packet has been received and successfully handled by the broker.                                                                                                                                                                                                                                                              |
| OnSubscribe            | Called when a client subscribes to one or more filters. Allows packet modification.                                                                                                                                                                                                                                                        |
| OnSubscribed           | Called when a client successfully subscribes to one or more filters.                                                                                                                                                                                                                                                                     |
| OnSelectSubscribers   | Called when subscribers have been collected for a topic, but before shared subscription subscribers have been selected. Allows receipient modification.                                                                                                                                                                                          |
| OnUnsubscribe          | Called when a client unsubscribes from one or more filters. Allows packet modification.                                                                                                                                                                                                                                                        |
| OnUnsubscribed         | Called when a client successfully unsubscribes from one or more filters.                                                                                                                                                                                                                                                                 |
| OnPublish              | Called when a client publishes a message. Allows packet modification.                                                                                                                                                                                                                                                                   |
| OnPublished            | Called when a client has published a message to subscribers.                                                                                                                                                                                                                                                                             |
| OnPublishDropped       | Called when a message to a client is dropped before delivery, such as if the client is taking too long to respond.                                                                                                                                                                                                                                |
| OnRetainMessage        | Called then a published message is retained.                                                                                                                                                                                                                                                                                              |
| OnRetainPublished      | Called then a retained message is published to a client.                                                                                                                                                                                                                                                                                  |
| OnQosPublish           | Called when a publish packet with Qos >= 1 is issued to a subscriber.                                                                                                                                                                                                                                                                   |
| OnQosComplete          | Called when the Qos flow for a message has been completed.                                                                                                                                                                                                                                                                               |
| OnQosDropped           | Called when an inflight message expires before completion.                                                                                                                                                                                                                                                                              |
| OnPacketIDExhausted    | Called when a client runs out of unused packet ids to assign.                                                                                                                                                                                                                                                                          |
| OnWill                 | Called when a client disconnects and intends to issue a will message. Allows packet modification.                                                                                                                                                                                                                                         |
| OnWillSent             | Called when an LWT message has been issued from a disconnecting client.                                                                                                                                                                                                                                                                   |
| OnClientExpired        | Called when a client session has expired and should be deleted.                                                                                                                                                                                                                                                                         |
| OnRetainedExpired     | Called when a retained message has expired and should be deleted.                                                                                                                                                                                                                                                                     |
| StoredClients          | Returns clients, eg. from a persistent store.                                                                                                                                                                                                                                                                                           |
| StoredSubscriptions    | Returns client subscriptions, eg. from a persistent store.                                                                                                                                                                                                                                                                                             |
| StoredInflightMessages | Returns inflight messages, eg. from a persistent store.                                                                                                                                                                                                                                                                                             |
| StoredRetainedMessages | Returns retained messages, eg. from a persistent store.                                                                                                                                                                                                                                                                                          |
| StoredSysInfo          | Returns stored system info values, eg. from a persistent store.                                                                                                                                                                                                                                                                                          |

To use PostgreSQL for authentication and ACL, you must first set up the required schema. Comqtt expects bcrypt hashed passwords for client authentication.

PostgreSQL Schema:

BEGIN;
CREATE TABLE mqtt_user (
    id serial PRIMARY KEY,
    username TEXT NOT NULL UNIQUE,
    password TEXT NOT NULL,
    allow smallint DEFAULT 1 NOT NULL,
    created timestamp with time zone DEFAULT NOW(),
    updated timestamp
);

CREATE TABLE mqtt_acl(
    id serial PRIMARY KEY,
    username TEXT NOT NULL,
    topic TEXT NOT NULL,
    access smallint DEFAULT 3 NOT NULL,
    created timestamp with time zone DEFAULT NOW(),
    updated timestamp
);
CREATE INDEX mqtt_acl_username_idx ON mqtt_acl(username);
COMMIT;

Generating Bcrypt Hashes: Since passwords must be bcrypt hashed, use the following snippets to prepare client credentials:

Go:

import "golang.org/x/crypto/bcrypt"
hashed, err := bcrypt.GenerateFromPassword(pwd, bcrypt.DefaultCost)
if err != nil {
	return 
}
println("Password hash for MQTT client: ", hashed)

Python:

import bcrypt
salt = bcrypt.gensalt(rounds=10)
hashed = bcrypt.hashpw(b"VeryVerySecretPa55w0rd", salt)
print(f"Password hash for MQTT client: {hashed}")

To create a cluster, you must start nodes sequentially. The first node must be bootstrapped, and subsequent nodes must point to an existing member using the -members flag.

1. Start the first node (Bootstrap)

Use the -raft-bootstrap=true flag to allow the first node to elect a leader.

./comqtt --node-name=c01 --gossip-port=7946 --raft-port=8946 --raft-bootstrap=true

2. Start the second node

Provide a seed node from the existing cluster using the -members flag. Ensure ports for gossip, raft, and listeners do not conflict with the first node.

./comqtt --node-name=c02 --gossip-port=7947 --raft-port=8947 --members=localhost:7946 --tcp=:1885 --ws=:1886 --http=:1881

3. Start the third node

Repeat the process, pointing to a member of the existing cluster.

./comqtt --node-name=c03 --gossip-port=7948 --raft-port=8948 --members=localhost:7946 --tcp=:1887 --ws=:1888 --http=:1882

Note: For advanced features like bridging to Kafka or multiple authentication methods, it is highly recommended to use a configuration file via the -conf flag instead of CLI arguments.

# Start first node
./comqtt --node-name=c01 --gossip-port=7946 --raft-port=8946 --raft-bootstrap=true

# Start second node
./comqtt --node-name=c02 --gossip-port=7947 --raft-port=8947 --members=localhost:7946 --tcp=:1885 --ws=:1886 --http=:1881

# Start third node
./comqtt --node-name=c03 --gossip-port=7948 --raft-port=8948 --members=localhost:7946 --tcp=:1887 --ws=:1888 --http=:1882

The chart prefers the Gateway API over legacy Ingress. You must provide a Gateway resource and a compatible provider (e.g., Envoy Gateway, Cilium, Istio).

Supported Routes

1. HTTPRoute (gateway.api.enabled): Enabled by default. Serves the broker's HTTP listener on port 8080 for REST APIs and /metrics.
2. TCPRoute (gateway.mqtt.enabled): Used for raw MQTT. This is currently an alpha API and requires a TCP-aware provider. Defaults to false.

Example Configuration (Envoy Gateway)

# values.yaml
gateway:
  enabled: true
  parentRefs:
    - name: eg
      namespace: envoy-gateway-system
  api:
    enabled: true
    hostnames: ["comqtt.example.com"]
  mqtt:
    enabled: true

TLS Configuration

• At the Gateway: Configure listeners on your Gateway resource.
• At the Broker: Set tls.existingSecret and configure config.mqtt.tls.{ca-cert,server-cert,server-key}.