Eventing

流程

1. 在 OpenShift Container Platform Web 控制台的 Administrator 视角中，导航到 Serverless → Eventing。
2. 在 Create 列表中，选择 Event Source。您将被定向到 Event Sources 页面。
3. 选择您要创建的事件源类型。

1. 以 YAML 文件形式,为事件源创建服务帐户、角色和角色绑定：

   apiVersion: v1
   kind: ServiceAccount
   metadata:
     name: events-sa
     namespace: default 1
   
   ---
   apiVersion: rbac.authorization.k8s.io/v1
   kind: Role
   metadata:
     name: event-watcher
     namespace: default 2
   rules:
     - apiGroups:
         - ""
       resources:
         - events
       verbs:
         - get
         - list
         - watch
   
   ---
   apiVersion: rbac.authorization.k8s.io/v1
   kind: RoleBinding
   metadata:
     name: k8s-ra-event-watcher
     namespace: default 3
   roleRef:
     apiGroup: rbac.authorization.k8s.io
     kind: Role
     name: event-watcher
   subjects:
     - kind: ServiceAccount
       name: events-sa
       namespace: default 4

   1 2 3 4

   将这个命名空间更改为已选择安装事件源的命名空间。

2. 应用 YAML 文件：

   $ oc apply -f <filename>

3. 在 Developer 视角中，导航到 +Add → Event Source。此时会显示 Event Sources 页面。
4. 可选：如果您的事件源有多个供应商，请从 Providers 列表中选择所需的供应商，以过滤供应商的可用事件源。
5. 选择 ApiServerSource，然后点 Create Event Source。此时会显示 Create Event Source 页面。

6. 使用 Form view 或 YAML view 配置 ApiServerSource 设置：

   注意

   您可以在 Form view 和 YAML view 间进行切换。在不同视图间切换时数据会被保留。

   1. 输入 v1 作为 APIVERSION 和 Event 作为 KIND。
   2. 为您创建的服务帐户选择 Service Account Name。
   3. 为事件源选择 Sink。Sink 可以是一个 资源，如频道、代理或服务，也可以是一个 URI。
7. 点 Create。

删除 API 服务器源

1. 导航到 Topology 视图。

2. 右键点击 API 服务器源并选择 Delete ApiServerSource。

   

1. 以 YAML 文件形式,为事件源创建服务帐户、角色和角色绑定：

   apiVersion: v1
   kind: ServiceAccount
   metadata:
     name: events-sa
     namespace: default 1
   
   ---
   apiVersion: rbac.authorization.k8s.io/v1
   kind: Role
   metadata:
     name: event-watcher
     namespace: default 2
   rules:
     - apiGroups:
         - ""
       resources:
         - events
       verbs:
         - get
         - list
         - watch
   
   ---
   apiVersion: rbac.authorization.k8s.io/v1
   kind: RoleBinding
   metadata:
     name: k8s-ra-event-watcher
     namespace: default 3
   roleRef:
     apiGroup: rbac.authorization.k8s.io
     kind: Role
     name: event-watcher
   subjects:
     - kind: ServiceAccount
       name: events-sa
       namespace: default 4

   1 2 3 4

   将这个命名空间更改为已选择安装事件源的命名空间。

2. 应用 YAML 文件：

   $ oc apply -f <filename>

3. 创建具有事件 sink 的 API 服务器源。在以下示例中，sink 是一个代理：

   $ kn source apiserver create <event_source_name> --sink broker:<broker_name> --resource "event:v1" --service-account <service_account_name> --mode Resource

4. 要检查 API 服务器源是否已正确设置，请创建一个 Knative 服务，在日志中转储传入的信息：

   $ kn service create <service_name> --image quay.io/openshift-knative/knative-eventing-sources-event-display:latest

5. 如果您使用代理作为事件 sink，请创建一个触发器将事件从 default 代理过滤到服务：

   $ kn trigger create <trigger_name> --sink ksvc:<service_name>

6. 通过在 default 命名空间中启动 pod 来创建事件：

   $ oc create deployment hello-node --image quay.io/openshift-knative/knative-eventing-sources-event-display:latest

7. 通过检查以下命令生成的输出来检查是否正确映射了控制器：

   $ kn source apiserver describe <source_name>

   输出示例

   Name:                mysource
   Namespace:           default
   Annotations:         sources.knative.dev/creator=developer, sources.knative.dev/lastModifier=developer
   Age:                 3m
   ServiceAccountName:  events-sa
   Mode:                Resource
   Sink:
     Name:       default
     Namespace:  default
     Kind:       Broker (eventing.knative.dev/v1)
   Resources:
     Kind:        event (v1)
     Controller:  false
   Conditions:
     OK TYPE                     AGE REASON
     ++ Ready                     3m
     ++ Deployed                  3m
     ++ SinkProvided              3m
     ++ SufficientPermissions     3m
     ++ EventTypesProvided        3m

1. 获取 pod:

   $ oc get pods

