Knative 系列（四）：Eventing 篇

本节对 Knative 的第三大组件—Eventing进行介绍，Kubernetes 用户在实现开发和部署阶段服务之间松耦合的同时，服务间常要通过不同的事件机制来进行事件传递，为了解决大部分的云原生消息通信需求，Knative 提供了 Eventing 组件。

与Serving和Building两个组件相同，Eventing也是基于 Kubernetes 的控制器模式实现的。Eventing 在版本迭代过程中陆续更新一些特性，版本之间有些许差距，本文基于Knative Eventing 0.6 进行介绍。由于 Eventing 涉及的概念比较多，我们根据版本迭代过程来讲解 Eventing 组件，从而呈现整个 Eventing 的设计思路。同时，鉴于 0.6 版本解除了对Istio的依赖，本篇也不会提及 Knative 的好朋友 Istio。

1、事件的订阅发布

在 0.5 版本之前，Eventing 最重要的两个对象是 Channel 和 Subscriptions，如下图所示，Channel 是事件生产者与事件消费者的中间层，它会从事件源（Event Source）收集事件并且分发给事件消费者（Service）。创建一个 Channel 对象会将一个事件源绑定到该 Channel，这里我们定义的 Channel 必须是可寻址的，目前 Knative 只支持 HTTP 协议，这个地址对应是一个 URI，事件源可以通过这个 URI 将事件发送给 Channel。那么事件消费者从哪里订阅事件呢？ Eventing 提出了另一个对象 Subscription，用户可以通过创建一个 Subscription 对象来指定一个事件消费者对事件的订阅，需要分发给哪些消费者是通过 Subscription 来指定的，同样消费者需要提供分发路径 URI。

可以看出，Channel 起到了一个订阅分发事件的作用，但实际上这里的 Channel 只是一个概念，底层存储可以有不同形式的实现，在 Knative Eventing 中称为 Provisioner。目前 Knative 已经支持多种不同的 Provisioner：Apache Kafka、GCP PubSub、NATS。除了对这些消息中间件的支持，为方便用户学习和测试使用，Knative 也提供便捷的 Provisioner：In-Memory Channel。但是 In-Memory 将事件信息缓存在内存中，服务一旦退出会导致事件信息全部丢失，因为生产环境使用是不安全的。

我们通过一组示例来展示怎样利用 Eventing 组件实现订阅发布目的。

1、首先定义一个 Channel 对象，Spec 字段需要指定 Channel 的底层实现，这里采用的是 In-memory 类型的 Provisioner。

2、创建事件源。这里我们定义一个从Github获取消息的事件源。目前，Eventing 组件已经支持多种事件源，GitHubSource 是其中的一类，参数中需要配置关注的事件点，这里关注的事件点是某个 Git 库的项目 Pull 事件，Sink 中指定消息会发送到我们在第一步创建的 Channel 中。

3、创建事件订阅方。创建一个 Subscription 对象绑定一个 Subscriber 到 Channel 的绑定，创建之后，集群中名称为 test 的 Service 就可以成功订阅 Channel，也可以获取 Github 的项目 Pull 事件。同样，也可以创建多个 Subecription 来绑定更多的 Subscriber。

如上所述，我们就实现了一组订阅发布，实现了集群内部服务对 Github 事件的订阅。

下面，我们讲一下上述操作背后的控制器实现，对于 In-Memory 的 Channel 底层存储支持以及 Github 事件源支持的实现，我们在第四部分“扩展 Knative Eventing”中会进行介绍，这里我们主要讲一下其余两个主要的控制器，Subscribtion 控制器和 Channel 控制器。

首先是 Channel Controller。这个控制器的逻辑很简单，它关注 Channel 资源的变动，检测 Channel 底层的 Provisioner 是否已经安装支持，如果不支持会在 Status 标记。

