kube-plugin

1.controller

1.1.资源监听

• 相比其它插件，kubernetes插件实现更复杂，是coredns的核心，负责监听不同资源的事件变更及缓存，以作为dns解析的数据来源，插件注册会调用newdnsController()初始化informer触发监听及缓存。

• // newdnsController creates a controller for CoreDNS.
  func newdnsController(ctx context.Context, kubeClient kubernetes.Interface, opts dnsControlOpts) *dnsControl {
  	...
  	// svc监听及缓存
  	dns.svcLister, dns.svcController = object.NewIndexerInformer(
  		...
  		&api.Service{},
  		cache.ResourceEventHandlerFuncs{AddFunc: dns.Add, UpdateFunc: dns.Update, DeleteFunc: dns.Delete},
  		cache.Indexers{svcNameNamespaceIndex: svcNameNamespaceIndexFunc, svcIPIndex: svcIPIndexFunc, svcExtIPIndex: svcExtIPIndexFunc},
  		...
  	)
  
      // pod监听及缓存(默认关闭)
  	if opts.initPodCache {
  		dns.podLister, dns.podController = object.NewIndexerInformer(
  			...
  			&api.Pod{},
  			cache.ResourceEventHandlerFuncs{AddFunc: dns.Add, UpdateFunc: dns.Update, DeleteFunc: dns.Delete},
  			cache.Indexers{podIPIndex: podIPIndexFunc},
  			object.DefaultProcessor(object.ToPod, nil),
              ...
  		)
  	}
  
      // eps监听及缓存(默认开启)
  	if opts.initEndpointsCache {
  		dns.epLock.Lock()
  		dns.epLister, dns.epController = object.NewIndexerInformer(
  			...
  			&discovery.EndpointSlice{},
  			cache.ResourceEventHandlerFuncs{AddFunc: dns.Add, UpdateFunc: dns.Update, DeleteFunc: dns.Delete},
  			cache.Indexers{epNameNamespaceIndex: epNameNamespaceIndexFunc, epIPIndex: epIPIndexFunc},
  			object.DefaultProcessor(object.EndpointSliceToEndpoints, dns.EndpointSliceLatencyRecorder()),
              ...
  		)
  		dns.epLock.Unlock()
  	}
  
      // namespace监听及缓存
  	dns.nsLister, dns.nsController = object.NewIndexerInformer(
  		...
  		&api.Namespace{},
  		cache.ResourceEventHandlerFuncs{},
  		cache.Indexers{},
  		object.DefaultProcessor(object.ToNamespace, nil),
          ...
  	)
  
  	return &dns
  }
  
  // Run starts the controller.
  func (dns *dnsControl) Run() {
  	go dns.svcController.Run(dns.stopCh)
  	if dns.epController != nil {
  		go func() {
  			dns.epLock.RLock()
  			dns.epController.Run(dns.stopCh)
  			dns.epLock.RUnlock()
  		}()
  	}
  	if dns.podController != nil {
  		go dns.podController.Run(dns.stopCh)
  	}
  	go dns.nsController.Run(dns.stopCh)
  	<-dns.stopCh
  }
  

• 

• 注意

  1.dnsController会监听资源事件，相应的对象基于informer indexer缓存

  2.dnsController收到请求会从各资源informer indexer获取对象，数据组装后返回给dns客户端

1.2.事件处理

• 社区其它controller会将事件拆解放入本地队列workqueue，基于worker循环调用syncHandler进行状态和配置同步，dnsController仅更新时间戳。

