Add TLS, QoS and retain options to the MQTT receiver

receivers/mqtt/client.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/tls"
 	"errors"
+	"fmt"
 	"time"
 
 	mqttLib "github.com/at-wat/mqtt-go"
@@ -63,10 +64,27 @@ func (c *mqttClient) Publish(ctx context.Context, message message) error {
 		return errors.New("failed to publish: client is not connected to the broker")
 	}
 
+	var mqttQoS mqttLib.QoS
+	var err error
+	switch message.qos {
+	case 0:
+		mqttQoS = mqttLib.QoS0
+	case 1:
+		mqttQoS = mqttLib.QoS1
+	case 2:
+		mqttQoS = mqttLib.QoS2
+	default:
+		err = fmt.Errorf("failed to publish: invalid QoS level %d", message.qos)
+	}
+
+	if err != nil {
+		return err
+	}
+
 	return c.client.Publish(ctx, &mqttLib.Message{
 		Topic:   message.topic,
-		QoS:     mqttLib.QoS0,
-		Retain:  false,
+		QoS:     mqttQoS,
+		Retain:  message.retain,
 		Payload: message.payload,
 	})
 }

receivers/mqtt/client_test.go

@@ -58,11 +58,15 @@ func TestMqttClientPublish(t *testing.T) {
 		name    string
 		topic   string
 		payload []byte
+		retain  bool
+		qos     int
 	}{
 		{
 			name:    "Simple publish",
 			topic:   "test",
 			payload: []byte("test"),
+			retain:  true,
+			qos:     1,
 		},
 	}
 
@@ -73,17 +77,31 @@ func TestMqttClientPublish(t *testing.T) {
 				client: mc,
 			}
 
+			var expectedQoS mqttLib.QoS
+			switch tc.qos {
+			case 0:
+				expectedQoS = mqttLib.QoS0
+			case 1:
+				expectedQoS = mqttLib.QoS1
+			case 2:
+				expectedQoS = mqttLib.QoS2
+			default:
+				require.Fail(t, "invalid QoS level")
+			}
+
 			ctx := context.Background()
 			mc.On("Publish", ctx, &mqttLib.Message{
 				Topic:   tc.topic,
 				Payload: tc.payload,
-				QoS:     mqttLib.QoS0,
-				Retain:  false,
+				QoS:     expectedQoS,
+				Retain:  tc.retain,
 			}).Return(nil)
 
 			err := c.Publish(ctx, message{
 				topic:   tc.topic,
 				payload: tc.payload,
+				retain:  tc.retain,
+				qos:     tc.qos,
 			})
 
 			require.NoError(t, err)

receivers/mqtt/config.go

@@ -16,14 +16,16 @@ const (
 )
 
 type Config struct {
-	BrokerURL          string `json:"brokerUrl,omitempty" yaml:"brokerUrl,omitempty"`
-	ClientID           string `json:"clientId,omitempty" yaml:"clientId,omitempty"`
-	Topic              string `json:"topic,omitempty" yaml:"topic,omitempty"`
-	Message            string `json:"message,omitempty" yaml:"message,omitempty"`
-	MessageFormat      string `json:"messageFormat,omitempty" yaml:"messageFormat,omitempty"`
-	Username           string `json:"username,omitempty" yaml:"username,omitempty"`
-	Password           string `json:"password,omitempty" yaml:"password,omitempty"`
-	InsecureSkipVerify bool   `json:"insecureSkipVerify,omitempty" yaml:"insecureSkipVerify,omitempty"`
+	BrokerURL     string                   `json:"brokerUrl,omitempty" yaml:"brokerUrl,omitempty"`
+	ClientID      string                   `json:"clientId,omitempty" yaml:"clientId,omitempty"`
+	Topic         string                   `json:"topic,omitempty" yaml:"topic,omitempty"`
+	Message       string                   `json:"message,omitempty" yaml:"message,omitempty"`
+	MessageFormat string                   `json:"messageFormat,omitempty" yaml:"messageFormat,omitempty"`
+	Username      string                   `json:"username,omitempty" yaml:"username,omitempty"`
+	Password      string                   `json:"password,omitempty" yaml:"password,omitempty"`
+	QoS           receivers.OptionalNumber `json:"qos,omitempty" yaml:"qos,omitempty"`
+	Retain        bool                     `json:"retain,omitempty" yaml:"retain,omitempty"`
+	TLSConfig     *receivers.TLSConfig     `json:"tlsConfig,omitempty" yaml:"tlsConfig,omitempty"`
 }
 
 func NewConfig(jsonData json.RawMessage, decryptFn receivers.DecryptFunc) (Config, error) {
@@ -56,8 +58,23 @@ func NewConfig(jsonData json.RawMessage, decryptFn receivers.DecryptFunc) (Confi
 		return Config{}, errors.New("Invalid message format, must be 'json' or 'text'")
 	}
 
-	password := decryptFn("password", settings.Password)
-	settings.Password = password
+	qos, err := settings.QoS.Int64()
+	if err != nil {
+		return Config{}, fmt.Errorf("Failed to parse QoS: %w", err)
+	}
+	if qos < 0 || qos > 2 {
+		return Config{}, fmt.Errorf("Invalid QoS level: %d. Must be 0, 1 or 2", qos)
+	}
+
+	settings.Password = decryptFn("password", settings.Password)
+
+	if settings.TLSConfig == nil {
+		settings.TLSConfig = &receivers.TLSConfig{}
+	}
+
+	settings.TLSConfig.CACertificate = decryptFn("tlsConfig.caCertificate", settings.TLSConfig.CACertificate)
+	settings.TLSConfig.ClientCertificate = decryptFn("tlsConfig.clientCertificate", settings.TLSConfig.ClientCertificate)
+	settings.TLSConfig.ClientKey = decryptFn("tlsConfig.clientKey", settings.TLSConfig.ClientKey)
 
 	return settings, nil
 }