2. 查看 pod 的消息转储程序功能日志：

   $ oc logs $(oc get pod -o name | grep event-display) -c user-container

   输出示例

   ☁️  cloudevents.Event
   Validation: valid
   Context Attributes,
     specversion: 1.0
     type: dev.knative.apiserver.resource.update
     datacontenttype: application/json
     ...
   Data,
     {
       "apiVersion": "v1",
       "involvedObject": {
         "apiVersion": "v1",
         "fieldPath": "spec.containers{hello-node}",
         "kind": "Pod",
         "name": "hello-node",
         "namespace": "default",
          .....
       },
       "kind": "Event",
       "message": "Started container",
       "metadata": {
         "name": "hello-node.159d7608e3a3572c",
         "namespace": "default",
         ....
       },
       "reason": "Started",
       ...
     }

删除 API 服务器源

1. 删除触发器：

   $ kn trigger delete <trigger_name>

2. 删除事件源：

   $ kn source apiserver delete <source_name>

3. 删除服务帐户、集群角色和集群绑定：

   $ oc delete -f authentication.yaml

1. 以 YAML 文件形式,为事件源创建服务帐户、角色和角色绑定：

   apiVersion: v1
   kind: ServiceAccount
   metadata:
     name: events-sa
     namespace: default 1
   
   ---
   apiVersion: rbac.authorization.k8s.io/v1
   kind: Role
   metadata:
     name: event-watcher
     namespace: default 2
   rules:
     - apiGroups:
         - ""
       resources:
         - events
       verbs:
         - get
         - list
         - watch
   
   ---
   apiVersion: rbac.authorization.k8s.io/v1
   kind: RoleBinding
   metadata:
     name: k8s-ra-event-watcher
     namespace: default 3
   roleRef:
     apiGroup: rbac.authorization.k8s.io
     kind: Role
     name: event-watcher
   subjects:
     - kind: ServiceAccount
       name: events-sa
       namespace: default 4

   1 2 3 4

   将这个命名空间更改为已选择安装事件源的命名空间。

2. 应用 YAML 文件：

   $ oc apply -f <filename>

3. 将 API 服务器源创建为 YAML 文件：

   apiVersion: sources.knative.dev/v1alpha1
   kind: ApiServerSource
   metadata:
     name: testevents
   spec:
     serviceAccountName: events-sa
     mode: Resource
     resources:
       - apiVersion: v1
         kind: Event
     sink:
       ref:
         apiVersion: eventing.knative.dev/v1
         kind: Broker
         name: default

4. 应用 ApiServerSource YAML 文件：

   $ oc apply -f <filename>

5. 要检查 API 服务器源是否已正确设置，请创建一个 Knative 服务作为 YAML 文件，在日志中转储传入的信息：

   apiVersion: serving.knative.dev/v1
   kind: Service
   metadata:
     name: event-display
     namespace: default
   spec:
     template:
       spec:
         containers:
           - image: quay.io/openshift-knative/knative-eventing-sources-event-display:latest

6. 应用 Service YAML 文件：

   $ oc apply -f <filename>

7. 创建一个 Trigger 对象作为一个 YAML 文件，该文件将事件从 default 代理过滤到上一步中创建的服务：

   apiVersion: eventing.knative.dev/v1
   kind: Trigger
   metadata:
     name: event-display-trigger
     namespace: default
   spec:
     broker: default
     subscriber:
       ref:
         apiVersion: serving.knative.dev/v1
         kind: Service
         name: event-display

8. 应用 Trigger YAML 文件：

   $ oc apply -f <filename>

9. 通过在 default 命名空间中启动 pod 来创建事件：

   $ oc create deployment hello-node --image=quay.io/openshift-knative/knative-eventing-sources-event-display

10. 输入以下命令并检查输出，检查是否正确映射了控制器：

    $ oc get apiserversource.sources.knative.dev testevents -o yaml

    输出示例

    apiVersion: sources.knative.dev/v1alpha1
    kind: ApiServerSource
    metadata:
      annotations:
      creationTimestamp: "2020-04-07T17:24:54Z"
      generation: 1
      name: testevents
      namespace: default
      resourceVersion: "62868"
      selfLink: /apis/sources.knative.dev/v1alpha1/namespaces/default/apiserversources/testevents2
      uid: 1603d863-bb06-4d1c-b371-f580b4db99fa
    spec:
      mode: Resource
      resources:
      - apiVersion: v1
        controller: false
        controllerSelector:
          apiVersion: ""
          kind: ""
          name: ""
          uid: ""
        kind: Event
        labelSelector: {}
      serviceAccountName: events-sa
      sink:
        ref:
          apiVersion: eventing.knative.dev/v1
          kind: Broker
          name: default

1. 输入以下命令来获取 pod：

   $ oc get pods

2. 输入以下命令来查看 pod 的消息转储程序功能日志：

   $ oc logs $(oc get pod -o name | grep event-display) -c user-container

   输出示例

   ☁️  cloudevents.Event
   Validation: valid
   Context Attributes,
     specversion: 1.0
     type: dev.knative.apiserver.resource.update
     datacontenttype: application/json
     ...
   Data,
     {
       "apiVersion": "v1",
       "involvedObject": {
         "apiVersion": "v1",
         "fieldPath": "spec.containers{hello-node}",
         "kind": "Pod",
         "name": "hello-node",
         "namespace": "default",
          .....
       },
       "kind": "Event",
       "message": "Started container",
       "metadata": {
         "name": "hello-node.159d7608e3a3572c",
         "namespace": "default",
         ....
       },
       "reason": "Started",
       ...
     }

删除 API 服务器源