• func (dns *dnsControl) Add(obj interface{})               { dns.updateModified() }
  func (dns *dnsControl) Delete(obj interface{})            { dns.updateModified() }
  func (dns *dnsControl) Update(oldObj, newObj interface{}) { dns.detectChanges(oldObj, newObj) }
  
  // detectChanges detects changes in objects, and updates the modified timestamp
  func (dns *dnsControl) detectChanges(oldObj, newObj interface{}) {
  	if newObj != nil && oldObj != nil && (oldObj.(meta.Object).GetResourceVersion() == newObj.(meta.Object).GetResourceVersion()) {
  		return
  	}
  	obj := newObj
  	if obj == nil {
  		obj = oldObj
  	}
  	switch ob := obj.(type) {
  	case *object.Service:
          // 获取需要更新哪些时间戳
  		imod, emod := serviceModified(oldObj, newObj)
  		if imod {
              // 更新时间戳
  			dns.updateModified()
  		}
          // 当service含有externalIPS时,修改extModified时间戳
  		if emod {
  			dns.updateExtModifed()
  		}
  	case *object.Pod:
          // 更新时间戳
  		dns.updateModified()
  	case *object.Endpoints:
          // 只有endpoints地址变更时才更新时间戳
  		if !endpointsEquivalent(oldObj.(*object.Endpoints), newObj.(*object.Endpoints)) {
  			dns.updateModified()
  		}
  	default:
  		...
  	}
  }
  
  func (dns *dnsControl) updateModified() {
  	unix := time.Now().Unix()
  	atomic.StoreInt64(&dns.modified, unix)
  }
  
  // updateExtModified set dns.extModified to the current time.
  func (dns *dnsControl) updateExtModifed() {
  	unix := time.Now().Unix()
  	atomic.StoreInt64(&dns.extModified, unix)
  }
  

• 注意

  1.dnsController仅需要informer indexer缓存对象，以解析dns请求

  2.dnsController监听事件仅更新时间戳标识资源版本，确保dns cache及slave node感知到配置变化

2.plugin

2.1.插件注册

• 所有的插件注册均需实现Handler接口，init()阶段调用plugin.Register()将setup()回调注册到caddy，setup()回调注册的是创建回调，用于NewServer()阶段初始化plugin及关联顺序。

• func init() { 
  	// setuo回调注册到caddy
      plugin.Register(pluginName, setup) 
  }
  
  func setup(c *caddy.Controller) error {
  	// 解析corefile及初始化k8s插件对象
  	k, err := kubernetesParse(c)
  	...
  	// 初始化dnsController及构造启动、终止回调
  	onStart, onShut, err := k.InitKubeCache(context.Background())
  	...
      // 启动回调注册到caddy controller(启动dns服务前调用)
  	if onStart != nil {
  		c.OnStartup(onStart)
  	}
      // 终止回调注册到caddy controller(启动dns服务后调用)
  	if onShut != nil {
  		c.OnShutdown(onShut)
  	}
  	
      // plugin初始化回调
  	dnsserver.GetConfig(c).AddPlugin(func(next plugin.Handler) plugin.Handler {
  		k.Next = next
  		return k
  	})
  
  	// 本地非回环IP获取回调
  	c.OnStartup(func() error {
          // 当前监听非回环的IP列表
  		k.localIPs = boundIPs(c)
  		return nil
  	})
  
  	return nil
  }
  
  // InitKubeCache initializes a new Kubernetes cache.
  func (k *Kubernetes) InitKubeCache(ctx context.Context) (onStart func() error, onShut func() error, err error) {
      // 获取kubeconfig
      config, err := k.getClientConfig()
      
  	...
  	// 初始化client
  	kubeClient, err := kubernetes.NewForConfig(config)
      
  	...
      // 启用pod缓存(defaultPodMode=podModeDisabled)
  	k.opts.initPodCache = k.podMode == podModeVerified
  
  	...
      // 初始化dnsController
  	k.APIConn = newdnsController(ctx, kubeClient, k.opts)
  
      // 启用ep监听(default=true)
  	initEndpointWatch := k.opts.initEndpointsCache
  
      // 启动回调
  	onStart = func() error {
  		go func() {
             	// 默认开启ep监听
  			if initEndpointWatch {
  				// 检查支持eps
  				ok, v := k.endpointSliceSupported(kubeClient)
  				if !ok {
                      // 回退到ep监听
  					k.APIConn.(*dnsControl).WatchEndpoints(ctx)
  				}
  				// 仅支持vabeta1版本eps,监听旧版本
  				if ok && v == discoveryV1beta1.SchemeGroupVersion.String() {
  					k.APIConn.(*dnsControl).WatchEndpointSliceV1beta1(ctx)
  				}
  			}
              // 启用informer监听
  			k.APIConn.Run()
  		}()
  		...
  		for {
  			select {
              // 间隔100ms周期检查资源同步完成
  			case <-checkSyncTicker.C:
  				if k.APIConn.HasSynced() {
  					return nil
  				}
              // 间隔500ms记录日志
  			case <-logTicker.C:
  				log.Info("waiting for Kubernetes API before starting server")
              // 最长等5s
  			case <-timeoutTicker.C:
  				log.Warning("starting server with unsynced Kubernetes API")
  				return nil
  			}
  		}
  	}
  
      // 关闭回调
  	onShut = func() error {
  		return k.APIConn.Stop()
  	}
  
  	return onStart, onShut, err
  }
  

