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

   この 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

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

手順

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

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

手順

1. KafkaSink オブジェクト定義を YAML ファイルとして作成します。

   Kafka シンク YAML

   apiVersion: eventing.knative.dev/v1alpha1
   kind: KafkaSink
   metadata:
     name: <sink-name>
     namespace: <namespace>
   spec:
     topic: <topic-name>
     bootstrapServers:
      - <bootstrap-server>

2. Kafka シンクを作成するには、KafkaSink YAML ファイルを適用します。

   $ oc apply -f <filename>

3. シンクが仕様で指定されるようにイベントソースを設定します。

   API サーバーソースに接続された Kafka シンクの例

   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

   イベントソースの namespace。

   3

   イベントソースのサービスアカウント。

   4

   Kafka シンクの名前。

手順

1. Developer パースペクティブで、+Add ビューに移動します。
2. Eventing カタログ で Event Sink をクリックします。
3. カタログ項目で KafkaSink を検索してクリックします。
4. イベントシンクの作成 をクリックします。

5. フォームビューで、ホスト名とポートの組み合わせであるブートストラップサーバーの URL を入力します。

   
6. イベントデータを送信するトピックの名前を入力します。
7. イベントシンクの名前を入力します。
8. Create をクリックします。

検証

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

手順

1. KafkaSink オブジェクトと同じ namespace に証明書ファイルをシークレットとして作成します。

   重要

   証明書とキーは 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>

   • SASL を使用した認証と TLS を使用した暗号化の場合:

     $ oc create secret -n <namespace> generic <secret_name> \
       --from-literal=protocol=SASL_SSL \
       --from-literal=sasl.mechanism=<sasl_mechanism> \
       --from-file=ca.crt=<my_caroot.pem_file_path> \ 1
       --from-literal=user=<username> \
       --from-literal=password=<password>

     1

     パブリッククラウドで管理される Kafka サービスを使用している場合は、ca.crt を省略してシステムのルート CA セットを使用できます。

   • TLS を使用した認証と暗号化の場合:

     $ oc create secret -n <namespace> generic <secret_name> \
       --from-literal=protocol=SSL \
       --from-file=ca.crt=<my_caroot.pem_file_path> \ 1
       --from-file=user.crt=<my_cert.pem_file_path> \
       --from-file=user.key=<my_key.pem_file_path>

     1

     パブリッククラウドで管理される Kafka サービスを使用している場合は、ca.crt を省略してシステムのルート CA セットを使用できます。

2. KafkaSink オブジェクトを作成または変更し、auth 仕様にシークレットへの参照を追加します。

   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 オブジェクトを、eventing.knative.dev/injection: enabled アノテーションを付けて YAML ファイルとして作成します。

   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

   トリガーがイベントを送信するイベントシンクまたは サブスクライバー の詳細を指定します。

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 ラベルを namespace から削除します。

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

   アノテーションを削除すると、Knative では削除後にブローカーを再作成できなくなります。

2. 選択された namespace からブローカーを削除します。

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

手順

1. Developer パースペクティブで、+Add → Broker に移動します。Broker ページが表示されます。
2. オプション:ブローカーの Name を更新します。名前を更新しないと、生成されたブローカーの名前は 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 ConfigMap を変更するか、以下の設定が含まれる独自の ConfigMap を作成します。

   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

   ConfigMap 名。

   2

   ConfigMap が存在する namespace。

   3

   Kafka ブローカーのトピックパーティションの数。これは、イベントをブローカーに送信する速度を制御します。パーティションが多い場合には、コンピュートリソースが多く必要です。

   4

   トピックメッセージのレプリケーション係数。これにより、データ損失を防ぐことができます。レプリケーション係数を増やすには、より多くのコンピュートリソースとストレージが必要になります。

   5

   ブートストラップサーバーのコンマ区切りリスト。これは、OpenShift Container Platform クラスターの内部または外部にある可能性があり、ブローカーがイベントを受信してイベントを送信する Kafka クラスターのリストです。

   重要

   default.topic.replication.factor の値は、クラスター内の Kafka ブローカーインスタンスの数以下である必要があります。たとえば、Kafka ブローカーが 1 つしかない場合、default.topic.replication.factor の値は "1" より大きな値にすることはできません。

   Kafka ブローカーの ConfigMap の例

   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. ConfigMap を適用します。

   $ oc apply -f <config_map_filename>

