Kafka サンプルで必要な資格情報を使用したシークレットの作成

Kafka メッセージを送信する Code Engine アプリケーションを作成する前に、必要な資格情報を含む Code Engine 秘密を作成します。

開始前に

• 使用する Code Engine プロジェクトを決定し、このプロジェクトが選択されていることを確認します。 プロジェクトの作成と操作 を参照してください。

このシナリオを簡略化するため、 Kafka の送信者サンプルアプリと、 Kafka 受信者サンプルを使用する Kafka のイベントサブスクリプションの両方に必要な認証情報を格納する「 kafka-subscription-secret 」という1つのシークレットを作成します。 これらの認証情報は、サンプルアプリ「 Kafka 」の送信アプリおよび「 Code Engine 」の「 Kafka 」イベントサブスクリプションが、「 Event Streams 」のサービスインスタンスと通信するために必要です。 Kafka の送信者アプリやイベントサブスクリプションを作成する前に、このシークレットを作成する必要はありませんが、作成しておくと必要な手順が簡略化されます。

イベントを送信するための Code Engine アプリの作成

Event Streams インスタンスに接続し、 Kafka メッセージを Kafka メッセージの受信者（ Kafka コンシューマー）に送信するための Code Engine アプリを作成します。

コンソールから Kafka イベントを受信するための Code Engine アプリの作成

1. Kafka メッセージのイベント・コンシューマーとして機能し、 Kafka イベントを受信する Code Engine アプリケーションを作成します。 例えば、icr.io/codeengine/kafka-receiverイメージを使用するkafka-receiver-appというアプリケーションを作成します。 このイメージは receiver.go からビルドされています。これは、 Samples for IBM Cloud Code Engine GitHub repoから入手できます。 このサンプルでは、環境変数は必要ありません。
2. このアプリをデプロイした後、アプリの状況が ready であることを確認します。

コンソールを使用する場合、Kafka イベントを受信するために使用するアプリやジョブは、Kafka イベントのサブスクリプションを作成する前に存在している必要はありません。 ただし、イベント・サブスクリプションの作成時にアプリケーションまたはジョブが存在しない場合は、コンシューマーが存在しないことがサブスクリプションの状況に反映されます。 サブスクリプションが準備完了状態になり、このサブスクリプションを介してイベントを受信できるようになる前に、アプリまたはジョブを作成する必要があります。

コンソールからアプリの Code Engine Kafka イベント・サブスクリプションを作成する

Kafka のイベントサブスクリプションは、 Kafka のイベントプロデューサー（送信者）とコンシューマー（受信者）間の関係を定義します。

開始前に

• 使用する Code Engine プロジェクトを決定し、このプロジェクトが選択されていることを確認します。 プロジェクトの作成と操作 を参照してください。

コンソールからアプリケーションの「 Kafka 」イベントのサブスクリプションを作成するには、以下の手順を実行してください。

