MQTTS support?

[ben-auo]

Does the Vector MQTT sink support MQTTS (MQTT over TLS)?

I see https://github.com/vectordotdev/vector/blob/master/src/sinks/websocket/sink.rs#L463 exists for wss but I don't see an equivalent on https://github.com/vectordotdev/vector/blob/master/src/sinks/mqtt/sink.rs

[jszwedko]

Hey! The mqtt sink does support TLS configurable via the https://vector.dev/docs/reference/configuration/sinks/mqtt/#tls options.

[ben-auo]

Thanks, @jszwedko. Is that just for self-signed certificates? I tried connecting to an MQTT server (Emitter) that's using a Let's Encrypt certificate and couldn't connect whereas MQTTX connected fine.

[ben-auo]

Some troubleshooting info:

• Both compute instances are running Ubuntu 22.04.5 LTS minimal with nothing else installed and fully updated
• All ports open and iptables configured to accept all traffic (iptables -I INPUT -j ACCEPT)

Emitter host was able to successfully retrieve and install a Let's Encrypt cert for the host domain:

Issued To
Common Name (CN)	my.realdomainhere.com
Organization (O)	<Not Part Of Certificate>
Organizational Unit (OU)	<Not Part Of Certificate>
Common Name (CN)	E6
Organization (O)	Let's Encrypt
Organizational Unit (OU)	<Not Part Of Certificate>
Issued On	Saturday, September 28, 2024 at 11:36:51 AM
Expires On	Friday, December 27, 2024 at 11:36:50 AM

MQTTX can connect, subscribe, and publish:

But Vector can't connect:

# tail -n20 /root/.vector/config/vector.yaml

# Print parsed logs to stdout
sinks:
  print:
    type: "mqtt"
    port: 443
    host: "my.realdomainhere.com"
    topic: "REDACTED...p9Q3j/db/"
    inputs: ["parse_logs"]
    encoding:
      codec: "json"
      json:
        pretty: true

# Vector's GraphQL API (disabled by default)
# Uncomment to try it out with the `vector top` command or
# in your browser at http://localhost:8686
# api:
#   enabled: true
#   address: "127.0.0.1:8686"

