vlambda博客
学习文章列表

2.2 nginx controller实例化

主要包含：

• nginx controller初始化
• 实例化store(本地缓存)
• 创建同步队列：此处会对队列的sync属性进行赋值

• 监听本地配置文件

#代码路径 internal/ingress/controller/nginx.go
func NewNGINXController(config *Configuration, mc metric.Collector) *NGINXController {
 eventBroadcaster := record.NewBroadcaster()
 eventBroadcaster.StartLogging(klog.Infof)
 eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{
  Interface: config.Client.CoreV1().Events(config.Namespace),
 })

 ...
 // 实例化nginx controller
 n := &NGINXController{
  isIPV6Enabled: ing_net.IsIPv6Enabled(),
  resolver:        h,
  cfg:             config,
  syncRateLimiter: flowcontrol.NewTokenBucketRateLimiter(config.SyncRateLimit, 1),

  recorder: eventBroadcaster.NewRecorder(scheme.Scheme, apiv1.EventSource{
   Component: "nginx-ingress-controller",
  }),
  stopCh:   make(chan struct{}),
  updateCh: channels.NewRingChannel(1024),
  ngxErrCh: make(chan error),
  stopLock: &sync.Mutex{},
  runningConfig: new(ingress.Configuration),
  Proxy: &TCPProxy{},
  metricCollector: mc,
  command: NewNginxCommand(),
 }
 ...
 //实例化store，实现了Informer机制和thread safe stores的封装
 n.store = store.New(
  config.Namespace,
  config.ConfigMapName,
  config.TCPConfigMapName,
  config.UDPConfigMapName,
  config.DefaultSSLCertificate,
  config.ResyncPeriod,
  config.Client,
  n.updateCh,
  pod,
  config.DisableCatchAll)

 // 初始化syncQueue，传入syncIngress函数对队列的sync属性进行赋值
 n.syncQueue = task.NewTaskQueue(n.syncIngress)


 onTemplateChange := func() {
  template, err := ngx_template.NewTemplate(nginx.TemplatePath)
  ...
   return
  }
  n.t = template
  n.syncQueue.EnqueueTask(task.GetDummyObject("template-change"))
 }

 ngxTpl, err := ngx_template.NewTemplate(nginx.TemplatePath)
 n.t = ngxTpl
 // watch nginx模板文件更新
 _, err = watch.NewFileWatcher(nginx.TemplatePath, onTemplateChange)

 return n
}

2.2.1 实例化store(本地缓存)

controller需要监听ingress、service、configmap、endpoint、secret、pod等资源的增、删、改动作。

store是对informer和k8s thread safe store的封装，New函数包含：

// 代码路径：internal/ingress/controller/store/store.go
func New(
 namespace, configmap, tcp, udp, defaultSSLCertificate string,
 resyncPeriod time.Duration,
 client clientset.Interface,
 updateCh *channels.RingChannel,
 pod *k8s.PodInfo,
 disableCatchAll bool) Storer {

 // 实例化k8sStore对象
 store := &k8sStore{
  informers:             &Informer{},
  listers:               &Lister{},
  sslStore:              NewSSLCertTracker(),
  updateCh:              updateCh,
  backendConfig:         ngx_config.NewDefault(),
  syncSecretMu:          &sync.Mutex{},
  backendConfigMu:       &sync.RWMutex{},
  secretIngressMap:      NewObjectRefMap(),
  defaultSSLCertificate: defaultSSLCertificate,
  pod:                   pod,
 }
 ...
 
 // 实例化informer factory，并且实例化ingress、secret、configmap、service、pods等资源的informer
 infFactory := informers.NewSharedInformerFactoryWithOptions(client, resyncPeriod,
  informers.WithNamespace(namespace),
  informers.WithTweakListOptions(func(*metav1.ListOptions) {}))
 
 store.listers.Ingress.Store = store.informers.Ingress.GetStore()

 store.informers.Endpoint = infFactory.Core().V1().Endpoints().Informer()
 store.listers.Endpoint.Store = store.informers.Endpoint.GetStore()

 store.informers.Secret = infFactory.Core().V1().Secrets().Informer()
 store.listers.Secret.Store = store.informers.Secret.GetStore()

 store.informers.ConfigMap = infFactory.Core().V1().ConfigMaps().Informer()
 store.listers.ConfigMap.Store = store.informers.ConfigMap.GetStore()

 store.informers.Service = infFactory.Core().V1().Services().Informer()
 store.listers.Service.Store = store.informers.Service.GetStore()

 labelSelector := labels.SelectorFromSet(store.pod.Labels)
 store.informers.Pod = cache.NewSharedIndexInformer(
  &cache.ListWatch{
   ListFunc: func(options metav1.ListOptions) (k8sruntime.Object, error) {
    options.LabelSelector = labelSelector.String()
    return client.CoreV1().Pods(store.pod.Namespace).List(context.TODO(), options)
   },
   WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
    options.LabelSelector = labelSelector.String()
    return client.CoreV1().Pods(store.pod.Namespace).Watch(context.TODO(), options)
   },
  },
  &corev1.Pod{},
  resyncPeriod,
  cache.Indexers{},
 )
 store.listers.Pod.Store = store.informers.Pod.GetStore()

 // ingDeleteHandler用于处理ingress Delete事件
 ingDeleteHandler := func(obj interface{}) {
  ing, ok := toIngress(obj)
  ...
  store.listers.IngressWithAnnotation.Delete(ing)
  key := k8s.MetaNamespaceKey(ing)
  store.secretIngressMap.Delete(key)
  updateCh.In() <- Event{
   Type: DeleteEvent,
   Obj:  obj,
  }
 }
 // ingEventHandler用于初始化ingress资源的事件回调方法
 ingEventHandler := cache.ResourceEventHandlerFuncs{
  AddFunc: func(obj interface{}) {
   ing, _ := toIngress(obj)
   ...
   // 
   store.syncIngress(ing)
   store.updateSecretIngressMap(ing)
   store.syncSecrets(ing)

   updateCh.In() <- Event{
    Type: CreateEvent,
    Obj:  obj,
   }
  },
  DeleteFunc: ingDeleteHandler,
  UpdateFunc: func(old, cur interface{}) {
   oldIng, _ := toIngress(old)
   curIng, _ := toIngress(cur)
   ...
   store.syncIngress(curIng)
   store.updateSecretIngressMap(curIng)
   store.syncSecrets(curIng)
   updateCh.In() <- Event{
    Type: UpdateEvent,
    Obj:  cur,
   }
  },
 }
 // 以下为初始化处理endpoint、secret、configmap、service、pods等资源的事件回调方法
 ...
 // 通过informer.AddEventHandler函数为各资源添加事件回调方法
 store.informers.Ingress.AddEventHandler(ingEventHandler)
 store.informers.Endpoint.AddEventHandler(epEventHandler)
 store.informers.Secret.AddEventHandler(secrEventHandler)
 store.informers.ConfigMap.AddEventHandler(cmEventHandler)
 store.informers.Service.AddEventHandler(serviceHandler)
 store.informers.Pod.AddEventHandler(podEventHandler)
 ...
 return store
}

