feat(outputs.mqtt): Add client trace logging, resolve MQTT5 reconnect login (#15429)

This commit is contained in:
Joshua Powers 2024-06-06 03:06:23 -06:00 committed by GitHub
parent 079c9d285a
commit f0c72586cc
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 49 additions and 27 deletions

View File

@ -35,6 +35,7 @@ type MqttConfig struct {
KeepAlive int64 `toml:"keep_alive"`
PersistentSession bool `toml:"persistent_session"`
PublishPropertiesV5 *PublishProperties `toml:"v5"`
ClientTrace bool `toml:"client_trace"`
tls.ClientConfig

View File

@ -1,4 +1,4 @@
package mqtt_consumer
package mqtt
import (
"github.com/influxdata/telegraf"

View File

@ -7,6 +7,7 @@ import (
mqttv3 "github.com/eclipse/paho.mqtt.golang" // Library that supports v3.1.1
"github.com/influxdata/telegraf/internal"
"github.com/influxdata/telegraf/logger"
)
type mqttv311Client struct {
@ -77,6 +78,14 @@ func NewMQTTv311Client(cfg *MqttConfig) (*mqttv311Client, error) {
opts.AddBroker(broker)
}
if cfg.ClientTrace {
log := &mqttLogger{logger.NewLogger("paho", "", "")}
mqttv3.ERROR = log
mqttv3.CRITICAL = log
mqttv3.WARN = log
mqttv3.DEBUG = log
}
return &mqttv311Client{
client: mqttv3.NewClient(opts),
timeout: time.Duration(cfg.Timeout),

View File

@ -12,6 +12,7 @@ import (
"github.com/influxdata/telegraf/config"
"github.com/influxdata/telegraf/internal"
"github.com/influxdata/telegraf/logger"
)
type mqttv5Client struct {
@ -22,6 +23,7 @@ type mqttv5Client struct {
timeout time.Duration
qos int
retain bool
clientTrace bool
properties *mqttv5.PublishProperties
}
@ -101,6 +103,7 @@ func NewMQTTv5Client(cfg *MqttConfig) (*mqttv5Client, error) {
qos: cfg.QoS,
retain: cfg.Retain,
properties: properties,
clientTrace: cfg.ClientTrace,
}, nil
}
@ -115,8 +118,14 @@ func (m *mqttv5Client) Connect() (bool, error) {
return false, fmt.Errorf("getting password failed: %w", err)
}
defer pass.Destroy()
m.options.ConnectUsername = user.TemporaryString()
m.options.ConnectPassword = pass.Bytes()
m.options.ConnectUsername = user.String()
m.options.ConnectPassword = []byte(pass.String())
if m.clientTrace {
log := mqttLogger{logger.NewLogger("paho", "", "")}
m.options.Debug = log
m.options.Errors = log
}
client, err := mqttv5auto.NewConnection(context.Background(), m.options)
if err != nil {

View File

@ -73,7 +73,6 @@ type MQTTConsumer struct {
Password config.Secret `toml:"password"`
QoS int `toml:"qos"`
ConnectionTimeout config.Duration `toml:"connection_timeout"`
ClientTrace bool `toml:"client_trace"`
MaxUndeliveredMessages int `toml:"max_undelivered_messages"`
PersistentSession bool `toml:"persistent_session"`
ClientID string `toml:"client_id"`
@ -105,14 +104,6 @@ func (m *MQTTConsumer) SetParser(parser telegraf.Parser) {
m.parser = parser
}
func (m *MQTTConsumer) Init() error {
if m.ClientTrace {
log := &mqttLogger{m.Log}
mqtt.ERROR = log
mqtt.CRITICAL = log
mqtt.WARN = log
mqtt.DEBUG = log
}
m.state = Disconnected
if m.PersistentSession && m.ClientID == "" {
return errors.New("persistent_session requires client_id")

View File

@ -102,6 +102,12 @@ to use them.
## actually reads it
# retain = false
## Client trace messages
## When set to true, and debug mode enabled in the agent settings, the MQTT
## client's messages are included in telegraf logs. These messages are very
## noisey, but essential for debugging issues.
# client_trace = false
## Layout of the topics published.
## The following choices are available:
## non-batch -- send individual messages, one for each metric

View File

@ -67,6 +67,12 @@
## actually reads it
# retain = false
## Client trace messages
## When set to true, and debug mode enabled in the agent settings, the MQTT
## client's messages are included in telegraf logs. These messages are very
## noisey, but essential for debugging issues.
# client_trace = false
## Layout of the topics published.
## The following choices are available:
## non-batch -- send individual messages, one for each metric