diff --git a/.gitignore b/.gitignore index 783fe77f3..305f3cb50 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ +.claude .idea .run *.iml diff --git a/client/iface/configurer/kernel_unix.go b/client/iface/configurer/kernel_unix.go index a29fe181a..da69c2a35 100644 --- a/client/iface/configurer/kernel_unix.go +++ b/client/iface/configurer/kernel_unix.go @@ -17,12 +17,15 @@ import ( type KernelConfigurer struct { deviceName string + statsCache *statsCache } func NewKernelConfigurer(deviceName string) *KernelConfigurer { - return &KernelConfigurer{ + c := &KernelConfigurer{ deviceName: deviceName, } + c.statsCache = newStatsCache(statsCacheTTL, c.fetchStats) + return c } func (c *KernelConfigurer) ConfigureInterface(privateKey string, port int) error { @@ -246,12 +249,6 @@ func (c *KernelConfigurer) configure(config wgtypes.Config) error { } }() - // validate if device with name exists - _, err = wg.Device(c.deviceName) - if err != nil { - return err - } - return wg.ConfigureDevice(c.deviceName, config) } @@ -300,6 +297,14 @@ func (c *KernelConfigurer) FullStats() (*Stats, error) { } func (c *KernelConfigurer) GetStats() (map[string]WGStats, error) { + return c.statsCache.get() +} + +func (c *KernelConfigurer) LastActivities() map[string]monotime.Time { + return nil +} + +func (c *KernelConfigurer) fetchStats() (map[string]WGStats, error) { stats := make(map[string]WGStats) wg, err := wgctrl.New() if err != nil { @@ -326,7 +331,3 @@ func (c *KernelConfigurer) GetStats() (map[string]WGStats, error) { } return stats, nil } - -func (c *KernelConfigurer) LastActivities() map[string]monotime.Time { - return nil -} diff --git a/client/iface/configurer/stats_cache.go b/client/iface/configurer/stats_cache.go new file mode 100644 index 000000000..71a4e88fc --- /dev/null +++ b/client/iface/configurer/stats_cache.go @@ -0,0 +1,52 @@ +package configurer + +import ( + "sync" + "time" + + "golang.org/x/sync/singleflight" +) + +const statsCacheTTL = 1 * time.Second + +type statsCache struct { + ttl time.Duration + fetch func() (map[string]WGStats, error) + + mu sync.RWMutex + value map[string]WGStats + expireAt time.Time + + sf singleflight.Group +} + +func newStatsCache(ttl time.Duration, fetch func() (map[string]WGStats, error)) *statsCache { + return &statsCache{ttl: ttl, fetch: fetch} +} + +func (c *statsCache) get() (map[string]WGStats, error) { + c.mu.RLock() + if c.value != nil && time.Now().Before(c.expireAt) { + value := c.value + c.mu.RUnlock() + return value, nil + } + c.mu.RUnlock() + + value, err, _ := c.sf.Do("stats", func() (interface{}, error) { + res, err := c.fetch() + if err != nil { + return nil, err + } + + c.mu.Lock() + c.value = res + c.expireAt = time.Now().Add(c.ttl) + c.mu.Unlock() + return res, nil + }) + if err != nil { + return nil, err + } + return value.(map[string]WGStats), nil +} diff --git a/client/iface/configurer/stats_cache_test.go b/client/iface/configurer/stats_cache_test.go new file mode 100644 index 000000000..bcee5cd52 --- /dev/null +++ b/client/iface/configurer/stats_cache_test.go @@ -0,0 +1,70 @@ +package configurer + +import ( + "errors" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestStatsCache_CachesWithinTTL(t *testing.T) { + var calls atomic.Int64 + c := newStatsCache(50*time.Millisecond, func() (map[string]WGStats, error) { + calls.Add(1) + return map[string]WGStats{"p": {}}, nil + }) + + for i := 0; i < 10; i++ { + _, err := c.get() + require.NoError(t, err) + } + require.Equal(t, int64(1), calls.Load(), "within TTL only one underlying fetch") + + time.Sleep(60 * time.Millisecond) + _, err := c.get() + require.NoError(t, err) + require.Equal(t, int64(2), calls.Load(), "after TTL expiry a fresh fetch happens") +} + +func TestStatsCache_SingleFlight(t *testing.T) { + var calls atomic.Int64 + release := make(chan struct{}) + c := newStatsCache(time.Minute, func() (map[string]WGStats, error) { + calls.Add(1) + <-release + return map[string]WGStats{}, nil + }) + + const n = 50 + var wg sync.WaitGroup + wg.Add(n) + for i := 0; i < n; i++ { + go func() { + defer wg.Done() + _, _ = c.get() + }() + } + time.Sleep(20 * time.Millisecond) + close(release) + wg.Wait() + + require.Equal(t, int64(1), calls.Load(), "concurrent misses collapse into one fetch") +} + +func TestStatsCache_ErrorNotCached(t *testing.T) { + var calls atomic.Int64 + wantErr := errors.New("dump failed") + c := newStatsCache(time.Minute, func() (map[string]WGStats, error) { + calls.Add(1) + return nil, wantErr + }) + + _, err := c.get() + require.ErrorIs(t, err, wantErr) + _, err = c.get() + require.ErrorIs(t, err, wantErr) + require.Equal(t, int64(2), calls.Load(), "errors are not cached; each call retries") +} diff --git a/client/iface/configurer/usp.go b/client/iface/configurer/usp.go index 9b070aab8..0a25c55bc 100644 --- a/client/iface/configurer/usp.go +++ b/client/iface/configurer/usp.go @@ -40,6 +40,7 @@ type WGUSPConfigurer struct { device *device.Device deviceName string activityRecorder *bind.ActivityRecorder + statsCache *statsCache uapiListener net.Listener } @@ -50,16 +51,19 @@ func NewUSPConfigurer(device *device.Device, deviceName string, activityRecorder deviceName: deviceName, activityRecorder: activityRecorder, } + wgCfg.statsCache = newStatsCache(statsCacheTTL, wgCfg.fetchStats) wgCfg.startUAPI() return wgCfg } func NewUSPConfigurerNoUAPI(device *device.Device, deviceName string, activityRecorder *bind.ActivityRecorder) *WGUSPConfigurer { - return &WGUSPConfigurer{ + wgCfg := &WGUSPConfigurer{ device: device, deviceName: deviceName, activityRecorder: activityRecorder, } + wgCfg.statsCache = newStatsCache(statsCacheTTL, wgCfg.fetchStats) + return wgCfg } func (c *WGUSPConfigurer) ConfigureInterface(privateKey string, port int) error { @@ -348,6 +352,10 @@ func (t *WGUSPConfigurer) Close() { } func (t *WGUSPConfigurer) GetStats() (map[string]WGStats, error) { + return t.statsCache.get() +} + +func (t *WGUSPConfigurer) fetchStats() (map[string]WGStats, error) { ipc, err := t.device.IpcGet() if err != nil { return nil, fmt.Errorf("ipc get: %w", err)