From 69f13e3733c40cc9d83a14c3e044a6de8c54bbf6 Mon Sep 17 00:00:00 2001 From: Bethuel Mmbaga Date: Fri, 23 Oct 2026 16:04:09 +0300 Subject: [PATCH 1/5] [management] Move reverse proxy access logs to a dedicated database (#8303) --- combined/cmd/admin_config_test.go | 20 +++ combined/cmd/config.go | 1 + combined/cmd/root.go | 16 ++ combined/config.yaml.example | 7 + .../modules/reverseproxy/accesslogs/filter.go | 2 +- .../reverseproxy/accesslogs/manager/conn.go | 92 ++++++++++++ .../accesslogs/manager/conn_test.go | 142 ++++++++++++++++++ .../accesslogs/manager/repository.go | 4 +- .../accesslogs/manager/repository_test.go | 34 +++++ management/internals/server/boot.go | 21 ++- management/internals/shared/db/conn.go | 32 +++- management/internals/shared/db/conn_test.go | 19 +++ management/internals/shared/db/open.go | 19 ++- .../testing/testing_tools/channel/channel.go | 16 +- management/server/store/sql_store.go | 3 +- management/server/store/store.go | 7 +- shared/management/http/api/openapi.yml | 2 +- shared/management/http/api/types.gen.go | 2 +- 18 files changed, 421 insertions(+), 18 deletions(-) create mode 100644 management/internals/modules/reverseproxy/accesslogs/manager/conn.go create mode 100644 management/internals/modules/reverseproxy/accesslogs/manager/conn_test.go diff --git a/combined/cmd/admin_config_test.go b/combined/cmd/admin_config_test.go index 5cec26552..c1ea2e8da 100644 --- a/combined/cmd/admin_config_test.go +++ b/combined/cmd/admin_config_test.go @@ -7,6 +7,7 @@ import ( "github.com/stretchr/testify/require" + accesslogsmanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs/manager" nbconfig "github.com/netbirdio/netbird/management/internals/server/config" ) @@ -41,3 +42,22 @@ func TestApplyServerStoreEnv(t *testing.T) { require.Equal(t, "postgres-dsn", os.Getenv("NB_STORE_ENGINE_POSTGRES_DSN")) require.Equal(t, "store.db", os.Getenv("NB_STORE_ENGINE_SQLITE_FILE")) } + +func TestApplyAccessLogStoreEnv(t *testing.T) { + keys := []string{accesslogsmanager.StoreEngineEnv, accesslogsmanager.PostgresDSNEnv, accesslogsmanager.SqliteFileEnv} + for _, key := range keys { + t.Setenv(key, "") + require.NoError(t, os.Unsetenv(key)) + } + + applyAccessLogStoreEnv(StoreConfig{}) + for _, key := range keys { + _, ok := os.LookupEnv(key) + require.False(t, ok, "%s should stay unset for an empty accessLogStore block", key) + } + + applyAccessLogStoreEnv(StoreConfig{Engine: "Postgres", DSN: "postgres-dsn", File: "access_logs.db"}) + require.Equal(t, "postgres", os.Getenv(accesslogsmanager.StoreEngineEnv)) + require.Equal(t, "postgres-dsn", os.Getenv(accesslogsmanager.PostgresDSNEnv)) + require.Equal(t, "access_logs.db", os.Getenv(accesslogsmanager.SqliteFileEnv)) +} diff --git a/combined/cmd/config.go b/combined/cmd/config.go index ca3b25617..0c7302fe5 100644 --- a/combined/cmd/config.go +++ b/combined/cmd/config.go @@ -71,6 +71,7 @@ type ServerConfig struct { Auth AuthConfig `yaml:"auth"` Store StoreConfig `yaml:"store"` ActivityStore StoreConfig `yaml:"activityStore"` + AccessLogStore StoreConfig `yaml:"accessLogStore"` AuthStore StoreConfig `yaml:"authStore"` ReverseProxy ReverseProxyConfig `yaml:"reverseProxy"` diff --git a/combined/cmd/root.go b/combined/cmd/root.go index fedb0e1cc..7f4549643 100644 --- a/combined/cmd/root.go +++ b/combined/cmd/root.go @@ -26,6 +26,7 @@ import ( "github.com/netbirdio/netbird/encryption" agentnetworkpricing "github.com/netbirdio/netbird/management/internals/modules/agentnetwork/pricing" + accesslogsmanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs/manager" mgmtServer "github.com/netbirdio/netbird/management/internals/server" nbconfig "github.com/netbirdio/netbird/management/internals/server/config" "github.com/netbirdio/netbird/management/server/telemetry" @@ -152,6 +153,20 @@ func applyActivityStoreEnv(storeConfig StoreConfig) error { return nil } +// applyAccessLogStoreEnv leaves unset fields to follow the main store, so a +// postgres engine without a DSN reuses the main store DSN. +func applyAccessLogStoreEnv(storeConfig StoreConfig) { + if engine := storeConfig.Engine; engine != "" { + os.Setenv(accesslogsmanager.StoreEngineEnv, strings.ToLower(engine)) + } + if dsn := storeConfig.DSN; dsn != "" { + os.Setenv(accesslogsmanager.PostgresDSNEnv, dsn) + } + if file := storeConfig.File; file != "" { + os.Setenv(accesslogsmanager.SqliteFileEnv, file) + } +} + func initializeConfig() error { var err error config, err = LoadConfig(configPath) @@ -172,6 +187,7 @@ func initializeConfig() error { if err := applyActivityStoreEnv(config.Server.ActivityStore); err != nil { return err } + applyAccessLogStoreEnv(config.Server.AccessLogStore) log.Infof("Starting combined NetBird server") logConfig(config) diff --git a/combined/config.yaml.example b/combined/config.yaml.example index 68dbf701e..7f2d85210 100644 --- a/combined/config.yaml.example +++ b/combined/config.yaml.example @@ -121,6 +121,13 @@ server: # dsn: "" # Connection string for postgres # file: "" # Custom SQLite file path (optional, defaults to {dataDir}/events.db) + # Reverse proxy access log store configuration (optional, follows the store engine: + # sqlite uses its own file in dataDir, postgres reuses the store DSN with a separate pool) + # accessLogStore: + # engine: "postgres" # sqlite or postgres + # dsn: "" # Connection string for a dedicated postgres database + # file: "" # Custom SQLite file path (optional, defaults to {dataDir}/access_logs.db) + # Auth (embedded IdP) store configuration (optional, defaults to sqlite3 in dataDir/idp.db) # authStore: # engine: "sqlite3" # sqlite3 or postgres diff --git a/management/internals/modules/reverseproxy/accesslogs/filter.go b/management/internals/modules/reverseproxy/accesslogs/filter.go index a1fa28312..996828c51 100644 --- a/management/internals/modules/reverseproxy/accesslogs/filter.go +++ b/management/internals/modules/reverseproxy/accesslogs/filter.go @@ -46,7 +46,7 @@ type AccessLogFilter struct { SortOrder string // Sort order: asc or desc (default: desc) // Filtering parameters - Search *string // General search across log ID, host, path, source IP, and user fields + Search *string // General search across log ID, host, path, and source IP SourceIP *string // Filter by source IP address Host *string // Filter by host header Path *string // Filter by request path (supports LIKE pattern) diff --git a/management/internals/modules/reverseproxy/accesslogs/manager/conn.go b/management/internals/modules/reverseproxy/accesslogs/manager/conn.go new file mode 100644 index 000000000..b1ca05d16 --- /dev/null +++ b/management/internals/modules/reverseproxy/accesslogs/manager/conn.go @@ -0,0 +1,92 @@ +package manager + +import ( + "context" + "fmt" + "os" + "strconv" + "strings" + + log "github.com/sirupsen/logrus" + + "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs" + "github.com/netbirdio/netbird/management/internals/shared/db" + "github.com/netbirdio/netbird/management/server/store" +) + +const ( + // StoreEngineEnv selects the access log store engine. Unset follows the main store engine. + StoreEngineEnv = "NB_ACCESS_LOG_STORE_ENGINE" + // PostgresDSNEnv is the access log Postgres DSN. Unset reuses the main store DSN. + PostgresDSNEnv = "NB_ACCESS_LOG_POSTGRES_DSN" + // SqliteFileEnv is the access log SQLite file, resolved against the data directory when relative. + SqliteFileEnv = "NB_ACCESS_LOG_SQLITE_FILE" + // MaxOpenConnsEnv sizes the access log connection pool. Unset applies NB_SQL_MAX_OPEN_CONNS. + MaxOpenConnsEnv = "NB_ACCESS_LOG_SQL_MAX_OPEN_CONNS" + + sqliteFileName = "access_logs.db" +) + +// OpenConn opens and migrates the access log database. Unless configured +// otherwise, it follows mainEngine: SQLite uses its own file in dataDir and +// Postgres reuses the main DSN. +func OpenConn(ctx context.Context, dataDir string, mainEngine db.Engine) (*db.Conn, error) { + engine := mainEngine + if v := os.Getenv(StoreEngineEnv); v != "" { + engine = db.Engine(strings.ToLower(v)) + } + + conn, err := openConn(ctx, dataDir, engine) + if err != nil { + return nil, err + } + if err := conn.AutoMigrate(&accesslogs.AccessLogEntry{}); err != nil { + _ = conn.Close() + return nil, fmt.Errorf("migrate access log store: %w", err) + } + return conn, nil +} + +func openConn(ctx context.Context, dataDir string, engine db.Engine) (*db.Conn, error) { + switch engine { + case db.SqliteStoreEngine: + log.WithContext(ctx).Info("using SQLite access log store engine") + file := sqliteFileName + if v := os.Getenv(SqliteFileEnv); v != "" { + file = v + } + return db.OpenSqliteFile(ctx, dataDir, file) + case db.PostgresStoreEngine: + log.WithContext(ctx).Info("using Postgres access log store engine") + dsn, err := postgresDSN() + if err != nil { + return nil, err + } + return db.OpenPostgresWithoutPool(ctx, dsn, connOptions(ctx)...) + default: + return nil, fmt.Errorf("unsupported access log store engine: %s", engine) + } +} + +func postgresDSN() (string, error) { + if dsn := os.Getenv(PostgresDSNEnv); dsn != "" { + return dsn, nil + } + if dsn, ok := store.LookupPostgresDSN(); ok && dsn != "" { + return dsn, nil + } + return "", fmt.Errorf("neither %s nor %s is set", PostgresDSNEnv, store.PostgresDsnEnv) +} + +func connOptions(ctx context.Context) []db.Option { + val := os.Getenv(MaxOpenConnsEnv) + if val == "" { + return nil + } + conns, err := strconv.Atoi(val) + if err != nil || conns <= 0 { + log.WithContext(ctx).Warnf("ignoring invalid %s value %q", MaxOpenConnsEnv, val) + return nil + } + return []db.Option{db.WithMaxOpenConns(conns)} +} diff --git a/management/internals/modules/reverseproxy/accesslogs/manager/conn_test.go b/management/internals/modules/reverseproxy/accesslogs/manager/conn_test.go new file mode 100644 index 000000000..90fdf8ffa --- /dev/null +++ b/management/internals/modules/reverseproxy/accesslogs/manager/conn_test.go @@ -0,0 +1,142 @@ +package manager + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gorm.io/driver/sqlite" + "gorm.io/gorm" + + "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs" + "github.com/netbirdio/netbird/management/internals/shared/db" + "github.com/netbirdio/netbird/management/server/store" + "github.com/netbirdio/netbird/management/server/testutil" +) + +func unsetStoreEnv(t *testing.T) { + t.Helper() + for _, key := range []string{StoreEngineEnv, PostgresDSNEnv, SqliteFileEnv, MaxOpenConnsEnv, store.PostgresDsnEnv, store.PostgresDsnEnvLegacy} { + t.Setenv(key, "") + require.NoError(t, os.Unsetenv(key)) + } +} + +func openTestConn(t *testing.T, dataDir string, mainEngine db.Engine) *db.Conn { + t.Helper() + conn, err := OpenConn(context.Background(), dataDir, mainEngine) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + return conn +} + +func assertStoresEntries(t *testing.T, conn *db.Conn) { + t.Helper() + ctx := context.Background() + repo := NewRepository(conn) + require.NoError(t, repo.Create(ctx, newEntry("entry", "acc", "GET", time.Minute))) + + logs, total, err := repo.ListByAccount(ctx, db.LockingStrengthNone, "acc", accesslogs.AccessLogFilter{Page: 1, PageSize: 10}) + require.NoError(t, err) + assert.EqualValues(t, 1, total, "the stored entry should be listed") + require.Len(t, logs, 1) + assert.Equal(t, "entry", logs[0].ID) +} + +func TestOpenConn_SqliteUsesItsOwnFile(t *testing.T) { + unsetStoreEnv(t) + dataDir := t.TempDir() + + conn := openTestConn(t, dataDir, db.SqliteStoreEngine) + + assert.Equal(t, db.SqliteStoreEngine, conn.Engine()) + assert.FileExists(t, filepath.Join(dataDir, sqliteFileName)) + assert.NoFileExists(t, filepath.Join(dataDir, db.SqliteFileName), "the main store file must not be opened") + assertStoresEntries(t, conn) +} + +func TestOpenConn_SqliteFileFromEnv(t *testing.T) { + unsetStoreEnv(t) + dataDir := t.TempDir() + t.Setenv(SqliteFileEnv, "custom_logs.db") + + conn := openTestConn(t, dataDir, db.SqliteStoreEngine) + + assert.FileExists(t, filepath.Join(dataDir, "custom_logs.db")) + assert.NoFileExists(t, filepath.Join(dataDir, sqliteFileName)) + assertStoresEntries(t, conn) +} + +func TestOpenConn_EngineFromEnvOverridesMainEngine(t *testing.T) { + unsetStoreEnv(t) + t.Setenv(StoreEngineEnv, "SQLite") + + conn := openTestConn(t, t.TempDir(), db.PostgresStoreEngine) + + assert.Equal(t, db.SqliteStoreEngine, conn.Engine(), "the configured engine should win over the main store engine") +} + +func TestOpenConn_PostgresWithoutDSNFails(t *testing.T) { + unsetStoreEnv(t) + + _, err := OpenConn(context.Background(), t.TempDir(), db.PostgresStoreEngine) + + require.Error(t, err) + assert.Contains(t, err.Error(), PostgresDSNEnv) +} + +func TestOpenConn_UnsupportedEngineFails(t *testing.T) { + unsetStoreEnv(t) + t.Setenv(StoreEngineEnv, "mysql") + + _, err := OpenConn(context.Background(), t.TempDir(), db.SqliteStoreEngine) + + require.Error(t, err) + assert.Contains(t, err.Error(), "mysql") +} + +func TestOpenConn_PostgresReusesMainDSNWithItsOwnPool(t *testing.T) { + if os.Getenv("NETBIRD_STORE_ENGINE") != string(db.PostgresStoreEngine) { + t.Skip("requires NETBIRD_STORE_ENGINE=postgres") + } + unsetStoreEnv(t) + cleanup, dsn, err := testutil.CreatePostgresTestContainer() + require.NoError(t, err) + t.Cleanup(cleanup) + t.Setenv(store.PostgresDsnEnv, dsn) + t.Setenv(MaxOpenConnsEnv, "3") + + conn := openTestConn(t, t.TempDir(), db.PostgresStoreEngine) + + assert.Equal(t, db.PostgresStoreEngine, conn.Engine()) + assert.Nil(t, conn.Pool(nil), "access logs never take the pgx read paths, so no pgx pool should be opened") + sqlDB, err := conn.DB(nil).DB() + require.NoError(t, err) + assert.Equal(t, 3, sqlDB.Stats().MaxOpenConnections, "the pool should be sized by %s", MaxOpenConnsEnv) + assertStoresEntries(t, conn) +} + +func TestConnOptions_SizesPoolFromEnv(t *testing.T) { + // Invalid values fall back to the store-wide setting instead of failing boot. + cases := map[string]int{"3": 3, "": 7, "zero": 7, "0": 7, "-1": 7} + for value, want := range cases { + t.Run(value, func(t *testing.T) { + t.Setenv("NB_SQL_MAX_OPEN_CONNS", "7") + t.Setenv(MaxOpenConnsEnv, value) + + gormDB, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "pool.db")), db.GormConfig()) + require.NoError(t, err) + conn, err := db.NewConn(context.Background(), gormDB, db.PostgresStoreEngine, nil, connOptions(context.Background())...) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + + sqlDB, err := conn.DB(nil).DB() + require.NoError(t, err) + assert.Equal(t, want, sqlDB.Stats().MaxOpenConnections, "max open connections for %s=%q", MaxOpenConnsEnv, value) + }) + } +} diff --git a/management/internals/modules/reverseproxy/accesslogs/manager/repository.go b/management/internals/modules/reverseproxy/accesslogs/manager/repository.go index 105696ab7..3967442d8 100644 --- a/management/internals/modules/reverseproxy/accesslogs/manager/repository.go +++ b/management/internals/modules/reverseproxy/accesslogs/manager/repository.go @@ -85,8 +85,8 @@ func applyFilters(query *gorm.DB, filter accesslogs.AccessLogFilter) *gorm.DB { if filter.Search != nil { searchPattern := "%" + *filter.Search + "%" query = query.Where( - "id LIKE ? OR location_connection_ip LIKE ? OR host LIKE ? OR path LIKE ? OR CONCAT(host, path) LIKE ? OR user_id IN (SELECT id FROM users WHERE email LIKE ? OR name LIKE ?)", - searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, + "id LIKE ? OR location_connection_ip LIKE ? OR host LIKE ? OR path LIKE ? OR CONCAT(host, path) LIKE ?", + searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, ) } diff --git a/management/internals/modules/reverseproxy/accesslogs/manager/repository_test.go b/management/internals/modules/reverseproxy/accesslogs/manager/repository_test.go index 9931cc27c..7fea12a85 100644 --- a/management/internals/modules/reverseproxy/accesslogs/manager/repository_test.go +++ b/management/internals/modules/reverseproxy/accesslogs/manager/repository_test.go @@ -123,3 +123,37 @@ func TestSqlRepository_ListByAccount_StatusFilter(t *testing.T) { assert.Equal(t, []string{"info", "notfound"}, listIDs("failed")) assert.Equal(t, []string{"ok"}, listIDs("success")) } + +func TestSqlRepository_ListByAccount_Search(t *testing.T) { + repo, _ := newTestRepository(t) + ctx := context.Background() + api := newEntry("req-api", "acc", "GET", time.Hour) + api.Host = "api.example.com" + api.Path = "/v1/peers" + web := newEntry("req-web", "acc", "GET", 2*time.Hour) + web.Host = "web.example.com" + web.Path = "/login" + foreign := newEntry("foreign", "other", "GET", time.Hour) + foreign.Host = "api.example.com" + for _, entry := range []*accesslogs.AccessLogEntry{api, web, foreign} { + require.NoError(t, repo.Create(ctx, entry)) + } + + search := func(term string) []string { + logs, total, err := repo.ListByAccount(ctx, db.LockingStrengthNone, "acc", accesslogs.AccessLogFilter{Page: 1, PageSize: 10, Search: &term}) + require.NoError(t, err) + require.EqualValues(t, len(logs), total) + ids := make([]string, 0, len(logs)) + for _, entry := range logs { + ids = append(ids, entry.ID) + } + return ids + } + + assert.Equal(t, []string{"req-api"}, search("api.example"), "search should match the host") + assert.Equal(t, []string{"req-web"}, search("/login"), "search should match the path") + assert.Equal(t, []string{"req-api"}, search("example.com/v1"), "search should match host and path joined") + assert.Equal(t, []string{"req-web"}, search("req-web"), "search should match the log ID") + assert.Equal(t, []string{"req-api", "req-web"}, search("example.com"), "search should list every match newest first") + assert.Empty(t, search("nothing-matches")) +} diff --git a/management/internals/server/boot.go b/management/internals/server/boot.go index e466005fe..28f6294d6 100644 --- a/management/internals/server/boot.go +++ b/management/internals/server/boot.go @@ -180,7 +180,8 @@ func (s *BaseServer) APIHandler() http.Handler { s.GroupsManager(), s.GeoLocationManager(), s.PermissionsManager(), s.SettingsManager(), s.ZonesManager(), s.RecordsManager(), s.NetworkMapController(), s.IdpManager(), s.ServiceManager(), s.ReverseProxyDomainManager(), s.AccessLogsManager(), s.ReverseProxyGRPCServer(), s.Config.ReverseProxy.TrustedHTTPProxies, - s.AgentNetworkManager(), nil) + s.AgentNetworkManager(), nil, + ) if err != nil { log.Fatalf("failed to create API handler: %v", err) } @@ -360,14 +361,30 @@ func (s *BaseServer) ProxyActivityManager() proxyactivity.Manager { }) } +func (s *BaseServer) AccessLogDBConn() *db.Conn { + return CreateNamed(s, "accesslogs", func() *db.Conn { + conn, err := accesslogsmanager.OpenConn(context.Background(), s.Config.Datadir, s.DBConn().Engine()) + if err != nil { + log.Fatalf("failed to open access log database connection: %v", err) + } + s.OnStop(func() { + if err := conn.Close(); err != nil { + log.Debugf("failed to close access log database connection: %v", err) + } + }) + return conn + }) +} + func (s *BaseServer) AccessLogsManager() accesslogs.Manager { return Create(s, func() accesslogs.Manager { - accessLogManager := accesslogsmanager.NewManager(accesslogsmanager.NewRepository(s.DBConn()), s.Store(), s.PermissionsManager(), s.GeoLocationManager()) + accessLogManager := accesslogsmanager.NewManager(accesslogsmanager.NewRepository(s.AccessLogDBConn()), s.Store(), s.PermissionsManager(), s.GeoLocationManager()) accessLogManager.StartPeriodicCleanup( context.Background(), s.Config.ReverseProxy.AccessLogRetentionDays, s.Config.ReverseProxy.AccessLogCleanupIntervalHours, ) + s.OnStop(accessLogManager.StopPeriodicCleanup) return accessLogManager }) } diff --git a/management/internals/shared/db/conn.go b/management/internals/shared/db/conn.go index 8edaa1e4e..b21b066e2 100644 --- a/management/internals/shared/db/conn.go +++ b/management/internals/shared/db/conn.go @@ -24,8 +24,8 @@ type TxMetrics interface { CountTransactionDuration(duration time.Duration) } -// Conn is the database connection shared by all repositories: one gorm handle, -// the pgx pool of a Postgres deployment and the engine they talk to. +// Conn is a database connection the repositories run on: one gorm handle, the +// pgx pool of a Postgres deployment and the engine they talk to. type Conn struct { db *gorm.DB pool *pgxpool.Pool @@ -34,15 +34,36 @@ type Conn struct { metrics TxMetrics } +// Option tunes a Conn as it is opened. +type Option func(*connOptions) + +type connOptions struct { + maxOpenConns int +} + +// WithMaxOpenConns sizes the gorm connection pool of this Conn, taking +// precedence over NB_SQL_MAX_OPEN_CONNS. A pgx pool keeps its PoolConfig size, +// and SQLite always uses one connection. +func WithMaxOpenConns(n int) Option { + return func(o *connOptions) { + o.maxOpenConns = n + } +} + // NewConn takes ownership of an open gorm handle and pool once it returns // without error, applying the connection limits and transaction timeout -// configured through the environment. -func NewConn(ctx context.Context, gormDB *gorm.DB, engine Engine, pool *pgxpool.Pool) (*Conn, error) { +// configured through the environment and opts. +func NewConn(ctx context.Context, gormDB *gorm.DB, engine Engine, pool *pgxpool.Pool, opts ...Option) (*Conn, error) { sqlDB, err := gormDB.DB() if err != nil { return nil, err } + var options connOptions + for _, opt := range opts { + opt(&options) + } + txTimeout := defaultTransactionTimeout if v := os.Getenv("NB_STORE_TRANSACTION_TIMEOUT"); v != "" { if parsed, err := time.ParseDuration(v); err == nil { @@ -57,6 +78,9 @@ func NewConn(ctx context.Context, gormDB *gorm.DB, engine Engine, pool *pgxpool. if connsConfigured { conns = configuredConns } + if options.maxOpenConns > 0 { + conns = options.maxOpenConns + } if engine == SqliteStoreEngine { if connsConfigured { log.WithContext(ctx).Warnf("setting NB_SQL_MAX_OPEN_CONNS is not supported for sqlite, using default value 1") diff --git a/management/internals/shared/db/conn_test.go b/management/internals/shared/db/conn_test.go index 4f1a234b2..cd7488422 100644 --- a/management/internals/shared/db/conn_test.go +++ b/management/internals/shared/db/conn_test.go @@ -146,3 +146,22 @@ func TestNewConn_MaxOpenConnsFromEnv(t *testing.T) { require.NoError(t, err) assert.Equal(t, 1, sqliteDB.Stats().MaxOpenConnections) } + +func TestNewConn_MaxOpenConnsOptionOverridesEnv(t *testing.T) { + t.Setenv("NB_SQL_MAX_OPEN_CONNS", "7") + + open := func(engine Engine, opts ...Option) int { + gormDB, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "store.db")), GormConfig()) + require.NoError(t, err) + conn, err := NewConn(context.Background(), gormDB, engine, nil, opts...) + require.NoError(t, err) + t.Cleanup(func() { _ = conn.Close() }) + sqlDB, err := conn.DB(nil).DB() + require.NoError(t, err) + return sqlDB.Stats().MaxOpenConnections + } + + assert.Equal(t, 3, open(PostgresStoreEngine, WithMaxOpenConns(3)), "option should take precedence over the environment") + assert.Equal(t, 7, open(PostgresStoreEngine, WithMaxOpenConns(0)), "a non-positive option should fall back to the environment") + assert.Equal(t, 1, open(SqliteStoreEngine, WithMaxOpenConns(3)), "sqlite should stay on a single connection") +} diff --git a/management/internals/shared/db/open.go b/management/internals/shared/db/open.go index c573e8c08..3c3299bb9 100644 --- a/management/internals/shared/db/open.go +++ b/management/internals/shared/db/open.go @@ -101,7 +101,7 @@ func OpenSqliteFile(ctx context.Context, dataDir, storeFile string) (*Conn, erro } // OpenPostgres opens a Postgres database through gorm and a pgx pool sized by pool. -func OpenPostgres(ctx context.Context, dsn string, pool PoolConfig) (*Conn, error) { +func OpenPostgres(ctx context.Context, dsn string, pool PoolConfig, opts ...Option) (*Conn, error) { gormDB, err := gorm.Open(postgres.Open(dsn), GormConfig()) if err != nil { return nil, err @@ -111,7 +111,7 @@ func OpenPostgres(ctx context.Context, dsn string, pool PoolConfig) (*Conn, erro closeGorm(gormDB) return nil, err } - conn, err := NewConn(ctx, gormDB, PostgresStoreEngine, pgxPool) + conn, err := NewConn(ctx, gormDB, PostgresStoreEngine, pgxPool, opts...) if err != nil { pgxPool.Close() closeGorm(gormDB) @@ -120,6 +120,21 @@ func OpenPostgres(ctx context.Context, dsn string, pool PoolConfig) (*Conn, erro return conn, nil } +// OpenPostgresWithoutPool opens a Postgres database through gorm alone, for +// repositories that never take the pgx read paths. +func OpenPostgresWithoutPool(ctx context.Context, dsn string, opts ...Option) (*Conn, error) { + gormDB, err := gorm.Open(postgres.Open(dsn), GormConfig()) + if err != nil { + return nil, err + } + conn, err := NewConn(ctx, gormDB, PostgresStoreEngine, nil, opts...) + if err != nil { + closeGorm(gormDB) + return nil, err + } + return conn, nil +} + func newPgxPool(ctx context.Context, dsn string, cfg PoolConfig) (*pgxpool.Pool, error) { config, err := pgxpool.ParseConfig(dsn) if err != nil { diff --git a/management/server/http/testing/testing_tools/channel/channel.go b/management/server/http/testing/testing_tools/channel/channel.go index 88ab70b9c..326dfb9a0 100644 --- a/management/server/http/testing/testing_tools/channel/channel.go +++ b/management/server/http/testing/testing_tools/channel/channel.go @@ -14,10 +14,12 @@ import ( "github.com/netbirdio/management-integrations/integrations" + "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs" accesslogsmanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs/manager" "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/domain/manager" proxymanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy/manager" reverseproxymanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service/manager" + "github.com/netbirdio/netbird/management/internals/shared/db" nbgrpc "github.com/netbirdio/netbird/management/internals/shared/grpc" zonesManager "github.com/netbirdio/netbird/management/internals/modules/zones/manager" @@ -107,7 +109,7 @@ func BuildApiBlackBoxWithDBState(t testing_tools.TB, sqlFile string, expectedPee t.Fatalf("Failed to create manager: %v", err) } - accessLogsManager := accesslogsmanager.NewManager(accesslogsmanager.NewRepository(store.(*nbstore.SqlStore).Conn()), store, permissionsManager, nil) + accessLogsManager := accesslogsmanager.NewManager(newAccessLogsRepository(t), store, permissionsManager, nil) proxyTokenStore := nbgrpc.NewOneTimeTokenStore(ctx, cacheStore) singleUseStore := nbgrpc.NewSingleUseStore(ctx, cacheStore) noopMeter := noop.NewMeterProvider().Meter("") @@ -249,7 +251,7 @@ func BuildApiBlackBoxWithDBStateAndPeerChannel(t testing_tools.TB, sqlFile strin t.Fatalf("Failed to create manager: %v", err) } - accessLogsManager := accesslogsmanager.NewManager(accesslogsmanager.NewRepository(store.(*nbstore.SqlStore).Conn()), store, permissionsManager, nil) + accessLogsManager := accesslogsmanager.NewManager(newAccessLogsRepository(t), store, permissionsManager, nil) proxyTokenStore := nbgrpc.NewOneTimeTokenStore(ctx, cacheStore) singleUseStore := nbgrpc.NewSingleUseStore(ctx, cacheStore) noopMeter := noop.NewMeterProvider().Meter("") @@ -299,6 +301,16 @@ func BuildApiBlackBoxWithDBStateAndPeerChannel(t testing_tools.TB, sqlFile strin return apiHandler, am, updMsg } +func newAccessLogsRepository(t testing_tools.TB) accesslogs.Repository { + t.Helper() + conn, err := accesslogsmanager.OpenConn(context.Background(), t.TempDir(), db.SqliteStoreEngine) + if err != nil { + t.Fatalf("Failed to open access log store: %v", err) + } + t.Cleanup(func() { _ = conn.Close() }) + return accesslogsmanager.NewRepository(conn) +} + func mockValidateAndParseToken(_ context.Context, token string) (auth.UserAuth, *jwt.Token, error) { userAuth := auth.UserAuth{} diff --git a/management/server/store/sql_store.go b/management/server/store/sql_store.go index 1734129c3..dc1edca88 100644 --- a/management/server/store/sql_store.go +++ b/management/server/store/sql_store.go @@ -12,7 +12,6 @@ import ( nbdns "github.com/netbirdio/netbird/dns" agentNetworkTypes "github.com/netbirdio/netbird/management/internals/modules/agentnetwork/types" - "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs" "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/domain" "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy" rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service" @@ -83,7 +82,7 @@ func NewSqlStore(ctx context.Context, conn *db.Conn, metrics telemetry.AppMetric &installation{}, &types.ExtraSettings{}, &posture.Checks{}, &nbpeer.NetworkAddress{}, &networkTypes.Network{}, &routerTypes.NetworkRouter{}, &resourceTypes.NetworkResource{}, &types.AccountOnboarding{}, &types.Job{}, &zones.Zone{}, &records.Record{}, &types.UserInviteRecord{}, &rpservice.Service{}, &rpservice.Target{}, &domain.Domain{}, - &accesslogs.AccessLogEntry{}, &proxy.Proxy{}, + &proxy.Proxy{}, &agentNetworkTypes.Provider{}, &agentNetworkTypes.Policy{}, &agentNetworkTypes.Guardrail{}, &agentNetworkTypes.Settings{}, &agentNetworkTypes.Consumption{}, &agentNetworkTypes.AccountBudgetRule{}, &agentNetworkTypes.AgentNetworkAccessLog{}, &agentNetworkTypes.AgentNetworkAccessLogGroup{}, diff --git a/management/server/store/store.go b/management/server/store/store.go index 6d20c1e2b..670bd3956 100644 --- a/management/server/store/store.go +++ b/management/server/store/store.go @@ -460,6 +460,11 @@ func lookupDSNEnv(nbKey, legacyKey string) (string, bool) { return os.LookupEnv(legacyKey) } +// LookupPostgresDSN returns the DSN the main store opens a Postgres engine with. +func LookupPostgresDSN() (string, bool) { + return lookupDSNEnv(PostgresDsnEnv, PostgresDsnEnvLegacy) +} + var supportedEngines = []types.Engine{types.SqliteStoreEngine, types.PostgresStoreEngine} func getStoreEngineFromEnv() types.Engine { @@ -530,7 +535,7 @@ func OpenConn(ctx context.Context, kind types.Engine, dataDir string) (*db.Conn, return db.OpenSqlite(ctx, dataDir) case types.PostgresStoreEngine: log.WithContext(ctx).Info("using Postgres store engine") - dsn, ok := lookupDSNEnv(PostgresDsnEnv, PostgresDsnEnvLegacy) + dsn, ok := LookupPostgresDSN() if !ok { return nil, fmt.Errorf("%s is not set", PostgresDsnEnv) } diff --git a/shared/management/http/api/openapi.yml b/shared/management/http/api/openapi.yml index f129ada26..e64c67ce0 100644 --- a/shared/management/http/api/openapi.yml +++ b/shared/management/http/api/openapi.yml @@ -9381,7 +9381,7 @@ paths: name: search schema: type: string - description: General search across request ID, host, path, source IP, user email, and user name + description: General search across request ID, host, path, and source IP - in: query name: source_ip schema: diff --git a/shared/management/http/api/types.gen.go b/shared/management/http/api/types.gen.go index a5ffe4145..6a7450153 100644 --- a/shared/management/http/api/types.gen.go +++ b/shared/management/http/api/types.gen.go @@ -6051,7 +6051,7 @@ type GetApiEventsProxyParams struct { // SortOrder Sort order (ascending or descending) SortOrder *GetApiEventsProxyParamsSortOrder `form:"sort_order,omitempty" json:"sort_order,omitempty"` - // Search General search across request ID, host, path, source IP, user email, and user name + // Search General search across request ID, host, path, and source IP Search *string `form:"search,omitempty" json:"search,omitempty"` // SourceIp Filter by source IP address From c051ca8f8767617091ce8fdb23edccd83487cba0 Mon Sep 17 00:00:00 2001 From: Maycon Santos Date: Sat, 24 Oct 2026 01:07:20 +0200 Subject: [PATCH 2/5] [client,proxy,management,signal,relay] Distrust compromised eMudhra CA keys (#8326) * [client,proxy,management,signal,relay] Distrust compromised eMudhra CA keys Eight eMudhra intermediate CA private keys were compromised and browsers blocked them on 2026-10-17. Our binaries were built with go1.26.7, which accepts chains through those intermediates whenever the trust store still holds emSign ECC Root CA - G3. go1.26.10 rejects them in crypto/x509 on every platform, regardless of the trust store. The UBI base still ships the root and Red Hat has no fixed package yet, so the UBI images blocklist it with p11-kit. The ingest image, the UI server image and the pinned Chainguard digests also carried the root and move to bases that no longer do. * [client,proxy,management,signal,relay] Use the fixed UBI base instead of a blocklist Red Hat shipped ca-certificates 2026.2.92_v9.0.319-90.3 on 2026-10-22, which drops emSign ECC Root CA - G3 and AAA Certificate Services. The current ubi9-minimal image carries it, so pin that digest and remove the p11-kit blocklist step. * [misc] Pin upload-server images by digest only and restore go.mod comment --- .devcontainer/Dockerfile | 2 +- .github/workflows/golang-test-linux.yml | 2 +- client/Dockerfile-rootless.ubi | 2 +- client/internal/metrics/infra/ingest/Dockerfile | 4 ++-- client/testutil/privileged/runner_test.go | 2 +- client/ui/build/docker/Dockerfile.cross | 2 +- client/ui/build/docker/Dockerfile.server | 4 ++-- combined/Dockerfile.multistage | 2 +- combined/Dockerfile.ubi | 2 +- docs/testing-privileged.md | 2 +- e2e/harness/Dockerfile.client | 2 +- e2e/harness/Dockerfile.nblink | 2 +- go.mod | 2 +- management/Dockerfile.multistage | 2 +- management/Dockerfile.ubi | 2 +- proxy/Dockerfile | 2 +- proxy/Dockerfile.multistage | 2 +- proxy/Dockerfile.ubi | 2 +- relay/Dockerfile.ubi | 2 +- signal/Dockerfile.ubi | 2 +- upload-server/Dockerfile | 11 ++++++----- 21 files changed, 28 insertions(+), 27 deletions(-) diff --git a/.devcontainer/Dockerfile b/.devcontainer/Dockerfile index 9b6a0edfd..dc7475a06 100644 --- a/.devcontainer/Dockerfile +++ b/.devcontainer/Dockerfile @@ -1,4 +1,4 @@ -FROM golang:1.26.7-bookworm +FROM golang:1.26.10-bookworm RUN apt-get update && export DEBIAN_FRONTEND=noninteractive \ && apt-get -y install --no-install-recommends\ diff --git a/.github/workflows/golang-test-linux.yml b/.github/workflows/golang-test-linux.yml index b3e4dd4f7..eea69eaa7 100644 --- a/.github/workflows/golang-test-linux.yml +++ b/.github/workflows/golang-test-linux.yml @@ -303,7 +303,7 @@ jobs: -e GOCACHE=${CONTAINER_GOCACHE} \ -e GOMODCACHE=${CONTAINER_GOMODCACHE} \ -e CONTAINER=${CONTAINER} \ - golang:1.26.7-alpine \ + golang:1.26.10-alpine \ sh -c ' \ apk update; apk add --no-cache \ ca-certificates iptables ip6tables dbus dbus-dev libpcap-dev build-base; \ diff --git a/client/Dockerfile-rootless.ubi b/client/Dockerfile-rootless.ubi index 4701728c1..396e0d562 100644 --- a/client/Dockerfile-rootless.ubi +++ b/client/Dockerfile-rootless.ubi @@ -1,4 +1,4 @@ -FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:7fbeae18dc9476399f565e68255f602a3374ea8614ba3d14843565131a13ff93 +FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:2d4c4551af6831dcd26bbf86108c4385de5183ec8fe6b5868f6f9640fe47f4ed ARG TARGETPLATFORM ARG NETBIRD_BINARY=$TARGETPLATFORM/netbird diff --git a/client/internal/metrics/infra/ingest/Dockerfile b/client/internal/metrics/infra/ingest/Dockerfile index 3620c524b..1b4862a7d 100644 --- a/client/internal/metrics/infra/ingest/Dockerfile +++ b/client/internal/metrics/infra/ingest/Dockerfile @@ -1,9 +1,9 @@ -FROM golang:1.25-alpine AS build +FROM golang:1.26.10-alpine AS build WORKDIR /app COPY go.mod main.go ./ RUN CGO_ENABLED=0 go build -o ingest . -FROM alpine:3.20 +FROM alpine:3.24 RUN adduser -D -H ingest COPY --from=build /app/ingest /usr/local/bin/ingest USER ingest diff --git a/client/testutil/privileged/runner_test.go b/client/testutil/privileged/runner_test.go index 157005d3e..c7bde1316 100644 --- a/client/testutil/privileged/runner_test.go +++ b/client/testutil/privileged/runner_test.go @@ -25,7 +25,7 @@ import ( // (.github/workflows/golang-test-linux.yml, test_client_on_docker). const ( containerImage = "golang" - containerTag = "1.26.7-alpine" + containerTag = "1.26.10-alpine" ) const ( diff --git a/client/ui/build/docker/Dockerfile.cross b/client/ui/build/docker/Dockerfile.cross index 55c0d69e1..650087fb3 100644 --- a/client/ui/build/docker/Dockerfile.cross +++ b/client/ui/build/docker/Dockerfile.cross @@ -13,7 +13,7 @@ # docker run --rm -v $(pwd):/app wails-cross windows amd64 # docker run --rm -v $(pwd):/app wails-cross windows arm64 -FROM golang:1.26.7-bookworm +FROM golang:1.26.10-bookworm ARG TARGETARCH diff --git a/client/ui/build/docker/Dockerfile.server b/client/ui/build/docker/Dockerfile.server index 57183f1d2..7a7f36575 100644 --- a/client/ui/build/docker/Dockerfile.server +++ b/client/ui/build/docker/Dockerfile.server @@ -2,7 +2,7 @@ # Multi-stage build for minimal image size # Build stage -FROM golang:1.26.7-alpine AS builder +FROM golang:1.26.10-alpine AS builder WORKDIR /app @@ -22,7 +22,7 @@ RUN go mod tidy RUN go build -tags server -ldflags="-s -w" -o server . # Runtime stage - minimal image -FROM gcr.io/distroless/static-debian12 +FROM gcr.io/distroless/static-debian13 # Copy the binary COPY --from=builder /app/server /server diff --git a/combined/Dockerfile.multistage b/combined/Dockerfile.multistage index 011379c2f..0551516fd 100644 --- a/combined/Dockerfile.multistage +++ b/combined/Dockerfile.multistage @@ -1,4 +1,4 @@ -FROM golang:1.26.7-bookworm AS builder +FROM golang:1.26.10-bookworm AS builder WORKDIR /app # Install build dependencies diff --git a/combined/Dockerfile.ubi b/combined/Dockerfile.ubi index 66ef55a34..dbbb920c9 100644 --- a/combined/Dockerfile.ubi +++ b/combined/Dockerfile.ubi @@ -1,4 +1,4 @@ -FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:7fbeae18dc9476399f565e68255f602a3374ea8614ba3d14843565131a13ff93 +FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:2d4c4551af6831dcd26bbf86108c4385de5183ec8fe6b5868f6f9640fe47f4ed ARG TARGETPLATFORM ARG VERSION=dev diff --git a/docs/testing-privileged.md b/docs/testing-privileged.md index 939770de9..7d9552c4a 100644 --- a/docs/testing-privileged.md +++ b/docs/testing-privileged.md @@ -32,7 +32,7 @@ list; both are optional and default to the full privileged suite. 1. Skips immediately when it detects it is already inside the container (`DOCKER_CI=true`), so the privileged tests run in place instead of recursing. -2. Otherwise spins up a `golang:1.26.7-alpine` container (matching CI), +2. Otherwise spins up a `golang:1.26.10-alpine` container (matching CI), bind-mounts the repo and the host Go build/module caches, installs the required packages, and runs `go test -tags 'devcert privileged'` over the client packages. diff --git a/e2e/harness/Dockerfile.client b/e2e/harness/Dockerfile.client index 4c76b95c6..4bcb2b079 100644 --- a/e2e/harness/Dockerfile.client +++ b/e2e/harness/Dockerfile.client @@ -3,7 +3,7 @@ # artifact), so this mirrors its alpine runtime + entrypoint while compiling the # CGO-free client inline. BuildKit cache mounts keep rebuilds incremental. -FROM golang:1.26.7-bookworm AS builder +FROM golang:1.26.10-bookworm AS builder WORKDIR /src COPY go.mod go.sum ./ RUN --mount=type=cache,target=/go/pkg/mod go mod download diff --git a/e2e/harness/Dockerfile.nblink b/e2e/harness/Dockerfile.nblink index dac3aceec..0628aa083 100644 --- a/e2e/harness/Dockerfile.nblink +++ b/e2e/harness/Dockerfile.nblink @@ -8,7 +8,7 @@ # that quietly hands it NET_ADMIN would make the suite prove the opposite of # what it is for. -FROM golang:1.26.7-bookworm AS builder +FROM golang:1.26.10-bookworm AS builder WORKDIR /src COPY go.mod go.sum ./ RUN --mount=type=cache,target=/go/pkg/mod go mod download diff --git a/go.mod b/go.mod index f93141d5c..0d1492731 100644 --- a/go.mod +++ b/go.mod @@ -4,7 +4,7 @@ go 1.26.0 // Pin the toolchain to a patch release >= go1.26.2 // See https://go.dev/issue/77875. -toolchain go1.26.7 +toolchain go1.26.10 require ( cunicu.li/go-rosenpass v0.5.42 diff --git a/management/Dockerfile.multistage b/management/Dockerfile.multistage index 5d037f1b1..1e9d11227 100644 --- a/management/Dockerfile.multistage +++ b/management/Dockerfile.multistage @@ -1,4 +1,4 @@ -FROM golang:1.26.7-bookworm AS builder +FROM golang:1.26.10-bookworm AS builder WORKDIR /app # Install build dependencies diff --git a/management/Dockerfile.ubi b/management/Dockerfile.ubi index 628d1a086..163302aa9 100644 --- a/management/Dockerfile.ubi +++ b/management/Dockerfile.ubi @@ -1,4 +1,4 @@ -FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:7fbeae18dc9476399f565e68255f602a3374ea8614ba3d14843565131a13ff93 +FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:2d4c4551af6831dcd26bbf86108c4385de5183ec8fe6b5868f6f9640fe47f4ed ARG TARGETPLATFORM ARG VERSION=dev diff --git a/proxy/Dockerfile b/proxy/Dockerfile index 5944d944e..e4f9a0239 100644 --- a/proxy/Dockerfile +++ b/proxy/Dockerfile @@ -1,4 +1,4 @@ -FROM golang:1.26.7-alpine AS builder +FROM golang:1.26.10-alpine AS builder WORKDIR /app RUN echo "netbird:x:1000:1000:netbird:/var/lib/netbird:/sbin/nologin" > /tmp/passwd && \ diff --git a/proxy/Dockerfile.multistage b/proxy/Dockerfile.multistage index d1db32296..ff46280d2 100644 --- a/proxy/Dockerfile.multistage +++ b/proxy/Dockerfile.multistage @@ -1,4 +1,4 @@ -FROM golang:1.26.7-alpine AS builder +FROM golang:1.26.10-alpine AS builder WORKDIR /app COPY go.mod go.sum ./ diff --git a/proxy/Dockerfile.ubi b/proxy/Dockerfile.ubi index a74280a49..013eb423d 100644 --- a/proxy/Dockerfile.ubi +++ b/proxy/Dockerfile.ubi @@ -1,4 +1,4 @@ -FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:7fbeae18dc9476399f565e68255f602a3374ea8614ba3d14843565131a13ff93 +FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:2d4c4551af6831dcd26bbf86108c4385de5183ec8fe6b5868f6f9640fe47f4ed ARG TARGETPLATFORM ARG VERSION=dev diff --git a/relay/Dockerfile.ubi b/relay/Dockerfile.ubi index 86b0cbc01..b4987a424 100644 --- a/relay/Dockerfile.ubi +++ b/relay/Dockerfile.ubi @@ -1,4 +1,4 @@ -FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:7fbeae18dc9476399f565e68255f602a3374ea8614ba3d14843565131a13ff93 +FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:2d4c4551af6831dcd26bbf86108c4385de5183ec8fe6b5868f6f9640fe47f4ed ARG TARGETPLATFORM ARG VERSION=dev diff --git a/signal/Dockerfile.ubi b/signal/Dockerfile.ubi index f0a6552da..5b4036469 100644 --- a/signal/Dockerfile.ubi +++ b/signal/Dockerfile.ubi @@ -1,4 +1,4 @@ -FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:7fbeae18dc9476399f565e68255f602a3374ea8614ba3d14843565131a13ff93 +FROM registry.access.redhat.com/ubi9/ubi-minimal@sha256:2d4c4551af6831dcd26bbf86108c4385de5183ec8fe6b5868f6f9640fe47f4ed ARG TARGETPLATFORM ARG VERSION=dev diff --git a/upload-server/Dockerfile b/upload-server/Dockerfile index 8098a0186..16fd79871 100644 --- a/upload-server/Dockerfile +++ b/upload-server/Dockerfile @@ -16,13 +16,14 @@ # VARIANT=release|debug debug swaps the base for Chainguard busybox (a shell) # VERSION stamped into the binary the same way goreleaser does # -# Chainguard publishes only :latest for free, so the bases are pinned by digest -# and moved by Dependabot. +# Images are pinned by digest alone. The builder is golang:1.26.10-bookworm and +# the bases are Chainguard's :latest, the only free tag; bump them by hand with +# docker buildx imagetools inspect . ARG VARIANT=release # Pure Go: cross-compile from the build host instead of emulating the target. -FROM --platform=$BUILDPLATFORM golang:1.26.7-bookworm@sha256:e8c859f5632dcfde7b32d2012b4351728f6437930887c2f6a91ea242459e5514 AS builder +FROM --platform=$BUILDPLATFORM golang@sha256:632dff7351799a5a9954a48dc06670ade72ee44e8c351cdc0e5a57af2fea21af AS builder WORKDIR /app COPY go.mod go.sum ./ @@ -39,8 +40,8 @@ RUN --mount=type=cache,target=/go/pkg/mod \ -o /out/netbird-upload ./upload-server \ && mkdir -p /out/var/lib/netbird -FROM cgr.dev/chainguard/static:latest@sha256:41e17ed83c594a64a9396b6ab96dd26d5ddc290dacf4c177464712ff21ad534f AS base-release -FROM cgr.dev/chainguard/busybox:latest@sha256:b2953ab1cae4a6265e18cf675851bd99975211b150d7a014774911d76eb309ba AS base-debug +FROM cgr.dev/chainguard/static@sha256:17367e4d3ae2b46c3290858e483fac09be085c92b372360fe3bf6a6150848fde AS base-release +FROM cgr.dev/chainguard/busybox@sha256:98e9e7306c899a8af252ef1ff50c54def91c2fe04677298eafdf0b902f2980ed AS base-debug # hadolint ignore=DL3006 FROM base-${VARIANT} From 9dd51ca42e975780bc6c9691111a712b7f08f430 Mon Sep 17 00:00:00 2001 From: Viktor Liu <17948409+lixmal@users.noreply.github.com> Date: Sat, 24 Oct 2026 17:49:51 +0900 Subject: [PATCH 3/5] [client] Drop SSH privileges through libc in PKCS#11 package builds (#8323) --- .github/workflows/golang-test-linux.yml | 8 ++ .golangci.yaml | 12 ++ .../executor_drop_privileged_linux_test.go | 102 +++++++++++++++++ client/ssh/server/executor_unix.go | 11 +- client/ssh/server/setid_libc_linux.go | 108 ++++++++++++++++++ client/ssh/server/setid_unix.go | 25 ++++ 6 files changed, 260 insertions(+), 6 deletions(-) create mode 100644 client/ssh/server/executor_drop_privileged_linux_test.go create mode 100644 client/ssh/server/setid_libc_linux.go create mode 100644 client/ssh/server/setid_unix.go diff --git a/.github/workflows/golang-test-linux.yml b/.github/workflows/golang-test-linux.yml index eea69eaa7..e4388bfe3 100644 --- a/.github/workflows/golang-test-linux.yml +++ b/.github/workflows/golang-test-linux.yml @@ -244,6 +244,14 @@ jobs: NB_TEST_PKCS11_DISPOSABLE: "1" run: CGO_ENABLED=0 go test -tags pkcs11 -timeout 5m ./client/internal/pkcs11/... ./client/internal/certproof/... + - name: Test SSH privilege drop + # purego marks the runtime as cgo without libc's setxid hooks, so the syscall + # package's setuid family panics in this build. Drop privileges as root to cover it. + run: | + CGO_ENABLED=0 go test -c -tags 'pkcs11 privileged' -o "$RUNNER_TEMP/ssh-server.test" ./client/ssh/server/ + sudo "$RUNNER_TEMP/ssh-server.test" -test.run '^TestDropPrivilegesAllThreads$' -test.v + CGO_ENABLED=0 GOARCH=arm64 go test -c -tags 'pkcs11 privileged' -o "$RUNNER_TEMP/ssh-server-arm64.test" ./client/ssh/server/ + test_client_on_docker: name: "Client (Docker) / Unit" needs: [build-cache] diff --git a/.golangci.yaml b/.golangci.yaml index e350b9de7..f52f822dd 100644 --- a/.golangci.yaml +++ b/.golangci.yaml @@ -24,6 +24,15 @@ linters: settings: errcheck: check-type-assertions: false + forbidigo: + analyze-types: true + forbid: + - pattern: ^(fmt\.Print(|f|ln)|print|println)$ + # Builds that link libc through purego mark the runtime as cgo without + # registering the libc setxid hooks, so these panic in AllThreadsSyscall. + - pattern: ^(syscall|unix)\.(Setgroups|Set(e|re|res)?[ug]id|AllThreadsSyscall6?)$ + pkg: ^(syscall|golang\.org/x/sys/unix)$ + msg: use the setgroups/setgid/setuid helpers in client/ssh/server, which go through libc in purego builds gocritic: disabled-checks: - commentFormatting @@ -86,6 +95,9 @@ linters: - linters: - forbidigo path: signal/cmd/root\.go + - linters: + - forbidigo + path: client/ssh/server/setid_unix\.go - linters: - unused path: sharedsock/filter\.go diff --git a/client/ssh/server/executor_drop_privileged_linux_test.go b/client/ssh/server/executor_drop_privileged_linux_test.go new file mode 100644 index 000000000..27cc1384b --- /dev/null +++ b/client/ssh/server/executor_drop_privileged_linux_test.go @@ -0,0 +1,102 @@ +//go:build linux && privileged + +package server + +import ( + "bufio" + "fmt" + "os" + "os/exec" + "path/filepath" + "runtime" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const envDropPrivilegesChild = "NB_TEST_DROP_PRIVILEGES_CHILD" + +const ( + dropTargetUID = 65534 + dropTargetGID = 65534 + dropExtraGID = 65533 +) + +// TestDropPrivilegesAllThreads drops privileges in a child process, because the drop +// cannot be undone, and checks that every OS thread of that process carries the target +// credentials. A drop applied to the calling thread only would leave the others as root. +func TestDropPrivilegesAllThreads(t *testing.T) { + if os.Getenv(envDropPrivilegesChild) == "1" { + dropPrivilegesChild(t) + return + } + if os.Geteuid() != 0 { + t.Skip("requires root") + } + + cmd := exec.Command(os.Args[0], "-test.run=^TestDropPrivilegesAllThreads$", "-test.v") + cmd.Env = append(os.Environ(), envDropPrivilegesChild+"=1") + out, err := cmd.CombinedOutput() + t.Logf("child output:\n%s", out) + require.NoError(t, err, "child must drop privileges on all threads") + assert.Contains(t, string(out), "--- PASS: TestDropPrivilegesAllThreads", "child test must run and pass") +} + +func dropPrivilegesChild(t *testing.T) { + // Park goroutines on their own OS threads so the process has threads other than + // the one that performs the drop. + release := make(chan struct{}) + defer close(release) + ready := make(chan struct{}) + for range 4 { + go func() { + runtime.LockOSThread() + defer runtime.UnlockOSThread() + ready <- struct{}{} + <-release + }() + <-ready + } + + pd := NewPrivilegeDropper() + require.NoError(t, pd.DropPrivileges(dropTargetUID, dropTargetGID, []uint32{dropTargetGID, dropExtraGID})) + + tasks, err := filepath.Glob("/proc/self/task/*/status") + require.NoError(t, err) + require.Greater(t, len(tasks), 4, "process must have several threads for the check to mean anything") + + want := map[string]string{ + "Uid": fmt.Sprintf("%d\t%d\t%d\t%d", dropTargetUID, dropTargetUID, dropTargetUID, dropTargetUID), + "Gid": fmt.Sprintf("%d\t%d\t%d\t%d", dropTargetGID, dropTargetGID, dropTargetGID, dropTargetGID), + "Groups": fmt.Sprintf("%d %d", dropExtraGID, dropTargetGID), + } + for _, task := range tasks { + got := readStatusFields(t, task) + for field, value := range want { + assert.Equal(t, value, got[field], "%s of %s", field, task) + } + } + + assert.Error(t, seteuid(0), "euid 0 must not be restorable after the drop") + assert.Error(t, setegid(0), "egid 0 must not be restorable after the drop") +} + +func readStatusFields(t *testing.T, path string) map[string]string { + f, err := os.Open(path) + require.NoError(t, err) + defer f.Close() + + fields := make(map[string]string) + scanner := bufio.NewScanner(f) + for scanner.Scan() { + key, value, ok := strings.Cut(scanner.Text(), ":") + if !ok { + continue + } + fields[key] = strings.TrimSpace(value) + } + require.NoError(t, scanner.Err()) + return fields +} diff --git a/client/ssh/server/executor_unix.go b/client/ssh/server/executor_unix.go index ee0b0ff78..5b42190ab 100644 --- a/client/ssh/server/executor_unix.go +++ b/client/ssh/server/executor_unix.go @@ -11,7 +11,6 @@ import ( "path/filepath" "runtime" "strings" - "syscall" log "github.com/sirupsen/logrus" ) @@ -151,15 +150,15 @@ func (pd *PrivilegeDropper) setGroupsAndIDs(targetUID, targetGID uint32, supplem } } - if err := syscall.Setgroups(groups); err != nil { + if err := setgroups(groups); err != nil { return fmt.Errorf("setgroups to %v: %w", groups, err) } - if err := syscall.Setgid(int(targetGID)); err != nil { + if err := setgid(int(targetGID)); err != nil { return fmt.Errorf("setgid to %d: %w", targetGID, err) } - if err := syscall.Setuid(int(targetUID)); err != nil { + if err := setuid(int(targetUID)); err != nil { return fmt.Errorf("setuid to %d: %w", targetUID, err) } @@ -182,12 +181,12 @@ func (pd *PrivilegeDropper) validatePrivilegeDropSuccess(targetUID, targetGID ui // validatePrivilegeDropReversibility ensures privileges cannot be restored func (pd *PrivilegeDropper) validatePrivilegeDropReversibility(targetUID, targetGID uint32, originalUID, originalGID int) error { if originalGID != int(targetGID) { - if err := syscall.Setegid(originalGID); err == nil { + if err := setegid(originalGID); err == nil { return fmt.Errorf("privilege drop validation failed: able to restore original GID %d", originalGID) } } if originalUID != int(targetUID) { - if err := syscall.Seteuid(originalUID); err == nil { + if err := seteuid(originalUID); err == nil { return fmt.Errorf("privilege drop validation failed: able to restore original UID %d", originalUID) } } diff --git a/client/ssh/server/setid_libc_linux.go b/client/ssh/server/setid_libc_linux.go new file mode 100644 index 000000000..13a9f254c --- /dev/null +++ b/client/ssh/server/setid_libc_linux.go @@ -0,0 +1,108 @@ +//go:build pkcs11 && linux && !android && (amd64 || arm64) + +package server + +import ( + "fmt" + "runtime" + "sync" + "syscall" + + "github.com/ebitengine/purego" +) + +// This build links libc through purego, whose fakecgo runtime marks the binary as cgo +// without registering the libc hooks a real cgo build provides. syscall's setuid family +// then panics in AllThreadsSyscall, so the calls go to libc directly: glibc applies them +// to every thread, and all Go threads here are created through pthread_create. + +type libcSetID struct { + setgroups func(size uintptr, list *uint32) int32 + setgid func(gid uint32) int32 + setuid func(uid uint32) int32 + setegid func(egid uint32) int32 + seteuid func(euid uint32) int32 + errnoLocation func() *int32 +} + +var ( + libcSetIDOnce sync.Once + libcSetIDFns libcSetID + libcSetIDErr error +) + +func loadLibcSetID() (*libcSetID, error) { + libcSetIDOnce.Do(func() { + syms := []struct { + fn any + name string + }{ + {&libcSetIDFns.setgroups, "setgroups"}, + {&libcSetIDFns.setgid, "setgid"}, + {&libcSetIDFns.setuid, "setuid"}, + {&libcSetIDFns.setegid, "setegid"}, + {&libcSetIDFns.seteuid, "seteuid"}, + {&libcSetIDFns.errnoLocation, "__errno_location"}, + } + for _, s := range syms { + addr, err := purego.Dlsym(purego.RTLD_DEFAULT, s.name) + if err != nil { + libcSetIDErr = fmt.Errorf("resolve libc %s: %w", s.name, err) + return + } + purego.RegisterFunc(s.fn, addr) + } + }) + if libcSetIDErr != nil { + return nil, libcSetIDErr + } + return &libcSetIDFns, nil +} + +// libcCall runs call on a locked thread so errno is read from the thread that set it. +func libcCall(call func(*libcSetID) int32) error { + fns, err := loadLibcSetID() + if err != nil { + return err + } + + runtime.LockOSThread() + defer runtime.UnlockOSThread() + + if call(fns) == 0 { + return nil + } + if errno := *fns.errnoLocation(); errno != 0 { + return syscall.Errno(errno) + } + return syscall.EINVAL +} + +func setgroups(gids []int) error { + list := make([]uint32, len(gids)) + for i, g := range gids { + list[i] = uint32(g) + } + return libcCall(func(f *libcSetID) int32 { + if len(list) == 0 { + return f.setgroups(0, nil) + } + return f.setgroups(uintptr(len(list)), &list[0]) + }) +} + +func setgid(gid int) error { + return libcCall(func(f *libcSetID) int32 { return f.setgid(uint32(gid)) }) +} + +func setuid(uid int) error { + return libcCall(func(f *libcSetID) int32 { return f.setuid(uint32(uid)) }) +} + +func setegid(egid int) error { + return libcCall(func(f *libcSetID) int32 { return f.setegid(uint32(egid)) }) +} + +func seteuid(euid int) error { + return libcCall(func(f *libcSetID) int32 { return f.seteuid(uint32(euid)) }) +} diff --git a/client/ssh/server/setid_unix.go b/client/ssh/server/setid_unix.go new file mode 100644 index 000000000..bd9fe3856 --- /dev/null +++ b/client/ssh/server/setid_unix.go @@ -0,0 +1,25 @@ +//go:build unix && !(pkcs11 && linux && !android && (amd64 || arm64)) + +package server + +import "syscall" + +func setgroups(gids []int) error { + return syscall.Setgroups(gids) +} + +func setgid(gid int) error { + return syscall.Setgid(gid) +} + +func setuid(uid int) error { + return syscall.Setuid(uid) +} + +func setegid(egid int) error { + return syscall.Setegid(egid) +} + +func seteuid(euid int) error { + return syscall.Seteuid(euid) +} From dd38fabc82d7314c11500d7fb8f1ee8161258250 Mon Sep 17 00:00:00 2001 From: Misha Bragin Date: Sat, 24 Oct 2026 14:01:45 -0400 Subject: [PATCH 4/5] [client] Add --table output to netbird status (#8305) --- client/cmd/status.go | 8 +- client/cmd/status_test.go | 59 ++++++++++++ client/status/table.go | 55 +++++++++++ client/status/table_test.go | 176 ++++++++++++++++++++++++++++++++++++ 4 files changed, 296 insertions(+), 2 deletions(-) create mode 100644 client/status/table.go create mode 100644 client/status/table_test.go diff --git a/client/cmd/status.go b/client/cmd/status.go index f2e5bcc66..6049d701a 100644 --- a/client/cmd/status.go +++ b/client/cmd/status.go @@ -19,6 +19,7 @@ import ( var ( detailFlag bool + tableFlag bool ipv4Flag bool ipv6Flag bool jsonFlag bool @@ -43,11 +44,12 @@ func init() { ipsFilterMap = make(map[string]struct{}) prefixNamesFilterMap = make(map[string]struct{}) statusCmd.PersistentFlags().BoolVarP(&detailFlag, "detail", "d", false, "display detailed status information in human-readable format") + statusCmd.PersistentFlags().BoolVarP(&tableFlag, "table", "t", false, "display peers in a table, supports the --filter-by-* flags") statusCmd.PersistentFlags().BoolVarP(&jsonFlag, "json", "j", false, "display detailed status information in json format") statusCmd.PersistentFlags().BoolVarP(&yamlFlag, "yaml", "y", false, "display detailed status information in yaml format") statusCmd.PersistentFlags().BoolVarP(&ipv4Flag, "ipv4", "4", false, "display only NetBird IPv4 of this peer, e.g., --ipv4 will output 100.64.0.33") statusCmd.PersistentFlags().BoolVarP(&ipv6Flag, "ipv6", "6", false, "display only NetBird IPv6 of this peer") - statusCmd.MarkFlagsMutuallyExclusive("detail", "json", "yaml", "ipv4", "ipv6") + statusCmd.MarkFlagsMutuallyExclusive("detail", "table", "json", "yaml", "ipv4", "ipv6") statusCmd.PersistentFlags().StringSliceVarP(&ipsFilter, "filter-by-ips", "I", []string{}, "filters the detailed output by a list of one or more IPs (v4 or v6), e.g., --filter-by-ips 100.64.0.100,fd00::1") statusCmd.PersistentFlags().StringSliceVarP(&prefixNamesFilter, "filter-by-names", "N", []string{}, "filters the detailed output by a list of one or more peer FQDN or hostnames, e.g., --filter-by-names peer-a,peer-b.netbird.cloud") statusCmd.PersistentFlags().StringVarP(&statusFilter, "filter-by-status", "S", "", "filters the detailed output by connection status(idle|connecting|connected), e.g., --filter-by-status connected") @@ -143,6 +145,8 @@ func statusFunc(cmd *cobra.Command, args []string) error { switch { case detailFlag: statusOutputString = outputInformationHolder.FullDetailSummary() + case tableFlag: + statusOutputString = outputInformationHolder.PeersTable() case jsonFlag: statusOutputString, err = outputInformationHolder.JSON() case yamlFlag: @@ -238,7 +242,7 @@ func parseFilters() error { } func enableDetailFlagWhenFilterFlag() { - if !detailFlag && !jsonFlag && !yamlFlag { + if !detailFlag && !jsonFlag && !yamlFlag && !tableFlag { detailFlag = true } } diff --git a/client/cmd/status_test.go b/client/cmd/status_test.go index 03608eab0..ca0fd97dd 100644 --- a/client/cmd/status_test.go +++ b/client/cmd/status_test.go @@ -4,6 +4,7 @@ import ( "testing" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" ) func TestParsingOfIP(t *testing.T) { @@ -13,3 +14,61 @@ func TestParsingOfIP(t *testing.T) { assert.Equal(t, "192.168.178.123\n", parsedIP) } + +// resetStatusFlags restores only the status-specific output flags. The global +// root flags (setup-key and friends) are shared with other tests in the +// package, so they must not be touched here. +func resetStatusFlags(t *testing.T) { + t.Helper() + t.Cleanup(func() { + for _, name := range []string{"detail", "table", "json", "yaml", "ipv4", "ipv6"} { + f := statusCmd.Flags().Lookup(name) + require.NotNil(t, f, "flag %s must exist", name) + require.NoError(t, f.Value.Set(f.DefValue)) + f.Changed = false + } + }) +} + +func TestStatusTableFlagExclusive(t *testing.T) { + for _, other := range []string{"detail", "json", "yaml", "ipv4", "ipv6"} { + t.Run(other, func(t *testing.T) { + resetStatusFlags(t) + require.NoError(t, statusCmd.Flags().Set("table", "true")) + require.NoError(t, statusCmd.Flags().Set(other, "true")) + + assert.Error(t, statusCmd.ValidateFlagGroups(), "--table with --%s must be rejected", other) + }) + } + + t.Run("alone", func(t *testing.T) { + resetStatusFlags(t) + require.NoError(t, statusCmd.Flags().Set("table", "true")) + + assert.NoError(t, statusCmd.ValidateFlagGroups()) + }) +} + +func TestFilterEnablesDetailOnlyWithoutOutputFormat(t *testing.T) { + tests := []struct { + name string + table bool + json bool + wantDetail bool + }{ + {name: "no format", wantDetail: true}, + {name: "table", table: true, wantDetail: false}, + {name: "json", json: true, wantDetail: false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + resetStatusFlags(t) + tableFlag, jsonFlag = tt.table, tt.json + + enableDetailFlagWhenFilterFlag() + + assert.Equal(t, tt.wantDetail, detailFlag) + }) + } +} diff --git a/client/status/table.go b/client/status/table.go new file mode 100644 index 000000000..f1cd7e01d --- /dev/null +++ b/client/status/table.go @@ -0,0 +1,55 @@ +package status + +import ( + "fmt" + "strings" + "text/tabwriter" + "time" +) + +// PeersTable renders the peers as an aligned table, one row per peer. +func (o *OutputOverview) PeersTable() string { + var sb strings.Builder + fmt.Fprintf(&sb, "Peers (%d/%d connected):\n", o.Peers.Connected, o.Peers.Total) + + tw := tabwriter.NewWriter(&sb, 0, 0, 3, ' ', 0) + fmt.Fprintln(tw, "NAME\tIP\tSTATUS\tCONN\tLATENCY\tRX / TX") + for _, p := range o.Peers.Details { + fmt.Fprintf(tw, "%s\t%s\t%s\t%s\t%s\t%s / %s\n", + shortName(p.FQDN), + dashIfEmpty(p.IP), + p.Status, + dashIfEmpty(p.ConnType), + tableLatency(p), + toIEC(p.TransferReceived), + toIEC(p.TransferSent), + ) + } + // The writer targets a strings.Builder, which never returns an error. + _ = tw.Flush() + + return sb.String() +} + +// shortName drops the domain suffix, which is the same for every peer. +func shortName(fqdn string) string { + name, _, _ := strings.Cut(fqdn, ".") + return dashIfEmpty(name) +} + +func tableLatency(p PeerStateDetailOutput) string { + if p.Latency <= 0 { + return "-" + } + if p.Latency >= time.Millisecond { + return p.Latency.Round(time.Millisecond).String() + } + return p.Latency.Round(time.Microsecond).String() +} + +func dashIfEmpty(s string) string { + if s == "" { + return "-" + } + return s +} diff --git a/client/status/table_test.go b/client/status/table_test.go new file mode 100644 index 000000000..c60e57917 --- /dev/null +++ b/client/status/table_test.go @@ -0,0 +1,176 @@ +package status + +import ( + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + protobuf "google.golang.org/protobuf/proto" + + "github.com/netbirdio/netbird/client/anonymize" + "github.com/netbirdio/netbird/client/proto" +) + +func TestPeersTable(t *testing.T) { + overview := OutputOverview{Peers: PeersStateOutput{ + Total: 2, + Connected: 1, + Details: []PeerStateDetailOutput{ + { + FQDN: "web-01.netbird.cloud", + IP: "100.64.0.12", + PubKey: "k3Vx9pQ2abcdefghijklmnopqrstuvwxyz0123456789=", + Status: "Connected", + ConnType: "P2P", + Latency: 12 * time.Millisecond, + TransferReceived: 1024, + TransferSent: 2048, + }, + { + FQDN: "old-nas.netbird.cloud", + IP: "100.64.0.77", + Status: "Idle", + }, + }, + }} + + lines := strings.Split(strings.TrimRight(overview.PeersTable(), "\n"), "\n") + require.Len(t, lines, 4, "title, header and one line per peer") + + assert.Equal(t, "Peers (1/2 connected):", lines[0]) + assert.Equal(t, + []string{"NAME", "IP", "STATUS", "CONN", "LATENCY", "RX", "/", "TX"}, + strings.Fields(lines[1])) + + connected := strings.Fields(lines[2]) + assert.Equal(t, "web-01", connected[0], "domain suffix must be dropped") + assert.Equal(t, "Connected", connected[2]) + assert.Equal(t, "P2P", connected[3]) + assert.Equal(t, "12ms", connected[4]) + assert.NotContains(t, lines[2], "k3Vx9pQ2", "public key must not be shown") + assert.Contains(t, lines[2], "1.0 KiB / 2.0 KiB") + + idle := strings.Fields(lines[3]) + assert.Equal(t, []string{"old-nas", "100.64.0.77", "Idle", "-", "-", "0", "B", "/", "0", "B"}, idle, + "missing values render as dashes") +} + +func TestPeersTable_NoPeers(t *testing.T) { + out := (&OutputOverview{}).PeersTable() + assert.True(t, strings.HasPrefix(out, "Peers (0/0 connected):\n")) + assert.Contains(t, out, "NAME") +} + +func tableRows(t *testing.T, opts ConvertOptions) []string { + t.Helper() + + opts.DaemonVersion = resp.GetDaemonVersion() + opts.DaemonStatus = ParseDaemonStatus(resp.GetStatus()) + // The anonymizer rewrites slices in place, so work on a copy to keep the + // shared fixture intact for the other tests. + fullStatus := protobuf.Clone(resp.GetFullStatus()).(*proto.FullStatus) + overview := ConvertToStatusOutputOverview(fullStatus, opts) + out := overview.PeersTable() + + lines := strings.Split(strings.TrimRight(out, "\n"), "\n") + require.GreaterOrEqual(t, len(lines), 2, "title and header are always present") + return lines +} + +// The fixture holds peer-1 (connected, P2P) and peer-2 (connected, relayed). +func TestPeersTable_Filters(t *testing.T) { + tests := []struct { + name string + opts ConvertOptions + wantTitle string + wantPeers []string + }{ + { + name: "no filter lists every peer sorted by IP", + wantTitle: "Peers (2/2 connected):", + wantPeers: []string{"peer-1", "peer-2"}, + }, + { + name: "connection type P2P", + opts: ConvertOptions{ConnectionTypeFilter: "P2P"}, + wantTitle: "Peers (1/1 connected):", + wantPeers: []string{"peer-1"}, + }, + { + name: "connection type Relayed", + opts: ConvertOptions{ConnectionTypeFilter: "relayed"}, + wantTitle: "Peers (1/1 connected):", + wantPeers: []string{"peer-2"}, + }, + { + name: "status with no match", + opts: ConvertOptions{StatusFilter: "idle"}, + wantTitle: "Peers (0/0 connected):", + }, + { + name: "name prefix", + opts: ConvertOptions{ + PrefixNamesFilter: []string{"peer-2"}, + PrefixNamesFilterMap: map[string]struct{}{"peer-2": {}}, + }, + wantTitle: "Peers (1/1 connected):", + wantPeers: []string{"peer-2"}, + }, + { + name: "IPv4", + opts: ConvertOptions{IPsFilter: map[string]struct{}{"192.168.178.101": {}}}, + wantTitle: "Peers (1/1 connected):", + wantPeers: []string{"peer-1"}, + }, + { + name: "IPv6", + opts: ConvertOptions{IPsFilter: map[string]struct{}{"fd00::1": {}}}, + wantTitle: "Peers (1/1 connected):", + wantPeers: []string{"peer-1"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + lines := tableRows(t, tt.opts) + + assert.Equal(t, tt.wantTitle, lines[0]) + rows := lines[2:] + require.Len(t, rows, len(tt.wantPeers), "one row per matching peer") + for i, name := range tt.wantPeers { + assert.True(t, strings.HasPrefix(rows[i], name+" "), "row %d should be %s, got %q", i, name, rows[i]) + } + }) + } +} + +// The table shows whatever the anonymizer left in the overview, so it must +// carry the redacted names rather than the originals. +func TestPeersTable_Anonymized(t *testing.T) { + lines := tableRows(t, ConvertOptions{Anonymize: true, AnonymizeLevel: anonymize.LevelStrict}) + + for _, row := range lines[2:] { + assert.NotContains(t, row, "awesome-domain.com", "FQDN must be anonymized") + assert.NotContains(t, row, "peer-", "peer name must be anonymized") + } +} + +func TestShortName(t *testing.T) { + assert.Equal(t, "web-01", shortName("web-01.netbird.cloud")) + assert.Equal(t, "web-01", shortName("web-01")) + assert.Equal(t, "-", shortName("")) +} + +func TestTableLatency(t *testing.T) { + tests := map[time.Duration]string{ + 0: "-", + 145939625 * time.Nanosecond: "146ms", + 12 * time.Millisecond: "12ms", + 420*time.Microsecond + 700: "421µs", + } + for in, want := range tests { + assert.Equal(t, want, tableLatency(PeerStateDetailOutput{Latency: in}), "latency %s", in) + } +} From fdc51d6f9539756c38b777f091534c804eb2f9a3 Mon Sep 17 00:00:00 2001 From: Pascal Fischer <32096965+pascal-fischer@users.noreply.github.com> Date: Sun, 25 Oct 2026 22:34:44 +0100 Subject: [PATCH 5/5] [management] shift network map data store onto the read replica (#8302) --- .../network_map_db/network_map_data_test.go | 75 ++++- .../network_map/controller/controller.go | 2 +- management/internals/modules/peers/manager.go | 4 + .../internals/modules/peers/manager_test.go | 32 ++ .../service/manager/l4_port_test.go | 3 +- .../reverseproxy/service/manager/manager.go | 31 ++ .../service/manager/manager_test.go | 58 +++- .../network_map_db/factory/db_store.go | 45 ++- .../internals/network_map_db/metrics.go | 96 ++++++ .../network_map_db/network_map_data.go | 163 ++++++++-- .../network_map_db/network_map_data_test.go | 278 +++++++++++++++++- .../internals/network_map_db/shared_types.go | 17 +- management/internals/server/boot.go | 24 ++ management/internals/shared/db/open.go | 5 +- .../internals/shared/requestbuffer/buffer.go | 31 ++ .../shared/requestbuffer/buffer_test.go | 71 +++++ management/server/account.go | 17 +- management/server/account_test.go | 45 ++- management/server/peer.go | 42 ++- management/server/peer_test.go | 68 +++++ management/server/user.go | 13 +- management/server/user_test.go | 22 ++ 22 files changed, 1095 insertions(+), 47 deletions(-) create mode 100644 management/internals/network_map_db/metrics.go diff --git a/integration_tests/management/network_map_db/network_map_data_test.go b/integration_tests/management/network_map_db/network_map_data_test.go index 00c0ec03f..96172bd49 100644 --- a/integration_tests/management/network_map_db/network_map_data_test.go +++ b/integration_tests/management/network_map_db/network_map_data_test.go @@ -10,14 +10,21 @@ import ( "path/filepath" "runtime" "strings" + "sync" "testing" + "time" + "github.com/jackc/pgx/v5/pgxpool" networkmapdb "github.com/netbirdio/netbird/management/internals/network_map_db" "github.com/netbirdio/netbird/management/server/integrations/integrated_validator" "github.com/netbirdio/netbird/management/server/settings" "github.com/netbirdio/netbird/management/server/types" log "github.com/sirupsen/logrus" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/attribute" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" "go.uber.org/mock/gomock" ) @@ -29,6 +36,19 @@ var goldenNMap string const EnvUpdateGoldenData = "NMAP_UPDATE_GOLDEN_DATA" +var seedNetworkMapDataOnce sync.Once + +func seedNetworkMapData(t *testing.T, ctx context.Context) { + t.Helper() + seedNetworkMapDataOnce.Do(func() { + for _, query := range strings.Split(nmapData, ";") { + if err := store(t).Exec(ctx, query); err != nil { + log.Fatalf("error initializing nmap test: %s", err.Error()) + } + } + }) +} + func TestGetNetworkMapData(t *testing.T) { ctx := context.TODO() @@ -51,11 +71,7 @@ func TestGetNetworkMapData(t *testing.T) { IntegratedPeerValidator: peerValidators, } - for _, query := range strings.Split(nmapData, ";") { - if err := store(t).Exec(ctx, query); err != nil { - log.Fatalf("error initializing nmap test: %s", err.Error()) - } - } + seedNetworkMapData(t, ctx) nmap, err := storeImpl.GetNetworkMapData(ctx, "account-33") assert.NoError(t, err) @@ -72,3 +88,52 @@ func TestGetNetworkMapData(t *testing.T) { } assert.Equal(t, goldenNMap, string(serializedNMap)) } + +func TestGetNetworkMapData_FromReplica(t *testing.T) { + if engine != string(types.PostgresStoreEngine) { + t.Skip("replicas exist only for Postgres") + } + ctx := context.TODO() + + extraSettingsManager := settings.NewMockManager(gomock.NewController(t)) + extraSettingsManager.EXPECT().GetExtraSettings(gomock.Any(), gomock.Any()).Return(&types.ExtraSettings{}, nil) + peerValidators := integrated_validator.NewMockIntegratedValidator(gomock.NewController(t)) + peerValidators.EXPECT().GetValidatedPeers(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(map[string]struct{}{}, nil) + + seedNetworkMapData(t, ctx) + + reader := sdkmetric.NewManualReader() + metrics, err := networkmapdb.NewMetrics(sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)).Meter("test"), + map[string]*pgxpool.Pool{"replica": pgstore.Pool}) + require.NoError(t, err) + + primary := store(t) + storeImpl := networkmapdb.NetworkMapDBStoreImpl{ + Metrics: metrics, + Store: primary, + Replica: primary, + ReplicaWait: time.Second, + ExtraSettingsManager: extraSettingsManager, + IntegratedPeerValidator: peerValidators, + } + + nmap, err := storeImpl.GetNetworkMapData(ctx, "account-33") + require.NoError(t, err) + assert.NotEmpty(t, nmap.Peers, "data must be read through the replica transaction after its attempt deadline is released") + + var rm metricdata.ResourceMetrics + require.NoError(t, reader.Collect(ctx, &rm)) + var totalConns int64 + for _, sm := range rm.ScopeMetrics { + for _, m := range sm.Metrics { + if gauge, ok := m.Data.(metricdata.Gauge[int64]); ok { + for _, dp := range gauge.DataPoints { + if state, _ := dp.Attributes.Value(attribute.Key("state")); state.AsString() == "total" { + totalConns += dp.Value + } + } + } + } + } + assert.Positive(t, totalConns, "the replica pool gauge reports its open connections") +} diff --git a/management/internals/controllers/network_map/controller/controller.go b/management/internals/controllers/network_map/controller/controller.go index 08d018e5f..0554811db 100644 --- a/management/internals/controllers/network_map/controller/controller.go +++ b/management/internals/controllers/network_map/controller/controller.go @@ -110,7 +110,7 @@ func NewController(ctx context.Context, store store.Store, metrics telemetry.App if nmdataStore != nil { interval := requestbuffer.Interval(ctx, "NB_NETWORK_MAP_DATA_BUFFER_INTERVAL", defaultNetworkMapDataBufferInterval) log.WithContext(ctx).Infof("set network map data request buffer interval to %s", interval) - c.nmdataBuffer = requestbuffer.New(ctx, "network map data request buffer", interval, c.fetchNetworkMapData) + c.nmdataBuffer = requestbuffer.NewSerialized(ctx, "network map data request buffer", interval, c.fetchNetworkMapData) } return c diff --git a/management/internals/modules/peers/manager.go b/management/internals/modules/peers/manager.go index 6f9705efe..b49667948 100644 --- a/management/internals/modules/peers/manager.go +++ b/management/internals/modules/peers/manager.go @@ -199,6 +199,10 @@ func (m *managerImpl) DeletePeers(ctx context.Context, accountID string, peerIDs return err } + if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + log.WithContext(ctx).Debugf("DeletePeers: deleted peer %s", peerID) if !(peer.ProxyMeta.Embedded || peer.Meta.KernelVersion == "wasm") { diff --git a/management/internals/modules/peers/manager_test.go b/management/internals/modules/peers/manager_test.go index 2645276b7..eba565fbf 100644 --- a/management/internals/modules/peers/manager_test.go +++ b/management/internals/modules/peers/manager_test.go @@ -5,6 +5,8 @@ import ( "fmt" "testing" + "github.com/netbirdio/netbird/management/internals/controllers/network_map" + "github.com/netbirdio/netbird/management/server/account" "github.com/netbirdio/netbird/management/server/peer" "github.com/netbirdio/netbird/management/server/permissions" "github.com/netbirdio/netbird/management/server/permissions/modules" @@ -195,3 +197,33 @@ func TestManager_GetPeersPaginated_NoPermission_User_IsRestrictable_ViewBlocked( assert.Equal(t, 0, total) assert.NoError(t, err) } + +func TestManager_DeletePeers_IncrementsNetworkSerial(t *testing.T) { + ctrl := gomock.NewController(t) + mockStore := store.NewMockStore(ctrl) + txStore := store.NewMockStore(ctrl) + accountManager := account.NewMockManager(ctrl) + networkMapController := network_map.NewMockController(ctrl) + + m := NewManager(mockStore, permissions.NewMockManager(ctrl)).(*managerImpl) + m.accountManager = accountManager + m.networkMapController = networkMapController + + ctx := context.Background() + mockStore.EXPECT().GetAccountSettings(gomock.Any(), store.LockingStrengthNone, "test-account").Return(&types.Settings{}, nil) + networkMapController.EXPECT().GetDNSDomain(gomock.Any()).Return("netbird.selfhosted") + mockStore.EXPECT().ExecuteInTransaction(gomock.Any(), gomock.Any()).Times(2). + DoAndReturn(func(_ context.Context, fn func(store.Store) error) error { return fn(txStore) }) + + for _, id := range []string{"peer-1", "peer-2"} { + txStore.EXPECT().GetPeerByID(gomock.Any(), store.LockingStrengthNone, "test-account", id).Return(&peer.Peer{ID: id, Status: &peer.PeerStatus{}}, nil) + txStore.EXPECT().RemovePeerFromAllGroups(gomock.Any(), id).Return(nil) + txStore.EXPECT().GetPolicyRulesByResourceID(gomock.Any(), store.LockingStrengthNone, "test-account", id).Return(nil, nil) + txStore.EXPECT().DeletePeer(gomock.Any(), "test-account", id).Return(nil) + } + txStore.EXPECT().IncrementNetworkSerial(gomock.Any(), "test-account").Times(2).Return(nil) + accountManager.EXPECT().StoreEvent(gomock.Any(), "test-user", gomock.Any(), "test-account", gomock.Any(), gomock.Any()).Times(2) + accountManager.EXPECT().UpdateAccountPeers(gomock.Any(), "test-account", gomock.Any()) + + assert.NoError(t, m.DeletePeers(ctx, "test-account", []string{"peer-1", "peer-2"}, "test-user", false)) +} diff --git a/management/internals/modules/reverseproxy/service/manager/l4_port_test.go b/management/internals/modules/reverseproxy/service/manager/l4_port_test.go index a44e759c4..5ec34a073 100644 --- a/management/internals/modules/reverseproxy/service/manager/l4_port_test.go +++ b/management/internals/modules/reverseproxy/service/manager/l4_port_test.go @@ -6,9 +6,9 @@ import ( "testing" "time" - "go.uber.org/mock/gomock" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy" rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service" @@ -38,6 +38,7 @@ func setupL4Test(t *testing.T, customPortsSupported *bool) (*Manager, store.Stor err = testStore.SaveAccount(ctx, &types.Account{ Id: testAccountID, CreatedBy: testUserID, + Network: types.NewNetwork(), Settings: &types.Settings{ PeerExposeEnabled: true, PeerExposeGroups: []string{testGroupID}, diff --git a/management/internals/modules/reverseproxy/service/manager/manager.go b/management/internals/modules/reverseproxy/service/manager/manager.go index ebb52c74b..6173a381d 100644 --- a/management/internals/modules/reverseproxy/service/manager/manager.go +++ b/management/internals/modules/reverseproxy/service/manager/manager.go @@ -363,6 +363,10 @@ func (m *Manager) persistNewService(ctx context.Context, accountID string, svc * return fmt.Errorf("create service: %w", err) } + if err := transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + return nil }) } @@ -534,6 +538,10 @@ func (m *Manager) persistNewEphemeralService(ctx context.Context, accountID, pee return fmt.Errorf("create service: %w", err) } + if err := transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + return nil }) } @@ -744,6 +752,10 @@ func (m *Manager) executeServiceUpdate(ctx context.Context, transaction store.St return fmt.Errorf("update service: %w", err) } + if err := transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + return nil } @@ -1027,6 +1039,10 @@ func (m *Manager) DeleteService(ctx context.Context, accountID, userID, serviceI return fmt.Errorf("failed to delete service: %w", err) } + if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + return nil }) if err != nil { @@ -1069,6 +1085,12 @@ func (m *Manager) DeleteAllServices(ctx context.Context, accountID, userID strin } } + if len(services) > 0 { + if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + } + return nil }) if err != nil { @@ -1425,6 +1447,10 @@ func (m *Manager) deletePeerService(ctx context.Context, accountID, peerID, serv return fmt.Errorf("delete service: %w", err) } + if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + return nil }) if err != nil { @@ -1477,6 +1503,11 @@ func (m *Manager) deleteExpiredPeerService(ctx context.Context, accountID, peerI if err = transaction.DeleteService(ctx, accountID, serviceID); err != nil { return fmt.Errorf("delete service: %w", err) } + + if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + deleted = true return nil diff --git a/management/internals/modules/reverseproxy/service/manager/manager_test.go b/management/internals/modules/reverseproxy/service/manager/manager_test.go index f9273d594..85d7c4574 100644 --- a/management/internals/modules/reverseproxy/service/manager/manager_test.go +++ b/management/internals/modules/reverseproxy/service/manager/manager_test.go @@ -227,6 +227,13 @@ func TestCheckDomainAvailable_EdgeCases(t *testing.T) { }) } +func networkSerial(t *testing.T, s store.Store, accountID string) uint64 { + t.Helper() + account, err := s.GetAccount(context.Background(), accountID) + require.NoError(t, err) + return account.Network.Serial +} + func TestPersistNewService(t *testing.T) { ctx := context.Background() accountID := "test-account" @@ -254,6 +261,9 @@ func TestPersistNewService(t *testing.T) { txMock.EXPECT(). CreateService(ctx, service). Return(nil) + txMock.EXPECT(). + IncrementNetworkSerial(ctx, accountID). + Return(nil) return fn(txMock) }) @@ -465,6 +475,9 @@ func TestDeletePeerService_SourcePeerValidation(t *testing.T) { txMock.EXPECT(). DeleteService(ctx, accountID, serviceID). Return(nil) + txMock.EXPECT(). + IncrementNetworkSerial(ctx, accountID). + Return(nil) return fn(txMock) }) mockStore.EXPECT(). @@ -570,6 +583,9 @@ func TestDeletePeerService_SourcePeerValidation(t *testing.T) { txMock.EXPECT(). DeleteService(ctx, accountID, serviceID). Return(nil) + txMock.EXPECT(). + IncrementNetworkSerial(ctx, accountID). + Return(nil) return fn(txMock) }) mockStore.EXPECT(). @@ -617,6 +633,9 @@ func TestDeletePeerService_SourcePeerValidation(t *testing.T) { txMock.EXPECT(). DeleteService(ctx, accountID, serviceID). Return(nil) + txMock.EXPECT(). + IncrementNetworkSerial(ctx, accountID). + Return(nil) return fn(txMock) }) mockStore.EXPECT(). @@ -679,6 +698,7 @@ func setupIntegrationTest(t *testing.T) (*Manager, store.Store) { err = testStore.SaveAccount(ctx, &types.Account{ Id: testAccountID, CreatedBy: testUserID, + Network: types.NewNetwork(), Settings: &types.Settings{ PeerExposeEnabled: true, PeerExposeGroups: []string{testGroupID}, @@ -843,6 +863,7 @@ func TestCreateServiceFromPeer(t *testing.T) { assert.Equal(t, rpservice.SourceEphemeral, persisted.Source, "source should be ephemeral") assert.Equal(t, testPeerID, persisted.SourcePeer, "source peer should be set") assert.NotNil(t, persisted.Meta.LastRenewedAt, "last renewed should be set") + assert.Equal(t, uint64(1), networkSerial(t, testStore, testAccountID), "network serial should be incremented") }) t.Run("creates service with custom domain", func(t *testing.T) { @@ -1049,8 +1070,10 @@ func TestDeleteService_DeletesEphemeralExpose(t *testing.T) { svc, err := testStore.GetServiceByDomain(ctx, resp.Domain) require.NoError(t, err) + serial := networkSerial(t, testStore, testAccountID) err = mgr.DeleteService(ctx, testAccountID, testUserID, svc.ID) require.NoError(t, err) + assert.Equal(t, serial+1, networkSerial(t, testStore, testAccountID), "network serial should be incremented") count, err = mgr.store.CountEphemeralServicesByPeer(ctx, store.LockingStrengthNone, testAccountID, testPeerID) require.NoError(t, err) @@ -1065,7 +1088,7 @@ func TestDeleteService_DeletesEphemeralExpose(t *testing.T) { func TestDeleteAllServices_DeletesEphemeralExposes(t *testing.T) { ctx := context.Background() - mgr, _ := setupIntegrationTest(t) + mgr, testStore := setupIntegrationTest(t) for i := range 3 { _, err := mgr.CreateServiceFromPeer(ctx, testAccountID, testPeerID, &rpservice.ExposeServiceRequest{ @@ -1079,8 +1102,10 @@ func TestDeleteAllServices_DeletesEphemeralExposes(t *testing.T) { require.NoError(t, err) assert.Equal(t, int64(3), count, "all ephemeral services should exist") + serial := networkSerial(t, testStore, testAccountID) err = mgr.DeleteAllServices(ctx, testAccountID, testUserID) require.NoError(t, err) + assert.Equal(t, serial+1, networkSerial(t, testStore, testAccountID), "network serial should be incremented") count, err = mgr.store.CountEphemeralServicesByPeer(ctx, store.LockingStrengthNone, testAccountID, testPeerID) require.NoError(t, err) @@ -1223,8 +1248,10 @@ func TestDeleteExpiredPeerService_DeletesTargets(t *testing.T) { require.Len(t, targets, 1, "ephemeral peer-exposed service should have exactly one persisted target before reaping") expireEphemeralService(t, testStore, testAccountID, resp.Domain) + serial := networkSerial(t, testStore, testAccountID) err = mgr.deleteExpiredPeerService(ctx, testAccountID, testPeerID, svcID) require.NoError(t, err) + assert.Equal(t, serial+1, networkSerial(t, testStore, testAccountID), "network serial should be incremented") _, err = testStore.GetServiceByDomain(ctx, resp.Domain) require.Error(t, err, "expired peer-exposed service should be deleted") @@ -1253,8 +1280,10 @@ func TestDeleteServiceFromPeer_DeletesTargets(t *testing.T) { require.NoError(t, err) require.Len(t, targets, 1, "ephemeral peer-exposed service should have exactly one persisted target before stopping") + serial := networkSerial(t, testStore, testAccountID) err = mgr.StopServiceFromPeer(ctx, testAccountID, testPeerID, svcID) require.NoError(t, err) + assert.Equal(t, serial+1, networkSerial(t, testStore, testAccountID), "network serial should be incremented") _, err = testStore.GetServiceByDomain(ctx, resp.Domain) require.Error(t, err, "stopped peer-exposed service should be deleted") @@ -1521,3 +1550,30 @@ func TestReloadAllServicesForAccount_ContinuesAfterFailedService(t *testing.T) { assert.Contains(t, err.Error(), "svc-broken") assert.NotContains(t, err.Error(), "svc-healthy") } + +func TestServiceWrites_IncrementNetworkSerial(t *testing.T) { + ctx := context.Background() + mgr, testStore, _ := setupL4Test(t, nil) + + created, err := mgr.CreateService(ctx, testAccountID, testUserID, &rpservice.Service{ + Name: "serial", Domain: "serial.test.netbird.io", Mode: rpservice.ModeHTTP, Enabled: true, + Targets: []*rpservice.Target{{ + AccountID: testAccountID, TargetId: testPeerID, + TargetType: rpservice.TargetTypePeer, Protocol: "http", Port: 8080, Enabled: true, + }}, + }) + require.NoError(t, err) + serial := networkSerial(t, testStore, testAccountID) + assert.Equal(t, uint64(1), serial, "create should increment the network serial") + + updated := created.Copy() + updated.Name = "serial renamed" + _, err = mgr.UpdateService(ctx, testAccountID, testUserID, updated) + require.NoError(t, err) + assert.Equal(t, serial+1, networkSerial(t, testStore, testAccountID), "update should increment the network serial") + serial++ + + require.NoError(t, mgr.SetStatus(ctx, testAccountID, created.ID, rpservice.StatusActive)) + require.NoError(t, mgr.SetCertificateIssuedAt(ctx, testAccountID, created.ID)) + assert.Equal(t, serial, networkSerial(t, testStore, testAccountID), "meta-only writes must not increment the network serial") +} diff --git a/management/internals/network_map_db/factory/db_store.go b/management/internals/network_map_db/factory/db_store.go index 3eea0ae69..e60c1f6f9 100644 --- a/management/internals/network_map_db/factory/db_store.go +++ b/management/internals/network_map_db/factory/db_store.go @@ -5,10 +5,15 @@ import ( "errors" "fmt" "os" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "go.opentelemetry.io/otel/metric" networkmapdb "github.com/netbirdio/netbird/management/internals/network_map_db" networkmap_pgsql "github.com/netbirdio/netbird/management/internals/network_map_db/pgsql" networkmap_sqlite "github.com/netbirdio/netbird/management/internals/network_map_db/sqlite" + "github.com/netbirdio/netbird/management/internals/shared/requestbuffer" "github.com/netbirdio/netbird/management/server/integrations/integrated_validator" "github.com/netbirdio/netbird/management/server/settings" "github.com/netbirdio/netbird/management/server/store" @@ -16,7 +21,16 @@ import ( log "github.com/sirupsen/logrus" ) -const storeSqliteFileName = "store.db" +const ( + storeSqliteFileName = "store.db" + + // ReplicaDsnEnv names the Postgres DSN of a read replica serving network map data reads. + ReplicaDsnEnv = "NB_STORE_ENGINE_POSTGRES_REPLICA_DSN" + // replicaWaitEnv bounds how long a network map data read waits for the replica. + replicaWaitEnv = "NB_NETWORK_MAP_DATA_REPLICA_WAIT" + + defaultReplicaWait = 2 * time.Second +) var ErrNotSupportedStoreEngine = errors.New("unsupported store engine") @@ -25,7 +39,9 @@ func NewNetworkMapDBStore( kind types.Engine, dataDir string, integratedPeerValidator integrated_validator.IntegratedValidator, - extraSettingsManager settings.Manager) (*networkmapdb.NetworkMapDBStoreImpl, error) { + extraSettingsManager settings.Manager, + replicaPool *pgxpool.Pool, + meter metric.Meter) (*networkmapdb.NetworkMapDBStoreImpl, error) { switch kind { case types.SqliteStoreEngine: log.WithContext(ctx).Info("networkmap store is using SQLite") @@ -54,11 +70,32 @@ func NewNetworkMapDBStore( return nil, err } - return &networkmapdb.NetworkMapDBStoreImpl{ + impl := &networkmapdb.NetworkMapDBStoreImpl{ Store: store, IntegratedPeerValidator: integratedPeerValidator, ExtraSettingsManager: extraSettingsManager, - }, nil + } + + pools := map[string]*pgxpool.Pool{"primary": store.Pool} + if replicaPool != nil { + impl.Replica = &networkmap_pgsql.PgStore{Pool: replicaPool} + impl.ReplicaWait = requestbuffer.Interval(ctx, replicaWaitEnv, defaultReplicaWait) + pools["replica"] = replicaPool + replicaConfig := replicaPool.Config().ConnConfig + log.WithContext(ctx).Infof("networkmap store reads from Postgres replica %s:%d/%s, waiting up to %s for replication", + replicaConfig.Host, replicaConfig.Port, replicaConfig.Database, impl.ReplicaWait) + } else { + log.WithContext(ctx).Info("networkmap store reads from the primary, no replica configured") + } + + if meter != nil { + impl.Metrics, err = networkmapdb.NewMetrics(meter, pools) + if err != nil { + return nil, fmt.Errorf("create network map data metrics: %w", err) + } + } + + return impl, nil } return nil, fmt.Errorf("networkmap store doesn't support engine %s, %w", kind, ErrNotSupportedStoreEngine) diff --git a/management/internals/network_map_db/metrics.go b/management/internals/network_map_db/metrics.go new file mode 100644 index 000000000..0cfd8f92d --- /dev/null +++ b/management/internals/network_map_db/metrics.go @@ -0,0 +1,96 @@ +package networkmapdb + +import ( + "context" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgxpool" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/metric" +) + +const ( + readSourceReplica = "replica" + readSourcePrimary = "primary" + + readReasonNoReplica = "no_replica" + readReasonCaughtUp = "caught_up" + readReasonLagTimeout = "lag_timeout" + readReasonReplicaError = "replica_error" + readReasonSerialReadError = "serial_read_error" +) + +// Metrics records which database serves network map data reads, how long reads +// wait for the replica and the state of the connection pools. +type Metrics struct { + reads metric.Int64Counter + replicaWait metric.Int64Histogram +} + +// NewMetrics creates the network map data metrics on meter and observes the +// connection pools keyed by pool name. +func NewMetrics(meter metric.Meter, pools map[string]*pgxpool.Pool) (*Metrics, error) { + reads, err := meter.Int64Counter("management.nmdata.reads", + metric.WithUnit("1"), + metric.WithDescription("Network map data reads by the database that served them and the reason it was chosen"), + ) + if err != nil { + return nil, fmt.Errorf("create reads counter: %w", err) + } + + replicaWait, err := meter.Int64Histogram("management.nmdata.replica.wait.ms", + metric.WithUnit("milliseconds"), + metric.WithDescription("Time a network map data read waited for the replica to replicate the primary's network serial"), + ) + if err != nil { + return nil, fmt.Errorf("create replica wait histogram: %w", err) + } + + connections, err := meter.Int64ObservableGauge("management.nmdata.pool.connections", + metric.WithUnit("1"), + metric.WithDescription("Connections of the network map data connection pools by state"), + ) + if err != nil { + return nil, fmt.Errorf("create pool connections gauge: %w", err) + } + + _, err = meter.RegisterCallback(func(_ context.Context, observer metric.Observer) error { + for name, pool := range pools { + stat := pool.Stat() + observePool(observer, connections, name, "acquired", stat.AcquiredConns()) + observePool(observer, connections, name, "idle", stat.IdleConns()) + observePool(observer, connections, name, "total", stat.TotalConns()) + } + return nil + }, connections) + if err != nil { + return nil, fmt.Errorf("create pool connections callback: %w", err) + } + + return &Metrics{reads: reads, replicaWait: replicaWait}, nil +} + +func observePool(observer metric.Observer, gauge metric.Int64ObservableGauge, pool, state string, value int32) { + observer.ObserveInt64(gauge, int64(value), metric.WithAttributes( + attribute.String("pool", pool), + attribute.String("state", state), + )) +} + +func (m *Metrics) countRead(ctx context.Context, source, reason string) { + if m == nil { + return + } + m.reads.Add(ctx, 1, metric.WithAttributes( + attribute.String("source", source), + attribute.String("reason", reason), + )) +} + +func (m *Metrics) recordReplicaWait(ctx context.Context, wait time.Duration) { + if m == nil { + return + } + m.replicaWait.Record(ctx, wait.Milliseconds()) +} diff --git a/management/internals/network_map_db/network_map_data.go b/management/internals/network_map_db/network_map_data.go index 1e83bfb34..2c1e7abcc 100644 --- a/management/internals/network_map_db/network_map_data.go +++ b/management/internals/network_map_db/network_map_data.go @@ -2,7 +2,9 @@ package networkmapdb import ( "context" + "errors" "fmt" + "time" log "github.com/sirupsen/logrus" "golang.org/x/exp/maps" @@ -11,12 +13,147 @@ import ( "github.com/netbirdio/netbird/shared/management/networkmap/nmdata" ) +// GetNetworkMapData reads the network map data of the account. With a Replica +// configured it reads from the replica once that has replicated the account's +// current network serial, and from Store when it does not catch up within +// ReplicaWait or the replica read fails. func (s *NetworkMapDBStoreImpl) GetNetworkMapData(ctx context.Context, accountId string) (*networkmap.NetworkMapData, error) { - tx, err := s.Store.BeginTx(ctx) + toret, err := s.readNetworkMapData(ctx, accountId) if err != nil { return nil, err } + extraSettings, err := s.ExtraSettingsManager.GetExtraSettings(ctx, accountId) + if err != nil { + return nil, err + } + + toret.ValidatedPeers, err = s.IntegratedPeerValidator.GetValidatedPeers(ctx, accountId, maps.Values(toret.Groups), maps.Values(toret.Peers), extraSettings) + if err != nil { + return nil, err + } + + return toret, nil +} + +func (s *NetworkMapDBStoreImpl) readNetworkMapData(ctx context.Context, accountId string) (*networkmap.NetworkMapData, error) { + reason := readReasonNoReplica + if s.Replica != nil { + toret, err := s.readFromReplica(ctx, accountId) + if err == nil { + s.Metrics.countRead(ctx, readSourceReplica, readReasonCaughtUp) + return toret, nil + } + if ctx.Err() != nil { + return nil, ctx.Err() + } + reason = fallbackReason(err) + if reason == readReasonLagTimeout { + log.WithContext(ctx).Debugf("reading network map data of account %s from primary: %v", accountId, err) + } else { + log.WithContext(ctx).Warnf("reading network map data of account %s from primary: %v", accountId, err) + } + } + + tx, network, err := beginRead(ctx, s.Store, accountId) + if err != nil { + return nil, err + } + toret, err := readFromTx(ctx, tx, accountId, network) + if err != nil { + return nil, err + } + s.Metrics.countRead(ctx, readSourcePrimary, reason) + return toret, nil +} + +func fallbackReason(err error) string { + switch { + case errors.Is(err, errReplicaLagging): + return readReasonLagTimeout + case errors.Is(err, errPrimarySerialRead): + return readReasonSerialReadError + default: + return readReasonReplicaError + } +} + +func (s *NetworkMapDBStoreImpl) readFromReplica(ctx context.Context, accountId string) (*networkmap.NetworkMapData, error) { + tx, network, err := s.beginOnReplica(ctx, accountId) + if err != nil { + return nil, err + } + return readFromTx(ctx, tx, accountId, network) +} + +// beginOnReplica opens a read transaction on the replica whose snapshot holds at +// least the account's network serial on the primary at the time of the call. +func (s *NetworkMapDBStoreImpl) beginOnReplica(ctx context.Context, accountId string) (NetworkMapDBStoreConn, nmdata.Network, error) { + primaryTx, primaryNetwork, err := beginRead(ctx, s.Store, accountId) + if err != nil { + return nil, nmdata.Network{}, fmt.Errorf("%w: %w", errPrimarySerialRead, err) + } + rollbackTx(ctx, primaryTx) + + start := time.Now() + deadline := start.Add(s.ReplicaWait) + + for { + tx, network, err := beginReadBefore(ctx, s.Replica, accountId, deadline) + if err != nil { + if ctx.Err() == nil && errors.Is(err, context.DeadlineExceeded) { + s.Metrics.recordReplicaWait(ctx, time.Since(start)) + return nil, nmdata.Network{}, fmt.Errorf("%w: no answer within %s: %w", errReplicaLagging, s.ReplicaWait, err) + } + return nil, nmdata.Network{}, err + } + if network.Serial >= primaryNetwork.Serial { + s.Metrics.recordReplicaWait(ctx, time.Since(start)) + return tx, network, nil + } + rollbackTx(ctx, tx) + + remaining := time.Until(deadline) + if remaining <= 0 { + s.Metrics.recordReplicaWait(ctx, time.Since(start)) + return nil, nmdata.Network{}, fmt.Errorf("%w: at network serial %d after %s, primary at %d", errReplicaLagging, network.Serial, s.ReplicaWait, primaryNetwork.Serial) + } + + select { + case <-ctx.Done(): + return nil, nmdata.Network{}, ctx.Err() + case <-time.After(min(replicaPollInterval, remaining)): + } + } +} + +// beginReadBefore runs beginRead with its connect, begin and network read bounded +// by deadline, so a stalled replica cannot hold the read past it. +func beginReadBefore(ctx context.Context, store NetworkMapDBStore, accountId string, deadline time.Time) (NetworkMapDBStoreConn, nmdata.Network, error) { + attemptCtx, cancel := context.WithDeadline(ctx, deadline) + defer cancel() + return beginRead(attemptCtx, store, accountId) +} + +// beginRead opens a read transaction and reads the account's network as its +// first statement, so the network serial belongs to the transaction's snapshot. +func beginRead(ctx context.Context, store NetworkMapDBStore, accountId string) (NetworkMapDBStoreConn, nmdata.Network, error) { + tx, err := store.BeginTx(ctx) + if err != nil { + return nil, nmdata.Network{}, err + } + + network, err := tx.GetNetwork(ctx, accountId) + if err != nil { + rollbackTx(ctx, tx) + return nil, nmdata.Network{}, fmt.Errorf("get network: %w", err) + } + + return tx, network, nil +} + +// readFromTx reads the remaining network map data in tx and ends it. +func readFromTx(ctx context.Context, tx NetworkMapDBStoreConn, accountId string, network nmdata.Network) (*networkmap.NetworkMapData, error) { acctSettings, err := tx.GetAccountSettings(ctx, accountId) if err != nil { return rollbackAndReturnError(ctx, tx, fmt.Errorf("failed to get account settings: %w", err)) @@ -41,10 +178,6 @@ func (s *NetworkMapDBStoreImpl) GetNetworkMapData(ctx context.Context, accountId if err != nil { return rollbackAndReturnError(ctx, tx, fmt.Errorf("failed to get network routers: %w", err)) } - network, err := tx.GetNetwork(ctx, accountId) - if err != nil { - return rollbackAndReturnError(ctx, tx, fmt.Errorf("failed to get network: %w", err)) - } peers, _, err := tx.GetPeers(ctx, accountId) if err != nil { return rollbackAndReturnError(ctx, tx, fmt.Errorf("failed to get peers: %w", err)) @@ -111,26 +244,20 @@ func (s *NetworkMapDBStoreImpl) GetNetworkMapData(ctx context.Context, accountId ProxyTargetedDomainResourceIDs: proxyTargetedDomainResourceIDs, } - extraSettings, err := s.ExtraSettingsManager.GetExtraSettings(ctx, accountId) - if err != nil { - return nil, err - } - - toret.ValidatedPeers, err = s.IntegratedPeerValidator.GetValidatedPeers(ctx, accountId, maps.Values(toret.Groups), maps.Values(toret.Peers), extraSettings) - if err != nil { - return nil, err - } - return &toret, nil } func rollbackAndReturnError(ctx context.Context, tx NetworkMapDBStoreConn, err error) (*networkmap.NetworkMapData, error) { - if errr := tx.RollbackTx(ctx); errr != nil { - log.WithContext(ctx).Warnf("failed to rollback network map read transaction: %v", errr) - } + rollbackTx(ctx, tx) return nil, err } +func rollbackTx(ctx context.Context, tx NetworkMapDBStoreConn) { + if err := tx.RollbackTx(ctx); err != nil { + log.WithContext(ctx).Warnf("failed to rollback network map read transaction: %v", err) + } +} + func toMap[T any](all []T, id func(t T) string) map[string]*T { toret := make(map[string]*T, len(all)) for _, t := range all { diff --git a/management/internals/network_map_db/network_map_data_test.go b/management/internals/network_map_db/network_map_data_test.go index 2b694f5bc..d2d24ecc6 100644 --- a/management/internals/network_map_db/network_map_data_test.go +++ b/management/internals/network_map_db/network_map_data_test.go @@ -1,10 +1,21 @@ package networkmapdb import ( + "context" + "errors" + "sync/atomic" "testing" + "testing/synctest" + "time" - "github.com/netbirdio/netbird/shared/management/networkmap/nmdata" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/attribute" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" + + "github.com/netbirdio/netbird/shared/management/networkmap" + "github.com/netbirdio/netbird/shared/management/networkmap/nmdata" ) // disabled network resource shouldn't be in the resulting map @@ -156,3 +167,268 @@ func TestBuildResourcePolicies_ViaPolicyToDestinationGroupIdx(t *testing.T) { }, }, resourceToPolicies) } + +type fakeNmdataStore struct { + name string + serial atomic.Int64 + readErr error + stall bool + begins atomic.Int32 + finishes atomic.Int32 +} + +func (f *fakeNmdataStore) BeginTx(ctx context.Context) (NetworkMapDBStoreConn, error) { + if f.stall { + <-ctx.Done() + return nil, ctx.Err() + } + f.begins.Add(1) + return &fakeNmdataConn{store: f}, nil +} + +func (f *fakeNmdataStore) Exec(context.Context, string, ...any) error { return nil } + +type fakeNmdataConn struct { + store *fakeNmdataStore +} + +func (c *fakeNmdataConn) GetNetwork(context.Context, string) (nmdata.Network, error) { + return nmdata.Network{Identifier: c.store.name, Serial: c.store.serial.Load()}, nil +} + +func (c *fakeNmdataConn) GetPeers(context.Context, string) ([]nmdata.Peer, map[string][]*nmdata.Peer, error) { + return nil, nil, c.store.readErr +} + +func (c *fakeNmdataConn) GetGroups(context.Context, string) ([]nmdata.Group, map[string]map[string]any, error) { + return nil, nil, nil +} +func (c *fakeNmdataConn) GetDomains(context.Context, string) ([]Domain, error) { return nil, nil } +func (c *fakeNmdataConn) GetPolicies(context.Context, string) ([]nmdata.Policy, map[string]map[string]any, map[string]map[string]any, error) { + return nil, nil, nil, nil +} +func (c *fakeNmdataConn) GetRoutes(context.Context, string) ([]nmdata.Route, error) { return nil, nil } +func (c *fakeNmdataConn) GetNameServerGroups(context.Context, string) ([]nmdata.NameServerGroup, error) { + return nil, nil +} +func (c *fakeNmdataConn) GetNetworkResources(context.Context, string) ([]nmdata.NetworkResource, error) { + return nil, nil +} +func (c *fakeNmdataConn) GetNetworkRouters(context.Context, string) (map[string]map[string]*nmdata.NetworkRouter, error) { + return map[string]map[string]*nmdata.NetworkRouter{}, nil +} +func (c *fakeNmdataConn) GetAppliedZoneCandidates(context.Context, string) ([]networkmap.AppliedZoneCandidate, error) { + return nil, nil +} +func (c *fakeNmdataConn) GetAccountSettings(context.Context, string) (nmdata.AccountSettingsInfo, error) { + return nmdata.AccountSettingsInfo{}, nil +} +func (c *fakeNmdataConn) GetPostureChecks(context.Context, string) ([]nmdata.PostureChecks, map[string]string, error) { + return nil, nil, nil +} +func (c *fakeNmdataConn) GetAllowedUsers(context.Context, string) (map[string]struct{}, map[string][]string, error) { + return nil, nil, nil +} +func (c *fakeNmdataConn) GetDnsSettings(context.Context, string) (nmdata.DNSSettings, error) { + return nmdata.DNSSettings{}, nil +} +func (c *fakeNmdataConn) GetNetworkXIDToPublicIdMap(context.Context, string) (map[string]string, error) { + return map[string]string{}, nil +} +func (c *fakeNmdataConn) GetPrivateServices(context.Context, string) ([]Service, error) { + return nil, nil +} +func (c *fakeNmdataConn) GetProxyTargetedDomainResourceIDs(context.Context, string) (map[string]struct{}, error) { + return map[string]struct{}{}, nil +} + +func (c *fakeNmdataConn) CommitTx(context.Context) error { + c.store.finishes.Add(1) + return nil +} + +func (c *fakeNmdataConn) RollbackTx(context.Context) error { + c.store.finishes.Add(1) + return nil +} + +func newReplicaTestStores(primarySerial, replicaSerial int64) (*NetworkMapDBStoreImpl, *fakeNmdataStore, *fakeNmdataStore) { + primary := &fakeNmdataStore{name: "primary"} + primary.serial.Store(primarySerial) + replica := &fakeNmdataStore{name: "replica"} + replica.serial.Store(replicaSerial) + return &NetworkMapDBStoreImpl{Store: primary, Replica: replica, ReplicaWait: time.Second}, primary, replica +} + +func TestReadNetworkMapData_ReplicaCaughtUp(t *testing.T) { + s, primary, replica := newReplicaTestStores(5, 5) + + data, err := s.readNetworkMapData(context.Background(), "account") + require.NoError(t, err) + + assert.Equal(t, "replica", data.Network.Identifier) + assert.Equal(t, int32(1), replica.begins.Load()) + assert.Equal(t, primary.begins.Load(), primary.finishes.Load()) + assert.Equal(t, replica.begins.Load(), replica.finishes.Load()) +} + +func TestReadNetworkMapData_WaitsForReplica(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + s, _, replica := newReplicaTestStores(5, 4) + + time.AfterFunc(100*time.Millisecond, func() { replica.serial.Store(5) }) + + start := time.Now() + data, err := s.readNetworkMapData(context.Background(), "account") + require.NoError(t, err) + + assert.Equal(t, "replica", data.Network.Identifier) + assert.Equal(t, int64(5), data.Network.Serial) + assert.GreaterOrEqual(t, time.Since(start), 100*time.Millisecond) + assert.Equal(t, replica.begins.Load(), replica.finishes.Load()) + }) +} + +func TestReadNetworkMapData_FallsBackToPrimaryWhenReplicaLags(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + s, primary, replica := newReplicaTestStores(5, 4) + + start := time.Now() + data, err := s.readNetworkMapData(context.Background(), "account") + require.NoError(t, err) + + assert.Equal(t, "primary", data.Network.Identifier) + assert.GreaterOrEqual(t, time.Since(start), s.ReplicaWait) + assert.Equal(t, primary.begins.Load(), primary.finishes.Load()) + assert.Equal(t, replica.begins.Load(), replica.finishes.Load()) + }) +} + +func TestReadNetworkMapData_FallsBackToPrimaryOnReplicaReadError(t *testing.T) { + s, primary, replica := newReplicaTestStores(5, 5) + replica.readErr = errors.New("canceling statement due to conflict with recovery") + + data, err := s.readNetworkMapData(context.Background(), "account") + require.NoError(t, err) + + assert.Equal(t, "primary", data.Network.Identifier) + assert.Equal(t, primary.begins.Load(), primary.finishes.Load()) + assert.Equal(t, replica.begins.Load(), replica.finishes.Load()) +} + +func TestReadNetworkMapData_WithoutReplicaReadsPrimaryOnce(t *testing.T) { + primary := &fakeNmdataStore{name: "primary"} + s := &NetworkMapDBStoreImpl{Store: primary} + + data, err := s.readNetworkMapData(context.Background(), "account") + require.NoError(t, err) + + assert.Equal(t, "primary", data.Network.Identifier) + assert.Equal(t, int32(1), primary.begins.Load()) + assert.Equal(t, int32(1), primary.finishes.Load()) +} + +func TestReadNetworkMapData_CancelledWhileWaitingForReplica(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + s, primary, replica := newReplicaTestStores(5, 4) + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + + _, err := s.readNetworkMapData(ctx, "account") + require.ErrorIs(t, err, context.DeadlineExceeded) + + assert.Equal(t, int32(1), primary.begins.Load(), "primary should only be read for the serial") + assert.Equal(t, primary.begins.Load(), primary.finishes.Load()) + assert.Equal(t, replica.begins.Load(), replica.finishes.Load()) + }) +} + +func TestReadNetworkMapData_FallsBackToPrimaryWhenReplicaStalls(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + reader := sdkmetric.NewManualReader() + metrics, err := NewMetrics(sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)).Meter("test"), nil) + require.NoError(t, err) + + s, primary, replica := newReplicaTestStores(5, 5) + s.Metrics = metrics + replica.stall = true + + start := time.Now() + data, err := s.readNetworkMapData(context.Background(), "account") + require.NoError(t, err) + + assert.Equal(t, "primary", data.Network.Identifier) + assert.Equal(t, s.ReplicaWait, time.Since(start), "a stalled replica must be abandoned at the wait deadline") + assert.Equal(t, primary.begins.Load(), primary.finishes.Load()) + + reads, waits := collectReadMetrics(t, reader) + assert.Equal(t, map[[2]string]int64{{readSourcePrimary, readReasonLagTimeout}: 1}, reads) + assert.Equal(t, uint64(1), waits) + }) +} + +func TestReadNetworkMapData_CountsReadsBySourceAndReason(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + reader := sdkmetric.NewManualReader() + metrics, err := NewMetrics(sdkmetric.NewMeterProvider(sdkmetric.WithReader(reader)).Meter("test"), nil) + require.NoError(t, err) + + s, _, replica := newReplicaTestStores(5, 5) + s.Metrics = metrics + + _, err = s.readNetworkMapData(context.Background(), "account") + require.NoError(t, err) + + replica.serial.Store(4) + _, err = s.readNetworkMapData(context.Background(), "account") + require.NoError(t, err) + + replica.serial.Store(5) + replica.readErr = errors.New("canceling statement due to conflict with recovery") + _, err = s.readNetworkMapData(context.Background(), "account") + require.NoError(t, err) + + s.Replica = nil + _, err = s.readNetworkMapData(context.Background(), "account") + require.NoError(t, err) + + reads, waits := collectReadMetrics(t, reader) + + assert.Equal(t, map[[2]string]int64{ + {readSourceReplica, readReasonCaughtUp}: 1, + {readSourcePrimary, readReasonLagTimeout}: 1, + {readSourcePrimary, readReasonReplicaError}: 1, + {readSourcePrimary, readReasonNoReplica}: 1, + }, reads) + assert.Equal(t, uint64(3), waits, "every replica attempt that reached the serial check records its wait") + }) +} + +// collectReadMetrics returns the read counts keyed by source and reason and the +// number of recorded replica waits. +func collectReadMetrics(t *testing.T, reader *sdkmetric.ManualReader) (map[[2]string]int64, uint64) { + t.Helper() + var rm metricdata.ResourceMetrics + require.NoError(t, reader.Collect(context.Background(), &rm)) + + reads := map[[2]string]int64{} + var waits uint64 + for _, sm := range rm.ScopeMetrics { + for _, m := range sm.Metrics { + switch data := m.Data.(type) { + case metricdata.Sum[int64]: + for _, dp := range data.DataPoints { + source, _ := dp.Attributes.Value(attribute.Key("source")) + reason, _ := dp.Attributes.Value(attribute.Key("reason")) + reads[[2]string{source.AsString(), reason.AsString()}] = dp.Value + } + case metricdata.Histogram[int64]: + for _, dp := range data.DataPoints { + waits += dp.Count + } + } + } + } + return reads, waits +} diff --git a/management/internals/network_map_db/shared_types.go b/management/internals/network_map_db/shared_types.go index dc065073c..ca92d4eea 100644 --- a/management/internals/network_map_db/shared_types.go +++ b/management/internals/network_map_db/shared_types.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "reflect" + "time" "github.com/miekg/dns" "github.com/netbirdio/netbird/management/server/integrations/integrated_validator" @@ -17,6 +18,13 @@ import ( var ErrDnsUnsupportedRecordType = errors.New("unsupported record type") +const replicaPollInterval = 10 * time.Millisecond + +var ( + errReplicaLagging = errors.New("replica lagging") + errPrimarySerialRead = errors.New("read primary network serial") +) + type NetworkMapDBStore interface { //nolint:revive // established name across the codebase BeginTx(ctx context.Context) (NetworkMapDBStoreConn, error) Exec(ctx context.Context, query string, args ...any) error @@ -46,7 +54,14 @@ type NetworkMapDBStoreConn interface { //nolint:revive // established name acros } type NetworkMapDBStoreImpl struct { //nolint:revive // established name across the codebase - Store NetworkMapDBStore + Store NetworkMapDBStore + // Replica is an optional read replica of Store that serves reads once it has + // replicated the account's current network serial. + Replica NetworkMapDBStore + // ReplicaWait bounds how long a read waits for Replica before it falls back to Store. + ReplicaWait time.Duration + // Metrics records how reads are served. It is optional. + Metrics *Metrics IntegratedPeerValidator integrated_validator.IntegratedValidator ExtraSettingsManager settings.Manager } diff --git a/management/internals/server/boot.go b/management/internals/server/boot.go index 28f6294d6..64d5ee4bc 100644 --- a/management/internals/server/boot.go +++ b/management/internals/server/boot.go @@ -8,12 +8,14 @@ import ( "errors" "net/http" "net/netip" + "os" "slices" "time" "github.com/gorilla/mux" grpcMiddleware "github.com/grpc-ecosystem/go-grpc-middleware/v2" "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/realip" + "github.com/jackc/pgx/v5/pgxpool" "github.com/rs/cors" "github.com/rs/xid" log "github.com/sirupsen/logrus" @@ -44,6 +46,7 @@ import ( "github.com/netbirdio/netbird/management/server/idp" "github.com/netbirdio/netbird/management/server/store" "github.com/netbirdio/netbird/management/server/telemetry" + "github.com/netbirdio/netbird/management/server/types" mgmtProto "github.com/netbirdio/netbird/shared/management/proto" "github.com/netbirdio/netbird/shared/ratelimit" "github.com/netbirdio/netbird/util/crypt" @@ -130,6 +133,8 @@ func (s *BaseServer) NetworkMapStore() *networkmapdb.NetworkMapDBStoreImpl { s.Config.Datadir, s.IntegratedValidator(), s.SettingsManager(), + s.NetworkMapReplicaPool(), + s.Metrics().GetMeter(), ) // networkmap db store supports postgres and sqlite backends only // for other backends a fallback is used, so NotSupportedStoreEngineError @@ -141,6 +146,25 @@ func (s *BaseServer) NetworkMapStore() *networkmapdb.NetworkMapDBStoreImpl { }) } +// NetworkMapReplicaPool opens the pgx pool to the Postgres read replica serving +// network map data reads. It is nil unless the store engine is Postgres and +// networkmapdbfactory.ReplicaDsnEnv is set. +func (s *BaseServer) NetworkMapReplicaPool() *pgxpool.Pool { + return CreateNamed(s, "NetworkMapReplica", func() *pgxpool.Pool { + dsn := os.Getenv(networkmapdbfactory.ReplicaDsnEnv) + if dsn == "" || s.Config.StoreConfig.Engine != types.PostgresStoreEngine { + return nil + } + pool, err := db.OpenPgxPool(context.Background(), dsn, db.DefaultPoolConfig) + if err != nil { + log.Errorf("network map data reads use the primary, connecting to the replica failed: %v", err) + return nil + } + s.OnStop(pool.Close) + return pool + }) +} + func (s *BaseServer) EventStore() activity.Store { return Create(s, func() activity.Store { var err error diff --git a/management/internals/shared/db/open.go b/management/internals/shared/db/open.go index 3c3299bb9..f813b0ce8 100644 --- a/management/internals/shared/db/open.go +++ b/management/internals/shared/db/open.go @@ -106,7 +106,7 @@ func OpenPostgres(ctx context.Context, dsn string, pool PoolConfig, opts ...Opti if err != nil { return nil, err } - pgxPool, err := newPgxPool(ctx, dsn, pool) + pgxPool, err := OpenPgxPool(ctx, dsn, pool) if err != nil { closeGorm(gormDB) return nil, err @@ -135,7 +135,8 @@ func OpenPostgresWithoutPool(ctx context.Context, dsn string, opts ...Option) (* return conn, nil } -func newPgxPool(ctx context.Context, dsn string, cfg PoolConfig) (*pgxpool.Pool, error) { +// OpenPgxPool opens a pgx pool to dsn sized by cfg and verifies it with a ping. +func OpenPgxPool(ctx context.Context, dsn string, cfg PoolConfig) (*pgxpool.Pool, error) { config, err := pgxpool.ParseConfig(dsn) if err != nil { return nil, fmt.Errorf("unable to parse database config: %w", err) diff --git a/management/internals/shared/requestbuffer/buffer.go b/management/internals/shared/requestbuffer/buffer.go index c3823776c..22f9d1952 100644 --- a/management/internals/shared/requestbuffer/buffer.go +++ b/management/internals/shared/requestbuffer/buffer.go @@ -26,6 +26,9 @@ type Buffer[T any] struct { mu sync.Mutex waiting map[string][]chan result[T] + // inFlight holds the keys with a running fetch and whether their next flush + // is due once it ends. It is nil unless fetches are serialized. + inFlight map[string]bool } type result[T any] struct { @@ -45,6 +48,15 @@ func New[T any](ctx context.Context, name string, interval time.Duration, fetch } } +// NewSerialized returns a Buffer like New that runs at most one fetch per key at +// a time. Requests arriving during a fetch form the next batch, which is fetched +// once both its window closed and the running fetch ended. +func NewSerialized[T any](ctx context.Context, name string, interval time.Duration, fetch FetchFunc[T]) *Buffer[T] { + b := New(ctx, name, interval, fetch) + b.inFlight = make(map[string]bool) + return b +} + // Get returns the value for key, sharing one fetch with the other callers of // the current batch. The value is shared as is, so callers must treat it as // read-only unless the fetch hands out copies. @@ -71,8 +83,16 @@ func (b *Buffer[T]) Get(ctx context.Context, key string) (T, error) { func (b *Buffer[T]) flush(key string) { b.mu.Lock() + if _, running := b.inFlight[key]; running { + b.inFlight[key] = true + b.mu.Unlock() + return + } waiting := b.waiting[key] delete(b.waiting, key) + if b.inFlight != nil && len(waiting) > 0 { + b.inFlight[key] = false + } b.mu.Unlock() if len(waiting) == 0 { @@ -86,6 +106,17 @@ func (b *Buffer[T]) flush(key string) { for _, ch := range waiting { ch <- result[T]{value: value, err: err} } + + if b.inFlight == nil { + return + } + b.mu.Lock() + due := b.inFlight[key] + delete(b.inFlight, key) + b.mu.Unlock() + if due { + go b.flush(key) + } } // Interval reads a buffer interval from envVar, falling back to def. diff --git a/management/internals/shared/requestbuffer/buffer_test.go b/management/internals/shared/requestbuffer/buffer_test.go index 9e356145e..3bfb73de4 100644 --- a/management/internals/shared/requestbuffer/buffer_test.go +++ b/management/internals/shared/requestbuffer/buffer_test.go @@ -6,6 +6,7 @@ import ( "sync" "sync/atomic" "testing" + "testing/synctest" "time" "github.com/stretchr/testify/assert" @@ -104,3 +105,73 @@ func TestBufferHonorsCallerContext(t *testing.T) { _, err := buffer.Get(ctx, "account") assert.ErrorIs(t, err, context.DeadlineExceeded) } + +func TestSerializedBufferQueuesRequestsDuringFetch(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + var fetches, running, maxRunning atomic.Int32 + buffer := NewSerialized(context.Background(), "test", 10*time.Millisecond, + func(ctx context.Context, key string) (int32, error) { + n := running.Add(1) + defer running.Add(-1) + if n > maxRunning.Load() { + maxRunning.Store(n) + } + time.Sleep(time.Second) + return fetches.Add(1), nil + }) + + var wg sync.WaitGroup + get := func(want int32) { + wg.Add(1) + go func() { + defer wg.Done() + value, err := buffer.Get(context.Background(), "account") + assert.NoError(t, err) + assert.Equal(t, want, value) + }() + } + + get(1) + time.Sleep(100 * time.Millisecond) + synctest.Wait() + require.Equal(t, int32(1), running.Load(), "first fetch should be running") + + for range 5 { + get(2) + } + wg.Wait() + + assert.Equal(t, int32(2), fetches.Load(), "requests queued during the fetch should share one fetch") + assert.Equal(t, int32(1), maxRunning.Load(), "fetches of a key should not overlap") + }) +} + +func TestUnserializedBufferOverlapsFetches(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + var running, maxRunning atomic.Int32 + buffer := New(context.Background(), "test", 10*time.Millisecond, + func(ctx context.Context, key string) (string, error) { + n := running.Add(1) + defer running.Add(-1) + if n > maxRunning.Load() { + maxRunning.Store(n) + } + time.Sleep(time.Second) + return key, nil + }) + + var wg sync.WaitGroup + wg.Go(func() { + _, err := buffer.Get(context.Background(), "account") + assert.NoError(t, err) + }) + time.Sleep(100 * time.Millisecond) + wg.Go(func() { + _, err := buffer.Get(context.Background(), "account") + assert.NoError(t, err) + }) + wg.Wait() + + assert.Equal(t, int32(2), maxRunning.Load()) + }) +} diff --git a/management/server/account.go b/management/server/account.go index ddc8d4bbc..21b9fa431 100644 --- a/management/server/account.go +++ b/management/server/account.go @@ -1825,10 +1825,10 @@ func (am *DefaultAccountManager) SyncUserJWTGroups(ctx context.Context, userAuth return fmt.Errorf("reconcile IPv6 for group changes: %w", err) } change.ChangedPeerIDs = append(change.ChangedPeerIDs, ipv6Changed...) + } - if err = transaction.IncrementNetworkSerial(ctx, userAuth.AccountId); err != nil { - return fmt.Errorf("error incrementing network serial: %w", err) - } + if err = transaction.IncrementNetworkSerial(ctx, userAuth.AccountId); err != nil { + return fmt.Errorf("error incrementing network serial: %w", err) } if snap, err = affectedpeers.Load(ctx, transaction, userAuth.AccountId, change); err != nil { @@ -2844,6 +2844,10 @@ func (am *DefaultAccountManager) savePeerIPUpdate(ctx context.Context, transacti return fmt.Errorf("save peer: %w", err) } + if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + eventMeta["old_ip"] = oldIP eventMeta["ip"] = newIP.String() am.StoreEvent(ctx, userID, peer.ID, accountID, activity.PeerIPUpdated, eventMeta) @@ -2884,7 +2888,8 @@ func (am *DefaultAccountManager) UpdatePeerIPv6(ctx context.Context, accountID, // updatePeerIPv6InTransaction validates and applies an IPv6 address change within a store transaction. func (am *DefaultAccountManager) updatePeerIPv6InTransaction(ctx context.Context, transaction store.Store, accountID, peerID string, newIPv6 netip.Addr) (bool, error) { - network, err := transaction.GetAccountNetwork(ctx, store.LockingStrengthShare, accountID) + // Locked for update because the serial increment below writes the same row. + network, err := transaction.GetAccountNetwork(ctx, store.LockingStrengthUpdate, accountID) if err != nil { return false, fmt.Errorf("get network: %w", err) } @@ -2928,6 +2933,10 @@ func (am *DefaultAccountManager) updatePeerIPv6InTransaction(ctx context.Context return false, fmt.Errorf("save peer: %w", err) } + if err := transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return false, fmt.Errorf("increment network serial: %w", err) + } + return true, nil } diff --git a/management/server/account_test.go b/management/server/account_test.go index 7acc72d1f..c343575b8 100644 --- a/management/server/account_test.go +++ b/management/server/account_test.go @@ -3690,7 +3690,7 @@ func createManagerWithNetworkMapStore(t testing.TB) (*DefaultAccountManager, *up store, err := createStoreAt(t, dataDir) require.NoError(t, err) - nmdataStore, err := networkmapdbfactory.NewNetworkMapDBStore(context.Background(), types.SqliteStoreEngine, dataDir, MockIntegratedValidator{}, newSettingsMockManager(t)) + nmdataStore, err := networkmapdbfactory.NewNetworkMapDBStore(context.Background(), types.SqliteStoreEngine, dataDir, MockIntegratedValidator{}, newSettingsMockManager(t), nil, nil) require.NoError(t, err) sqliteStore, ok := nmdataStore.Store.(*networkmap_sqlite.SqliteStore) require.True(t, ok, "network map store is %T, want *networkmap_sqlite.SqliteStore", nmdataStore.Store) @@ -5110,3 +5110,46 @@ func TestDefaultAccountManager_ExpirationJobsStopForDeletedAccount(t *testing.T) }) } } + +func networkSerial(t *testing.T, manager *DefaultAccountManager, accountID string) uint64 { + t.Helper() + + network, err := manager.Store.GetAccountNetwork(context.Background(), store.LockingStrengthNone, accountID) + require.NoError(t, err) + return network.Serial +} + +func TestDefaultAccountManager_UpdatePeerIP_BumpsNetworkSerial(t *testing.T) { + manager, _, account, peer1, peer2, peer3 := setupNetworkMapTest(t) + + newIP, err := types.AllocatePeerIP(netip.MustParsePrefix(account.Network.Net.String()), []netip.Addr{peer1.IP, peer2.IP, peer3.IP}) + require.NoError(t, err) + + before := networkSerial(t, manager, account.Id) + require.NoError(t, manager.UpdatePeerIP(context.Background(), account.Id, userID, peer1.ID, newIP)) + assert.Greater(t, networkSerial(t, manager, account.Id), before) +} + +func TestDefaultAccountManager_SyncUserJWTGroups_BumpsNetworkSerialWithoutPropagation(t *testing.T) { + manager, _, err := createManager(t) + require.NoError(t, err) + + accountID, err := manager.GetAccountIDByUserID(context.Background(), auth.UserAuth{UserId: "jwt-user", Domain: "test.domain"}) + require.NoError(t, err) + acc, err := manager.Store.GetAccount(context.Background(), accountID) + require.NoError(t, err) + acc.Settings.JWTGroupsEnabled = true + acc.Settings.JWTGroupsClaimName = "idp-groups" + acc.Settings.GroupsPropagationEnabled = false + require.NoError(t, manager.Store.SaveAccount(context.Background(), acc)) + + before := networkSerial(t, manager, accountID) + err = manager.SyncUserJWTGroups(context.Background(), auth.UserAuth{ + AccountId: accountID, + Domain: "test.domain", + UserId: "jwt-user", + Groups: []string{"group1"}, + }) + require.NoError(t, err) + assert.Greater(t, networkSerial(t, manager, accountID), before) +} diff --git a/management/server/peer.go b/management/server/peer.go index 5b33ca9f8..c3454179e 100644 --- a/management/server/peer.go +++ b/management/server/peer.go @@ -103,7 +103,9 @@ func (am *DefaultAccountManager) MarkPeerConnected(ctx context.Context, peerPubK return err } - updated, err := am.Store.MarkPeerConnectedIfNewerSession(ctx, accountID, peer.ID, sessionStartedAt) + updated, err := am.writePeerStatus(ctx, accountID, peer, func(st store.Store) (bool, error) { + return st.MarkPeerConnectedIfNewerSession(ctx, accountID, peer.ID, sessionStartedAt) + }) if err != nil { am.metrics.AccountManagerMetrics().CountPeerStatusUpdate(telemetry.PeerStatusConnect, telemetry.PeerStatusError) return err @@ -140,6 +142,28 @@ func (am *DefaultAccountManager) MarkPeerConnected(ctx context.Context, peerPubK return nil } +// writePeerStatus runs a peer status write. The connection state of an embedded +// proxy peer is part of the network map, so for those peers an applied write also +// increments the account network serial in the same transaction. +func (am *DefaultAccountManager) writePeerStatus(ctx context.Context, accountID string, peer *nbpeer.Peer, write func(store.Store) (bool, error)) (bool, error) { + if !peer.ProxyMeta.Embedded { + return write(am.Store) + } + + var updated bool + err := am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error { + var err error + if updated, err = write(transaction); err != nil || !updated { + return err + } + if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + return nil + }) + return updated, err +} + // schedulePeerExpirations arms the account's login/inactivity expiration // timers for an SSO peer that just connected. func (am *DefaultAccountManager) schedulePeerExpirations(ctx context.Context, accountID string, peer *nbpeer.Peer) error { @@ -179,7 +203,9 @@ func (am *DefaultAccountManager) MarkPeerDisconnected(ctx context.Context, peerP return err } - updated, err := am.Store.MarkPeerDisconnectedIfSameSession(ctx, accountID, peer.ID, sessionStartedAt) + updated, err := am.writePeerStatus(ctx, accountID, peer, func(st store.Store) (bool, error) { + return st.MarkPeerDisconnectedIfSameSession(ctx, accountID, peer.ID, sessionStartedAt) + }) if err != nil { am.metrics.AccountManagerMetrics().CountPeerStatusUpdate(telemetry.PeerStatusDisconnect, telemetry.PeerStatusError) return err @@ -349,7 +375,7 @@ func (am *DefaultAccountManager) UpdatePeer(ctx context.Context, accountID, user } if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { - return fmt.Errorf("failed to increment network serial: %w", err) + return fmt.Errorf("increment network serial: %w", err) } return transaction.SavePeer(ctx, accountID, peer) @@ -618,7 +644,7 @@ func (am *DefaultAccountManager) deletePeerInTransaction(ctx context.Context, ac } if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { - return fmt.Errorf("failed to increment network serial: %w", err) + return fmt.Errorf("increment network serial: %w", err) } return nil @@ -956,7 +982,7 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, accountID, setupKe err = transaction.IncrementNetworkSerial(ctx, accountID) if err != nil { - return fmt.Errorf("failed to increment network serial: %w", err) + return fmt.Errorf("increment network serial: %w", err) } if ephemeral { @@ -1097,6 +1123,9 @@ func (am *DefaultAccountManager) SyncPeer(ctx context.Context, sync types.PeerSy if err = transaction.SavePeer(ctx, accountID, peer); err != nil { return err } + if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } } return nil }) @@ -1272,6 +1301,9 @@ func (am *DefaultAccountManager) LoginPeer(ctx context.Context, login types.Peer if err = transaction.SavePeer(ctx, accountID, peer); err != nil { return err } + if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } } return nil diff --git a/management/server/peer_test.go b/management/server/peer_test.go index b75dd1b72..bf8093c1d 100644 --- a/management/server/peer_test.go +++ b/management/server/peer_test.go @@ -3255,3 +3255,71 @@ func TestAddPeer_ExhaustedIPAllocationFails(t *testing.T) { require.Len(t, peers, 1) assert.Equal(t, netip.MustParseAddr("100.64.0.1"), peers[0].IP) } + +func TestPeerWrites_BumpNetworkSerial(t *testing.T) { + manager, _, account, peer1, peer2, _ := setupNetworkMapTest(t) + ctx := context.Background() + + t.Run("SyncPeer with changed meta bumps, unchanged does not", func(t *testing.T) { + current, err := manager.Store.GetPeerByID(ctx, store.LockingStrengthNone, account.Id, peer1.ID) + require.NoError(t, err) + + before := networkSerial(t, manager, account.Id) + _, _, _, _, err = manager.SyncPeer(ctx, types.PeerSync{WireGuardPubKey: peer1.Key, Meta: current.Meta}, account.Id) + require.NoError(t, err) + assert.Equal(t, before, networkSerial(t, manager, account.Id)) + + meta := current.Meta + meta.Hostname = "renamed-host" + _, _, _, _, err = manager.SyncPeer(ctx, types.PeerSync{WireGuardPubKey: peer1.Key, Meta: meta}, account.Id) + require.NoError(t, err) + assert.Greater(t, networkSerial(t, manager, account.Id), before) + }) + + t.Run("LoginPeer with changed ssh key bumps, unchanged does not", func(t *testing.T) { + current, err := manager.Store.GetPeerByID(ctx, store.LockingStrengthNone, account.Id, peer1.ID) + require.NoError(t, err) + + before := networkSerial(t, manager, account.Id) + _, _, _, _, err = manager.LoginPeer(ctx, types.PeerLogin{WireGuardPubKey: peer1.Key, SSHKey: current.SSHKey, Meta: current.Meta}) + require.NoError(t, err) + assert.Equal(t, before, networkSerial(t, manager, account.Id)) + + _, _, _, _, err = manager.LoginPeer(ctx, types.PeerLogin{WireGuardPubKey: peer1.Key, SSHKey: "ssh-ed25519 AAAAnew", Meta: current.Meta}) + require.NoError(t, err) + assert.Greater(t, networkSerial(t, manager, account.Id), before) + }) + + t.Run("MarkPeerConnected and MarkPeerDisconnected", func(t *testing.T) { + session := time.Now().UTC().UnixNano() + + before := networkSerial(t, manager, account.Id) + require.NoError(t, manager.MarkPeerConnected(ctx, peer2.Key, account.Id, session, nil)) + require.NoError(t, manager.MarkPeerDisconnected(ctx, peer2.Key, account.Id, session)) + assert.Equal(t, before, networkSerial(t, manager, account.Id), "regular peers must not bump") + + expired, err := manager.Store.GetPeerByID(ctx, store.LockingStrengthNone, account.Id, peer2.ID) + require.NoError(t, err) + expired.Status.LoginExpired = true + require.NoError(t, manager.Store.SavePeer(ctx, account.Id, expired)) + + session++ + require.NoError(t, manager.MarkPeerConnected(ctx, peer2.Key, account.Id, session, nil)) + afterExpired := networkSerial(t, manager, account.Id) + assert.Equal(t, before, afterExpired, "login-expired reconnect writes no network map data") + + embedded, err := manager.Store.GetPeerByID(ctx, store.LockingStrengthNone, account.Id, peer2.ID) + require.NoError(t, err) + embedded.Status.LoginExpired = false + embedded.ProxyMeta.Embedded = true + require.NoError(t, manager.Store.SavePeer(ctx, account.Id, embedded)) + + require.NoError(t, manager.MarkPeerDisconnected(ctx, peer2.Key, account.Id, session)) + afterDisconnect := networkSerial(t, manager, account.Id) + assert.Greater(t, afterDisconnect, afterExpired, "embedded proxy disconnect must bump") + + session++ + require.NoError(t, manager.MarkPeerConnected(ctx, peer2.Key, account.Id, session, nil)) + assert.Greater(t, networkSerial(t, manager, account.Id), afterDisconnect, "embedded proxy connect must bump") + }) +} diff --git a/management/server/user.go b/management/server/user.go index 3e84fcd03..89149367d 100644 --- a/management/server/user.go +++ b/management/server/user.go @@ -670,6 +670,12 @@ func (am *DefaultAccountManager) SaveOrAddUsers(ctx context.Context, accountID, var updatedUsersInfo = make([]*types.UserInfo, 0, len(usersToSave)) + if len(usersToSave) > 0 { + if err = am.Store.IncrementNetworkSerial(ctx, accountID); err != nil { + return nil, fmt.Errorf("increment network serial: %w", err) + } + } + userInfos, err := am.GetUsersFromAccount(ctx, accountID, initiatorUserID) if err != nil { return nil, err @@ -693,9 +699,6 @@ func (am *DefaultAccountManager) SaveOrAddUsers(ctx context.Context, accountID, return nil, err } } else if len(usersToSave) > 0 { - if err = am.Store.IncrementNetworkSerial(ctx, accountID); err != nil { - return nil, fmt.Errorf("failed to increment network serial: %w", err) - } go am.dispatchAffectedPeerIDs(ctx, accountID, affected.ids()) } @@ -1425,6 +1428,10 @@ func (am *DefaultAccountManager) deleteRegularUser(ctx context.Context, accountI return fmt.Errorf("failed to delete user: %s %w", targetUserInfo.ID, err) } + if err = transaction.IncrementNetworkSerial(ctx, accountID); err != nil { + return fmt.Errorf("increment network serial: %w", err) + } + return nil }) if err != nil { diff --git a/management/server/user_test.go b/management/server/user_test.go index 305ed5a24..2e1b51572 100644 --- a/management/server/user_test.go +++ b/management/server/user_test.go @@ -14,6 +14,7 @@ import ( "github.com/netbirdio/netbird/management/internals/controllers/network_map" nbcache "github.com/netbirdio/netbird/management/server/cache" + nbcontext "github.com/netbirdio/netbird/management/server/context" "github.com/netbirdio/netbird/management/server/permissions" "github.com/netbirdio/netbird/management/server/permissions/modules" "github.com/netbirdio/netbird/management/server/permissions/roles" @@ -2352,3 +2353,24 @@ func TestUser_Operations_WithEmbeddedIDP(t *testing.T) { t.Logf("Duplicate email error: %v", err) }) } + +func TestDefaultAccountManager_UserMutations_BumpNetworkSerial(t *testing.T) { + manager, _, account, _, _, _ := setupNetworkMapTest(t) + + account, err := manager.Store.GetAccount(context.Background(), account.Id) + require.NoError(t, err) + account.Users["regular-user"] = types.NewRegularUser("regular-user", "", "") + require.NoError(t, manager.Store.SaveAccount(context.Background(), account)) + + ctx := nbcontext.WithRole(context.Background(), string(types.UserRoleOwner)) + + before := networkSerial(t, manager, account.Id) + _, err = manager.SaveUser(ctx, account.Id, userID, &types.User{Id: "regular-user", Role: types.UserRoleUser, Blocked: true}) + require.NoError(t, err) + afterSave := networkSerial(t, manager, account.Id) + assert.Greater(t, afterSave, before, "SaveUser must bump the serial") + + _, err = manager.deleteRegularUser(ctx, account.Id, userID, &types.UserInfo{ID: "regular-user"}) + require.NoError(t, err) + assert.Greater(t, networkSerial(t, manager, account.Id), afterSave, "deleteRegularUser must bump the serial") +}