【kubernetes/k8s源码分析】coredns 源码分析之四 cache 插件

一时失言乱红尘 2024-04-17 15:12 86阅读 0赞

## ![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70][] ##

## Description ##

>     With *cache* enabled, all records except zone transfers and metadata records will be cached for up to 3600s. Caching is mostly useful in a scenario when fetching data from the backend (upstream, database, etc.) is expensive.
> 
>     This plugin can only be used once per Server Block.

# #

## Syntax ##

cache \[TTL\] \[ZONES...\]

TTL 单位为秒，如果不指定 TTL ，则默认为 3600 秒

cache [TTL] [ZONES...] {
        success CAPACITY [TTL] [MINTTL]
        denial CAPACITY [TTL] [MINTTL]
        prefetch AMOUNT [[DURATION] [PERCENTAGE%]]
    }

success：覆盖缓存成功响应的设置。**CAPACITY** 代表在开始移除前可以缓存最大的数据包数量，TTL 会覆盖最大的 TTL 值，MINTTL 将会覆盖最小的 TTL 值，默认为 5，success 可以用来限制查询到后端

denial：覆盖缓存拒绝存在响应的设置。**CAPACITY** 代表了在开始移除前（LRU）可以缓存的最大最大数据包量，

prefetch：将预取即将从缓存中删除的热门条目。热门意味着数量查询之间没有间隔的 DURATION 或者没有比这更多的。**DURATION 默认为 1分钟，当 TTL 低于 PERCENTAGE（默认为10%）或 TTL 过期前的最后一秒时，将发生预取。值范围应在\[10%，90%\]**

注意： 如果 CAPACITY 未设置，则默认的 cahe 大小为 9984 ，最小的 cache 为 1024，如果设置 CAPACITY，实际的 cache size 四舍五入最接近除以 256，

## Examples ##

Enable caching for all zones, but cap everything to a TTL of 10 seconds:

. {
        cache 10
        whoami
    }

Proxy to Google Public DNS and only cache responses for example.org (or below).

. {
        forward . 8.8.8.8:53
        cache example.org
    }

Enable caching for all zones, keep a positive cache size of 5000 and a negative cache size of 2500:

. {
         cache {
             success 5000
             denial 2500
        }
     }

## 结构体 Cache ##

pcap：Success CAPACITY 值，默认设置为 10000

ncap：Denial CAPACITY 值，默认设置为 10000

pttl：默认设置为 3600，最大的 TTL

minpttl：默认设置为 5，最小的 TTL

nttl：默认设置为 1800，Denial TTL 最大值

minnttl：默认为 5，Denial TTL 最小值

>     type Cache struct {
>        Next  plugin.Handler
>        Zones []string
>     
>        ncache  *cache.Cache
>        ncap    int
>        nttl    time.Duration
>        minnttl time.Duration
>     
>        pcache  *cache.Cache
>        pcap    int
>        pttl    time.Duration
>        minpttl time.Duration
>     
>        // Prefetch.
>        prefetch   int
>        duration   time.Duration
>        percentage int
>     
>        // Testing.
>        now func() time.Time
>     }

# 1. init 初始化注册 cahce 服务类型为 dns #

func init() {
    	caddy.RegisterPlugin("cache", caddy.Plugin{
    		ServerType: "dns",
    		Action:     setup,
    	})
    }

# 2. setup 函数 #

调用 AddPlugin 函数添加插件实现了 plugin Handler 接口

func setup(c *caddy.Controller) error {
    	ca, err := cacheParse(c)
    	if err != nil {
    		return plugin.Error("cache", err)
    	}
    	dnsserver.GetConfig(c).AddPlugin(func(next plugin.Handler) plugin.Handler {
    		ca.Next = next
    		return ca
    	})
    
    	c.OnStartup(func() error {
    		metrics.MustRegister(c,
    			cacheSize, cacheHits, cacheMisses,
    			cachePrefetches, cacheDrops)
    		return nil
    	})
    
    	return nil
    }

