Flexway - Appareils MQTT : l'écouteur en proxy devant… (Exploitation)

Guide d'exploitation : configurer, déployer et observer la passerelle. Ce que fait la fonctionnalité côté produit : guide produit ; brancher un appareil : guide développeur.

Capteurs, automates et passerelles d'atelier se connectent à la passerelle en MQTT 5 ou 3.1.1, sur TLS ou WebSocket, au lieu de votre courtier. La passerelle est un proxy, jamais un courtier : chaque appareil a sa connexion à votre courtier (Mosquitto, EMQX, HiveMQ...), ouverte au nom de la passerelle ; chaque paquet est jugé contre le contrat AsyncAPI - qui est l'appareil, quels topics son tenant et son identité ouvrent, ses débits et son quota - puis relayé tel quel. Sessions, messages retenus, testaments et QoS 2 restent chez votre courtier. Disponible depuis 1.7.0.

Sur cet écouteur : Sparkplug B et OPC UA PubSub JSON ; l'écran Appareils de la console : Console : l'écran Appareils. Les défauts max_connections, max_connections_per_ip et max_connection_rate_per_ip sont provisoires, en attendant les mesures de performance : écrivez-les si vous en dépendez.

# 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 } ]

Déclarer ce qu'un appareil peut faire

Une opération AsyncAPI marquée x-flexway-device ouvre son canal aux appareils, dans son sens : send, l'appareil publie ; receive, il s'abonne. Rien d'autre ne passe.

Clé de `x-flexway-device`	Défaut	Effet
`qos_max`	2	QoS la plus haute ; un `PUBLISH` au-delà est refusé `0x9B`, un abonnement accordé à elle
`messages_per_second`, `bytes_per_second`	ceux de `listeners.mqtt.device`	Les débits d'un `PUBLISH` sur ce canal (`send`)
`shared`	`false`	Admet `$share/<groupe>/<filtre>` sur un canal `receive` ; le groupe est écrit `<tenant>~<groupe>` chez le courtier

Chaque paramètre du canal est lié à un fait vérifié (x-flexway-bind) et l'un d'eux au tenant - refusé au chargement sinon : tenant, subject (l'identifiant de la clé, le sujet du jeton, l'identifiant du client auth.mtls), claim:<nom>, ou un champ du certificat de l'appareil, cert:<champ> : cn, serial, fingerprint (SHA-256 du certificat, hexadécimal), ou avec son indice ou[i], o[i], san.dns[i], san.uri[i], san.ip[i], san.email[i]. Une valeur vide ou qui contient /, +, #, $, {, } ou un caractère nul laisse l'opération fermée : jamais écrite dans un topic, et une valeur écrite n'est jamais relue comme paramètre. Un joker (+, #) s'écrit dans un canal receive seulement, après les niveaux liés : un filtre d'appareil doit y être inclus.

Tous les canaux d'appareils écrivent le tenant au même niveau, après le même préfixe littéral (plant/{site}/... partout) - refusé au chargement sinon : un canal avec le tenant au niveau 1 et un autre au niveau 2 laisseraient le tenant nommé comme un niveau littéral de l'un lire les topics de l'autre.

Identifier les appareils

L'écouteur lit, dans l'ordre : le certificat vérifié par tls.client_ca_file et déclaré dans auth.mtls, sinon le mot de passe du CONNECT - une clé d'API dont l'identifiant est le nom d'utilisateur, sinon un jeton (exp obligatoire ; à son terme, DISCONNECT 0xA0). Une alternative de la security de l'opération nomme ce qu'elle admet : X509, userPassword (ou plain, apiKey in: password), oauth2/openIdConnect.

Certificats d'appareil (TLS obligatoire) :

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 } ]

Jetons d'un fournisseur OIDC, le tenant dans une claim :

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

Chez votre courtier, l'appareil a le client id <tenant>/<mécanisme>:<identité> (acme/apikey:press-07), jamais celui qu'il a choisi : un appareil d'un tenant n'évince jamais celui d'un autre. Celui d'un jeton porte une empreinte de son émetteur, acme/oidc:<8 hex>:<sujet> : deux émetteurs qui délivrent le même sujet ne partagent jamais une session. Une seconde connexion de la même identité reprend la session (0x8E à la première) ; device.sessions: per_client_id donne une session par client id de l'appareil sous une identité partagée, au plus device.max_sessions_per_identity à la fois (0x97 au-delà).

subject et plusieurs émetteurs - x-flexway-bind: subject lie le sujet seul : deux émetteurs du même tenant qui délivrent le même sujet atteignent les mêmes topics. Donnez un émetteur par tenant, ou liez une claim qu'un seul émetteur pose (claim:<nom>). Une liaison issuer est proposée pour un lot ultérieur, non construite.

