mirror of
https://github.com/netbirdio/netbird.git
synced 2026-09-19 05:09:06 +02:00
use gorm cache
This commit is contained in:
+48
@@ -0,0 +1,48 @@
|
||||
package cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
|
||||
"github.com/go-gorm/caches/v4"
|
||||
)
|
||||
|
||||
type MemoryCacher struct {
|
||||
store *sync.Map
|
||||
}
|
||||
|
||||
func (c *MemoryCacher) init() {
|
||||
if c.store == nil {
|
||||
c.store = &sync.Map{}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *MemoryCacher) Get(ctx context.Context, key string, q *caches.Query[any]) (*caches.Query[any], error) {
|
||||
c.init()
|
||||
val, ok := c.store.Load(key)
|
||||
if !ok {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
if err := q.Unmarshal(val.([]byte)); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return q, nil
|
||||
}
|
||||
|
||||
func (c *MemoryCacher) Store(ctx context.Context, key string, val *caches.Query[any]) error {
|
||||
c.init()
|
||||
res, err := val.Marshal()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
c.store.Store(key, res)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *MemoryCacher) Invalidate(ctx context.Context) error {
|
||||
c.store = &sync.Map{}
|
||||
return nil
|
||||
}
|
||||
+73
@@ -0,0 +1,73 @@
|
||||
package cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/go-gorm/caches/v4"
|
||||
"github.com/redis/go-redis/v9"
|
||||
)
|
||||
|
||||
type RedisCacher struct {
|
||||
rdb *redis.Client
|
||||
}
|
||||
|
||||
func NewRedisCacher(rdb *redis.Client) *RedisCacher {
|
||||
return &RedisCacher{rdb: rdb}
|
||||
}
|
||||
|
||||
func (c *RedisCacher) Get(ctx context.Context, key string, q *caches.Query[any]) (*caches.Query[any], error) {
|
||||
res, err := c.rdb.Get(ctx, key).Result()
|
||||
if err == redis.Nil {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if err := q.Unmarshal([]byte(res)); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return q, nil
|
||||
}
|
||||
|
||||
func (c *RedisCacher) Store(ctx context.Context, key string, val *caches.Query[any]) error {
|
||||
res, err := val.Marshal()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
c.rdb.Set(ctx, key, res, 300*time.Second) // Set proper cache time
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *RedisCacher) Invalidate(ctx context.Context) error {
|
||||
var (
|
||||
cursor uint64
|
||||
keys []string
|
||||
)
|
||||
for {
|
||||
var (
|
||||
k []string
|
||||
err error
|
||||
)
|
||||
k, cursor, err = c.rdb.Scan(ctx, cursor, fmt.Sprintf("%s*", caches.IdentifierPrefix), 0).Result()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
keys = append(keys, k...)
|
||||
if cursor == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if len(keys) > 0 {
|
||||
if _, err := c.rdb.Del(ctx, keys...).Result(); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -15,6 +15,8 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/go-gorm/caches/v4"
|
||||
"github.com/redis/go-redis/v9"
|
||||
log "github.com/sirupsen/logrus"
|
||||
"gorm.io/driver/mysql"
|
||||
"gorm.io/driver/postgres"
|
||||
@@ -30,6 +32,7 @@ import (
|
||||
networkTypes "github.com/netbirdio/netbird/management/server/networks/types"
|
||||
nbpeer "github.com/netbirdio/netbird/management/server/peer"
|
||||
"github.com/netbirdio/netbird/management/server/posture"
|
||||
"github.com/netbirdio/netbird/management/server/store/cache"
|
||||
"github.com/netbirdio/netbird/management/server/telemetry"
|
||||
"github.com/netbirdio/netbird/management/server/types"
|
||||
"github.com/netbirdio/netbird/management/server/util"
|
||||
@@ -46,11 +49,15 @@ const (
|
||||
accountAndIDsQueryCondition = "account_id = ? AND id IN ?"
|
||||
accountIDCondition = "account_id = ?"
|
||||
peerNotFoundFMT = "peer %s not found"
|
||||
|
||||
storeCacheEnabledEnv = "NB_STORE_CACHE_ENABLED"
|
||||
storeCacheRedisAddrEnv = "NB_STORE_CACHE_REDIS_ADDR"
|
||||
)
|
||||
|
||||
// SqlStore represents an account storage backed by a Sql DB persisted to disk
|
||||
type SqlStore struct {
|
||||
db *gorm.DB
|
||||
uncachedDB *gorm.DB
|
||||
globalAccountLock sync.Mutex
|
||||
metrics telemetry.AppMetrics
|
||||
installationPK int
|
||||
@@ -66,6 +73,13 @@ type migrationFunc func(*gorm.DB) error
|
||||
|
||||
// NewSqlStore creates a new SqlStore instance.
|
||||
func NewSqlStore(ctx context.Context, db *gorm.DB, storeEngine types.Engine, metrics telemetry.AppMetrics, skipMigration bool) (*SqlStore, error) {
|
||||
if os.Getenv(storeCacheEnabledEnv) == "true" {
|
||||
err := configureStoreCache(ctx, db)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to configure store cache: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
sql, err := db.DB()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -116,6 +130,26 @@ func NewSqlStore(ctx context.Context, db *gorm.DB, storeEngine types.Engine, met
|
||||
return &SqlStore{db: db, storeEngine: storeEngine, metrics: metrics, installationPK: 1}, nil
|
||||
}
|
||||
|
||||
func configureStoreCache(ctx context.Context, db *gorm.DB) error {
|
||||
var cacher caches.Cacher = &cache.MemoryCacher{}
|
||||
if addr := os.Getenv(storeCacheRedisAddrEnv); addr != "" {
|
||||
opt, err := redis.ParseURL(addr)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to parse redis url from %s: %w", addr, err)
|
||||
}
|
||||
cacher = cache.NewRedisCacher(redis.NewClient(opt))
|
||||
log.WithContext(ctx).Infof("using redis store cache at %s", addr)
|
||||
} else {
|
||||
log.WithContext(ctx).Infof("using in-memory store cache")
|
||||
}
|
||||
|
||||
cachesPlugin := &caches.Caches{Conf: &caches.Config{
|
||||
Cacher: cacher,
|
||||
}}
|
||||
|
||||
return db.Use(cachesPlugin)
|
||||
}
|
||||
|
||||
func GetKeyQueryCondition(s *SqlStore) string {
|
||||
if s.storeEngine == types.MysqlStoreEngine {
|
||||
return mysqlKeyQueryCondition
|
||||
|
||||
@@ -20,6 +20,8 @@ import (
|
||||
log "github.com/sirupsen/logrus"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"github.com/testcontainers/testcontainers-go"
|
||||
testcontainersredis "github.com/testcontainers/testcontainers-go/modules/redis"
|
||||
|
||||
nbdns "github.com/netbirdio/netbird/dns"
|
||||
resourceTypes "github.com/netbirdio/netbird/management/server/networks/resources/types"
|
||||
@@ -3717,3 +3719,320 @@ func TestSqlStore_GetPeersByGroupIDs(t *testing.T) {
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSqlStore_CacheHit(t *testing.T) {
|
||||
if runtime.GOOS == "windows" {
|
||||
t.Skip("The SQLite store is not properly supported by Windows yet")
|
||||
}
|
||||
|
||||
t.Setenv(storeCacheEnabledEnv, "true")
|
||||
|
||||
store, cleanUp, err := NewTestStoreFromSQL(context.Background(), "../testdata/store.sql", t.TempDir())
|
||||
t.Cleanup(cleanUp)
|
||||
require.NoError(t, err)
|
||||
|
||||
ctx := context.Background()
|
||||
accountID := "bf1c8084-ba50-4ce7-9439-34653001fc3b"
|
||||
peerID := "ct286bi7qv930dsrrug0"
|
||||
|
||||
sqlStore := store.(*SqlStore)
|
||||
|
||||
// First call - should hit the database
|
||||
peer1, err := sqlStore.GetPeerByID(ctx, LockingStrengthShare, accountID, peerID)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, peer1)
|
||||
|
||||
// Get the underlying database connection
|
||||
db, err := sqlStore.db.DB()
|
||||
require.NoError(t, err)
|
||||
|
||||
// Get DB stats before second call
|
||||
statsBefore := db.Stats()
|
||||
|
||||
// Second call - should hit the cache, not the database
|
||||
peer2, err := sqlStore.GetPeerByID(ctx, LockingStrengthShare, accountID, peerID)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, peer2)
|
||||
|
||||
// Get DB stats after second call
|
||||
statsAfter := db.Stats()
|
||||
|
||||
// Verify no additional database connections were opened for the cached query
|
||||
// The OpenConnections count should be the same or very similar
|
||||
assert.Equal(t, statsBefore.InUse, statsAfter.InUse, "Cache hit should not open new database connections")
|
||||
|
||||
// Verify both peers are equal
|
||||
assert.Equal(t, peer1.ID, peer2.ID)
|
||||
assert.Equal(t, peer1.Name, peer2.Name)
|
||||
}
|
||||
|
||||
func TestSqlStore_CacheInvalidationAcrossInstances(t *testing.T) {
|
||||
if runtime.GOOS == "windows" {
|
||||
t.Skip("The SQLite store is not properly supported by Windows yet")
|
||||
}
|
||||
|
||||
t.Setenv(storeCacheEnabledEnv, "true")
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
// Start Redis container for shared cache
|
||||
redisContainer, err := testcontainersredis.RunContainer(ctx, testcontainers.WithImage("redis:7"))
|
||||
require.NoError(t, err)
|
||||
defer func() {
|
||||
if err := redisContainer.Terminate(ctx); err != nil {
|
||||
t.Logf("failed to terminate container: %s", err)
|
||||
}
|
||||
}()
|
||||
|
||||
redisURL, err := redisContainer.ConnectionString(ctx)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Set the Redis URL environment variable for both stores
|
||||
t.Setenv(storeCacheRedisAddrEnv, redisURL)
|
||||
|
||||
// Create a shared SQLite database in a temp directory with cache=shared mode
|
||||
// This allows multiple connections to the same database
|
||||
tempDir := t.TempDir()
|
||||
|
||||
// Create first store instance with shared database
|
||||
store1, cleanUp1, err := NewTestStoreFromSQL(ctx, "../testdata/store.sql", tempDir)
|
||||
t.Cleanup(cleanUp1)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Create second store instance connecting to the SAME database file
|
||||
// Both stores will share the same underlying database AND the same Redis cache
|
||||
store2, cleanUp2, err := NewTestStoreFromSQL(ctx, "", tempDir)
|
||||
t.Cleanup(cleanUp2)
|
||||
require.NoError(t, err)
|
||||
|
||||
accountID := "bf1c8084-ba50-4ce7-9439-34653001fc3b"
|
||||
peerID := "ct286bi7qv930dsrrug0"
|
||||
|
||||
// Store 1: Fetch peer (populates cache)
|
||||
peer1, err := store1.GetPeerByID(ctx, LockingStrengthShare, accountID, peerID)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, peer1)
|
||||
|
||||
// Store 2: Fetch same peer (should use cache)
|
||||
peer2, err := store2.GetPeerByID(ctx, LockingStrengthShare, accountID, peerID)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, peer2)
|
||||
assert.Equal(t, peer1.ID, peer2.ID)
|
||||
|
||||
// Store 1: Modify the peer
|
||||
peer1.Name = "updated-peer-name"
|
||||
err = store1.SavePeer(ctx, accountID, peer1)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Store 2: Fetch the peer again - should get updated data (cache was invalidated)
|
||||
peer2Updated, err := store2.GetPeerByID(ctx, LockingStrengthShare, accountID, peerID)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, peer2Updated)
|
||||
|
||||
// Verify the name was updated via cache invalidation
|
||||
assert.Equal(t, "updated-peer-name", peer2Updated.Name,
|
||||
"Cache should have been invalidated, store 2 should see the update from store 1")
|
||||
}
|
||||
|
||||
func TestSqlStore_CacheGetAccountWithAssociations(t *testing.T) {
|
||||
if runtime.GOOS == "windows" {
|
||||
t.Skip("The SQLite store is not properly supported by Windows yet")
|
||||
}
|
||||
|
||||
t.Setenv(storeCacheEnabledEnv, "true")
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
// Start Redis container for shared cache
|
||||
redisContainer, err := testcontainersredis.RunContainer(ctx, testcontainers.WithImage("redis:7"))
|
||||
require.NoError(t, err)
|
||||
defer func() {
|
||||
if err := redisContainer.Terminate(ctx); err != nil {
|
||||
t.Logf("failed to terminate container: %s", err)
|
||||
}
|
||||
}()
|
||||
|
||||
redisURL, err := redisContainer.ConnectionString(ctx)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Set the Redis URL environment variable for both stores
|
||||
t.Setenv(storeCacheRedisAddrEnv, redisURL)
|
||||
|
||||
// Create a shared SQLite database in a temp directory with cache=shared mode
|
||||
// This allows multiple connections to the same database
|
||||
tempDir := t.TempDir()
|
||||
|
||||
// Create first store instance with shared database
|
||||
store1, cleanUp1, err := NewTestStoreFromSQL(ctx, "", tempDir)
|
||||
t.Cleanup(cleanUp1)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Create second store instance connecting to the SAME database file
|
||||
// Both stores will share the same underlying database AND the same Redis cache
|
||||
store2, cleanUp2, err := NewTestStoreFromSQL(ctx, "", tempDir)
|
||||
t.Cleanup(cleanUp2)
|
||||
require.NoError(t, err)
|
||||
|
||||
accountID := "bf1c8084-ba50-4ce7-9439-34653001fc3b"
|
||||
userID := "edafee4e-63fb-11ec-90d6-0242ac120003"
|
||||
|
||||
// Create a fresh account
|
||||
account := newAccountWithId(ctx, accountID, userID, "test.com")
|
||||
|
||||
err = store1.SaveAccount(ctx, account)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Store 1: Fetch account (populates cache)
|
||||
account1, err := store1.GetAccount(ctx, accountID)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, account1)
|
||||
|
||||
// Store 2: Fetch same account (should use cache)
|
||||
account2, err := store2.GetAccount(ctx, accountID)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, account2)
|
||||
assert.Equal(t, account1.Id, account2.Id)
|
||||
|
||||
// Store 1: Modify the account
|
||||
account1.Domain = "updated-domain.example.com"
|
||||
err = store1.SaveAccount(ctx, account1)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Store 2: Fetch the account again - should get updated data (cache was invalidated)
|
||||
account2Updated, err := store2.GetAccount(ctx, accountID)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, account2Updated)
|
||||
|
||||
// Verify the domain was updated via cache invalidation
|
||||
assert.Equal(t, "updated-domain.example.com", account2Updated.Domain,
|
||||
"Cache should have been invalidated, store 2 should see the update from store 1")
|
||||
}
|
||||
|
||||
func TestSqlStore_CacheGetGroupWithAssociations(t *testing.T) {
|
||||
if runtime.GOOS == "windows" {
|
||||
t.Skip("The SQLite store is not properly supported by Windows yet")
|
||||
}
|
||||
|
||||
t.Setenv(storeCacheEnabledEnv, "true")
|
||||
|
||||
ctx := context.Background()
|
||||
|
||||
// Start Redis container for shared cache
|
||||
redisContainer, err := testcontainersredis.RunContainer(ctx, testcontainers.WithImage("redis:7"))
|
||||
require.NoError(t, err)
|
||||
defer func() {
|
||||
if err := redisContainer.Terminate(ctx); err != nil {
|
||||
t.Logf("failed to terminate container: %s", err)
|
||||
}
|
||||
}()
|
||||
|
||||
redisURL, err := redisContainer.ConnectionString(ctx)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Set the Redis URL environment variable for both stores
|
||||
t.Setenv(storeCacheRedisAddrEnv, redisURL)
|
||||
|
||||
// Create a shared SQLite database in a temp directory with cache=shared mode
|
||||
// This allows multiple connections to the same database
|
||||
tempDir := t.TempDir()
|
||||
|
||||
// Create first store instance with shared database
|
||||
store1, cleanUp1, err := NewTestStoreFromSQL(ctx, "", tempDir)
|
||||
t.Cleanup(cleanUp1)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Create second store instance connecting to the SAME database file
|
||||
// Both stores will share the same underlying database AND the same Redis cache
|
||||
store2, cleanUp2, err := NewTestStoreFromSQL(ctx, "", tempDir)
|
||||
t.Cleanup(cleanUp2)
|
||||
require.NoError(t, err)
|
||||
|
||||
accountID := "bf1c8084-ba50-4ce7-9439-34653001fc3b"
|
||||
userID := "edafee4e-63fb-11ec-90d6-0242ac120003"
|
||||
|
||||
// Create a fresh account
|
||||
account := newAccountWithId(ctx, accountID, userID, "test.com")
|
||||
|
||||
// Add peers to the account
|
||||
peer1 := &nbpeer.Peer{
|
||||
Key: "peer-key-1",
|
||||
ID: "peer-id-1",
|
||||
IP: net.IP{100, 64, 0, 1},
|
||||
Meta: nbpeer.PeerSystemMeta{Hostname: "test-peer-1"},
|
||||
Name: "Test Peer 1",
|
||||
DNSLabel: "test-peer-1",
|
||||
Status: &nbpeer.PeerStatus{Connected: false, LastSeen: time.Now().UTC()},
|
||||
CreatedAt: time.Now().UTC(),
|
||||
UserID: userID,
|
||||
}
|
||||
account.Peers[peer1.ID] = peer1
|
||||
|
||||
peer2 := &nbpeer.Peer{
|
||||
Key: "peer-key-2",
|
||||
ID: "peer-id-2",
|
||||
IP: net.IP{100, 64, 0, 2},
|
||||
Meta: nbpeer.PeerSystemMeta{Hostname: "test-peer-2"},
|
||||
Name: "Test Peer 2",
|
||||
DNSLabel: "test-peer-2",
|
||||
Status: &nbpeer.PeerStatus{Connected: false, LastSeen: time.Now().UTC()},
|
||||
CreatedAt: time.Now().UTC(),
|
||||
UserID: userID,
|
||||
}
|
||||
account.Peers[peer2.ID] = peer2
|
||||
|
||||
// Create a group with peers (SaveAccount will convert to GroupPeers)
|
||||
group := &types.Group{
|
||||
ID: "group-id-1",
|
||||
AccountID: accountID,
|
||||
Name: "Test Group",
|
||||
Issued: "api",
|
||||
Peers: []string{peer1.ID, peer2.ID},
|
||||
Resources: []types.Resource{},
|
||||
}
|
||||
account.Groups = map[string]*types.Group{
|
||||
group.ID: group,
|
||||
}
|
||||
|
||||
// Save the account with all data using store1
|
||||
err = store1.SaveAccount(ctx, account)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Store 1: Fetch group (populates cache)
|
||||
group1, err := store1.GetGroupByID(ctx, LockingStrengthShare, accountID, group.ID)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, group1)
|
||||
require.NotEmpty(t, group1.Peers, "First call should load Peers (converted from GroupPeers)")
|
||||
require.Len(t, group1.Peers, 2, "First call should load both Peers")
|
||||
|
||||
// Store 2: Fetch same group (should use cache)
|
||||
group2, err := store2.GetGroupByID(ctx, LockingStrengthShare, accountID, group.ID)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, group2)
|
||||
require.NotEmpty(t, group2.Peers, "Cached group should have Peers")
|
||||
require.Len(t, group2.Peers, 2, "Cached group should have both Peers")
|
||||
|
||||
// Verify data matches between both stores
|
||||
assert.Equal(t, len(group1.Peers), len(group2.Peers))
|
||||
assert.Equal(t, group1.Name, group2.Name)
|
||||
assert.ElementsMatch(t, group1.Peers, group2.Peers)
|
||||
|
||||
// Modify the group with store1 (update name and remove one peer using UpdateGroup and RemovePeerFromGroup)
|
||||
group1.Name = "Modified Group Name"
|
||||
err = store1.UpdateGroup(ctx, group1)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Remove peer2 from the group
|
||||
err = store1.RemovePeerFromGroup(ctx, peer2.ID, group.ID)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Store2: Fetch the modified group (should get updated data, not stale cache)
|
||||
group3, err := store2.GetGroupByID(ctx, LockingStrengthShare, accountID, group.ID)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, group3)
|
||||
|
||||
// Verify the updated data is visible from store2
|
||||
assert.Equal(t, "Modified Group Name", group3.Name, "Store2 should see the updated group name")
|
||||
assert.Len(t, group3.Peers, 1, "Store2 should see only one peer after modification")
|
||||
assert.Contains(t, group3.Peers, peer1.ID, "Store2 should see peer1")
|
||||
assert.NotContains(t, group3.Peers, peer2.ID, "Store2 should NOT see peer2 after removal")
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user