GoSuda

Amélioration de la réactivité grâce à un cache côté client Redis

By snowmerak views ...

Qu'est-ce que Redis ?

Je pense que peu de personnes ne connaissent pas Redis. Cependant, pour les mentionner brièvement avant de poursuivre, on peut les résumer comme suit.

  • Les opérations sont exécutées dans un thread unique, de sorte que toutes les opérations possèdent une atomicité.
  • Les données sont stockées et traitées In-Memory, de sorte que toutes les opérations sont rapides.
  • Selon les options, Redis peut enregistrer un WAL, ce qui permet de sauvegarder et de restaurer rapidement l'état le plus récent.
  • Il prend en charge plusieurs types tels que Set, Hash, Bit et List, ce qui lui confère une grande productivité.
  • Il dispose d'une grande communauté, permettant de partager diverses expériences, problèmes et solutions.
  • Il est développé et exploité depuis longtemps, ce qui lui confère une stabilité digne de confiance.

Maintenant, au fond du sujet

Imaginez-vous ?

Que se passerait-il si le cache de votre service répondait aux deux conditions suivantes ?

  1. Les données fréquemment consultées doivent être fournies à l'utilisateur dans leur état le plus récent, mais leur mise à jour est irrégulière, ce qui impose de rafraîchir le cache fréquemment
  2. Les données ne sont pas mises à jour, mais le même élément de cache est fréquemment consulté

Le premier cas peut être illustré par le classement en temps réel des produits populaires d'une boutique en ligne. Si ce classement est stocké sous forme de sorted set, il est inefficace de le lire dans Redis à chaque fois qu'un utilisateur accède à la page principale.
Le second cas concerne les données de taux de change : même si ces taux sont publiés environ toutes les 10 minutes, les consultations réelles sont très fréquentes. Pour les couples won-dollar, won-yen et won-yuan en particulier, le cache est interrogé de manière très intensive.
Dans ces cas, il serait plus efficace que le serveur API dispose d'un cache local distinct, puis qu'il interroge de nouveau Redis pour se mettre à jour lorsque les données changent.

Comment peut-on alors mettre en œuvre ce comportement dans une architecture base de données - Redis - serveur API ??

Redis PubSub ne suffirait-il pas ?

Lors de l'utilisation d'un cache, abonnons-nous à un canal permettant de recevoir les notifications de mise à jour !

  • Il faut alors créer une logique qui envoie un message lors de chaque mise à jour.
  • Le traitement supplémentaire induit par PubSub affecte les performances.

pubsub-write

pubsub-read

Et si Redis détectait lui-même les modifications ?

Et si nous recevions des notifications de commandes pour une clé donnée grâce à Keyspace Notification ?

  • Il faut stocker à l'avance et partager les clés et les commandes utilisées pour la mise à jour, ce qui est fastidieux.
  • Par exemple, pour certaines clés, la commande de mise à jour est un simple Set, tandis que pour d'autres elle est LPush, RPush, SAdd ou SRem, ce qui rend le système plus complexe.
  • Cela augmente considérablement le risque d'erreurs de communication pendant le développement et d'erreurs humaines dans le code.

Et si nous recevions des notifications par commande grâce à Keyevent Notification ?

  • Il faut s'abonner à toutes les commandes utilisées pour la mise à jour, puis filtrer correctement les clés reçues.
  • Par exemple, parmi toutes les clés reçues via Del, le client n'a probablement pas de cache local pour certaines d'entre elles.
  • Cela peut entraîner un gaspillage inutile de ressources.

C'est donc là qu'intervient l'Invalidation Message !

Qu'est-ce qu'un Invalidation Message ?

Les Invalidation Messages sont un concept fourni dans le cadre du Server Assisted Client-Side Cache, ajouté à partir de Redis 6.0. Un Invalidation Message est transmis selon le flux suivant.

  1. On suppose que ClientB a déjà lu une fois la clé.
  2. ClientA définit à nouveau cette clé.
  3. Redis détecte la modification et publie un Invalidation Message vers ClientB pour lui indiquer de supprimer son cache.
  4. ClientB reçoit ce message et prend les mesures appropriées.

invalidation-message

