netmaker/mq/emqx_cloud.go
Abhishek K 66069fbc34
NET-1082: Scale Testing Fixes (#2894)
* add additional mutex lock on node acls func

* increase verbosity

* disable acls on cloud emqx

* add emqx creds creation to go routine

* add debug log of mq client id

* comment port check

* uncomment port check

* check for connection mq connection open

* use username for client id

* add write mutex on acl is allowed

* add mq connection lost handler on server

* spin off zombie init as go routine

* get whole api path from config

* Revert "get whole api path from config"

This reverts commit 392f5f4c5f.

* update extclient acls async

* add additional mutex lock on node acls func

(cherry picked from commit 5325f0e7d7)

* increase verbosity

(cherry picked from commit 705b3cf0bf)

* add emqx creds creation to go routine

(cherry picked from commit c8e65f4820)

* add debug log of mq client id

(cherry picked from commit 29c5d6ceca)

* comment port check

(cherry picked from commit db8d6d95ea)

* check for connection mq connection open

(cherry picked from commit 13b11033b0)

* use username for client id

(cherry picked from commit e90c7386de)

* add write mutex on acl is allowed

(cherry picked from commit 4cae1b0bb4)

* add mq connection lost handler on server

(cherry picked from commit c82918ad35)

* spin off zombie init as go routine

(cherry picked from commit 6d65c44c43)

* update extclient acls async

(cherry picked from commit 6557ef1ebe)

* additionl logs for oauth user flow

(cherry picked from commit 61703038ae)

* add more debug logs

(cherry picked from commit 5980beacd1)

* add more debug logs

(cherry picked from commit 4d001f0d27)

* add set auth secret

(cherry picked from commit f41cef5da5)

* fix fetch pass

(cherry picked from commit 825caf4b60)

* make sure auth secret is set only once

(cherry picked from commit ba33ed02aa)

* make sure auth secret is set only once

(cherry picked from commit 920ac4c507)

* comment usage of emqx acls

* replace  read lock with write lock on acls

* replace  read lock with write lock on acls

(cherry picked from commit 808d2135c8)

* use deadlock pkg for visibility

* add additional mutex locks

* remove race flag

* on mq re-connecting donot exit if failed

* on mq re-connecting donot exit if failed

* revert mutex package change

* set mq clean session

* remove debug log

* go mod tidy

* revert on prem emqx acls del
2024-04-11 21:18:57 +05:30

129 lines
2.9 KiB
Go

package mq
import (
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"strings"
"github.com/gravitl/netmaker/servercfg"
)
type EmqxCloud struct {
URL string
AppID string
AppSecret string
}
type userCreateReq struct {
UserName string `json:"username"`
Password string `json:"password"`
}
func (e *EmqxCloud) GetType() servercfg.Emqxdeploy { return servercfg.EmqxCloudDeploy }
func (e *EmqxCloud) CreateEmqxUser(username, pass string) error {
payload := userCreateReq{
UserName: username,
Password: pass,
}
data, _ := json.Marshal(payload)
client := &http.Client{}
req, err := http.NewRequest(http.MethodPost, fmt.Sprintf("%s/api/auth_username", e.URL), strings.NewReader(string(data)))
if err != nil {
return err
}
req.SetBasicAuth(e.AppID, e.AppSecret)
req.Header.Add("Content-Type", "application/json")
res, err := client.Do(req)
if err != nil {
return err
}
defer res.Body.Close()
body, err := io.ReadAll(res.Body)
if err != nil {
return err
}
if res.StatusCode != http.StatusOK {
return errors.New("request failed " + string(body))
}
return nil
}
func (e *EmqxCloud) CreateEmqxUserforServer() error {
payload := userCreateReq{
UserName: servercfg.GetMqUserName(),
Password: servercfg.GetMqPassword(),
}
data, _ := json.Marshal(payload)
client := &http.Client{}
req, err := http.NewRequest(http.MethodPost, fmt.Sprintf("%s/api/auth_username", e.URL), strings.NewReader(string(data)))
if err != nil {
return err
}
req.SetBasicAuth(e.AppID, e.AppSecret)
req.Header.Add("Content-Type", "application/json")
res, err := client.Do(req)
if err != nil {
return err
}
defer res.Body.Close()
body, err := io.ReadAll(res.Body)
if err != nil {
return err
}
if res.StatusCode != http.StatusOK {
return errors.New("request failed " + string(body))
}
return nil
}
func (e *EmqxCloud) CreateEmqxDefaultAuthenticator() error { return nil } // ignore
func (e *EmqxCloud) CreateEmqxDefaultAuthorizer() error { return nil } // ignore
func (e *EmqxCloud) CreateDefaultDenyRule() error {
return nil
}
func (e *EmqxCloud) CreateHostACL(hostID, serverName string) error {
return nil
}
func (e *EmqxCloud) AppendNodeUpdateACL(hostID, nodeNetwork, nodeID, serverName string) error {
return nil
}
func (e *EmqxCloud) GetUserACL(username string) (*aclObject, error) { return nil, nil } // ununsed on cloud since it doesn't overwrite acls list
func (e *EmqxCloud) DeleteEmqxUser(username string) error {
client := &http.Client{}
req, err := http.NewRequest(http.MethodDelete, fmt.Sprintf("%s/api/auth_username/%s", e.URL, username), nil)
if err != nil {
return err
}
req.SetBasicAuth(e.AppID, e.AppSecret)
res, err := client.Do(req)
if err != nil {
return err
}
defer res.Body.Close()
body, err := io.ReadAll(res.Body)
if err != nil {
return err
}
if res.StatusCode != http.StatusOK {
return errors.New("request failed " + string(body))
}
return nil
}