2023-10-10 21:14:40 +02:00
|
|
|
package sync
|
|
|
|
|
|
|
|
import (
|
|
|
|
"context"
|
|
|
|
"errors"
|
2023-10-23 19:13:08 +02:00
|
|
|
"io"
|
2023-10-10 21:14:40 +02:00
|
|
|
"time"
|
|
|
|
|
|
|
|
"github.com/tim-beatham/wgmesh/pkg/ctrlserver"
|
2023-10-20 13:41:06 +02:00
|
|
|
logging "github.com/tim-beatham/wgmesh/pkg/log"
|
2023-10-26 17:53:12 +02:00
|
|
|
"github.com/tim-beatham/wgmesh/pkg/mesh"
|
2023-10-10 21:14:40 +02:00
|
|
|
"github.com/tim-beatham/wgmesh/pkg/rpc"
|
|
|
|
)
|
|
|
|
|
|
|
|
// SyncRequester: coordinates the syncing of meshes
|
|
|
|
type SyncRequester interface {
|
2023-10-24 17:00:46 +02:00
|
|
|
GetMesh(meshId string, ifName string, port int, endPoint string) error
|
2023-12-07 02:44:54 +01:00
|
|
|
SyncMesh(meshid string, meshNode mesh.MeshNode) error
|
2023-10-10 21:14:40 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
type SyncRequesterImpl struct {
|
2023-10-20 18:35:02 +02:00
|
|
|
server *ctrlserver.MeshCtrlServer
|
|
|
|
errorHdlr SyncErrorHandler
|
2023-10-10 21:14:40 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// GetMesh: Retrieves the local state of the mesh at the endpoint
|
2023-10-24 17:00:46 +02:00
|
|
|
func (s *SyncRequesterImpl) GetMesh(meshId string, ifName string, port int, endPoint string) error {
|
2023-10-10 21:14:40 +02:00
|
|
|
peerConnection, err := s.server.ConnectionManager.GetConnection(endPoint)
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
client, err := peerConnection.GetClient()
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
c := rpc.NewSyncServiceClient(client)
|
|
|
|
|
2023-10-24 01:12:38 +02:00
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
|
2023-10-10 21:14:40 +02:00
|
|
|
defer cancel()
|
|
|
|
|
|
|
|
reply, err := c.GetConf(ctx, &rpc.GetConfRequest{MeshId: meshId})
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2023-10-28 17:38:25 +02:00
|
|
|
err = s.server.MeshManager.AddMesh(&mesh.AddMeshParams{
|
|
|
|
MeshId: meshId,
|
|
|
|
WgPort: port,
|
|
|
|
MeshBytes: reply.Mesh,
|
|
|
|
})
|
2023-10-10 21:14:40 +02:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2023-12-07 02:44:54 +01:00
|
|
|
func (s *SyncRequesterImpl) handleErr(meshId, pubKey string, err error) error {
|
|
|
|
ok := s.errorHdlr.Handle(meshId, pubKey, err)
|
2023-10-20 18:35:02 +02:00
|
|
|
|
|
|
|
if ok {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
2023-10-10 21:14:40 +02:00
|
|
|
// SyncMesh: Proactively send a sync request to the other mesh
|
2023-12-07 02:44:54 +01:00
|
|
|
func (s *SyncRequesterImpl) SyncMesh(meshId string, meshNode mesh.MeshNode) error {
|
|
|
|
endpoint := meshNode.GetHostEndpoint()
|
|
|
|
pubKey, _ := meshNode.GetPublicKey()
|
|
|
|
|
2023-10-10 21:14:40 +02:00
|
|
|
peerConnection, err := s.server.ConnectionManager.GetConnection(endpoint)
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
client, err := peerConnection.GetClient()
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
mesh := s.server.MeshManager.GetMesh(meshId)
|
|
|
|
|
|
|
|
if mesh == nil {
|
|
|
|
return errors.New("mesh does not exist")
|
|
|
|
}
|
|
|
|
|
|
|
|
c := rpc.NewSyncServiceClient(client)
|
|
|
|
|
2023-12-10 20:21:54 +01:00
|
|
|
syncTimeOut := float64(s.server.Conf.SyncRate) * float64(time.Second)
|
2023-11-06 10:54:06 +01:00
|
|
|
|
|
|
|
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(syncTimeOut))
|
2023-10-10 21:14:40 +02:00
|
|
|
defer cancel()
|
|
|
|
|
2023-11-06 10:54:06 +01:00
|
|
|
err = s.syncMesh(mesh, ctx, c)
|
2023-10-10 21:14:40 +02:00
|
|
|
|
|
|
|
if err != nil {
|
2023-12-07 02:44:54 +01:00
|
|
|
return s.handleErr(meshId, pubKey.String(), err)
|
2023-10-10 21:14:40 +02:00
|
|
|
}
|
|
|
|
|
2023-10-24 01:12:38 +02:00
|
|
|
logging.Log.WriteInfof("Synced with node: %s meshId: %s\n", endpoint, meshId)
|
2023-10-10 21:14:40 +02:00
|
|
|
return nil
|
|
|
|
}
|
2023-10-20 13:41:06 +02:00
|
|
|
|
2023-11-06 10:54:06 +01:00
|
|
|
func (s *SyncRequesterImpl) syncMesh(mesh mesh.MeshProvider, ctx context.Context, client rpc.SyncServiceClient) error {
|
2023-10-23 19:13:08 +02:00
|
|
|
stream, err := client.SyncMesh(ctx)
|
|
|
|
|
|
|
|
syncer := mesh.GetSyncer()
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
for {
|
|
|
|
msg, moreMessages := syncer.GenerateMessage()
|
|
|
|
|
2023-10-26 17:53:12 +02:00
|
|
|
err := stream.Send(&rpc.SyncMeshRequest{MeshId: mesh.GetMeshId(), Changes: msg})
|
2023-10-23 19:13:08 +02:00
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
in, err := stream.Recv()
|
|
|
|
|
|
|
|
if err != nil && err != io.EOF {
|
2023-10-24 01:12:38 +02:00
|
|
|
logging.Log.WriteInfof("Stream recv error: %s\n", err.Error())
|
2023-10-23 19:13:08 +02:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
if err != io.EOF && len(in.Changes) != 0 {
|
|
|
|
err = syncer.RecvMessage(in.Changes)
|
|
|
|
}
|
|
|
|
|
|
|
|
if err != nil {
|
2023-10-24 01:12:38 +02:00
|
|
|
logging.Log.WriteInfof("Syncer recv error: %s\n", err.Error())
|
2023-10-23 19:13:08 +02:00
|
|
|
return err
|
|
|
|
}
|
|
|
|
|
|
|
|
if !moreMessages {
|
|
|
|
break
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2023-10-24 17:00:46 +02:00
|
|
|
syncer.Complete()
|
2023-10-23 19:13:08 +02:00
|
|
|
stream.CloseSend()
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
2023-10-20 13:41:06 +02:00
|
|
|
func NewSyncRequester(s *ctrlserver.MeshCtrlServer) SyncRequester {
|
2023-12-29 12:12:40 +01:00
|
|
|
errorHdlr := NewSyncErrorHandler(s.MeshManager, s.ConnectionManager)
|
2023-10-20 18:35:02 +02:00
|
|
|
return &SyncRequesterImpl{server: s, errorHdlr: errorHdlr}
|
2023-10-20 13:41:06 +02:00
|
|
|
}
|