其次是 Subscribtion Controller。它会关注 Subscribtion 资源的变动，一方面生成生成 Subscriber 的 URI（如果 Subscriber 是一个 Kubernetes Service 对象，那么 Service 的访问路径就是该 URI）；另一方面会将 Subscribtion 注册到指定的 Channel，也就是将 Subscribtion 加入到 Channel 的 Spec 的 Subscrible 字段，这样 Channel 就知道要将自己的消息分发到哪些 Subscriber 了。

从控制器逻辑上看，完全没有事件采集、保存、转发的实际流程，那么这部分流程在哪里呢？前文已经介绍了 Channel 只是概念的逻辑，Channel Controller 自然也不会有实际的举动。事件的处理都在事件源组件与 Provisioner 两个部分完成，这两个部分是可以插件化的扩展的，主要的工作流程在第四节你就会明白。

2、事件的过滤转发

在 0.5 版本中，Knative Eventing 增加了 Broker 和 Trigger 对象的定义，方便对事件进行过滤，事件的消费方可以利用这种机制只获取自己感兴趣的事件，而不是无脑接收所有 Channel 中的事件信息。说到这里，你首先会想问“过滤条件”的指标怎么定义呢？Eventing 的过滤指标暂时支持的有两项：Type（事件的类型）和 Source（事件来源），后续版本应该会支持更多的筛选规则。那又怎样能保证所有的事件都有这两项呢？其实 Eventing 组件要求事件信息的格式符合 CloudEvents 的标准（https://github.com/cloudevents/spec/blob/master/spec.md#design-goals），这两项指标在标准参数之内。

如果你的 Eventing 版本是 0.5 之后的，那么你就可以使用 Trigger 来订阅自己感兴趣的事件了。订阅的方式也很简单，原先我们订阅一个事件是创建一个 Subscribtion 对象，现在创建一个 Trigger 对象即可。下图定义了一个 Trigger，Spec 字段中的 filter 字段定义了筛选逻辑，只订阅 Broker 名称为 default 中的 Type 为 dev.knative.foo 同时 Source 为 dev.knative.bar 的消息，订阅方是名称为 message-dumper 的 Service。

那 Broker 是用来做什么的呢？从逻辑图里很容易认为 Broker 是 Channel 的替代品，其实不是，它不是为了事件信息缓存转发存在的，而是为了事件过滤存在概念，这里展示一个 Broker 的定义就一目了然。可以看见定义一个 Broker 的时候仍旧需要指定一个 Channel，这个 Channel 是 Broker 的事件来源。Broker 根据 Trigger 中定义的规则，对事件进行过滤，过滤之后发送给事件的订阅方。

与上述过滤机制相关的控制器有两个：Trigger Controller 和 Broker Controller。其中 Trigger Controller 处理 Trigger 资源的逻辑，Broker Controller 处理 Broker 资源的逻辑。

Broker Controller 的处理逻辑主要如下：

1、创建 Broker 资源定义的 Channel 资源。

2、创建用于 Fliter 的 Deployment。该服务负责实现消息过滤。

3、 创建步骤 2 中的 Deployment 资源的上层 Service 资源。

Trigger Controller 主要为每个 Trigger 资源创建一个 Subscribtion，由于 Channel 中的事件只会分发给由 Subscribtion 指定的 Subscriber，这里创建 Subscribtion 对象本质上是用于将 Channel 中的事件导流，Subscriber 的 URI 参数为 Broker 控制器在第三步中创建的 Service 的访问路径。这样，Channel 中的事件被导流到用于 Flitter 的 Service，Flitter 根据 Trigger 中定义的规则过滤之后再将事件给真实的 Subscriber，从而实现事件过滤。

3、事件注册

在 0.6 版本中，Knative Eventing 增加了 EventType 对象的定义，方便用户得知从不同 Broker 获取的所有事件集合。在提出 EventType 概念之前，用户在定义了一系列的 Broker 之后很难管理集群中可以订阅的事件，需要逐个查看 Channel 资源的事件源，EventType 就是为了解决这个问题提出的。

