Knative Eventing入门解析

关于Knative被大家所熟知的标签应该是Serverless。的确，它算得上是这个领域的先驱者，但是它不仅仅只有Serverless，它还提供了事件驱动架构，事件驱动架构是微服务架构的延伸，它将事件作为微服务之间的通信方式，使得微服务之间可以更加灵活地进行通信和协作。

CloudEvents规范

事件代表事实，因此不包括目的地，而消息则传达意图，将数据从源头传送到特定的目的地。在服务器端代码中，事件通常用于连接不同的系统，其中一个系统的状态变化会导致另一个系统的代码执行。

源(Source)生成消息(Message)，其中事件(Event)被封装在协议中。事件到达目的地，触发一个由事件数据提供的动作(Action)。

源是源类型的一个特定实例，一个特定源类型的开放源软件可以由多个公司或供应商部署。

事件可以通过各种行业标准协议（如HTTP、AMQP、MQTT、SMTP）、开源协议（如Kafka、NATS）或平台/供应商特定的协议（AWS Kinesis、Azure Event Grid）来递送。

动作处理定义了由特定源的特定事件触发的行为或效果的事件。虽然不在本规范的范围内，但生成事件的目的通常是为了让其他系统能够轻松地对它们无法控制的源中的变化做出反应。源和动作通常是由不同的开发人员建立的。通常情况下，源是一个托管服务，而动作是无服务器函数（如AWS Lambda或Google Cloud Functions）中的自定义代码。

CloudEvents设计目标

CloudEvents 的核心是定义了一组关于系统间传输事件的元数据（被称为属性），以及这些元数据应该如何出现在该消息中。

CloudEvents 规范将不包括协议级的路由信息（例如，发送事件的目标 URL）。

CloudEvents 规范集定义了四种不同类型的协议元素，它们构成了一个分层架构模型。

1. 基本规范 定义了由属性（键值对）和相关规则组成的抽象信息模型，这些属性和相关规则构成了CloudEvents。
2. 扩展 添加了特定的、可能重叠的扩展属性和相关规则集，例如，支持不同的跟踪标准。
3. 事件格式编码（如 JSON）定义了基本规范的信息模型和所选扩展的编码方式，以将其映射到应用协议的头和有效载荷元素。
4. 协议绑定，例如 HTTP，定义了 CloudEvent 如何与应用协议的传输帧绑定，在 HTTP 的下是绑定 HTTP 消息。协议绑定并不约束传输帧的使用方式，这意味着HTTP绑定可以与任何HTTP方法以及请求和响应消息一起使用。

CloudEvents例子

{
    "id": "abc-123",
    "source": "/users",
    "specversion": "1.0",
    "type": "user.created",
    "time": "2022-01-01T00:00:00Z",
    "data": {
        "userId": "123",
        "name": "Alice",
        "email": "[email protected]"
    }
}

• id: 事件的唯一标识符。
• source: 事件的来源URL。
• specversion: 使用的CloudEvents规范版本。
• type: 描述事件类型的字符串。在这个例子中，事件类型是’user.created’，它指示了一个新用户已经创建。
• time: 事件发生的时间戳。
• data: 事件的具体内容，包含有关新创建的用户的数据，如用户ID、姓名和电子邮件地址。

Knative Eventing介绍

CRDs

• Channel是事件的源头，它定义了事件的格式和传输协议。

  apiVersion: messaging.knative.dev/v1beta1
  kind: Channel
  metadata:
    name: my-channel
  spec:
    type: InMemory
    sink:
      ref:
        apiVersion: serving.knative.dev/v1
        kind: Service
        name: my-service

  请注意，这个例子使用了InMemory类型的Channel，如果您需要使用其他类型的Channel，可以在spec.type字段中指定。sink字段指定了Channel的目标地址，这里是一个名为my-service的Knative Service。

• Subscription定义了事件的订阅者，它可以订阅一个或多个Channel，并定义了事件的过滤规则。

  apiVersion: messaging.knative.dev/v1beta1
  kind: Subscription
  metadata:
    name: github-push-subscription
  spec:
    channel:
      apiVersion: messaging.knative.dev/v1beta1
      kind: InMemoryChannel
      name: github-push-channel
    subscriber:
      ref:
        apiVersion: v1
        kind: Service
        name: github-push-service
    filter:
      attributes:
        type: com.github.push

  这个例子定义了一个名为example-subscription的Subscription，它订阅了一个名为example-channel的InMemoryChannel，并将事件发送到一个名为example-service的Service。你可以根据自己的需要修改这个例子。

• Broker是事件的中间件，它负责将事件从Channel路由到Subscription。

  apiVersion: eventing.knative.dev/v1
  kind: Broker
  metadata:
    name: my-broker
  spec:
    config:
      apiVersion: v1
      kind: ConfigMapKeySelector
      name: config-br-defaults
      namespace: knative-eventing

• Trigger定义了事件的触发器，它可以根据一些条件来触发相应的操作，动作。

  apiVersion: eventing.knative.dev/v1
  kind: Trigger
  metadata:
    name: example-trigger
  spec:
    broker: default
    filter:
      attributes:
        type: dev.knative.samples.helloworld
    subscriber:
      ref:
        apiVersion: serving.knative.dev/v1
        kind: Service
        name: helloworld-go

  这个例子中，Trigger的名称为example-trigger，它订阅了default broker中的dev.knative.samples.helloworld类型的事件，并将其发送到名为helloworld-go的Knative服务。

