1
0
forked from extern/smegmesh

Compare commits

..

8 Commits

Author SHA1 Message Date
8aab4e99d8 59-error-when-peer-not-selected
In the CLI when the peer is not selected
as the type throwing an error stating
either client or peer must be selected
2023-12-22 19:08:20 +00:00
cf4be1ccab Merge pull request #58 from tim-beatham/bugfix-pull-only
Bugfix pull only
2023-12-22 18:49:09 +00:00
6ed32f3a79 bugfix-push-pull
Organised groups as a tree so that there
isn't a limit to dissemination
2023-12-19 00:50:17 +00:00
b6199892f0 bugfix-pull-only
Bugfix with inter-cluster communication pull not working
2023-12-18 22:17:46 +00:00
ad22f04b0d bugfix-pull-only
After certain period of time if no changes have
occurred then pull
2023-12-18 20:45:56 +00:00
092d9a4af5 checking-latency-for-pull-only 2023-12-17 09:44:32 +00:00
19abf712a6 Fixing bug with nodes being removed 2023-12-12 12:45:41 +00:00
b296e1f45a Merge pull request #57 from tim-beatham/55-cli-option-for-peer-type
55-cli-optionifor-peer-type
2023-12-12 12:00:42 +00:00
7 changed files with 85 additions and 21 deletions

View File

