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

   この namespace を、イベントソースのインストールに選択した namespace に変更します。

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. APIVERSION に v1 を、KIND に Event を入力します。
   2. 作成したサービスアカウントの Service Account Name を選択します。
   3. イベントソースの Sink を選択します。Sink は、チャネル、ブローカー、またはサービスなどの Resource、または 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

   この namespace を、イベントソースのインストールに選択した namespace に変更します。

2. YAML ファイルを適用します。

   $ oc apply -f <filename>

3. イベントシンクを持つ API サーバーソースを作成します。次の例では、シンクはブローカーです。

   $ 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. ブローカーをイベントシンクとして使用した場合は、トリガーを作成して、default のブローカーからサービスへのイベントをフィルターリングします。

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

6. デフォルト namespace で 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

   この namespace を、イベントソースのインストールに選択した namespace に変更します。

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. 直接の手順で作成下サービスに、default ブローカーからイベントをフィルターする Trigger オブジェクトを YAML ファイルとして作成します。

   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. デフォルト namespace で 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. PingSource が機能していることを確認するには、受信メッセージをサービスのログにダンプする単純な 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. 直前の手順で作成したサービスと同じ namespace、またはイベントの送信先となる他のシンクと同じ namespace に 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 * * * * であり、2 分ごとにメッセージを送信する PingSource を作成します。
   5. オプション: Data の値を入力できます。これはメッセージのペイロードです。
   6. Sink を選択します。これは Resource または URI のいずれかになります。この例では、直前の手順で作成された event-display サービスが Resources シンクとして使用されます。
   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 イベントのセットごとに、PingSource をイベントコンシューマーと同じ namespace に作成します。

   $ 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 ソースをイベントコンシューマーと同じ namespace に作成します。

   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. ブートストラップサーバー のコンマ区切りの一覧を追加します。
   2. トピック のコンマ区切りの一覧を追加します。
   3. コンシューマーグループ を追加します。
   4. 作成したサービスアカウントの Service Account Name を選択します。
   5. イベントソースの Sink を選択します。Sink は、チャネル、ブローカー、またはサービスなどの Resource、または URI のいずれかになります。
   6. Kafka イベントソースの Name を入力します。
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 namespace にインストールされている。
   • 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

   トピックは、データの保存先を提供します。各トピックは、1 つまたは複数のパーティションに分割されます。

   3

   シンクは、イベントがソースから送信される場所を指定します。

   重要

   OpenShift Serverless 上の KafkaSource オブジェクトの API の v1beta1 バージョンのみがサポートされます。非推奨となった v1alpha1 バージョンの API は使用しないでください。

   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. 選択された namespace にシークレットとして証明書ファイルを作成します。

   $ 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

   Red Hat OpenShift Streams for Apache Kafka などのパブリッククラウド Kafka サービスを使用している場合は、caCert 仕様は必要ありません。

手順

1. +Add Event Source に移動して任意のタイプのイベントソースを作成し、作成するイベントソースを選択します。
2. イベントソースの作成 フォームビューの シンク セクションで、リソース リストからシンクを選択します。
3. Create をクリックします。

1. Developer パースペクティブで、Topology に移動します。
2. イベントソースを表示し、接続されたシンクをクリックして、右側のパネルにシンクの詳細を表示します。