3. Kafka Broker オブジェクトの ConfigMap を指定します。

   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

   ブローカーが存在する namespace。

   3

   ブローカークラスアノテーション。この例では、ブローカーはクラス値 Kafka を使用する Kafka ブローカーです。

   4

   ConfigMap 名。

   5

   ConfigMap が存在する namespace。

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

   順序付けられていないコンシューマーは、適切なオフセット管理を維持しながら、メッセージを順序付けずに配信するノンブロッキングコンシューマーです。

   ordered

   順序付きコンシューマーは、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

   順序付けられていないコンシューマーは、適切なオフセット管理を維持しながら、メッセージを順序付けずに配信するノンブロッキングコンシューマーです。

   ordered

   順序付きコンシューマーは、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 のダイアログボックスで、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. タイプ リストで、作成する Channel オブジェクトのタイプを選択します。

   注記

   現時点で、InMemoryChannel チャネルオブジェクトのみがデフォルトでサポートされます。Apache Kafka の Knative チャネルは、OpenShift Serverless に Apache Kafka の Knative ブローカー実装をインストールしている場合に使用できます。

4. Create をクリックします。

手順

1. Developer パースペクティブで、+Add → Channel に移動します。
2. タイプ リストで、作成する 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 オブジェクトの API の v1beta1 バージョンのみがサポートされます。非推奨となった v1alpha1 バージョンの API は使用しないでください。

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

   $ oc apply -f <filename>

手順

1. Developer パースペクティブで、Topology ページに移動します。

2. 以下の方法のいずれかを使用してサブスクリプションを作成します。

   1. サブスクリプションを作成するチャネルにカーソルを合わせ、矢印をドラッグします。Add Subscription オプションが表示されます。

      
      1. Subscriber 一覧でシンクを選択します。
      2. Add をクリックします。
   2. このサービスが、チャネルと同じ namespace またはプロジェクトにある Topology ビューで利用可能な場合は、サブスクリプションを作成するチャネルをクリックし、矢印をサービスに直接ドラッグして、チャネルからそのサービスにサブスクリプションを即時に作成します。

手順

1. OpenShift Container Platform Web コンソールの Administrator パースペクティブで、 Serverless → Eventing に移動します。
2. Channel タブで、サブスクリプションを追加するチャネルの Options メニュー を選択します。
3. 一覧で Add Subscription をクリックします。
4. Add Subscription のダイアログボックスで、サブスクリプションの Subscriber を選択します。サブスクライバーは、チャネルからイベントを受信する Knative サービスです。
5. Add をクリックします。

手順

1. 選択された namespace にシークレットとして証明書ファイルを作成します。

   $ 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. シークレットおよびシークレットの namespace を参照します。

   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

   注記

   ブートストラップサーバーで一致するポートを指定するようにしてください。

   以下に例を示します。

   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. 選択された 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" \
     --from-literal=user="my-sasl-user"

   • キー名に ca.crt、 password、および sasl.mechanism を使用します。これらの値は変更しないでください。

   • パブリック CA 証明書で SASL を使用する場合は、シークレットの作成時に ca.crt 引数ではなく tls.enabled=true フラグを使用する必要があります。以下に例を示します。

     $ 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. シークレットおよびシークレットの namespace を参照します。

   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

   注記

   ブートストラップサーバーで一致するポートを指定するようにしてください。

   以下に例を示します。

   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. このサービスが、チャネルと同じ namespace またはプロジェクトにある 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 namespace を選択します。
3. OpenShift Serverless Operator の Provided API 一覧で Knative Eventing をクリックし、Knative Eventing タブに移動します。

4. knative-serving をクリックしてから、knative-eventing ページの YAML タブに移動します。

   

5. KnativeEvening 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 namespace を選択します。
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