• 

• 注意

  1.kubernetes plugin会初始化controller，注册启动及终止回调以触发informer缓存

  2.controller实现serveDNS()，作为plugin一部分用于解析dns

2.2.请求分发

• coredns每个插件都要实现plugin handler接口，接口定义两个方法，Name()返回插件名，serveDNS()处理域名查询。dns解析会按序执行plugin chain中各插件的serveDNS()，基于QType类型调用不同方法解析域名。

• // ServeDNS implements the plugin.Handler interface.
  func (k Kubernetes) ServeDNS(ctx context.Context, w dns.ResponseWriter, r *dns.Msg) (int, error) {
  	state := request.Request{W: w, Req: r}
  	// 待解析域名
  	qname := state.QName()
      // 最匹配的zone
  	zone := plugin.Zones(k.Zones).Matches(qname)
  	...
  	zone = qname[len(qname)-len(zone):] // maintain case of original query
  	state.Zone = zone
  	...
  	// 根据qtype执行解析
  	switch state.QType() {
      // IPV4地址解析(dig A my-svc.my-namespace.svc.cluster.local)
  	case dns.TypeA:
  		records, truncated, err = plugin.A(ctx, &k, zone, state, nil, plugin.Options{})
      // IPV6地址解析(dig AAAA my-svc.my-namespace.svc.cluster.local)
  	case dns.TypeAAAA:
  		records, truncated, err = plugin.AAAA(ctx, &k, zone, state, nil, plugin.Options{})
      // TXT记录查询(dig TXT my-svc.my-namespace.svc.cluster.local)
  	case dns.TypeTXT:
  		records, truncated, err = plugin.TXT(ctx, &k, zone, state, nil, plugin.Options{})
      // 别名解析(dig CNAME my-svc.my-namespace.svc.cluster.local)
  	case dns.TypeCNAME:
  		records, err = plugin.CNAME(ctx, &k, zone, state, plugin.Options{})
      // 反向解析(dig -x 10.96.0.1)
  	case dns.TypePTR:
  		records, err = plugin.PTR(ctx, &k, zone, state, plugin.Options{})
      // MX记录查询,即邮件交换记录(dig MX my-svc.my-namespace.svc.cluster.local)
  	case dns.TypeMX:
  		records, extra, err = plugin.MX(ctx, &k, zone, state, plugin.Options{})
      // 服务发现(dig SRV _http._tcp.my-svc.my-namespace.svc.cluster.local)
  	case dns.TypeSRV:
  		records, extra, err = plugin.SRV(ctx, &k, zone, state, plugin.Options{})
      // 区域授权记录(dig SOA cluster.local)
  	case dns.TypeSOA:
  		if qname == zone {
  			records, err = plugin.SOA(ctx, &k, zone, state, plugin.Options{})
  		}
      // 区域传送,复制用(dig AXFR cluster.local @<dns-server-ip>)
  	case dns.TypeAXFR, dns.TypeIXFR:
  		return dns.RcodeRefused, nil
      // 区域NameServer查询(dig NS cluster.local-->ns.dns.cluster.local-->dns svc入口IP)
  	case dns.TypeNS:
  		// 域名匹配zone才返回ns
  		if state.Name() == zone {
  			records, extra, err = plugin.NS(ctx, &k, zone, state, plugin.Options{})
  			break
  		}
  		fallthrough
      // 默认地址解析
  	default:
  		// Do a fake A lookup, so we can distinguish between NODATA and NXDOMAIN
  		fake := state.NewWithQuestion(state.QName(), dns.TypeA)
  		fake.Zone = state.Zone
  		_, _, err = plugin.A(ctx, &k, zone, fake, nil, plugin.Options{})
  	}
  
      // 查询错误
  	if k.IsNameError(err) {
          // 配置fallthrough,转到next插件
  		if k.Fall.Through(state.Name()) {
  			return plugin.NextOrFailure(k.Name(), k.Next, ctx, w, r)
  		}
          ...
          // nxdomain响应
  		return plugin.BackendError(ctx, &k, zone, dns.RcodeNameError, state, nil /* err */, plugin.Options{})
  	}
  	...
  	// 未查到记录,nodata响应
  	if len(records) == 0 {
  		return plugin.BackendError(ctx, &k, zone, dns.RcodeSuccess, state, nil, plugin.Options{})
  	}
  
  	...
      // 响应解析结果
  	w.WriteMsg(m)
  	return dns.RcodeSuccess, nil
  }
  

