Amazon SQS

Author: everesio

CHANGELOG.md

@@ -1,9 +1,20 @@
 ## Changelog
 
+### [v0.3.0](https://github.com/grepplabs/mqtt-proxy/releases/tag/v0.3.0) - 11 April 2023
+
+* Update confluent-kafka-go to v2.0.2
+* Build docker `ARM64` image
+
+### [v0.2.0](https://github.com/grepplabs/mqtt-proxy/releases/tag/v0.2.0) - 29 December 2022
+
+* Add MQTT 5.0 support
+
 ### [v0.1.1](https://github.com/grepplabs/mqtt-proxy/releases/tag/v0.1.1) - 6 November 2022
+
 * Update dependencies
 * Migrate from [kingpin](https://github.com/alecthomas/kingpin) to [kong](https://github.com/alecthomas/kong)
 
 ### [v0.1.0](https://github.com/grepplabs/mqtt-proxy/releases/tag/v0.1.0) - 1 November 2022
+
 * Add server certificates rotation
 * Add client certificate revocation list

README.md

@@ -18,6 +18,9 @@ MQTT Proxy allows MQTT clients to send messages to other messaging systems
 * Publisher
     * [x] Noop
     * [x] [Apache Kafka](https://kafka.apache.org/)
+    * [x] [Amazon SQS](https://aws.amazon.com/sqs/)
+    * [ ] [Amazon SNS](https://aws.amazon.com/sns/)
+    * [ ] [Amazon Kinesis](https://aws.amazon.com/kinesis/)
     * [ ] Others
 * Authentication
     * [x] Noop
@@ -113,8 +116,6 @@ prerequisites
     watch -c 'curl -s localhost:9090/metrics | grep mqtt | egrep -v '^#''
     ```
 
-5. see also [cp-kafka](scripts/cp-kafka/Makefile) with SASL_PLAINTEXT and SASL_SSL configuration
-
 ### publish to Amazon MSK
 
 1. provision test MSK and EC2 running in [podman](https://podman.io/) 2 proxy containers
@@ -138,6 +139,27 @@ prerequisites
     mosquitto_pub -m "on" -t "dummy" -k 20 -i mqtt-proxy.clientv --repeat 1 -q 1 -h <ec2-ip> -p 1884
     ```
 
+### SQS publisher
+
+1. Create AWS SQS `test1` and `test2.fifo` queues
+2. Build and start MQTT Proxy
+
+    ```
+    make build
+   ./mqtt-proxy server \
+      --mqtt.publisher.name=sqs \
+      --mqtt.publisher.message-format=json \
+      --mqtt.publisher.sqs.queue-mappings=test1='^dummy$' \
+      --mqtt.publisher.sqs.default-queue=test2.fifo \
+      --mqtt.publisher.sqs.aws-profile=admin-dev
+    ```
+
+3. publish
+
+    ```
+    mosquitto_pub -L mqtt://localhost:1883/dummy -m "test qos 0" --repeat 1 -q 2
+    ```
+
 ### plain authenticator
 
 1. start server with `plain` authenticator

apis/publish.go

@@ -7,12 +7,12 @@ import "context"
 type PublishID interface{}
 
 type PublishRequest struct {
-	Dup       bool
-	Qos       byte
-	Retain    bool
-	TopicName string
-	MessageID uint16
-	Message   []byte
+	Dup       bool   `json:"dup"`
+	Qos       byte   `json:"qos"`
+	Retain    bool   `json:"retain"`
+	TopicName string `json:"topic_name"`
+	MessageID uint16 `json:"packet_id"`
+	Message   []byte `json:"payload"`
 }
 
 type PublishResponse struct {

cmd/server.go

@@ -16,6 +16,7 @@ import (
 	pubinst "github.com/grepplabs/mqtt-proxy/pkg/publisher/instrument"
 	pubkafka "github.com/grepplabs/mqtt-proxy/pkg/publisher/kafka"
 	pubnoop "github.com/grepplabs/mqtt-proxy/pkg/publisher/noop"
+	pubsqs "github.com/grepplabs/mqtt-proxy/pkg/publisher/sqs"
 	httpserver "github.com/grepplabs/mqtt-proxy/pkg/server/http"
 	mqttserver "github.com/grepplabs/mqtt-proxy/pkg/server/mqtt"
 	servertls "github.com/grepplabs/mqtt-proxy/pkg/tls"
@@ -102,10 +103,22 @@ func runServer(
 				pubkafka.WithConfigMap(cfg.MQTT.Publisher.Kafka.ConfArgs.ConfigMap()),
 				pubkafka.WithGracePeriod(cfg.MQTT.Publisher.Kafka.GracePeriod),
 				pubkafka.WithWorkers(cfg.MQTT.Publisher.Kafka.Workers),
+				pubkafka.WithMessageFormat(cfg.MQTT.Publisher.MessageFormat),
 			)
 			if err != nil {
 				return fmt.Errorf("setup kafka publisher: %w", err)
 			}
+		case config.PublisherSQS:
+			publisher, err = pubsqs.New(logger, registry,
+				pubsqs.WithAWSProfile(cfg.MQTT.Publisher.SQS.AWSProfile),
+				pubsqs.WithAWSRegion(cfg.MQTT.Publisher.SQS.AWSRegion),
+				pubsqs.WithQueueMappings(cfg.MQTT.Publisher.SQS.QueueMappings),
+				pubsqs.WithDefaultQueue(cfg.MQTT.Publisher.SQS.DefaultQueue),
+				pubsqs.WithMessageFormat(cfg.MQTT.Publisher.MessageFormat),
+			)
+			if err != nil {
+				return fmt.Errorf("setup sqs publisher: %w", err)
+			}
 		default:
 			return fmt.Errorf("unknown publisher %s", cfg.MQTT.Publisher.Name)
 		}

go.mod

@@ -5,6 +5,10 @@ go 1.20
 require (
 	github.com/alecthomas/kong v0.7.0
 	github.com/alecthomas/kong-yaml v0.1.1
+	github.com/aws/aws-sdk-go-v2 v1.17.8
+	github.com/aws/aws-sdk-go-v2/config v1.18.21
+	github.com/aws/aws-sdk-go-v2/service/sqs v1.20.8
+	github.com/aws/aws-sdk-go-v2/service/sts v1.18.9
 	github.com/confluentinc/confluent-kafka-go/v2 v2.0.2
 	github.com/go-playground/validator/v10 v10.11.1
 	github.com/oklog/run v1.1.0
@@ -18,6 +22,15 @@ require (
 )
 
 require (
+	github.com/aws/aws-sdk-go-v2/credentials v1.13.20 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.13.2 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.32 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.26 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.3.33 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.26 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.12.8 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.14.8 // indirect
+	github.com/aws/smithy-go v1.13.5 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect

go.sum

@@ -49,6 +49,32 @@ github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRF
 github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho=
 github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
+github.com/aws/aws-sdk-go-v2 v1.17.8 h1:GMupCNNI7FARX27L7GjCJM8NgivWbRgpjNI/hOQjFS8=
+github.com/aws/aws-sdk-go-v2 v1.17.8/go.mod h1:uzbQtefpm44goOPmdKyAlXSNcwlRgF3ePWVW6EtJvvw=
+github.com/aws/aws-sdk-go-v2/config v1.18.21 h1:ENTXWKwE8b9YXgQCsruGLhvA9bhg+RqAsL9XEMEsa2c=
+github.com/aws/aws-sdk-go-v2/config v1.18.21/go.mod h1:+jPQiVPz1diRnjj6VGqWcLK6EzNmQ42l7J3OqGTLsSY=
+github.com/aws/aws-sdk-go-v2/credentials v1.13.20 h1:oZCEFcrMppP/CNiS8myzv9JgOzq2s0d3v3MXYil/mxQ=
+github.com/aws/aws-sdk-go-v2/credentials v1.13.20/go.mod h1:xtZnXErtbZ8YGXC3+8WfajpMBn5Ga/3ojZdxHq6iI8o=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.13.2 h1:jOzQAesnBFDmz93feqKnsTHsXrlwWORNZMFHMV+WLFU=
+github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.13.2/go.mod h1:cDh1p6XkSGSwSRIArWRc6+UqAQ7x4alQ0QfpVR6f+co=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.32 h1:dpbVNUjczQ8Ae3QKHbpHBpfvaVkRdesxpTOe9pTouhU=
+github.com/aws/aws-sdk-go-v2/internal/configsources v1.1.32/go.mod h1:RudqOgadTWdcS3t/erPQo24pcVEoYyqj/kKW5Vya21I=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.26 h1:QH2kOS3Ht7x+u0gHCh06CXL/h6G8LQJFpZfFBYBNboo=
+github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.4.26/go.mod h1:vq86l7956VgFr0/FWQ2BWnK07QC3WYsepKzy33qqY5U=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.3.33 h1:HbH1VjUgrCdLJ+4lnnuLI4iVNRvBbBELGaJ5f69ClA8=
+github.com/aws/aws-sdk-go-v2/internal/ini v1.3.33/go.mod h1:zG2FcwjQarWaqXSCGpgcr3RSjZ6dHGguZSppUL0XR7Q=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.26 h1:uUt4XctZLhl9wBE1L8lobU3bVN8SNUP7T+olb0bWBO4=
+github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.9.26/go.mod h1:Bd4C/4PkVGubtNe5iMXu5BNnaBi/9t/UsFspPt4ram8=
+github.com/aws/aws-sdk-go-v2/service/sqs v1.20.8 h1:SDZBYFUp70hI2T0z9z+KD1iJBz9jGeT7xgU5hPPC9zs=
+github.com/aws/aws-sdk-go-v2/service/sqs v1.20.8/go.mod h1:w058QQWcK1MLEnIrD0DmkQtSvC1pLY0EWRQsPXPWppM=
+github.com/aws/aws-sdk-go-v2/service/sso v1.12.8 h1:5cb3D6xb006bPTqEfCNaEA6PPEfBXxxy4NNeX/44kGk=
+github.com/aws/aws-sdk-go-v2/service/sso v1.12.8/go.mod h1:GNIveDnP+aE3jujyUSH5aZ/rktsTM5EvtKnCqBZawdw=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.14.8 h1:NZaj0ngZMzsubWZbrEFSB4rgSQRbFq38Sd6KBxHuOIU=
+github.com/aws/aws-sdk-go-v2/service/ssooidc v1.14.8/go.mod h1:44qFP1g7pfd+U+sQHLPalAPKnyfTZjJsYR4xIwsJy5o=
+github.com/aws/aws-sdk-go-v2/service/sts v1.18.9 h1:Qf1aWwnsNkyAoqDqmdM3nHwN78XQjec27LjM6b9vyfI=
+github.com/aws/aws-sdk-go-v2/service/sts v1.18.9/go.mod h1:yyW88BEPXA2fGFyI2KCcZC3dNpiT0CZAHaF+i656/tQ=
+github.com/aws/smithy-go v1.13.5 h1:hgz0X/DX0dGqTYpGALqXJoRKRj5oQ7150i5FdTePzO8=
+github.com/aws/smithy-go v1.13.5/go.mod h1:Tg+OJXh4MB2R/uN61Ko2f6hTZwB/ZYGOtib8J3gBHzA=
 github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
 github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q=
 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
@@ -154,6 +180,7 @@ github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/go-cmp v0.5.8 h1:e6P7q2lk1O+qJJb4BtCQXlK8vWEO8V1ZeuEdJNOqZyg=
+github.com/google/go-cmp v0.5.8/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
 github.com/google/martian v2.1.0+incompatible/go.mod h1:9I4somxYTbIHy5NJKHRl3wXiIaQGbYVAs8BPL6v8lEs=
 github.com/google/martian/v3 v3.0.0/go.mod h1:y5Zk1BBys9G+gd6Jrk0W3cC1+ELVxBWuIGO+w/tUAp0=
@@ -185,6 +212,8 @@ github.com/jhump/gopoet v0.1.0/go.mod h1:me9yfT6IJSlOL3FCfrg+L6yzUEZ+5jW6WHt4Sk+
 github.com/jhump/goprotoc v0.5.0/go.mod h1:VrbvcYrQOrTi3i0Vf+m+oqQWk9l72mjkJCYo7UvLHRQ=
 github.com/jhump/protoreflect v1.11.0/go.mod h1:U7aMIjN0NWq9swDP7xDdoMfRHb35uiuTd3Z9nFXJf5E=
 github.com/jhump/protoreflect v1.12.0/go.mod h1:JytZfP5d0r8pVNLZvai7U/MCuTWITgrI4tTg7puQFKI=
+github.com/jmespath/go-jmespath v0.4.0/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo=
+github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U=
 github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/json-iterator/go v1.1.10/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=

pkg/config/config.go

@@ -15,6 +15,7 @@ import (
 const (
 	PublisherNoop  = "noop"
 	PublisherKafka = "kafka"
+	PublisherSQS   = "sqs"
 )
 
 // authenticator names
@@ -23,6 +24,13 @@ const (
 	AuthPlain = "plain"
 )
 
+// message format
+const (
+	MessageFormatPlain  = "plain"
+	MessageFormatBase64 = "base64"
+	MessageFormatJson   = "json"
+)
+
 // server certificate source
 const (
 	CertSourceFile = "file"
@@ -72,15 +80,22 @@ type Server struct {
 			} `embed:"" prefix:"auth."`
 		} `embed:"" prefix:"handler."`
 		Publisher struct {
-			Name  string `default:"${PublisherDefault}" enum:"${PublisherEnum}" help:"Publisher name. One of: [${PublisherEnum}]"`
-			Kafka struct {
+			Name          string `default:"${PublisherDefault}" enum:"${PublisherEnum}" help:"Publisher name. One of: [${PublisherEnum}]"`
+			MessageFormat string `default:"${MessageFormatDefault}" enum:"${MessageFormatEnum}" help:"Message format. One of: [${MessageFormatEnum}]"`
+			Kafka         struct {
 				BootstrapServers string          `default:"localhost:9092" help:"Kafka bootstrap servers."`
 				GracePeriod      time.Duration   `default:"10s" help:"Time to wait after an interrupt received for Kafka publisher." validate:"gte=0"`
 				ConfArgs         KafkaConfigArgs `name:"config" placeholder:"PROP=VAL" help:"Comma separated list of properties."`
 				DefaultTopic     string          `default:"" help:"Default Kafka topic for MQTT publish messages."`
 				TopicMappings    TopicMappings   `placeholder:"TOPIC=REGEX" help:"Comma separated list of Kafka topic to MQTT topic mappings."`
 				Workers          int             `default:"1" help:"Number of kafka publisher workers." validate:"gte=1"`
 			} `embed:"" prefix:"kafka."`
+			SQS struct {
+				AWSProfile    string        `default:"" help:"AWS Profile."`
+				AWSRegion     string        `default:"" help:"AWS Region."`
+				DefaultQueue  string        `default:"" help:"Default Kafka topic for MQTT publish messages."`
+				QueueMappings TopicMappings `placeholder:"QUEUE=REGEX" help:"Comma separated list of SQS queue to MQTT topic mappings."`
+			} `embed:"" prefix:"sqs."`
 		} `embed:"" prefix:"publisher."`
 	} `embed:"" prefix:"mqtt."`
 }
@@ -94,7 +109,9 @@ func ServerVars() kong.Vars {
 		"AuthDefault":              AuthNoop,
 		"AuthEnum":                 strings.Join([]string{AuthNoop, AuthPlain}, ", "),
 		"PublisherDefault":         PublisherNoop,
-		"PublisherEnum":            strings.Join([]string{PublisherNoop, PublisherKafka}, ", "),
+		"PublisherEnum":            strings.Join([]string{PublisherNoop, PublisherKafka, PublisherSQS}, ", "),
+		"MessageFormatDefault":     MessageFormatPlain,
+		"MessageFormatEnum":        strings.Join([]string{MessageFormatPlain, MessageFormatBase64, MessageFormatJson}, ", "),
 	}
 }
 

pkg/publisher/kafka/kafka.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/grepplabs/mqtt-proxy/pkg/util"
 	"reflect"
 	"strconv"
 	"strings"
@@ -25,6 +26,7 @@ const (
 	mqttDupHeader    = "mqtt.dup"
 	mqttRetainHeader = "mqtt.retain"
 	mqttMsgIDHeader  = "mqtt.packet.id"
+	mqttMsgFmtHeader = "mqtt.fmt"
 )
 const (
 	shutdownPollInterval = 500 * time.Millisecond
@@ -44,15 +46,11 @@ func (k *kafkaProducer) Close() {
 }
 
 type Publisher struct {
-	logger log.Logger
-
-	producers map[byte]*kafkaProducer
-
-	inShutdown atomic.Bool
-
+	logger      log.Logger
+	producers   map[byte]*kafkaProducer
+	inShutdown  atomic.Bool
 	workersDone *runtime.DoneChannel
-
-	opts options
+	opts        options
 }
 
 func New(logger log.Logger, _ *prometheus.Registry, opts ...Option) (*Publisher, error) {
@@ -236,11 +234,18 @@ func (s *Publisher) newKafkaMessage(req *apis.PublishRequest, opaque interface{}
 		{Key: mqttDupHeader, Value: []byte(strconv.FormatBool(req.Dup))},
 		{Key: mqttRetainHeader, Value: []byte(strconv.FormatBool(req.Retain))},
 		{Key: mqttMsgIDHeader, Value: []byte(strconv.FormatUint(uint64(req.MessageID), 10))},
+		{Key: mqttMsgFmtHeader, Value: []byte(s.opts.messageFormat)},
 	}
+
+	message, err := util.GetMessageBody(s.opts.messageFormat, req)
+	if err != nil {
+		return nil, err
+	}
+
 	return &kafka.Message{
 		TopicPartition: kafka.TopicPartition{Topic: &kafkaTopic, Partition: kafka.PartitionAny},
 		Key:            []byte(req.TopicName),
-		Value:          req.Message,
+		Value:          message,
 		Opaque:         opaque,
 		Headers:        headers,
 	}, nil
@@ -255,7 +260,7 @@ func (s *Publisher) getKafkaTopic(mqttTopic string) (string, error) {
 	if s.opts.defaultTopic != "" {
 		return s.opts.defaultTopic, nil
 	}
-	return "", fmt.Errorf("Kafka topic not found for MQTT topic %s", mqttTopic)
+	return "", fmt.Errorf("kafka topic not found for MQTT topic %s", mqttTopic)
 }
 
 func (s *Publisher) Publish(ctx context.Context, request *apis.PublishRequest) (*apis.PublishResponse, error) {

pkg/publisher/kafka/kafka_it_test.go

@@ -3,6 +3,7 @@ package kafka
 import (
 	"context"
 	"github.com/grepplabs/mqtt-proxy/apis"
+	"github.com/grepplabs/mqtt-proxy/pkg/config"
 	"github.com/oklog/run"
 	"os"
 	"testing"
@@ -30,7 +31,8 @@ func newTestPublisherOrExit() *Publisher {
 		WithBootstrapServers(bootstrapServers),
 		WithGracePeriod(60*time.Second),
 		WithConfigMap(defaultConfig),
-		WithDefaultTopic("mqtt-test"))
+		WithDefaultTopic("mqtt-test"),
+		WithMessageFormat(config.MessageFormatPlain))
 
 	if err != nil {
 		logger.WithError(err).Errorf("kafka publisher creation failed")

pkg/publisher/kafka/option.go

@@ -17,6 +17,7 @@ type options struct {
 
 	defaultTopic  string
 	topicMappings config.TopicMappings
+	messageFormat string
 }
 
 func (o options) validate() error {
@@ -29,6 +30,9 @@ func (o options) validate() error {
 	if o.defaultTopic == "" && len(o.topicMappings.Mappings) == 0 {
 		return errors.New("kafka default topic or topic mappings must be provided")
 	}
+	if o.messageFormat == "" {
+		return errors.New("publisher message format must not be empty")
+	}
 	return nil
 }
 
@@ -77,3 +81,9 @@ func WithWorkers(v int) Option {
 		o.workers = v
 	})
 }
+
+func WithMessageFormat(s string) Option {
+	return optionFunc(func(o *options) {
+		o.messageFormat = s
+	})
+}