1. Code Engine プロジェクトページ から、自分のプロジェクトに移動します。
2. 「概要」ページで、**「イベント・サブスクリプション (Event subscriptions)」**をクリックします。
3. 「イベント・サブスクリプション (Event subscriptions)」ページで**「作成」**をクリックして、サブスクリプションを作成します。
4. 「イベント・サブスクリプションの作成 (Create an event subscription)」ページで、 Event Streams / Kafka タイルを選択してイベント・タイプを指定します。
5. 「一般」 に、 Event Streams / Kafka サブスクリプションの名前を指定します。 **「次へ」**をクリックして、先に進みます。
6. メッセージブローカーの詳細については、
   1. このサブスクリプションを通じてイベントとしてメッセージを受信するメッセージキューの、 Kafka メッセージブローカーのホストを指定します。 ブローカー・ホスト、トピック、およびアクセス資格情報に関する情報を取得するには、 Event Streams コンソールでサービス・インスタンスのサービス資格情報の詳細を表示します。 例えば、 myeventstream-key サービス・インスタンスのメッセージ・ブローカー・ホストには、 "broker-0-abcdabcdabcdabcd.kafka.svc01.us-south.eventstreams.cloud.ibm.com:9093", "broker-1-abcdabcdabcdabcd.kafka.svc01.us-south.eventstreams.cloud.ibm.com:9093" を指定します。 サービス・インスタンスのブローカーは、 Event Streams の Kafka_brokers_sasl フィールドで確認できます。
   2. 「構成」 をクリックして、メッセージ・ブローカーへのアクセスを構成します。 Code Engine から Kafka または Event Streams インスタンスに対して認証するには、メッセージ・ブローカー・アクセス・シークレットを指定する必要があります。
      • 新しいシークレットを作成したり、既存のシークレットを選択したり、メッセージ・ブローカーへのアクセスに資格情報が必要ない場合は、 None を選択したりすることができます。
      • シークレットを作成するには、 「作成」 をクリックします。 シークレットの名前、および username と password の値を指定します。 username および password の値は、 Kafka または Event Streams インスタンスのサービス資格情報の値と一致している必要があります。 例えば、 username の値は、 Event Streams サービス・インスタンスのサービス資格情報の詳細にリストされている user の値です。 password の値は、 Event Streams サービス・インスタンスのサービス資格情報の詳細にリストされている apikey の値です。
      • この例では、以前に作成した既存の kafka-subscription-secret シークレットを使用します。
   3. メッセージ待ち行列の既存のトピックの名前を指定してください。 例えば、kafka-topic1 です。 サービス・インスタンスの既存のトピックに関する情報を取得するには、 Event Streams コンソールでサービス・インスタンスに移動し、 「トピック」 を表示します。
   4. (オプション) コンシューマー・グループを指定します。 Kafka メッセージのコンシューマーは、 コンシューマー・グループ にグループ化できます。 コンシューマー・グループを使用している場合、トピック構成は、コンシューマー・グループ内のコンシューマーへのメッセージ・フローを制御します。 コンシューマーがコンシューマー・グループに追加されたり、コンシューマー・グループから削除されたりするたびに、そのトピックからのメッセージ・フローが変更される可能性があります。 このアクションにより、既存のコンシューマーがそのトピックからメッセージを受信しなくなる可能性があります。
   5. **「次へ」**をクリックして、先に進みます。
7. イベントコンシューマについては、イベントを受信するアプリケーションとして「 Code Engine 」を指定してください。 定義済みのアプリケーションおよびジョブのリストから選択することも、まだ作成されていないアプリ (またはジョブ) の名前を指定することもできます。 コンソールでイベント・サブスクリプションを作成するときに、アプリケーションまたはジョブが存在している必要はありません。 ただし、サブスクリプションが作成されると、コンシューマーが存在しないことがサブスクリプションの状況に反映されます。 サブスクリプションが準備完了状態になり、このサブスクリプションを介してイベントを受信できるようになる前に、アプリまたはジョブを作成する必要があります。 この例では、icr.io/codeengine/kafka-receiverイメージを参照するkafka-receiver-appアプリケーションを使用します。 アプリが存在しない場合は、 Kafka サブスクリプションの作成後に、アプリケーションの名前を指定し、 アプリケーションを作成 します。 アプリケーションの場合のみ、オプションでパスを指定できます。 デフォルトでは、イベントは、宛先アプリケーションのルート URL にルーティングされます。 パスを指定すると、アプリ内の別の宛先にイベントを送信できます。 例えば、サブスクリプション・パスに/eventsが指定されている場合、イベントはhttps://<base application URL>/eventsに送信されます。 **「次へ」**をクリックして、先に進みます。
8. まとめとして、 Kafka のイベント購読設定を確認し、必要に応じて変更を行ってください。 準備ができたら、「 作成 」をクリックして、 Kafka のサブスクリプションを作成してください。

コンソールから受信側アプリへのイベントの送信

kafka-receiver-app アプリケーションを参照する Kafka イベントのサブスクリプションが作成されたので、 kafka-sender-app を使用して、受信アプリケーションにメッセージイベントを送信してください。

1. 受信側アプリケーションのロギングを開始して、 コンソールからアプリケーション・ログを表示 し、イベントを確認します。

2. (オプション) 送信側アプリケーションのロギングを開始して、 コンソールからアプリケーション・ログを表示 し、送信されたイベントに関する情報を確認します。

3. ロギングが開始されたら、 curl で kafka-sender-app アプリケーションを呼び出し、 kafka-sender-app の public URL、トピック名、送信するメッセージ数を指定する。 このアプリケーションの公開 URL は、アプリケーションの Domain mappings タブから取得できます。 以下に例を示します。

   curl "<public_URL_of_Kafka_sender_app>?topic=<your_topic_name>&num=<number_of_messages_to_produce>"
   