• 

• 注意

  kubernetes插件解析请求时，会根据qtype进行分发处理，解析成功会响应给dns客户端，解析失败则根据fallthrough分发给next插件或响应错误信息

2.3.DNS解析

• dns解析过程就是根据qtype进行任务分发，执行各类解析方法进行服务查询及结果响应，主要针对集群内的请求，以解析域名至pod/svc/cname。

• // returns A records from Backend or an error.
  func A/AAAA/TXT(...) (records []dns.RR, truncated bool, err error) {
      // 获取匹配的服务列表
  	services, err := checkForApex(ctx, b, zone, state, opt)
  	...
  	// 遍历服务
  	for _, serv := range services {
          // 根据IP解析类型(CNAME/TXT/A/AAAA)
  		what, ip := serv.HostType()
  	
  		switch what {
          // CNAME解析
  		case dns.TypeCNAME:
              // 检查别名循环(x-->x)
  			if Name(state.Name()).Matches(dns.Fqdn(serv.Host)) {
  				continue
  			}
  			// 构造CNAME记录
  			newRecord := serv.NewCNAME(state.QName(), serv.Host)
              // 限制递归7层
  			if len(previousRecords) > 7 {
  				// don't add it, and just continue
  				continue
  			}
              // 重复检查,避免CNAME重复添加
  			if dnsutil.DuplicateCNAME(newRecord, previousRecords) {
  				continue
  			}
              // CNAME指向的域名属于当前zone子域
  			if dns.IsSubDomain(zone, dns.Fqdn(serv.Host)) {
  				state1 := state.NewWithQuestion(serv.Host, state.QType())
  				state1.Zone = zone
                  // 递归调用A()/AAAA()/TXT()解析
  				nextRecords, tc, err := A/AAAA/TXT(ctx,b,zone,state1,append(previousRecords,newRecord),opt)
  				...
                  // 记录解析结果
                  records = append(records, newRecord)
                  records = append(records, nextRecords...)
  				...
  				continue
  			}
  			// 目标域名不在当前zone/递归解析失败
  			target := newRecord.Target
  			// 查询目标域名对应记录
  			m1, e1 := b.Lookup(ctx, state, target, state.QType())
  			...
  			// 记录解析结果
  			records = append(records, newRecord)
  			records = append(records, m1.Answer...)
  			continue
  		// IPV4解析
  		case dns.TypeA:
  			if _, ok := dup[serv.Host]; !ok {
                  // 标记srv已处理
  				dup[serv.Host] = struct{}{}
                  // 追加A记录
  				records = append(records, serv.NewA(state.QName(), ip))
  			}
  		// IPV6
  		case dns.TypeAAAA:
  			if _, ok := dup[serv.Host]; !ok {
                  // 标记srv已处理
  				dup[serv.Host] = struct{}{}
                  // 追加AAAA记录
  				records = append(records, serv.NewAAAA(state.QName(), ip))
  			}
          // TXT解析
          case dns.TypeTXT:
  			if _, ok := dup[serv.Text]; !ok {
                  // 标记srv已处理
  				dup[serv.Text] = struct{}{}
                  // 构造默认TXT记录
  				records = append(records, serv.NewTXT(state.QName()))
  			}
  		}
  	}
  	return records, truncated, nil
  }
  
  // CNAME returns CNAME records from the backend or an error.
  func CNAME(...) (records []dns.RR, err error) {
      // 获取服务列表
  	services, err := b.Services(ctx, state, true, opt)
  	...
  	serv := services[0]
  	ip := net.ParseIP(serv.Host)
      ...
      // 解析为指向域名
  	records = append(records, serv.NewCNAME(state.QName(), serv.Host))
  	return records, nil
  }
  
  // PTR returns the PTR records from the backend, only services that have a domain name as host are included.
  func PTR(...) (records []dns.RR, err error) {
      // 获取反向服务列表
  	services, err := b.Reverse(ctx, state, true, opt)
  	...
      // 遍历服务
  	for _, serv := range services {
  		...
          // 去重
  		if _, ok := dup[serv.Host]; !ok {
              // 标记已处理
  			dup[serv.Host] = struct{}{}
              // 生成反向解析记录
  			records = append(records, serv.NewPTR(state.QName(), serv.Host))
  		}
  	}
  	return records, nil
  }
  
  // SRV returns SRV records from the Backend.
  // If the Target is not a name but an IP address, a name is created on the fly.
  func SRV(...) (records, extra []dns.RR, err error) {
      // 获取服务列表
  	services, err := b.Services(ctx, state, false, opt)
  	...
  
      // 生成srv记录
  	for _, serv := range services {
  		...
          // 计算srv相对权重(至少为1)
  		weight := uint16(math.Floor(w1))
  
          // serv类型
  		what, ip := serv.HostType()
  
  		switch what {
          // CNAME类型
  		case dns.TypeCNAME:
  			...
  			records = append(records, serv.NewSRV(state.QName(), weight))
  			
              ...
              // 当前域不是plugin zone子域
  			if !dns.IsSubDomain(zone, srv.Target) {
                  // 执行Lookup解析
  				m1, e1 := b.Lookup(ctx, state, srv.Target, dns.TypeA)
  				...
  				m1, e1 = b.Lookup(ctx, state, srv.Target, dns.TypeAAAA)
  				...
                  // 结果追加到extra
                  extra = append(extra, m1)
  				break
  			}
  			// 否则递归调用
  			addr, _, e1 := A(ctx, b, zone, state1, nil, opt)
  			...
              extra = append(extra, addr...)
  			// TODO(miek): AAAA as well here.
  		// 地址类型
  		case dns.TypeA, dns.TypeAAAA:
  			...
  			srv := serv.NewSRV(state.QName(), weight)
  			
              ...
              // SRV的域名指向结果
              // _http._tcp.mysvc.ns.svc.cluster.local --> mysvc.ns.svc.cluster.local
              records = append(records, srv)
  
  			...
              // 额外的IP记录
              // mysvc.ns.svc.cluster.local --> addr
              extra = append(extra, newAddress(serv, srv.Target, ip, what))
  		}
  	}
  	return records, extra, nil
  }
  
  // returns NS records from  the backend
  func NS(ctx context.Context, b ServiceBackend, zone string, state request.Request, opt Options) (records, extra []dns.RR, err error) {
  	...
      // 更新为默认入口域名
  	state.Req.Question[0].Name = dnsutil.Join("ns.dns.", zone)
      // 获取NS对应服务(boundIPs)
  	services, err := b.Services(ctx, state, false, opt)
  	...
      // 恢复为请求域名
  	state.Req.Question[0].Name = old
  	...
  	for _, serv := range services {
  		what, ip := serv.HostType()
  		switch what {
  		...
          // IPV4/IPV6地址
  		case dns.TypeA, dns.TypeAAAA:
              // 原始IP替换为域名(ns.dns.zone)
  			serv.Host = msg.Domain(serv.Key)
              // 构造ns记录
  			ns := serv.NewNS(state.QName())
              // 附加A/AAAA记录
  			extra = append(extra, newAddress(serv, ns.Ns, ip, what))
  			...
  			records = append(records, ns)
  		}
  	}
  	return records, extra, nil
  }
  