1. 删除触发器：

   $ oc delete -f trigger.yaml

2. 删除事件源：

   $ oc delete -f k8s-events.yaml

3. 删除服务帐户、集群角色和集群绑定：

   $ oc delete -f authentication.yaml

流程

1. 要验证 ping 源是否可以工作，请创建一个简单的 Knative 服务，在服务日志中转储传入的信息。

   1. 在 Developer 视角中，导航到 +Add → YAML。

   2. 复制 YAML 示例：

      apiVersion: serving.knative.dev/v1
      kind: Service
      metadata:
        name: event-display
      spec:
        template:
          spec:
            containers:
              - image: quay.io/openshift-knative/knative-eventing-sources-event-display:latest

   3. 点 Create。

2. 在与上一步中创建的服务相同的命名空间中创建一个 ping 源，或您要将事件发送到的任何其他接收器。

   1. 在 Developer 视角中，导航到 +Add → Event Source。此时会显示 Event Sources 页面。
   2. 可选：如果您的事件源有多个供应商，请从 Providers 列表中选择所需的供应商，以过滤供应商的可用事件源。

   3. 选择 Ping Source，然后点击 Create Event Source。此时会显示 Create Event Source 页面。

      注意

      您可以使用 Form view 或 YAML view 配置 PingSource 设置，并可以在两者间切换。在不同视图间切换时数据会被保留。

   4. 为 Schedule 输入一个值。在本例中，值为 */2 * * *，它会创建一个 PingSource，每两分钟发送一条消息。
   5. 可选：您可以为 Data 输入一个值，它是消息的有效负载。
   6. 选择一个 Sink。这可以是 Resource 或一个 URI。在这个示例中，上一步中创建的 event-display 服务被用作 Resource sink。
   7. 点 Create。

1. 在 Developer 视角中，导航到 Topology。

2. 查看 ping 源和接收器。

   

删除 ping 源

1. 导航到 Topology 视图。
2. 右键单击 API 服务器源，再选择 Delete Ping Source。

流程

1. 要验证 ping 源是否可以工作，请创建一个简单的 Knative 服务，在服务日志中转储传入的信息：

   $ kn service create event-display \
       --image quay.io/openshift-knative/knative-eventing-sources-event-display:latest

2. 对于您要请求的每一组 ping 事件，请在与事件消费者相同的命名空间中创建一个 ping 源：

   $ kn source ping create test-ping-source \
       --schedule "*/2 * * * *" \
       --data '{"message": "Hello world!"}' \
       --sink ksvc:event-display

3. 输入以下命令并检查输出，检查是否正确映射了控制器：

   $ kn source ping describe test-ping-source

   输出示例

   Name:         test-ping-source
   Namespace:    default
   Annotations:  sources.knative.dev/creator=developer, sources.knative.dev/lastModifier=developer
   Age:          15s
   Schedule:     */2 * * * *
   Data:         {"message": "Hello world!"}
   
   Sink:
     Name:       event-display
     Namespace:  default
     Resource:   Service (serving.knative.dev/v1)
   
   Conditions:
     OK TYPE                 AGE REASON
     ++ Ready                 8s
     ++ Deployed              8s
     ++ SinkProvided         15s
     ++ ValidSchedule        15s
     ++ EventTypeProvided    15s
     ++ ResourcesCorrect     15s

1. 查看新创建的 pod：

   $ watch oc get pods

2. 使用 Ctrl+C 取消查看 pod，然后查看所创建 pod 的日志：

   $ oc logs $(oc get pod -o name | grep event-display) -c user-container

   输出示例

   ☁️  cloudevents.Event
   Validation: valid
   Context Attributes,
     specversion: 1.0
     type: dev.knative.sources.ping
     source: /apis/v1/namespaces/default/pingsources/test-ping-source
     id: 99e4f4f6-08ff-4bff-acf1-47f61ded68c9
     time: 2020-04-07T16:16:00.000601161Z
     datacontenttype: application/json
   Data,
     {
       "message": "Hello world!"
     }

流程

1. 要验证 ping 源是否可以工作，请创建一个简单的 Knative 服务，在服务日志中转储传入的信息。

   1. 创建服务 YAML 文件：

      apiVersion: serving.knative.dev/v1
      kind: Service
      metadata:
        name: event-display
      spec:
        template:
          spec:
            containers:
              - image: quay.io/openshift-knative/knative-eventing-sources-event-display:latest

   2. 创建服务：

      $ oc apply -f <filename>

2. 对于您要请求的每一组 ping 事件，请在与事件消费者相同的命名空间中创建一个 ping 源。

   1. 为 ping 源创建 YAML 文件：

      apiVersion: sources.knative.dev/v1
      kind: PingSource
      metadata:
        name: test-ping-source
      spec:
        schedule: "*/2 * * * *"
        data: '{"message": "Hello world!"}'
        sink:
          ref:
            apiVersion: serving.knative.dev/v1
            kind: Service
            name: event-display

   2. 创建 ping 源：

      $ oc apply -f <filename>

