• Apisix-Ingress服务发现详解

    apisix

    Apache APISIX 是一个基于微服务 API 网关，其不仅可以处理南北向的流量，也可以处理东西向的流量即服务之间的流量。Apache APISIX 集成了控制面板和数据面，与其他 API 网关相比，Apache APISIX 的上游、路由、插件全是动态的，修改这些东西时都不用重启。并且 Apache APISIX 的插件也是热加载，可以随时插拔、修改插件。

    Apache APISIX 其设计理念是基于API 网关的数据平面和控制平面分离。控制平面不仅仅能够控制 Apache APISIX ，同时其还能够控制其他组件；数据平面不仅仅能够被自身的控制平面控制，还能被其他组件所控制。由于其基于ETCD 来存储和分发路由数据，默认具备高可用，无单点故障风险。除此之外，其能够友好地支持 Prometheus、SkyWalking 动态追踪、流量复制、故障注入等相关功能。

    apisix ingress

    在 K8s 生态中，Ingress 作为表示 K8s 流量入口的一种资源，想要让其生效，就需要有一个 Ingress Controller 去监听 K8s 中的 Ingress 资源，并对这些资源进行相应规则的解析和实际承载流量。在当下趋势中，像 Kubernetes Ingress Nginx 就是使用最广泛的 Ingress Controller 实现。

    而 APISIX Ingress 则是另一种 Ingress Controller 的实现。跟 Kubernetes Ingress Nginx 的区别主要在于 APISIX Ingress 是以 Apache APISIX 作为实际承载业务流量的数据面。

    Apache APISIX Ingress Controller 除了覆盖 NGINX Ingress Controller 已有的能力外，还解决了一些 Nginx Ingress Controller 的痛点。具体如下：

    • 1、配置的动态化加载
      通常情况下，作为接入层的 Ingress Controller ，其承载着服务的入口流量引入，在生产环境中，我们的业务对系统的可靠性有着更高的要求，然而，基于 Apache APISIX Ingress Controller 其能够支持动态配置，即时生效，降低生产事故的意外及风险，有助于提高运维可维护性。

    • 2、较强的灰度能力
      在实际的业务场景中，有的时候，往往会依据某些特定的需求进行权重调整，结合业务需求按比例进行流量控制，Apache APISIX Ingress Controller 可以支持 Service和 Pod 级别的权重调整，配置清晰而且可读性更强。
      除此，相对于NGINX Ingress Controller 中通过 Annotation 的方式提供 Canary 灰度方案，Apache APISIX Ingress Controller 能够解决其缺陷，从而能够更好的提供灰度策略。

    • 3、较好的扩展能力
      基于 Apache APISIX 强大的插件能力，Apache APISIX Ingress Controller 通过动态绑定插件来增强功能。Apache APISIX 通过插件封装逻辑，易于管理；完善的文档，易于使用和理解。Apache APISIX Ingress Controller 通过配置即可绑定和解绑插件，无需操作脚本。

    APISIX Ingress 目前已经支持的自定义资源主要是以下 5 类，涉及到路由、上游、消费者、证书相关和集群公共配置的相关类别。

    内置服务发现

    APISIX 内置了下面这些服务发现机制：

    • 基于 Eureka 的服务发现

    • 基于 Nacos 的服务发现

    • 基于 Consul 的服务发现

    • 基于 Consul KV 的服务发现

    • 基于 DNS 的服务发现

    • 基于 APISIX-Seed 架构的控制面服务发现

    • 基于 Kubernetes 的服务发现

    上面介绍的这些都是基于数据面apisix配置信息手动变更操作的方案集成。其实，在基于k8s的云原生场景下，apisix还提供了一个控制面组件来对apisix的服务发现进行自动管理，那就是apisix ingress。

    下面就从源码的角度来看看apisix ingress是怎么做到自动的服务发现的

    apisix ingress启动

    从main.go启动入口一路跟踪，进入providers/controller.go的run方法:

    1. func (c *Controller) run(ctx context.Context) {
    2.    log.Infow("controller tries to leading ...",
    3.       zap.String("namespace", c.namespace),
    4.       zap.String("pod", c.name),
    5.    )
    6.  
    7.    var cancelFunc context.CancelFunc
    8.    ctx, cancelFunc = context.WithCancel(ctx)
    9.    defer cancelFunc()
    10.  
    11.    // give up leader
    12.    defer c.leaderContextCancelFunc()
    13.  
    14.    clusterOpts := &apisix.ClusterOptions{
    15.       AdminAPIVersion: c.cfg.APISIX.AdminAPIVersion,
    16.       Name: c.cfg.APISIX.DefaultClusterName,
    17.       AdminKey: c.cfg.APISIX.DefaultClusterAdminKey,
    18.       BaseURL: c.cfg.APISIX.DefaultClusterBaseURL,
    19.       MetricsCollector: c.MetricsCollector,
    20.    }
    21.    err := c.apisix.AddCluster(ctx, clusterOpts)
    22.    if err != nil && err != apisix.ErrDuplicatedCluster {
    23.       // TODO give up the leader role
    24.       log.Errorf("failed to add default cluster: %s", err)
    25.       return
    26.    }
    27.  
    28.    if err := c.apisix.Cluster(c.cfg.APISIX.DefaultClusterName).HasSynced(ctx); err != nil {
    29.       // TODO give up the leader role
    30.       log.Errorf("failed to wait the default cluster to be ready: %s", err)
    31.  
    32.       // re-create apisix cluster, used in next c.run
    33.       if err = c.apisix.UpdateCluster(ctx, clusterOpts); err != nil {
    34.          log.Errorf("failed to update default cluster: %s", err)
    35.          return
    36.       }
    37.       return
    38.    }
    39.  
    40.    // Creation Phase
    41.  
    42.    c.informers = c.initSharedInformers()
    43.    common := &providertypes.Common{
    44.       ControllerNamespace: c.namespace,
    45.       ListerInformer: c.informers,
    46.       Config: c.cfg,
    47.       APISIX: c.apisix,
    48.       KubeClient: c.kubeClient,
    49.       MetricsCollector: c.MetricsCollector,
    50.       Recorder: c.recorder,
    51.    }
    52.  
    53.    c.namespaceProvider, err = namespace.NewWatchingNamespaceProvider(ctx, c.kubeClient, c.cfg)
    54.    if err != nil {
    55.       ctx.Done()
    56.       return
    57.    }
    58.  
    59.    c.podProvider, err = pod.NewProvider(common, c.namespaceProvider)
    60.    if err != nil {
    61.       ctx.Done()
    62.       return
    63.    }
    64.  
    65.    c.translator = translation.NewTranslator(&translation.TranslatorOptions{
    66.       APIVersion: c.cfg.Kubernetes.APIVersion,
    67.       EndpointLister: c.informers.EpLister,
    68.       ServiceLister: c.informers.SvcLister,
    69.       SecretLister: c.informers.SecretLister,
    70.       PodLister: c.informers.PodLister,
    71.       ApisixUpstreamLister: c.informers.ApisixUpstreamLister,
    72.       PodProvider: c.podProvider,
    73.    })
    74.  
    75.    c.apisixProvider, c.apisixTranslator, err = apisixprovider.NewProvider(common, c.namespaceProvider, c.translator)
    76.    if err != nil {
    77.       ctx.Done()
    78.       return
    79.    }
    80.  
    81.    c.ingressProvider, err = ingressprovider.NewProvider(common, c.namespaceProvider, c.translator, c.apisixTranslator)
    82.    if err != nil {
    83.       ctx.Done()
    84.       return
    85.    }
    86.  
    87.    c.kubeProvider, err = k8s.NewProvider(common, c.translator, c.namespaceProvider, c.apisixProvider, c.ingressProvider)
    88.    if err != nil {
    89.       ctx.Done()
    90.       return
    91.    }
    92.  
    93.    if c.cfg.Kubernetes.EnableGatewayAPI {
    94.       c.gatewayProvider, err = gateway.NewGatewayProvider(&gateway.ProviderOptions{
    95.          Cfg: c.cfg,
    96.          APISIX: c.apisix,
    97.          APISIXClusterName: c.cfg.APISIX.DefaultClusterName,
    98.          KubeTranslator: c.translator,
    99.          RestConfig: nil,
    100.          KubeClient: c.kubeClient.Client,
    101.          MetricsCollector: c.MetricsCollector,
    102.          NamespaceProvider: c.namespaceProvider,
    103.       })
    104.       if err != nil {
    105.          ctx.Done()
    106.          return
    107.       }
    108.    }
    109.  
    110.    // Init Phase
    111.  
    112.    if err = c.namespaceProvider.Init(ctx); err != nil {
    113.       ctx.Done()
    114.       return
    115.    }
    116.    if err = c.apisixProvider.Init(ctx); err != nil {
    117.       ctx.Done()
    118.       return
    119.    }
    120.  
    121.    // Run Phase
    122.  
    123.    e := utils.ParallelExecutor{}
    124.  
    125.    e.Add(func() {
    126.       c.checkClusterHealth(ctx, cancelFunc)
    127.    })
    128.  
    129.    e.Add(func() {
    130.       c.informers.Run(ctx)
    131.    })
    132.  
    133.    e.Add(func() {
    134.       c.namespaceProvider.Run(ctx)
    135.    })
    136.  
    137.    e.Add(func() {
    138.       c.kubeProvider.Run(ctx)
    139.    })
    140.  
    141.    e.Add(func() {
    142.       c.apisixProvider.Run(ctx)
    143.    })
    144.  
    145.    e.Add(func() {
    146.       c.ingressProvider.Run(ctx)
    147.    })
    148.  
    149.    if c.cfg.Kubernetes.EnableGatewayAPI {
    150.       e.Add(func() {
    151.          c.gatewayProvider.Run(ctx)
    152.       })
    153.    }
    154.  
    155.    e.Add(func() {
    156.       c.resourceSyncLoop(ctx, c.cfg.ApisixResourceSyncInterval.Duration)
    157.    })
    158.    c.MetricsCollector.ResetLeader(true)
    159.  
    160.    log.Infow("controller now is running as leader",
    161.       zap.String("namespace", c.namespace),
    162.       zap.String("pod", c.name),
    163.    )
    164.  
    165.    <-ctx.Done()
    166.    e.Wait()
    167.  
    168.    for _, execErr := range e.Errors() {
    169.       log.Error(execErr.Error())
    170.    }
    171.    if len(e.Errors()) > 0 {
    172.       log.Error("Start failed, abort...")
    173.       cancelFunc()
    174.    }
    175. }

    上面代码逻辑大致如下：

    • 初始化apisix集群配置信息：用于跟数据面服务apisix通信进行相关的配置操作

    • 初始化k8s资源inforrmers信息：用户对k8s各个资源进行监听及获取资源信息

    • 监听k8s集群namespace资源并处理

    • 监听k8s集群pod资源并处理

    • 监听k8s集群ingress资源并处理

    • 监听k8s集群中apisix自定义资源并处理，比如：apisixRoute等

    • 监听k8s集群endpoint资源并处理

    • 监听k8s集群secret资源并处理

    • 监听k8s集群configmap资源并处理

    • 监听k8s集群gateway资源并处理

    下面以处理endpoint资源为例进行说明，其他资源的监听处理类似，就不一一讲解了。

    服务发现

    进入到k8s/endpoint/provider.go中，我们先来看看实例初始化方法：

    1. func NewProvider(common *providertypes.Common, translator translation.Translator, namespaceProvider namespace.WatchingNamespaceProvider) (Provider, error) {
    2.    p := &endpointProvider{
    3.       cfg: common.Config,
    4.    }
    5.  
    6.    base := &baseEndpointController{
    7.       Common: common,
    8.       translator: translator,
    9.  
    10.       svcLister: common.SvcLister,
    11.       apisixUpstreamLister: common.ApisixUpstreamLister,
    12.    }
    13.  
    14.    if common.Kubernetes.WatchEndpointSlices {
    15.       p.endpointSliceController = newEndpointSliceController(base, namespaceProvider)
    16.    } else {
    17.       p.endpointsController = newEndpointsController(base, namespaceProvider)
    18.    }
    19.  
    20.    return p, nil
    21. }
    22. func newEndpointsController(base *baseEndpointController, namespaceProvider namespace.WatchingNamespaceProvider) *endpointsController {
    23.    ctl := &endpointsController{
    24.       baseEndpointController: base,
    25.  
    26.       workqueue: workqueue.NewNamedRateLimitingQueue(workqueue.NewItemFastSlowRateLimiter(1*time.Second, 60*time.Second, 5), "endpoints"),
    27.       workers: 1,
    28.  
    29.       namespaceProvider: namespaceProvider,
    30.  
    31.       epLister: base.EpLister,
    32.       epInformer: base.EpInformer,
    33.    }
    34.  
    35.    ctl.epInformer.AddEventHandler(
    36.       cache.ResourceEventHandlerFuncs{
    37.          AddFunc: ctl.onAdd,
    38.          UpdateFunc: ctl.onUpdate,
    39.          DeleteFunc: ctl.onDelete,
    40.       },
    41.    )
    42.  
    43.    return ctl
    44. }

    注意到最后代码的AddEventHandler，这里就是我们经常见到的informer的处理回调方法设置的地方。

    我们看到针对endpoint资源的增删改，设置了对应的回调方法。这里来看看onAdd方法：

    1. func (c *endpointsController) onAdd(obj interface{}) {
    2.    key, err := cache.MetaNamespaceKeyFunc(obj)
    3.    if err != nil {
    4.       log.Errorf("found endpoints object with bad namespace/name: %s, ignore it", err)
    5.       return
    6.    }
    7.    if !c.namespaceProvider.IsWatchingNamespace(key) {
    8.       return
    9.    }
    10.    log.Debugw("endpoints add event arrived",
    11.       zap.String("object-key", key))
    12.  
    13.    c.workqueue.Add(&types.Event{
    14.       Type: types.EventAdd,
    15.       // TODO pass key.
    16.       Object: kube.NewEndpoint(obj.(*corev1.Endpoints)),
    17.    })
    18.  
    19.    c.MetricsCollector.IncrEvents("endpoints", "add")
    20. }

    该方法的参数表示增加的endpoint资源对象信息。该方法主要是向endpoint的队列workqueue中增加一个事件对象：包含事件类型、增加的endpoint对象。

    在最开始main我们介绍provider的启动方法中提到：执行了每个provider的run方法，下面我们来看下endpoint的provider的run方法：

    1. func (c *endpointsController) run(ctx context.Context) {
    2.    log.Info("endpoints controller started")
    3.    defer log.Info("endpoints controller exited")
    4.    defer c.workqueue.ShutDown()
    5.  
    6.    if ok := cache.WaitForCacheSync(ctx.Done(), c.epInformer.HasSynced); !ok {
    7.       log.Error("informers sync failed")
    8.       return
    9.    }
    10.  
    11.    handler := func() {
    12.       for {
    13.          obj, shutdown := c.workqueue.Get()
    14.          if shutdown {
    15.             return
    16.          }
    17.  
    18.          err := c.sync(ctx, obj.(*types.Event))
    19.          c.workqueue.Done(obj)
    20.          c.handleSyncErr(obj, err)
    21.       }
    22.    }
    23.  
    24.    for i := 0; i < c.workers; i++ {
    25.       go handler()
    26.    }
    27.  
    28.    <-ctx.Done()
    29. }
    30. func (c *endpointsController) sync(ctx context.Context, ev *types.Event) error {
    31.    ep := ev.Object.(kube.Endpoint)
    32.    ns, err := ep.Namespace()
    33.    if err != nil {
    34.       return err
    35.    }
    36.    newestEp, err := c.epLister.GetEndpoint(ns, ep.ServiceName())
    37.    if err != nil {
    38.       if errors.IsNotFound(err) {
    39.          return c.syncEmptyEndpoint(ctx, ep)
    40.       }
    41.       return err
    42.    }
    43.    return c.syncEndpoint(ctx, newestEp)
    44. }
    45. func (c *baseEndpointController) syncEndpoint(ctx context.Context, ep kube.Endpoint) error {
    46.    log.Debugw("endpoint controller syncing endpoint",
    47.       zap.Any("endpoint", ep),
    48.    )
    49.  
    50.    namespace, err := ep.Namespace()
    51.    if err != nil {
    52.       return err
    53.    }
    54.    svcName := ep.ServiceName()
    55.    svc, err := c.svcLister.Services(namespace).Get(svcName)
    56.    if err != nil {
    57.       if k8serrors.IsNotFound(err) {
    58.          return c.syncEmptyEndpoint(ctx, ep)
    59.       }
    60.       log.Errorf("failed to get service %s/%s: %s", namespace, svcName, err)
    61.       return err
    62.    }
    63.  
    64.    switch c.Kubernetes.APIVersion {
    65.    case config.ApisixV2beta3:
    66.       var subsets []configv2beta3.ApisixUpstreamSubset
    67.       subsets = append(subsets, configv2beta3.ApisixUpstreamSubset{})
    68.       auKube, err := c.apisixUpstreamLister.V2beta3(namespace, svcName)
    69.       if err != nil {
    70.          if !k8serrors.IsNotFound(err) {
    71.             log.Errorf("failed to get ApisixUpstream %s/%s: %s", namespace, svcName, err)
    72.             return err
    73.          }
    74.       } else if auKube.V2beta3().Spec != nil && len(auKube.V2beta3().Spec.Subsets) > 0 {
    75.          subsets = append(subsets, auKube.V2beta3().Spec.Subsets...)
    76.       }
    77.       clusters := c.APISIX.ListClusters()
    78.       for _, port := range svc.Spec.Ports {
    79.          for _, subset := range subsets {
    80.             nodes, err := c.translator.TranslateEndpoint(ep, port.Port, subset.Labels)
    81.             if err != nil {
    82.                log.Errorw("failed to translate upstream nodes",
    83.                   zap.Error(err),
    84.                   zap.Any("endpoints", ep),
    85.                   zap.Int32("port", port.Port),
    86.                )
    87.             }
    88.             name := apisixv1.ComposeUpstreamName(namespace, svcName, subset.Name, port.Port, types.ResolveGranularity.Endpoint)
    89.             for _, cluster := range clusters {
    90.                if err := c.SyncUpstreamNodesChangeToCluster(ctx, cluster, nodes, name); err != nil {
    91.                   return err
    92.                }
    93.             }
    94.          }
    95.       }
    96.    case config.ApisixV2:
    97.       var subsets []configv2.ApisixUpstreamSubset
    98.       subsets = append(subsets, configv2.ApisixUpstreamSubset{})
    99.       auKube, err := c.apisixUpstreamLister.V2(namespace, svcName)
    100.       if err != nil {
    101.          if !k8serrors.IsNotFound(err) {
    102.             log.Errorf("failed to get ApisixUpstream %s/%s: %s", namespace, svcName, err)
    103.             return err
    104.          }
    105.       } else if auKube.V2().Spec != nil && len(auKube.V2().Spec.Subsets) > 0 {
    106.          subsets = append(subsets, auKube.V2().Spec.Subsets...)
    107.       }
    108.       clusters := c.APISIX.ListClusters()
    109.       for _, port := range svc.Spec.Ports {
    110.          for _, subset := range subsets {
    111.             nodes, err := c.translator.TranslateEndpoint(ep, port.Port, subset.Labels)
    112.             if err != nil {
    113.                log.Errorw("failed to translate upstream nodes",
    114.                   zap.Error(err),
    115.                   zap.Any("endpoints", ep),
    116.                   zap.Int32("port", port.Port),
    117.                )
    118.             }
    119.             name := apisixv1.ComposeUpstreamName(namespace, svcName, subset.Name, port.Port, types.ResolveGranularity.Endpoint)
    120.             for _, cluster := range clusters {
    121.                if err := c.SyncUpstreamNodesChangeToCluster(ctx, cluster, nodes, name); err != nil {
    122.                   return err
    123.                }
    124.             }
    125.          }
    126.       }
    127.    default:
    128.       panic(fmt.Errorf("unsupported ApisixUpstream version %v", c.Kubernetes.APIVersion))
    129.    }
    130.    return nil
    131. }

    上面代码主要逻辑就是：

    • 从endpoint的队列workqueue中获取事件对象

    • 根据endpoint信息从k8s集群中获取最新的namespace和service等信息

    • 根据namespace和servicename从k8s集群中获取apisix upstream资源信息

    • 对每一个service端口，向数据面服务apisix发送配置更新请求

    https://xiaorui.cc/archives/7369

  • 相关阅读:
    Java实现SQL分页
    合成孔径雷达地面运动目标检测技术研究——基于概率图（Matlab代码实现）
    Java SE 19 新增特性
    Request和Response介绍 [Tomcat][Servlet]
    计算机毕业设计之java+ssm植物养护管理系统
    conda创建环境、安装包到环境迁移
    命名空间和作用域
    Vmware 静态网络配置
    OpenSIPS 防扫描处理
    java中的IO流之序列化与反序列化（对象数据和文件的读写交互）
  • 原文地址：https://blog.csdn.net/m0_47495420/article/details/133781847