From 2e80438c51b7381d882478ae01cba62f921bb87c Mon Sep 17 00:00:00 2001 From: Pascal Fischer Date: Tue, 10 Jun 2025 23:46:40 +0200 Subject: [PATCH] set up modules and controller --- .../controllers/network_map/controller.go | 35 +--- .../network_map/controller/controller.go | 74 +++++++ .../network_map/controller/metrics.go | 150 ++++++++++++++ .../network_map/controller/repository.go | 41 ++++ .../network_map/controller/updatechannel.go | 162 +++++++++++++++ internals/controllers/network_map/metrics.go | 31 --- .../controllers/network_map/network_map.go | 5 + .../{repository.go => network_map_data.go} | 36 ---- .../controllers/network_map/updatechannel.go | 185 +----------------- .../controllers/network_map/updatemessage.go | 18 ++ internals/server/controllers.go | 11 +- 11 files changed, 465 insertions(+), 283 deletions(-) create mode 100644 internals/controllers/network_map/controller/controller.go create mode 100644 internals/controllers/network_map/controller/metrics.go create mode 100644 internals/controllers/network_map/controller/repository.go create mode 100644 internals/controllers/network_map/controller/updatechannel.go delete mode 100644 internals/controllers/network_map/metrics.go create mode 100644 internals/controllers/network_map/network_map.go rename internals/controllers/network_map/{repository.go => network_map_data.go} (62%) create mode 100644 internals/controllers/network_map/updatemessage.go diff --git a/internals/controllers/network_map/controller.go b/internals/controllers/network_map/controller.go index c2503b9..dea2d5e 100644 --- a/internals/controllers/network_map/controller.go +++ b/internals/controllers/network_map/controller.go @@ -1,35 +1,8 @@ package network_map -import ( - "fmt" +import "context" - log "github.com/sirupsen/logrus" - - "github.com/netbirdio/management-refactor/internals/shared/db" - appmetrics "github.com/netbirdio/management-refactor/internals/shared/metrics" -) - -type Controller struct { - repo Repository - metrics *metrics -} - -func NewController(store *db.Store, metrics *appmetrics.AppMetrics) *Controller { - cMetrics, err := appmetrics.RegisterMetrics(metrics, newMetrics) - if err != nil { - log.Fatalf("Failed to register app metrics: %v", err) - } - return &Controller{ - repo: newRepository(store, cMetrics), - metrics: cMetrics, - } -} - -func (c *Controller) UpdatePeers(accountID string) error { - _, err := c.repo.GetNetworkMapData(accountID) - if err != nil { - return fmt.Errorf("get network map data: %w", err) - } - - return nil +type Controller interface { + UpdatePeers(ctx context.Context, accountID string) error + CalculateNetworkMap(data *NetworkMapData) (*NetworkMap, error) } diff --git a/internals/controllers/network_map/controller/controller.go b/internals/controllers/network_map/controller/controller.go new file mode 100644 index 0000000..42b8072 --- /dev/null +++ b/internals/controllers/network_map/controller/controller.go @@ -0,0 +1,74 @@ +package controller + +import ( + "context" + "fmt" + "sync" + + nbpeer "github.com/netbirdio/netbird/management/server/peer" + log "github.com/sirupsen/logrus" + + "github.com/netbirdio/management-refactor/internals/controllers/network_map" + "github.com/netbirdio/management-refactor/internals/shared/db" + appmetrics "github.com/netbirdio/management-refactor/internals/shared/metrics" +) + +type Controller struct { + repo Repository + metrics *metrics + updateChannel *UpdateChannel +} + +func NewController(store *db.Store, metrics *appmetrics.AppMetrics) *Controller { + cMetrics, err := appmetrics.RegisterMetrics(metrics, newMetrics) + if err != nil { + log.Fatalf("Failed to register app metrics: %v", err) + } + return &Controller{ + repo: newRepository(store, cMetrics), + metrics: cMetrics, + updateChannel: NewUpdateChannel(cMetrics), + } +} + +func (c *Controller) CalculateNetworkMap(data *network_map.NetworkMapData) (*network_map.NetworkMap, error) { + // Do calc on data + + return &network_map.NetworkMap{}, nil +} + +func (c *Controller) UpdatePeers(ctx context.Context, accountID string) error { + data, err := c.repo.GetNetworkMapData(accountID) + if err != nil { + return fmt.Errorf("get network map data: %w", err) + } + + var wg sync.WaitGroup + semaphore := make(chan struct{}, 10) + + for _, peer := range data.Peers { + if !c.updateChannel.HasChannel(peer.ID) { + log.WithContext(ctx).Tracef("peer %s doesn't have a channel, skipping network map update", peer.ID) + continue + } + + wg.Add(1) + semaphore <- struct{}{} + go func(p *nbpeer.Peer) { + defer wg.Done() + defer func() { <-semaphore }() + + // TODO: posture checks + + _, err := c.CalculateNetworkMap(data) + if err != nil { + log.WithContext(ctx).Errorf("failed to calculate network map for peer %s: %v", p.ID, err) + return + } + + c.updateChannel.SendUpdate(ctx, p.ID, &network_map.UpdateMessage{}) + }(&peer) + } + + return nil +} diff --git a/internals/controllers/network_map/controller/metrics.go b/internals/controllers/network_map/controller/metrics.go new file mode 100644 index 0000000..a39ce22 --- /dev/null +++ b/internals/controllers/network_map/controller/metrics.go @@ -0,0 +1,150 @@ +package controller + +import ( + "context" + "time" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" +) + +type metrics struct { + dbAccessDuration metric.Int64Histogram + createChannelDurationMicro metric.Int64Histogram + closeChannelDurationMicro metric.Int64Histogram + closeChannelsDurationMicro metric.Int64Histogram + closeChannels metric.Int64Histogram + sendUpdateDurationMicro metric.Int64Histogram + getAllConnectedPeersDurationMicro metric.Int64Histogram + getAllConnectedPeers metric.Int64Histogram + hasChannelDurationMicro metric.Int64Histogram +} + +func newMetrics(meter metric.Meter) (*metrics, error) { + dbAccessDuration, err := meter.Int64Histogram( + "sync_request_duration_seconds", + metric.WithDescription("Duration of sync requests in seconds"), + metric.WithUnit("s"), + ) + if err != nil { + return nil, err + } + + createChannelDurationMicro, err := meter.Int64Histogram("management.updatechannel.create.duration.micro", + metric.WithUnit("microseconds"), + metric.WithDescription("Duration of how long it takes to create a new peer update channel"), + ) + if err != nil { + return nil, err + } + + closeChannelDurationMicro, err := meter.Int64Histogram("management.updatechannel.close.one.duration.micro", + metric.WithUnit("microseconds"), + metric.WithDescription("Duration of how long it takes to close a peer update channel"), + ) + if err != nil { + return nil, err + } + + closeChannelsDurationMicro, err := meter.Int64Histogram("management.updatechannel.close.multiple.duration.micro", + metric.WithUnit("microseconds"), + metric.WithDescription("Duration of how long it takes to close a set of peer update channels"), + ) + + if err != nil { + return nil, err + } + + closeChannels, err := meter.Int64Histogram("management.updatechannel.close.multiple.channels", + metric.WithUnit("1"), + metric.WithDescription("Number of peer update channels that have been closed"), + ) + + if err != nil { + return nil, err + } + + sendUpdateDurationMicro, err := meter.Int64Histogram("management.updatechannel.send.duration.micro", + metric.WithUnit("microseconds"), + metric.WithDescription("Duration of how long it takes to send an network map update to a peer"), + ) + if err != nil { + return nil, err + } + + getAllConnectedPeersDurationMicro, err := meter.Int64Histogram("management.updatechannel.get.all.duration.micro", + metric.WithUnit("microseconds"), + metric.WithDescription("Duration of how long it takes to get all connected peers"), + ) + if err != nil { + return nil, err + } + + getAllConnectedPeers, err := meter.Int64Histogram("management.updatechannel.get.all.peers", + metric.WithUnit("1"), + metric.WithDescription("Number of connected peers"), + ) + if err != nil { + return nil, err + } + + hasChannelDurationMicro, err := meter.Int64Histogram("management.updatechannel.haschannel.duration.micro", + metric.WithUnit("microseconds"), + metric.WithDescription("Duration of how long it takes to check if a peer has a channel"), + ) + if err != nil { + return nil, err + } + + return &metrics{ + dbAccessDuration: dbAccessDuration, + createChannelDurationMicro: createChannelDurationMicro, + closeChannelDurationMicro: closeChannelDurationMicro, + closeChannelsDurationMicro: closeChannelsDurationMicro, + closeChannels: closeChannels, + sendUpdateDurationMicro: sendUpdateDurationMicro, + getAllConnectedPeersDurationMicro: getAllConnectedPeersDurationMicro, + getAllConnectedPeers: getAllConnectedPeers, + hasChannelDurationMicro: hasChannelDurationMicro, + }, nil +} + +func (m *metrics) RecordDBAccessDuration(duration time.Duration) { + m.dbAccessDuration.Record(context.Background(), duration.Milliseconds(), metric.WithAttributes()) +} + +// CountCreateChannelDuration counts the duration of the CreateChannel method, +// closed indicates if existing channel was closed before creation of a new one +func (m *metrics) CountCreateChannelDuration(duration time.Duration, closed bool) { + opts := metric.WithAttributeSet(attribute.NewSet(attribute.Bool("closed", closed))) + m.createChannelDurationMicro.Record(context.Background(), duration.Microseconds(), opts) +} + +// CountCloseChannelDuration counts the duration of the CloseChannel method +func (m *metrics) CountCloseChannelDuration(duration time.Duration) { + m.closeChannelDurationMicro.Record(context.Background(), duration.Microseconds()) +} + +// CountCloseChannelsDuration counts the duration of the CloseChannels method and the number of channels have been closed +func (m *metrics) CountCloseChannelsDuration(duration time.Duration, channels int) { + m.closeChannelsDurationMicro.Record(context.Background(), duration.Microseconds()) + m.closeChannels.Record(context.Background(), int64(channels)) +} + +// CountSendUpdateDuration counts the duration of the SendUpdate method +// found indicates if peer had channel, dropped indicates if the message was dropped due channel buffer overload +func (m *metrics) CountSendUpdateDuration(duration time.Duration, found, dropped bool) { + opts := metric.WithAttributeSet(attribute.NewSet(attribute.Bool("found", found), attribute.Bool("dropped", dropped))) + m.sendUpdateDurationMicro.Record(context.Background(), duration.Microseconds(), opts) +} + +// CountGetAllConnectedPeersDuration counts the duration of the GetAllConnectedPeers method and the number of peers have been returned +func (m *metrics) CountGetAllConnectedPeersDuration(duration time.Duration, peers int) { + m.getAllConnectedPeersDurationMicro.Record(context.Background(), duration.Microseconds()) + m.getAllConnectedPeers.Record(context.Background(), int64(peers)) +} + +// CountHasChannelDuration counts the duration of the HasChannel method +func (m *metrics) CountHasChannelDuration(duration time.Duration) { + m.hasChannelDurationMicro.Record(context.Background(), duration.Microseconds()) +} diff --git a/internals/controllers/network_map/controller/repository.go b/internals/controllers/network_map/controller/repository.go new file mode 100644 index 0000000..d322909 --- /dev/null +++ b/internals/controllers/network_map/controller/repository.go @@ -0,0 +1,41 @@ +package controller + +import ( + "time" + + "github.com/netbirdio/management-refactor/internals/controllers/network_map" + "github.com/netbirdio/management-refactor/internals/shared/db" +) + +type Repository interface { + GetNetworkMapData(accountID string) (*network_map.NetworkMapData, error) +} + +type repository struct { + store *db.Store + metrics *metrics +} + +func newRepository(s *db.Store, metrics *metrics) Repository { + return &repository{ + store: s, + metrics: metrics, + } +} + +func (r *repository) GetNetworkMapData(accountID string) (*network_map.NetworkMapData, error) { + start := time.Now() + var networkMapData network_map.NetworkMapData + err := r.store.GetOne(nil, db.LockingStrengthShare, &networkMapData, "id = ?", accountID) + if err != nil { + return nil, err + } + + // if err := r.store.Load(&networkMapData, "Peers", "Groups", "Policies", "Networks", "NetworkRouters", "NetworkResources"); err != nil { + // return nil, err + // } + + r.metrics.RecordDBAccessDuration(time.Since(start)) + + return &networkMapData, nil +} diff --git a/internals/controllers/network_map/controller/updatechannel.go b/internals/controllers/network_map/controller/updatechannel.go new file mode 100644 index 0000000..5514ec3 --- /dev/null +++ b/internals/controllers/network_map/controller/updatechannel.go @@ -0,0 +1,162 @@ +package controller + +import ( + "context" + "sync" + "time" + + log "github.com/sirupsen/logrus" + + "github.com/netbirdio/management-refactor/internals/controllers/network_map" +) + +const channelBufferSize = 100 + +type UpdateChannel struct { + // peerChannels is an update channel indexed by Peer.ID + peerChannels map[string]chan *network_map.UpdateMessage + // channelsMux keeps the mutex to access peerChannels + channelsMux *sync.RWMutex + // metrics provides method to collect application metrics + metrics *metrics +} + +// NewUpdateChannel returns a new instance of UpdateChannel +func NewUpdateChannel(metrics *metrics) *UpdateChannel { + return &UpdateChannel{ + peerChannels: make(map[string]chan *network_map.UpdateMessage), + channelsMux: &sync.RWMutex{}, + metrics: metrics, + } +} + +// SendUpdate sends update message to the peer's channel +func (p *UpdateChannel) SendUpdate(ctx context.Context, peerID string, update *network_map.UpdateMessage) { + start := time.Now() + var found, dropped bool + + p.channelsMux.RLock() + + defer func() { + p.channelsMux.RUnlock() + p.metrics.CountSendUpdateDuration(time.Since(start), found, dropped) + }() + + if channel, ok := p.peerChannels[peerID]; ok { + found = true + select { + case channel <- update: + log.WithContext(ctx).Debugf("update was sent to channel for peer %s", peerID) + default: + dropped = true + log.WithContext(ctx).Warnf("channel for peer %s is %d full or closed", peerID, len(channel)) + } + } else { + log.WithContext(ctx).Debugf("peer %s has no channel", peerID) + } +} + +// CreateChannel creates a go channel for a given peer used to deliver updates relevant to the peer. +func (p *UpdateChannel) CreateChannel(ctx context.Context, peerID string) chan *network_map.UpdateMessage { + start := time.Now() + + closed := false + + p.channelsMux.Lock() + defer func() { + p.channelsMux.Unlock() + p.metrics.CountCreateChannelDuration(time.Since(start), closed) + }() + + if channel, ok := p.peerChannels[peerID]; ok { + closed = true + delete(p.peerChannels, peerID) + close(channel) + } + // mbragin: todo shouldn't it be more? or configurable? + channel := make(chan *network_map.UpdateMessage, channelBufferSize) + p.peerChannels[peerID] = channel + + log.WithContext(ctx).Debugf("opened updates channel for a peer %s", peerID) + + return channel +} + +func (p *UpdateChannel) closeChannel(ctx context.Context, peerID string) { + if channel, ok := p.peerChannels[peerID]; ok { + delete(p.peerChannels, peerID) + close(channel) + + log.WithContext(ctx).Debugf("closed updates channel of a peer %s", peerID) + return + } + + log.WithContext(ctx).Debugf("closing updates channel: peer %s has no channel", peerID) +} + +// CloseChannels closes updates channel for each given peer +func (p *UpdateChannel) CloseChannels(ctx context.Context, peerIDs []string) { + start := time.Now() + + p.channelsMux.Lock() + defer func() { + p.channelsMux.Unlock() + p.metrics.CountCloseChannelsDuration(time.Since(start), len(peerIDs)) + }() + + for _, id := range peerIDs { + p.closeChannel(ctx, id) + } +} + +// CloseChannel closes updates channel of a given peer +func (p *UpdateChannel) CloseChannel(ctx context.Context, peerID string) { + start := time.Now() + + p.channelsMux.Lock() + defer func() { + p.channelsMux.Unlock() + p.metrics.CountCloseChannelDuration(time.Since(start)) + }() + + p.closeChannel(ctx, peerID) +} + +// GetAllConnectedPeers returns a copy of the connected peers map +func (p *UpdateChannel) GetAllConnectedPeers() map[string]struct{} { + start := time.Now() + + p.channelsMux.RLock() + + m := make(map[string]struct{}) + + defer func() { + p.channelsMux.RUnlock() + p.metrics.CountGetAllConnectedPeersDuration(time.Since(start), len(m)) + }() + + for ID := range p.peerChannels { + m[ID] = struct{}{} + } + + return m +} + +// HasChannel returns true if peers has channel in update manager, otherwise false +func (p *UpdateChannel) HasChannel(peerID string) bool { + start := time.Now() + + p.channelsMux.RLock() + + defer func() { + p.channelsMux.RUnlock() + // TODO; remove condition as it is useless + if p.metrics != nil { + p.metrics.CountHasChannelDuration(time.Since(start)) + } + }() + + _, ok := p.peerChannels[peerID] + + return ok +} diff --git a/internals/controllers/network_map/metrics.go b/internals/controllers/network_map/metrics.go deleted file mode 100644 index 1143540..0000000 --- a/internals/controllers/network_map/metrics.go +++ /dev/null @@ -1,31 +0,0 @@ -package network_map - -import ( - "context" - "time" - - "go.opentelemetry.io/otel/metric" -) - -type metrics struct { - dbAccessDuration metric.Int64Histogram -} - -func newMetrics(meter metric.Meter) (*metrics, error) { - dbAccessDuration, err := meter.Int64Histogram( - "sync_request_duration_seconds", - metric.WithDescription("Duration of sync requests in seconds"), - metric.WithUnit("s"), - ) - if err != nil { - return nil, err - } - - return &metrics{ - dbAccessDuration: dbAccessDuration, - }, nil -} - -func (m *metrics) RecordDBAccessDuration(duration time.Duration) { - m.dbAccessDuration.Record(context.Background(), duration.Milliseconds(), metric.WithAttributes()) -} diff --git a/internals/controllers/network_map/network_map.go b/internals/controllers/network_map/network_map.go new file mode 100644 index 0000000..db89576 --- /dev/null +++ b/internals/controllers/network_map/network_map.go @@ -0,0 +1,5 @@ +package network_map + +type NetworkMap struct { + Data []*string +} diff --git a/internals/controllers/network_map/repository.go b/internals/controllers/network_map/network_map_data.go similarity index 62% rename from internals/controllers/network_map/repository.go rename to internals/controllers/network_map/network_map_data.go index 26d77a9..c992c69 100644 --- a/internals/controllers/network_map/repository.go +++ b/internals/controllers/network_map/network_map_data.go @@ -1,8 +1,6 @@ package network_map import ( - "time" - nbpeer "github.com/netbirdio/netbird/management/server/peer" "github.com/netbirdio/management-refactor/internals/modules/accounts" @@ -11,7 +9,6 @@ import ( "github.com/netbirdio/management-refactor/internals/modules/networks/resources" "github.com/netbirdio/management-refactor/internals/modules/networks/routers" "github.com/netbirdio/management-refactor/internals/modules/policies" - "github.com/netbirdio/management-refactor/internals/shared/db" ) type NetworkMapData struct { @@ -30,36 +27,3 @@ type NetworkMapData struct { NetworkRouters []*routers.NetworkRouter `gorm:"foreignKey:AccountID;references:id"` NetworkResources []*resources.NetworkResource `gorm:"foreignKey:AccountID;references:id"` } - -type Repository interface { - GetNetworkMapData(accountID string) (*NetworkMapData, error) -} - -type repository struct { - store *db.Store - metrics *metrics -} - -func newRepository(s *db.Store, metrics *metrics) Repository { - return &repository{ - store: s, - metrics: metrics, - } -} - -func (r *repository) GetNetworkMapData(accountID string) (*NetworkMapData, error) { - start := time.Now() - var networkMapData NetworkMapData - err := r.store.GetOne(nil, db.LockingStrengthShare, &networkMapData, "id = ?", accountID) - if err != nil { - return nil, err - } - - // if err := r.store.Load(&networkMapData, "Peers", "Groups", "Policies", "Networks", "NetworkRouters", "NetworkResources"); err != nil { - // return nil, err - // } - - r.metrics.RecordDBAccessDuration(time.Since(start)) - - return &networkMapData, nil -} diff --git a/internals/controllers/network_map/updatechannel.go b/internals/controllers/network_map/updatechannel.go index d44baec..e0e037e 100644 --- a/internals/controllers/network_map/updatechannel.go +++ b/internals/controllers/network_map/updatechannel.go @@ -1,185 +1,8 @@ package network_map -import ( - "context" - "sync" - "time" +import "context" - "github.com/netbirdio/netbird/management/server/groups" - log "github.com/sirupsen/logrus" - - "github.com/netbirdio/netbird/management/proto" - "github.com/netbirdio/netbird/management/server/telemetry" - "github.com/netbirdio/netbird/management/server/types" - - "github.com/netbirdio/management-refactor/internals/modules/peers" - "github.com/netbirdio/management-refactor/internals/modules/policies" -) - -const channelBufferSize = 100 - -type UpdateMessage struct { - Update *proto.SyncResponse - NetworkMap *types.NetworkMap - PeerManager *peers.Manager - PolicyManager *policies.Manager - GroupManager *groups.Manager -} - -type UpdateChannel struct { - // peerChannels is an update channel indexed by Peer.ID - peerChannels map[string]chan *UpdateMessage - // channelsMux keeps the mutex to access peerChannels - channelsMux *sync.RWMutex - // metrics provides method to collect application metrics - metrics telemetry.AppMetrics -} - -// NewUpdateChannel returns a new instance of UpdateChannel -func NewUpdateChannel(metrics telemetry.AppMetrics) *UpdateChannel { - return &UpdateChannel{ - peerChannels: make(map[string]chan *UpdateMessage), - channelsMux: &sync.RWMutex{}, - metrics: metrics, - } -} - -// SendUpdate sends update message to the peer's channel -func (p *UpdateChannel) SendUpdate(ctx context.Context, peerID string, update *UpdateMessage) { - start := time.Now() - var found, dropped bool - - p.channelsMux.RLock() - - defer func() { - p.channelsMux.RUnlock() - if p.metrics != nil { - p.metrics.UpdateChannelMetrics().CountSendUpdateDuration(time.Since(start), found, dropped) - } - }() - - if channel, ok := p.peerChannels[peerID]; ok { - found = true - select { - case channel <- update: - log.WithContext(ctx).Debugf("update was sent to channel for peer %s", peerID) - default: - dropped = true - log.WithContext(ctx).Warnf("channel for peer %s is %d full or closed", peerID, len(channel)) - } - } else { - log.WithContext(ctx).Debugf("peer %s has no channel", peerID) - } -} - -// CreateChannel creates a go channel for a given peer used to deliver updates relevant to the peer. -func (p *UpdateChannel) CreateChannel(ctx context.Context, peerID string) chan *UpdateMessage { - start := time.Now() - - closed := false - - p.channelsMux.Lock() - defer func() { - p.channelsMux.Unlock() - if p.metrics != nil { - p.metrics.UpdateChannelMetrics().CountCreateChannelDuration(time.Since(start), closed) - } - }() - - if channel, ok := p.peerChannels[peerID]; ok { - closed = true - delete(p.peerChannels, peerID) - close(channel) - } - // mbragin: todo shouldn't it be more? or configurable? - channel := make(chan *UpdateMessage, channelBufferSize) - p.peerChannels[peerID] = channel - - log.WithContext(ctx).Debugf("opened updates channel for a peer %s", peerID) - - return channel -} - -func (p *UpdateChannel) closeChannel(ctx context.Context, peerID string) { - if channel, ok := p.peerChannels[peerID]; ok { - delete(p.peerChannels, peerID) - close(channel) - - log.WithContext(ctx).Debugf("closed updates channel of a peer %s", peerID) - return - } - - log.WithContext(ctx).Debugf("closing updates channel: peer %s has no channel", peerID) -} - -// CloseChannels closes updates channel for each given peer -func (p *UpdateChannel) CloseChannels(ctx context.Context, peerIDs []string) { - start := time.Now() - - p.channelsMux.Lock() - defer func() { - p.channelsMux.Unlock() - if p.metrics != nil { - p.metrics.UpdateChannelMetrics().CountCloseChannelsDuration(time.Since(start), len(peerIDs)) - } - }() - - for _, id := range peerIDs { - p.closeChannel(ctx, id) - } -} - -// CloseChannel closes updates channel of a given peer -func (p *UpdateChannel) CloseChannel(ctx context.Context, peerID string) { - start := time.Now() - - p.channelsMux.Lock() - defer func() { - p.channelsMux.Unlock() - if p.metrics != nil { - p.metrics.UpdateChannelMetrics().CountCloseChannelDuration(time.Since(start)) - } - }() - - p.closeChannel(ctx, peerID) -} - -// GetAllConnectedPeers returns a copy of the connected peers map -func (p *UpdateChannel) GetAllConnectedPeers() map[string]struct{} { - start := time.Now() - - p.channelsMux.RLock() - - m := make(map[string]struct{}) - - defer func() { - p.channelsMux.RUnlock() - if p.metrics != nil { - p.metrics.UpdateChannelMetrics().CountGetAllConnectedPeersDuration(time.Since(start), len(m)) - } - }() - - for ID := range p.peerChannels { - m[ID] = struct{}{} - } - - return m -} - -// HasChannel returns true if peers has channel in update manager, otherwise false -func (p *UpdateChannel) HasChannel(peerID string) bool { - start := time.Now() - - p.channelsMux.RLock() - - defer func() { - p.channelsMux.RUnlock() - if p.metrics != nil { - p.metrics.UpdateChannelMetrics().CountHasChannelDuration(time.Since(start)) - } - }() - - _, ok := p.peerChannels[peerID] - - return ok +type UpdateChannel interface { + UpdatePeers(accountID string) error + SendUpdate(ctx context.Context, peerID string, update *UpdateMessage) } diff --git a/internals/controllers/network_map/updatemessage.go b/internals/controllers/network_map/updatemessage.go new file mode 100644 index 0000000..6f5af80 --- /dev/null +++ b/internals/controllers/network_map/updatemessage.go @@ -0,0 +1,18 @@ +package network_map + +import ( + "github.com/netbirdio/netbird/management/proto" + "github.com/netbirdio/netbird/management/server/groups" + "github.com/netbirdio/netbird/management/server/types" + + "github.com/netbirdio/management-refactor/internals/modules/peers" + "github.com/netbirdio/management-refactor/internals/modules/policies" +) + +type UpdateMessage struct { + Update *proto.SyncResponse + NetworkMap *types.NetworkMap + PeerManager *peers.Manager + PolicyManager *policies.Manager + GroupManager *groups.Manager +} diff --git a/internals/server/controllers.go b/internals/server/controllers.go index 06bb0bd..8adeb9c 100644 --- a/internals/server/controllers.go +++ b/internals/server/controllers.go @@ -1,11 +1,14 @@ package server -import "github.com/netbirdio/management-refactor/internals/controllers/network_map" +import ( + "github.com/netbirdio/management-refactor/internals/controllers/network_map" + "github.com/netbirdio/management-refactor/internals/controllers/network_map/controller" +) -func (s *BaseServer) NetworkMapController() *network_map.Controller { - return Create(s, func() *network_map.Controller { +func (s *BaseServer) NetworkMapController() network_map.Controller { + return Create(s, func() network_map.Controller { store := s.Store() metrics := s.Metrics() - return network_map.NewController(store, metrics) + return controller.NewController(store, metrics) }) }