Keep-alive. Un appareil MQTT 5 reçoit max_keepalive (Server Keep Alive) quand il demande plus, ou rien. MQTT 3.1.1 ne permet pas de l'imposer : un appareil qui demande plus que max_keepalive est refusé 0x05, celui qui ne demande rien (0) aussi, sauf avec allow_zero_keepalive: true - un appareil muet ne serait jamais vu parti. Toute connexion prend fin après max_session_lifetime (24 h) quel que soit son justificatif, et se reconnecte avec un justificatif vérifié à nouveau.

Révocation. À chaque rechargement, chaque connexion ouverte est vérifiée de nouveau en moins d'une demi-seconde contre la nouvelle configuration - un appareil qui ne fait qu'écouter aussi : une clé, un client ou une opération retirés y mettent fin par DISCONNECT 0x87, et à sa connexion au courtier par DISCONNECT 0x00, si bien que le courtier écarte son testament (un appareil révoqué n'est pas un appareil perdu). Chaque livraison du courtier est jugée sur la configuration en vigueur.

Réglages et défauts

Chaque défaut est réglable ; les bornes s'appliquent au paquet ou au CONNECT suivant, sauf l'adresse, TLS, mesh_only, le WebSocket et max_connections, lus au démarrage.

Clé (`listeners.mqtt.`)	Défaut	Pourquoi
`address`, `tls`, `mesh_only`	aucun écouteur sans déclaration ; TLS ou `mesh_only`	Le jeton ou la clé d'un appareil ne traverse jamais le réseau en clair hors d'un maillage
`broker_mesh_only`	`false`	Un courtier en clair (`mqtt://`, `ws://`) est refusé au chargement hors `mesh_only` : à poser quand un maillage chiffre le saut vers le courtier
`websocket.path`	`/mqtt`	Le chemin des clients MQTT sur WebSocket (navigateurs, passerelles d'atelier)
`max_packet_bytes`	256 Kio	Un paquet est lu en entier avant d'être jugé : la borne de mémoire par paquet
`connect_timeout`	`5s`	Un `CONNECT` lent ou une poignée TLS qui traîne ne tient pas une place
`max_connections`	50 000 par réplique	Provisoire : à confirmer au banc M1 ; au-delà `0x97`
`max_unauthenticated_connections`	1 024	Une rafale sans `CONNECT` ne prive pas les appareils authentifiés
`max_connections_per_ip`, `max_connection_rate_per_ip`	64 ; 20 par seconde	Provisoire (banc M1) ; à poser à `0` derrière un équilibreur ou un proxy qui masque les adresses (le chart les garde : `externalTrafficPolicy: Local`) - sinon tous les appareils partagent une adresse et une borne
`max_connections_per_tenant`, `max_connection_rate_per_tenant`	10 000 ; 100 par seconde	Un tenant ne prend pas la réplique aux autres ; une tempête de reconnexions est étalée
`max_keepalive`, `max_session_expiry`, `topic_alias_maximum`	1 200 s ; 1 h ; 10 (1 000 au plus)	Les bornes des courtiers du marché (AWS, Event Grid, HiveMQ) ; un alias n'est retenu qu'une fois son topic accordé
`allow_zero_keepalive`	`false`	Un appareil 3.1.1 qui ne demande aucun keep-alive est refusé : il ne serait jamais vu parti
`max_session_lifetime`	24 h (1 min à 30 jours)	Une connexion ne survit jamais à une révocation que le rechargement manquerait
`device.messages_per_second`, `bytes_per_second`, `burst`	100 ; 512 Kio ; 2	Ceux d'AWS IoT Core ; contre-pression au-delà, jamais un refus
`device.receive_maximum`, `max_subscriptions`, `max_filters_per_subscribe`	100 ; 50 ; 10	Bornent ce qu'un appareil tient en vol et souscrit
`device.connect_rate_per_identity`	1 par seconde	Un appareil qui boucle ne s'évince pas lui-même sans fin (`0x9F`) ; compté sur l'identité vérifiée
`device.max_refusals_per_minute`	60	Chaque refus dépense le débit de messages de l'appareil, son audit est échantillonné (une entrée par seconde et par cause) ; au-delà `DISCONNECT 0x96`
`device.max_user_properties_bytes`, `max_correlation_data_bytes`	8 Kio ; 8 Kio	Ce qu'un paquet porte à côté de sa charge
`device.sessions`	`single`	Une session par identité : un second appareil évince le premier (`0x8E`)
`device.max_sessions_per_identity`	10 (ou `max_connections` s'il est plus bas)	Avec `per_client_id`, une identité ne prend pas la réplique (`0x97`)
`tenant.messages_per_second`, `bytes_per_second`	5 000 ; 20 Mio	Le plafond d'un site entier ; le plan d'abonnement viendra en `1.9.0`
`failover.mode`, `reason`	`redirect` ; `use_another_server`	Un arrêt redirige plutôt que de couper ; `0x9C` est temporaire
`failover.server_reference`	`advertised_address`	Toujours joignable, même par un client qui ignore la référence
`failover.redirect_above`, `handover_timeout`	95 % ; 30 s	Redirige avant la saturation ; laisse revenir un client courant avec recul
`failover.will_delay_on_handover`	`false`	Activé : un testament MQTT 5 sans délai survit à une reprise, celui d'un appareil perdu part jusqu'à `handover_timeout` plus tard
`server.shutdown.mqtt_drain_spread`	30 s (au plus la moitié de `drain`)	Pas de tempête de reconnexions sur les autres répliques
`observability.mqtt.closures`	`true`	Compter les fermetures normales à part des refus

Débits, contre-pression et quota

Au-delà d'un débit de l'appareil ou de son tenant, la passerelle cesse de lire la socket de l'appareil le temps nécessaire - jamais un refus, jamais une fermeture qui déclencherait son testament - et tient le keep-alive vers le courtier pendant ce temps. Le quota jour ou mois du tenant (une politique quota, comme en HTTP) est, lui, un refus : PUBACK 0x97, QoS 0 jeté et compté, fermeture en 3.1.1.

device.max_subscriptions compte les filtres qu'un appareil souscrit sur cette connexion. Un appareil qui revient avec Session Present=1 garde chez le courtier les abonnements de sa connexion précédente, que la passerelle ne voit pas : bornez-les aussi au courtier (sa limite d'abonnements par client), la borne de la passerelle étant par connexion.

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

Bascule : arrêt et réplique pleine

À l'arrêt d'une réplique, chaque appareil MQTT 5 reçoit DISCONNECT 0x9C et la Server Reference (l'adresse de l'équilibreur par défaut) ; un appareil 3.1.1 est fermé et se reconnecte par l'équilibreur. Sa connexion au courtier est tenue handover_timeout : revenu par une autre réplique, le courtier reprend sa session et aucun testament ne part ; pas revenu, la connexion est fermée par DISCONNECT 0x04 et le testament part - l'appareil est vraiment perdu. Au-delà de redirect_above, un nouvel appareil est redirigé dès son CONNECT. Une réplique perdue brutalement coupe ses connexions : le courtier publie les testaments, vrais.

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 : un testament sans délai reçoit handover_timeout (R11)

Le délai de grâce du pod suit server.shutdown.drain : couvrez l'étalement et la reprise.

Testament à la reprise - Mosquitto, EMQX et HiveMQ CE publient le testament d'une session reprise quand son délai de testament (Will Delay Interval) est nul : c'est la règle de MQTT 5, pas la passerelle. La reprise garde la session (abonnements, messages en vol) ; pour garder aussi le testament, un appareil MQTT 5 déclare un délai de testament supérieur à sa reconnexion et une expiration de session qui le couvre (20 s et 40 s par exemple). Un appareil 3.1.1 n'a pas de délai : son testament part à la reprise. flexway.mqtt.failover{outcome="taken_over"} compte ces reprises à part.

Pour des appareils que vous ne pouvez pas modifier, failover.will_delay_on_handover: true (désactivé par défaut) fait écrire par la passerelle vers le courtier un délai de testament égal à handover_timeout, et une expiration de session qui le couvre, quand le testament n'en déclare pas : le testament survit à la reprise. La contrepartie : le testament d'un appareil vraiment perdu part jusqu'à handover_timeout plus tard. Sans effet sur un appareil MQTT 3.1.1 - sa connexion au courtier est en 3.1.1, sans délai de testament.

Le chart

Les ports du conteneur, le Service et sa NetworkPolicy suivent listeners.mqtt et sont rendus seulement avec mqtt.enabled (désactivé par défaut) ; le Certificate cert-manager avec listeners.mqtt.tls. Le certificat est monté sous /etc/flexway-mqtt-tls (tls.crt, tls.key, ca.crt). Avec mesh_only, le Service reste ClusterIP et mqtt.networkPolicy.from est exigé, mqtt.enabled ou non : le processus écoute en clair dans les deux cas. Un Service LoadBalancer (ou NodePort) reçoit externalTrafficPolicy: Local (mqtt.service.externalTrafficPolicy) : les adresses des appareils parviennent à la passerelle, et max_connections_per_ip borne un appareil - Cluster les masque, posez alors max_connections_per_ip: 0. L'écouteur WebSocket compte ses sockets dès leur acceptation, avant TLS et la montée en WebSocket.

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

Une adresse par réplique (un Service par pod, pour une Server Reference par réplique) n'est pas rendue : l'équilibreur suffit, et beaucoup de clients ignorent la référence.

Observer et simuler

Signal	Ce qu'il dit
`flexway.mqtt.connections{listener, tenant, protocol_version, auth}`	Appareils connectés : l'occupation de `max_connections`
`flexway.mqtt.packets{type, decision, cause, tenant}`	Paquets admis, refusés, jetés, et pourquoi
`flexway.mqtt.connections.refused{bound}`, `flexway.mqtt.connections.closed{cause}`	Refus par borne ; fermetures normales à part
`flexway.mqtt.publish.duration{tenant, qos}`	Du `PUBLISH` de l'appareil à l'accusé du courtier
`flexway.mqtt.throttled`, `flexway.mqtt.throttled.devices`	Retenues de contre-pression ; appareils retenus maintenant
`flexway.mqtt.failover{reason, outcome}`	Redirections, reprises, reprises expirées
`flexway.mqtt.upstream.connections{outcome}`	Connexions au courtier, coupe-circuit compris

Un span par paquet échantillonné ; chaque refus dans l'audit des décisions (audit.decisions). /config publie listeners.mqtt et devices (les opérations, leurs canaux et leurs liaisons). POST /simulate/mqtt juge un paquet décrit sans rien envoyer au courtier :

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"}

Dépannage

Ce que voit l'appareil	Pourquoi	Où regarder
`CONNACK 0x86` (3.1.1 `0x04`)	Clé ou jeton refusé, nom d'utilisateur qui n'est pas l'identifiant de la clé, jeton sans `exp`	Le journal `mqtt connection refused` dit la cause ; `flexway.mqtt.packets{type="connect",cause="authentication"}`
`CONNACK 0x87` (3.1.1 `0x05`)	Identité sans tenant, certificat non déclaré dans `auth.mtls`, aucune opération ne l'admet, testament hors de ses topics	Vérifier `operations` de la clé ou du client, `x-flexway-bind` du canal du testament
`CONNACK 0x8C`	Le client demande une authentification étendue (`AUTH`)	Non servie : clé, jeton ou certificat
`CONNACK 0x88` (3.1.1 `0x03`)	Courtier injoignable ou coupe-circuit ouvert	`flexway.mqtt.upstream.connections{outcome}`, l'amont `brokers`
`CONNACK 0x97`, `0x9F`	`max_connections`, une borne du tenant ou `max_sessions_per_identity` ; débit de connexions du tenant ou de l'identité	`flexway.mqtt.connections.refused{bound}`
Code de retour 3.1.1 `0x05` d'emblée	Un keep-alive de `0` ou au-delà de `max_keepalive`	Régler le keep-alive de l'appareil, ou `allow_zero_keepalive`
`CONNACK 0x9C` + `Server Reference`	Réplique au-delà de `redirect_above`	Normal : le client va à `server_reference`
`PUBACK 0x87`	Topic hors de ses canaux (`binding`, `not_authorized`), topic `$`, Response Topic hors de ses abonnements	`POST /simulate/mqtt` avec le topic et l'identité
`PUBACK 0x99`	Charge refusée par le schéma (`x-flexway-validation: full`), ou déclarée UTF-8 (Payload Format Indicator 1) sans l'être	Le message du contrat
`PUBACK 0x97`	Quota jour ou mois du tenant épuisé - jamais un débit	`listeners.mqtt.tenant.quota`
`PUBACK`/`PUBREC 0x9B`	QoS au-delà du `qos_max` de l'opération	`x-flexway-device.qos_max`
`SUBACK 0x87` (3.1.1 `0x80`)	Filtre hors d'un canal lu (`wildcard`, `binding`), `$share/` sur un canal non `shared`	`POST /simulate/mqtt` avec `packet: subscribe`
`DISCONNECT 0x93`, `0x94`, `0x95`	Receive Maximum dépassé, alias inconnu ou au-delà du maximum, paquet trop grand	Les bornes `device.*` et `max_packet_bytes`
`DISCONNECT 0xA0`	Fin du justificatif (jeton, clé, certificat) ou de `max_session_lifetime`	Le client se reconnecte avec un justificatif neuf
`DISCONNECT 0x87`	Un rechargement a retiré sa clé, son client ou son opération	`/config`, le magasin de clés
`DISCONNECT 0x96`	Plus de `device.max_refusals_per_minute` paquets refusés	Les refus dans `flexway.mqtt.packets{decision="refused"}` et l'audit échantillonné
`DISCONNECT 0x8E`	Une autre connexion de la même identité a repris la session	`device.sessions: per_client_id` pour plusieurs appareils sous une identité

Appareils MQTT : l'écouteur en proxy devant votre courtier