> cache \[TTL\] \[ZONES...\] \{  
>     success CAPACITY \[TTL\] \[MINTTL\]  
>     denial CAPACITY \[TTL\] \[MINTTL\]  
>     prefetch AMOUNT \[\[DURATION\] \[PERCENTAGE%\]\]  
> \}

###     2.1 cacheParse 分析提取块数据 ###

2.1.1 如果后面参数为数值类型的话，比如 cache 30，则设置 pttl （最大的 TTL）和 nttl 值，单位为秒

if len(args) > 0 {
    	// first args may be just a number, then it is the ttl, if not it is a zone
    	ttl, err := strconv.Atoi(args[0])
    	if err == nil {
    		// Reserve 0 (and smaller for future things)
    		if ttl <= 0 {
    			return nil, fmt.Errorf("cache TTL can not be zero or negative: %d", ttl)
    		}
    		ca.pttl = time.Duration(ttl) * time.Second
    		ca.nttl = time.Duration(ttl) * time.Second
    		args = args[1:]
    	}
    	if len(args) > 0 {
    		copy(origins, args)
    	}
    }

如果有包含块 \{\}，如下，则分析 2.1.2 2.1.3 章节

2.1.2 提取 Success 设置的参数值

>      cache \[TTL\] \[ZONES...\] \{  
>     success CAPACITY \[TTL\] \[MINTTL\]

提取 CAPACITY 保存 pcap，如果有参数，则设置 pttl 最大的 TTL，设置 minpttl 最小 TTL

case Success:
    	args := c.RemainingArgs()
    	if len(args) == 0 {
    		return nil, c.ArgErr()
    	}
    	pcap, err := strconv.Atoi(args[0])
    	if err != nil {
    		return nil, err
    	}
    	ca.pcap = pcap
    	if len(args) > 1 {
    		pttl, err := strconv.Atoi(args[1])
    		if err != nil {
    			return nil, err
    		}
    		// Reserve 0 (and smaller for future things)
    		if pttl <= 0 {
    			return nil, fmt.Errorf("cache TTL can not be zero or negative: %d", pttl)
    		}
    		ca.pttl = time.Duration(pttl) * time.Second
    		if len(args) > 2 {
    			minpttl, err := strconv.Atoi(args[2])
    			if err != nil {
    				return nil, err
    			}
    			// Reserve < 0
    			if minpttl < 0 {
    				return nil, fmt.Errorf("cache min TTL can not be negative: %d", minpttl)
    			}
    			ca.minpttl = time.Duration(minpttl) * time.Second
    		}
    	}

2.1.3 提取 Denial 值

ncap 保存 Denial 的 CAPACITY，nttl 设置为最大的 TTL，minnttl 设置为最小的 TTL

case Denial:
    	args := c.RemainingArgs()
    	if len(args) == 0 {
    		return nil, c.ArgErr()
    	}
    	ncap, err := strconv.Atoi(args[0])
    	if err != nil {
    		return nil, err
    	}
    	ca.ncap = ncap
    	if len(args) > 1 {
    		nttl, err := strconv.Atoi(args[1])
    		if err != nil {
    			return nil, err
    		}
    		// Reserve 0 (and smaller for future things)
    		if nttl <= 0 {
    			return nil, fmt.Errorf("cache TTL can not be zero or negative: %d", nttl)
    		}
    		ca.nttl = time.Duration(nttl) * time.Second
    		if len(args) > 2 {
    			minnttl, err := strconv.Atoi(args[2])
    			if err != nil {
    				return nil, err
    			}
    			// Reserve < 0
    			if minnttl < 0 {
    				return nil, fmt.Errorf("cache min TTL can not be negative: %d", minnttl)
    			}
    			ca.minnttl = time.Duration(minnttl) * time.Second
    		}
    	}

2.1.4 提取 prefetch 参数值，范围为 【10% - 90%】

case "prefetch":
    	args := c.RemainingArgs()
    	if len(args) == 0 || len(args) > 3 {
    		return nil, c.ArgErr()
    	}
    	amount, err := strconv.Atoi(args[0])
    	if err != nil {
    		return nil, err
    	}
    	if amount < 0 {
    		return nil, fmt.Errorf("prefetch amount should be positive: %d", amount)
    	}
    	ca.prefetch = amount
    
    	if len(args) > 1 {
    		dur, err := time.ParseDuration(args[1])
    		if err != nil {
    			return nil, err
    		}
    		ca.duration = dur
    	}
    	if len(args) > 2 {
    		pct := args[2]
    		if x := pct[len(pct)-1]; x != '%' {
    			return nil, fmt.Errorf("last character of percentage should be `%%`, but is: %q", x)
    		}
    		pct = pct[:len(pct)-1]
    
    		num, err := strconv.Atoi(pct)
    		if err != nil {
    			return nil, err
    		}
    		if num < 10 || num > 90 {
    			return nil, fmt.Errorf("percentage should fall in range [10, 90]: %d", num)
    		}
    		ca.percentage = num
    	}

2.1.5 New Cache 函数

分为 256个 shard，每个 shard 包括 size 和 items

// New returns a new cache.
    func New(size int) *Cache {
    	ssize := size / shardSize
    	if ssize < 4 {
    		ssize = 4
    	}
    
    	c := &Cache{}
    
    	// Initialize all the shards
    	for i := 0; i < shardSize; i++ {
    		c.shards[i] = newShard(ssize)
    	}
    	return c
    }

# 3. ServeDNS 函数 #

路径 plugin/cache/handler.go

// ServeDNS implements the plugin.Handler interface.
    func (c *Cache) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg) (int, error) {
    	state := request.Request{W: w, Req: r}
    
    	zone := plugin.Zones(c.Zones).Matches(state.Name())
    	if zone == "" {
    		return plugin.NextOrFailure(c.Name(), c.Next, ctx, w, r)
    	}

###     3.1 get 函数 ###

key 值经过 hash 值，看看落到哪个 shard 中，是根据 key & 255 值

如果命中则相应的 pcache 或者 ncache 中 hint 值增 1，如果未命中则 misses 值增加 1

func (c *Cache) get(now time.Time, state request.Request, server string) (*item, bool) {
    	k := hash(state.Name(), state.QType(), state.Do())
    
    	if i, ok := c.ncache.Get(k); ok && i.(*item).ttl(now) > 0 {
    		cacheHits.WithLabelValues(server, Denial).Inc()
    		return i.(*item), true
    	}
    
    	if i, ok := c.pcache.Get(k); ok && i.(*item).ttl(now) > 0 {
    		cacheHits.WithLabelValues(server, Success).Inc()
    		return i.(*item), true
    	}
    	cacheMisses.WithLabelValues(server).Inc()
    	return nil, false
    }

###     3.2 拷贝到 Msg 结构中 ###

// toMsg turns i into a message, it tailors the reply to m.
    // The Authoritative bit is always set to 0, because the answer is from the cache.
    func (i *item) toMsg(m *dns.Msg, now time.Time) *dns.Msg {
    	m1 := new(dns.Msg)
    	m1.SetReply(m)
    
    	m1.Authoritative = false
    	m1.AuthenticatedData = i.AuthenticatedData
    	m1.RecursionAvailable = i.RecursionAvailable
    	m1.Rcode = i.Rcode
    
    	m1.Answer = make([]dns.RR, len(i.Answer))
    	m1.Ns = make([]dns.RR, len(i.Ns))
    	m1.Extra = make([]dns.RR, len(i.Extra))
    
    	ttl := uint32(i.ttl(now))
    	for j, r := range i.Answer {
    		m1.Answer[j] = dns.Copy(r)
    		m1.Answer[j].Header().Ttl = ttl
    	}
    	for j, r := range i.Ns {
    		m1.Ns[j] = dns.Copy(r)
    		m1.Ns[j].Header().Ttl = ttl
    	}
    	// newItem skips OPT records, so we can just use i.Extra as is.
    	for j, r := range i.Extra {
    		m1.Extra[j] = dns.Copy(r)
    		m1.Extra[j].Header().Ttl = ttl
    	}
    	return m1
    }

# 4. WriteMsg 函数 #

> ; <<>> DiG 9.9.4-RedHat-9.9.4-74.el7\_6.2 <<>> www.baidu.com  
> ;; global options: +cmd  
> ;; Got answer:  
> ;; ->>HEADER<<- opcode: QUERY, status: NOERROR, id: 60775  
> ;; flags: qr rd ra; QUERY: 1, ANSWER: 3, AUTHORITY: 0, ADDITIONAL: 1
> 
> ;; OPT PSEUDOSECTION:  
> ; EDNS: version: 0, flags:; udp: 4096  
> ;; QUESTION SECTION:  
> ;www.baidu.com.            IN    A
> 
> ;; ANSWER SECTION:  
> www.baidu.com.        60    IN    CNAME    www.a.shifen.com.  
> www.a.shifen.com.    60    IN    A    61.135.169.121  
> www.a.shifen.com.    60    IN    A    61.135.169.125
> 
> ;; Query time: 16 msec  
> ;; SERVER: 10.200.254.254\#53(10.200.254.254)  
> ;; WHEN: Wed Aug 28 11:27:24 UTC 2019  
> ;; MSG SIZE  rcvd: 149

";; opcode: QUERY, status: NOERROR, id: 63258\\n;; flags: qr rd ra; QUERY: 1, ANSWER: 3, AUTHORITY: 0, ADDITIONAL: 0\\n\\n;; QUESTION SECTION:\\n;www.baidu.com.\\tIN\\t A\\n\\n;; ANSWER SECTION:\\nwww.baidu.com.\\t340\\tIN\\tCNAME\\twww.a.shifen.com.\\nwww.a.shifen.com.\\t45\\tIN\\tA\\t61.135.169.121\\nwww.a.shifen.com.\\t45\\tIN\\tA\\t61.135.169.125\\n

路径 plugin/cache/cache.go，设置 TTL，具体设置多大，逻辑看代码吧

// WriteMsg implements the dns.ResponseWriter interface.
    func (w *ResponseWriter) WriteMsg(res *dns.Msg) error {
    	do := false
    	mt, opt := response.Typify(res, w.now().UTC())
    	if opt != nil {
    		do = opt.Do()
    	}
    
    	// key returns empty string for anything we don't want to cache.
    	hasKey, key := key(w.state.Name(), res, mt, do)
    
    	msgTTL := dnsutil.MinimalTTL(res, mt)
    	var duration time.Duration
    	if mt == response.NameError || mt == response.NoData {
    		duration = computeTTL(msgTTL, w.minnttl, w.nttl)
    	} else if mt == response.ServerError {
    		// use default ttl which is 5s
    		duration = minTTL
    	} else {
    		duration = computeTTL(msgTTL, w.minpttl, w.pttl)
    	}

###      4.1 set 函数会更新 cache  ###

如果成功的则加入到 pcache 中，失败的则加入到 ncache 中

if hasKey && duration > 0 {
    	if w.state.Match(res) {
    		w.set(res, key, mt, duration)
    		cacheSize.WithLabelValues(w.server, Success).Set(float64(w.pcache.Len()))
    		cacheSize.WithLabelValues(w.server, Denial).Set(float64(w.ncache.Len()))
    	} else {
    		// Don't log it, but increment counter
    		cacheDrops.WithLabelValues(w.server).Inc()
    	}
    }

## 总结： ##

cache 分为 shard 256个，如果 CAPACITY 为 10000，则每个 shard 可以缓存 39 个

域名根据 hash 值确定落入哪个 shard 中，如果查找命中，并且在 TTL 内则算成功

如果未命中，则通过查询成功，则加入到 cache 的 pcache中，如果失败则加入到 ncache 中

如果往缓存添加时，已经满了，则随机踢除

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pob25nbGluemhhbmc_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/04/17/73f6582fc41f4b06a77d36131bca80e4.png