• 

• 注意

  1.不同解析类型处理逻辑类似，CNAME服务会继续递归或转到chain处理，其它类型服务会解析结果

  2.不同解析类型均依赖Services()/Reverse()获取服务列表

2.4.正向解析

• 不同的dns查询类型会调用不同的plugin.XXX()方法，本质都是调用k.Services()获取缓存的svc/pod，用于dns解析及处理。

• // Services implements the ServiceBackend interface.
  func (k *Kubernetes) Services(...) (svcs []msg.Service, err error) {
  	// 查询类型
  	switch state.QType() {
     	// TXT查询
  	case dns.TypeTXT:
  		...
          // 限制dns-version.zone域名,构造TXT文本记录
  		svc := msg.Service{Text: DNSSchemaVersion, TTL: 28800, Key: msg.Path(state.QName(), coredns)}
  		return []msg.Service{svc}, nil
  
      // NS查询
  	case dns.TypeNS:
  		// qname==zone,获取zone的ns地址
  		nss := k.nsAddrs(false, false, state.Zone)
  		var svcs []msg.Service
  		for _, ns := range nss {
              // IPV4地址
  			if ns.Header().Rrtype == dns.TypeA {
                  svcs = append(svcs, msg.Service{ns.A})
  				continue
  			}
              // IPV6地址
  			if ns.Header().Rrtype == dns.TypeAAAA {
  				svcs = append(svcs, msg.Service{ns.AAA})
  			}
  		}
  		return svcs, nil
  	}
  
      // qname==zone
  	if isDefaultNS(state.Name(), state.Zone) {
  		// 获取zone的ns地址
          nss := k.nsAddrs(false, false, state.Zone)
  		var svcs []msg.Service
  		for _, ns := range nss {
              // IPV4地址
  			if ns.Header().Rrtype == dns.TypeA {
                  svcs = append(svcs, msg.Service{ns.A})
  				continue
  			}
              // IPV6地址
  			if ns.Header().Rrtype == dns.TypeAAAA {
  				svcs = append(svcs, msg.Service{ns.AAA})
  			}
  		}
  		return svcs, nil
  	}
  
      // 查询后端匹配记录
  	s, e := k.Records(ctx, state, false)
  
  	// 不是SRV类型
  	if state.QType() != dns.TypeSRV {
  		return s, e
  	}
  
      // SRV类型查询,过滤CNAME类型记录
      // 1.SRV Target必须直接解析为A/AAAA地址
      // 2.coredns暂未实现 external svc的完整SRV支持
  	internal := []msg.Service{}
  	for _, svc := range s {
  		if t, _ := svc.HostType(); t != dns.TypeCNAME {
  			internal = append(internal, svc)
  		}
  	}
  
  	return internal, e
  }
  