3. 输入以下命令检查是否正确映射了控制器：

   $ oc get pingsource.sources.knative.dev <ping_source_name> -oyaml

   输出示例

   apiVersion: sources.knative.dev/v1
   kind: PingSource
   metadata:
     annotations:
       sources.knative.dev/creator: developer
       sources.knative.dev/lastModifier: developer
     creationTimestamp: "2020-04-07T16:11:14Z"
     generation: 1
     name: test-ping-source
     namespace: default
     resourceVersion: "55257"
     selfLink: /apis/sources.knative.dev/v1/namespaces/default/pingsources/test-ping-source
     uid: 3d80d50b-f8c7-4c1b-99f7-3ec00e0a8164
   spec:
     data: '{ value: "hello" }'
     schedule: '*/2 * * * *'
     sink:
       ref:
         apiVersion: serving.knative.dev/v1
         kind: Service
         name: event-display
         namespace: default

1. 查看新创建的 pod：

   $ watch oc get pods

2. 使用 Ctrl+C 取消查看 pod，然后查看所创建 pod 的日志：

   $ oc logs $(oc get pod -o name | grep event-display) -c user-container

   输出示例

   ☁️  cloudevents.Event
   Validation: valid
   Context Attributes,
     specversion: 1.0
     type: dev.knative.sources.ping
     source: /apis/v1/namespaces/default/pingsources/test-ping-source
     id: 042ff529-240e-45ee-b40c-3a908129853e
     time: 2020-04-07T16:22:00.000791674Z
     datacontenttype: application/json
   Data,
     {
       "message": "Hello world!"
     }

流程

1. 在 Developer 视角中，进入到 +Add 页面并选择 Event Source。
2. 在 Event Sources 页面中，在 Type 部分选择 Kafka Source。

3. 配置 Kafka Source 设置：

   1. 添加用逗号分开的 Bootstrap 服务器列表。
   2. 添加以逗号分隔的标题列表。
   3. 添加一个 消费者组。
   4. 为您创建的服务帐户选择 Service Account Name。
   5. 为事件源选择 Sink。Sink 可以是一个 资源，如频道、代理或服务，也可以是一个 URI。
   6. 输入 Kafka 事件源的 名称。
4. 点 Create。

1. 在 Developer 视角中，导航到 Topology。

2. 查看 Kafka 事件源和接收器。

   

流程

1. 要验证 Kafka 事件源是否可以工作，请创建一个 Knative 服务，在服务日志中转储传入的事件：

   $ kn service create event-display \
       --image quay.io/openshift-knative/knative-eventing-sources-event-display

2. 创建 KafkaSource CR：

   $ kn source kafka create <kafka_source_name> \
       --servers <cluster_kafka_bootstrap>.kafka.svc:9092 \
       --topics <topic_name> --consumergroup my-consumer-group \
       --sink event-display

   注意

   将此命令中的占位符值替换为源名称、引导服务器和主题的值。

   --servers、--topics 和 --consumergroup 选项指定到 Kafka 集群的连接参数。--consumergroup 选项是可选的。

3. 可选：查看您创建的 KafkaSource CR 的详情：

   $ kn source kafka describe <kafka_source_name>

   输出示例

   Name:              example-kafka-source
   Namespace:         kafka
   Age:               1h
   BootstrapServers:  example-cluster-kafka-bootstrap.kafka.svc:9092
   Topics:            example-topic
   ConsumerGroup:     example-consumer-group
   
   Sink:
     Name:       event-display
     Namespace:  default
     Resource:   Service (serving.knative.dev/v1)
   
   Conditions:
     OK TYPE            AGE REASON
     ++ Ready            1h
     ++ Deployed         1h
     ++ SinkProvided     1h

验证步骤

1. 触发 Kafka 实例将信息发送到主题：

   $ oc -n kafka run kafka-producer \
       -ti --image=quay.io/strimzi/kafka:latest-kafka-2.7.0 --rm=true \
       --restart=Never -- bin/kafka-console-producer.sh \
       --broker-list <cluster_kafka_bootstrap>:9092 --topic my-topic

   在提示符后输入信息。这个命令假设：

   • Kafka 集群安装在 kafka 命名空间中。
   • KafkaSource 对象已被配置为使用 my-topic 主题。

2. 通过查看日志来验证消息是否显示：

   $ oc logs $(oc get pod -o name | grep event-display) -c user-container

   输出示例

   ☁️  cloudevents.Event
   Validation: valid
   Context Attributes,
     specversion: 1.0
     type: dev.knative.kafka.event
     source: /apis/v1/namespaces/default/kafkasources/example-kafka-source#example-topic
     subject: partition:46#0
     id: partition:46/offset:0
     time: 2021-03-10T11:21:49.4Z
   Extensions,
     traceparent: 00-161ff3815727d8755848ec01c866d1cd-7ff3916c44334678-00
   Data,
     Hello!

流程

1. 创建 KafkaSource 对象作为 YAML 文件：

   apiVersion: sources.knative.dev/v1beta1
   kind: KafkaSource
   metadata:
     name: <source_name>
   spec:
     consumerGroup: <group_name> 1
     bootstrapServers:
     - <list_of_bootstrap_servers>
     topics:
     - <list_of_topics> 2
     sink:
     - <list_of_sinks> 3

   1

   用户组是一组使用相同组群 ID 的用户，并消耗一个标题中的数据。

   2

   主题提供数据存储的目的地。每个主题都被分成一个或多个分区。

   3

   sink 指定事件从源发送到的位置。

   重要

   仅支持 OpenShift Serverless 上的 KafkaSource 对象的 v1beta1 API 版本。不要使用这个 API 的 v1alpha1 版本，因为这个版本现已弃用。

   KafkaSource 对象示例

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