値が単一ストリングとして扱われるように、必ず値を curl に引用符で囲んでください。

CLI で Kafka イベントを受信するための Code Engine アプリの作成

開始前に

• Code Engine の CLI 環境をセットアップします。

• プロジェクトを作成して使用できるようにします。

1. Kafka メッセージのイベント・コンシューマーとして機能し、 Kafka イベントを受信する Code Engine アプリケーションを作成します。 例えば、icr.io/codeengine/kafka-receiverイメージを使用するkafka-receiver-app2というアプリケーションを作成します。 このイメージは receiver.go からビルドされています。これは、 Samples for IBM Cloud Code Engine GitHub repoから入手できます。 このサンプルでは、環境変数は必要ありません。 必要に応じて、 --min-scale=1 オプションを指定することで、アプリが常に実行中のインスタンスを保持し、スケールがゼロにならないようにすることができます。 ログを参照するときには、実行中のインスタンスが常に存在するようにアプリを構成すると役に立ちます。 実稼働環境で実行している場合は、アプリの実行インスタンスを保持するコスト、または Code Engine をゼロに自動スケーリングするかどうかを考慮してください。 デフォルトでは、使用されていない場合、アプリはゼロにスケーリングされます。

   ibmcloud ce app create -n kafka-receiver-app2 --image icr.io/codeengine/kafka-receiver --min-scale 1
   

   デフォルトでは、イベントは、宛先アプリケーションのルート URL にルーティングされます。 --path オプションを使用すると、アプリ内の異なる宛先にイベントを送信できます。 例えば、サブスクリプションで--path /eventが指定されている場合、イベントはhttps://<base application URL>/eventsに送信されます。

2. このアプリをデプロイした後、 app get コマンドを実行して、アプリの状況が ready であることを確認します。

   ibmcloud ce app get -n kafka-receiver-app2
   

CLI を使用したアプリの Code Engine Kafka イベント・サブスクリプションの作成

CLI を使用して、 Kafka のイベントサブスクリプションを作成できます。これにより、 Kafka のイベントプロデューサー（送信者）とコンシューマー（受信者）間の関係が定義されます。

1. ibmcloud ce sub kafka create コマンドを使用して、 Kafka イベント用の Code Engine Kafka イベント・サブスクリプションを作成します。 以前に作成した kafka-subscription-secret シークレットを使用して、メッセージ・ブローカーにアクセスします。 Kafka リソースのサービス資格情報に基づいてブローカー情報を指定します。 この例では、 ibmcloud resource service-key myeventstream-key コマンドの出力からブローカー情報を取得できます。 トピックのブローカーごとに --broker オプションを指定する必要があることに注意してください。 --destination オプションは、イベントを受信する Code Engine リソースを指定します。

   ibmcloud ce sub kafka create --name mykafkasubscription --destination kafka-receiver-app2 --secret kafka-subscription-secret --topic kafka-topic1 --broker broker-3-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-5-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker  broker-0-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-1-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-4-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-2-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
   

2. Kafka イベントのサブスクリプションの詳細を表示します。

   ibmcloud ce sub kafka get -n mykafkasubscription
   

   出力例

   Getting Kafka event subscription 'mykafkasubscription'...
   OK
   Name:          mykafkasubscription
   [...]
   Destination Type:                 app
   Destination:                      kafka-receiver-app2
   Brokers:
   broker-3-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
   broker-5-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
   broker-0-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
   broker-1-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
   broker-4-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
   broker-2-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
   Consumer Group:                   knative-kafka-source-a4072fe1-1dfa-4470-9d07-bf7a0ff8e340
   Topics:
   kafka-topic1
   Secret key reference (user):      kafka-subscription-secret.username
   Secret key reference (password):  kafka-subscription-secret.password
   Ready:                            true
   Conditions:
   Type                     OK    Age  Reason
   ConnectionEstablished    true  24s
   InitialOffsetsCommitted  true  24s
   Ready                    true  24s
   Scheduled                true  24s
   SinkProvided             true  24s
   Events:
   Type     Reason           Age  Source                  Messages
   Normal   FinalizerUpdate  26s  kafkasource-controller  Updated "mykafkasubscription" finalizers
   

