Flexway - MQTT devices: the listener in proxy in front… (Operations)

Operations guide: configure, deploy and observe the gateway. What the feature does for the product: product guide; connecting a device: developer guide.

Sensors, controllers and plant gateways connect to the gateway in MQTT 5 or 3.1.1, over TLS or WebSocket, instead of your broker. The gateway is a proxy, never a broker: each device has its own connection to your broker (Mosquitto, EMQX, HiveMQ...), opened in the gateway's name; each packet is judged against the AsyncAPI contract - who the device is, which topics its tenant and identity open, its rates and quota - then relayed as it is. Sessions, retained messages, wills and QoS 2 stay at your broker. Available since 1.7.0.

On this listener: Sparkplug B and OPC UA PubSub JSON; the console's Devices screen: Console: the Devices screen. The defaults of max_connections, max_connections_per_ip and max_connection_rate_per_ip are provisional, pending performance measurements: write them down if you depend on them.

# Devices behind the gateway (1.7.0): sensors and controllers connect in MQTT 5 or 3.1.1 and reach the
# plant's broker through the gateway - one broker connection per device, each packet judged against the
# AsyncAPI contract, each tenant inside its own topics. Here clear text inside a mesh, devices by API key
# or token; the TLS listener and device certificates are in the guide.
version: 1
contracts:
  asyncapi: ./asyncapi.yaml
upstreams:
  plant:
    protocol: mqtt
    brokers: ["mqtts://broker.plant.svc.cluster.local:8883"]
    sasl: { mechanism: PLAIN, username: flexway, password: "file:/etc/flexway-secrets/broker/password" }
    timeout: 5s                                  # a connection to the broker, per device
auth:
  issuers:
    - name: devices
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: plant-devices
      tenant_claim: site                         # every topic a device reaches is its tenant's
  api_keys:
    store: "file:./api-keys.yaml"                # the key's id as the user name, the key as the password
listeners:
  mqtt:
    address: ":1883"
    mesh_only: true                              # clear text only where the mesh encrypts between pods
    websocket: { address: ":8884" }              # MQTT over WebSocket, path /mqtt
    advertised_address: "mqtt.plant.example.com:1883"   # the load balancer: where a drained device goes
    device:
      messages_per_second: 50                    # past it the device is held, never refused
    failover: { handover_timeout: 30s }

# examples/mqtt-devices/asyncapi.yaml
asyncapi: 3.0.0
info: { title: Plant devices, version: "1.0.0" }
channels:
  telemetry:
    address: "plant/{site}/{device}/telemetry"
    parameters:
      site: { x-flexway-bind: tenant }           # the verified tenant, never read from the topic
      device: { x-flexway-bind: subject }        # the key's id, the token's subject
    messages:
      reading:
        payload: { type: object, required: [v], properties: { v: { type: number } } }
  commands:
    address: "plant/{site}/{device}/cmd"
    parameters:
      site: { x-flexway-bind: tenant }
      device: { x-flexway-bind: subject }
    messages: { command: { payload: { type: object } } }
  fleet:
    address: "plant/{site}/fleet/#"                    # a wildcard after the bound level: the tenant's fleet
    parameters:
      site: { x-flexway-bind: tenant }
    messages: { event: { payload: { type: object } } }
operations:
  publishTelemetry:
    action: send                                 # the device publishes
    channel: { $ref: "#/channels/telemetry" }
    x-flexway-upstream: plant
    x-flexway-validation: full                   # a reading against its schema: 0x99 otherwise
    x-flexway-device: { qos_max: 1 }
    security:
      - type: userPassword
      - type: oauth2
        flows: { clientCredentials: { tokenUrl: "https://login.example.net/token", availableScopes: { "telemetry:write": publish } } }
        scopes: ["telemetry:write"]
  readCommands:
    action: receive                              # the device subscribes
    channel: { $ref: "#/channels/commands" }
    x-flexway-upstream: plant
    x-flexway-device: true
    security: [ { type: userPassword } ]
  readFleet:
    action: receive
    channel: { $ref: "#/channels/fleet" }
    x-flexway-upstream: plant
    x-flexway-device: { shared: true, qos_max: 1 }   # $share/<group>/plant/<site>/fleet/#, the group in the tenant's name
    security: [ { type: userPassword } ]

Declaring what a device may do

