mirror of
https://github.com/0xERR0R/blocky.git
synced 2026-10-11 21:59:54 +00:00
refactor: redis write through cache (#2025)
This commit is contained in:
Dimitri Herzog
@@ -5,6 +5,7 @@
|
||||
bin/
|
||||
dist/
|
||||
docs/docs.go
|
||||
docs/superpowers/
|
||||
site/
|
||||
config.yml
|
||||
todo.txt
|
||||
|
||||
Vendored
+449
@@ -0,0 +1,449 @@
|
||||
package cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/0xERR0R/blocky/log"
|
||||
"github.com/0xERR0R/blocky/metrics"
|
||||
blockredis "github.com/0xERR0R/blocky/redis"
|
||||
goredis "github.com/go-redis/redis/v8"
|
||||
"github.com/google/uuid"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/prometheus/client_golang/prometheus/promauto"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
const (
|
||||
defaultBatchSize = 100
|
||||
defaultFlushInterval = 100 * time.Millisecond
|
||||
defaultSendBufSize = 1000
|
||||
redisScanCount = 100
|
||||
drainTimeout = 5 * time.Second
|
||||
)
|
||||
|
||||
//nolint:gochecknoglobals
|
||||
var redisBufferDrops = promauto.With(metrics.Reg).NewCounter(
|
||||
prometheus.CounterOpts{
|
||||
Name: "blocky_redis_cache_buffer_drops_total",
|
||||
Help: "Number of cache entries dropped because the Redis send buffer was full",
|
||||
},
|
||||
)
|
||||
|
||||
// RedisOptions configures the RedisExpiringCache decorator.
|
||||
type RedisOptions[T any] struct {
|
||||
// Prefix is prepended to all Redis keys.
|
||||
Prefix string
|
||||
// Channel is the pub/sub channel name used for cross-instance sync.
|
||||
Channel string
|
||||
// Encode serializes a cache value to bytes.
|
||||
Encode func(*T) ([]byte, error)
|
||||
// Decode deserializes bytes back to a cache value.
|
||||
Decode func([]byte) (*T, error)
|
||||
// BatchSize is the maximum number of entries per Redis pipeline flush.
|
||||
BatchSize int
|
||||
// FlushInterval is the maximum time between flushes.
|
||||
FlushInterval time.Duration
|
||||
// SendBufSize is the capacity of the internal send buffer channel.
|
||||
SendBufSize int
|
||||
}
|
||||
|
||||
type redisSyncEntry struct {
|
||||
Key string `json:"k"`
|
||||
Data []byte `json:"d"`
|
||||
}
|
||||
|
||||
type redisSyncMessage struct {
|
||||
Entries []redisSyncEntry `json:"e"`
|
||||
Client string `json:"c"`
|
||||
}
|
||||
|
||||
type sendBufferEntry[T any] struct {
|
||||
key string
|
||||
val *T
|
||||
expiration time.Duration
|
||||
}
|
||||
|
||||
// RedisExpiringCache wraps an ExpiringCache[T] with transparent Redis
|
||||
// write-through and pub/sub synchronisation across instances.
|
||||
type RedisExpiringCache[T any] struct {
|
||||
inner ExpiringCache[T]
|
||||
client *goredis.Client
|
||||
opts RedisOptions[T]
|
||||
instanceID string
|
||||
sendBuf chan sendBufferEntry[T]
|
||||
logger *logrus.Entry
|
||||
}
|
||||
|
||||
// NewRedisExpiringByteCache creates a RedisExpiringCache for []byte values,
|
||||
// using identity encode/decode when no codec is provided.
|
||||
func NewRedisExpiringByteCache(
|
||||
ctx context.Context,
|
||||
inner ExpiringCache[[]byte],
|
||||
client *goredis.Client,
|
||||
opts RedisOptions[[]byte],
|
||||
) (*RedisExpiringCache[[]byte], error) {
|
||||
if opts.Encode == nil && opts.Decode == nil {
|
||||
opts.Encode = func(b *[]byte) ([]byte, error) { return *b, nil }
|
||||
opts.Decode = func(b []byte) (*[]byte, error) {
|
||||
cp := make([]byte, len(b))
|
||||
copy(cp, b)
|
||||
|
||||
return &cp, nil
|
||||
}
|
||||
}
|
||||
|
||||
return NewRedisExpiringCache(ctx, inner, client, opts)
|
||||
}
|
||||
|
||||
// NewRedisExpiringCache creates a new RedisExpiringCache decorator.
|
||||
//
|
||||
// It performs a blocking startup scan of existing Redis keys and loads them
|
||||
// into inner before launching the background writer and subscriber goroutines.
|
||||
// The goroutines run until ctx is cancelled.
|
||||
func NewRedisExpiringCache[T any](
|
||||
ctx context.Context,
|
||||
inner ExpiringCache[T],
|
||||
client *goredis.Client,
|
||||
opts RedisOptions[T],
|
||||
) (*RedisExpiringCache[T], error) {
|
||||
if opts.Encode == nil || opts.Decode == nil {
|
||||
return nil, errors.New("RedisOptions: Encode and Decode must both be set")
|
||||
}
|
||||
|
||||
if opts.BatchSize <= 0 {
|
||||
opts.BatchSize = defaultBatchSize
|
||||
}
|
||||
|
||||
if opts.FlushInterval <= 0 {
|
||||
opts.FlushInterval = defaultFlushInterval
|
||||
}
|
||||
|
||||
if opts.SendBufSize <= 0 {
|
||||
opts.SendBufSize = defaultSendBufSize
|
||||
}
|
||||
|
||||
c := &RedisExpiringCache[T]{
|
||||
inner: inner,
|
||||
client: client,
|
||||
opts: opts,
|
||||
instanceID: uuid.NewString(),
|
||||
sendBuf: make(chan sendBufferEntry[T], opts.SendBufSize),
|
||||
logger: log.PrefixedLog("redis-cache"),
|
||||
}
|
||||
|
||||
// Blocking startup load.
|
||||
if err := c.loadFromRedis(ctx); err != nil {
|
||||
c.logger.WithError(err).Warn("startup Redis scan failed – starting with empty local cache")
|
||||
}
|
||||
|
||||
go c.runSubscriber(ctx)
|
||||
go c.runWriter(ctx)
|
||||
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// Put stores the value in the inner cache immediately, then enqueues a
|
||||
// non-blocking send to Redis. If the send buffer is full the entry is dropped
|
||||
// and a warning is logged – the DNS path is never blocked.
|
||||
func (c *RedisExpiringCache[T]) Put(key string, val *T, expiration time.Duration) {
|
||||
c.inner.Put(key, val, expiration)
|
||||
|
||||
if expiration <= 0 {
|
||||
return
|
||||
}
|
||||
|
||||
entry := sendBufferEntry[T]{key: key, val: val, expiration: expiration}
|
||||
|
||||
select {
|
||||
case c.sendBuf <- entry:
|
||||
default:
|
||||
redisBufferDrops.Inc()
|
||||
c.logger.Warn("redis send buffer full, dropping cache entry")
|
||||
}
|
||||
}
|
||||
|
||||
// Get reads from the inner cache only.
|
||||
func (c *RedisExpiringCache[T]) Get(key string) (*T, time.Duration) {
|
||||
return c.inner.Get(key)
|
||||
}
|
||||
|
||||
// TotalCount delegates to the inner cache.
|
||||
func (c *RedisExpiringCache[T]) TotalCount() int {
|
||||
return c.inner.TotalCount()
|
||||
}
|
||||
|
||||
const clearTimeout = 10 * time.Second
|
||||
|
||||
// Clear removes all entries from the inner cache and from Redis (all keys
|
||||
// matching the configured prefix).
|
||||
func (c *RedisExpiringCache[T]) Clear() {
|
||||
c.inner.Clear()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), clearTimeout)
|
||||
defer cancel()
|
||||
|
||||
err := c.scanKeys(ctx, func(ctx context.Context, keys []string) error {
|
||||
return c.client.Del(ctx, keys...).Err()
|
||||
})
|
||||
if err != nil {
|
||||
c.logger.WithError(err).Warn("Redis Clear failed")
|
||||
}
|
||||
}
|
||||
|
||||
// redisKey returns the prefixed Redis key for the given cache key.
|
||||
func (c *RedisExpiringCache[T]) redisKey(key string) string {
|
||||
return c.opts.Prefix + key
|
||||
}
|
||||
|
||||
// scanKeys iterates over all Redis keys matching the configured prefix
|
||||
// and calls fn for each batch of keys returned by SCAN.
|
||||
func (c *RedisExpiringCache[T]) scanKeys(
|
||||
ctx context.Context, fn func(ctx context.Context, keys []string) error,
|
||||
) error {
|
||||
pattern := c.opts.Prefix + "*"
|
||||
|
||||
var cursor uint64
|
||||
|
||||
for {
|
||||
keys, nextCursor, err := c.client.Scan(ctx, cursor, pattern, redisScanCount).Result()
|
||||
if err != nil {
|
||||
return fmt.Errorf("SCAN: %w", err)
|
||||
}
|
||||
|
||||
if len(keys) > 0 {
|
||||
if err := fn(ctx, keys); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
cursor = nextCursor
|
||||
if cursor == 0 {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// loadFromRedis scans existing Redis keys and populates the inner cache.
|
||||
// It uses a pipeline per SCAN page to minimise round trips.
|
||||
func (c *RedisExpiringCache[T]) loadFromRedis(ctx context.Context) error {
|
||||
return c.scanKeys(ctx, func(ctx context.Context, keys []string) error {
|
||||
if err := c.loadKeysViaPipeline(ctx, keys); err != nil {
|
||||
c.logger.WithError(err).Warn("pipeline load failed for SCAN page")
|
||||
}
|
||||
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
// loadKeysViaPipeline fetches TTL and value for a batch of keys in a single pipeline round trip.
|
||||
func (c *RedisExpiringCache[T]) loadKeysViaPipeline(ctx context.Context, keys []string) error {
|
||||
pipe := c.client.Pipeline()
|
||||
|
||||
ttlCmds := make([]*goredis.DurationCmd, len(keys))
|
||||
getCmds := make([]*goredis.StringCmd, len(keys))
|
||||
|
||||
for i, rk := range keys {
|
||||
ttlCmds[i] = pipe.TTL(ctx, rk)
|
||||
getCmds[i] = pipe.Get(ctx, rk)
|
||||
}
|
||||
|
||||
if _, err := pipe.Exec(ctx); err != nil && ctx.Err() == nil {
|
||||
return fmt.Errorf("pipeline exec: %w", err)
|
||||
}
|
||||
|
||||
for i, rk := range keys {
|
||||
dur, err := ttlCmds[i].Result()
|
||||
if err != nil || dur <= 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
data, err := getCmds[i].Bytes()
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
val, err := c.opts.Decode(data)
|
||||
if err != nil {
|
||||
c.logger.WithError(err).Warn("failed to decode Redis entry on startup")
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
cacheKey := strings.TrimPrefix(rk, c.opts.Prefix)
|
||||
c.inner.Put(cacheKey, val, dur)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// runWriter is the background goroutine that drains sendBuf and flushes
|
||||
// batches to Redis via a pipeline + a single PUBLISH.
|
||||
func (c *RedisExpiringCache[T]) runWriter(ctx context.Context) {
|
||||
ticker := time.NewTicker(c.opts.FlushInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
batch := make([]sendBufferEntry[T], 0, c.opts.BatchSize)
|
||||
|
||||
flush := func(fctx context.Context) {
|
||||
if len(batch) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
c.flushBatch(fctx, batch)
|
||||
batch = make([]sendBufferEntry[T], 0, c.opts.BatchSize)
|
||||
}
|
||||
|
||||
for {
|
||||
select {
|
||||
case entry := <-c.sendBuf:
|
||||
batch = append(batch, entry)
|
||||
if len(batch) >= c.opts.BatchSize {
|
||||
flush(ctx)
|
||||
}
|
||||
|
||||
case <-ticker.C:
|
||||
flush(ctx)
|
||||
|
||||
case <-ctx.Done():
|
||||
drainCtx, drainCancel := context.WithTimeout(context.Background(), drainTimeout)
|
||||
|
||||
// Drain remaining buffer before exiting.
|
||||
drainLoop:
|
||||
for {
|
||||
select {
|
||||
case entry := <-c.sendBuf:
|
||||
batch = append(batch, entry)
|
||||
if len(batch) >= c.opts.BatchSize {
|
||||
flush(drainCtx) //nolint:contextcheck
|
||||
}
|
||||
default:
|
||||
break drainLoop
|
||||
}
|
||||
}
|
||||
|
||||
flush(drainCtx) //nolint:contextcheck
|
||||
drainCancel()
|
||||
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// flushBatch writes a slice of entries to Redis via a pipeline and publishes a
|
||||
// single sync message on the configured channel.
|
||||
func (c *RedisExpiringCache[T]) flushBatch(ctx context.Context, batch []sendBufferEntry[T]) {
|
||||
syncEntries := make([]redisSyncEntry, 0, len(batch))
|
||||
|
||||
pipe := c.client.Pipeline()
|
||||
|
||||
for _, e := range batch {
|
||||
data, err := c.opts.Encode(e.val)
|
||||
if err != nil {
|
||||
c.logger.WithError(err).Warn("failed to encode cache entry for Redis")
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
pipe.Set(ctx, c.redisKey(e.key), data, e.expiration)
|
||||
|
||||
syncEntries = append(syncEntries, redisSyncEntry{
|
||||
Key: e.key,
|
||||
Data: data,
|
||||
})
|
||||
}
|
||||
|
||||
if _, err := pipe.Exec(ctx); err != nil && ctx.Err() == nil {
|
||||
c.logger.WithError(err).Warn("Redis pipeline exec failed")
|
||||
}
|
||||
|
||||
if len(syncEntries) == 0 || c.opts.Channel == "" {
|
||||
return
|
||||
}
|
||||
|
||||
msg := redisSyncMessage{
|
||||
Entries: syncEntries,
|
||||
Client: c.instanceID,
|
||||
}
|
||||
|
||||
payload, err := json.Marshal(msg)
|
||||
if err != nil {
|
||||
c.logger.WithError(err).Warn("failed to marshal sync message")
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
if err := c.client.Publish(ctx, c.opts.Channel, payload).Err(); err != nil && ctx.Err() == nil {
|
||||
c.logger.WithError(err).Warn("Redis PUBLISH failed")
|
||||
}
|
||||
}
|
||||
|
||||
// runSubscriber subscribes to the pub/sub channel and applies remote entries
|
||||
// to the inner cache, filtering out messages originating from this instance.
|
||||
// It reconnects automatically on channel closure with exponential backoff.
|
||||
func (c *RedisExpiringCache[T]) runSubscriber(ctx context.Context) {
|
||||
if c.opts.Channel == "" {
|
||||
return
|
||||
}
|
||||
|
||||
loop := &blockredis.PubSubLoop{
|
||||
Client: c.client,
|
||||
Channel: c.opts.Channel,
|
||||
Logger: c.logger,
|
||||
Handler: func(ctx context.Context, payload string) {
|
||||
c.handleSyncMessage(ctx, []byte(payload))
|
||||
},
|
||||
}
|
||||
|
||||
loop.Run(ctx)
|
||||
}
|
||||
|
||||
// handleSyncMessage decodes and applies a pub/sub sync message, skipping
|
||||
// entries that originated from this instance.
|
||||
func (c *RedisExpiringCache[T]) handleSyncMessage(ctx context.Context, payload []byte) {
|
||||
var msg redisSyncMessage
|
||||
if err := json.Unmarshal(payload, &msg); err != nil {
|
||||
c.logger.WithError(err).Warn("failed to decode sync message")
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
if msg.Client == c.instanceID {
|
||||
return
|
||||
}
|
||||
|
||||
pipe := c.client.Pipeline()
|
||||
|
||||
ttlCmds := make([]*goredis.DurationCmd, len(msg.Entries))
|
||||
for i, e := range msg.Entries {
|
||||
ttlCmds[i] = pipe.TTL(ctx, c.redisKey(e.Key))
|
||||
}
|
||||
|
||||
if _, err := pipe.Exec(ctx); err != nil {
|
||||
c.logger.WithError(err).Warn("failed to fetch TTLs for sync entries")
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
for i, e := range msg.Entries {
|
||||
ttl, err := ttlCmds[i].Result()
|
||||
if err != nil || ttl <= 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
val, err := c.opts.Decode(e.Data)
|
||||
if err != nil {
|
||||
c.logger.WithError(err).Warn("failed to decode sync entry")
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
c.inner.Put(e.Key, val, ttl)
|
||||
}
|
||||
}
|
||||
Vendored
+18
@@ -0,0 +1,18 @@
|
||||
package cache
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/0xERR0R/blocky/log"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
func init() {
|
||||
log.Silence()
|
||||
}
|
||||
|
||||
func TestCacheSuite(t *testing.T) {
|
||||
RegisterFailHandler(Fail)
|
||||
RunSpecs(t, "Cache Suite")
|
||||
}
|
||||
Vendored
+733
@@ -0,0 +1,733 @@
|
||||
package cache
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
expirationcache "github.com/0xERR0R/expiration-cache"
|
||||
"github.com/alicebob/miniredis/v2"
|
||||
goredis "github.com/go-redis/redis/v8"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
// testValue is a simple string wrapper used as cache value type in tests.
|
||||
type testValue struct {
|
||||
Data string `json:"data"`
|
||||
}
|
||||
|
||||
func encodeTestValue(v *testValue) ([]byte, error) {
|
||||
return json.Marshal(v)
|
||||
}
|
||||
|
||||
func decodeTestValue(b []byte) (*testValue, error) {
|
||||
var v testValue
|
||||
if err := json.Unmarshal(b, &v); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return &v, nil
|
||||
}
|
||||
|
||||
// newTestInner creates a fresh in-memory expiration cache.
|
||||
func newTestInner(ctx context.Context) ExpiringCache[testValue] {
|
||||
return expirationcache.NewCache[testValue](ctx, expirationcache.Options{})
|
||||
}
|
||||
|
||||
// defaultOpts returns a RedisOptions with the given prefix suitable for tests.
|
||||
func defaultOpts(prefix string) RedisOptions[testValue] {
|
||||
return RedisOptions[testValue]{
|
||||
Prefix: prefix,
|
||||
Channel: "test-sync-" + prefix,
|
||||
Encode: encodeTestValue,
|
||||
Decode: decodeTestValue,
|
||||
BatchSize: 100,
|
||||
FlushInterval: 20 * time.Millisecond,
|
||||
SendBufSize: 1000,
|
||||
}
|
||||
}
|
||||
|
||||
// newRedisClient creates a goredis client pointing at the given miniredis server.
|
||||
func newRedisClient(srv *miniredis.Miniredis) *goredis.Client {
|
||||
return goredis.NewClient(&goredis.Options{Addr: srv.Addr()})
|
||||
}
|
||||
|
||||
var _ = Describe("RedisExpiringCache", func() {
|
||||
var (
|
||||
srv *miniredis.Miniredis
|
||||
client *goredis.Client
|
||||
)
|
||||
|
||||
BeforeEach(func() {
|
||||
var err error
|
||||
srv, err = miniredis.Run()
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
DeferCleanup(srv.Close)
|
||||
|
||||
client = newRedisClient(srv)
|
||||
DeferCleanup(func() { _ = client.Close() })
|
||||
})
|
||||
|
||||
Describe("Put", func() {
|
||||
When("a value is stored", func() {
|
||||
It("stores in inner AND Redis", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
inner := newTestInner(ctx)
|
||||
opts := defaultOpts("put:")
|
||||
|
||||
c, err := NewRedisExpiringCache(ctx, inner, client, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
c.Put("foo", &testValue{Data: "bar"}, time.Minute)
|
||||
|
||||
// Inner cache must have the entry immediately.
|
||||
val, _ := inner.Get("foo")
|
||||
Expect(val).ToNot(BeNil())
|
||||
Expect(val.Data).To(Equal("bar"))
|
||||
|
||||
// Redis key must appear after async flush.
|
||||
Eventually(func() bool {
|
||||
exists, _ := client.Exists(ctx, "put:foo").Result()
|
||||
|
||||
return exists > 0
|
||||
}).WithTimeout(2 * time.Second).WithPolling(20 * time.Millisecond).Should(BeTrue())
|
||||
})
|
||||
})
|
||||
|
||||
When("the send buffer is full", func() {
|
||||
It("completes without blocking", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
inner := newTestInner(ctx)
|
||||
opts := defaultOpts("full:")
|
||||
opts.SendBufSize = 2
|
||||
opts.FlushInterval = 10 * time.Minute // prevent flush during test
|
||||
|
||||
c, err := NewRedisExpiringCache(ctx, inner, client, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
defer close(done)
|
||||
|
||||
for i := 0; i < 10; i++ {
|
||||
c.Put("key", &testValue{Data: "v"}, time.Minute)
|
||||
}
|
||||
}()
|
||||
|
||||
Eventually(done).WithTimeout(time.Second).Should(BeClosed())
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Get", func() {
|
||||
When("a value exists only in Redis (not in inner cache)", func() {
|
||||
It("returns nil", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
// Pre-populate Redis with a key under a different prefix so
|
||||
// the decorator's startup scan does not pick it up.
|
||||
data, _ := encodeTestValue(&testValue{Data: "direct"})
|
||||
Expect(client.Set(ctx, "other:foo", data, time.Minute).Err()).ToNot(HaveOccurred())
|
||||
|
||||
// Create decorator with a prefix that does NOT match the key above.
|
||||
innerEmpty := newTestInner(ctx)
|
||||
optsEmpty := defaultOpts("get-empty:")
|
||||
|
||||
c, err := NewRedisExpiringCache(ctx, innerEmpty, client, optsEmpty)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
val, _ := c.Get("foo")
|
||||
Expect(val).To(BeNil())
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Clear", func() {
|
||||
When("entries exist in both inner cache and Redis", func() {
|
||||
It("clears both", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
inner := newTestInner(ctx)
|
||||
opts := defaultOpts("clear:")
|
||||
|
||||
c, err := NewRedisExpiringCache(ctx, inner, client, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
c.Put("foo", &testValue{Data: "bar"}, time.Minute)
|
||||
|
||||
// Wait for Redis write.
|
||||
Eventually(func() bool {
|
||||
exists, _ := client.Exists(ctx, "clear:foo").Result()
|
||||
|
||||
return exists > 0
|
||||
}).WithTimeout(2 * time.Second).WithPolling(20 * time.Millisecond).Should(BeTrue())
|
||||
|
||||
c.Clear()
|
||||
|
||||
Expect(inner.TotalCount()).To(Equal(0))
|
||||
|
||||
keys, err := client.Keys(ctx, "clear:*").Result()
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(keys).To(BeEmpty())
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Cross-instance sync", func() {
|
||||
When("two instances share the same Redis", func() {
|
||||
It("propagates Put from instance1 to instance2's inner cache", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
sharedOpts := defaultOpts("sync:")
|
||||
sharedOpts.FlushInterval = 10 * time.Millisecond
|
||||
|
||||
inner1 := newTestInner(ctx)
|
||||
c1, err := NewRedisExpiringCache(ctx, inner1, client, sharedOpts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
inner2 := newTestInner(ctx)
|
||||
c2, err := NewRedisExpiringCache(ctx, inner2, client, sharedOpts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
_ = c2 // used via inner2
|
||||
|
||||
c1.Put("shared", &testValue{Data: "hello"}, time.Minute)
|
||||
|
||||
Eventually(func() *testValue {
|
||||
val, _ := inner2.Get("shared")
|
||||
|
||||
return val
|
||||
}).WithTimeout(3 * time.Second).WithPolling(20 * time.Millisecond).ShouldNot(BeNil())
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Echo filtering", func() {
|
||||
When("an instance publishes a message", func() {
|
||||
It("does not double-count the entry in its own cache", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
inner := newTestInner(ctx)
|
||||
opts := defaultOpts("echo:")
|
||||
opts.FlushInterval = 10 * time.Millisecond
|
||||
|
||||
c, err := NewRedisExpiringCache(ctx, inner, client, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
c.Put("key", &testValue{Data: "v"}, time.Minute)
|
||||
|
||||
// Wait for the flush to happen.
|
||||
Eventually(func() bool {
|
||||
exists, _ := client.Exists(ctx, "echo:key").Result()
|
||||
|
||||
return exists > 0
|
||||
}).WithTimeout(2 * time.Second).WithPolling(10 * time.Millisecond).Should(BeTrue())
|
||||
|
||||
// TotalCount must remain 1, not grow due to echo.
|
||||
Consistently(func() int {
|
||||
return c.TotalCount()
|
||||
}, "300ms", "30ms").Should(Equal(1))
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Startup cache load", func() {
|
||||
When("Redis already contains keys with the configured prefix", func() {
|
||||
It("loads those entries into the inner cache on creation", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
data, _ := encodeTestValue(&testValue{Data: "pre-existing"})
|
||||
Expect(client.Set(ctx, "load:existing", data, time.Minute).Err()).ToNot(HaveOccurred())
|
||||
|
||||
inner := newTestInner(ctx)
|
||||
opts := defaultOpts("load:")
|
||||
|
||||
_, err := NewRedisExpiringCache(ctx, inner, client, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
val, _ := inner.Get("existing")
|
||||
Expect(val).ToNot(BeNil())
|
||||
Expect(val.Data).To(Equal("pre-existing"))
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Put with zero expiration", func() {
|
||||
When("expiration is zero", func() {
|
||||
It("stores in inner but does not enqueue to Redis", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
inner := newTestInner(ctx)
|
||||
opts := defaultOpts("zero-exp:")
|
||||
|
||||
c, err := NewRedisExpiringCache(ctx, inner, client, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
c.Put("no-redis", &testValue{Data: "val"}, 0)
|
||||
|
||||
// Inner should not cache it either (expiration <= 0 means no cache)
|
||||
val, _ := inner.Get("no-redis")
|
||||
Expect(val).To(BeNil())
|
||||
|
||||
// Redis should never get this key
|
||||
Consistently(func() bool {
|
||||
exists, _ := client.Exists(ctx, "zero-exp:no-redis").Result()
|
||||
|
||||
return exists > 0
|
||||
}, "200ms", "50ms").Should(BeFalse())
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("TotalCount", func() {
|
||||
When("multiple values are stored", func() {
|
||||
It("returns the count from the inner cache", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
inner := newTestInner(ctx)
|
||||
opts := defaultOpts("count:")
|
||||
|
||||
c, err := NewRedisExpiringCache(ctx, inner, client, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
c.Put("a", &testValue{Data: "1"}, time.Minute)
|
||||
c.Put("b", &testValue{Data: "2"}, time.Minute)
|
||||
c.Put("c", &testValue{Data: "3"}, time.Minute)
|
||||
|
||||
Expect(c.TotalCount()).To(Equal(3))
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("handleSyncMessage", func() {
|
||||
When("the payload is invalid JSON", func() {
|
||||
It("does not add entries to the cache", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
inner := newTestInner(ctx)
|
||||
opts := defaultOpts("bad-json:")
|
||||
|
||||
c, err := NewRedisExpiringCache(ctx, inner, client, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
c.handleSyncMessage(ctx, []byte("not-valid-json{{{"))
|
||||
|
||||
Expect(inner.TotalCount()).To(Equal(0))
|
||||
})
|
||||
})
|
||||
|
||||
When("the message is from the same instance", func() {
|
||||
It("skips the message", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
inner := newTestInner(ctx)
|
||||
opts := defaultOpts("self:")
|
||||
|
||||
c, err := NewRedisExpiringCache(ctx, inner, client, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
data, _ := encodeTestValue(&testValue{Data: "val"})
|
||||
msg := redisSyncMessage{
|
||||
Entries: []redisSyncEntry{{Key: "k", Data: data}},
|
||||
Client: c.instanceID,
|
||||
}
|
||||
payload, _ := json.Marshal(msg)
|
||||
|
||||
c.handleSyncMessage(ctx, payload)
|
||||
|
||||
Expect(inner.TotalCount()).To(Equal(0))
|
||||
})
|
||||
})
|
||||
|
||||
When("a sync entry has invalid data", func() {
|
||||
It("skips the bad entry but processes valid ones", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
inner := newTestInner(ctx)
|
||||
opts := defaultOpts("bad-entry:")
|
||||
|
||||
c, err := NewRedisExpiringCache(ctx, inner, client, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
goodData, _ := encodeTestValue(&testValue{Data: "good"})
|
||||
|
||||
// Store keys in Redis so handleSyncMessage can fetch their TTLs.
|
||||
client.Set(ctx, "bad-entry:bad", []byte("not-json"), time.Minute)
|
||||
client.Set(ctx, "bad-entry:good", goodData, time.Minute)
|
||||
|
||||
msg := redisSyncMessage{
|
||||
Entries: []redisSyncEntry{
|
||||
{Key: "bad", Data: []byte("not-json")},
|
||||
{Key: "good", Data: goodData},
|
||||
},
|
||||
Client: "other-instance",
|
||||
}
|
||||
payload, _ := json.Marshal(msg)
|
||||
|
||||
c.handleSyncMessage(ctx, payload)
|
||||
|
||||
Expect(inner.TotalCount()).To(Equal(1))
|
||||
val, _ := inner.Get("good")
|
||||
Expect(val).ToNot(BeNil())
|
||||
Expect(val.Data).To(Equal("good"))
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("flushBatch", func() {
|
||||
When("encoding fails for an entry", func() {
|
||||
It("skips the bad entry and flushes the rest", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
inner := newTestInner(ctx)
|
||||
opts := defaultOpts("enc-err:")
|
||||
|
||||
callCount := 0
|
||||
opts.Encode = func(v *testValue) ([]byte, error) {
|
||||
callCount++
|
||||
if v.Data == "fail" {
|
||||
return nil, errors.New("encode error")
|
||||
}
|
||||
|
||||
return encodeTestValue(v)
|
||||
}
|
||||
|
||||
c, err := NewRedisExpiringCache(ctx, inner, client, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
batch := []sendBufferEntry[testValue]{
|
||||
{key: "fail-key", val: &testValue{Data: "fail"}, expiration: time.Minute},
|
||||
{key: "ok-key", val: &testValue{Data: "ok"}, expiration: time.Minute},
|
||||
}
|
||||
c.flushBatch(ctx, batch)
|
||||
|
||||
// Only the "ok" entry should be in Redis
|
||||
exists, _ := client.Exists(ctx, "enc-err:ok-key").Result()
|
||||
Expect(exists).To(Equal(int64(1)))
|
||||
|
||||
exists, _ = client.Exists(ctx, "enc-err:fail-key").Result()
|
||||
Expect(exists).To(Equal(int64(0)))
|
||||
})
|
||||
})
|
||||
|
||||
When("the channel is empty", func() {
|
||||
It("writes to Redis but does not publish sync messages", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
inner := newTestInner(ctx)
|
||||
opts := defaultOpts("no-chan:")
|
||||
opts.Channel = ""
|
||||
|
||||
c, err := NewRedisExpiringCache(ctx, inner, client, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
batch := []sendBufferEntry[testValue]{
|
||||
{key: "k", val: &testValue{Data: "v"}, expiration: time.Minute},
|
||||
}
|
||||
c.flushBatch(ctx, batch)
|
||||
|
||||
// Entry should still be in Redis
|
||||
exists, _ := client.Exists(ctx, "no-chan:k").Result()
|
||||
Expect(exists).To(Equal(int64(1)))
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("NewRedisExpiringCache defaults", func() {
|
||||
When("options have zero values for batch size, flush interval, and send buf size", func() {
|
||||
It("applies defaults", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
inner := newTestInner(ctx)
|
||||
opts := RedisOptions[testValue]{
|
||||
Prefix: "defaults:",
|
||||
Channel: "defaults-chan",
|
||||
Encode: encodeTestValue,
|
||||
Decode: decodeTestValue,
|
||||
// All zero — should get defaults
|
||||
}
|
||||
|
||||
c, err := NewRedisExpiringCache(ctx, inner, client, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
Expect(c.opts.BatchSize).To(Equal(100))
|
||||
Expect(c.opts.FlushInterval).To(Equal(100 * time.Millisecond))
|
||||
Expect(c.opts.SendBufSize).To(Equal(1000))
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Clear with scan error", func() {
|
||||
When("the Redis connection fails during Clear", func() {
|
||||
It("does not panic and clears the inner cache", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
inner := newTestInner(ctx)
|
||||
opts := defaultOpts("clear-err:")
|
||||
|
||||
c, err := NewRedisExpiringCache(ctx, inner, client, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
c.Put("foo", &testValue{Data: "bar"}, time.Minute)
|
||||
|
||||
// Close the Redis server to cause scan failure
|
||||
srv.Close()
|
||||
|
||||
c.Clear()
|
||||
|
||||
// Inner cache should still be cleared
|
||||
Expect(inner.TotalCount()).To(Equal(0))
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("loadKeysViaPipeline with decode errors", func() {
|
||||
When("a stored value cannot be decoded", func() {
|
||||
It("skips that key and continues", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
// Store a key with bad data directly in Redis
|
||||
Expect(client.Set(ctx, "decode-err:bad", "not-valid-json", time.Minute).Err()).ToNot(HaveOccurred())
|
||||
|
||||
// Store a key with good data
|
||||
goodData, _ := encodeTestValue(&testValue{Data: "good"})
|
||||
Expect(client.Set(ctx, "decode-err:good", goodData, time.Minute).Err()).ToNot(HaveOccurred())
|
||||
|
||||
inner := newTestInner(ctx)
|
||||
opts := defaultOpts("decode-err:")
|
||||
|
||||
_, err := NewRedisExpiringCache(ctx, inner, client, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
// Only the good key should be loaded
|
||||
val, _ := inner.Get("good")
|
||||
Expect(val).ToNot(BeNil())
|
||||
Expect(val.Data).To(Equal("good"))
|
||||
|
||||
// The bad key should be skipped
|
||||
badVal, _ := inner.Get("bad")
|
||||
Expect(badVal).To(BeNil())
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("runWriter batch size trigger", func() {
|
||||
When("enough entries are buffered to fill a batch", func() {
|
||||
It("flushes immediately without waiting for the timer", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
inner := newTestInner(ctx)
|
||||
opts := defaultOpts("batch:")
|
||||
opts.BatchSize = 3
|
||||
opts.FlushInterval = 10 * time.Minute // prevent timer flush
|
||||
|
||||
c, err := NewRedisExpiringCache(ctx, inner, client, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
// Put exactly BatchSize entries to trigger immediate flush
|
||||
c.Put("a", &testValue{Data: "1"}, time.Minute)
|
||||
c.Put("b", &testValue{Data: "2"}, time.Minute)
|
||||
c.Put("c", &testValue{Data: "3"}, time.Minute)
|
||||
|
||||
Eventually(func() bool {
|
||||
exists, _ := client.Exists(ctx, "batch:a", "batch:b", "batch:c").Result()
|
||||
|
||||
return exists == 3
|
||||
}).WithTimeout(2 * time.Second).WithPolling(20 * time.Millisecond).Should(BeTrue())
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("runSubscriber with empty channel", func() {
|
||||
When("the Channel option is empty", func() {
|
||||
It("does not subscribe to pub/sub", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
inner := newTestInner(ctx)
|
||||
opts := defaultOpts("no-sub:")
|
||||
opts.Channel = ""
|
||||
|
||||
c, err := NewRedisExpiringCache(ctx, inner, client, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
// Put a value via the cache (will write to Redis but not publish sync)
|
||||
c.Put("local", &testValue{Data: "local-only"}, time.Minute)
|
||||
|
||||
val, _ := c.Get("local")
|
||||
Expect(val).ToNot(BeNil())
|
||||
Expect(val.Data).To(Equal("local-only"))
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Graceful degradation", func() {
|
||||
When("the Redis client points to a dead server", func() {
|
||||
It("Put and Get still work locally", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
deadClient := goredis.NewClient(&goredis.Options{
|
||||
Addr: "127.0.0.1:0",
|
||||
DialTimeout: 50 * time.Millisecond,
|
||||
ReadTimeout: 50 * time.Millisecond,
|
||||
})
|
||||
DeferCleanup(deadClient.Close)
|
||||
|
||||
inner := newTestInner(ctx)
|
||||
opts := defaultOpts("dead:")
|
||||
|
||||
// Startup scan will fail but should not return error (just log).
|
||||
c, err := NewRedisExpiringCache(ctx, inner, deadClient, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
c.Put("local", &testValue{Data: "value"}, time.Minute)
|
||||
|
||||
val, _ := c.Get("local")
|
||||
Expect(val).ToNot(BeNil())
|
||||
Expect(val.Data).To(Equal("value"))
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("handleSyncMessage with TTL fetch failure", func() {
|
||||
When("the Redis server is down during TTL fetch", func() {
|
||||
It("skips all entries without panic", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
inner := newTestInner(ctx)
|
||||
opts := defaultOpts("ttl-fail:")
|
||||
|
||||
c, err := NewRedisExpiringCache(ctx, inner, client, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
goodData, _ := encodeTestValue(&testValue{Data: "val"})
|
||||
|
||||
// Close Redis to cause TTL fetch failure
|
||||
srv.Close()
|
||||
|
||||
msg := redisSyncMessage{
|
||||
Entries: []redisSyncEntry{{Key: "k", Data: goodData}},
|
||||
Client: "other-instance",
|
||||
}
|
||||
payload, _ := json.Marshal(msg)
|
||||
|
||||
c.handleSyncMessage(ctx, payload)
|
||||
|
||||
Expect(inner.TotalCount()).To(Equal(0))
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("handleSyncMessage with missing TTL", func() {
|
||||
When("a sync entry's key has no expiration in Redis", func() {
|
||||
It("skips the entry", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
inner := newTestInner(ctx)
|
||||
opts := defaultOpts("no-ttl:")
|
||||
|
||||
c, err := NewRedisExpiringCache(ctx, inner, client, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
data, _ := encodeTestValue(&testValue{Data: "val"})
|
||||
|
||||
// Store key WITHOUT expiration (TTL = -1 in Redis)
|
||||
client.Set(ctx, "no-ttl:k", data, 0)
|
||||
|
||||
msg := redisSyncMessage{
|
||||
Entries: []redisSyncEntry{{Key: "k", Data: data}},
|
||||
Client: "other-instance",
|
||||
}
|
||||
payload, _ := json.Marshal(msg)
|
||||
|
||||
c.handleSyncMessage(ctx, payload)
|
||||
|
||||
Expect(inner.TotalCount()).To(Equal(0))
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("flushBatch with publish failure", func() {
|
||||
When("the Redis server goes down after entries are written", func() {
|
||||
It("does not panic", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
inner := newTestInner(ctx)
|
||||
opts := defaultOpts("pub-fail:")
|
||||
|
||||
c, err := NewRedisExpiringCache(ctx, inner, client, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
// Close Redis so both pipeline exec and publish will fail
|
||||
srv.Close()
|
||||
|
||||
batch := []sendBufferEntry[testValue]{
|
||||
{key: "k", val: &testValue{Data: "v"}, expiration: time.Minute},
|
||||
}
|
||||
Expect(func() { c.flushBatch(ctx, batch) }).ToNot(Panic())
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("NewRedisExpiringByteCache", func() {
|
||||
When("no codec is provided", func() {
|
||||
It("uses identity encode/decode functions", func() {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
inner := expirationcache.NewCache[[]byte](ctx, expirationcache.Options{})
|
||||
|
||||
opts := RedisOptions[[]byte]{
|
||||
Prefix: "bytes:",
|
||||
Channel: "bytes-chan",
|
||||
// Encode and Decode intentionally nil — defaults should apply
|
||||
}
|
||||
|
||||
c, err := NewRedisExpiringByteCache(ctx, inner, client, opts)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
original := []byte("hello world")
|
||||
c.Put("test", &original, time.Minute)
|
||||
|
||||
// Wait for Redis write
|
||||
Eventually(func() bool {
|
||||
exists, _ := client.Exists(ctx, "bytes:test").Result()
|
||||
|
||||
return exists > 0
|
||||
}).WithTimeout(2 * time.Second).WithPolling(20 * time.Millisecond).Should(BeTrue())
|
||||
|
||||
// Verify round-trip: read from Redis and decode
|
||||
data, err := client.Get(ctx, "bytes:test").Bytes()
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(data).To(Equal(original))
|
||||
})
|
||||
})
|
||||
})
|
||||
})
|
||||
@@ -2,6 +2,8 @@ package e2e
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
. "github.com/0xERR0R/blocky/helpertest"
|
||||
@@ -168,6 +170,228 @@ var _ = Describe("Redis configuration tests", func() {
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Blocking state sync via Redis", func() {
|
||||
When("Redis, blocking, and 2 blocky instances are configured", func() {
|
||||
BeforeEach(func(ctx context.Context) {
|
||||
_, err = createDNSMokkaContainer(ctx, "moka2", e2eNet,
|
||||
`A blocked.com/NOERROR("A 5.6.7.8 123")`,
|
||||
)
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
_, err = createHTTPServerContainer(ctx, "httpserver", e2eNet, "list.txt", "blocked.com")
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
blocky1, err = createBlockyContainerFromString(ctx, e2eNet, dedent(`
|
||||
log:
|
||||
level: warn
|
||||
upstreams:
|
||||
groups:
|
||||
default:
|
||||
- moka2
|
||||
ports:
|
||||
http: 4000
|
||||
blocking:
|
||||
denylists:
|
||||
ads:
|
||||
- http://httpserver:8080/list.txt
|
||||
clientGroupsBlock:
|
||||
default:
|
||||
- ads
|
||||
redis:
|
||||
address: redis:6379
|
||||
`))
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
blocky2, err = createBlockyContainerFromString(ctx, e2eNet, dedent(`
|
||||
log:
|
||||
level: warn
|
||||
upstreams:
|
||||
groups:
|
||||
default:
|
||||
- moka2
|
||||
ports:
|
||||
http: 4000
|
||||
blocking:
|
||||
denylists:
|
||||
ads:
|
||||
- http://httpserver:8080/list.txt
|
||||
clientGroupsBlock:
|
||||
default:
|
||||
- ads
|
||||
redis:
|
||||
address: redis:6379
|
||||
`))
|
||||
Expect(err).Should(Succeed())
|
||||
})
|
||||
|
||||
It("syncs blocking disable/enable between instances", func(ctx context.Context) {
|
||||
msg := util.NewMsgWithQuestion("blocked.com.", A)
|
||||
|
||||
By("verifying blocking works on both instances", func() {
|
||||
Eventually(doDNSRequest, "5s", "2ms").WithArguments(ctx, blocky1, msg).
|
||||
Should(BeDNSRecord("blocked.com.", A, "0.0.0.0"))
|
||||
Eventually(doDNSRequest, "5s", "2ms").WithArguments(ctx, blocky2, msg).
|
||||
Should(BeDNSRecord("blocked.com.", A, "0.0.0.0"))
|
||||
})
|
||||
|
||||
By("disabling blocking on instance1 via API", func() {
|
||||
host, port, err := getContainerHostPort(ctx, blocky1, "4000/tcp")
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
resp, err := http.Get("http://" + net.JoinHostPort(host, port) + "/api/blocking/disable")
|
||||
Expect(err).Should(Succeed())
|
||||
defer resp.Body.Close()
|
||||
Expect(resp.StatusCode).Should(Equal(http.StatusOK))
|
||||
})
|
||||
|
||||
By("verifying instance2 also has blocking disabled", func() {
|
||||
Eventually(doDNSRequest, "5s", "100ms").WithArguments(ctx, blocky2, msg).
|
||||
Should(BeDNSRecord("blocked.com.", A, "5.6.7.8"))
|
||||
})
|
||||
|
||||
By("re-enabling blocking on instance1", func() {
|
||||
host, port, err := getContainerHostPort(ctx, blocky1, "4000/tcp")
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
resp, err := http.Get("http://" + net.JoinHostPort(host, port) + "/api/blocking/enable")
|
||||
Expect(err).Should(Succeed())
|
||||
defer resp.Body.Close()
|
||||
Expect(resp.StatusCode).Should(Equal(http.StatusOK))
|
||||
})
|
||||
|
||||
By("verifying instance2 re-enables blocking", func() {
|
||||
Eventually(doDNSRequest, "5s", "100ms").WithArguments(ctx, blocky2, msg).
|
||||
Should(BeDNSRecord("blocked.com.", A, "0.0.0.0"))
|
||||
})
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Cache clear propagation via Redis", func() {
|
||||
When("Redis and 2 blocky instances share cached entries", func() {
|
||||
BeforeEach(func(ctx context.Context) {
|
||||
blocky1, err = createBlockyContainerFromString(ctx, e2eNet, dedent(`
|
||||
log:
|
||||
level: warn
|
||||
upstreams:
|
||||
groups:
|
||||
default:
|
||||
- moka1
|
||||
ports:
|
||||
http: 4000
|
||||
caching:
|
||||
minTime: 5m
|
||||
redis:
|
||||
address: redis:6379
|
||||
`))
|
||||
Expect(err).Should(Succeed())
|
||||
})
|
||||
|
||||
It("clears cache entries from Redis when flushed via API", func(ctx context.Context) {
|
||||
msg := util.NewMsgWithQuestion("google.de.", A)
|
||||
|
||||
By("populating cache via instance1", func() {
|
||||
Eventually(doDNSRequest, "5s", "2ms").WithArguments(ctx, blocky1, msg).
|
||||
Should(BeDNSRecord("google.de.", A, "1.2.3.4"))
|
||||
})
|
||||
|
||||
By("waiting for Redis to have the entry", func() {
|
||||
Eventually(dbSize, "5s", "2ms").WithArguments(ctx, redisClient).
|
||||
Should(BeNumerically(">=", 1))
|
||||
})
|
||||
|
||||
By("flushing cache via instance1 API", func() {
|
||||
host, port, err := getContainerHostPort(ctx, blocky1, "4000/tcp")
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
resp, err := http.Post(
|
||||
"http://"+net.JoinHostPort(host, port)+"/api/cache/flush",
|
||||
"application/json", nil)
|
||||
Expect(err).Should(Succeed())
|
||||
defer resp.Body.Close()
|
||||
Expect(resp.StatusCode).Should(Equal(http.StatusOK))
|
||||
})
|
||||
|
||||
By("verifying Redis cache was cleared", func() {
|
||||
Eventually(dbSize, "5s", "100ms").WithArguments(ctx, redisClient).
|
||||
Should(BeNumerically("==", 0))
|
||||
})
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Multiple query types via Redis cache", func() {
|
||||
When("both A and AAAA queries are cached", func() {
|
||||
BeforeEach(func(ctx context.Context) {
|
||||
_, err = createDNSMokkaContainer(ctx, "moka2", e2eNet,
|
||||
`A google/NOERROR("A 1.2.3.4 123")`,
|
||||
`AAAA google/NOERROR("AAAA 2001:db8::1 123")`,
|
||||
)
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
blocky1, err = createBlockyContainerFromString(ctx, e2eNet, dedent(`
|
||||
log:
|
||||
level: warn
|
||||
upstreams:
|
||||
groups:
|
||||
default:
|
||||
- moka2
|
||||
redis:
|
||||
address: redis:6379
|
||||
`))
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
blocky2, err = createBlockyContainerFromString(ctx, e2eNet, dedent(`
|
||||
log:
|
||||
level: warn
|
||||
upstreams:
|
||||
groups:
|
||||
default:
|
||||
- moka2
|
||||
redis:
|
||||
address: redis:6379
|
||||
`))
|
||||
Expect(err).Should(Succeed())
|
||||
})
|
||||
|
||||
It("shares both A and AAAA cached responses via Redis", func(ctx context.Context) {
|
||||
By("querying A and AAAA on instance1", func() {
|
||||
Eventually(doDNSRequest, "5s", "2ms").
|
||||
WithArguments(ctx, blocky1, util.NewMsgWithQuestion("google.de.", A)).
|
||||
Should(BeDNSRecord("google.de.", A, "1.2.3.4"))
|
||||
|
||||
Eventually(doDNSRequest, "5s", "2ms").
|
||||
WithArguments(ctx, blocky1, util.NewMsgWithQuestion("google.de.", AAAA)).
|
||||
Should(BeDNSRecord("google.de.", AAAA, "2001:db8::1"))
|
||||
})
|
||||
|
||||
By("waiting for Redis entries", func() {
|
||||
Eventually(dbSize, "5s", "2ms").WithArguments(ctx, redisClient).
|
||||
Should(BeNumerically(">=", 2))
|
||||
})
|
||||
|
||||
By("shutting down upstream", func() {
|
||||
Expect(mokka.Terminate(ctx)).Should(Succeed())
|
||||
})
|
||||
|
||||
By("querying instance2 for both types from cache", func() {
|
||||
Eventually(doDNSRequest, "5s", "2ms").
|
||||
WithArguments(ctx, blocky2, util.NewMsgWithQuestion("google.de.", A)).
|
||||
Should(BeDNSRecord("google.de.", A, "1.2.3.4"))
|
||||
|
||||
Eventually(doDNSRequest, "5s", "2ms").
|
||||
WithArguments(ctx, blocky2, util.NewMsgWithQuestion("google.de.", AAAA)).
|
||||
Should(BeDNSRecord("google.de.", AAAA, "2001:db8::1"))
|
||||
})
|
||||
|
||||
By("No warnings/errors in log", func() {
|
||||
Expect(getContainerLogs(ctx, blocky1)).Should(BeEmpty())
|
||||
Expect(getContainerLogs(ctx, blocky2)).Should(BeEmpty())
|
||||
})
|
||||
})
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
func dbSize(ctx context.Context, redisClient *redis.Client) (int64, error) {
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
package evt
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"github.com/asaskevich/EventBus"
|
||||
)
|
||||
|
||||
@@ -8,6 +10,14 @@ const (
|
||||
// BlockingEnabledEvent fires if blocking status will be changed. Parameter: boolean (enabled = true)
|
||||
BlockingEnabledEvent = "blocking:enabled"
|
||||
|
||||
// BlockingStateChanged fires when blocking state changes locally (for Redis bridge).
|
||||
// Parameter: BlockingState
|
||||
BlockingStateChanged = "blocking:stateChanged"
|
||||
|
||||
// BlockingStateChangedRemote fires when blocking state changes from a remote instance via Redis.
|
||||
// Parameter: BlockingState
|
||||
BlockingStateChangedRemote = "blocking:stateChangedRemote"
|
||||
|
||||
// BlockingCacheGroupChanged fires, if a list group is changed. Parameter: list type, group name, element count
|
||||
BlockingCacheGroupChanged = "blocking:cachingGroupChanged"
|
||||
|
||||
@@ -37,3 +47,10 @@ var evtBus = EventBus.New()
|
||||
func Bus() EventBus.Bus {
|
||||
return evtBus
|
||||
}
|
||||
|
||||
// BlockingState carries the full blocking state for cross-instance sync.
|
||||
type BlockingState struct {
|
||||
Enabled bool `json:"enabled"`
|
||||
Duration time.Duration `json:"duration,omitempty"`
|
||||
Groups []string `json:"groups,omitempty"`
|
||||
}
|
||||
|
||||
@@ -61,6 +61,7 @@ func AssertRegistryComplete(t *testing.T, reg *prometheus.Registry) {
|
||||
"blocky_cache_entries",
|
||||
"blocky_cache_hits_total",
|
||||
"blocky_cache_misses_total",
|
||||
"blocky_redis_cache_buffer_drops_total",
|
||||
"blocky_last_list_group_refresh_timestamp_seconds",
|
||||
"blocky_prefetches_total",
|
||||
"blocky_prefetch_hits_total",
|
||||
|
||||
@@ -0,0 +1,144 @@
|
||||
package redis
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/0xERR0R/blocky/evt"
|
||||
"github.com/0xERR0R/blocky/log"
|
||||
goredis "github.com/go-redis/redis/v8"
|
||||
"github.com/google/uuid"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
const (
|
||||
EventBridgeChannel = "blocky_sync_enabled"
|
||||
bridgePublishTimeout = 5 * time.Second
|
||||
)
|
||||
|
||||
type bridgeMessage struct {
|
||||
State evt.BlockingState `json:"s"`
|
||||
Client string `json:"c"`
|
||||
}
|
||||
|
||||
// EventBusBridge connects the local event bus with Redis pub/sub for blocking state synchronization.
|
||||
type EventBusBridge struct {
|
||||
client *goredis.Client
|
||||
id string
|
||||
channel string
|
||||
l *logrus.Entry
|
||||
cancel context.CancelFunc
|
||||
done <-chan struct{}
|
||||
once sync.Once
|
||||
}
|
||||
|
||||
// NewEventBusBridge creates a new EventBusBridge that synchronizes blocking state
|
||||
// between local event bus and Redis pub/sub.
|
||||
func NewEventBusBridge(ctx context.Context, client *goredis.Client) (*EventBusBridge, error) {
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
|
||||
b := &EventBusBridge{
|
||||
client: client,
|
||||
id: uuid.NewString(),
|
||||
channel: EventBridgeChannel,
|
||||
l: log.PrefixedLog("redis-event-bridge"),
|
||||
cancel: cancel,
|
||||
done: ctx.Done(),
|
||||
}
|
||||
|
||||
if err := evt.Bus().Subscribe(evt.BlockingStateChanged, b.onLocalStateChanged); err != nil {
|
||||
cancel()
|
||||
|
||||
return nil, err
|
||||
}
|
||||
|
||||
ps := client.Subscribe(ctx, b.channel)
|
||||
|
||||
if _, err := ps.Receive(ctx); err != nil {
|
||||
_ = ps.Close()
|
||||
_ = evt.Bus().Unsubscribe(evt.BlockingStateChanged, b.onLocalStateChanged)
|
||||
cancel()
|
||||
|
||||
return nil, err
|
||||
}
|
||||
|
||||
loop := &PubSubLoop{
|
||||
Client: client,
|
||||
Channel: b.channel,
|
||||
Logger: b.l,
|
||||
Handler: b.handleMessage,
|
||||
}
|
||||
|
||||
go func() {
|
||||
defer b.Close()
|
||||
|
||||
loop.RunWithSub(ctx, ps)
|
||||
}()
|
||||
|
||||
return b, nil
|
||||
}
|
||||
|
||||
// Close stops the subscriber goroutine and unsubscribes from the local event bus.
|
||||
// It is safe to call multiple times.
|
||||
func (b *EventBusBridge) Close() error {
|
||||
var unsubErr error
|
||||
|
||||
b.once.Do(func() {
|
||||
b.cancel()
|
||||
unsubErr = evt.Bus().Unsubscribe(evt.BlockingStateChanged, b.onLocalStateChanged)
|
||||
})
|
||||
|
||||
return unsubErr
|
||||
}
|
||||
|
||||
// onLocalStateChanged is called when a local blocking state change occurs.
|
||||
// It marshals the state and publishes it to Redis.
|
||||
func (b *EventBusBridge) onLocalStateChanged(state evt.BlockingState) {
|
||||
msg := bridgeMessage{
|
||||
State: state,
|
||||
Client: b.id,
|
||||
}
|
||||
|
||||
payload, err := json.Marshal(msg)
|
||||
if err != nil {
|
||||
b.l.Error("failed to marshal bridge message: ", err)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
pubCtx, cancel := context.WithTimeout(context.Background(), bridgePublishTimeout)
|
||||
defer cancel()
|
||||
|
||||
select {
|
||||
case <-b.done:
|
||||
return
|
||||
default:
|
||||
}
|
||||
|
||||
if err := b.client.Publish(pubCtx, b.channel, payload).Err(); err != nil {
|
||||
b.l.Error("failed to publish to Redis: ", err)
|
||||
}
|
||||
}
|
||||
|
||||
// handleMessage processes a single Redis pub/sub message, publishing remote
|
||||
// state changes to the local event bus and filtering out echoes.
|
||||
func (b *EventBusBridge) handleMessage(_ context.Context, payload string) {
|
||||
if len(payload) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
var bm bridgeMessage
|
||||
if err := json.Unmarshal([]byte(payload), &bm); err != nil {
|
||||
b.l.Error("failed to unmarshal bridge message: ", err)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
if bm.Client == b.id {
|
||||
return
|
||||
}
|
||||
|
||||
evt.Bus().Publish(evt.BlockingStateChangedRemote, bm.State)
|
||||
}
|
||||
@@ -0,0 +1,334 @@
|
||||
package redis
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"time"
|
||||
|
||||
"github.com/0xERR0R/blocky/evt"
|
||||
"github.com/alicebob/miniredis/v2"
|
||||
goredis "github.com/go-redis/redis/v8"
|
||||
"github.com/google/uuid"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
var _ = Describe("EventBusBridge", func() {
|
||||
var (
|
||||
redisServer *miniredis.Miniredis
|
||||
redisClient *goredis.Client
|
||||
bridge *EventBusBridge
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
)
|
||||
|
||||
BeforeEach(func() {
|
||||
var err error
|
||||
redisServer, err = miniredis.Run()
|
||||
Expect(err).Should(Succeed())
|
||||
DeferCleanup(redisServer.Close)
|
||||
|
||||
redisClient = goredis.NewClient(&goredis.Options{
|
||||
Addr: redisServer.Addr(),
|
||||
})
|
||||
DeferCleanup(func() { _ = redisClient.Close() })
|
||||
|
||||
ctx, cancel = context.WithCancel(context.Background())
|
||||
DeferCleanup(cancel)
|
||||
|
||||
bridge, err = NewEventBusBridge(ctx, redisClient)
|
||||
Expect(err).Should(Succeed())
|
||||
DeferCleanup(func() { bridge.Close() })
|
||||
})
|
||||
|
||||
Describe("NewEventBusBridge errors", func() {
|
||||
When("Redis is not reachable", func() {
|
||||
It("should return an error", func() {
|
||||
deadClient := goredis.NewClient(&goredis.Options{
|
||||
Addr: "127.0.0.1:0",
|
||||
DialTimeout: 50 * time.Millisecond,
|
||||
})
|
||||
DeferCleanup(func() { _ = deadClient.Close() })
|
||||
|
||||
newCtx, newCancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
|
||||
DeferCleanup(newCancel)
|
||||
|
||||
_, err := NewEventBusBridge(newCtx, deadClient)
|
||||
Expect(err).Should(HaveOccurred())
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Local event → Redis publish", func() {
|
||||
When("BlockingStateChanged is published on the local bus", func() {
|
||||
It("should publish a message to the Redis channel that a second subscriber receives", func(specCtx context.Context) {
|
||||
// Create a second bridge to act as receiver
|
||||
bridge2, err := NewEventBusBridge(ctx, redisClient)
|
||||
Expect(err).Should(Succeed())
|
||||
DeferCleanup(func() { bridge2.Close() })
|
||||
|
||||
receivedStates := make(chan evt.BlockingState, 1)
|
||||
|
||||
handler := func(state evt.BlockingState) {
|
||||
receivedStates <- state
|
||||
}
|
||||
|
||||
Expect(evt.Bus().Subscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed())
|
||||
DeferCleanup(func() {
|
||||
Expect(evt.Bus().Unsubscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed())
|
||||
})
|
||||
|
||||
expectedState := evt.BlockingState{Enabled: true}
|
||||
evt.Bus().Publish(evt.BlockingStateChanged, expectedState)
|
||||
|
||||
Eventually(receivedStates).WithTimeout(2 * time.Second).WithPolling(50 * time.Millisecond).Should(Receive(Equal(expectedState)))
|
||||
}, SpecTimeout(3*time.Second))
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Redis message → local event", func() {
|
||||
When("a bridgeMessage with a different UUID arrives on Redis", func() {
|
||||
It("should fire BlockingStateChangedRemote on the local bus with the correct state", func(specCtx context.Context) {
|
||||
receivedStates := make(chan evt.BlockingState, 1)
|
||||
|
||||
handler := func(state evt.BlockingState) {
|
||||
receivedStates <- state
|
||||
}
|
||||
|
||||
Expect(evt.Bus().Subscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed())
|
||||
DeferCleanup(func() {
|
||||
Expect(evt.Bus().Unsubscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed())
|
||||
})
|
||||
|
||||
otherID := uuid.NewString()
|
||||
|
||||
expectedState := evt.BlockingState{
|
||||
Enabled: true,
|
||||
Duration: 5 * time.Minute,
|
||||
Groups: []string{"ads", "malware"},
|
||||
}
|
||||
|
||||
payload, err := json.Marshal(bridgeMessage{
|
||||
State: expectedState,
|
||||
Client: otherID,
|
||||
})
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
redisServer.Publish(EventBridgeChannel, string(payload))
|
||||
|
||||
Eventually(receivedStates).Should(Receive(Equal(expectedState)))
|
||||
}, SpecTimeout(3*time.Second))
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Echo filtering", func() {
|
||||
When("a bridgeMessage with the bridge's own UUID arrives on Redis", func() {
|
||||
It("should NOT fire BlockingStateChangedRemote on the local bus", func(specCtx context.Context) {
|
||||
receivedStates := make(chan evt.BlockingState, 1)
|
||||
|
||||
handler := func(state evt.BlockingState) {
|
||||
receivedStates <- state
|
||||
}
|
||||
|
||||
Expect(evt.Bus().Subscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed())
|
||||
DeferCleanup(func() {
|
||||
Expect(evt.Bus().Unsubscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed())
|
||||
})
|
||||
|
||||
payload, err := json.Marshal(bridgeMessage{
|
||||
State: evt.BlockingState{Enabled: true},
|
||||
Client: bridge.id,
|
||||
})
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
redisServer.Publish(EventBridgeChannel, string(payload))
|
||||
|
||||
Consistently(receivedStates, "300ms", "50ms").Should(BeEmpty())
|
||||
}, SpecTimeout(3*time.Second))
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Close", func() {
|
||||
When("Close is called multiple times", func() {
|
||||
It("should be idempotent and not panic", func() {
|
||||
Expect(bridge.Close()).Should(Succeed())
|
||||
Expect(bridge.Close()).Should(Succeed())
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("consumeMessages", func() {
|
||||
When("a message with empty payload arrives", func() {
|
||||
It("should ignore it without error", func(specCtx context.Context) {
|
||||
receivedStates := make(chan evt.BlockingState, 1)
|
||||
|
||||
handler := func(state evt.BlockingState) {
|
||||
receivedStates <- state
|
||||
}
|
||||
|
||||
Expect(evt.Bus().Subscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed())
|
||||
DeferCleanup(func() {
|
||||
Expect(evt.Bus().Unsubscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed())
|
||||
})
|
||||
|
||||
// Publish empty payload
|
||||
redisServer.Publish(EventBridgeChannel, "")
|
||||
|
||||
Consistently(receivedStates, "300ms", "50ms").Should(BeEmpty())
|
||||
}, SpecTimeout(3*time.Second))
|
||||
})
|
||||
|
||||
When("a message with invalid JSON arrives", func() {
|
||||
It("should skip it without firing an event", func(specCtx context.Context) {
|
||||
receivedStates := make(chan evt.BlockingState, 1)
|
||||
|
||||
handler := func(state evt.BlockingState) {
|
||||
receivedStates <- state
|
||||
}
|
||||
|
||||
Expect(evt.Bus().Subscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed())
|
||||
DeferCleanup(func() {
|
||||
Expect(evt.Bus().Unsubscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed())
|
||||
})
|
||||
|
||||
redisServer.Publish(EventBridgeChannel, "not-valid-json{{{")
|
||||
|
||||
Consistently(receivedStates, "300ms", "50ms").Should(BeEmpty())
|
||||
}, SpecTimeout(3*time.Second))
|
||||
})
|
||||
})
|
||||
|
||||
Describe("onLocalStateChanged after close", func() {
|
||||
When("the bridge context is cancelled before publishing", func() {
|
||||
It("should not attempt to publish to Redis", func() {
|
||||
// Cancel the bridge context
|
||||
cancel()
|
||||
|
||||
// onLocalStateChanged should return early due to done channel
|
||||
// This should not panic or block
|
||||
bridge.onLocalStateChanged(evt.BlockingState{Enabled: true})
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("subscribeLoop reconnection", func() {
|
||||
When("the Redis pub/sub connection closes unexpectedly", func() {
|
||||
It("should not panic when pub/sub connection drops", func(specCtx context.Context) {
|
||||
receivedStates := make(chan evt.BlockingState, 5)
|
||||
|
||||
handler := func(state evt.BlockingState) {
|
||||
receivedStates <- state
|
||||
}
|
||||
|
||||
Expect(evt.Bus().Subscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed())
|
||||
DeferCleanup(func() {
|
||||
Expect(evt.Bus().Unsubscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed())
|
||||
})
|
||||
|
||||
otherID := uuid.NewString()
|
||||
|
||||
// Send a message before restart to verify connectivity
|
||||
preState := evt.BlockingState{Enabled: true}
|
||||
payload, err := json.Marshal(bridgeMessage{State: preState, Client: otherID})
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
redisServer.Publish(EventBridgeChannel, string(payload))
|
||||
|
||||
Eventually(receivedStates, "2s", "50ms").Should(Receive(Equal(preState)))
|
||||
|
||||
// Restart miniredis to simulate connection drop + recovery
|
||||
redisServer.Close()
|
||||
redisServer2, err := miniredis.Run()
|
||||
Expect(err).Should(Succeed())
|
||||
DeferCleanup(redisServer2.Close)
|
||||
|
||||
// Note: since miniredis restarts on a new port, the existing bridge
|
||||
// won't reconnect to it. But we verify the reconnect loop runs
|
||||
// without panicking by cancelling context shortly after.
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
}, SpecTimeout(5*time.Second))
|
||||
})
|
||||
|
||||
When("the context is cancelled during reconnect", func() {
|
||||
It("should stop and return without error", func(specCtx context.Context) {
|
||||
// Create a new bridge with its own context
|
||||
bridgeCtx, bridgeCancel := context.WithCancel(context.Background())
|
||||
|
||||
b, err := NewEventBusBridge(bridgeCtx, redisClient)
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
// Close the Redis server to trigger reconnection
|
||||
redisServer.Close()
|
||||
|
||||
// Give the bridge time to notice the disconnection
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
// Cancel the context to stop reconnection
|
||||
bridgeCancel()
|
||||
|
||||
// Verify the bridge shuts down cleanly
|
||||
Eventually(func() error {
|
||||
return b.Close()
|
||||
}).WithTimeout(2 * time.Second).Should(Succeed())
|
||||
}, SpecTimeout(5*time.Second))
|
||||
})
|
||||
})
|
||||
|
||||
Describe("onLocalStateChanged error handling", func() {
|
||||
When("Redis is unreachable during publish", func() {
|
||||
It("does not panic", func() {
|
||||
// Close Redis to cause publish failure
|
||||
redisServer.Close()
|
||||
|
||||
Expect(func() {
|
||||
bridge.onLocalStateChanged(evt.BlockingState{Enabled: true})
|
||||
}).ToNot(Panic())
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Full payload round-trip", func() {
|
||||
When("BlockingStateChanged is published locally with Duration and Groups", func() {
|
||||
It("should preserve Duration and Groups through local → Redis → remote", func(specCtx context.Context) {
|
||||
receivedStates := make(chan evt.BlockingState, 1)
|
||||
|
||||
handler := func(state evt.BlockingState) {
|
||||
receivedStates <- state
|
||||
}
|
||||
|
||||
Expect(evt.Bus().Subscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed())
|
||||
DeferCleanup(func() {
|
||||
Expect(evt.Bus().Unsubscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed())
|
||||
})
|
||||
|
||||
// Create a second bridge acting as the "remote" receiver
|
||||
bridge2, err := NewEventBusBridge(ctx, redisClient)
|
||||
Expect(err).Should(Succeed())
|
||||
DeferCleanup(func() { bridge2.Close() })
|
||||
|
||||
// Override bridge2's handler to forward to our channel
|
||||
Expect(evt.Bus().Unsubscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed())
|
||||
Expect(evt.Bus().Subscribe(evt.BlockingStateChangedRemote, handler)).Should(Succeed())
|
||||
|
||||
// Publish from bridge (bridge2 should receive since UUIDs differ)
|
||||
originalState := evt.BlockingState{
|
||||
Enabled: false,
|
||||
Duration: 2*time.Hour + 30*time.Minute,
|
||||
Groups: []string{"group1", "group2", "group3"},
|
||||
}
|
||||
|
||||
// Directly publish to Redis as bridge (simulate local→Redis path)
|
||||
payload, err := json.Marshal(bridgeMessage{
|
||||
State: originalState,
|
||||
Client: bridge.id, // from bridge, NOT bridge2
|
||||
})
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
redisServer.Publish(EventBridgeChannel, string(payload))
|
||||
|
||||
// bridge2 should receive it (different UUID)
|
||||
Eventually(receivedStates, "2s", "50ms").Should(Receive(Equal(originalState)))
|
||||
}, SpecTimeout(5*time.Second))
|
||||
})
|
||||
})
|
||||
})
|
||||
+119
@@ -0,0 +1,119 @@
|
||||
package redis
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
|
||||
goredis "github.com/go-redis/redis/v8"
|
||||
"github.com/sirupsen/logrus"
|
||||
)
|
||||
|
||||
const (
|
||||
reconnectBaseDelay = 500 * time.Millisecond
|
||||
reconnectMaxDelay = 30 * time.Second
|
||||
)
|
||||
|
||||
// PubSubLoop manages a Redis pub/sub subscription with automatic reconnection.
|
||||
// It calls handler for each received message payload and reconnects with
|
||||
// exponential backoff when the channel closes unexpectedly.
|
||||
type PubSubLoop struct {
|
||||
Client *goredis.Client
|
||||
Channel string
|
||||
Logger *logrus.Entry
|
||||
Handler func(ctx context.Context, payload string)
|
||||
}
|
||||
|
||||
// Run subscribes to the channel and processes messages until ctx is cancelled.
|
||||
// It reconnects automatically on unexpected channel closure.
|
||||
func (p *PubSubLoop) Run(ctx context.Context) {
|
||||
p.RunWithSub(ctx, nil)
|
||||
}
|
||||
|
||||
// RunWithSub is like Run but uses an existing subscription for the first iteration.
|
||||
// If initial is nil, a new subscription is created.
|
||||
func (p *PubSubLoop) RunWithSub(ctx context.Context, initial *goredis.PubSub) {
|
||||
sub := initial
|
||||
if sub == nil {
|
||||
sub = p.Client.Subscribe(ctx, p.Channel)
|
||||
|
||||
if _, err := sub.Receive(ctx); err != nil {
|
||||
p.Logger.WithError(err).Warn("Redis pub/sub initial subscribe failed, attempting to reconnect")
|
||||
_ = sub.Close()
|
||||
|
||||
sub = p.reconnect(ctx)
|
||||
if sub == nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for {
|
||||
p.consume(ctx, sub)
|
||||
_ = sub.Close()
|
||||
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
|
||||
p.Logger.Warn("Redis pub/sub channel closed unexpectedly, attempting to reconnect")
|
||||
|
||||
sub = p.reconnect(ctx)
|
||||
if sub == nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (p *PubSubLoop) consume(ctx context.Context, sub *goredis.PubSub) {
|
||||
ch := sub.Channel()
|
||||
|
||||
for {
|
||||
select {
|
||||
case msg, ok := <-ch:
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
if msg == nil {
|
||||
continue
|
||||
}
|
||||
|
||||
p.Handler(ctx, msg.Payload)
|
||||
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// reconnect attempts to re-subscribe with exponential backoff.
|
||||
// Returns nil if the context is cancelled before reconnecting.
|
||||
func (p *PubSubLoop) reconnect(ctx context.Context) *goredis.PubSub {
|
||||
delay := reconnectBaseDelay
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
case <-time.After(delay):
|
||||
}
|
||||
|
||||
sub := p.Client.Subscribe(ctx, p.Channel)
|
||||
|
||||
if _, err := sub.Receive(ctx); err != nil {
|
||||
_ = sub.Close()
|
||||
p.Logger.WithError(err).Warn("Redis pub/sub reconnect failed, retrying")
|
||||
|
||||
delay *= 2
|
||||
if delay > reconnectMaxDelay {
|
||||
delay = reconnectMaxDelay
|
||||
}
|
||||
|
||||
continue
|
||||
}
|
||||
|
||||
p.Logger.Info("Redis pub/sub reconnected successfully")
|
||||
|
||||
return sub
|
||||
}
|
||||
}
|
||||
+10
-315
@@ -1,81 +1,22 @@
|
||||
package redis
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"math"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/0xERR0R/blocky/config"
|
||||
"github.com/0xERR0R/blocky/log"
|
||||
"github.com/0xERR0R/blocky/model"
|
||||
"github.com/0xERR0R/blocky/util"
|
||||
"github.com/go-redis/redis/v8"
|
||||
"github.com/google/uuid"
|
||||
"github.com/miekg/dns"
|
||||
"github.com/sirupsen/logrus"
|
||||
goredis "github.com/go-redis/redis/v8"
|
||||
)
|
||||
|
||||
const (
|
||||
SyncChannelName = "blocky_sync"
|
||||
CacheStorePrefix = "blocky:cache:"
|
||||
chanCap = 1000
|
||||
cacheReason = "EXTERNAL_CACHE"
|
||||
defaultCacheTime = 1 * time.Second
|
||||
messageTypeCache = 0
|
||||
messageTypeEnable = 1
|
||||
)
|
||||
|
||||
// sendBuffer message
|
||||
type bufferMessage struct {
|
||||
Key string
|
||||
Message *dns.Msg
|
||||
}
|
||||
|
||||
// redis pubsub message
|
||||
type redisMessage struct {
|
||||
Key string `json:"k,omitempty"`
|
||||
Type int `json:"t"`
|
||||
Message []byte `json:"m"`
|
||||
Client []byte `json:"c"`
|
||||
}
|
||||
|
||||
// CacheChannel message
|
||||
type CacheMessage struct {
|
||||
Key string
|
||||
Response *model.Response
|
||||
}
|
||||
|
||||
type EnabledMessage struct {
|
||||
State bool `json:"s"`
|
||||
Duration time.Duration `json:"d,omitempty"`
|
||||
Groups []string `json:"g,omitempty"`
|
||||
}
|
||||
|
||||
// Client for redis communication
|
||||
type Client struct {
|
||||
config *config.Redis
|
||||
client *redis.Client
|
||||
l *logrus.Entry
|
||||
id []byte
|
||||
sendBuffer chan *bufferMessage
|
||||
CacheChannel chan *CacheMessage
|
||||
EnabledChannel chan *EnabledMessage
|
||||
}
|
||||
|
||||
// New creates a new redis client
|
||||
func New(ctx context.Context, cfg *config.Redis) (*Client, error) {
|
||||
// disable redis if no address is provided
|
||||
// New creates a new Redis connection. Returns nil if Redis is not configured.
|
||||
func New(ctx context.Context, cfg *config.Redis) (*goredis.Client, error) {
|
||||
if cfg == nil || len(cfg.Address) == 0 {
|
||||
return nil, nil //nolint:nilnil
|
||||
}
|
||||
|
||||
var baseClient *redis.Client
|
||||
var client *goredis.Client
|
||||
if len(cfg.SentinelAddresses) > 0 {
|
||||
baseClient = redis.NewFailoverClient(&redis.FailoverOptions{
|
||||
client = goredis.NewFailoverClient(&goredis.FailoverOptions{
|
||||
MasterName: cfg.Address,
|
||||
SentinelUsername: cfg.Username,
|
||||
SentinelPassword: cfg.SentinelPassword,
|
||||
@@ -87,7 +28,7 @@ func New(ctx context.Context, cfg *config.Redis) (*Client, error) {
|
||||
MaxRetryBackoff: cfg.ConnectionCooldown.ToDuration(),
|
||||
})
|
||||
} else {
|
||||
baseClient = redis.NewClient(&redis.Options{
|
||||
client = goredis.NewClient(&goredis.Options{
|
||||
Addr: cfg.Address,
|
||||
Username: cfg.Username,
|
||||
Password: cfg.Password,
|
||||
@@ -97,257 +38,11 @@ func New(ctx context.Context, cfg *config.Redis) (*Client, error) {
|
||||
})
|
||||
}
|
||||
|
||||
rdb := baseClient.WithContext(ctx)
|
||||
rdb := client.WithContext(ctx)
|
||||
|
||||
_, err := rdb.Ping(ctx).Result()
|
||||
if err == nil {
|
||||
var id []byte
|
||||
|
||||
id, err = uuid.New().MarshalBinary()
|
||||
if err == nil {
|
||||
// construct client
|
||||
res := &Client{
|
||||
config: cfg,
|
||||
client: rdb,
|
||||
l: log.PrefixedLog("redis"),
|
||||
id: id,
|
||||
sendBuffer: make(chan *bufferMessage, chanCap),
|
||||
CacheChannel: make(chan *CacheMessage, chanCap),
|
||||
EnabledChannel: make(chan *EnabledMessage, chanCap),
|
||||
}
|
||||
|
||||
// start channel handling go routine
|
||||
err = res.startup(ctx)
|
||||
|
||||
return res, err
|
||||
}
|
||||
if _, err := rdb.Ping(ctx).Result(); err != nil {
|
||||
return nil, fmt.Errorf("failed to connect to Redis at '%s': %w", cfg.Address, err)
|
||||
}
|
||||
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// PublishCache publish cache to redis async
|
||||
func (c *Client) PublishCache(key string, message *dns.Msg) {
|
||||
if len(key) > 0 && message != nil {
|
||||
c.sendBuffer <- &bufferMessage{
|
||||
Key: key,
|
||||
Message: message,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) PublishEnabled(ctx context.Context, state *EnabledMessage) {
|
||||
binState, sErr := json.Marshal(state)
|
||||
if sErr == nil {
|
||||
binMsg, mErr := json.Marshal(redisMessage{
|
||||
Type: messageTypeEnable,
|
||||
Message: binState,
|
||||
Client: c.id,
|
||||
})
|
||||
|
||||
if mErr == nil {
|
||||
c.client.Publish(ctx, SyncChannelName, binMsg)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// GetRedisCache reads the redis cache and publish it to the channel
|
||||
func (c *Client) GetRedisCache(ctx context.Context) {
|
||||
c.l.Debug("GetRedisCache")
|
||||
|
||||
go func() {
|
||||
iter := c.client.Scan(ctx, 0, prefixKey("*"), 0).Iterator()
|
||||
if err := iter.Err(); err != nil {
|
||||
c.l.Error("GetRedisCache ", err)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
for iter.Next(ctx) {
|
||||
response, err := c.getResponse(ctx, iter.Val())
|
||||
if err == nil {
|
||||
if response != nil {
|
||||
if !util.CtxSend(ctx, c.CacheChannel, response) {
|
||||
return
|
||||
}
|
||||
}
|
||||
} else {
|
||||
c.l.Error("GetRedisCache ", err)
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// startup starts a new goroutine for subscription and translation
|
||||
func (c *Client) startup(ctx context.Context) error {
|
||||
ps := c.client.Subscribe(ctx, SyncChannelName)
|
||||
|
||||
_, err := ps.Receive(ctx)
|
||||
if err == nil {
|
||||
go func() {
|
||||
for {
|
||||
select {
|
||||
// received message from subscription
|
||||
case msg := <-ps.Channel():
|
||||
c.l.Debug("Received message: ", msg)
|
||||
|
||||
if msg != nil && len(msg.Payload) > 0 {
|
||||
// message is not empty
|
||||
c.processReceivedMessage(ctx, msg)
|
||||
}
|
||||
// publish message from buffer
|
||||
case s := <-c.sendBuffer:
|
||||
c.publishMessageFromBuffer(ctx, s)
|
||||
// context is done
|
||||
case <-ctx.Done():
|
||||
c.client.Close()
|
||||
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
} else {
|
||||
return fmt.Errorf("failed to receive from Redis subscription channel '%s': %w", SyncChannelName, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Client) publishMessageFromBuffer(ctx context.Context, s *bufferMessage) {
|
||||
origRes := s.Message
|
||||
origRes.Compress = true
|
||||
binRes, pErr := origRes.Pack()
|
||||
|
||||
if pErr == nil {
|
||||
binMsg, mErr := json.Marshal(redisMessage{
|
||||
Key: s.Key,
|
||||
Type: messageTypeCache,
|
||||
Message: binRes,
|
||||
Client: c.id,
|
||||
})
|
||||
|
||||
if mErr == nil {
|
||||
c.client.Publish(ctx, SyncChannelName, binMsg)
|
||||
}
|
||||
|
||||
c.client.Set(ctx,
|
||||
prefixKey(s.Key),
|
||||
binRes,
|
||||
c.getTTL(origRes))
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) processReceivedMessage(ctx context.Context, msg *redis.Message) {
|
||||
var rm redisMessage
|
||||
|
||||
if err := json.Unmarshal([]byte(msg.Payload), &rm); err != nil {
|
||||
c.l.Error("Processing error: ", err)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
// message was sent from a different blocky instance
|
||||
if !bytes.Equal(rm.Client, c.id) {
|
||||
switch rm.Type {
|
||||
case messageTypeCache:
|
||||
var cm *CacheMessage
|
||||
|
||||
cm, err := convertMessage(&rm, 0)
|
||||
if err != nil {
|
||||
c.l.Error("Processing CacheMessage error: ", err)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
util.CtxSend(ctx, c.CacheChannel, cm)
|
||||
case messageTypeEnable:
|
||||
var msg EnabledMessage
|
||||
|
||||
if err := json.Unmarshal(rm.Message, &msg); err != nil {
|
||||
c.l.Error("Processing EnabledMessage error: ", err)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
util.CtxSend(ctx, c.EnabledChannel, &msg)
|
||||
default:
|
||||
c.l.Warn("Unknown message type: ", rm.Type)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// getResponse returns model.Response for a key
|
||||
func (c *Client) getResponse(ctx context.Context, key string) (*CacheMessage, error) {
|
||||
resp, err := c.client.Get(ctx, key).Result()
|
||||
if err == nil {
|
||||
var ttl time.Duration
|
||||
ttl, err = c.client.TTL(ctx, key).Result()
|
||||
|
||||
if err == nil {
|
||||
var result *CacheMessage
|
||||
|
||||
result, err = convertMessage(&redisMessage{
|
||||
Key: cleanKey(key),
|
||||
Message: []byte(resp),
|
||||
}, ttl)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("conversion error: %w", err)
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
}
|
||||
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// convertMessage converts redisMessage to CacheMessage
|
||||
func convertMessage(message *redisMessage, ttl time.Duration) (*CacheMessage, error) {
|
||||
msg := dns.Msg{}
|
||||
|
||||
err := msg.Unpack(message.Message)
|
||||
if err == nil {
|
||||
if ttl > 0 {
|
||||
for _, a := range msg.Answer {
|
||||
a.Header().Ttl = uint32(ttl.Seconds())
|
||||
}
|
||||
}
|
||||
|
||||
res := &CacheMessage{
|
||||
Key: message.Key,
|
||||
Response: &model.Response{
|
||||
RType: model.ResponseTypeCACHED,
|
||||
Reason: cacheReason,
|
||||
Res: &msg,
|
||||
},
|
||||
}
|
||||
|
||||
return res, nil
|
||||
}
|
||||
|
||||
return nil, fmt.Errorf("failed to unpack DNS message from Redis for key '%s': %w", message.Key, err)
|
||||
}
|
||||
|
||||
// getTTL of dns message or return defaultCacheTime if 0
|
||||
func (c *Client) getTTL(dns *dns.Msg) time.Duration {
|
||||
ttl := uint32(math.MaxInt32)
|
||||
for _, a := range dns.Answer {
|
||||
ttl = min(ttl, a.Header().Ttl)
|
||||
}
|
||||
|
||||
if ttl == 0 {
|
||||
return defaultCacheTime
|
||||
}
|
||||
|
||||
return time.Duration(ttl) * time.Second
|
||||
}
|
||||
|
||||
// prefixKey with CacheStorePrefix
|
||||
func prefixKey(key string) string {
|
||||
return fmt.Sprintf("%s%s", CacheStorePrefix, key)
|
||||
}
|
||||
|
||||
// cleanKey trims CacheStorePrefix prefix
|
||||
func cleanKey(key string) string {
|
||||
return strings.TrimPrefix(key, CacheStorePrefix)
|
||||
return rdb, nil
|
||||
}
|
||||
|
||||
@@ -1,25 +1,18 @@
|
||||
package redis
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/0xERR0R/blocky/log"
|
||||
"github.com/go-redis/redis/v8"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
func init() {
|
||||
log.Silence()
|
||||
redis.SetLogger(NoLogs{})
|
||||
}
|
||||
|
||||
func TestRedisClient(t *testing.T) {
|
||||
RegisterFailHandler(Fail)
|
||||
RunSpecs(t, "Redis Suite")
|
||||
}
|
||||
|
||||
type NoLogs struct{}
|
||||
|
||||
func (l NoLogs) Printf(context.Context, string, ...any) {}
|
||||
|
||||
+41
-314
@@ -2,31 +2,16 @@ package redis
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"time"
|
||||
|
||||
"github.com/0xERR0R/blocky/config"
|
||||
"github.com/0xERR0R/blocky/util"
|
||||
"github.com/alicebob/miniredis/v2"
|
||||
"github.com/creasty/defaults"
|
||||
"github.com/google/uuid"
|
||||
"github.com/miekg/dns"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
)
|
||||
|
||||
const (
|
||||
exampleComKey = CacheStorePrefix + "example.com"
|
||||
)
|
||||
|
||||
var _ = Describe("Redis client", func() {
|
||||
var (
|
||||
redisConfig *config.Redis
|
||||
|
||||
redisClient *Client
|
||||
|
||||
err error
|
||||
)
|
||||
var _ = Describe("Redis connection factory", func() {
|
||||
var redisConfig *config.Redis
|
||||
|
||||
BeforeEach(func() {
|
||||
var rcfg config.Redis
|
||||
@@ -34,323 +19,65 @@ var _ = Describe("Redis client", func() {
|
||||
redisConfig = &rcfg
|
||||
})
|
||||
|
||||
Describe("Client creation", func() {
|
||||
When("redis configuration has no address", func() {
|
||||
It("should return nil without error", func(ctx context.Context) {
|
||||
Expect(New(ctx, redisConfig)).Should(BeNil())
|
||||
})
|
||||
})
|
||||
|
||||
When("redis configuration has invalid address", func() {
|
||||
BeforeEach(func() {
|
||||
redisConfig.Address = "127.0.0.1:0"
|
||||
})
|
||||
|
||||
It("should fail with error", func(ctx context.Context) {
|
||||
_, err = New(ctx, redisConfig)
|
||||
Expect(err).Should(HaveOccurred())
|
||||
})
|
||||
})
|
||||
|
||||
When("sentinel is enabled without servers", func() {
|
||||
BeforeEach(func() {
|
||||
redisConfig.Address = "test"
|
||||
redisConfig.SentinelAddresses = []string{"127.0.0.1:0"}
|
||||
})
|
||||
|
||||
It("should fail with error", func(ctx context.Context) {
|
||||
_, err = New(ctx, redisConfig)
|
||||
Expect(err).Should(HaveOccurred())
|
||||
})
|
||||
})
|
||||
|
||||
When("redis configuration has invalid password", func() {
|
||||
BeforeEach(func() {
|
||||
setupRedisServer(redisConfig)
|
||||
redisConfig.Password = "wrong"
|
||||
})
|
||||
|
||||
It("should fail with error", func(ctx context.Context) {
|
||||
_, err = New(ctx, redisConfig)
|
||||
Expect(err).Should(HaveOccurred())
|
||||
})
|
||||
When("configuration has no address", func() {
|
||||
It("should return nil without error", func(ctx context.Context) {
|
||||
client, err := New(ctx, redisConfig)
|
||||
Expect(err).Should(Succeed())
|
||||
Expect(client).Should(BeNil())
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Publish message", func() {
|
||||
var redisServer *miniredis.Miniredis
|
||||
When("configuration has invalid address", func() {
|
||||
BeforeEach(func() {
|
||||
redisServer = setupRedisServer(redisConfig)
|
||||
redisConfig.Address = "127.0.0.1:0"
|
||||
})
|
||||
|
||||
When("Redis client publishes 'cache' message", func() {
|
||||
It("One new entry with TTL > 0 should be persisted in the database", func(ctx context.Context) {
|
||||
redisClient, err = New(ctx, redisConfig)
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
By("Database is empty", func() {
|
||||
Eventually(func() []string {
|
||||
return redisServer.DB(redisConfig.Database).Keys()
|
||||
}).Should(BeEmpty())
|
||||
})
|
||||
|
||||
By("publish new message with TTL > 0", func() {
|
||||
res, err := util.NewMsgWithAnswer("example.com.", 123, dns.Type(dns.TypeA), "123.124.122.123")
|
||||
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
redisClient.PublishCache("example.com", res)
|
||||
})
|
||||
|
||||
By("Database has one entry with correct TTL", func() {
|
||||
Eventually(func() bool {
|
||||
return redisServer.DB(redisConfig.Database).Exists(exampleComKey)
|
||||
}).Should(BeTrue())
|
||||
|
||||
ttl := redisServer.DB(redisConfig.Database).TTL(exampleComKey)
|
||||
Expect(ttl.Seconds()).Should(BeNumerically("~", 123))
|
||||
})
|
||||
})
|
||||
|
||||
It("One new entry with default TTL should be persisted in the database", func(ctx context.Context) {
|
||||
redisClient, err = New(ctx, redisConfig)
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
By("Database is empty", func() {
|
||||
Eventually(func() []string {
|
||||
return redisServer.DB(redisConfig.Database).Keys()
|
||||
}).Should(BeEmpty())
|
||||
})
|
||||
|
||||
By("publish new message with TTL = 0", func() {
|
||||
res, err := util.NewMsgWithAnswer("example.com.", 0, dns.Type(dns.TypeA), "123.124.122.123")
|
||||
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
redisClient.PublishCache("example.com", res)
|
||||
})
|
||||
|
||||
By("Database has one entry with default TTL", func() {
|
||||
Eventually(func() bool {
|
||||
return redisServer.DB(redisConfig.Database).Exists(exampleComKey)
|
||||
}).Should(BeTrue())
|
||||
|
||||
ttl := redisServer.DB(redisConfig.Database).TTL(exampleComKey)
|
||||
Expect(ttl.Seconds()).Should(BeNumerically("~", defaultCacheTime.Seconds()))
|
||||
})
|
||||
})
|
||||
})
|
||||
When("Redis client publishes 'enabled' message", func() {
|
||||
It("should propagate the message over redis", func(ctx context.Context) {
|
||||
redisClient, err = New(ctx, redisConfig)
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
redisClient.PublishEnabled(ctx, &EnabledMessage{
|
||||
State: true,
|
||||
})
|
||||
Eventually(func() map[string]int {
|
||||
return redisServer.PubSubNumSub(SyncChannelName)
|
||||
}).Should(HaveLen(1))
|
||||
}, SpecTimeout(time.Second*3))
|
||||
It("should fail with error", func(ctx context.Context) {
|
||||
_, err := New(ctx, redisConfig)
|
||||
Expect(err).Should(HaveOccurred())
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Receive message", func() {
|
||||
var redisServer *miniredis.Miniredis
|
||||
When("sentinel is enabled without servers", func() {
|
||||
BeforeEach(func() {
|
||||
redisServer = setupRedisServer(redisConfig)
|
||||
redisConfig.Address = "test"
|
||||
redisConfig.SentinelAddresses = []string{"127.0.0.1:0"}
|
||||
})
|
||||
When("'enabled' message is received", func() {
|
||||
It("should propagate the message over the channel", func(ctx context.Context) {
|
||||
redisClient, err = New(ctx, redisConfig)
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
var binState []byte
|
||||
binState, err = json.Marshal(EnabledMessage{State: true})
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
var id []byte
|
||||
id, err = uuid.New().MarshalBinary()
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
var binMsg []byte
|
||||
binMsg, err = json.Marshal(redisMessage{
|
||||
Type: messageTypeEnable,
|
||||
Message: binState,
|
||||
Client: id,
|
||||
})
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
lenE := len(redisClient.EnabledChannel)
|
||||
|
||||
rec := redisServer.Publish(SyncChannelName, string(binMsg))
|
||||
Expect(rec).Should(Equal(1))
|
||||
|
||||
Eventually(func() chan *EnabledMessage {
|
||||
return redisClient.EnabledChannel
|
||||
}).Should(HaveLen(lenE + 1))
|
||||
})
|
||||
})
|
||||
When("'cache' message is received", func() {
|
||||
It("should propagate the message over the channel", func(ctx context.Context) {
|
||||
redisClient, err = New(ctx, redisConfig)
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
res, err := util.NewMsgWithAnswer("example.com.", 123, dns.Type(dns.TypeA), "123.124.122.123")
|
||||
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
var binState []byte
|
||||
binState, err = res.Pack()
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
var id []byte
|
||||
id, err = uuid.New().MarshalBinary()
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
var binMsg []byte
|
||||
binMsg, err = json.Marshal(redisMessage{
|
||||
Key: "example.com",
|
||||
Type: messageTypeCache,
|
||||
Message: binState,
|
||||
Client: id,
|
||||
})
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
lenE := len(redisClient.CacheChannel)
|
||||
|
||||
rec := redisServer.Publish(SyncChannelName, string(binMsg))
|
||||
Expect(rec).Should(Equal(1))
|
||||
|
||||
Eventually(func() chan *CacheMessage {
|
||||
return redisClient.CacheChannel
|
||||
}).Should(HaveLen(lenE + 1))
|
||||
}, SpecTimeout(time.Second*3))
|
||||
})
|
||||
When("wrong data is received", func() {
|
||||
It("should not propagate the message over the channel if data is wrong", func(ctx context.Context) {
|
||||
redisClient, err = New(ctx, redisConfig)
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
var id []byte
|
||||
id, err = uuid.New().MarshalBinary()
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
var binMsg []byte
|
||||
binMsg, err = json.Marshal(redisMessage{
|
||||
Key: "unknown",
|
||||
Type: messageTypeCache,
|
||||
Message: []byte("test"),
|
||||
Client: id,
|
||||
})
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
lenE := len(redisClient.EnabledChannel)
|
||||
lenC := len(redisClient.CacheChannel)
|
||||
|
||||
rec := redisServer.Publish(SyncChannelName, string(binMsg))
|
||||
Expect(rec).Should(Equal(1))
|
||||
|
||||
Eventually(func() chan *EnabledMessage {
|
||||
return redisClient.EnabledChannel
|
||||
}).Should(HaveLen(lenE))
|
||||
|
||||
Eventually(func() chan *CacheMessage {
|
||||
return redisClient.CacheChannel
|
||||
}).Should(HaveLen(lenC))
|
||||
}, SpecTimeout(time.Second*3))
|
||||
It("should not propagate the message over the channel if type is wrong", func(ctx context.Context) {
|
||||
redisClient, err = New(ctx, redisConfig)
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
var id []byte
|
||||
id, err = uuid.New().MarshalBinary()
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
var binMsg []byte
|
||||
binMsg, err = json.Marshal(redisMessage{
|
||||
Key: "unknown",
|
||||
Type: 99,
|
||||
Message: []byte("test"),
|
||||
Client: id,
|
||||
})
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
lenE := len(redisClient.EnabledChannel)
|
||||
lenC := len(redisClient.CacheChannel)
|
||||
|
||||
rec := redisServer.Publish(SyncChannelName, string(binMsg))
|
||||
Expect(rec).Should(Equal(1))
|
||||
|
||||
// Use Consistently to verify channels don't receive invalid messages
|
||||
Consistently(func() int {
|
||||
return len(redisClient.EnabledChannel)
|
||||
}, "200ms", "50ms").Should(Equal(lenE))
|
||||
|
||||
Consistently(func() int {
|
||||
return len(redisClient.CacheChannel)
|
||||
}, "200ms", "50ms").Should(Equal(lenC))
|
||||
}, SpecTimeout(time.Second*3))
|
||||
It("should fail with error", func(ctx context.Context) {
|
||||
_, err := New(ctx, redisConfig)
|
||||
Expect(err).Should(HaveOccurred())
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Read the redis cache and publish it to the channel", func() {
|
||||
var redisServer *miniredis.Miniredis
|
||||
When("configuration has valid address", func() {
|
||||
BeforeEach(func() {
|
||||
redisServer = setupRedisServer(redisConfig)
|
||||
redisServer, err := miniredis.Run()
|
||||
Expect(err).Should(Succeed())
|
||||
DeferCleanup(redisServer.Close)
|
||||
redisConfig.Address = redisServer.Addr()
|
||||
})
|
||||
When("GetRedisCache is called with valid database entries", func() {
|
||||
It("Should read data from Redis and propagate it via cache channel", func(ctx context.Context) {
|
||||
redisClient, err = New(ctx, redisConfig)
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
By("Database is empty", func() {
|
||||
Eventually(func() []string {
|
||||
return redisServer.DB(redisConfig.Database).Keys()
|
||||
}).Should(BeEmpty())
|
||||
})
|
||||
|
||||
By("Put valid data in Redis by publishing the cache entry", func() {
|
||||
var res *dns.Msg
|
||||
|
||||
res, err = util.NewMsgWithAnswer("example.com.", 123, dns.Type(dns.TypeA), "123.124.122.123")
|
||||
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
redisClient.PublishCache("example.com", res)
|
||||
})
|
||||
|
||||
By("Database has one entry now", func() {
|
||||
Eventually(func() []string {
|
||||
return redisServer.DB(redisConfig.Database).Keys()
|
||||
}).Should(HaveLen(1))
|
||||
})
|
||||
|
||||
By("call GetRedisCache - It should read one entry from redis and propagate it via channel", func() {
|
||||
redisClient.GetRedisCache(ctx)
|
||||
|
||||
Eventually(redisClient.CacheChannel).Should(HaveLen(1))
|
||||
})
|
||||
}, SpecTimeout(time.Second*4))
|
||||
It("should return a connected client", func(ctx context.Context) {
|
||||
client, err := New(ctx, redisConfig)
|
||||
Expect(err).Should(Succeed())
|
||||
Expect(client).ShouldNot(BeNil())
|
||||
})
|
||||
When("GetRedisCache is called and database contains not valid entry", func() {
|
||||
It("Should do nothing (only log error)", func(ctx context.Context) {
|
||||
redisClient, err = New(ctx, redisConfig)
|
||||
Expect(err).Should(Succeed())
|
||||
})
|
||||
|
||||
Expect(redisServer.DB(redisConfig.Database).Set(CacheStorePrefix+"test", "test")).Should(Succeed())
|
||||
redisClient.GetRedisCache(ctx)
|
||||
Consistently(redisClient.CacheChannel).Should(BeEmpty())
|
||||
}, SpecTimeout(time.Second*2))
|
||||
When("configuration has invalid password", func() {
|
||||
BeforeEach(func() {
|
||||
redisServer, err := miniredis.Run()
|
||||
Expect(err).Should(Succeed())
|
||||
DeferCleanup(redisServer.Close)
|
||||
redisServer.RequireAuth("correct-password")
|
||||
redisConfig.Address = redisServer.Addr()
|
||||
redisConfig.Password = "wrong"
|
||||
})
|
||||
|
||||
It("should fail with error", func(ctx context.Context) {
|
||||
_, err := New(ctx, redisConfig)
|
||||
Expect(err).Should(HaveOccurred())
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
func setupRedisServer(cfg *config.Redis) *miniredis.Miniredis {
|
||||
redisServer, err := miniredis.Run()
|
||||
Expect(err).Should(Succeed())
|
||||
DeferCleanup(redisServer.Close)
|
||||
cfg.Address = redisServer.Addr()
|
||||
|
||||
return redisServer
|
||||
}
|
||||
|
||||
@@ -24,7 +24,6 @@ import (
|
||||
"github.com/0xERR0R/blocky/lists"
|
||||
"github.com/0xERR0R/blocky/log"
|
||||
"github.com/0xERR0R/blocky/model"
|
||||
"github.com/0xERR0R/blocky/redis"
|
||||
"github.com/0xERR0R/blocky/util"
|
||||
|
||||
"github.com/miekg/dns"
|
||||
@@ -94,7 +93,6 @@ type BlockingResolver struct {
|
||||
allowlistOnlyGroups map[string]bool
|
||||
status *status
|
||||
clientGroupsBlock map[string][]string
|
||||
redisClient *redis.Client
|
||||
fqdnIPCache cache.ExpiringCache[[]net.IP]
|
||||
}
|
||||
|
||||
@@ -118,7 +116,6 @@ func clientGroupsBlock(cfg config.Blocking) map[string][]string {
|
||||
// NewBlockingResolver returns a new configured instance of the resolver
|
||||
func NewBlockingResolver(ctx context.Context,
|
||||
cfg config.Blocking,
|
||||
redis *redis.Client,
|
||||
bootstrap *Bootstrap,
|
||||
) (r *BlockingResolver, err error) {
|
||||
blockHandler, err := createBlockHandler(cfg)
|
||||
@@ -152,7 +149,6 @@ func NewBlockingResolver(ctx context.Context,
|
||||
enableTimer: time.NewTimer(0),
|
||||
},
|
||||
clientGroupsBlock: clientGroupsBlock(cfg),
|
||||
redisClient: redis,
|
||||
}
|
||||
|
||||
res.fqdnIPCache = expirationcache.NewCacheWithOnExpired[[]net.IP](ctx, expirationcache.Options{
|
||||
@@ -161,43 +157,44 @@ func NewBlockingResolver(ctx context.Context,
|
||||
return res.queryForFQIdentifierIPs(ctx, key)
|
||||
})
|
||||
|
||||
if res.redisClient != nil {
|
||||
go res.redisSubscriber(ctx)
|
||||
}
|
||||
|
||||
err = evt.Bus().SubscribeOnce(evt.ApplicationStarted, func(_ ...string) {
|
||||
go res.initFQDNIPCache(ctx)
|
||||
})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to subscribe to ApplicationStarted event: %w", err)
|
||||
if err := res.subscribeEvents(ctx); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return res, nil
|
||||
}
|
||||
|
||||
func (r *BlockingResolver) redisSubscriber(ctx context.Context) {
|
||||
ctx, logger := r.log(ctx)
|
||||
func (r *BlockingResolver) subscribeEvents(ctx context.Context) error {
|
||||
err := evt.Bus().SubscribeOnce(evt.ApplicationStarted, func(_ ...string) {
|
||||
go r.initFQDNIPCache(ctx)
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to subscribe to ApplicationStarted event: %w", err)
|
||||
}
|
||||
|
||||
for {
|
||||
select {
|
||||
case em := <-r.redisClient.EnabledChannel:
|
||||
if em != nil {
|
||||
logger.Debug("Received state from redis: ", em)
|
||||
|
||||
if em.State {
|
||||
r.internalEnableBlocking()
|
||||
} else {
|
||||
err := r.internalDisableBlocking(ctx, em.Duration, em.Groups)
|
||||
if err != nil {
|
||||
logger.Warn("Blocking couldn't be disabled:", err)
|
||||
}
|
||||
remoteHandler := func(state evt.BlockingState) {
|
||||
go func() {
|
||||
if state.Enabled {
|
||||
r.internalEnableBlocking()
|
||||
} else {
|
||||
if disableErr := r.internalDisableBlocking(ctx, state.Duration, state.Groups); disableErr != nil {
|
||||
log.PrefixedLog("blocking").Warn("blocking couldn't be disabled: ", disableErr)
|
||||
}
|
||||
}
|
||||
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
err = evt.Bus().Subscribe(evt.BlockingStateChangedRemote, remoteHandler)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to subscribe to %s: %w", evt.BlockingStateChangedRemote, err)
|
||||
}
|
||||
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
_ = evt.Bus().Unsubscribe(evt.BlockingStateChangedRemote, remoteHandler)
|
||||
}()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// RefreshLists triggers the refresh of all allow/denylists in the cache
|
||||
@@ -226,10 +223,7 @@ func (r *BlockingResolver) retrieveAllBlockingGroups() []string {
|
||||
// EnableBlocking enables the blocking against the denylists
|
||||
func (r *BlockingResolver) EnableBlocking(ctx context.Context) {
|
||||
r.internalEnableBlocking()
|
||||
|
||||
if r.redisClient != nil {
|
||||
r.redisClient.PublishEnabled(ctx, &redis.EnabledMessage{State: true})
|
||||
}
|
||||
evt.Bus().Publish(evt.BlockingStateChanged, evt.BlockingState{Enabled: true})
|
||||
}
|
||||
|
||||
func (r *BlockingResolver) internalEnableBlocking() {
|
||||
@@ -246,9 +240,9 @@ func (r *BlockingResolver) internalEnableBlocking() {
|
||||
// DisableBlocking deactivates the blocking for a particular duration (or forever if 0).
|
||||
func (r *BlockingResolver) DisableBlocking(ctx context.Context, duration time.Duration, disableGroups []string) error {
|
||||
err := r.internalDisableBlocking(ctx, duration, disableGroups)
|
||||
if err == nil && r.redisClient != nil {
|
||||
r.redisClient.PublishEnabled(ctx, &redis.EnabledMessage{
|
||||
State: false,
|
||||
if err == nil {
|
||||
evt.Bus().Publish(evt.BlockingStateChanged, evt.BlockingState{
|
||||
Enabled: false,
|
||||
Duration: duration,
|
||||
Groups: disableGroups,
|
||||
})
|
||||
|
||||
@@ -10,10 +10,7 @@ import (
|
||||
"github.com/0xERR0R/blocky/lists"
|
||||
"github.com/0xERR0R/blocky/log"
|
||||
. "github.com/0xERR0R/blocky/model"
|
||||
"github.com/0xERR0R/blocky/redis"
|
||||
"github.com/0xERR0R/blocky/util"
|
||||
"github.com/alicebob/miniredis/v2"
|
||||
"github.com/creasty/defaults"
|
||||
|
||||
"github.com/miekg/dns"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
@@ -71,7 +68,7 @@ var _ = Describe("BlockingResolver", Label("blockingResolver"), func() {
|
||||
m = &mockResolver{}
|
||||
m.On("Resolve", mock.Anything).Return(&Response{Res: mockAnswer}, nil)
|
||||
|
||||
sut, err = NewBlockingResolver(ctx, sutConfig, nil, systemResolverBootstrap)
|
||||
sut, err = NewBlockingResolver(ctx, sutConfig, systemResolverBootstrap)
|
||||
Expect(err).Should(Succeed())
|
||||
sut.Next(m)
|
||||
})
|
||||
@@ -112,7 +109,7 @@ var _ = Describe("BlockingResolver", Label("blockingResolver"), func() {
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
// recreate to trigger a reload
|
||||
sut, err = NewBlockingResolver(ctx, sutConfig, nil, systemResolverBootstrap)
|
||||
sut, err = NewBlockingResolver(ctx, sutConfig, systemResolverBootstrap)
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
Eventually(groupCnt, "1s").Should(HaveLen(2))
|
||||
@@ -1145,7 +1142,7 @@ var _ = Describe("BlockingResolver", Label("blockingResolver"), func() {
|
||||
It("should return error", func() {
|
||||
_, err := NewBlockingResolver(ctx, config.Blocking{
|
||||
BlockType: "wrong",
|
||||
}, nil, systemResolverBootstrap)
|
||||
}, systemResolverBootstrap)
|
||||
|
||||
Expect(err).Should(HaveOccurred())
|
||||
Expect(err.Error()).Should(ContainSubstring(
|
||||
@@ -1161,82 +1158,35 @@ var _ = Describe("BlockingResolver", Label("blockingResolver"), func() {
|
||||
Init: config.Init{Strategy: config.InitStrategyFailOnError},
|
||||
},
|
||||
BlockType: "zeroIp",
|
||||
}, nil, systemResolverBootstrap)
|
||||
}, systemResolverBootstrap)
|
||||
Expect(err).Should(HaveOccurred())
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Redis is configured", func() {
|
||||
var redisServer *miniredis.Miniredis
|
||||
var redisClient *redis.Client
|
||||
var err error
|
||||
JustBeforeEach(func() {
|
||||
redisServer, err = miniredis.Run()
|
||||
Describe("Remote blocking state via event bus", func() {
|
||||
When("BlockingStateChangedRemote with enabled=true is received", func() {
|
||||
It("should enable blocking", func() {
|
||||
Expect(sut.DisableBlocking(ctx, 0, []string{})).Should(Succeed())
|
||||
Expect(sut.BlockingStatus().Enabled).Should(BeFalse())
|
||||
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
var rcfg config.Redis
|
||||
err = defaults.Set(&rcfg)
|
||||
|
||||
Expect(err).Should(Succeed())
|
||||
rcfg.Address = redisServer.Addr()
|
||||
redisClient, err = redis.New(context.TODO(), &rcfg)
|
||||
|
||||
Expect(err).Should(Succeed())
|
||||
Expect(redisClient).ShouldNot(BeNil())
|
||||
sutConfig = config.Blocking{
|
||||
BlockType: "ZEROIP",
|
||||
BlockTTL: config.Duration(time.Minute),
|
||||
}
|
||||
|
||||
sut, err = NewBlockingResolver(ctx, sutConfig, redisClient, systemResolverBootstrap)
|
||||
Expect(err).Should(Succeed())
|
||||
})
|
||||
JustAfterEach(func() {
|
||||
redisServer.Close()
|
||||
})
|
||||
When("disable", func() {
|
||||
It("should return disable", func() {
|
||||
sut.EnableBlocking(context.TODO())
|
||||
|
||||
redisMockMsg := &redis.EnabledMessage{
|
||||
State: false,
|
||||
}
|
||||
redisClient.EnabledChannel <- redisMockMsg
|
||||
Bus().Publish(BlockingStateChangedRemote, BlockingState{Enabled: true})
|
||||
|
||||
Eventually(func() bool {
|
||||
return sut.BlockingStatus().Enabled
|
||||
}, "5s").Should(BeFalse())
|
||||
}).Should(BeTrue())
|
||||
})
|
||||
})
|
||||
When("disable", func() {
|
||||
It("should return disable", func() {
|
||||
sut.EnableBlocking(context.TODO())
|
||||
redisMockMsg := &redis.EnabledMessage{
|
||||
State: false,
|
||||
Groups: []string{"unknown"},
|
||||
}
|
||||
redisClient.EnabledChannel <- redisMockMsg
|
||||
When("BlockingStateChangedRemote with enabled=false is received", func() {
|
||||
It("should disable blocking", func() {
|
||||
Bus().Publish(BlockingStateChangedRemote, BlockingState{
|
||||
Enabled: false,
|
||||
Groups: []string{},
|
||||
})
|
||||
|
||||
Eventually(func() bool {
|
||||
return sut.BlockingStatus().Enabled
|
||||
}, "5s").Should(BeTrue())
|
||||
})
|
||||
})
|
||||
When("enable", func() {
|
||||
It("should return enable", func() {
|
||||
err = sut.DisableBlocking(context.TODO(), time.Hour, []string{})
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
redisMockMsg := &redis.EnabledMessage{
|
||||
State: true,
|
||||
}
|
||||
redisClient.EnabledChannel <- redisMockMsg
|
||||
|
||||
Eventually(func() bool {
|
||||
return sut.BlockingStatus().Enabled
|
||||
}, "5s").Should(BeTrue())
|
||||
}).Should(BeFalse())
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@@ -9,15 +9,14 @@ import (
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/0xERR0R/blocky/cache"
|
||||
"github.com/0xERR0R/blocky/config"
|
||||
"github.com/0xERR0R/blocky/evt"
|
||||
"github.com/0xERR0R/blocky/metrics"
|
||||
"github.com/0xERR0R/blocky/model"
|
||||
"github.com/0xERR0R/blocky/redis"
|
||||
"github.com/0xERR0R/blocky/util"
|
||||
expirationcache "github.com/0xERR0R/expiration-cache"
|
||||
|
||||
"github.com/0xERR0R/blocky/cache"
|
||||
"github.com/0xERR0R/blocky/cache/prefetching"
|
||||
"github.com/miekg/dns"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
@@ -27,6 +26,9 @@ import (
|
||||
|
||||
const defaultCachingCleanUpInterval = 5 * time.Second
|
||||
|
||||
// CacheDecorator optionally wraps the result cache (e.g., with Redis sync).
|
||||
type CacheDecorator func(cache.ExpiringCache[[]byte]) (cache.ExpiringCache[[]byte], error)
|
||||
|
||||
//nolint:gochecknoglobals
|
||||
var (
|
||||
cacheHits = promauto.With(metrics.Reg).NewCounter(
|
||||
@@ -54,38 +56,38 @@ type CachingResolver struct {
|
||||
|
||||
resultCache cache.ExpiringCache[[]byte]
|
||||
|
||||
redisClient *redis.Client
|
||||
|
||||
compiledExclusions []*regexp.Regexp
|
||||
}
|
||||
|
||||
// NewCachingResolver creates a new resolver instance
|
||||
func NewCachingResolver(ctx context.Context,
|
||||
cfg config.Caching,
|
||||
redis *redis.Client,
|
||||
decorator CacheDecorator,
|
||||
) (*CachingResolver, error) {
|
||||
return newCachingResolver(ctx, cfg, redis, true)
|
||||
return newCachingResolver(ctx, cfg, decorator, true)
|
||||
}
|
||||
|
||||
func newCachingResolver(ctx context.Context,
|
||||
cfg config.Caching,
|
||||
redis *redis.Client,
|
||||
decorator CacheDecorator,
|
||||
emitMetricEvents bool,
|
||||
) (*CachingResolver, error) {
|
||||
c := &CachingResolver{
|
||||
configurable: withConfig(&cfg),
|
||||
typed: withType("caching"),
|
||||
|
||||
redisClient: redis,
|
||||
configurable: withConfig(&cfg),
|
||||
typed: withType("caching"),
|
||||
emitMetricEvents: emitMetricEvents,
|
||||
}
|
||||
|
||||
configureCaches(ctx, c, &cfg)
|
||||
err := configureExclusions(c, &cfg)
|
||||
|
||||
if c.redisClient != nil {
|
||||
go c.redisSubscriber(ctx)
|
||||
c.redisClient.GetRedisCache(ctx)
|
||||
if decorator != nil {
|
||||
decorated, err := decorator(c.resultCache)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to apply cache decorator: %w", err)
|
||||
}
|
||||
|
||||
c.resultCache = decorated
|
||||
}
|
||||
|
||||
return c, err
|
||||
@@ -175,24 +177,6 @@ func (r *CachingResolver) reloadCacheEntry(ctx context.Context, cacheKey string)
|
||||
return &packed, r.adjustTTLs(response.Res.Answer)
|
||||
}
|
||||
|
||||
func (r *CachingResolver) redisSubscriber(ctx context.Context) {
|
||||
ctx, logger := r.log(ctx)
|
||||
|
||||
for {
|
||||
select {
|
||||
case rc := <-r.redisClient.CacheChannel:
|
||||
if rc != nil {
|
||||
logger.Debug("Received key from redis: ", rc.Key)
|
||||
ttl := r.adjustTTLs(rc.Response.Res.Answer)
|
||||
r.putInCache(ctx, rc.Key, rc.Response, ttl, false)
|
||||
}
|
||||
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// LogConfig implements `config.Configurable`.
|
||||
func (r *CachingResolver) LogConfig(logger *logrus.Entry) {
|
||||
r.cfg.LogConfig(logger)
|
||||
@@ -238,7 +222,7 @@ func (r *CachingResolver) Resolve(ctx context.Context, request *model.Request) (
|
||||
|
||||
if err == nil {
|
||||
cacheTTL := r.adjustTTLs(response.Res.Answer)
|
||||
r.putInCache(ctx, cacheKey, response, cacheTTL, true)
|
||||
r.putInCache(ctx, cacheKey, response, cacheTTL)
|
||||
} else {
|
||||
return nil, err
|
||||
}
|
||||
@@ -322,7 +306,7 @@ func isResponseCacheable(msg *dns.Msg) bool {
|
||||
}
|
||||
|
||||
func (r *CachingResolver) putInCache(
|
||||
ctx context.Context, cacheKey string, response *model.Response, ttl time.Duration, publish bool,
|
||||
ctx context.Context, cacheKey string, response *model.Response, ttl time.Duration,
|
||||
) {
|
||||
respCopy := response.Res.Copy()
|
||||
|
||||
@@ -343,11 +327,6 @@ func (r *CachingResolver) putInCache(
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if publish && r.redisClient != nil {
|
||||
res := *respCopy
|
||||
r.redisClient.PublishCache(cacheKey, &res)
|
||||
}
|
||||
}
|
||||
|
||||
// adjustTTLs calculates and returns the min TTL (considers also the min and max cache time)
|
||||
|
||||
@@ -12,9 +12,7 @@ import (
|
||||
. "github.com/0xERR0R/blocky/helpertest"
|
||||
"github.com/0xERR0R/blocky/log"
|
||||
. "github.com/0xERR0R/blocky/model"
|
||||
"github.com/0xERR0R/blocky/redis"
|
||||
"github.com/0xERR0R/blocky/util"
|
||||
"github.com/alicebob/miniredis/v2"
|
||||
"github.com/creasty/defaults"
|
||||
|
||||
"github.com/miekg/dns"
|
||||
@@ -719,80 +717,6 @@ var _ = Describe("CachingResolver", func() {
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Redis is configured", func() {
|
||||
var (
|
||||
redisServer *miniredis.Miniredis
|
||||
redisClient *redis.Client
|
||||
redisConfig *config.Redis
|
||||
err error
|
||||
)
|
||||
BeforeEach(func() {
|
||||
redisServer, err = miniredis.Run()
|
||||
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
var rcfg config.Redis
|
||||
err = defaults.Set(&rcfg)
|
||||
|
||||
Expect(err).Should(Succeed())
|
||||
|
||||
rcfg.Address = redisServer.Addr()
|
||||
redisConfig = &rcfg
|
||||
redisClient, err = redis.New(context.TODO(), redisConfig)
|
||||
|
||||
Expect(err).Should(Succeed())
|
||||
Expect(redisClient).ShouldNot(BeNil())
|
||||
})
|
||||
AfterEach(func() {
|
||||
redisServer.Close()
|
||||
})
|
||||
When("cache", func() {
|
||||
JustBeforeEach(func() {
|
||||
sutConfig = config.Caching{
|
||||
MaxCachingTime: config.Duration(time.Second * 10),
|
||||
}
|
||||
mockAnswer, _ = util.NewMsgWithAnswer("example.com.", 1000, A, "1.1.1.1")
|
||||
|
||||
sut, _ = NewCachingResolver(ctx, sutConfig, redisClient)
|
||||
m = &mockResolver{}
|
||||
m.On("Resolve", mock.Anything).Return(&Response{Res: mockAnswer}, nil)
|
||||
sut.Next(m)
|
||||
})
|
||||
|
||||
It("put in redis", func() {
|
||||
Expect(sut.Resolve(ctx, newRequest("example.com.", A))).
|
||||
Should(HaveResponseType(ResponseTypeRESOLVED))
|
||||
|
||||
Eventually(func() []string {
|
||||
return redisServer.DB(redisConfig.Database).Keys()
|
||||
}).Should(HaveLen(1))
|
||||
})
|
||||
|
||||
It("load", func() {
|
||||
request := newRequest("example2.com.", A)
|
||||
domain := util.ExtractDomain(request.Req.Question[0])
|
||||
cacheKey := util.GenerateCacheKey(A, domain)
|
||||
redisMockMsg := &redis.CacheMessage{
|
||||
Key: cacheKey,
|
||||
Response: &Response{
|
||||
RType: ResponseTypeCACHED,
|
||||
Reason: "MOCK_REDIS",
|
||||
Res: mockAnswer,
|
||||
},
|
||||
}
|
||||
redisClient.CacheChannel <- redisMockMsg
|
||||
|
||||
Eventually(sut.Resolve).
|
||||
WithContext(ctx).
|
||||
WithArguments(request).
|
||||
Should(
|
||||
SatisfyAll(
|
||||
HaveResponseType(ResponseTypeCACHED),
|
||||
HaveTTL(BeNumerically("<=", 10)),
|
||||
))
|
||||
})
|
||||
})
|
||||
})
|
||||
Context("isRequestCacheable", func() {
|
||||
var request *Request
|
||||
When("request is not cacheable", func() {
|
||||
|
||||
@@ -316,7 +316,7 @@ var _ = Describe("ParallelBestResolver", Label("parallelBestResolver"), func() {
|
||||
}
|
||||
for _, v := range resolverCount {
|
||||
// should be 500 ± 50
|
||||
Expect(v).Should(BeNumerically("~", 500, 75))
|
||||
Expect(v).Should(BeNumerically("~", 500, 100))
|
||||
}
|
||||
})
|
||||
By("perform 100 request, error upstream's weight will be reduced", func() {
|
||||
@@ -329,7 +329,7 @@ var _ = Describe("ParallelBestResolver", Label("parallelBestResolver"), func() {
|
||||
By("Resolvers without errors should be selected often", func() {
|
||||
resolverCount := make(map[*UpstreamResolver]int)
|
||||
|
||||
for range 100 {
|
||||
for range 1000 {
|
||||
resolvers := pickRandom(ctx, *sut.resolvers.Load(), parallelBestResolverCount)
|
||||
res1 := resolvers[0].resolver.(*UpstreamResolver)
|
||||
res2 := resolvers[1].resolver.(*UpstreamResolver)
|
||||
@@ -340,11 +340,11 @@ var _ = Describe("ParallelBestResolver", Label("parallelBestResolver"), func() {
|
||||
}
|
||||
for k, v := range resolverCount {
|
||||
if strings.Contains(k.String(), "wrong") {
|
||||
// error resolvers: should be 0 - 10
|
||||
Expect(v).Should(BeNumerically("~", 0, 10))
|
||||
// error resolvers: should be rare
|
||||
Expect(v).Should(BeNumerically("~", 0, 50))
|
||||
} else {
|
||||
// should be 90 ± 10
|
||||
Expect(v).Should(BeNumerically("~", 90, 10))
|
||||
// good resolvers get the vast majority of picks
|
||||
Expect(v).Should(BeNumerically("~", 950, 75))
|
||||
}
|
||||
}
|
||||
})
|
||||
@@ -516,7 +516,7 @@ var _ = Describe("ParallelBestResolver", Label("parallelBestResolver"), func() {
|
||||
By("Resolvers without errors should be selected often", func() {
|
||||
resolverCount := make(map[*UpstreamResolver]int)
|
||||
|
||||
for range 200 {
|
||||
for range 1000 {
|
||||
r := weightedRandom(ctx, *sut.resolvers.Load(), nil)
|
||||
res := r.resolver.(*UpstreamResolver)
|
||||
|
||||
@@ -524,11 +524,11 @@ var _ = Describe("ParallelBestResolver", Label("parallelBestResolver"), func() {
|
||||
}
|
||||
for k, v := range resolverCount {
|
||||
if strings.Contains(k.String(), "wrong") {
|
||||
// error resolvers: should be 0 - 10
|
||||
Expect(v).Should(BeNumerically("~", 0, 10))
|
||||
// error resolvers: should be rare
|
||||
Expect(v).Should(BeNumerically("~", 0, 25))
|
||||
} else {
|
||||
// should be 90 ± 10
|
||||
Expect(v).Should(BeNumerically("~", 95, 20))
|
||||
// good resolvers get the vast majority of picks
|
||||
Expect(v).Should(BeNumerically("~", 490, 50))
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
@@ -123,7 +123,7 @@ var _ = Describe("Resolver", func() {
|
||||
})
|
||||
When("'Name' is called", func() {
|
||||
It("should return resolver name", func() {
|
||||
br, _ := NewBlockingResolver(ctx, config.Blocking{BlockType: "zeroIP"}, nil, systemResolverBootstrap)
|
||||
br, _ := NewBlockingResolver(ctx, config.Blocking{BlockType: "zeroIP"}, systemResolverBootstrap)
|
||||
name := Name(br)
|
||||
Expect(name).Should(Equal("blocking"))
|
||||
})
|
||||
|
||||
+71
-8
@@ -5,6 +5,7 @@ import (
|
||||
"crypto/tls"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"runtime"
|
||||
@@ -12,6 +13,7 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/0xERR0R/blocky/cache"
|
||||
"github.com/0xERR0R/blocky/config"
|
||||
"github.com/0xERR0R/blocky/log"
|
||||
"github.com/0xERR0R/blocky/metrics"
|
||||
@@ -20,6 +22,7 @@ import (
|
||||
"github.com/0xERR0R/blocky/resolver"
|
||||
|
||||
"github.com/0xERR0R/blocky/util"
|
||||
goredis "github.com/go-redis/redis/v8"
|
||||
"github.com/google/uuid"
|
||||
"github.com/hashicorp/go-multierror"
|
||||
|
||||
@@ -41,6 +44,7 @@ type Server struct {
|
||||
cfg *config.Config
|
||||
|
||||
servers map[net.Listener]*httpServer
|
||||
closers []io.Closer
|
||||
}
|
||||
|
||||
func logger() *logrus.Entry {
|
||||
@@ -128,15 +132,24 @@ func NewServer(ctx context.Context, cfg *config.Config) (server *Server, err err
|
||||
return nil, fmt.Errorf("failed to create bootstrap resolver: %w", err)
|
||||
}
|
||||
|
||||
var redisClient *redis.Client
|
||||
var redisConn *goredis.Client
|
||||
if cfg.Redis.IsEnabled() {
|
||||
redisClient, err = redis.New(ctx, &cfg.Redis)
|
||||
if err != nil && cfg.Redis.Required {
|
||||
return nil, fmt.Errorf("failed to create required Redis client: %w", err)
|
||||
redisConn, err = redis.New(ctx, &cfg.Redis)
|
||||
if err != nil {
|
||||
if cfg.Redis.Required {
|
||||
return nil, fmt.Errorf("failed to create required Redis client: %w", err)
|
||||
}
|
||||
|
||||
logger().WithError(err).Warn("Redis is enabled but optional and could not be initialized, continuing without Redis")
|
||||
}
|
||||
}
|
||||
|
||||
queryResolver, queryError := createQueryResolver(ctx, cfg, bootstrap, redisClient)
|
||||
redisResult, err := createRedisCacheDecorator(ctx, redisConn, cfg.Redis.Required)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
queryResolver, queryError := createQueryResolver(ctx, cfg, bootstrap, redisResult.decorator)
|
||||
if queryError != nil {
|
||||
return nil, queryError
|
||||
}
|
||||
@@ -149,6 +162,14 @@ func NewServer(ctx context.Context, cfg *config.Config) (server *Server, err err
|
||||
servers: make(map[net.Listener]*httpServer),
|
||||
}
|
||||
|
||||
if redisResult.bridge != nil {
|
||||
server.closers = append(server.closers, redisResult.bridge)
|
||||
}
|
||||
|
||||
if redisConn != nil {
|
||||
server.closers = append(server.closers, redisConn)
|
||||
}
|
||||
|
||||
server.printConfiguration()
|
||||
|
||||
server.registerDNSHandlers(ctx)
|
||||
@@ -294,19 +315,55 @@ func createUDPServer(address string) (*dns.Server, error) {
|
||||
return createDNSServer("udp", address, nil)
|
||||
}
|
||||
|
||||
type redisBridgeResult struct {
|
||||
decorator resolver.CacheDecorator
|
||||
bridge *redis.EventBusBridge
|
||||
}
|
||||
|
||||
func createRedisCacheDecorator(
|
||||
ctx context.Context, redisConn *goredis.Client, required bool,
|
||||
) (*redisBridgeResult, error) {
|
||||
if redisConn == nil {
|
||||
return &redisBridgeResult{}, nil
|
||||
}
|
||||
|
||||
bridge, err := redis.NewEventBusBridge(ctx, redisConn)
|
||||
if err != nil {
|
||||
if required {
|
||||
return nil, fmt.Errorf("failed to create required Redis event bridge: %w", err)
|
||||
}
|
||||
|
||||
logger().Warn("failed to create Redis event bridge: ", err)
|
||||
}
|
||||
|
||||
decorator := func(inner cache.ExpiringCache[[]byte]) (cache.ExpiringCache[[]byte], error) {
|
||||
return cache.NewRedisExpiringByteCache(ctx, inner, redisConn, cache.RedisOptions[[]byte]{
|
||||
Prefix: "blocky:cache:",
|
||||
Channel: "blocky_cache_sync",
|
||||
})
|
||||
}
|
||||
|
||||
return &redisBridgeResult{decorator: decorator, bridge: bridge}, nil
|
||||
}
|
||||
|
||||
func createQueryResolver(
|
||||
ctx context.Context,
|
||||
cfg *config.Config,
|
||||
bootstrap *resolver.Bootstrap,
|
||||
redisClient *redis.Client,
|
||||
cacheDecorator resolver.CacheDecorator,
|
||||
) (resolver.ChainedResolver, error) {
|
||||
upstreamTree, utErr := resolver.NewUpstreamTreeResolver(ctx, cfg.Upstreams, bootstrap)
|
||||
blocking, blErr := resolver.NewBlockingResolver(ctx, cfg.Blocking, redisClient, bootstrap)
|
||||
blocking, blErr := resolver.NewBlockingResolver(ctx, cfg.Blocking, bootstrap)
|
||||
clientNames, cnErr := resolver.NewClientNamesResolver(ctx, cfg.ClientLookup, cfg.Upstreams, bootstrap)
|
||||
queryLogging, qlErr := resolver.NewQueryLoggingResolver(ctx, cfg.QueryLog)
|
||||
condUpstream, cuErr := resolver.NewConditionalUpstreamResolver(ctx, cfg.Conditional, cfg.Upstreams, bootstrap)
|
||||
hostsFile, hfErr := resolver.NewHostsFileResolver(ctx, cfg.HostsFile, bootstrap)
|
||||
cachingResolver, crErr := resolver.NewCachingResolver(ctx, cfg.Caching, redisClient)
|
||||
decorator := cacheDecorator
|
||||
if !cfg.Caching.IsEnabled() {
|
||||
decorator = nil
|
||||
}
|
||||
|
||||
cachingResolver, crErr := resolver.NewCachingResolver(ctx, cfg.Caching, decorator)
|
||||
// Pass upstreamTree to DNSSEC resolver so it can query for DNSKEY/DS records
|
||||
dnssecResolver, dsErr := resolver.NewDNSSECResolver(ctx, cfg.DNSSEC, upstreamTree)
|
||||
|
||||
@@ -429,6 +486,12 @@ func (s *Server) Start(ctx context.Context, errCh chan<- error) {
|
||||
func (s *Server) Stop(ctx context.Context) error {
|
||||
logger().Info("Stopping server")
|
||||
|
||||
for _, c := range s.closers {
|
||||
if err := c.Close(); err != nil {
|
||||
logger().Warn("failed to close resource: ", err)
|
||||
}
|
||||
}
|
||||
|
||||
for _, server := range s.dnsServers {
|
||||
if err := server.ShutdownContext(ctx); err != nil {
|
||||
return fmt.Errorf("stop %s listener failed: %w", server.Net, err)
|
||||
|
||||
Reference in New Issue
Block a user