Comment l'utiliser ?

Structure de fonctionnement de base

Le client connecté à Redis exécute CLIENT TRACKING ON REDIRECT <client-id> afin de recevoir les Invalidation Messages. Ensuite, le client qui doit recevoir ces messages s'abonne à SUBSCRIBE __redis__:invalidate pour les recevoir.

Default tracking

1# client 1
2> SET a 100

1# client 3
2> CLIENT ID
312
4> SUBSCRIBE __redis__:invalidate
51) "subscribe"
62) "__redis__:invalidate"
73) (integer) 1

1# client 2
2> CLIENT TRACKING ON REDIRECT 12
3> GET a # tracking

1# client 1
2> SET a 200

1# client 3
21) "message"
32) "__redis__:invalidate"
43) 1) "a"

Broadcasting tracking

1# client 3
2> CLIENT ID
312
4> SUBSCRIBE __redis__:invalidate
51) "subscribe"
62) "__redis__:invalidate"
73) (integer) 1

1# client 2
2CLIENT TRACKING ON BCAST PREFIX cache: REDIRECT 12

1# client 1
2> SET cache:name "Alice"
3> SET cache:age 26

1# client 3
21) "message"
32) "__redis__:invalidate"
43) 1) "cache:name"
51) "message"
62) "__redis__:invalidate"
73) 1) "cache:age"

Implémentation ! Implémentation ! Implémentation !

Redigo + Ristretto

Une simple description de ce type reste ambiguë quant à la manière de l'utiliser concrètement dans le code. Je vais donc d'abord construire un exemple simple avec redigo et ristretto.

Commençons par installer les deux dépendances.

  • github.com/gomodule/redigo
  • github.com/dgraph-io/ristretto
 1package main
 2
 3import (
 4	"context"
 5	"fmt"
 6	"log/slog"
 7	"time"
 8
 9	"github.com/dgraph-io/ristretto"
10	"github.com/gomodule/redigo/redis"
11)
12
13type RedisClient struct {
14	conn  redis.Conn
15	cache *ristretto.Cache[string, any]
16	addr  string
17}
18
19func NewRedisClient(addr string) (*RedisClient, error) {
20	cache, err := ristretto.NewCache(&ristretto.Config[string, any]{
21		NumCounters: 1e7,     // number of keys to track frequency of (10M).
22		MaxCost:     1 << 30, // maximum cost of cache (1GB).
23		BufferItems: 64,      // number of keys per Get buffer.
24	})
25	if err != nil {
26		return nil, fmt.Errorf("failed to generate cache: %w", err)
27	}
28
29	conn, err := redis.Dial("tcp", addr)
30	if err != nil {
31		return nil, fmt.Errorf("failed to connect to redis: %w", err)
32	}
33
34	return &RedisClient{
35		conn:  conn,
36		cache: cache,
37		addr:  addr,
38	}, nil
39}
40
41func (r *RedisClient) Close() error {
42	err := r.conn.Close()
43	if err != nil {
44		return fmt.Errorf("failed to close redis connection: %w", err)
45	}
46
47	return nil
48}