An AsyncAPI operation marked x-flexway-device opens its channel to devices, in its direction: send, the device publishes; receive, it subscribes. Nothing else passes.

Key of `x-flexway-device`	Default	Effect
`qos_max`	2	Highest QoS; a `PUBLISH` past it is refused `0x9B`, a subscription granted at it
`messages_per_second`, `bytes_per_second`	those of `listeners.mqtt.device`	The rates of a `PUBLISH` on this channel (`send`)
`shared`	`false`	Admits `$share/<group>/<filter>` on a `receive` channel; the group is written `<tenant>~<group>` at the broker

Every parameter of the channel is bound to a verified fact (x-flexway-bind) and one of them to the tenant - refused at load otherwise: tenant, subject (the key's id, the token's subject, the auth.mtls client's id), claim:<name>, or a field of the device certificate, cert:<field>: cn, serial, fingerprint (the certificate's SHA-256, hexadecimal), or with its index ou[i], o[i], san.dns[i], san.uri[i], san.ip[i], san.email[i]. A value that is empty or holds /, +, #, $, {, } or a NUL keeps the operation closed: never written into a topic, and a value written is never read again as a parameter. A wildcard (+, #) is written in a receive channel only, after the bound levels: a device's filter must be included in it.

All the device channels write the tenant at the same level, after the same literal prefix (plant/{site}/... everywhere) - refused at load otherwise: a channel with the tenant at level 1 and another with it at level 2 would let the tenant named like a literal level of one read the other's topics.

Identifying devices

The listener reads, in order: the certificate verified by tls.client_ca_file and declared in auth.mtls, else the CONNECT's password - an API key whose id is the user name, else a token (exp required; when it is reached, DISCONNECT 0xA0). An alternative of the operation's security names what it admits: X509, userPassword (or plain, apiKey in: password), oauth2/openIdConnect.

Device certificates (TLS required):

listeners:
  mqtt:
    address: ":8883"
    tls:
      cert_file: /etc/flexway-mqtt-tls/tls.crt
      key_file: /etc/flexway-mqtt-tls/tls.key
      client_ca_file: /etc/flexway-mqtt-tls/ca.crt   # verifies device certificates
    websocket: { address: ":8884", path: /mqtt }      # MQTT over WebSocket, same TLS
    advertised_address: "mqtt.plant.example.com:8883"
auth:
  mtls:
    clients:
      - { id: press-07, subject: "CN=press-07", tenant: acme, operations: [publishCertified] }

channels:
  certified:
    address: "plant/{site}/{cn}/certified/{dns}"
    parameters:
      site: { x-flexway-bind: tenant }
      cn:   { x-flexway-bind: "cert:cn" }           # the device certificate's common name
      dns:  { x-flexway-bind: "cert:san.dns[0]" }   # its first DNS name: a multi-valued field names its index
operations:
  publishCertified:
    action: send
    channel: { $ref: "#/channels/certified" }
    x-flexway-upstream: plant
    x-flexway-device: true
    security: [ { type: X509 } ]

Tokens of an OIDC provider, the tenant in a claim:

auth:
  issuers:
    - name: devices
      discovery: https://login.example.net/.well-known/openid-configuration
      audience: plant-devices
      tenant_claim: site

At your broker, the device has the client id <tenant>/<mechanism>:<identity> (acme/apikey:press-07), never the one it chose: a device of one tenant never evicts another's. A token's carries a digest of its issuer, acme/oidc:<8 hex>:<subject>: two issuers handing out the same subject never share a session. A second connection of the same identity takes the session over (0x8E to the first); device.sessions: per_client_id gives a session per device client id under a shared identity, at most device.max_sessions_per_identity at once (0x97 past it).

subject and several issuers - x-flexway-bind: subject binds the subject alone: two issuers of the same tenant that hand out the same subject reach the same topics. Give each tenant one issuer, or bind a claim only one issuer sets (claim:<name>). An issuer binding is proposed for a later lot, not built.

Keep-alive. An MQTT 5 device is told max_keepalive (Server Keep Alive) when it asks more, or none. MQTT 3.1.1 cannot be told one: a device asking more than max_keepalive is refused 0x05, and one asking none (0) too, unless allow_zero_keepalive: true - a silent device would never be seen gone. Every connection ends after max_session_lifetime (24 h) whatever its credential, and reconnects with a credential checked again.