CLI を使用した受信側アプリへのイベントの送信

kafka-receiver-app アプリケーションを参照する Kafka イベントのサブスクリプションが作成されたので、 kafka-sender-app を使用して、受信アプリケーションにメッセージイベントを送信してください。

1. を使用して、デスティネーション・アプリのパブリック URL、 kafka-sender-app。 ibmcloud ce app get コマンドに --output url オプションを付けて、アプリの URL。

   ibmcloud ce app get -n kafka-sender-app --output url
   

   出力例

   https://kafka-sender-app.abcdabcdabc.us-south.codeengine.appdomain.cloud
   

2. Kafka イベント・プロデューサー・アプリ kafka-sender-app を実行して、宛先 Code Engine アプリケーションにイベントを送信します。 curl を指定して kafka-sender-app アプリケーションを呼び出し、トピックの値とメッセージ数を指定します。 アプリのパブリックxml-ph-0000@deepl.internalを見つけるには、 ibmcloud ce app get コマンドの出力を参照して、ご自身のアプリのパブリック URL を確認してください。 値が単一ストリングとして扱われるように、必ず値を curl に引用符で囲んでください。

   curl "<public_URL_of_Kafka_sender_app>?topic=<your_topic_name>&num=<number_of_messages_to_produce>"
   

   以下に例を示します。

   curl "https://kafka-sender-app.abcdabcdabc.us-south.codeengine.appdomain.cloud?topic=kafka-topic1&num=1"
   

