CloudEvents三部曲:实践篇

随着云原生的发展（云原生的下一个五年在哪里？），逐步进入深水区，业界须要一种统一的事件定义和描述规范，以提供跨服务、跨平台的交互能力。CloudEvents事件规范应运而生，并获得了行业的普遍关注，包括主要的云提供商和 SaaS 公司。 对于CloudEvent的介绍、规范说明及实践落地，将以三篇系列文章进行说明，本文为《CloudEvent三部曲:实践篇》

1、产品接入

场景

Serverless是一项基于事件驱动的函数计算服务，经过使用函数计算产品，函数以弹性、免运维、高可靠的方式运行，用户能够无需采购和维护服务器等基础设施，能够更加专一于函数代码的编写。 目前 CloudEvents 在函数计算中已有普遍的应用，第三方服务接入函数计算服务，须要使用符合 CloudEvents 规范的消息传递数据，方便函数计算平台方对第三方服务的消息进行分发过滤，同时因为规范的通用性，第三方服务一次改造后能够无缝适配到各种符合 CloudEvents 规范的平台上。 此外消息类产品（例如：消息队列，消息服务，事件总线等）也可经过支持 CloudEvents 规范，统一云的事件标准，加速云原生生态集成。

开发

一般状况下，要构建一个CloudEvent，须要使用CloudEvents的软件开发工具包（SDK），利用SDK能够极大方便开发人员进行集成开发，截至 CloudEvents v1.0 规范的发布，CloudEvents 团队支持和维护如下6种SDK：

• CSharp
• Go SDK
• Java SDK
• JavaScript SDK
• Python SDK
• Ruby SDK

如下使用 Go，Python SDK构造符合CloudEvent 1.0 规范的消息接收发送，HTTP/JSON请求转化等功能的范例。

Golang

1. 获取依赖

go get github.com/cloudevents/sdk-go/v2@v2.0.0

2. 依赖引用

import cloudevents "github.com/cloudevents/sdk-go/v2"

3. 发送事件

package main

import (
    "log"

	cloudevents "github.com/cloudevents/sdk-go/v2"
)

func main() {
	// The default client is HTTP.
	c, err := cloudevents.NewDefaultClient()
	if err != nil {
		log.Fatalf("failed to create client, %v", err)
	}

	// Create an Event.
	event :=  cloudevents.NewEvent()
	event.SetSource("example/uri")
	event.SetType("example.type")
	event.SetData(cloudevents.ApplicationJSON, map[string]string{"hello": "world"})

	// Set a target.
	ctx := cloudevents.ContextWithTarget(context.Background(), "http://localhost:8080/")

	// Send that Event.
	if result := c.Send(ctx, event); !cloudevents.IsACK(result) {
		log.Fatalf("failed to send, %v", result)
	}
}
复制代码

4. 接受事件

package main

import (
    "log"

	cloudevents "github.com/cloudevents/sdk-go/v2"
)

func receive(event cloudevents.Event) {
	// do something with event.
    fmt.Printf("%s", event)
}

func main() {
	// The default client is HTTP.
	c, err := cloudevents.NewDefaultClient()
	if err != nil {
		log.Fatalf("failed to create client, %v", err)
	}
	log.Fatal(c.StartReceiver(context.Background(), receive));
}
复制代码

5. 序列化

序列化为JSON

event := cloudevents.NewEvent()
event.SetSource("example/uri")
event.SetType("example.type")
event.SetData(cloudevents.ApplicationJSON, map[string]string{"hello": "world"})

bytes, err := json.Marshal(event)
复制代码

反序列化

event :=  cloudevents.NewEvent()
err := json.Marshal(bytes, &event)
复制代码

Python

1. 依赖包安装

pip install cloudevents

2. 事件发送者

经过 Python SDK 中的 CloudEvent 类型构造 CloudEvents 事件，再利用 to_binary函数将其序列化为 JSON 格式的数据，使用 requests框架发送该 JSON 请求。

from cloudevents.http import CloudEvent, to_binary
import requests

# 构建CloudEvent结构体
# - The CloudEvent "id" is generated if omitted. "specversion" defaults to "1.0".
attributes = {
    "type": "com.example.sampletype1",
    "source": "https://example.com/event-producer",
}
data = {"message": "Hello World!"}
event = CloudEvent(attributes, data)

# 利用to_binary函数将其序列化为 JSON 格式的数据
headers, body = to_binary(event)

# POST
requests.post("<some-url>", data=body, headers=headers)
复制代码

3. 接受事件处理

经过 Python SDK 中的 from_http 函数解析出 CloudEvents 事件，并打印事件内容

from flask import Flask, request

from cloudevents.http import from_http

app = Flask(__name__)


# create an endpoint at http://localhost:/3000/
@app.route("/", methods=["POST"])
def home():
    # create a CloudEvent
    event = from_http(request.headers, request.get_data())

    # you can access cloudevent fields as seen below
    print(
        f"Found {event['id']} from {event['source']} with type "
        f"{event['type']} and specversion {event['specversion']}"
    )

    return "", 204