2. 应用 KafkaSource YAML 文件：

   $ oc apply -f <filename>

流程

1. 在所选命名空间中创建证书文件作为 secret：

   $ oc create secret -n <namespace> generic <kafka_auth_secret> \
     --from-file=ca.crt=caroot.pem \
     --from-literal=password="SecretPassword" \
     --from-literal=saslType="SCRAM-SHA-512" \ 1
     --from-literal=user="my-sasl-user"

   1

   SASL 类型可以是 PLAIN、SCRAM-SHA-256 或 SCRAM-SHA-512。

2. 创建或修改 Kafka 源，使其包含以下 spec 配置：

   apiVersion: sources.knative.dev/v1beta1
   kind: KafkaSource
   metadata:
     name: example-source
   spec:
   ...
     net:
       sasl:
         enable: true
         user:
           secretKeyRef:
             name: <kafka_auth_secret>
             key: user
         password:
           secretKeyRef:
             name: <kafka_auth_secret>
             key: password
         type:
           secretKeyRef:
             name: <kafka_auth_secret>
             key: saslType
       tls:
         enable: true
         caCert: 1
           secretKeyRef:
             name: <kafka_auth_secret>
             key: ca.crt
   ...

   1

   如果您使用公共云 Kafka 服务，则不需要 caCert spec。

流程

1. 进入 +Add → Event Source 并选择您要创建的事件源类型，创建任何类型的事件源。
2. 在 Create Event Source 表单视图的 Sink 部分，在 Resource 列表中选择您的接收器。
3. 点 Create。

1. 在 Developer 视角中，导航到 Topology。
2. 查看事件源并点连接的接收器查看右侧面板中的接收器详情。

流程

1. 创建一个 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>

2. 要创建 Kafka sink，请应用 KafkaSink YAML 文件：

   $ oc apply -f <filename>

3. 配置事件源，以便在其 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

   1

   事件源的名称。

   2

   事件源的命名空间。

   3

   事件源的服务帐户。

   4

   Kafka sink 名称。

流程

1. 在 Developer 视角中，进入到 +Add 视图。
2. 点 Eventing 目录中的 Event Sink。
3. 在目录项中搜索 KafkaSink 并点它。
4. 点 Create Event Sink。

5. 在表单视图中，键入 bootstrap 服务器的 URL，这是主机名和端口的组合。

   
6. 键入要发送事件数据的主题名称。
7. 键入事件 sink 的名称。
8. 点 Create。

验证

1. 在 Developer 视角中，进入 Topology 视图。
2. 点创建的事件 sink 在右侧面板中查看其详情。

流程

1. 在与 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 来使用系统的 root CA 设置。

   • 使用 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 来使用系统的 root CA 设置。

2. 创建或修改 KafkaSink 对象，并在 auth spec 中添加对 secret 的引用：

   apiVersion: eventing.knative.dev/v1alpha1
   kind: KafkaSink
   metadata:
      name: <sink_name>
      namespace: <namespace>
   spec:
   ...
      auth:
        secret:
          ref:
            name: <secret_name>
   ...

3. 应用 KafkaSink 对象：

   $ oc apply -f <filename>

验证

1. 使用 kn 命令列出所有现有代理：

   $ kn broker list

   输出示例

   NAME      URL                                                                     AGE   CONDITIONS   READY   REASON
   default   http://broker-ingress.knative-eventing.svc.cluster.local/test/default   45s   5 OK / 5     True

2. 可选：如果使用 OpenShift Container Platform Web 控制台，在 Developer 视角中进入 Topology 视图来查看存在的代理：

   

流程

1. 创建一个 Trigger 对象作为 YAML 文件，该文件带有 eventing.knative.dev/injection: enabled 注解：

   apiVersion: eventing.knative.dev/v1
   kind: Trigger
   metadata:
     annotations:
       eventing.knative.dev/injection: enabled
     name: <trigger_name>
   spec:
     broker: default
     subscriber: 1
       ref:
         apiVersion: serving.knative.dev/v1
         kind: Service
         name: <service_name>

   1

   指定触发器将事件发送到的事件 sink 或 subscriber。

2. 应用 Trigger YAML 文件：

   $ oc apply -f <filename>

1. 输入以下 oc 命令来获取代理：

   $ oc -n <namespace> get broker default

   输出示例

   NAME      READY     REASON    URL                                                                     AGE
   default   True                http://broker-ingress.knative-eventing.svc.cluster.local/test/default   3m56s

2. 可选：如果使用 OpenShift Container Platform Web 控制台，在 Developer 视角中进入 Topology 视图来查看存在的代理：

   

1. 使用 oc 命令获取代理：

   $ oc -n <namespace> get broker <broker_name>

   示例命令

   $ oc -n default get broker default

   输出示例

   NAME      READY     REASON    URL                                                                     AGE
   default   True                http://broker-ingress.knative-eventing.svc.cluster.local/test/default   3m56s

2. 可选：如果使用 OpenShift Container Platform Web 控制台，在 Developer 视角中进入 Topology 视图来查看存在的代理：

   

流程