@ -15,7 +15,6 @@ const SockAddr = "/tmp/wgmesh_ipc.sock"
type CreateMeshParams struct {
Client *ipcRpc.Client
Endpoint string
Role string
WgArgs ipc.WireGuardArgs
AdvertiseRoutes bool
AdvertiseDefault bool
@ -56,7 +55,6 @@ type JoinMeshParams struct {
MeshId string
IpAddress string
Endpoint string
Role string
WgArgs ipc.WireGuardArgs
AdvertiseRoutes bool
AdvertiseDefault bool
@ -203,6 +201,7 @@ func main() {
})
var newMeshRole *string = newMeshCmd.Selector("r", "role", []string{"peer", "client"}, &argparse.Options{
Default: "peer",
Help: "Role in the mesh network. A value of peer means that the node is publicly routeable and thus considered" +
" in the gossip protocol. Client means that the node is not publicly routeable and is not a candidate in the gossip" +
" protocol",
@ -235,7 +234,7 @@ func main() {
})
var joinMeshRole *string = joinMeshCmd.Selector("r", "role", []string{"peer", "client"}, &argparse.Options{
Default: "Peer",
Default: "peer",
Help: "Role in the mesh network. A value of peer means that the node is publicly routeable and thus considered" +
" in the gossip protocol. Client means that the node is not publicly routeable and is not a candidate in the gossip" +
" protocol",
@ -319,7 +318,6 @@ func main() {
fmt.Println(createMesh(&CreateMeshParams{
Client: client,
Endpoint: *newMeshEndpoint,
Role: *newMeshRole,
WgArgs: ipc.WireGuardArgs{
Endpoint: *newMeshEndpoint,
Role: *newMeshRole,
@ -341,7 +339,6 @@ func main() {
IpAddress: *joinMeshIpAddress,
MeshId: *joinMeshId,
Endpoint: *joinMeshEndpoint,
Role: *joinMeshRole,
WgArgs: ipc.WireGuardArgs{
Endpoint: *joinMeshEndpoint,
Role: *joinMeshRole,

14
go.mod
View File

@ -4,14 +4,18 @@ go 1.21.3
require (
github.com/akamensky/argparse v1.4.0
github.com/anandvarma/namegen v0.0.0-20230727084436-5197c6ea3255
github.com/automerge/automerge-go v0.0.0-20230903201930-b80ce8aadbb9
github.com/gin-gonic/gin v1.9.1
github.com/go-playground/validator/v10 v10.16.0
github.com/google/uuid v1.3.0
github.com/jmespath/go-jmespath v0.4.0
github.com/jsimonetti/rtnetlink v1.3.5
github.com/miekg/dns v1.1.57
github.com/sirupsen/logrus v1.9.3
golang.org/x/sys v0.14.0
golang.zx2c4.com/wireguard/wgctrl v0.0.0-20230429144221-925a1e7659e6
gonum.org/v1/gonum v0.14.0
google.golang.org/grpc v1.58.1
google.golang.org/protobuf v1.31.0
gopkg.in/yaml.v3 v3.0.1
@ -24,7 +28,6 @@ require (
github.com/gin-contrib/sse v0.1.0 // indirect
github.com/go-playground/locales v0.14.1 // indirect
github.com/go-playground/universal-translator v0.18.1 // indirect
github.com/go-playground/validator/v10 v10.14.0 // indirect
github.com/goccy/go-json v0.10.2 // indirect
github.com/golang/protobuf v1.5.3 // indirect
github.com/google/go-cmp v0.5.9 // indirect
@ -42,10 +45,13 @@ require (
github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
github.com/ugorji/go/codec v1.2.11 // indirect
golang.org/x/arch v0.3.0 // indirect
golang.org/x/crypto v0.13.0 // indirect
golang.org/x/net v0.15.0 // indirect
golang.org/x/sync v0.3.0 // indirect
golang.org/x/crypto v0.14.0 // indirect
golang.org/x/exp v0.0.0-20230321023759-10a507213a29 // indirect
golang.org/x/mod v0.12.0 // indirect
golang.org/x/net v0.17.0 // indirect
golang.org/x/sync v0.4.0 // indirect
golang.org/x/text v0.13.0 // indirect
golang.org/x/tools v0.13.0 // indirect
golang.zx2c4.com/wireguard v0.0.0-20230704135630-469159ecf7d1 // indirect
google.golang.org/genproto/googleapis/rpc v0.0.0-20230711160842-782d3b101e98 // indirect
)

2
go.sum
View File

@ -29,6 +29,8 @@ github.com/go-playground/universal-translator v0.18.1 h1:Bcnm0ZwsGyWbCzImXv+pAJn
github.com/go-playground/universal-translator v0.18.1/go.mod h1:xekY+UJKNuX9WP91TpwSH2VMlDf28Uj24BCp08ZFTUY=
github.com/go-playground/validator/v10 v10.14.0 h1:vgvQWe3XCz3gIeFDm/HnTIbj6UGmg/+t63MyGU2n5js=
github.com/go-playground/validator/v10 v10.14.0/go.mod h1:9iXMNT7sEkjXb0I+enO7QXmzG6QCsPWY4zveKFVRSyU=
github.com/go-playground/validator/v10 v10.16.0 h1:x+plE831WK4vaKHO/jpgUGsvLKIqRRkz6M78GuJAfGE=
github.com/go-playground/validator/v10 v10.16.0/go.mod h1:9iXMNT7sEkjXb0I+enO7QXmzG6QCsPWY4zveKFVRSyU=
github.com/goccy/go-json v0.10.2 h1:CrxCmQqYDkv1z7lO7Wbh2HN93uovUHgrECaO5ZrCXAU=
github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MGFi0w8I=
github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=

View File

@ -55,12 +55,14 @@ func (i *ConnClusterImpl) GetNeighbours(global []string, selfId string) []string
// you will communicate with a random node that is not in your cluster.
func (i *ConnClusterImpl) GetInterCluster(global []string, selfId string) string {
// Doesn't matter if not in it. Get index of where the node 'should' be
slices.Sort(global)
index, _ := binarySearch(global, selfId, 1)
numClusters := math.Ceil(float64(len(global)) / float64(i.clusterSize))
randomCluster := rand.Intn(int(numClusters)-1) + 1
randomCluster := rand.Intn(2) + 1
neighbourIndex := (index + randomCluster) % len(global)
// cluster is considered a heap
neighbourIndex := (2*index + (randomCluster * i.clusterSize)) % len(global)
return global[neighbourIndex]
}

View File

@ -386,7 +386,7 @@ func (m *TwoPhaseStoreMeshManager) SetAlias(nodeId string, alias string) error {
}
node := m.store.Get(nodeId)
node.Description = alias
node.Alias = alias
m.store.Put(nodeId, node)
return nil

View File

@ -68,9 +68,16 @@ func prepare(syncer *TwoPhaseSyncer) ([]byte, bool) {
return nil, false
}
// Increment the clock here so the clock gets
// distributed to everyone else in the mesh
syncer.manager.store.Clock.IncrementClock()
var buffer bytes.Buffer
enc := gob.NewEncoder(&buffer)
mapState := syncer.manager.store.GenerateMessage()
syncer.mapState = mapState
err = enc.Encode(*syncer.mapState)
if err != nil {
@ -94,7 +101,7 @@ func present(syncer *TwoPhaseSyncer) ([]byte, bool) {
if err != nil {
logging.Log.WriteErrorf(err.Error())
}
}
difference := syncer.mapState.Difference(syncer.manager.store.Clock.GetStaleCount(), &mapState)
syncer.manager.store.Clock.Merge(mapState.Vectors)
@ -164,9 +171,6 @@ func (t *TwoPhaseSyncer) RecvMessage(msg []byte) error {
func (t *TwoPhaseSyncer) Complete() {
logging.Log.WriteInfof("SYNC COMPLETED")
if t.state >= MERGE {
t.manager.store.Clock.IncrementClock()
}
}
func NewTwoPhaseSyncer(manager *TwoPhaseStoreMeshManager) *TwoPhaseSyncer {
@ -181,7 +185,6 @@ func NewTwoPhaseSyncer(manager *TwoPhaseStoreMeshManager) *TwoPhaseSyncer {
return &TwoPhaseSyncer{
manager: manager,
state: HASH,
mapState: manager.store.GenerateMessage(),
generateMessageFSM: generateMessageFsm,
}
}

View File

@ -1,6 +1,8 @@
package sync
import (
"errors"
"fmt"
"io"
"math/rand"
"time"
@ -25,7 +27,7 @@ type SyncerImpl struct {
syncCount int
cluster conn.ConnCluster
conf *conf.DaemonConfiguration
lastSync uint64
lastSync map[string]uint64
}
// Sync: Sync random nodes
@ -39,6 +41,12 @@ func (s *SyncerImpl) Sync(meshId string) error {
if self != nil && self.GetType() == conf.PEER_ROLE && !s.manager.HasChanges(meshId) && s.infectionCount == 0 {
logging.Log.WriteInfof("No changes for %s", meshId)
// If not synchronised in certain pull from random neighbour
if uint64(time.Now().Unix())-s.lastSync[meshId] > 20 {
return s.Pull(meshId)
}
return nil
}
@ -110,7 +118,7 @@ func (s *SyncerImpl) Sync(meshId string) error {
}
s.manager.GetMesh(meshId).SaveChanges()
s.lastSync = uint64(time.Now().Unix())
s.lastSync[meshId] = uint64(time.Now().Unix())
logging.Log.WriteInfof("UPDATING WG CONF")
err := s.manager.ApplyConfig()
@ -122,6 +130,51 @@ func (s *SyncerImpl) Sync(meshId string) error {
return nil
}
// Pull one node in the cluster, if there has not been message dissemination
// in a certain period of time pull a random node within the cluster
func (s *SyncerImpl) Pull(meshId string) error {
mesh := s.manager.GetMesh(meshId)
self, err := s.manager.GetSelf(meshId)
if err != nil {
return err
}
pubKey, _ := self.GetPublicKey()
if mesh == nil {
return errors.New("mesh is nil, invalid operation")
}
peers := mesh.GetPeers()
neighbours := s.cluster.GetNeighbours(peers, pubKey.String())
neighbour := lib.RandomSubsetOfLength(neighbours, 1)
if len(neighbour) == 0 {
logging.Log.WriteInfof("no neighbours")
return nil
}
logging.Log.WriteInfof("PULLING from node %s", neighbour[0])
pullNode, err := mesh.GetNode(neighbour[0])
if err != nil || pullNode == nil {
return fmt.Errorf("node %s does not exist in the mesh", neighbour[0])
}
err = s.requester.SyncMesh(meshId, pullNode)
if err == nil || err == io.EOF {
s.lastSync[meshId] = uint64(time.Now().Unix())
} else {
return err
}
s.syncCount++
return nil
}
// SyncMeshes: Sync all meshes
func (s *SyncerImpl) SyncMeshes() error {
for meshId := range s.manager.GetMeshes() {
@ -143,5 +196,6 @@ func NewSyncer(m mesh.MeshManager, conf *conf.DaemonConfiguration, r SyncRequest
requester: r,
infectionCount: 0,
syncCount: 0,
cluster: cluster}
cluster: cluster,
lastSync: make(map[string]uint64)}
}