// 代码路径: internal/ingress/controller/controller.go
func (n *NGINXController) syncIngress(interface{}) error {
 ...
    // List所有ingress资源
 ings := n.store.ListIngresses(nil)
 hosts, servers, pcfg := n.getConfiguration(ings)
 ...
}

// getConfiguration函数返回所有ingress对应的Configuration结构
func (n *NGINXController) getConfiguration(ingresses []*ingress.Ingress) (sets.String, []*ingress.Server, *ingress.Configuration) {
 upstreams, servers := n.getBackendServers(ingresses)
 ...
}

// getBackendServers函数返回ingress后端的使用的upstream和server列表
func (n *NGINXController) getBackendServers(ingresses []*ingress.Ingress) ([]*ingress.Backend, []*ingress.Server) {
 du := n.getDefaultUpstream()
 upstreams := n.createUpstreams(ingresses, du)
 servers := n.createServers(ingresses, upstreams, du)
 ...
}

// createServer函数会根据upstream结构体构造一个域名为key，server结构体为value的map
func (n *NGINXController) createServers(data []*ingress.Ingress,
 upstreams map[string]*ingress.Backend,
 du *ingress.Backend) map[string]*ingress.Server {

 servers := make(map[string]*ingress.Server, len(data))
 allAliases := make(map[string][]string, len(data))

 // 初始化nginx server和root location配置
    ...

 // 配置默认的location，alias，SSL
 for _, ing := range data {
  ingKey := k8s.MetaNamespaceKey(ing)
  anns := ing.ParsedAnnotations

  for _, rule := range ing.Spec.Rules {
   host := rule.Host
   ...
            if len(ing.Spec.TLS) == 0 {
    continue
   }
   // 验证secret包含的证书是否包含对应域名
   tlsSecretName := extractTLSSecretName(host, ing, n.store.GetLocalSSLCert)
   ...
  }
 }
 ...
}

func extractTLSSecretName(host string, ing *ingress.Ingress,
 getLocalSSLCert func(string) (*ingress.SSLCert, error)) string {

 // 当rule中的host和tls中host匹配，直接返回secret name
 lowercaseHost := toLowerCaseASCII(host)
 for _, tls := range ing.Spec.TLS {
  for _, tlsHost := range tls.Hosts {
   if toLowerCaseASCII(tlsHost) == lowercaseHost {
    return tls.SecretName
   }
  }
 }

 // 当rule中的host和tls中host不匹配，尝试去匹配整数中SAN或者CN
 for _, tls := range ing.Spec.TLS {
  secrKey := fmt.Sprintf("%v/%v", ing.Namespace, tls.SecretName)
  cert, err := getLocalSSLCert(secrKey)
  err = cert.Certificate.VerifyHostname(host)
  return tls.SecretName
 }

 return ""
}

syncIngress 经过的函数调用链如下：

n.syncIngress() ---> n.getConfiguration() ---> n.getBackendServers() ---> n.createServers() ---> extractTLSSecretName()