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. 在 Target 部分中，选择您的事件 sink。这可以是 Resource 或一个 URI ：

      1. 选择 Resource 来使用频道、代理或服务作为事件源的事件 sink。
      2. 选择 URI 来指定事件路由到的 Uniform Resource Identifier (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 event-display --image quay.io/openshift-knative/showcase

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

   $ kn trigger create <trigger_name> --sink ksvc:event-display

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

   $ oc create deployment event-origin --image quay.io/openshift-knative/showcase

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

删除 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/showcase

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 event-origin --image=quay.io/openshift-knative/showcase

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

删除 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/showcase

   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. 在 Target 部分中，选择您的事件 sink。这可以是 Resource 或一个 URI ：

      1. 选择 Resource 来使用频道、代理或服务作为事件源的事件 sink。在本例中，上一步中创建的 event-display 服务用作目标 资源。
      2. 选择 URI 来指定事件路由到的 Uniform Resource Identifier (URI)。
   7. 点 Create。

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

2. 查看 ping 源和接收器。

   

3. 在 Web 浏览器中查看 event-display 服务。您应该在 Web UI 中看到 ping 源事件。

   

删除 ping 源

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

流程

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

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

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/showcase

   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. 在 Target 部分中，选择您的事件 sink。这可以是 Resource 或一个 URI ：

      1. 选择 Resource 来使用频道、代理或服务作为事件源的事件 sink。
      2. 选择 URI 来指定事件路由到的 Uniform Resource Identifier (URI)。
   6. 输入 Kafka 事件源的 名称。
4. 点 Create。

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

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

   

流程

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

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

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 表单视图的 Target 部分，选择您的事件 sink。这可以是 Resource 或一个 URI ：

   1. 选择 Resource 来使用频道、代理或服务作为事件源的事件 sink。
   2. 选择 URI 来指定事件路由到的 Uniform Resource Identifier (URI)。
3. 点 Create。

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

流程

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. 在 Topology 视图中，指向您创建的代理。此时会出现一个箭头。将箭头拖到您要连接到代理的接收器。此操作将打开 Add Trigger 对话框。
2. 在 Add Trigger 对话框中，为触发器输入一个名称并点 Add。

1. 在 Developer 视角中，导航到 Topology。
2. 点击将代理连接到 sink 的行，以便在 Details 面板中查看触发器的详情。

流程

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