Merge branch 'main' into test/multiple-peer-logging

management/server/account.go

@@ -102,6 +102,8 @@ type DefaultAccountManager struct {
 
 	accountUpdateLocks               sync.Map
 	updateAccountPeersBufferInterval atomic.Int64
+
+	loginFilter *loginFilter
 }
 
 // getJWTGroupsChanges calculates the changes needed to sync a user's JWT groups.
@@ -195,6 +197,7 @@ func BuildManager(
 		proxyController:          proxyController,
 		settingsManager:          settingsManager,
 		permissionsManager:       permissionsManager,
+		loginFilter:              newLoginFilter(),
 	}
 
 	am.startWarmup(ctx)
@@ -1536,6 +1539,10 @@ func domainIsUpToDate(domain string, domainCategory string, userAuth nbcontext.U
 	return domainCategory == types.PrivateCategory || userAuth.DomainCategory != types.PrivateCategory || domain != userAuth.Domain
 }
 
+func (am *DefaultAccountManager) AllowSync(wgPubKey, metahash string) bool {
+	return true
+}
+
 func (am *DefaultAccountManager) SyncAndMarkPeer(ctx context.Context, accountID string, peerPubKey string, meta nbpeer.PeerSystemMeta, realIP net.IP) (*nbpeer.Peer, *types.NetworkMap, []*posture.Checks, error) {
 	start := time.Now()
 	defer func() {
@@ -1557,6 +1564,9 @@ func (am *DefaultAccountManager) SyncAndMarkPeer(ctx context.Context, accountID
 		log.WithContext(ctx).Warnf("failed marking peer as connected %s %v", peerPubKey, err)
 	}
 
+	metahash := metaHash(meta, realIP.String())
+	am.loginFilter.addLogin(peerPubKey, metahash)
+
 	return peer, netMap, postureChecks, nil
 }
 
@@ -1570,7 +1580,6 @@ func (am *DefaultAccountManager) OnPeerDisconnected(ctx context.Context, account
 	if err != nil {
 		log.WithContext(ctx).Warnf("failed marking peer as disconnected %s %v", peerPubKey, err)
 	}
-
 	return nil
 
 }

management/server/account/manager.go

@@ -117,4 +117,5 @@ type Manager interface {
 	UpdateToPrimaryAccount(ctx context.Context, accountId string) (*types.Account, error)
 	GetOwnerInfo(ctx context.Context, accountId string) (*types.UserInfo, error)
 	GetCurrentUserInfo(ctx context.Context, userAuth nbcontext.UserAuth) (*users.UserInfoWithPermissions, error)
+	AllowSync(string, string) bool
 }

management/server/grpcserver.go

@@ -166,7 +166,8 @@ func (s *GRPCServer) Sync(req *proto.EncryptedMessage, srv proto.ManagementServi
 	ctx = context.WithValue(ctx, nbContext.AccountIDKey, accountID)
 
 	realIP := getRealIP(ctx)
-	log.WithContext(ctx).Debugf("Sync request from peer [%s] [%s]", req.WgPubKey, realIP.String())
+	sip := realIP.String()
+	log.WithContext(ctx).Debugf("Sync request from peer [%s] [%s] [%s]", req.WgPubKey, sip, metaHash(extractPeerMeta(ctx, syncReq.GetMeta()), sip))
 
 	if syncReq.GetMeta() == nil {
 		log.WithContext(ctx).Tracef("peer system meta has to be provided on sync. Peer %s, remote addr %s", peerKey.String(), realIP)

management/server/loginfilter.go

@@ -0,0 +1,113 @@
+package server
+
+import (
+	"strings"
+	"sync"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+
+	nbpeer "github.com/netbirdio/netbird/management/server/peer"
+)
+
+const (
+	loginFilterSize = 100_000         // Size of the login filter map, making it large enough for a future
+	filterTimeout   = 5 * time.Minute // Duration to secure the previous login information in the filter
+
+	loggingLimit = 100
+
+	loggingLimitOnePeer    = 30
+	loggingTresholdOnePeer = 5 * time.Minute
+)
+
+type loginFilter struct {
+	mu     sync.RWMutex
+	logged map[string]metahash
+}
+
+type metahash struct {
+	hashes  map[string]struct{}
+	counter int
+	start   time.Time
+}
+
+func newLoginFilter() *loginFilter {
+	return &loginFilter{
+		logged: make(map[string]metahash, loginFilterSize),
+	}
+}
+
+func (l *loginFilter) addLogin(wgPubKey, metaHash string) {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	mh, ok := l.logged[wgPubKey]
+	if !ok {
+		mh = metahash{
+			hashes: make(map[string]struct{}, loggingLimit),
+			start:  time.Now(),
+		}
+	}
+	mh.hashes[metaHash] = struct{}{}
+	mh.counter++
+	if mh.counter >= loggingLimit && mh.counter%loggingLimit == 0 && len(mh.hashes) > 1 {
+		log.WithFields(log.Fields{
+			"wgPubKey":                            wgPubKey,
+			"number of different hashes":          len(mh.hashes),
+			"elapsed time for number of attempts": time.Since(mh.start),
+			"number of syncs":                     mh.counter,
+		}).Info(mh.prepareHashes())
+	} else if mh.counter%loggingLimitOnePeer == 0 && time.Since(mh.start) > loggingTresholdOnePeer && len(mh.hashes) == 1 {
+		log.WithFields(log.Fields{
+			"wgPubKey":                            wgPubKey,
+			"elapsed time for number of attempts": time.Since(mh.start),
+			"number of syncs":                     mh.counter,
+		}).Info(mh.prepareHashes())
+		mh.start = time.Now()
+	}
+	l.logged[wgPubKey] = mh
+}
+
+func (m *metahash) prepareHashes() string {
+	var sb strings.Builder
+	for hash := range m.hashes {
+		sb.WriteString(hash)
+		sb.WriteString(", ")
+	}
+	return sb.String()
+}
+
+func metaHash(meta nbpeer.PeerSystemMeta, pubip string) string {
+	mac := getMacAddress(meta.NetworkAddresses)
+	estimatedSize := len(meta.WtVersion) + len(meta.OSVersion) + len(meta.KernelVersion) + len(meta.Hostname) + len(meta.SystemSerialNumber) +
+		len(pubip) + len(mac) + 6
+
+	var b strings.Builder
+	b.Grow(estimatedSize)
+
+	b.WriteString(meta.WtVersion)
+	b.WriteByte('|')
+	b.WriteString(meta.OSVersion)
+	b.WriteByte('|')
+	b.WriteString(meta.KernelVersion)
+	b.WriteByte('|')
+	b.WriteString(meta.Hostname)
+	b.WriteByte('|')
+	b.WriteString(meta.SystemSerialNumber)
+	b.WriteByte('|')
+	b.WriteString(pubip)
+	b.WriteByte('|')
+	b.WriteString(mac)
+
+	return b.String()
+}
+
+func getMacAddress(nas []nbpeer.NetworkAddress) string {
+	if len(nas) == 0 {
+		return ""
+	}
+	macs := make([]string, 0, len(nas))
+	for _, na := range nas {
+		macs = append(macs, na.Mac)
+	}
+	return strings.Join(macs, "/")
+}

management/server/mock_server/account_mock.go

@@ -119,6 +119,8 @@ type MockAccountManager struct {
 	GetAccountMetaFunc                  func(ctx context.Context, accountID, userID string) (*types.AccountMeta, error)
 
 	GetOrCreateAccountByPrivateDomainFunc func(ctx context.Context, initiatorId, domain string) (*types.Account, bool, error)
+
+	AllowSyncFunc func(string, string) bool
 }
 
 func (am *MockAccountManager) UpdateAccountPeers(ctx context.Context, accountID string) {
@@ -890,3 +892,7 @@ func (am *MockAccountManager) GetCurrentUserInfo(ctx context.Context, userAuth n
 	}
 	return nil, status.Errorf(codes.Unimplemented, "method GetCurrentUserInfo is not implemented")
 }
+
+func (am *MockAccountManager) AllowSync(_, _ string) bool {
+	return true
+}

management/server/peer_test.go

@@ -1579,7 +1579,6 @@ func Test_LoginPeer(t *testing.T) {
 	testCases := []struct {
 		name                         string
 		setupKey                     string
-		wireGuardPubKey              string
 		expectExtraDNSLabelsMismatch bool
 		extraDNSLabels               []string
 		expectLoginError             bool

management/server/status/error.go

@@ -42,7 +42,10 @@ const (
 // Type is a type of the Error
 type Type int32
 
-var ErrExtraSettingsNotFound = fmt.Errorf("extra settings not found")
+var (
+	ErrExtraSettingsNotFound = fmt.Errorf("extra settings not found")
+	ErrPeerAlreadyLoggedIn   = errors.New("peer with the same public key is already logged in")
+)
 
 // Error is an internal error
 type Error struct {