mirror of
https://github.com/netbirdio/management-refactor.git
synced 2026-05-22 17:12:59 -07:00
set up modules and controller
This commit is contained in:
Pascal Fischer
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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())
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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())
|
||||
}
|
||||
@@ -0,0 +1,5 @@
|
||||
package network_map
|
||||
|
||||
type NetworkMap struct {
|
||||
Data []*string
|
||||
}
|
||||
-36
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
})
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user