Je commence par créer un RedisClient simple qui contient ristretto et redigo.

 1func (r *RedisClient) Tracking(ctx context.Context) error {
 2	psc, err := redis.Dial("tcp", r.addr)
 3	if err != nil {
 4		return fmt.Errorf("failed to connect to redis: %w", err)
 5	}
 6
 7	clientId, err := redis.Int64(psc.Do("CLIENT", "ID"))
 8	if err != nil {
 9		return fmt.Errorf("failed to get client id: %w", err)
10	}
11	slog.Info("client id", "id", clientId)
12
13	subscriptionResult, err := redis.String(r.conn.Do("CLIENT", "TRACKING", "ON", "REDIRECT", clientId))
14	if err != nil {
15		return fmt.Errorf("failed to enable tracking: %w", err)
16	}
17	slog.Info("subscription result", "result", subscriptionResult)
18
19	if err := psc.Send("SUBSCRIBE", "__redis__:invalidate"); err != nil {
20		return fmt.Errorf("failed to subscribe: %w", err)
21	}
22	psc.Flush()
23
24	for {
25		msg, err := psc.Receive()
26		if err != nil {
27			return fmt.Errorf("failed to receive message: %w", err)
28		}
29
30		switch msg := msg.(type) {
31		case redis.Message:
32			slog.Info("received message", "channel", msg.Channel, "data", msg.Data)
33			key := string(msg.Data)
34			r.cache.Del(key)
35		case redis.Subscription:
36			slog.Info("subscription", "kind", msg.Kind, "channel", msg.Channel, "count", msg.Count)
37		case error:
38			return fmt.Errorf("error: %w", msg)
39		case []interface{}:
40			if len(msg) != 3 || string(msg[0].([]byte)) != "message" || string(msg[1].([]byte)) != "__redis__:invalidate" {
41				slog.Warn("unexpected message", "message", msg)
42				continue
43			}
44
45			contents := msg[2].([]interface{})
46			keys := make([]string, len(contents))
47			for i, key := range contents {
48				keys[i] = string(key.([]byte))
49				r.cache.Del(keys[i])
50			}
51			slog.Info("received invalidation message", "keys", keys)
52		default:
53			slog.Warn("unexpected message", "type", fmt.Sprintf("%T", msg))
54		}
55	}
56}

Le code est un peu complexe.

  • Pour effectuer le Tracking, j'ouvre une connexion supplémentaire. Cette mesure vise à éviter que PubSub ne perturbe les autres opérations.
  • J'obtiens l'identifiant de la connexion ajoutée, puis je fais en sorte que le Tracking de la connexion qui lit les données soit redirigé vers cette connexion.
  • Ensuite, je m'abonne aux Invalidation Messages.
  • Le traitement des abonnements est un peu complexe : comme redigo ne parse pas les messages d'invalidation, il faut recevoir la réponse avant parsing et la traiter manuellement.
 1func (r *RedisClient) Get(key string) (any, error) {
 2	val, found := r.cache.Get(key)
 3	if found {
 4		switch v := val.(type) {
 5		case int64:
 6			slog.Info("cache hit", "key", key)
 7			return v, nil
 8		default:
 9			slog.Warn("unexpected type", "type", fmt.Sprintf("%T", v))
10		}
11	}
12	slog.Info("cache miss", "key", key)
13
14	val, err := redis.Int64(r.conn.Do("GET", key))
15	if err != nil {
16		return nil, fmt.Errorf("failed to get key: %w", err)
17	}
18
19	r.cache.SetWithTTL(key, val, 1, 10*time.Second)
20	return val, nil
21}

La méthode Get consulte d'abord ristretto comme suit, puis récupère la donnée dans Redis si elle n'y figure pas.

 1package main
 2
 3import (
 4	"context"
 5	"log/slog"
 6	"os"
 7	"os/signal"
 8	"time"
 9)
10
11func main() {
12	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
13	defer cancel()
14
15	client, err := NewRedisClient("localhost:6379")
16	if err != nil {
17		panic(err)
18	}
19	defer client.Close()
20
21	go func() {
22		if err := client.Tracking(ctx); err != nil {
23			slog.Error("failed to track invalidation message", "error", err)
24		}
25	}()
26
27	ticker := time.NewTicker(1 * time.Second)
28	defer ticker.Stop()
29	done := ctx.Done()
30
31	for {
32		select {
33		case <-done:
34			slog.Info("shutting down")
35			return
36		case <-ticker.C:
37			v, err := client.Get("key")
38			if err != nil {
39				slog.Error("failed to get key", "error", err)
40				return
41			}
42			slog.Info("got key", "value", v)
43		}
44	}
45}

Voici le code de test. Si vous l'essayez, vous constaterez que la valeur est mise à jour chaque fois que la donnée change dans Redis.

Cependant, cette approche est trop complexe. Surtout, pour une extension au cluster, il devient indispensable d'activer le Tracking sur tous les maîtres et tous les réplicas.

Rueidis

Puisque nous utilisons le langage Go, nous disposons de rueidis, la solution la plus moderne et la plus avancée. Je vais écrire un code utilisant le server assisted client side cache dans un environnement de cluster Redis avec rueidis.

Commençons par installer la dépendance.

  • github.com/redis/rueidis

