[management] Move reverse proxy access logs to a dedicated database (#8303)

This commit is contained in:
Bethuel Mmbaga
2026-10-23 16:04:09 +03:00
committed by GitHub
parent 8397bed4af
commit 69f13e3733
18 changed files with 421 additions and 18 deletions
+20
View File
@@ -7,6 +7,7 @@ import (
"github.com/stretchr/testify/require" "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" 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, "postgres-dsn", os.Getenv("NB_STORE_ENGINE_POSTGRES_DSN"))
require.Equal(t, "store.db", os.Getenv("NB_STORE_ENGINE_SQLITE_FILE")) 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))
}
+1
View File
@@ -71,6 +71,7 @@ type ServerConfig struct {
Auth AuthConfig `yaml:"auth"` Auth AuthConfig `yaml:"auth"`
Store StoreConfig `yaml:"store"` Store StoreConfig `yaml:"store"`
ActivityStore StoreConfig `yaml:"activityStore"` ActivityStore StoreConfig `yaml:"activityStore"`
AccessLogStore StoreConfig `yaml:"accessLogStore"`
AuthStore StoreConfig `yaml:"authStore"` AuthStore StoreConfig `yaml:"authStore"`
ReverseProxy ReverseProxyConfig `yaml:"reverseProxy"` ReverseProxy ReverseProxyConfig `yaml:"reverseProxy"`
+16
View File
@@ -26,6 +26,7 @@ import (
"github.com/netbirdio/netbird/encryption" "github.com/netbirdio/netbird/encryption"
agentnetworkpricing "github.com/netbirdio/netbird/management/internals/modules/agentnetwork/pricing" 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" mgmtServer "github.com/netbirdio/netbird/management/internals/server"
nbconfig "github.com/netbirdio/netbird/management/internals/server/config" nbconfig "github.com/netbirdio/netbird/management/internals/server/config"
"github.com/netbirdio/netbird/management/server/telemetry" "github.com/netbirdio/netbird/management/server/telemetry"
@@ -152,6 +153,20 @@ func applyActivityStoreEnv(storeConfig StoreConfig) error {
return nil 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 { func initializeConfig() error {
var err error var err error
config, err = LoadConfig(configPath) config, err = LoadConfig(configPath)
@@ -172,6 +187,7 @@ func initializeConfig() error {
if err := applyActivityStoreEnv(config.Server.ActivityStore); err != nil { if err := applyActivityStoreEnv(config.Server.ActivityStore); err != nil {
return err return err
} }
applyAccessLogStoreEnv(config.Server.AccessLogStore)
log.Infof("Starting combined NetBird server") log.Infof("Starting combined NetBird server")
logConfig(config) logConfig(config)
+7
View File
@@ -121,6 +121,13 @@ server:
# dsn: "" # Connection string for postgres # dsn: "" # Connection string for postgres
# file: "" # Custom SQLite file path (optional, defaults to {dataDir}/events.db) # 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) # Auth (embedded IdP) store configuration (optional, defaults to sqlite3 in dataDir/idp.db)
# authStore: # authStore:
# engine: "sqlite3" # sqlite3 or postgres # engine: "sqlite3" # sqlite3 or postgres
@@ -46,7 +46,7 @@ type AccessLogFilter struct {
SortOrder string // Sort order: asc or desc (default: desc) SortOrder string // Sort order: asc or desc (default: desc)
// Filtering parameters // 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 SourceIP *string // Filter by source IP address
Host *string // Filter by host header Host *string // Filter by host header
Path *string // Filter by request path (supports LIKE pattern) Path *string // Filter by request path (supports LIKE pattern)
@@ -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)}
}
@@ -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)
})
}
}
@@ -85,8 +85,8 @@ func applyFilters(query *gorm.DB, filter accesslogs.AccessLogFilter) *gorm.DB {
if filter.Search != nil { if filter.Search != nil {
searchPattern := "%" + *filter.Search + "%" searchPattern := "%" + *filter.Search + "%"
query = query.Where( 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 ?)", "id LIKE ? OR location_connection_ip LIKE ? OR host LIKE ? OR path LIKE ? OR CONCAT(host, path) LIKE ?",
searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, searchPattern, searchPattern,
) )
} }
@@ -123,3 +123,37 @@ func TestSqlRepository_ListByAccount_StatusFilter(t *testing.T) {
assert.Equal(t, []string{"info", "notfound"}, listIDs("failed")) assert.Equal(t, []string{"info", "notfound"}, listIDs("failed"))
assert.Equal(t, []string{"ok"}, listIDs("success")) 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"))
}
+19 -2
View File
@@ -180,7 +180,8 @@ func (s *BaseServer) APIHandler() http.Handler {
s.GroupsManager(), s.GeoLocationManager(), s.PermissionsManager(), s.SettingsManager(), s.ZonesManager(), s.GroupsManager(), s.GeoLocationManager(), s.PermissionsManager(), s.SettingsManager(), s.ZonesManager(),
s.RecordsManager(), s.NetworkMapController(), s.IdpManager(), s.ServiceManager(), s.ReverseProxyDomainManager(), s.RecordsManager(), s.NetworkMapController(), s.IdpManager(), s.ServiceManager(), s.ReverseProxyDomainManager(),
s.AccessLogsManager(), s.ReverseProxyGRPCServer(), s.Config.ReverseProxy.TrustedHTTPProxies, s.AccessLogsManager(), s.ReverseProxyGRPCServer(), s.Config.ReverseProxy.TrustedHTTPProxies,
s.AgentNetworkManager(), nil) s.AgentNetworkManager(), nil,
)
if err != nil { if err != nil {
log.Fatalf("failed to create API handler: %v", err) 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 { func (s *BaseServer) AccessLogsManager() accesslogs.Manager {
return Create(s, func() 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( accessLogManager.StartPeriodicCleanup(
context.Background(), context.Background(),
s.Config.ReverseProxy.AccessLogRetentionDays, s.Config.ReverseProxy.AccessLogRetentionDays,
s.Config.ReverseProxy.AccessLogCleanupIntervalHours, s.Config.ReverseProxy.AccessLogCleanupIntervalHours,
) )
s.OnStop(accessLogManager.StopPeriodicCleanup)
return accessLogManager return accessLogManager
}) })
} }
+28 -4
View File
@@ -24,8 +24,8 @@ type TxMetrics interface {
CountTransactionDuration(duration time.Duration) CountTransactionDuration(duration time.Duration)
} }
// Conn is the database connection shared by all repositories: one gorm handle, // Conn is a database connection the repositories run on: one gorm handle, the
// the pgx pool of a Postgres deployment and the engine they talk to. // pgx pool of a Postgres deployment and the engine they talk to.
type Conn struct { type Conn struct {
db *gorm.DB db *gorm.DB
pool *pgxpool.Pool pool *pgxpool.Pool
@@ -34,15 +34,36 @@ type Conn struct {
metrics TxMetrics 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 // NewConn takes ownership of an open gorm handle and pool once it returns
// without error, applying the connection limits and transaction timeout // without error, applying the connection limits and transaction timeout
// configured through the environment. // configured through the environment and opts.
func NewConn(ctx context.Context, gormDB *gorm.DB, engine Engine, pool *pgxpool.Pool) (*Conn, error) { func NewConn(ctx context.Context, gormDB *gorm.DB, engine Engine, pool *pgxpool.Pool, opts ...Option) (*Conn, error) {
sqlDB, err := gormDB.DB() sqlDB, err := gormDB.DB()
if err != nil { if err != nil {
return nil, err return nil, err
} }
var options connOptions
for _, opt := range opts {
opt(&options)
}
txTimeout := defaultTransactionTimeout txTimeout := defaultTransactionTimeout
if v := os.Getenv("NB_STORE_TRANSACTION_TIMEOUT"); v != "" { if v := os.Getenv("NB_STORE_TRANSACTION_TIMEOUT"); v != "" {
if parsed, err := time.ParseDuration(v); err == nil { 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 { if connsConfigured {
conns = configuredConns conns = configuredConns
} }
if options.maxOpenConns > 0 {
conns = options.maxOpenConns
}
if engine == SqliteStoreEngine { if engine == SqliteStoreEngine {
if connsConfigured { if connsConfigured {
log.WithContext(ctx).Warnf("setting NB_SQL_MAX_OPEN_CONNS is not supported for sqlite, using default value 1") log.WithContext(ctx).Warnf("setting NB_SQL_MAX_OPEN_CONNS is not supported for sqlite, using default value 1")
@@ -146,3 +146,22 @@ func TestNewConn_MaxOpenConnsFromEnv(t *testing.T) {
require.NoError(t, err) require.NoError(t, err)
assert.Equal(t, 1, sqliteDB.Stats().MaxOpenConnections) 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")
}
+17 -2
View File
@@ -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. // 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()) gormDB, err := gorm.Open(postgres.Open(dsn), GormConfig())
if err != nil { if err != nil {
return nil, err return nil, err
@@ -111,7 +111,7 @@ func OpenPostgres(ctx context.Context, dsn string, pool PoolConfig) (*Conn, erro
closeGorm(gormDB) closeGorm(gormDB)
return nil, err return nil, err
} }
conn, err := NewConn(ctx, gormDB, PostgresStoreEngine, pgxPool) conn, err := NewConn(ctx, gormDB, PostgresStoreEngine, pgxPool, opts...)
if err != nil { if err != nil {
pgxPool.Close() pgxPool.Close()
closeGorm(gormDB) closeGorm(gormDB)
@@ -120,6 +120,21 @@ func OpenPostgres(ctx context.Context, dsn string, pool PoolConfig) (*Conn, erro
return conn, nil 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) { func newPgxPool(ctx context.Context, dsn string, cfg PoolConfig) (*pgxpool.Pool, error) {
config, err := pgxpool.ParseConfig(dsn) config, err := pgxpool.ParseConfig(dsn)
if err != nil { if err != nil {
@@ -14,10 +14,12 @@ import (
"github.com/netbirdio/management-integrations/integrations" "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" accesslogsmanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/accesslogs/manager"
"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/domain/manager" "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/domain/manager"
proxymanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy/manager" proxymanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy/manager"
reverseproxymanager "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service/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" nbgrpc "github.com/netbirdio/netbird/management/internals/shared/grpc"
zonesManager "github.com/netbirdio/netbird/management/internals/modules/zones/manager" 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) 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) proxyTokenStore := nbgrpc.NewOneTimeTokenStore(ctx, cacheStore)
singleUseStore := nbgrpc.NewSingleUseStore(ctx, cacheStore) singleUseStore := nbgrpc.NewSingleUseStore(ctx, cacheStore)
noopMeter := noop.NewMeterProvider().Meter("") noopMeter := noop.NewMeterProvider().Meter("")
@@ -249,7 +251,7 @@ func BuildApiBlackBoxWithDBStateAndPeerChannel(t testing_tools.TB, sqlFile strin
t.Fatalf("Failed to create manager: %v", err) 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) proxyTokenStore := nbgrpc.NewOneTimeTokenStore(ctx, cacheStore)
singleUseStore := nbgrpc.NewSingleUseStore(ctx, cacheStore) singleUseStore := nbgrpc.NewSingleUseStore(ctx, cacheStore)
noopMeter := noop.NewMeterProvider().Meter("") noopMeter := noop.NewMeterProvider().Meter("")
@@ -299,6 +301,16 @@ func BuildApiBlackBoxWithDBStateAndPeerChannel(t testing_tools.TB, sqlFile strin
return apiHandler, am, updMsg 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) { func mockValidateAndParseToken(_ context.Context, token string) (auth.UserAuth, *jwt.Token, error) {
userAuth := auth.UserAuth{} userAuth := auth.UserAuth{}
+1 -2
View File
@@ -12,7 +12,6 @@ import (
nbdns "github.com/netbirdio/netbird/dns" nbdns "github.com/netbirdio/netbird/dns"
agentNetworkTypes "github.com/netbirdio/netbird/management/internals/modules/agentnetwork/types" 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/domain"
"github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy" "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/proxy"
rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service" 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{}, &installation{}, &types.ExtraSettings{}, &posture.Checks{}, &nbpeer.NetworkAddress{},
&networkTypes.Network{}, &routerTypes.NetworkRouter{}, &resourceTypes.NetworkResource{}, &types.AccountOnboarding{}, &networkTypes.Network{}, &routerTypes.NetworkRouter{}, &resourceTypes.NetworkResource{}, &types.AccountOnboarding{},
&types.Job{}, &zones.Zone{}, &records.Record{}, &types.UserInviteRecord{}, &rpservice.Service{}, &rpservice.Target{}, &domain.Domain{}, &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.Provider{}, &agentNetworkTypes.Policy{}, &agentNetworkTypes.Guardrail{}, &agentNetworkTypes.Settings{},
&agentNetworkTypes.Consumption{}, &agentNetworkTypes.AccountBudgetRule{}, &agentNetworkTypes.Consumption{}, &agentNetworkTypes.AccountBudgetRule{},
&agentNetworkTypes.AgentNetworkAccessLog{}, &agentNetworkTypes.AgentNetworkAccessLogGroup{}, &agentNetworkTypes.AgentNetworkAccessLog{}, &agentNetworkTypes.AgentNetworkAccessLogGroup{},
+6 -1
View File
@@ -460,6 +460,11 @@ func lookupDSNEnv(nbKey, legacyKey string) (string, bool) {
return os.LookupEnv(legacyKey) 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} var supportedEngines = []types.Engine{types.SqliteStoreEngine, types.PostgresStoreEngine}
func getStoreEngineFromEnv() types.Engine { 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) return db.OpenSqlite(ctx, dataDir)
case types.PostgresStoreEngine: case types.PostgresStoreEngine:
log.WithContext(ctx).Info("using Postgres store engine") log.WithContext(ctx).Info("using Postgres store engine")
dsn, ok := lookupDSNEnv(PostgresDsnEnv, PostgresDsnEnvLegacy) dsn, ok := LookupPostgresDSN()
if !ok { if !ok {
return nil, fmt.Errorf("%s is not set", PostgresDsnEnv) return nil, fmt.Errorf("%s is not set", PostgresDsnEnv)
} }
+1 -1
View File
@@ -9381,7 +9381,7 @@ paths:
name: search name: search
schema: schema:
type: string 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 - in: query
name: source_ip name: source_ip
schema: schema:
+1 -1
View File
@@ -6051,7 +6051,7 @@ type GetApiEventsProxyParams struct {
// SortOrder Sort order (ascending or descending) // SortOrder Sort order (ascending or descending)
SortOrder *GetApiEventsProxyParamsSortOrder `form:"sort_order,omitempty" json:"sort_order,omitempty"` 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"` Search *string `form:"search,omitempty" json:"search,omitempty"`
// SourceIp Filter by source IP address // SourceIp Filter by source IP address