在创建事件源对象资源的同时，控制器会创建一个对应的 EventType 对象来注册消息源对象。有了 EventType 的事件注册支持，用户通过 kubeclt get eventtypes 就可以获取集群中所有通过 Broker 定义的事件源，从而查看是否有自己需要订阅的事件，示例如下：

原图链接https://knative.dev/docs/eventing/event-registry/

EventType 只能用于搭配 Broker 和 Trigger 使用，用户在获取集群中的事件源集合之后，同时可以查看事件是否可以被订阅。如果允许订阅，就可以通过 Trigger 来订阅感兴趣的事件。

4、扩展 Knative Eventing

最后，我们来介绍对于 Knative Eventing 组件可以做的个性化扩展，读者可以通过这些扩展来支持自己的业务或者是组件。

4.1 扩展 Channel 底层存储实现的支持

前文我们说过，Channel 资源是一个概念上的含义，Channel Controller 并没有完成事件的存储转发等行为，而是由不同的 Provisioner 实现的。这里我们以 GCP PubSub 为例介绍怎样扩展 Channel 底层存储支持，同时也可以了解 Provisioner 的工作机制，清楚 Channel 消息是怎么接收、存储和分发的。

GCP PubSub 是一个分布式消息总线，使用时需要创建一个 Topic，然后才可以向 Topic 发布消息或者是订阅主题。对应到 Eventing，一个 Channel 也就对应了一个 Topic，对 Channel 的一个订阅对应了一个 GCP PubSub 的 Subscription 的创建，在实现时，会对 Channel 的 Status 字段中添加 Subscription 字段来记录该 Channel 实际创建的 GCP Pubsub Subscription 资源。

扩展对 GCP PubSub 的支持首先需要创建一个特定名称（gcp-pubsub）的 ClusterChannelProvisioner 对象，“ClusterChannelProvisioner”是由 Eventing 组件提供的资源类型。创建了这个对象就是向 Eventing 组件宣称：你已经可以支持名称为“gcp-pubsub”的 Provisioner 了，这种类型的 Channel 也可以正常工作了。

下一步就是实现控制器逻辑了，控制器需时刻关注 Channel 的动静，一旦有 Provisioner 类型为“gcp-pubsub”的 Channel 增删，控制器就要负起责任，实现底层的资源的增删。

控制器有两部分构成：Channel 控制器和 Dispatcher 分发器；首先介绍 Channel 控制器的处理逻辑：

1、创建 Kubernetes Service 对象。每个 Channel 资源会对应创建一个 Service，Channel 的 URI 也就是 Service 的访问路径，Service 的类型为 externalName，这种类型的 service 背后并没有真实存在的 Pod 支撑，是一个空的 Service，而进入 Service 的流量全部导入 externalName 指定的路径，这个路径是我们接下来要介绍的另一个组件的 Service 路径。

2、创建 Topic。每个 Channel 对应一个 GCP PubSub 的 Topic，如果这个 Channel 对象是新建的，那么需要创建一个 Topic。

3、创建 Subscription。将 Channel 的 Spec 字段中的 Subscrible（该字段有对该 Channel 进行订阅的所有的 Subscriber 对象集合，是 subscription 控制器在 subscription 对象创建的时候写入 Channel 的）与 Status 字段中的 subscription（GCP PubSub 中实际存在的订阅资源）进行对比，创建/删除 GCP PubSub 资源，保证一个 Channel 订阅对应一个真实的 Gcp PubSub 订阅资源。

Dispatcher 分发器以 Deployment 部署，上层有一个 Kubernetes Service 资源，我们在上文中说过，进入 Channel 的事件都会被转发到分发器，可以看出 Dispatcher 实际上是 Gcp PubSub 与 Eventing 之间的 Adapter，负责将集群内事件生产者进入 Channel 的事件发送到指定 Topic，同时将 Topic 中的事件分发给各个事件的消费者。