receivers/mqtt/config_test.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/stretchr/testify/require"
 
+	"github.com/grafana/alerting/receivers"
 	receiversTesting "github.com/grafana/alerting/receivers/testing"
 	"github.com/grafana/alerting/templates"
 )
@@ -46,17 +47,20 @@ func TestNewConfig(t *testing.T) {
 				BrokerURL:     "tcp://localhost:1883",
 				Topic:         "grafana/alerts",
 				MessageFormat: MessageFormatJSON,
+				TLSConfig:     &receivers.TLSConfig{},
 			},
 		},
 		{
 			name:     "Configuration with insecureSkipVerify",
-			settings: `{ "brokerUrl" : "tcp://localhost:1883", "topic": "grafana/alerts", "insecureSkipVerify": true}`,
+			settings: `{ "brokerUrl" : "tcp://localhost:1883", "topic": "grafana/alerts", "tlsConfig": {"insecureSkipVerify": true}}`,
 			expectedConfig: Config{
-				Message:            templates.DefaultMessageEmbed,
-				BrokerURL:          "tcp://localhost:1883",
-				Topic:              "grafana/alerts",
-				MessageFormat:      MessageFormatJSON,
-				InsecureSkipVerify: true,
+				Message:       templates.DefaultMessageEmbed,
+				BrokerURL:     "tcp://localhost:1883",
+				Topic:         "grafana/alerts",
+				MessageFormat: MessageFormatJSON,
+				TLSConfig: &receivers.TLSConfig{
+					InsecureSkipVerify: true,
+				},
 			},
 		},
 		{
@@ -68,6 +72,7 @@ func TestNewConfig(t *testing.T) {
 				Topic:         "grafana/alerts",
 				MessageFormat: MessageFormatJSON,
 				ClientID:      "test-client-id",
+				TLSConfig:     &receivers.TLSConfig{},
 			},
 		},
 		{
@@ -83,6 +88,28 @@ func TestNewConfig(t *testing.T) {
 				MessageFormat: MessageFormatJSON,
 				Username:      "grafana",
 				Password:      "testpasswd",
+				TLSConfig:     &receivers.TLSConfig{},
+			},
+		},
+		{
+			name:     "Configuration with tlsConfig",
+			settings: `{ "brokerUrl" : "tcp://localhost:1883", "topic": "grafana/alerts"}`,
+			secureSettings: map[string][]byte{
+				"tlsConfig.caCertificate":     []byte("test-ca-cert"),
+				"tlsConfig.clientCertificate": []byte("test-client-cert"),
+				"tlsConfig.clientKey":         []byte("test-client-key"),
+			},
+			expectedConfig: Config{
+				Message:       templates.DefaultMessageEmbed,
+				BrokerURL:     "tcp://localhost:1883",
+				Topic:         "grafana/alerts",
+				MessageFormat: MessageFormatJSON,
+				TLSConfig: &receivers.TLSConfig{
+					InsecureSkipVerify: false,
+					CACertificate:      "test-ca-cert",
+					ClientKey:          "test-client-key",
+					ClientCertificate:  "test-client-cert",
+				},
 			},
 		},
 	}