• k.Records()会从域名特征检查请求是pod/svc，pod域名调用findPods查询方法，svc域名调用findService方法。

• // Records looks up services in kubernetes.
  func (k *Kubernetes) Records(ctx context.Context, state request.Request, exact bool) ([]msg.Service, error) {
      // 解析请求
      // 命名空间、podOrSvc、podName/svcName
  	r, e := parseRequest(state.Name(), state.Zone)
  	...
  	// 反向解析跳过    
  	if dnsutil.IsReverse(state.Name()) > 0 {
  		return nil, errNoItems
  	}
  
      // 检查ns准入
  	if !k.namespaceExposed(r.namespace) {
  		return nil, errNsNotExposed
  	}
  
      // pod查询
  	if r.podOrSvc == Pod {
  		pods, err := k.findPods(r, state.Zone)
  		return pods, err
  	}
  
      // svc查询
  	services, err := k.findServices(r, state.Zone)
  	return services, err
  }
  

• k.findPods基于podName解析，insecure模式不会查询informer缓存，直接根据podName解析地址，verified模式基于podName解析地址作为索引查询podLister.Indexer缓存。

• // podName必须是IP格式
  func (k *Kubernetes) findPods(r recordRequest, zone string) (pods []msg.Service, err error) {
  	...
      // 域名格式化为url[/c/reverse(zone)]
      // /c/local/cluster
  	zonePath := msg.Path(zone, coredns)
  	...
  
      // Insecure模式(基于podName解析pod)
  	if k.podMode == podModeInsecure {
  		...
  		// Insecure模式,podName必须为ip拼接
  		return []msg.Service{{Key: strings.Join([]string{zonePath, Pod, namespace, podname}, "/"), Host: ip, TTL: k.ttl}}, err
  	}
  	...
  
      // 其它模式(基于索引查询缓存pod)
  	for _, p := range k.APIConn.PodIndex(ip) {
  		// 找到pod
  		if ip == p.PodIP && match(namespace, p.Namespace) {
              // 构造记录
  			s := msg.Service{Key: strings.Join([]string{zonePath, Pod, namespace, podname}, "/"), Host: ip, TTL: k.ttl}
  			pods = append(pods, s)
  		}
  	}
  	return pods, err
  }
  