Dispatcher 组件也是以控制器形式完成的，对于每个 Channel 会创建一个 Receiver 对象来负责接收集群中的事件，并将事件发送给 Gcp Pubsub，每一个 Subscriber 会创建一个 Dispatcher 协程负责接收 GCP PubSub 的事件并发送给指定的 Subscriber。

根据上述逻辑，你也可以愉快地开发支持自己的消息中间件作为 Channel 底层存储啦。介绍怎样实现 Channel 对外部消息中间件的支持，掌握了其中原理，你也就可以扩展自己的消息中间件作为 Channel 的 Provisioner。

4.2 扩展对新外部消息源的支持

除了 Channel 底层消息存储支持扩展之外，还有一部分可以实现个性化支持的就是消息源。对于 Channel 而言，事件源并非是真正产生事件的地方，而是集群中将事件源产生的事件收集到集群内的 Adapter。目前 Eventing 已经支持了很多的事件源，也开发完成了相应的 Adapter，例如 GitLab、GitHub、GCP PubSub、Apache Kafka 等等。还有一些事件源并不会直接被使用，而会让其他的事件源实现更加方便，例如 Container Source。

这里还是以 GCP PubSub 为例，讲解怎样实现对一个外部事件源的支持，实现模式仍旧是老朋友控制器模式。

首先需要定义一个 CRD，名称为“GcpPubSubSource”。这里介绍 Spec 中的几个比较重要的字段：GoogleCloudProject、Topic 和 Sink，其中 GoogleCloudProject 和 Topic 两个字段用于确定事件来源的 Gcp PubSub Topic；Sink 用于定义事件的去向，及事件的去向，例如可以是 Kubernetes 集群中的某个 Service 或者是一个 Channel。当指定 Channel 时，该 Topic 生成的事件就可以分发到 Channel 中，而订阅 Channel 的消费者可以获取 Topic 的事件了。当然 Sink 也可以是一个事件的直接消费者 Service，Service 拿到事件直接处理。

接下来介绍背后的控制器处理逻辑：

1、 创建一个 Gcp PubSub 的订阅对象 Subscribtion。订阅指定的 Project 以及 Topic。

2、 创建一个 ReceiverAdapter 对象。该 Adapter 组件以 Deployment 的形式运行在集群中，负责从 Subscribtion 中获取消息并且将消息发送给 Sink 字段的对象。

3、 创建一个 EventType 资源。EventType 对象中有个字段为 Broker，Broker的名称为sink.Name。

4.3  利用 Container Source 创建自定义事件源

很多情况下，我们也会希望将一些自己开发的服务作为事件的生产者，Eventing 提供了一种 Container Source 的事件源来方便用户创建自定义事件源，用户的服务只要满足几个条件就可以成为 Eventing 的事件生产者：

1、事件生产者服务需要打包成一个镜像。

2、镜像运行成容器时可以从环境变量中获取参数信息，该参数信息指定了服务的事件发送目标。

有了这样的镜像之后，用户可以创建一个 Container Source 资源来创建该服务的事件源，如下所示：

Container Source 控制器的处理逻辑如下：

1、 解析 ContainerSource 资源的 args（或者是 sink 变量）中定义的路径或者是 Kubernetes Service 对象解析成可以访问的 URI。

2、创建 Deployment。Deployment 中的镜像参数是用户的事件源服务对象，同时会将 sink 的 URI 作为容器的 Env 参数以及 Args 参数。这样镜像运行成容器之后就可以从参数中获取事件的发送路径，当这个路径对应一个 Channel 时，就可以实现事件的订阅分发了。

相关文章：

Knative 系列（一）：基本概念和原理解读

Knative 系列（二）：兵马未动粮草先行之 Build 篇

Knative 系列（三）：Serving 篇