Support proxy for MQTT Protocol

[thehellmaker]

Title: Support proxy for MQTT Protocol

Description:

The idea here is to setup MQTT Broker behind Envoy proxy load balancing between different nodes of MQTT cluster in our case a multi node EMQX MQTT Broker Cluster. Functional Requirements:

1. Since MQTT brokers have a large number of devices connected all the time (Single node 4GB RAM machine can have over 50000 connections) envoy proxy also should support large number of connection.
2. Support TLS termination for MQTT protocol
3. For an IOT usecase each device will be provisioned with its own unique TLS certificate for mTLS. The proxy should have support for CRL and OCSP validation of revoked certificates for Server and Client Side Certificates
4. Load balancing between multiple nodes of a multi node MQTT Broker Cluster

[optional Relevant Links:]

NGINX has stream support which supports MQTT here are the documents for the same http://nginx.org/en/docs/stream/ngx_stream_core_module.html https://www.emqx.io/docs/en/latest/deploy/cluster/lb-nginx.html#:~:text=Using%20Nginx%20to%20load%20balance,behalf%20of%20the%20EMQX%20cluster.

[domWinter]

Hi,

I am currently trying to setup envoy as tcp proxy for mqtt and facing problems.

docker-compose.yaml:

services:
  mqtt:
    image: eclipse-mosquitto:2.0.18
    ports:
      - "1883:1883"
    networks:
      - envoymesh
  envoy:
    image: envoyproxy/envoy:v1.30-latest
    hostname: envoy
    entrypoint: ["envoy", "-c", "/etc/envoy/envoy.yaml", "--log-level", "trace"]
    ports:
      - "10000:10000"
    volumes:
      - ./envoy.yaml:/etc/envoy/envoy.yaml
    networks:
      - envoymesh
networks:
  envoymesh: {}

envoy.yaml

static_resources:
  listeners:
    address:
      socket_address:
        address: 0.0.0.0
        port_value: 10000
    filter_chains:
      - filters:
          - name: envoy.filters.network.tcp_proxy
            typed_config:
              "@type": type.googleapis.com/envoy.extensions.filters.network.tcp_proxy.v3.TcpProxy
              stat_prefix: mqtt_server
              cluster: mqtt-server
              idle_timeout: 60s
  clusters:
    - name: mqtt-server
      connect_timeout: 2s
      type: strict_dns
      lb_policy: round_robin
      load_assignment:
        cluster_name: mqtt-service
        endpoints:
          - lb_endpoints:
            - endpoint:
                address:
                  socket_address:
                    address: mqtt
                    port_value: 1883

Python Client

#!python3
import paho.mqtt.client as mqtt
import logging

logging.basicConfig(level=logging.DEBUG)
mqttc = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)
mqttc.enable_logger()

def on_connect(client, userdata, flags, reason_code, properties):
    print(f"Connected with result code {reason_code}")
    client.subscribe("$SYS/#")

def on_message(client, userdata, msg):
    print(msg.topic+" "+str(msg.payload))

mqttc.on_connect = on_connect
mqttc.on_message = on_message

broker="127.0.0.1"
port=10000
mqttc.connect(broker, port, 60)
mqttc.loop_forever()

When trying to connect, envoy logs the following:

envoy-1  | [2024-07-24 12:08:58.029][27][debug][filter] [source/common/tcp_proxy/tcp_proxy.cc:264] [Tags: "ConnectionId":"0"] new tcp proxy session
envoy-1  | [2024-07-24 12:08:58.029][27][trace][connection] [source/common/network/connection_impl.cc:406] [Tags: "ConnectionId":"0"] readDisable: disable=true disable_count=0 state=0 buffer_length=0
envoy-1  | [2024-07-24 12:08:58.029][27][debug][filter] [source/common/tcp_proxy/tcp_proxy.cc:459] [Tags: "ConnectionId":"0"] Creating connection to cluster mqtt-server
envoy-1  | [2024-07-24 12:08:58.029][27][debug][misc] [source/common/upstream/cluster_manager_impl.cc:2327] Allocating TCP conn pool
envoy-1  | [2024-07-24 12:08:58.029][27][debug][pool] [source/common/conn_pool/conn_pool_base.cc:291] trying to create new connection
envoy-1  | [2024-07-24 12:08:58.029][27][trace][pool] [source/common/conn_pool/conn_pool_base.cc:292] ConnPoolImplBase 0x277fbfc2dcc0, ready_clients_.size(): 0, busy_clients_.size(): 0, connecting_clients_.size(): 0, connecting_stream_capacity_: 0, num_active_streams_: 0, pending_streams_.size(): 1 per upstream preconnect ratio: 1
envoy-1  | [2024-07-24 12:08:58.029][27][debug][pool] [source/common/conn_pool/conn_pool_base.cc:145] creating a new connection (connecting=0)
envoy-1  | [2024-07-24 12:08:58.029][27][debug][connection] [source/common/network/connection_impl.cc:1021] [Tags: "ConnectionId":"1"] connecting to 172.21.0.3:1883
envoy-1  | [2024-07-24 12:08:58.029][27][debug][connection] [source/common/network/connection_impl.cc:1040] [Tags: "ConnectionId":"1"] connection in progress
envoy-1  | [2024-07-24 12:08:58.029][27][trace][pool] [source/common/conn_pool/conn_pool_base.cc:131] not creating a new connection, shouldCreateNewConnection returned false.
envoy-1  | [2024-07-24 12:08:58.029][27][trace][connection] [source/common/network/connection_impl.cc:474] [Tags: "ConnectionId":"0"] raising connection event 2
envoy-1  | [2024-07-24 12:08:58.029][27][trace][filter] [source/common/tcp_proxy/tcp_proxy.cc:805] [Tags: "ConnectionId":"0"] on downstream event 2, has upstream = false
envoy-1  | [2024-07-24 12:08:58.029][27][debug][conn_handler] [source/common/listener_manager/active_tcp_listener.cc:160] [Tags: "ConnectionId":"0"] new connection from 172.21.0.1:36514
envoy-1  | [2024-07-24 12:08:58.029][27][trace][misc] [source/common/network/tcp_listener_impl.cc:114] TcpListener accepted 1 new connections.
envoy-1  | [2024-07-24 12:08:58.029][27][trace][connection] [source/common/network/connection_impl.cc:619] [Tags: "ConnectionId":"0"] socket event: 2
envoy-1  | [2024-07-24 12:08:58.029][27][trace][connection] [source/common/network/connection_impl.cc:742] [Tags: "ConnectionId":"0"] write ready
envoy-1  | [2024-07-24 12:08:58.029][27][trace][connection] [source/common/network/connection_impl.cc:619] [Tags: "ConnectionId":"1"] socket event: 3
envoy-1  | [2024-07-24 12:08:58.029][27][trace][connection] [source/common/network/connection_impl.cc:742] [Tags: "ConnectionId":"1"] write ready
envoy-1  | [2024-07-24 12:08:58.029][27][debug][connection] [source/common/network/connection_impl.cc:762] [Tags: "ConnectionId":"1"] delayed connect error: 111
envoy-1  | [2024-07-24 12:08:58.029][27][debug][connection] [source/common/network/connection_impl.cc:281] [Tags: "ConnectionId":"1"] closing socket: 0
envoy-1  | [2024-07-24 12:08:58.029][27][trace][connection] [source/common/network/connection_impl.cc:474] [Tags: "ConnectionId":"1"] raising connection event 0
envoy-1  | [2024-07-24 12:08:58.029][27][debug][pool] [source/common/conn_pool/conn_pool_base.cc:495] [Tags: "ConnectionId":"1"] client disconnected, failure reason: delayed connect error: 111