3. ログ内のイベントを表示します。 有効なブローカー、トピック、およびアクセス・シークレットを使用して Kafka イベント・サブスクリプションが作成され、そのトピックに関するメッセージを生成する Kafka アプリケーション ( kafka-sender-app など) がある場合、 kafka-receiver-app などの Kafka メッセージを受信する宛先 Code Engine アプリケーションのログにイベントを表示できます。 Kafka 受信側アプリケーション (icr.io/codeengine/kafka-receiver) を使用する場合は、受信側アプリケーションのログで Event data を検索して、受信したメッセージを確認します。

   ibmcloud ce app logs -n kafka-receiver-app2
   

   出力例

   Getting logs for all instances of application 'kafka-receiver-app2'...
   OK
   kafka-receiver-app2-00001-deployment-66976f7988-9xttm/user-container:
   2022/03/31 22:19:45 Listening on port 8080
   2022/03/31 22:19:46 ----------
   2022/03/31 22:19:46 Path: /
   2022/03/31 22:19:46 Header: Accept-Encoding=[gzip]
   2022/03/31 22:19:46 Header: Ce-Id=[partition:0/offset:167]
   2022/03/31 22:19:46 Header: Ce-Source=[/apis/v1/namespaces/glxo4k7nj7d/kafkasources/mykafkasubscription#kafka-topic1]
   2022/03/31 22:19:46 Header: Ce-Specversion=[1.0]
   2022/03/31 22:19:46 Header: Ce-Subject=[partition:0#167]
   2022/03/31 22:19:46 Header: Ce-Time=[2022-03-31T22:19:36.499Z]
   2022/03/31 22:19:46 Header: Ce-Type=[dev.knative.kafka.event]
   2022/03/31 22:19:46 Header: Content-Length=[8]
   2022/03/31 22:19:46 Header: Forwarded=[for=172.30.208.213;proto=http, for=127.0.0.6]
   2022/03/31 22:19:46 Header: K-Proxy-Request=[activator]
   2022/03/31 22:19:46 Header: Traceparent=[00-b033708685c715a7c2384cdf05797785-65540b0937e9b0ce-00]
   2022/03/31 22:19:46 Header: User-Agent=[Go-http-client/1.1]
   2022/03/31 22:19:46 Header: X-B3-Parentspanid=[e1a785d7fdbead6c]
   2022/03/31 22:19:46 Header: X-B3-Sampled=[1]
   2022/03/31 22:19:46 Header: X-B3-Spanid=[abcde9901e6bf83f]
   2022/03/31 22:19:46 Header: X-B3-Traceid=[abcde490a426573772fa0bf60caf5ddb]
   2022/03/31 22:19:46 Header: X-Envoy-Attempt-Count=[1]
   2022/03/31 22:19:46 Header: X-Forwarded-For=[172.30.208.213, 127.0.0.6, 127.0.0.6]
   2022/03/31 22:19:46 Header: X-Forwarded-Proto=[http]
   2022/03/31 22:19:46 Header: X-Request-Id=[abcdeb4e-c5ac-abcd-abcd-60e6278abcde]
   2022/03/31 22:19:46 Event data: test1: 1
   

   なお、アプリのログ情報は1時間しか保存されませんのでご注意ください。 アプリ (またはジョブ) のログの表示について詳しくは、 ログの表示 を参照してください。

アプリのヘッダー

出力例

Ce-Id=[partition:0/offset:0]
Ce-Source=[/apis/v1/namespaces/ewgz38l13ts/kafkasources/mykafkasubscription#kafka-topic1]
Ce-Specversion=[1.0]
Ce-Subject=[partition:0#0]
Ce-Time=[2021-09-27T16:39:01.36Z]
Ce-Type=[dev.knative.kafka.event]

HTTP アプリ本体

HTTP の本文には、 Kafka メッセージが含まれており、その形式は、サブスクリプションの作成または更新時に指定した形式となります。

コンソールから Kafka イベントを受信するための Code Engine ジョブの作成

1. Kafka メッセージのイベント・コンシューマーとして機能し、 Kafka イベントを受信する Code Engine ジョブ を作成します。 たとえば、「 kafka-receiver-job 」という名前のジョブを作成し、サンプルイメージ「 icr.io/codeengine/codeengine 」を使用するようにします。 この画像は、 IBM Cloud Code Engine GitHub リポジトリ用のサンプル から入手可能な「 codeengine.go 」を基に作成されています。 このサンプルでは、環境変数は必要ありません。
2. このジョブを作成した後、そのジョブが ready 状況であることを確認します。

コンソールを使用する場合、Kafka イベントを受信するために使用するアプリやジョブは、Kafka イベントのサブスクリプションを作成する前に存在している必要はありません。 ただし、イベント・サブスクリプションの作成時にアプリケーションまたはジョブが存在しない場合は、コンシューマーが存在しないことがサブスクリプションの状況に反映されます。 サブスクリプションが準備完了状態になり、このサブスクリプションを介してイベントを受信できるようになる前に、アプリまたはジョブを作成する必要があります。

コンソールからジョブの Code Engine Kafka イベント・サブスクリプションを作成する

Kafka のイベントサブスクリプションは、 Kafka のイベントプロデューサー（送信者）とコンシューマー（受信者）間の関係を定義します。

開始前に

• 使用する Code Engine プロジェクトを決定し、このプロジェクトが選択されていることを確認します。 プロジェクトの作成と操作 を参照してください。

コンソールからアプリケーションの「 Kafka 」イベントのサブスクリプションを作成するには、以下の手順を実行してください。

1. Code Engine プロジェクトページ から、自分のプロジェクトに移動します。
2. 「概要」ページで、**「イベント・サブスクリプション (Event subscriptions)」**をクリックします。
3. 「イベント・サブスクリプション (Event subscriptions)」ページで**「作成」**をクリックして、サブスクリプションを作成します。
4. 「イベント・サブスクリプションの作成 (Create an event subscription)」ページで、 Event Streams / Kafka タイルを選択してイベント・タイプを指定します。
5. 「一般」 に、 Event Streams / Kafka サブスクリプションの名前を指定します。 **「次へ」**をクリックして、先に進みます。
6. メッセージブローカーの詳細については、
   1. このサブスクリプションを通じてイベントとしてメッセージを受信するメッセージキューの、 Kafka メッセージブローカーのホストを指定します。 ブローカー・ホスト、トピック、およびアクセス資格情報に関する情報を取得するには、 Event Streams コンソールでサービス・インスタンスのサービス資格情報の詳細を表示します。 例えば、 myeventstream-key サービス・インスタンスのメッセージ・ブローカー・ホストには、 "broker-0-abcdabcdabcdabcd.kafka.svc01.us-south.eventstreams.cloud.ibm.com:9093", "broker-1-abcdabcdabcdabcd.kafka.svc01.us-south.eventstreams.cloud.ibm.com:9093" を指定します。
   2. 「構成」 をクリックして、メッセージ・ブローカーへのアクセスを構成します。 Code Engine から Kafka または Event Streams インスタンスに対して認証するには、メッセージ・ブローカー・アクセス・シークレットを指定する必要があります。
      • 新しいシークレットを作成したり、既存のシークレットを選択したり、メッセージ・ブローカーへのアクセスに資格情報が必要ない場合は、 None を選択したりすることができます。
      • シークレットを作成するには、 「作成」 をクリックします。 シークレットの名前、および username と password の値を指定します。 username および password の値は、 Kafka または Event Streams インスタンスのサービス資格情報の値と一致している必要があります。 例えば、 username の値は、 Event Streams サービス・インスタンスのサービス資格情報の詳細にリストされている user の値です。 password の値は、 Event Streams サービス・インスタンスのサービス資格情報の詳細にリストされている apikey の値です。
      • この例では、以前に作成した既存の kafka-subscription-secret シークレットを使用します。
   3. メッセージ待ち行列の既存のトピックの名前を指定してください。 例えば、kafka-topic1 です。 サービス・インスタンスの既存のトピックに関する情報を取得するには、 Event Streams コンソールでサービス・インスタンスに移動し、 「トピック」 を表示します。
   4. (オプション) コンシューマー・グループを指定します。 Kafka メッセージのコンシューマーは、 コンシューマー・グループ にグループ化できます。 コンシューマー・グループを使用している場合、トピック構成は、コンシューマー・グループ内のコンシューマーへのメッセージ・フローを制御します。 コンシューマーがコンシューマー・グループに追加されたり、コンシューマー・グループから削除されたりするたびに、そのトピックからのメッセージ・フローが変更される可能性があります。 このアクションにより、既存のコンシューマーがそのトピックからメッセージを受信しなくなる可能性があります。
   5. **「次へ」**をクリックして、先に進みます。
7. イベントコンシューマについては、イベントを受信する「 Code Engine 」ジョブを指定してください。 定義済みのジョブおよびアプリのリストから選択することも、まだ作成されていないジョブ (またはアプリ) の名前を指定することもできます。 コンソールでイベント・サブスクリプションを作成するときに、アプリケーションまたはジョブが存在している必要はありません。 ただし、サブスクリプションが作成されると、コンシューマーが存在しないことがサブスクリプションの状況に反映されます。 サブスクリプションが作動可能状態になり、このサブスクリプションを介してイベントを受信できるようになる前に、ジョブ (またはアプリ) を作成する必要があります。 この例では、コンポーネント・タイプとして job を選択し、イベントを受信するコンポーネントとして icr.io/codeengine/codeengine イメージを参照する kafka-receiver-job ジョブを使用します。 該当するジョブが存在しない場合は、ジョブの名前を指定し、 Kafka サブスクリプションの作成後に そのジョブを作成してください。 アプリケーションの場合のみ、オプションでパスを指定できます。 **「次へ」**をクリックして、先に進みます。
8. まとめとして、 Kafka のイベント購読設定を確認し、必要に応じて変更を行ってください。 準備ができたら、「 作成 」をクリックして、 Kafka のサブスクリプションを作成してください。

コンソールから受信ジョブへのイベントの送信

「 kafka-receiver-job 」ジョブを参照する Kafka イベントのサブスクリプションが作成されたので、 kafka-sender-app を使用して、メッセージイベントを受信側ジョブに送信してください。

1. レシーバー・ジョブのロギングを開始して、 コンソールからジョブ・ログを表示 し、イベントを確認します。

2. (オプション) 送信側アプリケーションのロギングを開始して、 コンソールからアプリケーション・ログを表示 し、送信されたイベントに関する情報を確認します。

3. ロギングが開始されたら、 curl で kafka-sender-app アプリケーションを呼び出し、 kafka-sender-app の public URL、トピック名、送信するメッセージ数を指定する。 このアプリケーションの公開 URL は、アプリケーションの Domain mappings タブから取得できます。 以下に例を示します。

   curl "<public_URL_of_Kafka_sender_app>?topic=<your_topic_name>&num=<number_of_messages_to_produce>"
   

値が単一ストリングとして扱われるように、必ず値を curl に引用符で囲んでください。

有効なブローカー、トピック、およびアクセス・シークレットを使用して Kafka サブスクリプションが作成され、そのトピックに関するメッセージを生成する Kafka ジョブ ( kafka-sender-app など) がある場合、 kafka-receiver-job などの Kafka メッセージを受信する Code Engine ジョブのログでイベントを確認できます。 Kafka レシーバー・ジョブ (icr.io/codeengine/codeengine) を使用する場合は、レシーバー・ジョブのログで CE_DATA を検索して、受信したメッセージを確認します。

CLI で Kafka イベントを受信するための Code Engine ジョブの作成

開始前に

• Code Engine の CLI 環境をセットアップします。

• プロジェクトを作成して使用できるようにします。

1. Kafka メッセージのイベント・コンシューマーとして機能し、 Kafka イベントを受信する Code Engine ジョブ を作成します。 例えば、icr.io/codeengine/codeengineイメージを使用するkafka-receiver-jobという名前のジョブを作成します。 この画像は、 IBM Cloud Code Engine GitHub リポジトリ用のサンプル から入手可能な「 codeengine.go 」を基に作成されています。

   ibmcloud ce job create -n kafka-receiver-job --image icr.io/codeengine/codeengine
   

2. (オプション) このジョブを作成した後、 job get コマンドを実行して、このジョブに関する情報を表示します。

   ibmcloud ce job get -n kafka-receiver-job
   

CLI を使用したジョブの Code Engine Kafka イベント・サブスクリプションの作成

CLI を使用して、 Kafka のイベントサブスクリプションを作成できます。これにより、 Kafka のイベントプロデューサー（送信者）とコンシューマー（受信者）間の関係が定義されます。

1. ibmcloud ce sub kafka create コマンドを使用して、 Kafka イベント用の Code Engine Kafka イベント・サブスクリプションを作成します。 以前に作成した kafka-subscription-secret シークレットを使用して、メッセージ・ブローカーにアクセスします。 Kafka リソースのサービス資格情報に基づいてブローカー情報を指定します。 この例では、 ibmcloud resource service-key myeventstream-key コマンドの出力からブローカー情報を取得できます。 トピックのブローカーごとに --broker オプションを指定する必要があることに注意してください。 --destination オプションは、イベントを受信する Code Engine リソースを指定します。 受信ジョブを処理する場合は、 --destination-type オプションも指定して、リソースがジョブであることを指定する必要があります。このオプションのデフォルトは app です。

   ibmcloud ce sub kafka create --name mykafkasubscription-withjob --destination-type job --destination kafka-receiver-job --secret kafka-subscription-secret --topic kafka-topic1 --broker broker-3-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-5-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker  broker-0-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-1-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-4-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093 --broker broker-2-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
   

2. Kafka イベントのサブスクリプションの詳細を表示します。

   ibmcloud ce sub kafka get -n mykafkasubscription-withjob
   

   出力例

   Getting Kafka event subscription 'mykafkasubscription-withjob'...
   OK
   Name:          mykafkasubscription-withjob
   [...]
   Destination Type:                 job
   Destination:                      kafka-receiver-job
   Brokers:
   broker-3-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
   broker-5-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
   broker-0-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
   broker-1-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
   broker-4-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
   broker-2-abcdabcdabcdabcd.kafka.svc07.us-south.eventstreams.cloud.ibm.com:9093
   Consumer Group:                   knative-kafka-source-a4072fe1-1dfa-4470-9d07-bf7a0ff8e340
   Topics:
   kafka-topic1
   Secret key reference (user):      kafka-subscription-secret.username
   Secret key reference (password):  kafka-subscription-secret.password
   Ready:                            true
   [...]
   

CLI を使用した受信ジョブへのイベントの送信

kafka-receiver-job アプリケーションを参照する Kafka イベントのサブスクリプションが作成されたので、 kafka-sender-app を使用して、受信アプリケーションにメッセージイベントを送信してください。

1. コマンドを使用して、インストール先アプリのパブリック URL、 kafka-sender-app。 ibmcloud ce app get コマンドを使ってアプリの URL。

   ibmcloud ce app get -n kafka-sender-app --output url
   

   出力例

   https://kafka-sender-app.abcdabcdabc.us-south.codeengine.appdomain.cloud
   

2. Kafka イベント・プロデューサー・アプリ kafka-sender-app を実行して、イベントを宛先 Code Engine ジョブに送信します。 curl を指定して kafka-sender-app アプリケーションを呼び出し、トピックの値とメッセージ数を指定します。 コマンドの出力を使って ibmcloud ce app get コマンドの出力を使って、イベント生成アプリのパブリック URL。 値が単一ストリングとして扱われるように、必ず値を curl に引用符で囲んでください。 以下に例を示します。

   curl "<public_URL_of_Kafka_sender_app>?topic=<your_topic_name>&num=<number_of_messages_to_produce>"
   

   以下に例を示します。

   curl "https://kafka-sender-app.abcdabcdabc.us-south.codeengine.appdomain.cloud?topic=kafka-topic1&num=1"
   

3. ログ内のイベントを表示します。 有効なブローカー、トピック、およびアクセス・シークレットを使用して Kafka イベント・サブスクリプションが作成され、そのトピックに関するメッセージを生成する Kafka アプリケーション ( kafka-sender-app など) がある場合、 kafka-receiver-job などの Kafka メッセージを受信する宛先 Code Engine ジョブのログでイベントを確認できます。 curl を使用して送信されるメッセージごとに、同じ数のジョブ実行が Kafka イベントによってトリガーされます。 ジョブに送信されたイベントを表示するには、 ibmcloud ce jobrun logs コマンドを使用します。

   1. ibmcloud ce jobrun list コマンドを使用して、 kafka-receiver-job ジョブのジョブ実行をリストします。

       ibmcloud ce jobrun list --job kafka-receiver-job
       ```
   2. **`ibmcloud ce jobrun logs`** コマンドを使用して、特定のジョブ実行のログを取得します。
   
   ```txt {: pre}
       ibmcloud ce jobrun logs -n kafka-receiver-job-abcde
       ```
       出力例
   
       ```txt {: screen}
       Getting logs for all instances of job run 'kafka-receiver-job-abcde'...
       Getting jobrun 'kafka-receiver-job-abcde'...
       Getting instances of jobrun 'kafka-receiver-job-abcde'...
       OK
       kafka-receiver-job-abcde-0-0/kafka-receiver-job:
       Hello from helloworld! I'm a batch job! Index: 0
       Hello World from:
       . ___  __  ____  ____
       ./ __)/  \(    \(  __)
       ( (__(  O )) D ( ) _)
       .\___)\__/(____/(____)
       .____  __ _   ___  __  __ _  ____
       (  __)(  ( \ / __)(  )(  ( \(  __)
       .) _) /    /( (_ \ )( /    / ) _)
       (____)\_)__) \___/(__)\_)__)(____)
       Some Env Vars:
       --------------
       CE_DATA=test1: 2
       CE_DOMAIN=us-south.codeengine.appdomain.cloud
       CE_ID=partition:0/offset:249
       CE_JOB=kafka-receiver-job
       CE_JOBRUN=kafka-receiver-job-abcde
       CE_SOURCE=/apis/v1/namespaces/p99k7iy919d/kafkasources/kafkasub-job-ui#kafka-topic1
       CE_SPECVERSION=1.0
       CE_SUBDOMAIN=p99k7iy919d
       CE_SUBJECT=partition:0#249
       CE_TIME=2022-06-21T12:19:24.06Z
       CE_TYPE=dev.knative.kafka.event
       HOME=/root
       HOSTNAME=kafka-receiver-job-abcde-0-0
       JOB_INDEX=0
       KUBERNETES_PORT=tcp://172.21.0.1:443
       KUBERNETES_PORT_443_TCP=tcp://172.21.0.1:443
       KUBERNETES_PORT_443_TCP_ADDR=172.21.0.1
       KUBERNETES_PORT_443_TCP_PORT=443
       KUBERNETES_PORT_443_TCP_PROTO=tcp
       KUBERNETES_SERVICE_HOST=172.21.0.1
       KUBERNETES_SERVICE_PORT=443
       KUBERNETES_SERVICE_PORT_HTTPS=443
       PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin
       PWD=/
       SHLVL=1
       z=Set env var 'SHOW' to see all variables
       ```
   Kafka レシーバー・ジョブ (`icr.io/codeengine/codeengine`) を使用する場合は、レシーバー・ジョブのログで `CE_DATA` を検索して、受信したメッセージを確認します。
   
   なお、ジョブの実行に関するログ情報は1時間しか保持されませんのでご注意ください。 アプリまたはジョブのログの表示について詳しくは、 [ログの表示](/docs/codeengine?topic=codeengine-logging) を参照してください。