• k.findServices()基于informer缓存获取svcList，根据svc类型获取clusterIP/CNAME/eps，解析过程会匹配svc后端池及端口协议。

• // svc名称解析
  func (k *Kubernetes) findServices(r recordRequest, zone string) (services []msg.Service, err error) {
      ...
  	
      // 生成svc的name/namespce
  	idx := object.ServiceKey(r.service, r.namespace)
      // 基于索引查询svcIndexer
  	serviceList = k.APIConn.SvcIndex(idx)
      // ep查询回调
  	endpointsListFunc = func() []*object.Endpoints { return k.APIConn.EpIndex(idx) }
  
      // 域名格式化为url[/c/reserve(zone)]
  	zonePath := msg.Path(zone, coredns)
  	for _, svc := range serviceList {
          // svc未匹配
  		if !(match(r.namespace, svc.Namespace) && match(r.service, svc.Name)) {
  			continue
  		}
  
  		// 忽略无可用ep的svc&非CNAME类型&非headless
  		if k.opts.ignoreEmptyService && svc.Type != api.ServiceTypeExternalName && !svc.Headless() {
  			podsCount := 0
              // 执行ep缓存查询
  			for _, ep := range endpointsListFunc() {
                  // 统计ep后端池
  				for _, eps := range ep.Subsets {
  					podsCount += len(eps.Addresses)
  				}
  			}
  			// 无后端跳过
  			if podsCount == 0 {
  				continue
  			}
  		}
  
  		// cname格式svc,返回svc指向域名
  		if svc.Type == api.ServiceTypeExternalName {
  			s := msg.Service{Key: strings.Join([]string{zonePath, Svc, svc.Namespace, svc.Name}, "/"), Host: svc.ExternalName, TTL: k.ttl}
  			if t, _ := s.HostType(); t == dns.TypeCNAME {
  				s.Key = strings.Join([]string{zonePath, Svc, svc.Namespace, svc.Name}, "/")
  				services = append(services, s)
  			}
  			continue
  		}
  
  		// headless svc&后端池存在
  		if svc.Headless() || r.endpoint != "" {
  			// 基于索引查询缓存ep
              if endpointsList == nil {
  				endpointsList = endpointsListFunc()
  			}
  			// 遍历ep
  			for _, ep := range endpointsList {
                  // svc不匹配
  				if object.EndpointsKey(svc.Name, svc.Namespace) != ep.Index {
  					continue
  				}
  				// 遍历ep地址
  				for _, eps := range ep.Subsets {
  					for _, addr := range eps.Addresses {
  						// 查询指定ep未命中
                          // podName.mysvc.ns.svc.cluster.local
  						if r.endpoint != "" {
                              // 1.pod设置hostname
                          	// 2.plugin配置endpointNameMode=true,返回podName
                          	// 3.IPV4地址生成
                          	// 4.IPV6地址生成
  							if !match(r.endpoint, endpointHostname(addr, k.endpointNameMode)) {
  								continue
  							}
  						}
  						// 遍历端口
  						for _, p := range eps.Ports {
                              // 请求端口和协议与ep不匹配
  							if !(matchPortAndProtocol(r.port, p.Name, r.protocol, p.Protocol)) {
  								continue
  							}
                              // 记录后端池地址
  							s := msg.Service{Host: addr.IP, Port: int(p.Port), TTL: k.ttl}
  							s.Key = strings.Join([]string{zonePath, Svc, svc.Namespace, svc.Name, endpointHostname(addr, k.endpointNameMode)}, "/")
  							services = append(services, s)
  						}
  					}
  				}
  			}
  			continue
  		}
  
  		// clusterIP svc
  		for _, p := range svc.Ports {
              // 请求端口及协议未匹配
  			if !(matchPortAndProtocol(r.port, p.Name, r.protocol, string(p.Protocol))) {
  				continue
  			}
              // 记录svc clusterIP
  			for _, ip := range svc.ClusterIPs {
  				s := msg.Service{Host: ip, Port: int(p.Port), TTL: k.ttl}
                  // 生成完整url
  				s.Key = strings.Join([]string{zonePath, Svc, svc.Namespace, svc.Name}, "/")
  				services = append(services, s)
  			}
  		}
  	}
  	return services, err
  }
  

