Kubernete Ingress Controller 如何实现配置动态发现(以 traefik 为例的源码分析)

// provider.go
// Provider defines methods of a provider.
type Provider interface {
  // Provide allows the provider to provide configurations to traefik
  // using the given configuration channel.
  Provide(configurationChan chan<- dynamic.Message, pool *safe.Pool) error
  Init() error
}

// pkg/provider/kubernetes/crd/kubernetes.go

// Init the provider.
func (p *Provider) Init() error {
  return nil
}

// pkg/provider/kubernetes/crd/kubernetes.go

func (p *Provider) Provide(configurationChan chan<- dynamic.Message, pool *safe.Pool) error {
  k8sClient, err := p.newK8sClient(ctxLog, p.LabelSelector)

  pool.Go(func(stop chan bool) {
    eventsChan, err := k8sClient.WatchAll(p.Namespaces, stopWatch)

    for {
      select {
      case event := <-eventsChan:
        conf := p.loadConfigurationFromCRD(ctxLog, k8sClient)

        confHash, err := hashstructure.Hash(conf, nil)
        switch {
        default:
          p.lastConfiguration.Set(confHash)
          configurationChan <- dynamic.Message{
            ProviderName:  providerName,
            Configuration: conf,
          }
        }
      }
    }
  })
  return nil
}

// pkg/provider/kubernetes/crd/kubernetes.go

func (p *Provider) newK8sClient(ctx context.Context, labelSelector string) (*clientWrapper, error) {
  var client *clientWrapper
  client, err = newInClusterClient(p.Endpoint)

  return client, err
}

//--------------------------------------
// pkg/provider/kubernetes/crd/client.go
func newClientImpl(csKube *kubernetes.Clientset, csCrd *versioned.Clientset) *clientWrapper {
  return &clientWrapper{
    csCrd:         csCrd,
    csKube:        csKube,
    factoriesCrd:  make(map[string]externalversions.SharedInformerFactory),
    factoriesKube: make(map[string]informers.SharedInformerFactory),
  }
}

func createClientFromConfig(c *rest.Config) (*clientWrapper, error) {
  csCrd, err := versioned.NewForConfig(c)
  csKube, err := kubernetes.NewForConfig(c)

  return newClientImpl(csKube, csCrd), nil
}

func newInClusterClient(endpoint string) (*clientWrapper, error) {
  config, err := rest.InClusterConfig()

  return createClientFromConfig(config)
}

// pkg/provider/kubernetes/crd/client.go

func (c *clientWrapper) WatchAll(namespaces []string, stopCh <-chan struct{}) (<-chan interface{}, error) {
  // 定义一个 chan
  eventCh := make(chan interface{}, 1)
  // 将 chan 包装成一个 handle 方法
  eventHandler := c.newResourceEventHandler(eventCh)

  // 进行消息订阅
  for _, ns := range namespaces {
    factoryCrd := externalversions.NewSharedInformerFactoryWithOptions(c.csCrd, resyncPeriod, externalversions.WithNamespace(ns))
    factoryCrd.Traefik().V1alpha1().IngressRoutes().Informer().AddEventHandler(eventHandler)
    factoryCrd.Traefik().V1alpha1().Middlewares().Informer().AddEventHandler(eventHandler)
    factoryCrd.Traefik().V1alpha1().IngressRouteTCPs().Informer().AddEventHandler(eventHandler)
    factoryCrd.Traefik().V1alpha1().TLSOptions().Informer().AddEventHandler(eventHandler)
    factoryCrd.Traefik().V1alpha1().TraefikServices().Informer().AddEventHandler(eventHandler)

    factoryKube := informers.NewSharedInformerFactoryWithOptions(c.csKube, resyncPeriod, informers.WithNamespace(ns))
    factoryKube.Extensions().V1beta1().Ingresses().Informer().AddEventHandler(eventHandler)
    factoryKube.Core().V1().Services().Informer().AddEventHandler(eventHandler)
    factoryKube.Core().V1().Endpoints().Informer().AddEventHandler(eventHandler)

    c.factoriesCrd[ns] = factoryCrd
    c.factoriesKube[ns] = factoryKube
  }

  return eventCh, nil
}

func (c *clientWrapper) newResourceEventHandler(events chan<- interface{}) cache.ResourceEventHandler {
  // 包装成 resourceEventHandler 解决事件通知
  return &cache.FilteringResourceEventHandler{
    Handler: &resourceEventHandler{ev: events},
  }
}

