diff --git a/cmd/management.go b/cmd/management.go index 1ddaf17..077c70b 100644 --- a/cmd/management.go +++ b/cmd/management.go @@ -13,7 +13,7 @@ import ( "github.com/netbirdio/management-refactor/pkg/logging" ) -var log = logging.LoggerForThisPackage() +var log = logging.LoggerForThisPackage // mgmtCmd starts the management server var mgmtCmd = &cobra.Command{ @@ -22,15 +22,15 @@ var mgmtCmd = &cobra.Command{ RunE: func(cmd *cobra.Command, args []string) error { err := logging.Init("logging.yaml") if err != nil { - log.Debugf("Failed to init logging: %v", err) + log().Debugf("Failed to init logging: %v", err) } srv := integrations.InitCloud(server.NewServer()) go func() { - log.Info("Starting server on :8080") + log().Info("Starting server on :8080") if err := srv.Start(); err != nil { - log.Fatalf("Server error: %v", err) + log().Fatalf("Server error: %v", err) } }() @@ -38,11 +38,11 @@ var mgmtCmd = &cobra.Command{ signal.Notify(stopChan, os.Interrupt, syscall.SIGTERM) <-stopChan - log.Info("Shutting down server...") + log().Info("Shutting down server...") if err := srv.Stop(); err != nil { - log.Errorf("Error stopping server: %v", err) + log().Errorf("Error stopping server: %v", err) } - log.Info("Server stopped gracefully.") + log().Info("Server stopped gracefully.") return nil }, diff --git a/internals/controllers/network_map/controller.go b/internals/controllers/network_map/controller.go index dea2d5e..c3db400 100644 --- a/internals/controllers/network_map/controller.go +++ b/internals/controllers/network_map/controller.go @@ -4,5 +4,4 @@ import "context" 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 index 42b8072..ddc46fb 100644 --- a/internals/controllers/network_map/controller/controller.go +++ b/internals/controllers/network_map/controller/controller.go @@ -2,73 +2,53 @@ 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" + "github.com/netbirdio/management-refactor/pkg/logging" ) +var log = logging.LoggerForThisPackage + type Controller struct { repo Repository metrics *metrics - updateChannel *UpdateChannel + UpdateChannel network_map.UpdateChannel } -func NewController(store *db.Store, metrics *appmetrics.AppMetrics) *Controller { +func NewController(store *db.Store, metrics *appmetrics.AppMetrics, updateChannel network_map.UpdateChannel) *Controller { cMetrics, err := appmetrics.RegisterMetrics(metrics, newMetrics) if err != nil { - log.Fatalf("Failed to register app metrics: %v", err) + log().Fatalf("Failed to register app metrics: %v", err) } return &Controller{ repo: newRepository(store, cMetrics), metrics: cMetrics, - updateChannel: NewUpdateChannel(cMetrics), + UpdateChannel: updateChannel, } } -func (c *Controller) CalculateNetworkMap(data *network_map.NetworkMapData) (*network_map.NetworkMap, error) { +func (c *Controller) CalculateNetworkMap(accountID string) (*network_map.NetworkMap, error) { + _, err := c.repo.GetNetworkMapData(accountID) + if err != nil { + // usually return error + } + // Do calc on data + log().Tracef("Calculating network map for account on public") + return &network_map.NetworkMap{}, nil } func (c *Controller) UpdatePeers(ctx context.Context, accountID string) error { - data, err := c.repo.GetNetworkMapData(accountID) + _, err := c.CalculateNetworkMap(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) + log().Errorf("Failed to calculate network map for account %s: %v", accountID, err) + return err } + c.UpdateChannel.SendUpdate(ctx, accountID, &network_map.UpdateMessage{}) return nil } diff --git a/internals/controllers/network_map/controller/metrics.go b/internals/controllers/network_map/controller/metrics.go index a39ce22..b4f4cb0 100644 --- a/internals/controllers/network_map/controller/metrics.go +++ b/internals/controllers/network_map/controller/metrics.go @@ -4,20 +4,11 @@ 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 + dbAccessDuration metric.Int64Histogram } func newMetrics(meter metric.Meter) (*metrics, error) { @@ -30,121 +21,11 @@ func newMetrics(meter metric.Meter) (*metrics, error) { 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, + dbAccessDuration: dbAccessDuration, }, 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/update_channel/metrics.go b/internals/controllers/network_map/update_channel/metrics.go new file mode 100644 index 0000000..2113cea --- /dev/null +++ b/internals/controllers/network_map/update_channel/metrics.go @@ -0,0 +1,135 @@ +package update_channel + +import ( + "context" + "time" + + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" +) + +type metrics struct { + 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) { + 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{ + createChannelDurationMicro: createChannelDurationMicro, + closeChannelDurationMicro: closeChannelDurationMicro, + closeChannelsDurationMicro: closeChannelsDurationMicro, + closeChannels: closeChannels, + sendUpdateDurationMicro: sendUpdateDurationMicro, + getAllConnectedPeersDurationMicro: getAllConnectedPeersDurationMicro, + getAllConnectedPeers: getAllConnectedPeers, + hasChannelDurationMicro: hasChannelDurationMicro, + }, nil +} + +// 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/updatechannel.go b/internals/controllers/network_map/update_channel/updatechannel.go similarity index 77% rename from internals/controllers/network_map/controller/updatechannel.go rename to internals/controllers/network_map/update_channel/updatechannel.go index 5514ec3..8d9c3c5 100644 --- a/internals/controllers/network_map/controller/updatechannel.go +++ b/internals/controllers/network_map/update_channel/updatechannel.go @@ -1,17 +1,19 @@ -package controller +package update_channel import ( "context" "sync" "time" - log "github.com/sirupsen/logrus" - "github.com/netbirdio/management-refactor/internals/controllers/network_map" + appmetrics "github.com/netbirdio/management-refactor/internals/shared/metrics" + "github.com/netbirdio/management-refactor/pkg/logging" ) const channelBufferSize = 100 +var log = logging.LoggerForThisPackage + type UpdateChannel struct { // peerChannels is an update channel indexed by Peer.ID peerChannels map[string]chan *network_map.UpdateMessage @@ -22,16 +24,22 @@ type UpdateChannel struct { } // NewUpdateChannel returns a new instance of UpdateChannel -func NewUpdateChannel(metrics *metrics) *UpdateChannel { +func NewUpdateChannel(metrics *appmetrics.AppMetrics) *UpdateChannel { + cMetrics, err := appmetrics.RegisterMetrics(metrics, newMetrics) + if err != nil { + log().Fatalf("Failed to register updatechannel metrics: %v", err) + } return &UpdateChannel{ peerChannels: make(map[string]chan *network_map.UpdateMessage), channelsMux: &sync.RWMutex{}, - metrics: metrics, + metrics: cMetrics, } } // SendUpdate sends update message to the peer's channel func (p *UpdateChannel) SendUpdate(ctx context.Context, peerID string, update *network_map.UpdateMessage) { + log().Debugf("Sending update message to peer %s on public update channel", peerID) + start := time.Now() var found, dropped bool @@ -46,13 +54,13 @@ func (p *UpdateChannel) SendUpdate(ctx context.Context, peerID string, update *n found = true select { case channel <- update: - log.WithContext(ctx).Debugf("update was sent to channel for peer %s", peerID) + 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)) + 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) + log().WithContext(ctx).Debugf("peer %s has no channel", peerID) } } @@ -77,7 +85,7 @@ func (p *UpdateChannel) CreateChannel(ctx context.Context, peerID string) chan * channel := make(chan *network_map.UpdateMessage, channelBufferSize) p.peerChannels[peerID] = channel - log.WithContext(ctx).Debugf("opened updates channel for a peer %s", peerID) + log().WithContext(ctx).Debugf("opened updates channel for a peer %s", peerID) return channel } @@ -87,11 +95,11 @@ func (p *UpdateChannel) closeChannel(ctx context.Context, peerID string) { delete(p.peerChannels, peerID) close(channel) - log.WithContext(ctx).Debugf("closed updates channel of a peer %s", peerID) + 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) + log().WithContext(ctx).Debugf("closing updates channel: peer %s has no channel", peerID) } // CloseChannels closes updates channel for each given peer diff --git a/internals/controllers/network_map/updatechannel.go b/internals/controllers/network_map/updatechannel.go index e0e037e..54b0f3d 100644 --- a/internals/controllers/network_map/updatechannel.go +++ b/internals/controllers/network_map/updatechannel.go @@ -3,6 +3,5 @@ package network_map import "context" type UpdateChannel interface { - UpdatePeers(accountID string) error SendUpdate(ctx context.Context, peerID string, update *UpdateMessage) } diff --git a/internals/modules/peers/interface.go b/internals/modules/peers/interface.go index 8881ccb..e1719a0 100644 --- a/internals/modules/peers/interface.go +++ b/internals/modules/peers/interface.go @@ -3,10 +3,12 @@ package peers import ( "context" + "github.com/netbirdio/management-refactor/internals/controllers/network_map" "github.com/netbirdio/management-refactor/internals/shared/db" ) type Manager interface { + SetNetworkMapController(networkMapController network_map.Controller) GetPeer(ctx context.Context, tx db.Transaction, strength db.LockingStrength, accountID, peerID string) (*Peer, error) GetPeers(ctx context.Context, tx db.Transaction, strength db.LockingStrength, accountID string) ([]*Peer, error) GetFilteredPeers(ctx context.Context, tx db.Transaction, strength db.LockingStrength, accountID, nameFilter, ipFilter string) ([]*Peer, error) diff --git a/internals/modules/peers/manager/api.go b/internals/modules/peers/manager/api.go index ebce2da..e15f27c 100644 --- a/internals/modules/peers/manager/api.go +++ b/internals/modules/peers/manager/api.go @@ -6,6 +6,7 @@ import ( "github.com/gorilla/mux" nbcontext "github.com/netbirdio/netbird/management/server/context" "github.com/netbirdio/netbird/management/server/http/util" + "github.com/netbirdio/netbird/management/server/status" "github.com/netbirdio/management-refactor/internals/modules/peers" "github.com/netbirdio/management-refactor/internals/shared/permissions" @@ -39,7 +40,24 @@ func (h *handler) getPeer(w http.ResponseWriter, r *http.Request, userAuth *nbco } func (h *handler) updatePeer(w http.ResponseWriter, r *http.Request, userAuth *nbcontext.UserAuth) { + vars := mux.Vars(r) + peerID, ok := vars["peerId"] + if !ok { + util.WriteError(r.Context(), status.Errorf(status.InvalidArgument, "peer ID field is missing"), w) + return + } + if len(peerID) == 0 { + util.WriteError(r.Context(), status.Errorf(status.InvalidArgument, "peer ID can't be empty"), w) + return + } + err := h.manager.UpdatePeer(r.Context(), nil, &peers.Peer{ID: peerID, AccountID: peerID}) + if err != nil { + util.WriteErrorResponse("Failed to update peer", http.StatusInternalServerError, w) + return + } + + util.WriteJSONObject(r.Context(), w, map[string]string{"status": "success"}) } func (h *handler) deletePeer(w http.ResponseWriter, r *http.Request, userAuth *nbcontext.UserAuth) { diff --git a/internals/modules/peers/manager/manager.go b/internals/modules/peers/manager/manager.go index 4ee2de9..301d98f 100644 --- a/internals/modules/peers/manager/manager.go +++ b/internals/modules/peers/manager/manager.go @@ -3,24 +3,30 @@ package manager import ( "context" + "github.com/netbirdio/management-refactor/internals/controllers/network_map" "github.com/netbirdio/management-refactor/internals/modules/peers" "github.com/netbirdio/management-refactor/internals/shared/activity" "github.com/netbirdio/management-refactor/internals/shared/db" "github.com/netbirdio/management-refactor/pkg/logging" ) -var log = logging.LoggerForThisPackage() +var log = logging.LoggerForThisPackage var _ peers.Manager = (*Manager)(nil) type Manager struct { - repo Repository - eventManager *activity.Manager + repo Repository + eventManager *activity.Manager + networkMapController network_map.Controller } func NewManager(store *db.Store) *Manager { return &Manager{repo: newRepository(store)} +} +func (m *Manager) SetNetworkMapController(networkMapController network_map.Controller) { + log().Tracef("Setting network map controller for peers manager") + m.networkMapController = networkMapController } func (m *Manager) GetPeer(ctx context.Context, tx db.Transaction, strength db.LockingStrength, accountID, peerID string) (*peers.Peer, error) { @@ -36,5 +42,12 @@ func (m *Manager) GetFilteredPeers(ctx context.Context, tx db.Transaction, stren } func (m *Manager) UpdatePeer(ctx context.Context, tx db.Transaction, peer *peers.Peer) error { - return m.repo.UpdatePeer(tx, peer) + // err := m.repo.UpdatePeer(tx, peer) + // if err != nil { + // return fmt.Errorf("failed to update peer: %w", err) + // } + + _ = m.networkMapController.UpdatePeers(ctx, peer.AccountID) + + return nil } diff --git a/internals/server/controllers.go b/internals/server/controllers.go index 8adeb9c..e58f6d5 100644 --- a/internals/server/controllers.go +++ b/internals/server/controllers.go @@ -3,12 +3,17 @@ package server import ( "github.com/netbirdio/management-refactor/internals/controllers/network_map" "github.com/netbirdio/management-refactor/internals/controllers/network_map/controller" + "github.com/netbirdio/management-refactor/internals/controllers/network_map/update_channel" ) func (s *BaseServer) NetworkMapController() network_map.Controller { return Create(s, func() network_map.Controller { - store := s.Store() - metrics := s.Metrics() - return controller.NewController(store, metrics) + return controller.NewController(s.Store(), s.Metrics(), s.NetworkMapUpdateChannel()) + }) +} + +func (s *BaseServer) NetworkMapUpdateChannel() network_map.UpdateChannel { + return Create(s, func() network_map.UpdateChannel { + return update_channel.NewUpdateChannel(s.Metrics()) }) } diff --git a/internals/server/modules.go b/internals/server/modules.go index f46251e..313cd27 100644 --- a/internals/server/modules.go +++ b/internals/server/modules.go @@ -14,7 +14,7 @@ import ( func (s *BaseServer) NetworksManager() networks.Manager { return Create(s, func() networks.Manager { - return manager.NewManager(s.Store(), s.Router(), s.PermissionsManager()) + return manager.NewManager(s.Store()) }) } @@ -36,6 +36,7 @@ func (s *BaseServer) PeersManager() peers.Manager { manager := peersManager.NewManager(s.Store()) s.AfterInit(func(s *BaseServer) { peersManager.RegisterEndpoints(s.Router(), s.PermissionsManager(), manager) + manager.SetNetworkMapController(s.NetworkMapController()) }) return manager }) diff --git a/internals/shared/db/database_connection.go b/internals/shared/db/database_connection.go index c361ee7..e1d6a95 100644 --- a/internals/shared/db/database_connection.go +++ b/internals/shared/db/database_connection.go @@ -28,10 +28,10 @@ type DatabaseConn struct { func NewDatabaseConn(ctx context.Context) (*DatabaseConn, error) { cfg, err := configuration.Parse[config]() if err != nil { - log.Fatalf("failed to parse config: %v", err) + log().Fatalf("failed to parse config: %v", err) } - log.WithContext(ctx).Infof("using %s store engine", cfg.Engine) + log().WithContext(ctx).Infof("using %s store engine", cfg.Engine) var db *gorm.DB switch Engine(cfg.Engine) { diff --git a/internals/shared/db/store.go b/internals/shared/db/store.go index 5d00e87..e7662af 100644 --- a/internals/shared/db/store.go +++ b/internals/shared/db/store.go @@ -9,7 +9,7 @@ import ( "github.com/netbirdio/management-refactor/pkg/logging" ) -var log = logging.LoggerForThisPackage() +var log = logging.LoggerForThisPackage type Store struct { db *gorm.DB diff --git a/logging.yaml b/logging.yaml index 221688f..71fcdfb 100644 --- a/logging.yaml +++ b/logging.yaml @@ -1,7 +1,5 @@ log_levels: - default: debug + default: trace management-refactor/internals/shared/activity: error - management-refactor/internals/shared/db: debug - management-refactor/internals/shared/permissions: debug - management-refactor/internals/modules/peers/manager: warn - management-integrations-refactor/integrations: info + management-refactor/internals/shared/db: error + management-integrations-refactor/integrations: trace diff --git a/pkg/logging/init.go b/pkg/logging/init.go index 0bd722f..cba80b5 100644 --- a/pkg/logging/init.go +++ b/pkg/logging/init.go @@ -62,7 +62,7 @@ func Init(configFilePath string) error { // Optionally, define a default logger for packages not explicitly listed if _, ok := loggers["default"]; !ok { defaultLogger := logrus.New() - defaultLogger.SetLevel(logrus.InfoLevel) + defaultLogger.SetLevel(logrus.TraceLevel) loggers["default"] = defaultLogger }