Revocation. At each reload, every open connection is checked again within half a second against the new configuration - a device that only listens too: a key, client or operation withdrawn ends it with DISCONNECT 0x87, and its broker connection with DISCONNECT 0x00, so the broker discards its will (a revoked device is not a lost one). Each delivery from the broker is judged on the configuration in force.

Settings and defaults

Every default can be changed; the bounds apply to the next packet or CONNECT, except the address, TLS, mesh_only, the WebSocket and max_connections, read at start.

Key (`listeners.mqtt.`)	Default	Why
`address`, `tls`, `mesh_only`	no listener unless declared; TLS or `mesh_only`	A device's token or key never crosses the network in clear outside a mesh
`broker_mesh_only`	`false`	A broker in clear (`mqtt://`, `ws://`) is refused at load outside `mesh_only`: set it when a mesh encrypts the hop to the broker
`websocket.path`	`/mqtt`	The path of MQTT-over-WebSocket clients (browsers, plant gateways)
`max_packet_bytes`	256 KiB	A packet is read whole before it is judged: the memory bound per packet
`connect_timeout`	`5s`	A slow `CONNECT` or a lingering TLS handshake never holds a place
`max_connections`	50,000 per replica	Provisional: to be confirmed by bench M1; past it `0x97`
`max_unauthenticated_connections`	1,024	A burst without `CONNECT` never starves authenticated devices
`max_connections_per_ip`, `max_connection_rate_per_ip`	64; 20 per second	Provisional (bench M1); set `0` behind a load balancer or proxy that hides addresses (the chart keeps them: `externalTrafficPolicy: Local`) - otherwise every device shares one address and one bound
`max_connections_per_tenant`, `max_connection_rate_per_tenant`	10,000; 100 per second	One tenant never takes the replica from the others; a reconnection storm is spread
`max_keepalive`, `max_session_expiry`, `topic_alias_maximum`	1,200 s; 1 h; 10 (1,000 at most)	The bounds of the market's brokers (AWS, Event Grid, HiveMQ); an alias is recorded only once its topic is granted
`allow_zero_keepalive`	`false`	A 3.1.1 device asking no keep-alive is refused: it would never be seen gone
`max_session_lifetime`	24 h (1 min to 30 days)	A connection never outlives a revocation the reload would miss
`device.messages_per_second`, `bytes_per_second`, `burst`	100; 512 KiB; 2	AWS IoT Core's; backpressure past them, never a refusal
`device.receive_maximum`, `max_subscriptions`, `max_filters_per_subscribe`	100; 50; 10	Bound what a device holds in flight and subscribes
`device.connect_rate_per_identity`	1 per second	A device in a loop never evicts itself again and again (`0x9F`); keyed on the verified identity
`device.max_refusals_per_minute`	60	Each refusal spends the device's message rate, its audit is sampled (one entry a second per cause); past it `DISCONNECT 0x96`
`device.max_user_properties_bytes`, `max_correlation_data_bytes`	8 KiB; 8 KiB	What a packet carries beside its payload
`device.sessions`	`single`	One session per identity: a second device evicts the first (`0x8E`)
`device.max_sessions_per_identity`	10 (or `max_connections` if lower)	With `per_client_id`, one identity never takes the replica (`0x97`)
`tenant.messages_per_second`, `bytes_per_second`	5,000; 20 MiB	The ceiling of a whole site; subscription plans come in `1.9.0`
`failover.mode`, `reason`	`redirect`; `use_another_server`	A shutdown redirects rather than cuts; `0x9C` is temporary
`failover.server_reference`	`advertised_address`	Always reachable, even by a client that ignores the reference
`failover.redirect_above`, `handover_timeout`	95 %; 30 s	Redirects before saturation; lets a common client with backoff come back
`failover.will_delay_on_handover`	`false`	On: an MQTT 5 will without delay keeps through a takeover, a lost device's will goes up to `handover_timeout` later
`server.shutdown.mqtt_drain_spread`	30 s (half of `drain` at most)	No reconnection storm on the other replicas
`observability.mqtt.closures`	`true`	Count normal closures apart from refusals

Rates, backpressure and quota

