5.3. 事件 sink
5.3.1. 事件 sink
创建事件源时，您可以指定事件从源发送到的接收器。sink 是一个可寻址或可调用的资源，可以从其他资源接收传入的事件。Knative 服务、频道和代理都是接收器示例。
可寻址的对象接收和确认通过 HTTP 发送的事件到其 status.address.url 字段中定义的地址。作为特殊情况，核心 Kubernetes Service 对象也履行可寻址的接口。
可调用的对象可以接收通过 HTTP 发送的事件并转换事件，并在 HTTP 响应中返回 0 或 1 新事件。这些返回的事件可能会象处理外部事件源中的事件一样进一步处理。
5.3.1.1. Knative CLI sink 标记
当使用 Knative (kn) CLI 创建事件源时，您可以使用 --sink 标志指定事件从该资源发送到的接收器。sink 可以是任何可寻址或可调用的资源，可以从其他资源接收传入的事件。
以下示例创建使用服务 http://event-display.svc.cluster.local 的接收器绑定作为接收器：
使用 sink 标记的命令示例
$ kn source binding create bind-heartbeat \
--namespace sinkbinding-example \
--subject "Job:batch/v1:app=heartbeat-cron" \
--sink http://event-display.svc.cluster.local \ 1
--ce-override "sink=bound"
- 1
http://event-display.svc.cluster.local中的svc确定接收器是一个 Knative 服务。其他默认的接收器前缀包括channel和broker。
您可以通过自定义 kn，配置哪些 CR 可在 Knative (kn) CLI 命令中使用 --sink 标记。
5.3.2. Kafka 接收器
如果集群管理员在集群中启用了 Kafka，则 Kafka sink 是事件 sink 类型。您可以使用 Kafka sink 直接从事件源发送到 Kafka 主题。
5.3.2.1. 使用 Kafka sink
您可以创建一个称为 Kafka sink 的事件 sink，用于将事件发送到 Kafka 主题。使用 YAML 文件创建 Knative 资源使用声明性 API，它允许您以声明性的方式描述应用程序，并以可重复的方式描述应用程序。默认情况下，Kafka sink 使用二进制内容模式，其效率比结构化模式更高效。要使用 YAML 创建 Kafka sink，您必须创建一个 YAML 文件来定义 KafkaSink 对象，然后使用 oc apply 命令应用它。
先决条件
-
在集群中安装了 OpenShift Serverless Operator、Knative Eventing 和
KnativeKafka自定义资源 (CR) 。 - 您已创建了一个项目，或者具有适当的角色和权限访问项目，以便在 OpenShift Container Platform 中创建应用程序和其他工作负载。
- 您可以访问 Red Hat AMQ Streams（Kafka）集群，该集群会生成您要导入的 Kafka 信息。
-
安装 OpenShift CLI (
oc) 。
流程
创建一个
KafkaSink对象定义作为一个 YAML 文件：Kafka sink YAML
apiVersion: eventing.knative.dev/v1alpha1 kind: KafkaSink metadata: name: <sink-name> namespace: <namespace> spec: topic: <topic-name> bootstrapServers: - <bootstrap-server>
要创建 Kafka sink，请应用
KafkaSinkYAML 文件：$ oc apply -f <filename>
配置事件源，以便在其 spec 中指定 sink:
连接到 API 服务器源的 Kafka sink 示例
apiVersion: sources.knative.dev/v1alpha2 kind: ApiServerSource metadata: name: <source-name> 1 namespace: <namespace> 2 spec: serviceAccountName: <service-account-name> 3 mode: Resource resources: - apiVersion: v1 kind: Event sink: ref: apiVersion: eventing.knative.dev/v1alpha1 kind: KafkaSink name: <sink-name> 4
5.3.2.2. 为 Kafka sink 配置安全性
Apache Kafka 客户端和服务器使用 传输层安全性 (TLS) 来加密 Knative 和 Kafka 之间的流量，以及用于身份验证。TLS 是 Knative Kafka 唯一支持的流量加密方法。
Apache Kafka 使用 简单身份验证和安全层 (SASL) 进行身份验证。如果在集群中使用 SASL 身份验证，用户则必须向 Knative 提供凭证才能与 Kafka 集群通信，否则无法生成或消耗事件。
先决条件
-
OpenShift Serverless Operator、Knative Eventing 和
KnativeKafka自定义资源（CR）已安装在 OpenShift Container Platform 集群中。 -
在
KnativeKafkaCR 中启用了 Kafka sink。 - 您已创建了一个项目，或者具有适当的角色和权限访问项目，以便在 OpenShift Container Platform 中创建应用程序和其他工作负载。
-
您有一个 Kafka 集群 CA 证书存储为一个
.pem文件。 -
您有一个 Kafka 集群客户端证书，并存储为
.pem文件的密钥。 -
已安装 OpenShift (
oc) CLI。 -
您已选择使用 SASL 机制，例如
PLAIN、SCRAM-SHA-256或SCRAM-SHA-512。
流程
在与
KafkaSink对象相同的命名空间中创建一个 secret：重要证书和密钥必须采用 PEM 格式。
对于使用 SASL 时没有加密的身份验证：
$ oc create secret -n <namespace> generic <secret_name> \ --from-literal=protocol=SASL_PLAINTEXT \ --from-literal=sasl.mechanism=<sasl_mechanism> \ --from-literal=user=<username> \ --from-literal=password=<password>
对于使用 TLS 的 SASL 和加密进行身份验证：
$ oc create secret -n <namespace> generic <secret_name> \ --from-literal=protocol=SASL_SSL \ --from-literal=sasl.mechanism=<sasl_mechanism> \ --from-file=ca.crt=<my_caroot.pem_file_path> \ 1 --from-literal=user=<username> \ --from-literal=password=<password>- 1
- 如果您使用公有云管理 Kafka 服务，可以省略
ca.crt来使用系统的根 CA，如用于 Apache Kafka 的 Red Hat OpenShift Streams。
使用 TLS 进行身份验证和加密：
$ oc create secret -n <namespace> generic <secret_name> \ --from-literal=protocol=SSL \ --from-file=ca.crt=<my_caroot.pem_file_path> \ 1 --from-file=user.crt=<my_cert.pem_file_path> \ --from-file=user.key=<my_key.pem_file_path>- 1
- 如果您使用公有云管理 Kafka 服务，可以省略
ca.crt来使用系统的根 CA，如用于 Apache Kafka 的 Red Hat OpenShift Streams。
创建或修改
KafkaSink对象，并在authspec 中添加对 secret 的引用：apiVersion: eventing.knative.dev/v1alpha1 kind: KafkaSink metadata: name: <sink_name> namespace: <namespace> spec: ... auth: secret: ref: name: <secret_name> ...应用
KafkaSink对象：$ oc apply -f <filename>
