GoSuda

Zvýšenie responzivity pomocou klientskeho cachovania v Redis

By snowmerak views ...

Čo je Redis?

Predpokladám, že len málokto nepozná Redis. Napriek tomu ho na úvod stručne pripomeniem a zhrniem jeho niekoľko charakteristík takto:

  • Operácie sa vykonávajú v jednom vlákne (single thread), a preto majú všetky operácie atomicitu.
  • Dáta sú uložené a spracovávané v pamäti (In-Memory), a preto sú všetky operácie rýchle.
  • Redis môže podľa nastavenia ukladať WAL, vďaka čomu možno rýchlo zálohovať aktuálny stav a obnoviť ho.
  • Podporuje viacero typov dát, ako sú Set, Hash, Bit, List a ďalšie, čo prináša vysokú produktivitu.
  • Má veľkú komunitu, vďaka čomu možno získavať a zdieľať rôzne skúsenosti, problémy a riešenia.
  • Je dlhodobo vyvíjaný a prevádzkovaný, a preto poskytuje dôveryhodnú stabilitu.

Teraz k podstate

Predstavte si

Čo ak by vaša služba mala cache, ktorá spĺňa tieto dve podmienky?

  1. Často dotazované dáta je potrebné poskytovať používateľom v aktuálnom stave, no ich aktualizácie prichádzajú nepravidelne, takže cache treba obnovovať časté.
  2. Dáta sa už neaktualizujú, ale k rovnakým cache dátam sa pristupuje a dopytuje veľmi často.

Prvý prípad môže predstavovať reálne poradie obľúbenosti produktov v internetovom obchode. Ak je poradie obľúbenosti produktov v reálnom čase uložené v sorted set, je neefektívne čítať ho z Redisu zakaždým, keď používateľ otvorí hlavnú stránku. Druhý prípad sa týka dát o výmenných kurzoch. Hoci sa kurzy vyhlasujú približne v 10-minútových intervaloch, skutočné dopyty prichádzajú veľmi často. Najmä pre kurzy won-dolár, won-jen a won-juan sa cache dotazuje mimoriadne často. V takýchto prípadoch je efektívnejšie, aby API server mal vlastnú lokálnu cache a pri zmene dát znovu načítal údaje z Redisu a obnovil ju.

Ako však túto funkcionalitu implementovať v štruktúre databáza - Redis - API server?

Nedá sa to riešiť cez Redis PubSub?

Pri používaní cache by sme sa mali prihlásiť na odber kanála, ktorý nám oznámi, či sa dáta aktualizovali!

  • Potom musíme vytvoriť logiku, ktorá pri aktualizácii odošle správu.
  • Dodatočné operácie spojené s PubSub ovplyvňujú výkon.

pubsub-write

pubsub-read

Čo ak Redis sám zistí zmenu?

Ak použijeme Keyspace Notification a dostaneme upozornenie na príkazy pre konkrétny kľúč?

  • Je potrebné vopred uložiť a zdieľať kľúče a príkazy používané na aktualizáciu, čo je zdĺhavé.
  • Napríklad pri niektorých kľúčoch je aktualizačným príkazom jednoduchý Set, pri iných LPush, prípadne RPush, SAdd a SRem. Situácia sa tak stáva zložitou.
  • To výrazne zvyšuje pravdepodobnosť komunikačných nedorozumení počas vývoja a ľudských chýb pri kódovaní.

Ak použijeme Keyevent Notification a dostaneme upozornenie na úrovni jednotlivých príkazov?

  • Je potrebný odber všetkých príkazov používaných na aktualizáciu. Prichádzajúce kľúče je následne nutné vhodne filtrovať.
  • Napríklad z množstva kľúčov prichádzajúcich pri príkaze Del môže byť pre daného klienta väčšina kľúčov bez lokálnej cache.
  • To môže viesť k zbytočnému plytvaniu zdrojmi.

Preto potrebujeme Invalidation Message!

Čo je Invalidation Message?

Invalidation Messages sú koncept poskytovaný ako súčasť Server Assisted Client-Side Cache, ktorý bol pridaný v Redis 6.0. Invalidation Message sa doručuje v nasledujúcom postupe.

  1. Predpokladajme, že ClientB už raz prečítal kľúč.
  2. ClientA kľúč znovu nastaví.
  3. Redis zmenu zistí a pošle ClientB Invalidation Message, ktorou ho informuje, aby vymazal cache.
  4. ClientB správu prijme a vykoná primeraný krok.

invalidation-message

Ako sa to používa

Základná štruktúra fungovania

Klient pripojený k Redisu spustí CLIENT TRACKING ON REDIRECT <client-id>, aby mohol prijímať invalidation messages. Klient, ktorý má tieto správy prijímať, sa potom prihlási na odber cez SUBSCRIBE __redis__:invalidate.

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"

Implementácia! Implementácia! Implementácia!

