mirror of
https://github.com/gravitl/netmaker.git
synced 2024-09-20 15:26:04 +08:00
5a561b3835
* NET-1440 scale test changes * fix UT error and add error info * load metric data into cacha in startup * remove debug info for metric * add server telemetry and hasSuperAdmin to cache * fix user UT case * update sqlite connection string for performance * update check-in TS in cache only if cache enabled * update metric data in cache only if cache enabled and write to DB once in stop * update server status in mq topic * add failover existed to server status update * only send mq messsage when there is server status change * batch peerUpdate * code changes for scale for review * update UT case * update mq client check * mq connection code change * revert server status update changes * revert batch peerUpdate * remove server status update info * batch peerUpdate * code changes based on review and setupmqtt in keepalive * set the mq message order to false for PIN * remove setupmqtt in keepalive * add peerUpdate batch size to config * update batch peerUpdate * recycle ip in node deletion * update ip allocation logic * remove ip addr cap * remove ippool file * update get extClient func * remove ip from cache map when extClient is removed * add batch peerUpdate switch * set batch peerUpdate to true by default --------- Co-authored-by: Max Ma <mayabin@gmail.com>
130 lines
3.2 KiB
Go
130 lines
3.2 KiB
Go
package mq
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"math"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/gravitl/netmaker/logic"
|
|
"github.com/gravitl/netmaker/models"
|
|
"github.com/gravitl/netmaker/netclient/ncutils"
|
|
"golang.org/x/exp/slog"
|
|
)
|
|
|
|
func decryptMsgWithHost(host *models.Host, msg []byte) ([]byte, error) {
|
|
if host.OS == models.OS_Types.IoT { // just pass along IoT messages
|
|
return msg, nil
|
|
}
|
|
|
|
trafficKey, trafficErr := logic.RetrievePrivateTrafficKey() // get server private key
|
|
if trafficErr != nil {
|
|
return nil, trafficErr
|
|
}
|
|
serverPrivTKey, err := ncutils.ConvertBytesToKey(trafficKey)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
nodePubTKey, err := ncutils.ConvertBytesToKey(host.TrafficKeyPublic)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return ncutils.DeChunk(msg, nodePubTKey, serverPrivTKey)
|
|
}
|
|
|
|
func DecryptMsg(node *models.Node, msg []byte) ([]byte, error) {
|
|
if len(msg) <= 24 { // make sure message is of appropriate length
|
|
return nil, fmt.Errorf("received invalid message from broker %v", msg)
|
|
}
|
|
host, err := logic.GetHost(node.HostID.String())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return decryptMsgWithHost(host, msg)
|
|
}
|
|
|
|
func BatchItems[T any](items []T, batchSize int) [][]T {
|
|
if batchSize <= 0 {
|
|
return nil
|
|
}
|
|
remainderBatchSize := len(items) % batchSize
|
|
nBatches := int(math.Ceil(float64(len(items)) / float64(batchSize)))
|
|
batches := make([][]T, nBatches)
|
|
for i := range batches {
|
|
if i == nBatches-1 && remainderBatchSize > 0 {
|
|
batches[i] = make([]T, remainderBatchSize)
|
|
} else {
|
|
batches[i] = make([]T, batchSize)
|
|
}
|
|
for j := range batches[i] {
|
|
batches[i][j] = items[i*batchSize+j]
|
|
}
|
|
}
|
|
return batches
|
|
}
|
|
|
|
func encryptMsg(host *models.Host, msg []byte) ([]byte, error) {
|
|
if host.OS == models.OS_Types.IoT {
|
|
return msg, nil
|
|
}
|
|
|
|
// fetch server public key to be certain hasn't changed in transit
|
|
trafficKey, trafficErr := logic.RetrievePrivateTrafficKey()
|
|
if trafficErr != nil {
|
|
return nil, trafficErr
|
|
}
|
|
|
|
serverPrivKey, err := ncutils.ConvertBytesToKey(trafficKey)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
nodePubKey, err := ncutils.ConvertBytesToKey(host.TrafficKeyPublic)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if strings.Contains(host.Version, "0.10.0") {
|
|
return ncutils.BoxEncrypt(msg, nodePubKey, serverPrivKey)
|
|
}
|
|
|
|
return ncutils.Chunk(msg, nodePubKey, serverPrivKey)
|
|
}
|
|
|
|
func publish(host *models.Host, dest string, msg []byte) error {
|
|
|
|
encrypted, encryptErr := encryptMsg(host, msg)
|
|
if encryptErr != nil {
|
|
return encryptErr
|
|
}
|
|
if mqclient == nil || !mqclient.IsConnectionOpen() {
|
|
return errors.New("cannot publish ... mqclient not connected")
|
|
}
|
|
|
|
if token := mqclient.Publish(dest, 0, true, encrypted); !token.WaitTimeout(MQ_TIMEOUT*time.Second) || token.Error() != nil {
|
|
var err error
|
|
if token.Error() == nil {
|
|
err = errors.New("connection timeout")
|
|
} else {
|
|
slog.Error("publish to mq error", "error", token.Error().Error())
|
|
err = token.Error()
|
|
}
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// decodes a message queue topic and returns the embedded node.ID
|
|
func GetID(topic string) (string, error) {
|
|
parts := strings.Split(topic, "/")
|
|
count := len(parts)
|
|
if count == 1 {
|
|
return "", fmt.Errorf("invalid topic")
|
|
}
|
|
//the last part of the topic will be the node.ID
|
|
return parts[count-1], nil
|
|
}
|