opensource/magistrala
1
0
Fork 0
mirror of https://github.com/absmach/magistrala.git synced 2026-08-07 07:14:46 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

MG-2456 - Refactor architecture (#2494)

Browse Source 
Signed-off-by: Felix Gateru <felix.gateru@gmail.com>
Signed-off-by: Arvindh <arvindh91@gmail.com>
Signed-off-by: Dusan Borovcanin <borovcanindusan1@gmail.com>
Co-authored-by: Arvindh <30824765+arvindh123@users.noreply.github.com>
Co-authored-by: Felix Gateru <felix.gateru@gmail.com>
This commit is contained in:
Dušan Borovčanin
2024-12-02 12:42:38 +01:00
committed by Dusan Borovcanin
parent e29b54c54d
commit 243ccade0b
535 changed files with 65453 additions and 39551 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/config/mockery.yaml
+24 -6
View File
@@ -6,23 +6,41 @@ filename: "{{.InterfaceName}}.go"
outpkg: "mocks"
boilerplate-file: "./tools/config/boilerplate.txt"
packages:
github.com/absmach/magistrala:
github.com/absmach/magistrala/internal/grpc/clients/v1:
interfaces:
ThingsServiceClient:
ClientsServiceClient:
config:
dir: "./things/mocks"
mockname: "ThingsServiceClient"
filename: "things_client.go"
dir: "./clients/mocks"
mockname: "ClientsServiceClient"
filename: "clients_client.go"
github.com/absmach/magistrala/internal/grpc/domains/v1:
interfaces:
DomainsServiceClient:
config:
dir: "./auth/mocks"
dir: "./domains/mocks"
mockname: "DomainsServiceClient"
filename: "domains_client.go"
github.com/absmach/magistrala/internal/grpc/token/v1:
interfaces:
TokenServiceClient:
config:
dir: "./auth/mocks"
mockname: "TokenServiceClient"
filename: "token_client.go"
github.com/absmach/magistrala/internal/grpc/channels/v1:
interfaces:
ChannelsServiceClient:
config:
dir: "./channels/mocks"
mockname: "ChannelsServiceClient"
filename: "channels_client.go"
github.com/absmach/magistrala/internal/grpc/groups/v1:
interfaces:
GroupsServiceClient:
config:
dir: "./groups/mocks"
mockname: "GroupsServiceClient"
filename: "groups_client.go"
github.com/absmach/magistrala/certs/pki/amcerts:
interfaces:
tools/e2e/README.md
+11 -11
View File
@@ -1,6 +1,6 @@
# Magistrala Users Groups Things and Channels E2E Testing Tool
# Magistrala Users Groups Clients and Channels E2E Testing Tool
A simple utility to create a list of groups and users connected to these groups and channels and things connected to these channels.
A simple utility to create a list of groups and users connected to these groups and channels and clients connected to these channels.
## Installation
@@ -14,9 +14,9 @@ make
```bash
./e2e --help
Tool for testing end-to-end flow of Magistrala by doing a couple of operations namely:
1. Creating, viewing, updating and changing status of users, groups, things and channels.
2. Connecting users and groups to each other and things and channels to each other.
3. Sending messages from things to channels on all 4 protocol adapters (HTTP, WS, CoAP and MQTT).
1. Creating, viewing, updating and changing status of users, groups, clients and channels.
2. Connecting users and groups to each other and clients and channels to each other.
3. Sending messages from clients to channels on all 4 protocol adapters (HTTP, WS, CoAP and MQTT).
Complete documentation is available at https://docs.magistrala.abstractmachines.fr
@@ -39,9 +39,9 @@ Flags:
-h, --help help for e2e
-H, --host string address for a running Magistrala instance (default "localhost")
-n, --num uint number of users, groups, channels and things to create and connect (default 10)
-n, --num uint number of users, groups, channels and clients to create and connect (default 10)
-N, --num_of_messages uint number of messages to send (default 10)
-p, --prefix string name prefix for users, groups, things and channels
-p, --prefix string name prefix for users, groups, clients and channels
```
To use `-H` option, you can specify the address for the Magistrala instance as an argument when running the program. For example, if the Magistrala instance is running on another computer with the IP address 192.168.0.1, you could use the following command:
@@ -74,7 +74,7 @@ c8fe4d9d-3ad6-4687-83c0-171356f3e4f6
513f7295-0923-4e21-b41a-3cfd1cb7b9b9
54bd71ea-3c22-401e-89ea-d58162b983c0
ae91b327-4c40-4e68-91fe-cd6223ee4e99
created things of ids:
created clients of ids:
5909a907-7413-47d4-b793-e1eb36988a5f
f9b6bc18-1862-4a24-8973-adde11cb3303
c2bd6eed-6f38-464c-989c-fe8ec8c084ba
@@ -86,8 +86,8 @@ d654948d-d6c1-4eae-b69a-29c853282c3d
2c2a5496-89cf-47e6-9d38-5fd5542337bd
7ab3319d-269c-4b07-9dc5-f9906693e894
5d8fa139-10e7-4683-94f3-4e881b4db041
created policies for users, groups, things and channels
viewed users, groups, things and channels
updated users, groups, things and channels
created policies for users, groups, clients and channels
viewed users, groups, clients and channels
updated users, groups, clients and channels
sent messages to channels
```
tools/e2e/cmd/main.go
+5 -5
View File
@@ -21,9 +21,9 @@ func main() {
Use: "e2e",
Short: "e2e is end-to-end testing tool for Magistrala",
Long: "Tool for testing end-to-end flow of magistrala by doing a couple of operations namely:\n" +
"1. Creating, viewing, updating and changing status of users, groups, things and channels.\n" +
"2. Connecting users and groups to each other and things and channels to each other.\n" +
"3. Sending messages from things to channels on all 4 protocol adapters (HTTP, WS, CoAP and MQTT).\n" +
"1. Creating, viewing, updating and changing status of users, groups, clients and channels.\n" +
"2. Connecting users and groups to each other and clients and channels to each other.\n" +
"3. Sending messages from clients to channels on all 4 protocol adapters (HTTP, WS, CoAP and MQTT).\n" +
"Complete documentation is available at https://docs.magistrala.abstractmachines.fr",
Example: "Here is a simple example of using e2e tool.\n" +
"Use the following commands from the root magistrala directory:\n\n" +
@@ -48,8 +48,8 @@ func main() {
// Root Flags
rootCmd.PersistentFlags().StringVarP(&econf.Host, "host", "H", "localhost", "address for a running magistrala instance")
rootCmd.PersistentFlags().StringVarP(&econf.Prefix, "prefix", "p", "", "name prefix for users, groups, things and channels")
rootCmd.PersistentFlags().Uint64VarP(&econf.Num, "num", "n", defNum, "number of users, groups, channels and things to create and connect")
rootCmd.PersistentFlags().StringVarP(&econf.Prefix, "prefix", "p", "", "name prefix for users, groups, clients and channels")
rootCmd.PersistentFlags().Uint64VarP(&econf.Num, "num", "n", defNum, "number of users, groups, channels and clients to create and connect")
rootCmd.PersistentFlags().Uint64VarP(&econf.NumOfMsg, "num_of_messages", "N", defNum, "number of messages to send")
if err := rootCmd.Execute(); err != nil {
tools/e2e/e2e.go
+96 -95
View File
@@ -26,7 +26,7 @@ const (
numAdapters = 4
batchSize = 99
usersPort = "9002"
thingsPort = "9000"
clientsPort = "9000"
domainsPort = "8189"
)
@@ -55,17 +55,17 @@ type Config struct {
// - Create groups using hierarchy
// - Do Read, Update and Change of Status operations on groups.
// - Create things
// - Do Read, Update and Change of Status operations on things.
// - Create clients
// - Do Read, Update and Change of Status operations on clients.
// - Create channels
// - Do Read, Update and Change of Status operations on channels.
// - Connect thing to channel
// - Connect client to channel
// - Publish message from HTTP, MQTT, WS and CoAP Adapters.
func Test(conf Config) {
sdkConf := sdk.Config{
ThingsURL: fmt.Sprintf("http://%s:%s", conf.Host, thingsPort),
ClientsURL: fmt.Sprintf("http://%s:%s", conf.Host, clientsPort),
UsersURL: fmt.Sprintf("http://%s:%s", conf.Host, usersPort),
DomainsURL: fmt.Sprintf("http://%s:%s", conf.Host, domainsPort),
HTTPAdapterURL: fmt.Sprintf("http://%s/http", conf.Host),
@@ -95,11 +95,11 @@ func Test(conf Config) {
}
color.Success.Printf("created groups of ids:\n%s\n", magenta(getIDS(groups)))
things, err := createThings(s, conf, domainID, token)
clients, err := createClients(s, conf, domainID, token)
if err != nil {
errExit(fmt.Errorf("unable to create things: %w", err))
errExit(fmt.Errorf("unable to create clients: %w", err))
}
color.Success.Printf("created things of ids:\n%s\n", magenta(getIDS(things)))
color.Success.Printf("created clients of ids:\n%s\n", magenta(getIDS(clients)))
channels, err := createChannels(s, conf, domainID, token)
if err != nil {
@@ -107,20 +107,20 @@ func Test(conf Config) {
}
color.Success.Printf("created channels of ids:\n%s\n", magenta(getIDS(channels)))
// List users, groups, things and channels
if err := read(s, conf, domainID, token, users, groups, things, channels); err != nil {
errExit(fmt.Errorf("unable to read users, groups, things and channels: %w", err))
// List users, groups, clients and channels
if err := read(s, conf, domainID, token, users, groups, clients, channels); err != nil {
errExit(fmt.Errorf("unable to read users, groups, clients and channels: %w", err))
}
color.Success.Println("viewed users, groups, things and channels")
color.Success.Println("viewed users, groups, clients and channels")
// Update users, groups, things and channels
if err := update(s, domainID, token, users, groups, things, channels); err != nil {
errExit(fmt.Errorf("unable to update users, groups, things and channels: %w", err))
// Update users, groups, clients and channels
if err := update(s, domainID, token, users, groups, clients, channels); err != nil {
errExit(fmt.Errorf("unable to update users, groups, clients and channels: %w", err))
}
color.Success.Println("updated users, groups, things and channels")
color.Success.Println("updated users, groups, clients and channels")
// Send messages to channels
if err := messaging(s, conf, domainID, token, things, channels); err != nil {
if err := messaging(s, conf, domainID, token, clients, channels); err != nil {
errExit(fmt.Errorf("unable to send messages to channels: %w", err))
}
color.Success.Println("sent messages to channels")
@@ -227,50 +227,50 @@ func createGroups(s sdk.SDK, conf Config, domainID, token string) ([]sdk.Group,
return groups, nil
}
func createThingsInBatch(s sdk.SDK, conf Config, domainID, token string, num uint64) ([]sdk.Thing, error) {
func createClientsInBatch(s sdk.SDK, conf Config, domainID, token string, num uint64) ([]sdk.Client, error) {
var err error
things := make([]sdk.Thing, num)
clients := make([]sdk.Client, num)
for i := uint64(0); i < num; i++ {
things[i] = sdk.Thing{
clients[i] = sdk.Client{
Name: fmt.Sprintf("%s%s", conf.Prefix, namesgenerator.Generate()),
}
}
things, err = s.CreateThings(things, domainID, token)
clients, err = s.CreateClients(clients, domainID, token)
if err != nil {
return []sdk.Thing{}, fmt.Errorf("failed to create the things: %w", err)
return []sdk.Client{}, fmt.Errorf("failed to create the clients: %w", err)
}
return things, nil
return clients, nil
}
func createThings(s sdk.SDK, conf Config, domainID, token string) ([]sdk.Thing, error) {
things := []sdk.Thing{}
func createClients(s sdk.SDK, conf Config, domainID, token string) ([]sdk.Client, error) {
clients := []sdk.Client{}
if conf.Num > batchSize {
batches := int(conf.Num) / batchSize
for i := 0; i < batches; i++ {
ths, err := createThingsInBatch(s, conf, domainID, token, batchSize)
ths, err := createClientsInBatch(s, conf, domainID, token, batchSize)
if err != nil {
return []sdk.Thing{}, fmt.Errorf("failed to create the things: %w", err)
return []sdk.Client{}, fmt.Errorf("failed to create the clients: %w", err)
}
things = append(things, ths...)
clients = append(clients, ths...)
}
ths, err := createThingsInBatch(s, conf, domainID, token, conf.Num%uint64(batchSize))
ths, err := createClientsInBatch(s, conf, domainID, token, conf.Num%uint64(batchSize))
if err != nil {
return []sdk.Thing{}, fmt.Errorf("failed to create the things: %w", err)
return []sdk.Client{}, fmt.Errorf("failed to create the clients: %w", err)
}
things = append(things, ths...)
clients = append(clients, ths...)
} else {
ths, err := createThingsInBatch(s, conf, domainID, token, conf.Num)
ths, err := createClientsInBatch(s, conf, domainID, token, conf.Num)
if err != nil {
return []sdk.Thing{}, fmt.Errorf("failed to create the things: %w", err)
return []sdk.Client{}, fmt.Errorf("failed to create the clients: %w", err)
}
things = append(things, ths...)
clients = append(clients, ths...)
}
return things, nil
return clients, nil
}
func createChannelsInBatch(s sdk.SDK, conf Config, domainID, token string, num uint64) ([]sdk.Channel, error) {
@@ -318,7 +318,7 @@ func createChannels(s sdk.SDK, conf Config, domainID, token string) ([]sdk.Chann
return channels, nil
}
func read(s sdk.SDK, conf Config, domainID, token string, users []sdk.User, groups []sdk.Group, things []sdk.Thing, channels []sdk.Channel) error {
func read(s sdk.SDK, conf Config, domainID, token string, users []sdk.User, groups []sdk.Group, clients []sdk.Client, channels []sdk.Channel) error {
for _, user := range users {
if _, err := s.User(user.ID, token); err != nil {
return fmt.Errorf("failed to get user %w", err)
@@ -343,17 +343,17 @@ func read(s sdk.SDK, conf Config, domainID, token string, users []sdk.User, grou
if gp.Total < conf.Num {
return fmt.Errorf("returned groups %d less than created groups %d", gp.Total, conf.Num)
}
for _, thing := range things {
if _, err := s.Thing(thing.ID, domainID, token); err != nil {
return fmt.Errorf("failed to get thing %w", err)
for _, c := range clients {
if _, err := s.Client(c.ID, domainID, token); err != nil {
return fmt.Errorf("failed to get client %w", err)
}
}
tp, err := s.Things(sdk.PageMetadata{}, domainID, token)
tp, err := s.Clients(sdk.PageMetadata{}, domainID, token)
if err != nil {
return fmt.Errorf("failed to get things %w", err)
return fmt.Errorf("failed to get clients %w", err)
}
if tp.Total < conf.Num {
return fmt.Errorf("returned things %d less than created things %d", tp.Total, conf.Num)
return fmt.Errorf("returned clients %d less than created clients %d", tp.Total, conf.Num)
}
for _, channel := range channels {
if _, err := s.Channel(channel.ID, domainID, token); err != nil {
@@ -371,7 +371,7 @@ func read(s sdk.SDK, conf Config, domainID, token string, users []sdk.User, grou
return nil
}
func update(s sdk.SDK, domainID, token string, users []sdk.User, groups []sdk.Group, things []sdk.Thing, channels []sdk.Channel) error {
func update(s sdk.SDK, domainID, token string, users []sdk.User, groups []sdk.Group, clients []sdk.Client, channels []sdk.Channel) error {
for _, user := range users {
user.FirstName = namesgenerator.Generate()
user.Metadata = sdk.Metadata{"Update": namesgenerator.Generate()}
@@ -458,48 +458,48 @@ func update(s sdk.SDK, domainID, token string, users []sdk.User, groups []sdk.Gr
return fmt.Errorf("failed to enable group before %s after %s", group.Status, rGroup.Status)
}
}
for _, thing := range things {
thing.Name = namesgenerator.Generate()
thing.Metadata = sdk.Metadata{"Update": namesgenerator.Generate()}
rThing, err := s.UpdateThing(thing, domainID, token)
for _, t := range clients {
t.Name = namesgenerator.Generate()
t.Metadata = sdk.Metadata{"Update": namesgenerator.Generate()}
rClient, err := s.UpdateClient(t, domainID, token)
if err != nil {
return fmt.Errorf("failed to update thing %w", err)
return fmt.Errorf("failed to update client %w", err)
}
if rThing.Name != thing.Name {
return fmt.Errorf("failed to update thing name before %s after %s", thing.Name, rThing.Name)
if rClient.Name != t.Name {
return fmt.Errorf("failed to update client name before %s after %s", t.Name, rClient.Name)
}
if rThing.Metadata["Update"] != thing.Metadata["Update"] {
return fmt.Errorf("failed to update thing metadata before %s after %s", thing.Metadata["Update"], rThing.Metadata["Update"])
if rClient.Metadata["Update"] != t.Metadata["Update"] {
return fmt.Errorf("failed to update client metadata before %s after %s", t.Metadata["Update"], rClient.Metadata["Update"])
}
thing = rThing
rThing, err = s.UpdateThingSecret(thing.ID, thing.Credentials.Secret, domainID, token)
t = rClient
rClient, err = s.UpdateClientSecret(t.ID, t.Credentials.Secret, domainID, token)
if err != nil {
return fmt.Errorf("failed to update thing secret %w", err)
return fmt.Errorf("failed to update client secret %w", err)
}
thing = rThing
thing.Tags = []string{namesgenerator.Generate()}
rThing, err = s.UpdateThingTags(thing, domainID, token)
t = rClient
t.Tags = []string{namesgenerator.Generate()}
rClient, err = s.UpdateClientTags(t, domainID, token)
if err != nil {
return fmt.Errorf("failed to update thing tags %w", err)
return fmt.Errorf("failed to update client tags %w", err)
}
if rThing.Tags[0] != thing.Tags[0] {
return fmt.Errorf("failed to update thing tags before %s after %s", thing.Tags[0], rThing.Tags[0])
if rClient.Tags[0] != t.Tags[0] {
return fmt.Errorf("failed to update client tags before %s after %s", t.Tags[0], rClient.Tags[0])
}
thing = rThing
rThing, err = s.DisableThing(thing.ID, domainID, token)
t = rClient
rClient, err = s.DisableClient(t.ID, domainID, token)
if err != nil {
return fmt.Errorf("failed to disable thing %w", err)
return fmt.Errorf("failed to disable client %w", err)
}
if rThing.Status != sdk.DisabledStatus {
return fmt.Errorf("failed to disable thing before %s after %s", thing.Status, rThing.Status)
if rClient.Status != sdk.DisabledStatus {
return fmt.Errorf("failed to disable client before %s after %s", t.Status, rClient.Status)
}
thing = rThing
rThing, err = s.EnableThing(thing.ID, domainID, token)
t = rClient
rClient, err = s.EnableClient(t.ID, domainID, token)
if err != nil {
return fmt.Errorf("failed to enable thing %w", err)
return fmt.Errorf("failed to enable client %w", err)
}
if rThing.Status != sdk.EnabledStatus {
return fmt.Errorf("failed to enable thing before %s after %s", thing.Status, rThing.Status)
if rClient.Status != sdk.EnabledStatus {
return fmt.Errorf("failed to enable client before %s after %s", t.Status, rClient.Status)
}
}
for _, channel := range channels {
@@ -536,15 +536,16 @@ func update(s sdk.SDK, domainID, token string, users []sdk.User, groups []sdk.Gr
return nil
}
func messaging(s sdk.SDK, conf Config, domainID, token string, things []sdk.Thing, channels []sdk.Channel) error {
for _, thing := range things {
func messaging(s sdk.SDK, conf Config, domainID, token string, clients []sdk.Client, channels []sdk.Channel) error {
for _, c := range clients {
for _, channel := range channels {
conn := sdk.Connection{
ThingID: thing.ID,
ChannelID: channel.ID,
ClientIDs: []string{c.ID},
ChannelIDs: []string{channel.ID},
Types: []string{"publish", "subscribe"},
}
if err := s.Connect(conn, domainID, token); err != nil {
return fmt.Errorf("failed to connect thing %s to channel %s", thing.ID, channel.ID)
return fmt.Errorf("failed to connect client %s to channel %s", c.ID, channel.ID)
}
}
}
@@ -553,26 +554,26 @@ func messaging(s sdk.SDK, conf Config, domainID, token string, things []sdk.Thin
bt := time.Now().Unix()
for i := uint64(0); i < conf.NumOfMsg; i++ {
for _, thing := range things {
for _, client := range clients {
for _, channel := range channels {
func(num int64, thing sdk.Thing, channel sdk.Channel) {
func(num int64, client sdk.Client, channel sdk.Channel) {
g.Go(func() error {
msg := fmt.Sprintf(msgFormat, num+1, rand.Int())
return sendHTTPMessage(s, msg, thing, channel.ID)
return sendHTTPMessage(s, msg, client, channel.ID)
})
g.Go(func() error {
msg := fmt.Sprintf(msgFormat, num+2, rand.Int())
return sendCoAPMessage(msg, thing, channel.ID)
return sendCoAPMessage(msg, client, channel.ID)
})
g.Go(func() error {
msg := fmt.Sprintf(msgFormat, num+3, rand.Int())
return sendMQTTMessage(msg, thing, channel.ID)
return sendMQTTMessage(msg, client, channel.ID)
})
g.Go(func() error {
msg := fmt.Sprintf(msgFormat, num+4, rand.Int())
return sendWSMessage(conf, msg, thing, channel.ID)
return sendWSMessage(conf, msg, client, channel.ID)
})
}(bt, thing, channel)
}(bt, client, channel)
bt += numAdapters
}
}
@@ -581,42 +582,42 @@ func messaging(s sdk.SDK, conf Config, domainID, token string, things []sdk.Thin
return g.Wait()
}
func sendHTTPMessage(s sdk.SDK, msg string, thing sdk.Thing, chanID string) error {
if err := s.SendMessage(chanID, msg, thing.Credentials.Secret); err != nil {
return fmt.Errorf("HTTP failed to send message from thing %s to channel %s: %w", thing.ID, chanID, err)
func sendHTTPMessage(s sdk.SDK, msg string, client sdk.Client, chanID string) error {
if err := s.SendMessage(chanID, msg, client.Credentials.Secret); err != nil {
return fmt.Errorf("HTTP failed to send message from client %s to channel %s: %w", client.ID, chanID, err)
}
return nil
}
func sendCoAPMessage(msg string, thing sdk.Thing, chanID string) error {
cmd := exec.Command("coap-cli", "post", fmt.Sprintf("channels/%s/messages", chanID), "--auth", thing.Credentials.Secret, "-d", msg)
func sendCoAPMessage(msg string, client sdk.Client, chanID string) error {
cmd := exec.Command("coap-cli", "post", fmt.Sprintf("channels/%s/messages", chanID), "--auth", client.Credentials.Secret, "-d", msg)
if _, err := cmd.CombinedOutput(); err != nil {
return fmt.Errorf("CoAP failed to send message from thing %s to channel %s: %w", thing.ID, chanID, err)
return fmt.Errorf("CoAP failed to send message from client %s to channel %s: %w", client.ID, chanID, err)
}
return nil
}
func sendMQTTMessage(msg string, thing sdk.Thing, chanID string) error {
cmd := exec.Command("mosquitto_pub", "--id-prefix", "magistrala", "-u", thing.ID, "-P", thing.Credentials.Secret, "-t", fmt.Sprintf("channels/%s/messages", chanID), "-h", "localhost", "-m", msg)
func sendMQTTMessage(msg string, client sdk.Client, chanID string) error {
cmd := exec.Command("mosquitto_pub", "--id-prefix", "magistrala", "-u", client.ID, "-P", client.Credentials.Secret, "-t", fmt.Sprintf("channels/%s/messages", chanID), "-h", "localhost", "-m", msg)
if _, err := cmd.CombinedOutput(); err != nil {
return fmt.Errorf("MQTT failed to send message from thing %s to channel %s: %w", thing.ID, chanID, err)
return fmt.Errorf("MQTT failed to send message from client %s to channel %s: %w", client.ID, chanID, err)
}
return nil
}
func sendWSMessage(conf Config, msg string, thing sdk.Thing, chanID string) error {
func sendWSMessage(conf Config, msg string, client sdk.Client, chanID string) error {
socketURL := fmt.Sprintf("ws://%s:%s/channels/%s/messages", conf.Host, defWSPort, chanID)
header := http.Header{"Authorization": []string{thing.Credentials.Secret}}
header := http.Header{"Authorization": []string{client.Credentials.Secret}}
conn, _, err := websocket.DefaultDialer.Dial(socketURL, header)
if err != nil {
return fmt.Errorf("unable to connect to websocket: %w", err)
}
defer conn.Close()
if err := conn.WriteMessage(websocket.TextMessage, []byte(msg)); err != nil {
return fmt.Errorf("WS failed to send message from thing %s to channel %s: %w", thing.ID, chanID, err)
return fmt.Errorf("WS failed to send message from client %s to channel %s: %w", client.ID, chanID, err)
}
return nil
tools/mqtt-bench/README.md
+4 -4
View File
@@ -2,10 +2,10 @@
A simple MQTT benchmarking tool for Magistrala platform.
It connects Magistrala things as subscribers over a number of channels and
uses other Magistrala things to publish messages and create MQTT load.
It connects Magistrala clients as subscribers over a number of channels and
uses other Magistrala clients to publish messages and create MQTT load.
Magistrala things used must be pre-provisioned first, and Magistrala `provision` tool can be used for this purpose.
Magistrala clients used must be pre-provisioned first, and Magistrala `provision` tool can be used for this purpose.
## Installation
@@ -46,7 +46,7 @@ Flags:
Two output formats supported: human-readable plain text and JSON.
Before use you need a `mgconn.toml` - a TOML file that describes Magistrala connection data (channels, thingIDs, thingKeys, certs).
Before use you need a `mgconn.toml` - a TOML file that describes Magistrala connection data (channels, clientIDs, clientKeys, certs).
You can use `provision` tool (in tools/provision) to create this TOML config file.
```bash
tools/mqtt-bench/bench.go
+6 -6
View File
@@ -72,16 +72,16 @@ func Benchmark(cfg Config) error {
go func(i int) {
defer wg.Done()
mgChan := mg.Channels[i%n]
mgThing := mg.Things[i%n]
mgCli := mg.Clients[i%n]
if cfg.MQTT.TLS.MTLS {
cert, err = tls.X509KeyPair([]byte(mgThing.MTLSCert), []byte(mgThing.MTLSKey))
cert, err = tls.X509KeyPair([]byte(mgCli.MTLSCert), []byte(mgCli.MTLSKey))
if err != nil {
errorChan <- err
return
}
}
c, err := makeClient(i, cfg, mgChan, mgThing, startStamp, caByte, cert)
c, err := makeClient(i, cfg, mgChan, mgCli, startStamp, caByte, cert)
if err != nil {
errorChan <- fmt.Errorf("unable to create message payload %s", err.Error())
return
@@ -171,12 +171,12 @@ func getBytePayload(size int, m message) (handler, error) {
return ret, nil
}
func makeClient(i int, cfg Config, mgChan mgChannel, mgThing mgThing, start time.Time, caCert []byte, clientCert tls.Certificate) (*Client, error) {
func makeClient(i int, cfg Config, mgChan mgChannel, cli mgClient, start time.Time, caCert []byte, clientCert tls.Certificate) (*Client, error) {
c := &Client{
ID: strconv.Itoa(i),
BrokerURL: cfg.MQTT.Broker.URL,
BrokerUser: mgThing.ThingID,
BrokerPass: mgThing.ThingKey,
BrokerUser: cli.ClientID,
BrokerPass: cli.ClientSecret,
MsgTopic: fmt.Sprintf("channels/%s/messages/%d/test", mgChan.ChannelID, start.UnixNano()),
MsgSize: cfg.MQTT.Message.Size,
MsgCount: cfg.Test.Count,
tools/mqtt-bench/config.go
+6 -6
View File
@@ -43,11 +43,11 @@ type magistralaFile struct {
ConnFile string `toml:"connections_file" mapstructure:"connections_file"`
}
type mgThing struct {
ThingID string `toml:"thing_id" mapstructure:"thing_id"`
ThingKey string `toml:"thing_key" mapstructure:"thing_key"`
MTLSCert string `toml:"mtls_cert" mapstructure:"mtls_cert"`
MTLSKey string `toml:"mtls_key" mapstructure:"mtls_key"`
type mgClient struct {
ClientID string `toml:"client_id" mapstructure:"client_id"`
ClientSecret string `toml:"client_secret" mapstructure:"client_secret"`
MTLSCert string `toml:"mtls_cert" mapstructure:"mtls_cert"`
MTLSKey string `toml:"mtls_key" mapstructure:"mtls_key"`
}
type mgChannel struct {
@@ -55,7 +55,7 @@ type mgChannel struct {
}
type magistrala struct {
Things []mgThing `toml:"things" mapstructure:"things"`
Clients []mgClient `toml:"clients" mapstructure:"clients"`
Channels []mgChannel `toml:"channels" mapstructure:"channels"`
}
tools/provision/README.md
+24 -22
View File
@@ -1,61 +1,63 @@
# Magistrala Things and Channels Provisioning Tool
# Magistrala Clients and Channels Provisioning Tool
A simple utility to create a list of channels and things connected to these channels with possibility to create certificates for mTLS use case.
A simple utility to create a list of channels and clients connected to these channels with possibility to create certificates for mTLS use case.
This tool is useful for testing, and it creates a TOML format output (on stdout, can be redirected into the file as needed)
that can be used by Magistrala MQTT benchmarking tool (`mqtt-bench`).
## Installation
```
```bash
cd tools/provision
make
```
### Usage
```
```bash
./provision --help
Tool for provisioning series of Magistrala channels and things and connecting them together.
Tool for provisioning series of Magistrala channels and clients and connecting them together.
Complete documentation is available at https://docs.magistrala.abstractmachines.fr
Usage:
provision [flags]
Flags:
--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")
--ca string CA for creating and signing clients certificate (default "ca.crt")
--cakey string ca.key for creating and signing clients certificate (default "ca.key")
-h, --help help for provision
--host string address for magistrala instance (default "https://localhost")
--num int number of channels and things to create and connect (default 10)
--num int number of channels and clients to create and connect (default 10)
-p, --password string magistrala users password
--ssl create certificates for mTLS access
-u, --username string magistrala user
--prefix string name prefix for things and channels
--prefix string name prefix for clients and channels
```
Example:
```
```bash
go run tools/provision/cmd/main.go -u test@magistrala.com -p test1234 --host https://142.93.118.47
```
If you want to create a list of channels with certificates:
```
```bash
go run tools/provision/cmd/main.go --host http://localhost --num 10 -u test@magistrala.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,
> `ca.crt` and `ca.key` are used for creating clients certificate and for HTTPS,
> if you are provisioning on remote server you will have to get these files to your local
> directory so that you can create certificates for things
> directory so that you can create certificates for clients
Example of output:
```
# List of things that can be connected to MQTT broker
[[things]]
thing_id = "0eac601b-6d54-4767-b8b7-594aaf9990d3"
thing_key = "07713103-513f-43c7-b7fe-500c1af23d7d"
```bash
# List of clients that can be connected to MQTT broker
[[clients]]
client_id = "0eac601b-6d54-4767-b8b7-594aaf9990d3"
client_key = "07713103-513f-43c7-b7fe-500c1af23d7d"
mtls_cert = """-----BEGIN CERTIFICATE-----
MIIEmTCCA4GgAwIBAgIRAO50qOfXsU+cHm/QY2NYu+0wDQYJKoZIhvcNAQELBQAw
VzESMBAGA1UEAwwJbG9jYWxob3N0MREwDwYDVQQKDAhNYWluZmx1eDEMMAoGA1UE
@@ -137,9 +139,9 @@ uCRt+TFMyEfqilipmNsV7esgbroiyEGXGMI8JdBY9OsnK6ZSlXaMnQ9vq2kK
-----END RSA PRIVATE KEY-----
"""
# List of channels that things can publish to
# each channel is connected to each thing from things list
# Things connected to channel 1f18afa1-29c4-4634-99d1-68dfa1b74e6a: 0eac601b-6d54-4767-b8b7-594aaf9990d3
# List of channels that clients can publish to
# each channel is connected to each client from clients list
# Clients connected to channel 1f18afa1-29c4-4634-99d1-68dfa1b74e6a: 0eac601b-6d54-4767-b8b7-594aaf9990d3
[[channels]]
channel_id = "1f18afa1-29c4-4634-99d1-68dfa1b74e6a"
tools/provision/cmd/main.go
+5 -5
View File
@@ -17,7 +17,7 @@ func main() {
rootCmd := &cobra.Command{
Use: "provision",
Short: "provision is provisioning tool for Magistrala",
Long: `Tool for provisioning series of Magistrala channels and things and connecting them together.
Long: `Tool for provisioning series of Magistrala channels and clients and connecting them together.
Complete documentation is available at https://docs.magistrala.abstractmachines.fr`,
Run: func(_ *cobra.Command, _ []string) {
if err := provision.Provision(pconf); err != nil {
@@ -28,13 +28,13 @@ Complete documentation is available at https://docs.magistrala.abstractmachines.
// Root Flags
rootCmd.PersistentFlags().StringVarP(&pconf.Host, "host", "", "https://localhost", "address for magistrala instance")
rootCmd.PersistentFlags().StringVarP(&pconf.Prefix, "prefix", "", "", "name prefix for things and channels")
rootCmd.PersistentFlags().StringVarP(&pconf.Prefix, "prefix", "", "", "name prefix for clients and channels")
rootCmd.PersistentFlags().StringVarP(&pconf.Username, "username", "u", "", "magistrala user")
rootCmd.PersistentFlags().StringVarP(&pconf.Password, "password", "p", "", "magistrala users password")
rootCmd.PersistentFlags().IntVarP(&pconf.Num, "num", "", 10, "number of channels and things to create and connect")
rootCmd.PersistentFlags().IntVarP(&pconf.Num, "num", "", 10, "number of channels and clients to create and connect")
rootCmd.PersistentFlags().BoolVarP(&pconf.SSL, "ssl", "", false, "create certificates for mTLS access")
rootCmd.PersistentFlags().StringVarP(&pconf.CAKey, "cakey", "", "ca.key", "ca.key for creating and signing things certificate")
rootCmd.PersistentFlags().StringVarP(&pconf.CA, "ca", "", "ca.crt", "CA for creating and signing things certificate")
rootCmd.PersistentFlags().StringVarP(&pconf.CAKey, "cakey", "", "ca.key", "ca.key for creating and signing clients certificate")
rootCmd.PersistentFlags().StringVarP(&pconf.CA, "ca", "", "ca.crt", "CA for creating and signing clients certificate")
if err := rootCmd.Execute(); err != nil {
log.Fatal(err)
tools/provision/doc.go
+1 -1
View File
@@ -2,6 +2,6 @@
// SPDX-License-Identifier: Apache-2.0
// Package provision is a simple utility to create
// a list of channels and things connected to these channels
// a list of channels and clients connected to these channels
// with possibility to create certificates for mTLS use case.
package provision
tools/provision/provision.go
+21 -20
View File
@@ -33,11 +33,11 @@ var namesgenerator = namegenerator.NewGenerator()
// MgConn - structure describing Magistrala connection set.
type MgConn struct {
ChannelID string
ThingID string
ThingKey string
MTLSCert string
MTLSKey string
ClientID string
ClinetSecret string
ChannelID string
MTLSCert string
MTLSKey string
}
// Config - provisioning configuration.
@@ -62,7 +62,7 @@ func Provision(conf Config) error {
msgContentType := string(sdk.CTJSONSenML)
sdkConf := sdk.Config{
ThingsURL: conf.Host,
ClientsURL: conf.Host,
UsersURL: conf.Host,
ReaderURL: defReaderURL,
HTTPAdapterURL: fmt.Sprintf("%s/http", conf.Host),
@@ -146,22 +146,22 @@ func Provision(conf Config) error {
}
}
// Create things and channels
things := make([]sdk.Thing, conf.Num)
// Create clients and channels
clients := make([]sdk.Client, conf.Num)
channels := make([]sdk.Channel, conf.Num)
cIDs := []string{}
tIDs := []string{}
fmt.Println("# List of things that can be connected to MQTT broker")
fmt.Println("# List of clients that can be connected to MQTT broker")
for i := 0; i < conf.Num; i++ {
things[i] = sdk.Thing{Name: fmt.Sprintf("%s-thing-%d", conf.Prefix, i)}
clients[i] = sdk.Client{Name: fmt.Sprintf("%s-client-%d", conf.Prefix, i)}
channels[i] = sdk.Channel{Name: fmt.Sprintf("%s-channel-%d", conf.Prefix, i)}
}
things, err = s.CreateThings(things, domain.ID, token.AccessToken)
clients, err = s.CreateClients(clients, domain.ID, token.AccessToken)
if err != nil {
return fmt.Errorf("failed to create the things: %s", err.Error())
return fmt.Errorf("failed to create the clients: %s", err.Error())
}
var chs []sdk.Channel
@@ -174,7 +174,7 @@ func Provision(conf Config) error {
}
channels = chs
for _, t := range things {
for _, t := range clients {
tIDs = append(tIDs, t.ID)
}
@@ -207,7 +207,7 @@ func Provision(conf Config) error {
SerialNumber: serialNumber,
Subject: pkix.Name{
Organization: []string{"Magistrala"},
CommonName: things[i].Credentials.Secret,
CommonName: clients[i].Credentials.Secret,
OrganizationalUnit: []string{"magistrala"},
},
NotBefore: notBefore,
@@ -241,7 +241,7 @@ func Provision(conf Config) error {
}
// Print output
fmt.Printf("[[things]]\nthing_id = \"%s\"\nthing_key = \"%s\"\n", things[i].ID, things[i].Credentials.Secret)
fmt.Printf("[[clients]]\nclient_id = \"%s\"\nclient_key = \"%s\"\n", clients[i].ID, clients[i].Credentials.Secret)
if conf.SSL {
fmt.Printf("mtls_cert = \"\"\"%s\"\"\"\n", cert)
fmt.Printf("mtls_key = \"\"\"%s\"\"\"\n", key)
@@ -249,8 +249,8 @@ func Provision(conf Config) error {
fmt.Println("")
}
fmt.Printf("# List of channels that things can publish to\n" +
"# each channel is connected to each thing from things list\n")
fmt.Printf("# List of channels that clients can publish to\n" +
"# each channel is connected to each client from clients list\n")
for i := 0; i < conf.Num; i++ {
fmt.Printf("[[channels]]\nchannel_id = \"%s\"\n\n", cIDs[i])
}
@@ -258,11 +258,12 @@ func Provision(conf Config) error {
for _, cID := range cIDs {
for _, tID := range tIDs {
conIDs := sdk.Connection{
ThingID: tID,
ChannelID: cID,
ClientIDs: []string{tID},
ChannelIDs: []string{cID},
Types: []string{"publish", "subscribe"},
}
if err := s.Connect(conIDs, domain.ID, token.AccessToken); err != nil {
log.Fatalf("Failed to connect things %s to channels %s: %s", tID, cID, err)
log.Fatalf("Failed to connect clients %s to channels %s: %s", tID, cID, err)
}
}
}