mirror of
https://github.com/absmach/supermq.git
synced 2026-06-23 06:50:18 +00:00
NOISSUE - update mqtt benchmark (#824)
* update gitignore Signed-off-by: Mirko Teodorovic <mirko.teodorovic@gmail.com> * fix dates for cert Signed-off-by: Mirko Teodorovic <mirko.teodorovic@gmail.com> * reorganize code Signed-off-by: Mirko Teodorovic <mirko.teodorovic@gmail.com> * update docs Signed-off-by: Mirko Teodorovic <mirko.teodorovic@gmail.com> * print to stdout for creating result file from script Signed-off-by: Mirko Teodorovic <mirko.teodorovic@gmail.com> * add script for testing Signed-off-by: Mirko Teodorovic <mirko.teodorovic@gmail.com> * add time measeuring Signed-off-by: Mirko Teodorovic <mirko.teodorovic@gmail.com> * separate test in one file each Signed-off-by: Mirko Teodorovic <mirko.teodorovic@gmail.com> * move config.toml Signed-off-by: Mirko Teodorovic <mirko.teodorovic@gmail.com> * update docs Signed-off-by: Mirko Teodorovic <mirko.teodorovic@gmail.com> * add connection check, and prevent blocking of publisher when connection fails Signed-off-by: Mirko Teodorovic <mirko.teodorovic@gmail.com>
This commit is contained in:
Mirko Teodorovic
committed by
Nikola Marčetić
Nikola Marčetić
parent
e7891ec6ab
commit
bfd12304ac
+2
-1
@@ -1,3 +1,4 @@
|
||||
build
|
||||
|
||||
.vscode/
|
||||
site/
|
||||
*.crt
|
||||
|
||||
+32
-22
@@ -7,23 +7,27 @@ The tool supports multiple concurrent clients, publishers and subscribers config
|
||||
|
||||
```
|
||||
cd benchmark
|
||||
go build -o mqtt-benchmark *.go
|
||||
go build -o mqtt-bench *.go
|
||||
|
||||
> mqtt-bench --help
|
||||
Flags:
|
||||
-b, --broker string address for mqtt broker, for secure use tcps and 8883 (default "tcp://localhost:1883")
|
||||
--ca string CA file (default "ca.crt")
|
||||
--channels string config file for channels (default "channels.toml")
|
||||
-g, --config string config file default is config.toml (default "config.toml")
|
||||
-n, --count int Number of messages sent per publisher (default 100)
|
||||
-f, --format string Output format: text|json (default "text")
|
||||
-h, --help help for mqtt-bench
|
||||
--msg string messg to be sent, SENML (default "{\"n\":\"current\",\"t\":-4,\"v\":1.3}")
|
||||
-m, --mtls Use mtls for connection
|
||||
--pubs int Number of publishers (default 10)
|
||||
-q, --qos int QoS for published messages, values 0 1 2
|
||||
--quiet Supress messages
|
||||
-r, --retain Retain mqtt messages
|
||||
-s, --size int Size of message payload bytes (default 100)
|
||||
-t, --skipTLSVer Skip tls verification
|
||||
--subs int Number of subscribers (default 10)
|
||||
|
||||
> mqtt-benchmark --help
|
||||
Usage of mqtt-benchmark:
|
||||
-broker="tcp://localhost:1883": MQTT broker endpoint as scheme://host:port
|
||||
-clients=10: Number of clients to start
|
||||
-count=100: Number of messages to send per client
|
||||
-format="text": Output format: text|json
|
||||
-password="": MQTT password (empty if auth disabled)
|
||||
-qos=1: QoS for published messages
|
||||
-quiet=false : Suppress logs while running (except errors and the result)
|
||||
-size=100: Size of the messages payload (bytes
|
||||
-subs=10 number of subscribers
|
||||
-pubs=10 number of publishers
|
||||
-config=connections.json , file with mainflux channels
|
||||
-mtls=false, use mtls
|
||||
-ca=ca.crt, use mqtts, pass ca to server validate certificate
|
||||
```
|
||||
|
||||
Two output formats supported: human-readable plain text and JSON.
|
||||
@@ -34,20 +38,20 @@ channels for testing
|
||||
Example use and output:
|
||||
|
||||
```
|
||||
go build -o mqtt-benchmark *.go
|
||||
go build -o mqtt-bench *.go
|
||||
|
||||
without mtls
|
||||
./mqtt-benchmark --broker tcp://localhost:1883 --count 100 --size 100 --qos 0 --format text --subs 100 --pubs 0 --channels channels.toml
|
||||
./mqtt-bench --broker tcp://localhost:1883 --count 100 --size 100 --qos 0 --format text --subs 100 --pubs 0 --channels channels.toml
|
||||
|
||||
with mtls
|
||||
./mqtt-benchmark --broker tcps://localhost:8883 --count 100 --size 100 --qos 0 --format text --subs 100 --pubs 0 --channels channels.toml --mtls -ca ca.crt
|
||||
./mqtt-bench --broker tcps://localhost:8883 --count 100 --size 100 --qos 0 --format text --subs 100 --pubs 0 --channels channels.toml --mtls -ca ca.crt
|
||||
```
|
||||
|
||||
|
||||
You can use config.toml to create tests with this tool
|
||||
./mqtt-benchmark --config config.toml it will read params from config.toml
|
||||
|
||||
|
||||
```
|
||||
./mqtt-bench --config config.toml it will read params from config.toml
|
||||
```
|
||||
```
|
||||
broker_url = "tcp://localhost:1883"
|
||||
qos = 2
|
||||
@@ -61,6 +65,12 @@ mtls = false
|
||||
skiptlsver = true
|
||||
ca_file = "ca.crt"
|
||||
channels_file = "channels.toml"
|
||||
```
|
||||
You can use script which will run series of tests using mqtt-bench
|
||||
```
|
||||
cd tools/mqtt-bench/scripts
|
||||
./mqtt-bench.sh mainflux mainflux.com channels.toml
|
||||
|
||||
```
|
||||
|
||||
For Example
|
||||
|
||||
@@ -6,8 +6,8 @@ pubs =3
|
||||
subs =1
|
||||
format = "text"
|
||||
quiet = true
|
||||
mtls = false
|
||||
mtls = true
|
||||
skiptlsver = true
|
||||
ca = "ca.crt"
|
||||
channels = "channels.toml"
|
||||
retain = true
|
||||
retain = true
|
||||
@@ -1,4 +1,4 @@
|
||||
package cmd
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
@@ -7,27 +7,22 @@ import (
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"log"
|
||||
"net"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/BurntSushi/toml"
|
||||
"github.com/mainflux/mainflux/tools/mqtt-bench/mqtt"
|
||||
"github.com/mainflux/mainflux/tools/mqtt-bench/res"
|
||||
res "github.com/mainflux/mainflux/tools/mqtt-bench/results"
|
||||
"github.com/spf13/cobra"
|
||||
"github.com/spf13/viper"
|
||||
"gonum.org/v1/gonum/mat"
|
||||
"gonum.org/v1/gonum/stat"
|
||||
)
|
||||
|
||||
// Execute - main command
|
||||
func Execute() {
|
||||
if err := benchCmd.Execute(); err != nil {
|
||||
log.Fatalf(err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
// Config - command optiopns from file configuration
|
||||
type Config struct {
|
||||
Broker string `toml:"broker"`
|
||||
@@ -75,6 +70,7 @@ var (
|
||||
format string
|
||||
conf string
|
||||
channels string
|
||||
msg string
|
||||
quiet bool
|
||||
retain bool
|
||||
mtls bool
|
||||
@@ -85,14 +81,25 @@ var (
|
||||
var benchCmd = &cobra.Command{
|
||||
Use: "mqtt-bench",
|
||||
Run: func(cmd *cobra.Command, args []string) {
|
||||
if len(args) < 1 {
|
||||
cmd.Help()
|
||||
fmt.Printf("broker - %s\n", broker)
|
||||
fmt.Printf("mtls - %v\n", mtls)
|
||||
fmt.Printf("retain - %v\n", retain)
|
||||
fmt.Printf("qos - %d\n", qos)
|
||||
fmt.Printf("pubs - %d\n", pubs)
|
||||
fmt.Printf("subs - %d\n", subs)
|
||||
if pubs < 1 && subs < 1 {
|
||||
log.Fatal("Invalid arguments")
|
||||
}
|
||||
|
||||
runBench()
|
||||
},
|
||||
}
|
||||
|
||||
func main() {
|
||||
if err := benchCmd.Execute(); err != nil {
|
||||
log.Fatalf(err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
func init() {
|
||||
cobra.OnInitialize(initConfig)
|
||||
|
||||
@@ -105,12 +112,12 @@ func init() {
|
||||
benchCmd.PersistentFlags().StringVarP(&format, "format", "f", "text", "Output format: text|json")
|
||||
benchCmd.PersistentFlags().StringVarP(&conf, "config", "g", "config.toml", "config file default is config.toml")
|
||||
benchCmd.PersistentFlags().StringVarP(&channels, "channels", "", "channels.toml", "config file for channels")
|
||||
benchCmd.PersistentFlags().StringVarP(&msg, "msg", "", "{\"n\":\"current\",\"t\":-4,\"v\":1.3}", "messg to be sent, SENML")
|
||||
benchCmd.PersistentFlags().StringVarP(&ca, "ca", "", "ca.crt", "CA file")
|
||||
benchCmd.PersistentFlags().BoolVarP(&quiet, "quiet", "", false, "Supress messages")
|
||||
benchCmd.PersistentFlags().BoolVarP(&retain, "retain", "r", false, "Retain mqtt messages")
|
||||
benchCmd.PersistentFlags().BoolVarP(&mtls, "mtls", "m", false, "Use mtls for connection")
|
||||
benchCmd.PersistentFlags().BoolVarP(&skipTLSVer, "skipTLSVer", "t", false, "Skip tls verification")
|
||||
|
||||
}
|
||||
|
||||
func initConfig() {
|
||||
@@ -126,9 +133,9 @@ func initConfig() {
|
||||
|
||||
err = viper.Unmarshal(&c)
|
||||
if err != nil {
|
||||
log.Printf("failed to load config - %s", err.Error())
|
||||
log.Printf("failed to load config - %s\n", err.Error())
|
||||
}
|
||||
log.Printf("config file: %s", viper.ConfigFileUsed())
|
||||
log.Printf("config file: %s\n", viper.ConfigFileUsed())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -136,12 +143,8 @@ func runBench() {
|
||||
var wg sync.WaitGroup
|
||||
var err error
|
||||
|
||||
checkConnection(broker, 1)
|
||||
subTimes := make(res.SubTimes)
|
||||
|
||||
if pubs < 1 && subs < 1 {
|
||||
log.Fatal("Invalid arguments")
|
||||
}
|
||||
|
||||
var caByte []byte
|
||||
if mtls {
|
||||
caFile, err := os.Open(ca)
|
||||
@@ -153,8 +156,8 @@ func runBench() {
|
||||
caByte, _ = ioutil.ReadAll(caFile)
|
||||
}
|
||||
|
||||
payload := string(make([]byte, size))
|
||||
c := Connections{}
|
||||
|
||||
loadChansConfig(&channels, &c)
|
||||
connections := c.Connection
|
||||
|
||||
@@ -190,9 +193,10 @@ func runBench() {
|
||||
CA: caByte,
|
||||
ClientCert: cert,
|
||||
Retain: retain,
|
||||
Message: payload,
|
||||
}
|
||||
wg.Add(1)
|
||||
go c.RunSubscriber(&wg, &subTimes, &done, mtls)
|
||||
go c.RunSubscriber(&wg, &subTimes, &done)
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
@@ -222,8 +226,9 @@ func runBench() {
|
||||
CA: caByte,
|
||||
ClientCert: cert,
|
||||
Retain: retain,
|
||||
Message: payload,
|
||||
}
|
||||
go c.RunPublisher(resCh, mtls)
|
||||
go c.RunPublisher(resCh)
|
||||
}
|
||||
|
||||
// collect the results
|
||||
@@ -320,7 +325,7 @@ func printResults(results []*res.RunResults, totals *res.TotalResults, format st
|
||||
}
|
||||
data, err := json.Marshal(jr)
|
||||
if err != nil {
|
||||
log.Printf("Failed to prepare results for printing - %s", err.Error())
|
||||
log.Printf("Failed to prepare results for printing - %s\n", err.Error())
|
||||
}
|
||||
var out bytes.Buffer
|
||||
json.Indent(&out, data, "", "\t")
|
||||
@@ -329,7 +334,7 @@ func printResults(results []*res.RunResults, totals *res.TotalResults, format st
|
||||
default:
|
||||
if !quiet {
|
||||
for _, res := range results {
|
||||
fmt.Printf("======= CLIENT %s =======\n", res.ID)
|
||||
fmt.Printf("======= CLIENT %d =======\n", res.ID)
|
||||
fmt.Printf("Ratio: %.3f (%d/%d)\n", float64(res.Successes)/float64(res.Successes+res.Failures), res.Successes, res.Successes+res.Failures)
|
||||
fmt.Printf("Runtime (s): %.3f\n", res.RunTime)
|
||||
fmt.Printf("Msg time min (us): %.3f\n", res.MsgTimeMin)
|
||||
@@ -365,3 +370,36 @@ func loadChansConfig(path *string, conns *Connections) {
|
||||
log.Fatalf("cannot load channels config %s \nuse tools/provision to create file", *path)
|
||||
}
|
||||
}
|
||||
|
||||
func checkConnection(broker string, timeoutSecs int) {
|
||||
s := strings.Split(broker, ":")
|
||||
if len(s) != 3 {
|
||||
log.Fatalf("wrong host address format")
|
||||
}
|
||||
|
||||
network := s[0]
|
||||
host := strings.Trim(s[1], "/")
|
||||
port := s[2]
|
||||
|
||||
conn, err := net.DialTimeout("tcp", fmt.Sprintf("%s:%s", host, port), time.Duration(timeoutSecs)*time.Second)
|
||||
conClose := func() {
|
||||
if conn != nil {
|
||||
log.Println("closing connection")
|
||||
conn.Close()
|
||||
}
|
||||
}
|
||||
|
||||
defer conClose()
|
||||
if err, ok := err.(*net.OpError); ok && err.Timeout() {
|
||||
fmt.Printf("Timeout error: %s\n", err.Error())
|
||||
log.Fatalf("Timeout error: %s\n", err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
fmt.Printf("Error: %s\n", err.Error())
|
||||
log.Fatalf("Error: %s\n", err)
|
||||
return
|
||||
}
|
||||
log.Printf("Connection to %s://%s:%s looks ok\n", network, host, port)
|
||||
}
|
||||
@@ -1,7 +0,0 @@
|
||||
package main
|
||||
|
||||
import "github.com/mainflux/mainflux/tools/mqtt-bench/cmd"
|
||||
|
||||
func main() {
|
||||
cmd.Execute()
|
||||
}
|
||||
@@ -11,7 +11,7 @@ import (
|
||||
"time"
|
||||
|
||||
mqtt "github.com/eclipse/paho.mqtt.golang"
|
||||
"github.com/mainflux/mainflux/tools/mqtt-bench/res"
|
||||
res "github.com/mainflux/mainflux/tools/mqtt-bench/results"
|
||||
mat "gonum.org/v1/gonum/mat"
|
||||
stat "gonum.org/v1/gonum/stat"
|
||||
)
|
||||
@@ -23,6 +23,7 @@ type Client struct {
|
||||
BrokerUser string
|
||||
BrokerPass string
|
||||
MsgTopic string
|
||||
Message string
|
||||
MsgSize int
|
||||
MsgCount int
|
||||
MsgQoS byte
|
||||
@@ -54,7 +55,7 @@ type message struct {
|
||||
}
|
||||
|
||||
// RunPublisher - runs publisher
|
||||
func (c *Client) RunPublisher(r chan *res.RunResults, mtls bool) {
|
||||
func (c *Client) RunPublisher(r chan *res.RunResults) {
|
||||
newMsgs := make(chan *message)
|
||||
pubMsgs := make(chan *message)
|
||||
doneGen := make(chan bool)
|
||||
@@ -65,7 +66,7 @@ func (c *Client) RunPublisher(r chan *res.RunResults, mtls bool) {
|
||||
// Start generator
|
||||
go c.genMessages(newMsgs, doneGen)
|
||||
// Start publisher
|
||||
go c.pubMessages(newMsgs, pubMsgs, doneGen, donePub, mtls)
|
||||
go c.pubMessages(newMsgs, pubMsgs, doneGen, donePub)
|
||||
|
||||
times := []float64{}
|
||||
|
||||
@@ -99,16 +100,18 @@ func (c *Client) RunPublisher(r chan *res.RunResults, mtls bool) {
|
||||
}
|
||||
|
||||
// RunSubscriber - runs a subscriber
|
||||
func (c *Client) RunSubscriber(wg *sync.WaitGroup, subTimes *res.SubTimes, done *chan bool, mtls bool) {
|
||||
func (c *Client) RunSubscriber(wg *sync.WaitGroup, subTimes *res.SubTimes, done *chan bool) {
|
||||
defer wg.Done()
|
||||
// Start subscriber
|
||||
c.subscribe(wg, subTimes, done, mtls)
|
||||
c.subscribe(wg, subTimes, done)
|
||||
|
||||
}
|
||||
|
||||
func (c *Client) genMessages(ch chan *message, done chan bool) {
|
||||
|
||||
for i := 0; i < c.MsgCount; i++ {
|
||||
msgPayload := messagePayload{Payload: make([]byte, c.MsgSize)}
|
||||
|
||||
msgPayload := messagePayload{Payload: c.Message}
|
||||
ch <- &message{
|
||||
Topic: c.MsgTopic,
|
||||
QoS: c.MsgQoS,
|
||||
@@ -119,7 +122,7 @@ func (c *Client) genMessages(ch chan *message, done chan bool) {
|
||||
return
|
||||
}
|
||||
|
||||
func (c *Client) subscribe(wg *sync.WaitGroup, subTimes *res.SubTimes, done *chan bool, mtls bool) {
|
||||
func (c *Client) subscribe(wg *sync.WaitGroup, subTimes *res.SubTimes, done *chan bool) {
|
||||
clientID := fmt.Sprintf("sub-%v-%v", time.Now().Format(time.RFC3339Nano), c.ID)
|
||||
c.ID = clientID
|
||||
|
||||
@@ -129,14 +132,17 @@ func (c *Client) subscribe(wg *sync.WaitGroup, subTimes *res.SubTimes, done *cha
|
||||
}
|
||||
}
|
||||
|
||||
c.connect(onConnected, mtls)
|
||||
connLost := func(client mqtt.Client, reason error) {
|
||||
log.Printf("CLIENT %v had lost connection to the broker: %s\n", c.ID, reason.Error())
|
||||
}
|
||||
c.connect(onConnected, connLost)
|
||||
|
||||
token := (*c.mqttClient).Subscribe(c.MsgTopic, 0, func(cl mqtt.Client, msg mqtt.Message) {
|
||||
token := (*c.mqttClient).Subscribe(c.MsgTopic, c.MsgQoS, func(cl mqtt.Client, msg mqtt.Message) {
|
||||
|
||||
mp := messagePayload{}
|
||||
err := json.Unmarshal(msg.Payload(), &mp)
|
||||
if err != nil {
|
||||
log.Printf("CLIENT %s failed to decode message", clientID)
|
||||
log.Printf("CLIENT %s failed to decode message\n", clientID)
|
||||
}
|
||||
})
|
||||
|
||||
@@ -144,20 +150,19 @@ func (c *Client) subscribe(wg *sync.WaitGroup, subTimes *res.SubTimes, done *cha
|
||||
|
||||
}
|
||||
|
||||
func (c *Client) pubMessages(in, out chan *message, doneGen chan bool, donePub chan bool, mtls bool) {
|
||||
func (c *Client) pubMessages(in, out chan *message, doneGen chan bool, donePub chan bool) {
|
||||
clientID := fmt.Sprintf("pub-%v-%v", time.Now().Format(time.RFC3339Nano), c.ID)
|
||||
c.ID = clientID
|
||||
ctr := 0
|
||||
onConnected := func(client mqtt.Client) {
|
||||
if !c.Quiet {
|
||||
log.Printf("CLIENT %v is connected to the broker %v\n", clientID, c.BrokerURL)
|
||||
}
|
||||
ctr := 0
|
||||
for {
|
||||
select {
|
||||
case m := <-in:
|
||||
m.Sent = time.Now()
|
||||
m.ID = clientID
|
||||
m.Payload.ID = clientID
|
||||
m.Payload.Sent = m.Sent
|
||||
|
||||
pload, err := json.Marshal(m.Payload)
|
||||
@@ -189,26 +194,38 @@ func (c *Client) pubMessages(in, out chan *message, doneGen chan bool, donePub c
|
||||
}
|
||||
}
|
||||
}
|
||||
connLost := func(client mqtt.Client, reason error) {
|
||||
log.Printf("CLIENT %v had lost connection to the broker: %s\n", c.ID, reason.Error())
|
||||
if ctr < c.MsgCount {
|
||||
flushMessages := make([]message, c.MsgCount-ctr)
|
||||
for _, m := range flushMessages {
|
||||
out <- &m
|
||||
}
|
||||
}
|
||||
donePub <- true
|
||||
}
|
||||
|
||||
c.connect(onConnected, mtls)
|
||||
if c.connect(onConnected, connLost) != nil {
|
||||
out <- &message{}
|
||||
donePub <- true
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func (c *Client) connect(onConnected func(client mqtt.Client), mtls bool) error {
|
||||
func (c *Client) connect(onConnected func(client mqtt.Client), connLost func(client mqtt.Client, reason error)) error {
|
||||
opts := mqtt.NewClientOptions().
|
||||
AddBroker(c.BrokerURL).
|
||||
SetClientID(c.ID).
|
||||
SetCleanSession(true).
|
||||
SetAutoReconnect(true).
|
||||
SetAutoReconnect(false).
|
||||
SetOnConnectHandler(onConnected).
|
||||
SetConnectionLostHandler(func(client mqtt.Client, reason error) {
|
||||
log.Printf("CLIENT %s lost connection to the broker: %v. Will reconnect...\n", c.ID, reason.Error())
|
||||
})
|
||||
SetConnectionLostHandler(connLost)
|
||||
if c.BrokerUser != "" && c.BrokerPass != "" {
|
||||
opts.SetUsername(c.BrokerUser)
|
||||
opts.SetPassword(c.BrokerPass)
|
||||
}
|
||||
if mtls {
|
||||
|
||||
if c.Mtls {
|
||||
|
||||
cfg := &tls.Config{
|
||||
InsecureSkipVerify: c.SkipTLSVer,
|
||||
@@ -216,9 +233,7 @@ func (c *Client) connect(onConnected func(client mqtt.Client), mtls bool) error
|
||||
|
||||
if c.CA != nil {
|
||||
cfg.RootCAs = x509.NewCertPool()
|
||||
if cfg.RootCAs.AppendCertsFromPEM(c.CA) {
|
||||
log.Printf("Successfully added certificate\n")
|
||||
}
|
||||
cfg.RootCAs.AppendCertsFromPEM(c.CA)
|
||||
}
|
||||
if c.ClientCert.Certificate != nil {
|
||||
cfg.Certificates = []tls.Certificate{c.ClientCert}
|
||||
@@ -235,7 +250,7 @@ func (c *Client) connect(onConnected func(client mqtt.Client), mtls bool) error
|
||||
c.mqttClient = &client
|
||||
|
||||
if token.Error() != nil {
|
||||
log.Printf("CLIENT %v had error connecting to the broker: %s\n", c.ID, token.Error())
|
||||
log.Printf("CLIENT %v had error connecting to the broker: %s\n", c.ID, token.Error().Error())
|
||||
return token.Error()
|
||||
}
|
||||
return nil
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
package res
|
||||
package results
|
||||
|
||||
// RunResults describes results of a single client / run
|
||||
type RunResults struct {
|
||||
Executable
+55
@@ -0,0 +1,55 @@
|
||||
#!/bin/bash
|
||||
|
||||
i=0
|
||||
echo "BEGIN TEST " > result.$1.out
|
||||
for mtls in true
|
||||
do
|
||||
for ret in false true
|
||||
do
|
||||
for qos in 0 1 2
|
||||
do
|
||||
for pub in 1 10 100
|
||||
do
|
||||
for sub in 1 10
|
||||
do
|
||||
for message in 100 1000
|
||||
do
|
||||
if [[ $pub -eq 100 && $message -eq 1000 ]];
|
||||
then
|
||||
continue
|
||||
fi
|
||||
|
||||
for size in 100 500
|
||||
do
|
||||
let "i += 1"
|
||||
echo "=================================TEST $i=========================================" >> $1-$i.out
|
||||
echo "MTLS: $mtls RETAIN: $ret, QOS $qos" >> $1-$i.out
|
||||
echo "Pub:" $pub ", Sub:" $sub ", MsgSize:" $size ", MsgPerPub:" $message >> $1-$i.out
|
||||
echo "=================================================================================" >> $1-$i.out
|
||||
if [ "$mtls" = true ];
|
||||
then
|
||||
echo "| " >> $1-$i.out
|
||||
echo "| ./mqtt-bench --channels $3 -s $size -n $message --subs $sub --pubs $pub -q $qos --retain=$ret -m=true -b tcps://$2:8883 --quiet=true --ca ../../../docker/ssl/certs/ca.crt -t=true" >> $1-$i.out
|
||||
echo "| " >> $1-$i.out
|
||||
../cmd/mqtt-bench --channels $3 -s $size -n $message --subs $sub --pubs $pub -q $qos --retain=$ret -m=true -b tcps://$2:8883 --quiet=true --ca ../../../docker/ssl/certs/ca.crt -t=true >> $1-$i.out
|
||||
else
|
||||
echo "| " >> $1-$i.out
|
||||
echo "| ./mqtt-bench --channels $3 -s $size -n $message --subs $sub --pubs $pub -q $qos --retain=$ret -b tcp://$2:1883 --quiet=true" >> $1-$i.out
|
||||
echo "| " >> $1-$i.out
|
||||
../cmd/mqtt-bench --channels $3 -s $size -n $message --subs $sub --pubs $pub -q $qos --retain=$ret -b tcp://$2:1883 --quiet=true >> $1-$i.out
|
||||
fi
|
||||
sleep 2
|
||||
done
|
||||
done
|
||||
done
|
||||
done
|
||||
done
|
||||
|
||||
done
|
||||
done
|
||||
files=`ls test*.out | sort --version-sort `
|
||||
for file in $files
|
||||
do
|
||||
cat $file >> result.$1.out
|
||||
done
|
||||
echo "END TEST " >> result.$1.out
|
||||
@@ -1,12 +0,0 @@
|
||||
broker_url = "tcp://localhost:1883"
|
||||
qos =2
|
||||
message_size =100
|
||||
message_count =100
|
||||
publishers_num =10
|
||||
subscribers_num =1
|
||||
format = "text"
|
||||
quiet = true
|
||||
mtls = true
|
||||
skiptlsver = true
|
||||
ca_file = "ca.crt"
|
||||
channels_file = "channels.toml"
|
||||
@@ -1,12 +0,0 @@
|
||||
broker_url = "tcp://localhost:1883"
|
||||
qos =2
|
||||
message_size =100
|
||||
message_count =100
|
||||
publishers_num =10
|
||||
subscribers_num =1
|
||||
format = "text"
|
||||
quiet = true
|
||||
mtls = false
|
||||
skiptlsver = true
|
||||
ca_file = "ca.crt"
|
||||
channels_file = "channels.toml"
|
||||
@@ -1,12 +0,0 @@
|
||||
broker_url = "tcp://localhost:1883"
|
||||
qos =2
|
||||
message_size =100
|
||||
message_count =100
|
||||
publishers_num =1
|
||||
subscribers_num =10
|
||||
format = "text"
|
||||
quiet = true
|
||||
mtls = true
|
||||
skiptlsver = true
|
||||
ca_file = "ca.crt"
|
||||
channels_file = "channels.toml"
|
||||
@@ -1,12 +0,0 @@
|
||||
broker_url = "tcp://localhost:1883"
|
||||
qos =2
|
||||
message_size =100
|
||||
message_count =100
|
||||
publishers_num =1
|
||||
subscribers_num =10
|
||||
format = "text"
|
||||
quiet = true
|
||||
mtls = false
|
||||
skiptlsver = true
|
||||
ca_file = "ca.crt"
|
||||
channels_file = "channels.toml"
|
||||
@@ -1,12 +0,0 @@
|
||||
broker_url = "tcp://localhost:1883"
|
||||
qos =2
|
||||
message_size =100
|
||||
message_count =100
|
||||
publishers_num =0
|
||||
subscribers_num =100
|
||||
format = "text"
|
||||
quiet = true
|
||||
mtls = true
|
||||
skiptlsver = true
|
||||
ca_file = "ca.crt"
|
||||
channels_file = "channels.toml"
|
||||
@@ -1,12 +0,0 @@
|
||||
broker_url = "tcp://localhost:1883"
|
||||
qos =2
|
||||
message_size =100
|
||||
message_count =100
|
||||
publishers_num =0
|
||||
subscribers_num =100
|
||||
format = "text"
|
||||
quiet = true
|
||||
mtls = false
|
||||
skiptlsver = true
|
||||
ca_file = "ca.crt"
|
||||
channels_file = "channels.toml"
|
||||
@@ -4,21 +4,33 @@ A simple utility to create a list of channels and things connected with possibil
|
||||
|
||||
### Usage
|
||||
|
||||
* host - hostname where mainflux instance is running
|
||||
* username - mainflux username
|
||||
* password - mainflux user Password
|
||||
* num - number of channels to creat
|
||||
* ssl - create certificates for each channel for mtls access
|
||||
* ca - CA crt file used for signing certificates
|
||||
* prefix - for channels and things Name
|
||||
* --ca string CA for creating and signing things certificate (default "ca.crt")
|
||||
* --cakey string ca.key for creating and signing things certificate (default "ca.key")
|
||||
* -h, --help help for mqtt-prov
|
||||
* --host string address for mainflux instance (default "https://localhost")
|
||||
* --num int number of channels and things to create and connect (default 10)
|
||||
* -p, --password string mainflux users password
|
||||
* --ssl create certificates for mTLS access
|
||||
* -u, --username string mainflux user (default "test@mainflux.com")
|
||||
|
||||
|
||||
|
||||
```
|
||||
go run main.go --host http://localhost --num 10 --username test@mainflux.com --password test1234 --ssl --ca ../../docker/ssl/certs/ca.crt
|
||||
go build -o mqtt-prov *.go
|
||||
|
||||
./mqtt-prov -u mirkot@mainflux.com -p test1234 --host https://142.93.118.47
|
||||
```
|
||||
if you want to create a list of channels with certificates
|
||||
```
|
||||
|
||||
example entry in channels.toml fail
|
||||
./mqtt-prov --host http://localhost --num 10 -u test@mainflux.com -p test1234 --ssl true --ca ../../docker/ssl/certs/ca.crt --cakey ../../docker/ssl/certs/ca.key
|
||||
|
||||
```
|
||||
ca.crt and ca.key are used for creating things certificate and for HTTPS, if you are provision on remote server you will have to get these files to your local
|
||||
|
||||
|
||||
|
||||
example entry in channels.toml file
|
||||
|
||||
```
|
||||
[[Connection]]
|
||||
|
||||
+56
-25
@@ -10,7 +10,6 @@ import (
|
||||
"crypto/x509"
|
||||
"crypto/x509/pkix"
|
||||
"encoding/pem"
|
||||
"flag"
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"log"
|
||||
@@ -20,6 +19,7 @@ import (
|
||||
|
||||
"github.com/BurntSushi/toml"
|
||||
sdk "github.com/mainflux/mainflux/sdk/go"
|
||||
"github.com/spf13/cobra"
|
||||
)
|
||||
|
||||
type connection struct {
|
||||
@@ -29,10 +29,22 @@ type connection struct {
|
||||
MTLSCert string
|
||||
MTLSKey string
|
||||
}
|
||||
|
||||
type connections struct {
|
||||
Connection []connection
|
||||
}
|
||||
|
||||
var (
|
||||
host string
|
||||
username string
|
||||
password string
|
||||
num int
|
||||
ssl bool
|
||||
ca string
|
||||
cakey string
|
||||
prefix string
|
||||
)
|
||||
|
||||
const (
|
||||
CRT_LOCATION = "certs"
|
||||
KEY = "default"
|
||||
@@ -42,26 +54,41 @@ const (
|
||||
CN = "localhost"
|
||||
CRT_FILE_NAME = "thing"
|
||||
rsaBits = 4096
|
||||
daysValid = 730
|
||||
daysValid = "2400h"
|
||||
)
|
||||
|
||||
func main() {
|
||||
var (
|
||||
host = flag.String("host", "http://localhost", "Mainflux host address")
|
||||
username = flag.String("username", "mirkot@mainflux.com", "mainflux user")
|
||||
password = flag.String("password", "test1234", "mainflux user password")
|
||||
num = flag.Int("num", 1, "number of created channels")
|
||||
ssl = flag.Bool("ssl", true, "create thing certs")
|
||||
ca = flag.String("ca", "ca.crt", "CA file for creating things certs")
|
||||
cakey = flag.String("cakey", "ca.key", "CA private key file")
|
||||
prefix = flag.String("prefix", "test", "name prefix for created channels and things")
|
||||
)
|
||||
var provisionCmd = &cobra.Command{
|
||||
Use: "mqtt-prov",
|
||||
Run: func(cmd *cobra.Command, args []string) {
|
||||
provision()
|
||||
},
|
||||
}
|
||||
|
||||
flag.Parse()
|
||||
func init() {
|
||||
provisionCmd.PersistentFlags().StringVarP(&host, "host", "", "https://localhost", "address for mainflux instance")
|
||||
provisionCmd.PersistentFlags().StringVarP(&username, "username", "u", "test@mainflux.com", "mainflux user")
|
||||
provisionCmd.PersistentFlags().StringVarP(&password, "password", "p", "", "mainflux users password")
|
||||
provisionCmd.PersistentFlags().IntVarP(&num, "num", "", 10, "number of channels and things to create and connect")
|
||||
provisionCmd.PersistentFlags().BoolVarP(&ssl, "ssl", "", false, "create certificates for mTLS access")
|
||||
provisionCmd.PersistentFlags().StringVarP(&cakey, "cakey", "", "ca.key", "ca.key for creating and signing things certificate")
|
||||
provisionCmd.PersistentFlags().StringVarP(&ca, "ca", "", "ca.crt", "CA for creating and signing things certificate")
|
||||
}
|
||||
|
||||
func main() {
|
||||
execute()
|
||||
}
|
||||
|
||||
func execute() {
|
||||
if err := provisionCmd.Execute(); err != nil {
|
||||
log.Fatalf(err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
func provision() {
|
||||
|
||||
msgContentType := string(sdk.CTJSONSenML)
|
||||
sdkConf := sdk.Config{
|
||||
BaseURL: *host,
|
||||
BaseURL: host,
|
||||
ReaderURL: "http://localhost:8905",
|
||||
ReaderPrefix: "",
|
||||
UsersPrefix: "",
|
||||
@@ -74,8 +101,8 @@ func main() {
|
||||
s := sdk.NewSDK(sdkConf)
|
||||
|
||||
user := sdk.User{
|
||||
Email: *username,
|
||||
Password: *password,
|
||||
Email: username,
|
||||
Password: password,
|
||||
}
|
||||
|
||||
token, err := s.CreateToken(user)
|
||||
@@ -90,13 +117,13 @@ func main() {
|
||||
var tlsCert tls.Certificate
|
||||
var caCert *x509.Certificate
|
||||
|
||||
if *ssl {
|
||||
tlsCert, err = tls.LoadX509KeyPair(*ca, *cakey)
|
||||
if ssl {
|
||||
tlsCert, err = tls.LoadX509KeyPair(ca, cakey)
|
||||
if err != nil {
|
||||
log.Fatalf("Failed to load CA cert")
|
||||
}
|
||||
|
||||
b, err := ioutil.ReadFile(*ca)
|
||||
b, err := ioutil.ReadFile(ca)
|
||||
if err != nil {
|
||||
log.Fatalf("Failed to load CA cert")
|
||||
}
|
||||
@@ -112,15 +139,15 @@ func main() {
|
||||
|
||||
}
|
||||
|
||||
for i := 0; i < *num; i++ {
|
||||
for i := 0; i < num; i++ {
|
||||
|
||||
m, err := createThing(s, fmt.Sprintf("%s-thing-%d", *prefix, i), token)
|
||||
m, err := createThing(s, fmt.Sprintf("%s-thing-%d", prefix, i), token)
|
||||
if err != nil {
|
||||
log.Println("Failed to create thing")
|
||||
return
|
||||
}
|
||||
|
||||
ch, err := createChannel(s, fmt.Sprintf("%s-channel-%d", *prefix, i), token)
|
||||
ch, err := createChannel(s, fmt.Sprintf("%s-channel-%d", prefix, i), token)
|
||||
|
||||
if err := s.ConnectThing(m.ID, ch.ID, token); err != nil {
|
||||
log.Println("Failed to create thing")
|
||||
@@ -132,12 +159,16 @@ func main() {
|
||||
things = append(things, m)
|
||||
cert := ""
|
||||
key := ""
|
||||
if *ssl {
|
||||
if ssl {
|
||||
var priv interface{}
|
||||
priv, err = rsa.GenerateKey(rand.Reader, rsaBits)
|
||||
|
||||
notBefore := time.Now()
|
||||
notAfter := notBefore.Add(daysValid)
|
||||
validFor, err := time.ParseDuration(daysValid)
|
||||
if err != nil {
|
||||
log.Fatalf("Failed to set date %v", validFor)
|
||||
}
|
||||
notAfter := notBefore.Add(validFor)
|
||||
|
||||
serialNumberLimit := new(big.Int).Lsh(big.NewInt(1), 128)
|
||||
serialNumber, err := rand.Int(rand.Reader, serialNumberLimit)
|
||||
|
||||
Reference in New Issue
Block a user