cache.go 1.3 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667
  1. package firewall
  2. import (
  3. "context"
  4. "log/slog"
  5. "sync/atomic"
  6. "time"
  7. )
  8. // ConntrackCache is used as a local routine cache to know if a given flow
  9. // has been seen in the conntrack table.
  10. type ConntrackCache map[Packet]struct{}
  11. type ConntrackCacheTicker struct {
  12. cacheV uint64
  13. cacheTick atomic.Uint64
  14. l *slog.Logger
  15. cache ConntrackCache
  16. }
  17. func NewConntrackCacheTicker(ctx context.Context, l *slog.Logger, d time.Duration) *ConntrackCacheTicker {
  18. if d == 0 {
  19. return nil
  20. }
  21. c := &ConntrackCacheTicker{
  22. l: l,
  23. cache: ConntrackCache{},
  24. }
  25. go c.tick(ctx, d)
  26. return c
  27. }
  28. func (c *ConntrackCacheTicker) tick(ctx context.Context, d time.Duration) {
  29. t := time.NewTicker(d)
  30. defer t.Stop()
  31. for {
  32. select {
  33. case <-ctx.Done():
  34. return
  35. case <-t.C:
  36. c.cacheTick.Add(1)
  37. }
  38. }
  39. }
  40. // Get checks if the cache ticker has moved to the next version before returning
  41. // the map. If it has moved, we reset the map.
  42. func (c *ConntrackCacheTicker) Get() ConntrackCache {
  43. if c == nil {
  44. return nil
  45. }
  46. if tick := c.cacheTick.Load(); tick != c.cacheV {
  47. c.cacheV = tick
  48. if ll := len(c.cache); ll > 0 {
  49. if c.l.Enabled(context.Background(), slog.LevelDebug) {
  50. c.l.Debug("resetting conntrack cache", "len", ll)
  51. }
  52. c.cache = make(ConntrackCache, ll)
  53. }
  54. }
  55. return c.cache
  56. }