1. 从命名空间中删除 eventing.knative.dev/injection=enabled 标识：

   $ oc label namespace <namespace> eventing.knative.dev/injection-

   移除注解可防止 Knative 在删除代理后重新创建代理。

2. 从所选命名空间中删除代理：

   $ oc -n <namespace> delete broker <broker_name>

流程

1. 在 Developer 视角中，进入到 +Add → Broker。此时会显示 Broker 页面。
2. 可选。更新代理的名称。如果您没有更新名称，则生成的代理名为 default。
3. 点 Create。

1. 在 Developer 视角中，导航到 Topology。

2. 查看 mt-broker-ingress、mt-broker-filter 和 mt-broker-controller 组件。

   

流程

1. 在 OpenShift Container Platform Web 控制台的 Administrator 视角中，导航到 Serverless → Eventing。
2. 在 Create 列表中，选择 Broker。您将进入 Create Broker 页面。
3. 可选：修改代理的 YAML 配置。
4. 点 Create。

流程

1. 修改 kafka-broker-config 配置映射，或创建自己的配置映射来包含以下配置：

   apiVersion: v1
   kind: ConfigMap
   metadata:
     name: <config_map_name> 1
     namespace: <namespace> 2
   data:
     default.topic.partitions: <integer> 3
     default.topic.replication.factor: <integer> 4
     bootstrap.servers: <list_of_servers> 5

   1

   配置映射名称。

   2

   配置映射所在的命名空间。

   3

   Kafka 代理的主题分区数量。这控制如何将事件发送到代理的速度。更多分区需要更多计算资源。

   4

   主题消息的复制因素。这可防止数据丢失。更高的复制因素需要更大的计算资源和更多存储。

   5

   以逗号分隔的 bootstrap 服务器列表。这可以位于 OpenShift Container Platform 集群内部或外部，是代理从发送事件发送到的 Kafka 集群列表。

   重要

   default.topic.replication.factor 值必须小于或等于集群中的 Kafka 代理实例数量。例如，如果您只有一个 Kafka 代理，则 default.topic.replication.factor 值应该不超过 "1"。

   Kafka 代理配置映射示例

   apiVersion: v1
   kind: ConfigMap
   metadata:
     name: kafka-broker-config
     namespace: knative-eventing
   data:
     default.topic.partitions: "10"
     default.topic.replication.factor: "3"
     bootstrap.servers: "my-cluster-kafka-bootstrap.kafka:9092"

2. 应用配置映射：

   $ oc apply -f <config_map_filename>

3. 指定 Kafka Broker 对象的配置映射：

   Broker 对象示例

   apiVersion: eventing.knative.dev/v1
   kind: Broker
   metadata:
     name: <broker_name> 1
     namespace: <namespace> 2
     annotations:
       eventing.knative.dev/broker.class: Kafka 3
   spec:
     config:
       apiVersion: v1
       kind: ConfigMap
       name: <config_map_name> 4
       namespace: <namespace> 5
   ...

   1

   代理名称。

   2

   代理存在的命名空间。

   3

   代理类注解。在本例中，代理是使用类值 Kafka 的 Kafka 代理。

   4

   配置映射名称。

   5

   配置映射所在的命名空间。

4. 应用代理：

   $ oc apply -f <broker_filename>

流程

1. 创建或修改 Trigger 对象并设置 kafka.eventing.knative.dev/delivery.order 注解：

   apiVersion: eventing.knative.dev/v1
   kind: Trigger
   metadata:
     name: <trigger_name>
     annotations:
        kafka.eventing.knative.dev/delivery.order: ordered
   ...

   支持的消费者交付保证有：

   unordered

   未排序的消费者是一种非阻塞消费者，它能以未排序的方式提供消息，同时保持正确的偏移管理。

   排序的

   一个订购的消费者是一个按分区阻止消费者，在提供分区的下一个消息前等待来自 CloudEvent 订阅者成功响应。

   默认排序保证是 unordered。

2. 应用 Trigger 对象：

   $ oc apply -f <filename>

流程

1. 创建或修改 Trigger 对象并设置 kafka.eventing.knative.dev/delivery.order 注解：

   apiVersion: eventing.knative.dev/v1
   kind: Trigger
   metadata:
     name: <trigger_name>
     annotations:
        kafka.eventing.knative.dev/delivery.order: ordered
   ...

   支持的消费者交付保证有：

   unordered

   未排序的消费者是一种非阻塞消费者，它能以未排序的方式提供消息，同时保持正确的偏移管理。

   排序的

   一个订购的消费者是一个按分区阻止消费者，在提供分区的下一个消息前等待来自 CloudEvent 订阅者成功响应。

   默认排序保证是 unordered。

2. 应用 Trigger 对象：

   $ oc apply -f <filename>

流程

1. 在 OpenShift Container Platform Web 控制台的 Administrator 视角中，导航到 Serverless → Eventing。
2. 在 Broker 选项卡中，为您要在其中添加触发器的代理选择 Options 菜单 。
3. 点列表中的 Add Trigger。
4. 在 Add Trigger 对话框中，为触发器选择 Subscriber。订阅者是可以从代理接收事件的 Knative 服务。
5. 点 Add。

流程