receivers/mqtt/mqtt.go

@@ -3,9 +3,11 @@ package mqtt
 import (
 	"context"
 	"crypto/tls"
+	"crypto/x509"
 	"encoding/json"
 	"errors"
 	"fmt"
+	"net/url"
 
 	"github.com/prometheus/alertmanager/notify"
 	"github.com/prometheus/alertmanager/types"
@@ -24,6 +26,8 @@ type client interface {
 type message struct {
 	topic   string
 	payload []byte
+	retain  bool
+	qos     int
 }
 
 type Notifier struct {
@@ -59,19 +63,18 @@ type mqttMessage struct {
 }
 
 func (n *Notifier) Notify(ctx context.Context, as ...*types.Alert) (bool, error) {
-	n.log.Debug("Sending an MQTT message")
+	n.log.Debug("Sending an MQTT message", "topic", n.settings.Topic, "qos", n.settings.QoS, "retain", n.settings.Retain)
 
 	msg, err := n.buildMessage(ctx, as...)
 	if err != nil {
 		n.log.Error("Failed to build MQTT message", "error", err.Error())
 		return false, err
 	}
 
-	var tlsCfg *tls.Config
-	if n.settings.InsecureSkipVerify {
-		tlsCfg = &tls.Config{
-			InsecureSkipVerify: true,
-		}
+	tlsCfg, err := n.buildTLSConfig()
+	if err != nil {
+		n.log.Error("Failed to build TLS config", "error", err.Error())
+		return false, fmt.Errorf("failed to build TLS config: %s", err.Error())
 	}
 
 	err = n.client.Connect(ctx, n.settings.BrokerURL, n.settings.ClientID, n.settings.Username, n.settings.Password, tlsCfg)
@@ -86,11 +89,19 @@ func (n *Notifier) Notify(ctx context.Context, as ...*types.Alert) (bool, error)
 		}
 	}()
 
+	qos, err := n.settings.QoS.Int64()
+	if err != nil {
+		n.log.Error("Failed to parse QoS", "error", err.Error())
+		return false, fmt.Errorf("Failed to parse QoS: %s", err.Error())
+	}
+
 	err = n.client.Publish(
 		ctx,
 		message{
 			topic:   n.settings.Topic,
 			payload: []byte(msg),
+			retain:  n.settings.Retain,
+			qos:     int(qos),
 		},
 	)
 
@@ -102,6 +113,39 @@ func (n *Notifier) Notify(ctx context.Context, as ...*types.Alert) (bool, error)
 	return true, nil
 }
 
+func (n *Notifier) buildTLSConfig() (*tls.Config, error) {
+	if n.settings.TLSConfig == nil {
+		return nil, nil
+	}
+
+	parsedURL, err := url.Parse(n.settings.BrokerURL)
+	if err != nil {
+		n.log.Error("Failed to parse broker URL", "error", err.Error())
+		return nil, err
+	}
+
+	tlsCfg := &tls.Config{
+		InsecureSkipVerify: n.settings.TLSConfig.InsecureSkipVerify,
+		ServerName:         parsedURL.Hostname(),
+	}
+
+	if n.settings.TLSConfig.CACertificate != "" {
+		tlsCfg.RootCAs = x509.NewCertPool()
+		tlsCfg.RootCAs.AppendCertsFromPEM([]byte(n.settings.TLSConfig.CACertificate))
+	}
+
+	if n.settings.TLSConfig.ClientCertificate != "" || n.settings.TLSConfig.ClientKey != "" {
+		cert, err := tls.X509KeyPair([]byte(n.settings.TLSConfig.ClientCertificate), []byte(n.settings.TLSConfig.ClientKey))
+		if err != nil {
+			n.log.Error("Failed to load client certificate", "error", err.Error())
+			return nil, err
+		}
+		tlsCfg.Certificates = append(tlsCfg.Certificates, cert)
+	}
+
+	return tlsCfg, nil
+}
+
 func (n *Notifier) buildMessage(ctx context.Context, as ...*types.Alert) (string, error) {
 	groupKey, err := notify.ExtractGroupKey(ctx)
 	if err != nil {