# RUST_BACKTRACE=full VECTOR_LOG=debug /root/.vector/bin/vector --config /root/.vector/config/vector.yaml
2024-09-28T19:46:24.884415Z DEBUG vector::app: Internal log rate limit configured. internal_log_rate_secs=10
2024-09-28T19:46:24.884465Z  INFO vector::app: Log level is enabled. level="debug"
2024-09-28T19:46:24.884561Z DEBUG vector::app: messaged="Building runtime." worker_threads=4
2024-09-28T19:46:24.885077Z  INFO vector::app: Loading configs. paths=["/root/.vector/config/vector.yaml"]
2024-09-28T19:46:24.885608Z DEBUG vector::config::loading: No secret placeholder found, skipping secret resolution.
2024-09-28T19:46:24.886637Z DEBUG vector::topology::builder: Building new source. component=dummy_logs
2024-09-28T19:46:24.886899Z DEBUG vector::topology::builder: Building new transform. component=parse_logs
2024-09-28T19:46:24.887190Z DEBUG vector::topology::builder: Building new sink. component=print
2024-09-28T19:46:24.887375Z  INFO vector::topology::running: Running healthchecks.
2024-09-28T19:46:24.887396Z DEBUG vector::topology::running: Connecting changed/added component(s).
2024-09-28T19:46:24.887408Z DEBUG vector::topology::running: Configuring outputs for source. component=dummy_logs
2024-09-28T19:46:24.887446Z  INFO vector::topology::builder: Healthcheck passed.
2024-09-28T19:46:24.887786Z DEBUG vector::topology::running: Configuring output for component. component=dummy_logs output_id=None
2024-09-28T19:46:24.887802Z DEBUG vector::topology::running: Configuring outputs for transform. component=parse_logs
2024-09-28T19:46:24.887813Z DEBUG vector::topology::running: Configuring output for component. component=parse_logs output_id=None
2024-09-28T19:46:24.887827Z DEBUG vector::topology::running: Connecting inputs for transform. component=parse_logs
2024-09-28T19:46:24.887842Z DEBUG vector::topology::running: Adding component input to fanout. component=parse_logs fanout_id=dummy_logs
2024-09-28T19:46:24.887854Z DEBUG vector::topology::running: Connecting inputs for sink. component=print
2024-09-28T19:46:24.887869Z DEBUG vector::topology::running: Adding component input to fanout. component=print fanout_id=parse_logs
2024-09-28T19:46:24.887884Z DEBUG vector::topology::running: Spawning new source. key=dummy_logs
2024-09-28T19:46:24.887916Z DEBUG vector::topology::running: Spawning new transform. key=parse_logs
2024-09-28T19:46:24.887964Z  INFO vector: Vector has started. debug="false" version="0.41.1" arch="aarch64" revision="745babd 2024-09-11 14:55:36.802851761"
2024-09-28T19:46:24.887974Z DEBUG source{component_kind="source" component_id=dummy_logs component_type=demo_logs}: vector::topology::builder: Source pump supervisor starting.
2024-09-28T19:46:24.888005Z  INFO vector::app: API is disabled, enable by setting `api.enabled` to `true` and use commands like `vector top`.
2024-09-28T19:46:24.888011Z DEBUG source{component_kind="source" component_id=dummy_logs component_type=demo_logs}: vector::topology::builder: Source pump starting.
2024-09-28T19:46:24.888021Z DEBUG source{component_kind="source" component_id=dummy_logs component_type=demo_logs}: vector::topology::builder: Source starting.
2024-09-28T19:46:24.888021Z DEBUG transform{component_kind="transform" component_id=parse_logs component_type=remap}: vector::topology::builder: Synchronous transform starting.
2024-09-28T19:46:24.888076Z DEBUG sink{component_kind="sink" component_id=print component_type=mqtt}: vector::topology::builder: Sink starting.
2024-09-28T19:46:24.889815Z DEBUG sink{component_kind="sink" component_id=print component_type=mqtt}: vector::utilization: utilization=0.15303463050438948
2024-09-28T19:46:24.891479Z ERROR vector::internal_events::mqtt: MQTT connection error. error=I/O: connection closed by peer error_code="mqtt_connection_error" error_type="writer_failed" stage="sending" internal_log_rate_limit=true
2024-09-28T19:46:24.894032Z ERROR vector::internal_events::mqtt: Internal log [MQTT connection error.] is being suppressed to avoid flooding.
2024-09-28T19:46:29.889023Z DEBUG sink{component_kind="sink" component_id=print component_type=mqtt}: vector::utilization: utilization=0.015306789796130267
2024-09-28T19:46:29.889619Z DEBUG transform{component_kind="transform" component_id=parse_logs component_type=remap}: vector::utilization: utilization=0.00024281765727596216
2024-09-28T19:46:34.888671Z DEBUG sink{component_kind="sink" component_id=print component_type=mqtt}: vector::utilization: utilization=0.001534149624532283
2024-09-28T19:46:34.893637Z ERROR vector::internal_events::mqtt: Internal log [MQTT connection error.] has been suppressed 4326 times.
2024-09-28T19:46:34.893661Z ERROR vector::internal_events::mqtt: MQTT connection error. error=I/O: connection closed by peer error_code="mqtt_connection_error" error_type="writer_failed" stage="sending" internal_log_rate_limit=true
2024-09-28T19:46:34.896073Z ERROR vector::internal_events::mqtt: Internal log [MQTT connection error.] is being suppressed to avoid flooding.
2024-09-28T19:46:35.889227Z DEBUG transform{component_kind="transform" component_id=parse_logs component_type=remap}: vector::utilization: utilization=0.00008763406537716234
^C2024-09-28T19:46:37.588380Z  INFO vector::signal: Signal received. signal="SIGINT"
2024-09-28T19:46:37.588450Z  INFO vector: Vector has stopped.
2024-09-28T19:46:37.588718Z  INFO vector::topology::running: Shutting down... Waiting on running components. remaining_components="dummy_logs, parse_logs, print" time_remaining="59 seconds left"
2024-09-28T19:46:37.889048Z DEBUG source{component_kind="source" component_id=dummy_logs component_type=demo_logs}: vector::topology::builder: Source finished normally.
2024-09-28T19:46:37.889101Z DEBUG source{component_kind="source" component_id=dummy_logs component_type=demo_logs}: vector::topology::builder: Source pump finished normally.
2024-09-28T19:46:37.889151Z DEBUG source{component_kind="source" component_id=dummy_logs component_type=demo_logs}: vector::topology::builder: Source pump supervisor task finished normally.
2024-09-28T19:46:37.889276Z DEBUG transform{component_kind="transform" component_id=parse_logs component_type=remap}: vector::topology::builder: Synchronous transform finished normally.
2024-09-28T19:46:37.889585Z DEBUG sink{component_kind="sink" component_id=print component_type=mqtt}: vector::topology::builder: Sink finished normally.

One guess is since Emitter requires MQTT version 3.1.1 and Vector supports MQTT version 5.0 perhaps that's the issue? Is there a way to tell Vector's sink to use MQTT version 3.1.1?

[ben-auo]

Using the https://www.emqx.com/en/blog/how-to-use-mqtt-in-rust guide, I created a simple rumqtt program and tried to connect to the Emitter host and got this error message, in case it helps:

0. Notification = Io(Custom { kind: ConnectionAborted, error: "connection closed by peer" })

[ben-auo]

Using MQTT CLI, I was able to identify the key conditions where my MQTT broker (Emitter) works and doesn't work:

# mqtt pub -t 'REDACTED...p9Q3j/db/' -m '{"foo":"bar"}' -h 'my.realdomainhere.com' -p 443 -V 3
Unable to connect. Reason: 'Server closed connection without DISCONNECT.'

# mqtt pub -t 'REDACTED...p9Q3j/db/' -m '{"foo":"bar"}' -h 'my.realdomainhere.com' -p 443 -V 3 --capath '/etc/ssl/certs'

# mqtt pub -t 'REDACTED...p9Q3j/db/' -m '{"foo":"bar"}' -h 'my.realdomainhere.com' -p 443 -V 5
Unable to connect. Reason: 'Server closed connection without DISCONNECT.'

# mqtt pub -t 'REDACTED...p9Q3j/db/' -m '{"foo":"bar"}' -h 'my.realdomainhere.com' -p 443 -V 5 --capath '/etc/ssl/certs'
Unable to connect. Reason: 'Exception while decoding CONNACK: remaining length too short'

As inferred by my MTTQX experience, Emitter only works with protocol 3.x, not 5.x. In addition, the client must be made aware of the local system certificate store is required.

Since I'm not sure how to force Vector to use MQTT protocol 3.x I can't troubleshoot the certificate store so I'll just close this for now.