1. 在 Developer 视角中，进入 Topology 页。
2. 将鼠标悬停在您要创建触发器的代理上，并拖动箭头。此时会显示 Add Trigger 选项。
3. 点 Add Trigger。
4. 在 Subscriber 列表中选择您的接收器。
5. 点 Add。

删除触发器

1. 在 Developer 视角中，进入 Topology 页。
2. 点您要删除的触发器。
3. 在 Actions 上下文菜单中，选择 Delete Trigger。

流程

1. 显示可用触发器列表：

   $ kn trigger list

   输出示例

   NAME    BROKER    SINK           AGE   CONDITIONS   READY   REASON
   email   default   ksvc:edisplay   4s    5 OK / 5     True
   ping    default   ksvc:edisplay   32s   5 OK / 5     True

2. 可选：以 JSON 格式输出触发器列表：

   $ kn trigger list -o json

验证

1. 列出现有触发器：

   $ kn trigger list

2. 验证触发器不再存在：

   输出示例

   No triggers found.

流程

1. 在 OpenShift Container Platform Web 控制台的 Administrator 视角中，导航到 Serverless → Eventing。
2. 在 Create 列表中，选择 Channel。您将被定向到 Channel 页。

3. 选择您要在 Type 列表中创建的 Channel 对象类型。

   注意

   目前默认只支持 InMemoryChannel 频道对象。如果您在 OpenShift Serverless 上安装了 Apache Kafka 的 Knative 代理实现，则 Apache Kafka 的 Knative 频道可用。

4. 点 Create。

流程

1. 在 Developer 视角中，导航到 +Add → Channel。
2. 选择您要在 Type 列表中创建的 Channel 对象类型。
3. 点 Create。

流程

1. 创建一个 Channel 对象作为一个 YAML 文件：

   apiVersion: messaging.knative.dev/v1
   kind: Channel
   metadata:
     name: example-channel
     namespace: default

2. 应用 YAML 文件：

   $ oc apply -f <filename>

流程

1. 创建一个 KafkaChannel 对象作为一个 YAML 文件：

   apiVersion: messaging.knative.dev/v1beta1
   kind: KafkaChannel
   metadata:
     name: example-channel
     namespace: default
   spec:
     numPartitions: 3
     replicationFactor: 1

   重要

   仅支持 OpenShift Serverless 上的 KafkaChannel 对象的 v1beta1 API 版本。不要使用这个 API 的 v1alpha1 版本，因为这个版本现已弃用。

2. 应用 KafkaChannel YAML 文件：

   $ oc apply -f <filename>

流程

1. 在 Developer 视角中，进入 Topology 页。

2. 使用以下方法之一创建订阅：

   1. 将鼠标悬停在您要为其创建订阅的频道上，并拖动箭头。此时会显示 Add Subscription 选项。

      
      1. 在 Subscriber 列表中选择您的接收器。
      2. 点 Add。
   2. 如果服务在与频道相同的命名空间或项目下的 Topology 视图中可用，点击您要为该频道创建订阅的频道，并将箭头直接拖到服务以立即从频道创建订阅到该服务。

流程

1. 在 OpenShift Container Platform Web 控制台的 Administrator 视角中，导航到 Serverless → Eventing。
2. 在 Channel 选项卡中，选择您要在其中添加订阅的频道的 Options 菜单 。
3. 点击列表中的 Add Subscription。
4. 在 Add Subscription 对话框中，为订阅选择 Subscriber。订阅者是可以从频道接收事件的 Knative 服务。
5. 点 Add。

流程

1. 在所选命名空间中创建证书文件作为 secret：

   $ oc create secret -n <namespace> generic <kafka_auth_secret> \
     --from-file=ca.crt=caroot.pem \
     --from-file=user.crt=certificate.pem \
     --from-file=user.key=key.pem

   重要

   使用密钥名称 ca.crt、user.crt 和 user.key。不要更改它们。

2. 编辑 KnativeKafka 自定义资源：

   $ oc edit knativekafka

3. 引用您的 secret 和 secret 的命名空间：

   apiVersion: operator.serverless.openshift.io/v1alpha1
   kind: KnativeKafka
   metadata:
     namespace: knative-eventing
     name: knative-kafka
   spec:
     channel:
       authSecretName: <kafka_auth_secret>
       authSecretNamespace: <kafka_auth_secret_namespace>
       bootstrapServers: <bootstrap_servers>
       enabled: true
     source:
       enabled: true

   注意

   确保指定 bootstrap 服务器中的匹配端口。

   例如：

   apiVersion: operator.serverless.openshift.io/v1alpha1
   kind: KnativeKafka
   metadata:
     namespace: knative-eventing
     name: knative-kafka
   spec:
     channel:
       authSecretName: tls-user
       authSecretNamespace: kafka
       bootstrapServers: eventing-kafka-bootstrap.kafka.svc:9094
       enabled: true
     source:
       enabled: true

流程

1. 在所选命名空间中创建证书文件作为 secret：

   $ oc create secret -n <namespace> generic <kafka_auth_secret> \
     --from-file=ca.crt=caroot.pem \
     --from-literal=password="SecretPassword" \
     --from-literal=saslType="SCRAM-SHA-512" \
     --from-literal=user="my-sasl-user"

   • 使用键名 ca.crt, password, 和 sasl.mechanism.不要更改它们。

   • 如果要将 SASL 与公共 CA 证书搭配使用，您必须在创建 secret 时使用 tls.enabled=true 标志，而不是使用 ca.crt 参数。例如：

     $ oc create secret -n <namespace> generic <kafka_auth_secret> \
       --from-literal=tls.enabled=true \
       --from-literal=password="SecretPassword" \
       --from-literal=saslType="SCRAM-SHA-512" \
       --from-literal=user="my-sasl-user"