Past a rate of the device or of its tenant, the gateway stops reading the device's socket for as long as needed - never a refusal, never a closure that would fire its will - and keeps its keep-alive toward the broker meanwhile. The tenant's day or month quota (a quota policy, as for HTTP) is a refusal: PUBACK 0x97, QoS 0 dropped and counted, 3.1.1 closed.

device.max_subscriptions counts the filters a device subscribes on this connection. A device that comes back with Session Present=1 keeps at the broker the subscriptions of its previous connection, which the gateway does not see: bound them at the broker too (its per-client subscription limit), the gateway's bound being per connection.

listeners:
  mqtt:
    device:
      messages_per_second: 100      # past it: held, never refused
      bytes_per_second: 524288
      burst: 2                      # seconds of each rate a burst may take
    tenant:
      messages_per_second: 5000
      bytes_per_second: 20971520
      quota: plant-monthly          # a policy of type quota: 0x97 once the month is spent

Failover: shutdown and full replica

When a replica shuts down, each MQTT 5 device receives DISCONNECT 0x9C and the Server Reference (the load balancer's address by default); a 3.1.1 device is closed and reconnects through the load balancer. Its broker connection is held for handover_timeout: back through another replica, the broker takes its session over and no will is published; not back, the connection is closed with DISCONNECT 0x04 and the will goes - the device is really gone. Past redirect_above, a new device is redirected at its CONNECT. A replica lost abruptly cuts its connections: the broker publishes the wills, which are true.

server:
  shutdown:
    drain: 45s                      # covers the spread and the handover
    mqtt_drain_spread: 30s
listeners:
  mqtt:
    advertised_address: "mqtt.plant.example.com:8883"
    failover:
      mode: redirect                # off: 0x8B and the will at once
      reason: use_another_server    # server_moved (0x9D) for a planned migration
      redirect_above: 95            # % of max_connections: new devices go elsewhere first
      handover_timeout: 30s
      will_delay_on_handover: false # true: a will without delay gets handover_timeout (R11)

The pod's grace period follows server.shutdown.drain: cover the spread and the handover.

Will at takeover - Mosquitto, EMQX and HiveMQ CE publish the will of a session taken over when its Will Delay Interval is 0: that is MQTT 5's rule, not the gateway's. The takeover keeps the session (subscriptions, messages in flight); to keep the will too, an MQTT 5 device declares a Will Delay Interval longer than its reconnection and a Session Expiry Interval that covers it (20 s and 40 s for instance). A 3.1.1 device has no delay: its will goes at the takeover. flexway.mqtt.failover{outcome="taken_over"} counts these takeovers apart.

For devices you cannot change, failover.will_delay_on_handover: true (off by default) has the gateway write toward the broker a Will Delay Interval of handover_timeout, and a session expiry that covers it, when the will declares none: the will survives the takeover. The cost: the will of a device really lost goes up to handover_timeout later. No effect on an MQTT 3.1.1 device - its broker connection is 3.1.1, which has no will delay.

The chart

The container's ports, the Service and its NetworkPolicy follow listeners.mqtt and are rendered only with mqtt.enabled (off by default); the cert-manager Certificate with listeners.mqtt.tls. The certificate is mounted at /etc/flexway-mqtt-tls (tls.crt, tls.key, ca.crt). With mesh_only, the Service stays ClusterIP and mqtt.networkPolicy.from is required, mqtt.enabled or not: the process listens in clear either way. A LoadBalancer (or NodePort) Service gets externalTrafficPolicy: Local (mqtt.service.externalTrafficPolicy): the devices' addresses reach the gateway, and max_connections_per_ip bounds a device - Cluster hides them, then set max_connections_per_ip: 0. The WebSocket listener counts its sockets from their acceptance, before TLS and the upgrade.

mqtt:
  enabled: true                     # the Service; off by default
  service: { type: LoadBalancer }   # externalTrafficPolicy: Local by default
  tls:
    dnsNames: [mqtt.plant.example.com]
    issuerRef: { name: devices-ca, kind: ClusterIssuer }
  networkPolicy:
    from: [{ ipBlock: { cidr: 10.20.0.0/16 } }]

An address per replica (a Service per pod, for a Server Reference per replica) is not rendered: the load balancer is enough, and many clients ignore the reference.

Observing and simulating

Signal	What it says
`flexway.mqtt.connections{listener, tenant, protocol_version, auth}`	Devices connected: the occupation of `max_connections`
`flexway.mqtt.packets{type, decision, cause, tenant}`	Packets admitted, refused, dropped, and why
`flexway.mqtt.connections.refused{bound}`, `flexway.mqtt.connections.closed{cause}`	Refusals by bound; normal closures apart
`flexway.mqtt.publish.duration{tenant, qos}`	From the device's `PUBLISH` to the broker's acknowledgement
`flexway.mqtt.throttled`, `flexway.mqtt.throttled.devices`	Backpressure holds; devices held now
`flexway.mqtt.failover{reason, outcome}`	Redirections, takeovers, expired handovers
`flexway.mqtt.upstream.connections{outcome}`	Connections to the broker, circuit breaker included

A span per sampled packet; each refusal in the decision audit (audit.decisions). /config publishes listeners.mqtt and devices (the operations, their channels and bindings). POST /simulate/mqtt judges a described packet without sending anything to the broker:

curl -s localhost:9090/simulate/mqtt -d '{
  "packet": "subscribe", "topic": "$share/workers/plant/acme/fleet/#", "qos": 2,
  "identity": { "kind": "api_key", "id": "press-07" } }'