• Source，KafkaBinding与KafkaSource，KafkaBinding定义了如何将事件发送到Kafka，而KafkaSource定义了如何从Kafka读取事件。当一个事件被发送到Kafka时，它会被保存在一个特定的topic中。KafkaSource会从这个topic中读取事件，并将它们转换为CloudEvents格式，然后将它们发送到HTTP sink中。

  apiVersion: sources.knative.dev/v1beta1
  kind: KafkaSource
  metadata:
    name: kafka-source
  spec:
    consumerGroup: knative-group
    bootstrapServers:
    - my-cluster-kafka-bootstrap.kafka:9092 # note the kafka namespace
    topics:
    - knative-demo-topic
    sink:
      ref:
        apiVersion: serving.knative.dev/v1
        kind: Service
        name: event-display

• SinkBinding用于将事件路由到特定的目标。它定义了一个事件源和一个或多个事件接收者之间的映射关系。当事件源发出事件时，SinkBinding将事件路由到与其关联的一个或多个事件接收者，这些事件接收者可以是Kubernetes服务、Knative服务或其他事件源。SinkBinding还可以定义过滤器和转换器，以便对事件进行处理和转换。通过使用SinkBinding，用户可以轻松地将事件路由到所需的目标，从而实现高效的事件驱动应用程序。

  apiVersion: sources.knative.dev/v1
  kind: SinkBinding
  metadata:
    name: my-sinkbinding
  spec:
    subject:
      apiVersion: v1
      kind: EventSource
      name: my-eventsource
    sink:
      apiVersion: serving.knative.dev/v1
      kind: Service
      name: my-knativeservice
    sink:
      apiVersion: v1
      kind: Service
      name: my-kubernetesservice

  在上面的定义中，我们指定了事件源“my-eventsource”，以及两个事件接收者：“my-knativeservice”和“my-kubernetesservice”。这意味着当事件源发出“order.created”事件时，事件将被路由到这两个服务中。

配合示例

一个名为my-channel的Channel，您可以使用以下命令创建它：

apiVersion: messaging.knative.dev/v1
kind: Channel
metadata:
  name: my-channel
spec:
  provisioner:
    apiVersion: messaging.knative.dev/v1
    kind: InMemoryChannel

然后，您可以使用以下命令创建一个名为my-broker的Broker，并将其与my-channel绑定：

apiVersion: eventing.knative.dev/v1
kind: Broker
metadata:
  name: my-broker
spec:
  channelTemplate:
    apiVersion: messaging.knative.dev/v1
    kind: Channel
    name: my-channel

上面是一个简单的例子，下面给出broker所有的配置示例：

apiVersion: eventing.knative.dev/v1
kind: Broker
metadata:
  name: default
  namespace: default
  annotations:
    eventing.knative.dev/broker.class: MTChannelBasedBroker
spec:
  config:
    apiVersion: v1
    kind: ConfigMap
    name: config-br-default-channel
    namespace: knative-eventing
  delivery:
    deadLetterSink:
      ref:
        kind: Service
        namespace: example-namespace
        name: example-service
        apiVersion: v1
      uri: example-uri
    retry: 5
    backoffPolicy: exponential
    backoffDelay: "2007-03-01T13:00:00Z/P1Y2M10DT2H30M"

现在，您可以使用以下命令创建一个名为my-trigger的Trigger，并将其与my-broker绑定：

apiVersion: eventing.knative.dev/v1
kind: Trigger
metadata:
  name: my-trigger
spec:
  broker: my-broker
  filter:
    attributes:
      type: com.example.eventType

这样，当您发送一个类型为com.example.eventType的事件时，它将被发送到my-channel，并通过my-broker和my-trigger进行传递。

curl -v "http://my-broker.default.svc.cluster.local" \
  -H "Content-Type: application/json" \
  -H "ce-specversion: 1.0" \
  -H "ce-type: com.example.eventType" \
  -H "ce-source: curl" \
  -H "ce-id: 1234" \
  -d '{"msg":"hello world"}'

附录：Apache EventMesh

Workflow (Base on ServerlessWorkflow Spec)

工作流设计参考

与MQ边界问题

与MQ中间件存在强依赖，边界不好划分

QA

• SinkBinding与Trigger有啥区别？

  SinkBinding和Trigger都是Knative Eventing的核心概念，它们的作用是将事件从事件源发送到目标服务。它们的区别在于：

  • SinkBinding是一种被动的事件路由机制，它仅在事件源发出事件时才会触发事件路由。SinkBinding需要先定义一个事件源和一个事件目标，当事件源发出事件时，事件会被路由到相应的事件目标。
  • Trigger是一种主动的事件路由机制，它可以通过定义触发规则来触发事件路由。Trigger需要先定义一个触发规则和一个事件目标，当触发规则被满足时，事件会被路由到相应的事件目标。

  因此，SinkBinding更适合被动接收事件的场景，而Trigger更适合主动触发事件的场景。同时，Trigger还可以根据条件进行过滤和转换事件，使得事件路由更加灵活和可控。