2. 编辑 KnativeKafka 自定义资源：

   $ oc edit knativekafka

3. 引用您的 secret 和 secret 的命名空间：

   apiVersion: operator.serverless.openshift.io/v1alpha1
   kind: KnativeKafka
   metadata:
     namespace: knative-eventing
     name: knative-kafka
   spec:
     channel:
       authSecretName: <kafka_auth_secret>
       authSecretNamespace: <kafka_auth_secret_namespace>
       bootstrapServers: <bootstrap_servers>
       enabled: true
     source:
       enabled: true

   注意

   确保指定 bootstrap 服务器中的匹配端口。

   例如：

   apiVersion: operator.serverless.openshift.io/v1alpha1
   kind: KnativeKafka
   metadata:
     namespace: knative-eventing
     name: knative-kafka
   spec:
     channel:
       authSecretName: scram-user
       authSecretNamespace: kafka
       bootstrapServers: eventing-kafka-bootstrap.kafka.svc:9093
       enabled: true
     source:
       enabled: true

流程

1. 在 OpenShift Container Platform Web 控制台的 Administrator 视角中，导航到 Serverless → Eventing。
2. 在 Channel 选项卡中，选择您要在其中添加订阅的频道的 Options 菜单 。
3. 点击列表中的 Add Subscription。
4. 在 Add Subscription 对话框中，为订阅选择 Subscriber。订阅者是可以从频道接收事件的 Knative 服务。
5. 点 Add。

流程

1. 在 Developer 视角中，进入 Topology 页。

2. 使用以下方法之一创建订阅：

   1. 将鼠标悬停在您要为其创建订阅的频道上，并拖动箭头。此时会显示 Add Subscription 选项。

      
      1. 在 Subscriber 列表中选择您的接收器。
      2. 点 Add。
   2. 如果服务在与频道相同的命名空间或项目下的 Topology 视图中可用，点击您要为该频道创建订阅的频道，并将箭头直接拖到服务以立即从频道创建订阅到该服务。

流程

1. 列出终端中的可用事件源类型：

   $ kn source list-types

   输出示例

   TYPE              NAME                                            DESCRIPTION
   ApiServerSource   apiserversources.sources.knative.dev            Watch and send Kubernetes API events to a sink
   PingSource        pingsources.sources.knative.dev                 Periodically send ping events to a sink
   SinkBinding       sinkbindings.sources.knative.dev                Binding for connecting a PodSpecable to a sink

2. 可选：在 OpenShift Container Platform 中，您还可以以 YAML 格式列出可用事件源类型：

   $ kn source list-types -o yaml

流程

1. 访问 Developer 视角。
2. 点 +Add。
3. 点 Event Source。
4. 查看可用的事件源类型。

流程

1. 列出终端中的现有事件源：

   $ kn source list

   输出示例

   NAME   TYPE              RESOURCE                               SINK         READY
   a1     ApiServerSource   apiserversources.sources.knative.dev   ksvc:eshow2   True
   b1     SinkBinding       sinkbindings.sources.knative.dev       ksvc:eshow3   False
   p1     PingSource        pingsources.sources.knative.dev        ksvc:eshow1   True

2. 可选： 您可以使用 --type 标志来只列出特定类型的事件源：

   $ kn source list --type <event_source_type>

   示例命令

   $ kn source list --type PingSource

   输出示例

   NAME   TYPE              RESOURCE                               SINK         READY
   p1     PingSource        pingsources.sources.knative.dev        ksvc:eshow1   True

流程

1. 在 OpenShift Container Platform web 控制台的 Administrator 视角中，进入 OperatorHub → Installed Operators。
2. 选择 knative-eventing 命名空间。
3. 点 OpenShift Serverless Operator 的 Provided APIs 列表中的 Knative Eventing 来进入 Knative Eventing 选项卡。

4. 点 knative-eventing，然后进入 knative-eventing 页面中的 YAML 选项卡。

   

5. 修改 KnativeEventing CR 中的副本数量：

   YAML 示例

   apiVersion: operator.knative.dev/v1beta1
   kind: KnativeEventing
   metadata:
     name: knative-eventing
     namespace: knative-eventing
   spec:
     high-availability:
       replicas: 3

流程

1. 在 OpenShift Container Platform web 控制台的 Administrator 视角中，进入 OperatorHub → Installed Operators。
2. 选择 knative-eventing 命名空间。
3. 点 OpenShift Serverless Operator 的 Provided APIs 列表中的 Knative Kafka 进入 Knative Kafka 标签页。

4. 点 knative-kafka，然后进入 knative-kafka 页面中的 YAML 选项卡。

   

5. 修改 KnativeKafka CR 中的副本数量：

   YAML 示例

   apiVersion: operator.serverless.openshift.io/v1alpha1
   kind: KnativeKafka
   metadata:
     name: knative-kafka
     namespace: knative-eventing
   spec:
     high-availability:
       replicas: 3