if __name__ == "__main__":
    app.run(port=3000)
复制代码

---

2、接入方式

架构

基于事件驱动服务是函数计算的核心功能之一，平台使用 Knative Eventing 的Broker/Trigger 事件处理模型对事件进行过滤分发，此外为了确保跨平台和互操做性，将采用CNCF定义的标准数据格式CloudEvents 进行事件传输。

如上图所示，架构分为三块内容，从左到右分别为事件源，事件接收/转发者，事件消费者。

1. 事件源

事件源是一种 Kubernetes 定制资源，它提供了一种机制，用于注册来自特定服务系统的一类事件。例如：对象存储事件源，Github事件源等等，所以不一样的事件源须要的不一样的自定义资源进行描述。

事件源负责获取特定服务系统的事件，并将事件转化为CloudEvents格式事件发送给 Knative Eventing 平台（即 Broker/Trigger事件处理模型）。

2. 事件接受/转发者

引入Broker/Trigger事件处理模型的目的，是为了搭建一些黑盒子，将具体的实现隐藏起来，用户无需关心底层实现细节。

• Broker如同事件桶，接收各类不一样的事件，这些事件能够经过属性来过滤。
• Trigger描述了一个过滤器，只有经过了过滤器选择的事件，能够被传送给事件消费者。

如图1所示，用户经过 filter指定感兴趣红色小球的事件，最终只有该类事件被传送给事件消费者（这里是Knative Service，即 KSvc函数）。

3. 事件消费者

事件消费者能够是某个服务或系统，这里的事件消费者是用户编写的KSvc函数（即处理事件的逻辑代码）。

实现方式

1. 第三方接入

第三方服务接入基于knative实现的serverless平台须要提供特定的事件源，Knative社区已维护部分事件源，具体列表请查看：github.com/knative/eve…

若是第三方服务不在社区提供的支持列表中，就须要自定义事件源，有以下经常使用的几种方式：

ContainerSource 实现简单，是目前大部分自定义事件源的实现方式，也是KNative平台推荐的方式。

ContainerSource 是 Kubernetes 中自定义的 CRD（Custom Resource Definition）资源类型，具体定义以下

主要看如下几个部分：

1. sink：事件转发的目标对象，这里即图1中介绍的Borker
2. image：须要开发的镜像，包括了监听具体数据源的事件和转发事件到sink的实现
3. arg和env：开发者自定义的一些数据经过 arg 和 env 传入镜像

ContainerSource 中 image 镜像部分即须要自定义实现的部分，实现方式根据获取第三方服务事件的不一样分为如下两种：

1. 消息队列方式

以下图 2所示，若是第三方服务已适配消息队列，能够将产生的事件发往消息队列，此时 ContainerSource 能够直接从消息队列中消费第三方服务的事件。

2. 直连方式

以下图 3所示，若是第三方服务未适配消息队列，但服务自己提供事件订阅能力（如 Redis 的键空间特性，Keyspace Notifications future），此时 ContainerSource 能够直接订阅第三方服务的事件，监听服务变化。

注意：不管采用以上哪一种方式，ContainerSource 在生成 CloudEvents 事件时，都须要携带 KSVC 目标函数的惟一标识，以供平台侧分发事件时使用。例如：1. 消息队列方式，因为全部事件都从同一个消息队列中获取，此时须要在第三方服务生产事件时就携带目标函数的标识（对象存储产品接入时采用该方式）；2. 直连方式，因为 ContainerSource 与第三方服务是一对一关系，因此能够在 ContainerSource 生成 CloudEvents 事件时添加目标函数的标识。

利用 Broker/Trigger 事件处理模型，极大简化了第三方服务接入函数计算的流程。不管使用消息队列方式仍是直连方式，产品侧只须要提供适配第三方服务的 ContainerSource 镜像，以供平台侧使用。

2. 平台侧纳管

平台侧的工做主要是纳管产品侧提供的 ContainerSource，并利用 Trigger 提供事件过滤的能力。

针对 ContainerSource 不一样的实现方式，平台侧工做内容也有所区别：

消息队列实现方式

平台侧会建立以下内容：

1. 一组相同的ContainerSource（用于高可用）
2. 一个 Broker 类型的资源，用于分发事件
3. 多个 Trigger 类型资源，用于事件过滤

平台侧会预先建立好 ContainerSource 和 Broker 资源，并提供纳管 Trigger 的增删改查接口用于事件过滤，此时 ContainerSource，Broker，Trigger 对应关系以下图所示：

直连方式

平台侧会建立以下内容：

1. 多个 ContainerSource 订阅监听不一样的服务实例
2. 一个 Broker 类型的资源，用于分发事件
3. 多个 Trigger 类型资源，用于事件过滤

平台侧会预先提供好 Broker 资源，并提供纳管 ContainerSource 和 Trigger 的增删改查接口，此时 ContainerSource，Broker，Trigger 对应关系以下图所示：

更多云原生相关内容，请查看公众号：DCOS

参考资料

• function
• go sdk
• sdk-go