5.2. 事件源

流程

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，如 Red Hat OpenShift Streams for Apache Kafka。

流程

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

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