separate update channel

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
 	},

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)
 }

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
 }

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())
-}

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())
+}

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

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)
 }

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)

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) {

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
 }

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())
 	})
 }

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
 	})

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) {

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

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

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
 	}