• 

• 注意

  1.mysvc.ns.svc.cluster.local域名会落在findServices()方法，根据svc类型解析为IP

  2.podName.mysvc.svc.cluster.local域名会落在findServices()方法，podName作为条件过滤指定ep解析为IP

  3.podName.ns.pod.cluster.local域名会落在findPods()方法，podName必须是IP格式生成(10-244-0-1)才能解析为IP

2.5.反向解析

• PTR反向解析会调用k.Reverse()方法，以解析IP-->domain关系，内部其实还是执行k.Services()方法查询informer缓存。

• // Reverse implements the ServiceBackend interface.
  func (k *Kubernetes) Reverse(...) ([]msg.Service, error) {
      // 反向域名解析为IP地址
  	ip := dnsutil.ExtractAddressFromReverse(state.Name())
  	if ip == "" {
          // IP为空,执行正向查询
  		_, e := k.Records(ctx, state, exact)
  		return nil, e
  	}
  	
      // 基于IP反向查询域名
  	records := k.serviceRecordForIP(ip, state.Name())
  	...
  	return records, nil
  }
  
  // serviceRecordForIP gets a service record with a cluster ip matching the ip argument
  func (k *Kubernetes) serviceRecordForIP(ip, name string) []msg.Service {
  	// 基于IP索引查询缓存svc
  	for _, service := range k.APIConn.SvcIndexReverse(ip) {
  		...
          // 构造svc domain
  		domain := strings.Join([]string{service.Name, service.Namespace, Svc, k.primaryZone()}, ".")
  		return []msg.Service{{Host: domain, TTL: k.ttl}}
  	}
  	// svc未命中,基于IP索引查询缓存ep
  	for _, ep := range k.APIConn.EpIndexReverse(ip) {
  		...
          // 比对ep地址
  		for _, eps := range ep.Subsets {
  			for _, addr := range eps.Addresses {
  				if addr.IP == ip {
                      // 构造ep域名
  					domain := strings.Join([]string{endpointHostname(addr, k.endpointNameMode), ep.Index, Svc, k.primaryZone()}, ".")
  					svcs = append(svcs, msg.Service{Host: domain, TTL: k.ttl})
  				}
  			}
  		}
  	}
  	return svcs
  }
  

• 

• 注意

  反向dns解析请求发起，IP地址会自动拼接.in-addr.arpa/.ip6.arpa后缀作为标识，反向解析阶段去掉标识，利用IP作为索引查询缓存，以生成匹配的svc/pod域名

2.6.整体流程

• coredns核心功能由插件实现，其中又以kubernetes plugin最重要，集群内dns解析主要围绕dns controller同步缓存数据展开，捕获到的tcp/udp请求又会依次送入plugin chain解析。

• --- 流程
  	1.启动时注册及初始化plugin chain
  	2.kubernetes plugin初始化dnsController监听svc/pod/ep/ns资源变化,构造索引缓存到informer
      4.dns解析过程尝试获取缓存内容,返回组装的dns数据
        4.1.域名为pod特性查询podIndexer获取addr
        4.2.查询externalName svc,返回CNAME记录
        4.3.查询的headLess svc,返回svc eps
        4.4.查询的clusterIP svc,返回svc clusterIP
  

•