# {"verdict":"served","operation":"readFleet","upstream":"plant",
#  "written":"$share/acme~workers/plant/acme/fleet/#","qos":1,"client_id":"acme/apikey:press-07"}

Troubleshooting

What the device sees	Why	Where to look
`CONNACK 0x86` (3.1.1 `0x04`)	Key or token refused, a user name that is not the key's id, a token without `exp`	The log line `mqtt connection refused` says why; `flexway.mqtt.packets{type="connect",cause="authentication"}`
`CONNACK 0x87` (3.1.1 `0x05`)	Identity without a tenant, a certificate not declared in `auth.mtls`, no operation admits it, a will outside its topics	Check the key's or client's `operations`, the will channel's `x-flexway-bind`
`CONNACK 0x8C`	The client asks for enhanced authentication (`AUTH`)	Not served: key, token or certificate
`CONNACK 0x88` (3.1.1 `0x03`)	Broker unreachable or circuit breaker open	`flexway.mqtt.upstream.connections{outcome}`, the upstream's `brokers`
`CONNACK 0x97`, `0x9F`	`max_connections`, a tenant bound or `max_sessions_per_identity`; connection rate of the tenant or the identity	`flexway.mqtt.connections.refused{bound}`
3.1.1 return code `0x05` at once	A keep-alive of `0` or past `max_keepalive`	Set the device's keep-alive, or `allow_zero_keepalive`
`CONNACK 0x9C` + Server Reference	Replica past `redirect_above`	Expected: the client goes to `server_reference`
`PUBACK 0x87`	Topic outside its channels (`binding`, `not_authorized`), a `$` topic, a Response Topic outside its subscriptions	`POST /simulate/mqtt` with the topic and the identity
`PUBACK 0x99`	Payload refused by the schema (`x-flexway-validation: full`), or declared UTF-8 (Payload Format Indicator 1) and not	The contract's message
`PUBACK 0x97`	The tenant's day or month quota is spent - never a rate	`listeners.mqtt.tenant.quota`
`PUBACK`/`PUBREC 0x9B`	QoS past the operation's `qos_max`	`x-flexway-device.qos_max`
`SUBACK 0x87` (3.1.1 `0x80`)	Filter outside a channel it reads (`wildcard`, `binding`), `$share/` on a channel not `shared`	`POST /simulate/mqtt` with `packet: subscribe`
`DISCONNECT 0x93`, `0x94`, `0x95`	Receive Maximum exceeded, unknown alias or past the maximum, packet too large	The `device.*` bounds and `max_packet_bytes`
`DISCONNECT 0xA0`	The credential ended (token, key, certificate) or `max_session_lifetime`	The client reconnects with a fresh credential
`DISCONNECT 0x87`	A reload withdrew its key, client or operation	`/config`, the key store
`DISCONNECT 0x96`	More than `device.max_refusals_per_minute` packets refused	The refusals in `flexway.mqtt.packets{decision="refused"}` and the sampled audit
`DISCONNECT 0x8E`	Another connection of the same identity took the session over	`device.sessions: per_client_id` for several devices under one identity

MQTT devices: the listener in proxy in front of your broker