type resourceEventHandler struct {
  ev chan<- interface{}
}
// 资源的创建, 更新和删除方法.
func (reh *resourceEventHandler) OnAdd(obj interface{}) { reh.ev <- obj }
func (reh *resourceEventHandler) OnUpdate(oldObj, newObj interface{}) { reh.ev <- obj }
func (reh *resourceEventHandler) OnDelete(obj interface{})  { reh.ev <- obj }

// pkg/provider/kubernetes/crd/kubernetes.go

func (p *Provider) loadConfigurationFromCRD(ctx context.Context, client Client) *dynamic.Configuration {
  conf := &dynamic.Configuration{
    HTTP: p.loadIngressRouteConfiguration(ctx, client, tlsConfigs),
  }

  cb := configBuilder{client}
  for _, service := range client.GetTraefikServices() {
    cb.buildTraefikService(ctx, service, conf.HTTP.Services)
  }

  return conf
}

// pkg/provider/kubernetes/crd/kubernetes_http.go

func (p *Provider) loadIngressRouteConfiguration(...) *dynamic.HTTPConfiguration {
  conf := &dynamic.HTTPConfiguration{}

  for _, ingressRoute := range client.GetIngressRoutes() {
    cb := configBuilder{client}
    for _, route := range ingressRoute.Spec.Routes {
      // 关心单个 service 关联的 pod 的 lb 配置方式 只看这一个分支即可
      if len(route.Services) == 1 {
        fullName, serversLB, err := cb.nameAndService(ctx, ingressRoute.Namespace, route.Services[0].LoadBalancerSpec)
        if serversLB != nil {
          // 这是重点, 接下来看 serversLB 的生成方式
          conf.Services[serviceName] = serversLB
        }
      }
    }
  }

  return conf
}

// pkg/provider/kubernetes/crd/kubernetes_http.go

func (c configBuilder) nameAndService(...) (...) {
  serversLB, err := c.buildServersLB(namespace, service)

  return
}

/* 
svc LoadBalancerSpec 对应 yaml 文件中的 spec.routes.services 部分
如:
- name: traefik-dashboard
  port: 80
*/
type LoadBalancerSpec struct {
  Name      string          `json:"name"`
  Kind      string          `json:"kind"`
  Namespace string          `json:"namespace"`
  Sticky    *dynamic.Sticky `json:"sticky,omitempty"`

  Port               int32                       `json:"port"`
  Scheme             string                      `json:"scheme,omitempty"`
  HealthCheck        *HealthCheck                `json:"healthCheck,omitempty"`
  Strategy           string                      `json:"strategy,omitempty"`
  PassHostHeader     *bool                       `json:"passHostHeader,omitempty"`
  ResponseForwarding *dynamic.ResponseForwarding `json:"responseForwarding,omitempty"`

  Weight *int `json:"weight,omitempty"`
}

// buildServersLB creates the configuration for the load-balancer of servers defined by svc.
func (c configBuilder) buildServersLB(namespace string, svc v1alpha1.LoadBalancerSpec) (*dynamic.Service, error) {
  servers, err := c.loadServers(namespace, svc)

  lb := &dynamic.ServersLoadBalancer{}
  lb.Servers = servers

  return &dynamic.Service{LoadBalancer: lb}, nil
}

func (c configBuilder) loadServers(fallbackNamespace string, svc v1alpha1.LoadBalancerSpec) ([]dynamic.Server, error) {
  strategy := svc.Strategy

  if strategy != roundRobinStrategy {
    return nil, fmt.Errorf("load balancing strategy %s is not supported", strategy) // 挺尴尬的
  }

  namespace := namespaceOrFallback(svc, fallbackNamespace)

  service, exists, err := c.client.GetService(namespace, sanitizedName)

  confPort := svc.Port
  var portSpec *corev1.ServicePort
  // 取到了 service 中 port 部分定义
  for _, p := range service.Spec.Ports {
    if confPort == p.Port {
      portSpec = &p
      break
    }
  }

  var servers []dynamic.Server

  endpoints, endpointsExists, endpointsErr := c.client.GetEndpoints(namespace, sanitizedName)

  // 取到对应的 endpoints
  var port int32
  for _, subset := range endpoints.Subsets {
    for _, p := range subset.Ports {
      if portSpec.Name == p.Name {
        port = p.Port
        break
      }
    }

    protocol := httpProtocol
    scheme := svc.Scheme
    // 包装配置返回
    for _, addr := range subset.Addresses {
      servers = append(servers, dynamic.Server{
        URL: fmt.Sprintf("%s://%s:%d", protocol, addr.IP, port),
      })
    }
  }

  return servers, nil
}