Écrivons ensuite le code qui interroge les données dans Redis.

 1package main
 2
 3import (
 4	"context"
 5	"log/slog"
 6	"os"
 7	"os/signal"
 8	"time"
 9
10	"github.com/redis/rueidis"
11)
12
13func main() {
14	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
15	defer cancel()
16
17	client, err := rueidis.NewClient(rueidis.ClientOption{
18		InitAddress: []string{"localhost:6379"},
19	})
20	if err != nil {
21		panic(err)
22	}
23
24	ticker := time.NewTicker(1 * time.Second)
25	defer ticker.Stop()
26	done := ctx.Done()
27
28	for {
29		select {
30		case <-done:
31			slog.Info("shutting down")
32			return
33		case <-ticker.C:
34			const key = "key"
35			resp := client.DoCache(ctx, client.B().Get().Key(key).Cache(), 10*time.Second)
36			if resp.Error() != nil {
37				slog.Error("failed to get key", "error", resp.Error())
38				continue
39			}
40			i, err := resp.AsInt64()
41			if err != nil {
42				slog.Error("failed to convert response to int64", "error", err)
43				continue
44			}
45			switch resp.IsCacheHit() {
46			case true:
47				slog.Info("cache hit", "key", key)
48			case false:
49				slog.Info("missed key", "key", key)
50			}
51			slog.Info("got key", "value", i)
52		}
53	}
54}

Avec rueidis, il suffit d'utiliser DoCache pour exploiter le client side cache. Vous pouvez ensuite définir la durée de conservation dans le cache local ; en appelant de nouveau DoCache, les données sont récupérées directement depuis le cache local. Bien entendu, les Invalidation Messages sont également traités correctement.

Pourquoi pas redis-go ?

Malheureusement, redis-go ne prend pas en charge le server assisted client side cache via son API officielle. De plus, lorsqu'il crée une nouvelle connexion pour PubSub, il n'offre aucune API permettant d'accéder directement à cette connexion, si bien qu'on ne peut même pas connaître son client id. J'ai donc jugé que la construction était impossible avec redis-go et je l'ai écarté.

Quelle élégance

Grâce à la structure de client side cache

  • Pour les données qui peuvent être préparées à l'avance, cette structure permet de minimiser les requêtes et le trafic vers Redis tout en fournissant toujours les données les plus récentes.
  • Elle permet également de créer une sorte de structure CQRS, ce qui peut améliorer considérablement les performances de lecture.

cqrs

Dans quelle mesure est-ce devenu plus élégant ?

Cette structure est effectivement utilisée sur le terrain ; j'ai donc mesuré la latence approximative de deux API. Veuillez excuser le caractère très abstrait de ces résultats.

  1. Première API
    1. Lors de la première consultation : moyenne de 14,63 ms
    2. Lors des consultations suivantes : moyenne de 2,82 ms
    3. Écart moyen : 10,98 ms
  2. Deuxième API
    1. Lors de la première consultation : moyenne de 14,05 ms
    2. Lors des consultations suivantes : moyenne de 1,60 ms
    3. Écart moyen : 11,57 ms

On observe une amélioration supplémentaire de la latence pouvant atteindre environ 82 % !

Nous attendions probablement les améliorations suivantes :

  • Suppression des échanges réseau entre le client et Redis, et économie de trafic
  • Diminution du nombre de commandes de lecture que Redis doit traiter
    • Cela améliore également les performances d'écriture.
  • Réduction du parsing du protocole Redis
    • Le parsing du protocole Redis n'est pas non plus gratuit. Pouvoir le réduire constitue une opportunité importante.

Cependant, toute solution implique des compromis. Pour y parvenir, nous avons sacrifié au minimum les deux éléments suivants :

  • La mise en œuvre, l'exploitation et la maintenance des composants de gestion du client side cache
  • L'augmentation de l'utilisation du CPU et de la mémoire côté client qui en résulte

Conclusion

Personnellement, j'estime qu'il s'agit d'un composant d'architecture satisfaisant, qui a par ailleurs très peu sollicité la latence et le serveur API. Je pense que, dans la mesure du possible, je continuerai à concevoir l'architecture selon cette structure.