Redigo + Ristretto

Keď sa to vysvetlí iba týmto spôsobom, nie je jasné, ako to použiť v skutočnom kóde. Preto skúsime najprv jednoducho zostaviť riešenie pomocou redigo a ristretto.

Najprv nainštalujeme dve závislosti:

  • 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}

Najprv jednoducho vytvoríme RedisClient, ktorý obsahuje ristretto a 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}

Kód je o niečo zložitejší.

  • Na sledovanie (Tracking) otvárame ešte jedno pripojenie. Týmto opatrením zohľadňujeme, že PubSub by mohol rušiť ostatné operácie.
  • Zistíme ID pridaného pripojenia a nastavíme, aby sa Tracking pre pripojenie, ktoré číta dáta, presmeroval na toto pripojenie.
  • Potom sa prihlásime na odber invalidation messages.
  • Spracovanie odberu je o niečo zložitejšie. Keďže redigo nevie parsovať invalidation messages, musíme prijať odpoveď pred parsovaním a spracovať ju sami.
 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}

Metóda Get najprv vyhľadá hodnotu v ristretto a ak tam nie je, načíta ju z Redisu, ako je uvedené vyššie.

 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}

Vyššie uvedený kód slúži na testovanie. Keď ho vyskúšate, uvidíte, že sa hodnota pri každej aktualizácii dát v Redisu znovu načíta.

Toto riešenie je však príliš zložité. Najmä kvôli škálovaniu na cluster je nevyhnutné aktivovať Tracking na všetkých masteroch aj replikách.

Rueidis

Keďže programujeme v jazyku Go, máme k dispozícii najmodernejší a najvyspelejší rueidis. Napíšeme kód, ktorý používa server assisted client side cache v prostredí Redis clusteru pomocou rueidis.

Najprv nainštalujeme závislosť:

  • github.com/redis/rueidis

Potom napíšeme kód, ktorý dotazuje dáta z Redisu.

 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}

Na použitie client side cache v rueidis stačí zavolať iba DoCache. Potom sa dáta pridajú do lokálnej cache s určením, ako dlho v nej majú zostať. Ak DoCache zavoláme znova, dáta sa získajú z lokálnej cache. Invalidation messages sa samozrejme spracúvajú správne.

Prečo nie redis-go?

redis-go bohužiaľ neposkytuje server assisted client side cache v oficiálnom API. Navyše pri vytváraní PubSub otvára nové pripojenie a neexistuje API, ktoré by umožnilo priamy prístup k tomuto pripojeniu, takže nie je možné zistiť ani client ID. Preto sme usúdili, že konfigurácia pomocou redis-go nie je možná, a tento variant sme vynechali.

Sexy

Vďaka štruktúre client side cache

  • Ak ide o dáta, ktoré je možné pripraviť vopred, touto štruktúrou môžeme minimalizovať dotazy a prevádzku voči Redisu a zároveň poskytovať vždy aktuálne dáta.
  • Týmto spôsobom môžeme vytvoriť akúsi formu CQRS štruktúry, ktorá výrazne zvýši výkon čítania.

cqrs

O koľko sexy sa to stalo?

Keďže táto štruktúra je práve používaná v produkcii, zmerali sme priemerné latencie pre dve API. Prosíme o pochopenie, že výsledky môžem uviesť len veľmi všeobecne.

  1. Prvé API
    1. Pri prvom dopyte: priemerne 14,63 ms
    2. Pri ďalších dopytoch: priemerne 2,82 ms
    3. Priemerný rozdiel: 10,98 ms
  2. Druhé API
    1. Pri prvom dopyte: priemerne 14,05 ms
    2. Pri ďalších dopytoch: priemerne 1,60 ms
    3. Priemerný rozdiel: 11,57 ms

Latencia sa zlepšila až o 82 % a je to naozaj výrazné zlepšenie!

Očakávame, že sme dosiahli najmä tieto zlepšenia:

  • Vynechanie sieťovej komunikácie medzi klientom a Redisom a úspora prenosu dát
  • Zníženie počtu čítacích príkazov, ktoré musí Redis sám vykonať
    • Toto má zároveň efekt zvýšenia výkonu zápisov.
  • Minimalizácia parsovania Redis protokolu
    • Parsovanie Redis protokolu tiež nie je bezplatné. Zníženie tejto záťaže je veľká príležitosť.

Ale každé riešenie je kompromis. Kvôli tomu sme obetovali aspoň tieto dve veci:

  • Implementácia, prevádzka a údržba komponentov na správu client side cache
  • S tým súvisiace zvýšenie využitia CPU a pamäte na strane klienta

Záver

Osobne považujem túto architektonickú komponentu za uspokojivú a latencia aj záťaž API servera boli veľmi nízke. Do budúcna by som chcel, ak to bude možné, budovať architektúru práve touto štruktúrou.