ノート:

このチュートリアルでは、Oracle Cloudへのアクセスが必要です。無料アカウントにサインアップするには、Oracle Cloud Infrastructure Free Tierの開始を参照してください。
Oracle Cloud Infrastructureの資格証明、テナンシおよびコンパートメントの値の例を使用します。演習を完了するときに、これらの値をクラウド環境に固有の値に置き換えます。

OCI Events ServiceおよびOCI Functionsを使用したOCIプライベート・ストリームへのイベントの書込み

イントロダクション

Oracle Cloud Infrastructure (OCI) Events Serviceのルールは、Cloud Native Computing Foundation (CNCF)のCloudEvents標準に準拠するイベントを使用して、OCIでのリソース変更を追跡します。これにより、開発者は、OCI Functionsでコードをトリガーしたり、OCI Streamingにデータを書き込んだり、OCI Notificationsを使用してアラートを送信したりすることで、リアルタイムの変更に対応できます。

パブリック・ストリームへのイベント送信はネイティブにサポートされますが、プライベート・ストリームへの書込みには、OCI関数またはカスタム・コードを使用したカスタム・アプローチが必要です。このチュートリアルでは、OCI関数を使用して、OCIイベント・サービス・ルールからイベント・ペイロードを取得し、プライベート・ストリームに書き込む方法を示します。

目標

OCIイベントを処理するための設定を自動化し、OCIプライベート・ストリームにスケーラブルな方法で送信することで、セキュアでカスタマイズされた処理を実現します。

前提条件

OCIテナンシへのアクセス。
OCI Events Serviceルール、Oracle ApplicationsおよびOCI Streamingサービスを管理する権限。

タスク1: 必要なポリシーおよびOracle Cloud Infrastructure Identity and Access Management (OCI IAM)権限の設定

このソリューションの各コンポーネントは、相互作用するOCIリソースにアクセスできる必要があります。このチュートリアルに従うには、開発者が次の権限を持っている必要があります。

ユーザー・ポリシー: OCIイベント・サービス・ルール、OCI通知、OCI関数、OCIストリームおよびOCI Vault (オプション)およびネットワーク・ファミリの使用を管理します。
サービス・ポリシー: OCIプライベート・ストリームにメッセージを書き込む権限をファンクションに付与します。

詳細なポリシーはここにあります。

タスク2: プライベート・ストリームの作成

OCI Streamingは、保存中および転送中にデータが暗号化されるフルマネージドOCIサービスであり、メッセージの整合性とセキュリティを確保します。セキュリティを強化するために、OCI Vaultサービスを使用して、特定のコンプライアンスまたはセキュリティ要件を満たす独自の暗号化キーを格納および管理できます。プライベート・エンドポイントは、Virtual Cloud Network (VCN)内で構成して、ストリームをさらに保護し、プライベートIPアドレスをストリーム・プールに関連付けることができます。これにより、OCIストリーミング・トラフィックがVCN内にとどまり、インターネット全体が回避されます。ただし、プライベート・エンドポイントを使用するストリームにはインターネットからアクセスできず、コンソールを介して最新のメッセージを表示する機能も制限されることに注意してください。プライベート・ストリームからのメッセージを消費するには、コンシューマには、プライベート・ストリームがホストされているネットワークへのルートとアクセスの両方が必要です。

ストリームおよびストリーム・プールを作成します。「ストリーム名」と入力し、「新規ストリーム・プールの作成」を選択してストリーム・プールを作成します。「ストリーム・プールの構成」セクションで、「ストリーム・プール名」を入力し、「プライベート・エンドポイント」を選択して、それに応じてVCNおよび「サブネット」およびネットワークの詳細を入力します。オプションですが、ネットワーク・セキュリティ・グループに、そのNSG内のすべてのトラフィックに対するイングレス・ルールを提供することをお薦めします。詳細は、ストリームの作成およびストリーム・プールの作成を参照してください。

NSGルールのイメージ

ストリームおよびストリーム・プール作成のイメージ

独自の暗号化キーを使用することで、キーのライフサイクルをより詳細に制御できます。ストリーム内のメッセージの保存を調整するオプションがあります。デフォルトは1日で、最大は7日です。

ストリーム暗号化のイメージ

ストリーム・プールのイメージ

ストリームOCIDおよびメッセージ・エンドポイントを書き留めます。この情報を関数に渡す必要があります。

タスク3: OCI通知トピックおよびOCIイベント・サービス・ルールの作成

リフレッシュとして、イベントの仕組みは次のとおりです。OCIサービスは、リソースまたはデータに対してイベントを発行します。対象のイベントを指定するフィルタを含むルールを作成します。フィルタが一致するイベントを見つけたときにアクションをトリガーする必要があります。アクションは、OCI通知トピック、OCIストリーミング・サービスまたはOCI Functionsです。

アプローチ1:要件がイベント・ペイロードをパブリック・ストリームに送信する場合、フローをネイティブに構成できるため、ファンクションは必要ありません。
アプローチ2:特定のイベント・ペイロードのみをプライベート・ストリームに送信することが要件である場合は、その関数をアクションとして直接指定できます。
アプローチ3:複数のイベント・ペイロードをプライベート・ストリームに送信することが要件である場合、理想的な方法は、イベント・ペイロードをOCI通知トピックに送信し、OCI FunctionsをOCI通知トピックにサブスクライブすることです。

このチュートリアルでは、最も複雑なアプローチ3を確認します。

OCI通知トピックを作成します。詳細は、トピックの作成を参照してください。
イベント・ルールを作成します。このアプローチを正当化するために、2つのイベント・ルールを作成します。詳細は、イベント・ルールの作成を参照してください。

タスク4: ファンクションの開発およびデプロイ

このファンクションは、イベント・ペイロードをエンコードし、指定されたメッセージ・エンドポイントおよびストリームにパブリッシュします。詳細は、ファンクションの作成を参照してください。

func.py:

import json
import logging
import oci
from base64 import b64encode

def handler(ctx, data):
    try:
        # Parse the incoming event payload
        body = json.loads(data.getvalue())
        cfg = ctx.Config()
        stream_ocid = cfg["stream_ocid"]
        stream_endpoint = cfg["stream_endpoint"]

        logging.getLogger().info(f'Function invoked for event: {body}')

        # Publish the event to the stream
        publish_to_stream(stream_ocid, stream_endpoint, body)

        return {"status": 200, "message": "Successfully processed event"}
    except (Exception, ValueError) as ex:
        logging.getLogger().error(f'Error processing event payload: {str(ex)}')
        return {"status": 500, "message": "Internal Server Error"}

def publish_to_stream(stream_ocid, stream_endpoint, event_data):
    signer = oci.auth.signers.get_resource_principals_signer()
    stream_client = oci.streaming.StreamClient(config={}, signer=signer, service_endpoint=stream_endpoint)

    # Convert event data to JSON string and encode
    event_payload = json.dumps(event_data).encode('utf-8')

    # Build the message list
    message_list = [
        oci.streaming.models.PutMessagesDetailsEntry(
            key=b64encode("partition-key-1".encode()).decode(),
            value=b64encode(event_payload).decode()
        ),
    ]

    try:
        logging.getLogger().info(f"Publishing {len(message_list)} messages to stream {stream_ocid}")
        put_message_details = oci.streaming.models.PutMessagesDetails(messages=message_list)
        put_message_result = stream_client.put_messages(stream_ocid, put_message_details)

        # Log publishing results
        for entry in put_message_result.data.entries:
            if entry.error:
                logging.getLogger().error(f"Error publishing message: {entry.error_message}")
            else:
                logging.getLogger().info(f"Published message to partition {entry.partition}, offset {entry.offset}")
    except Exception as ex:
        logging.getLogger().error(f"Failed to publish messages to stream: {str(ex)}")
        raise

func.yaml:

schema_version: 20180708
name: events-to-pvt-stream
version: 0.0.1
runtime: python
build_image: fnproject/python:3.9-dev
run_image: fnproject/python:3.9
entrypoint: /python/bin/fdk /function/func.py handler
memory: 256
config:
stream_ocid: ocid1.stream.123445
stream_endpoint: https://xyz.us-ashburn-1.oci.oraclecloud.com

requirements.txt:
```
fdk
oci
```

Fnのイメージ

最後のステップは、プライベート・ストリームがどこにあるかを関数に伝えることです。このファンクションは、構成パラメータを使用して、別のテナンシにデプロイする場合は再利用可能にします。

Fn構成のイメージ

このタスクでは、デプロイされたファンクションをOCI通知トピックにサブスクライブします。OCI通知トピックがメッセージを受信するたびに、ファンクションがトリガーされ、ファンクションはプライベート・ストリームのイベント・ペイロードを書き込みます。

要件を満たすように機能コードを変更することで、特定のフィールドを削除または拡張できます。ファンクション・サブスクリプションの確認は必要ありません。詳細は、ファンクション・サブスクリプションの作成を参照してください。

サブスクリプションのイメージ

Subscription2のイメージ

検証

データ・フローを検証できる場所は複数あります。

イベント・メトリック・ページを確認して、指定されたルールと一致するイベントがあるかどうかを確認します。障害グラフは空である必要があります。
OCI Notificationsのトピック・メトリックを確認します。すべてのメッセージが配信されていることを確認します。失敗したメッセージ・チャートにデータがありません。
ファンクションの起動メトリックを確認します。エラーがなく、関数がスロットルされていないことを確認します。
データがプライベート・ストリームに取り込まれていることを確認します。

ストリーム・メトリックのイメージ

次のいずれかのチャートにデータがない場合は、そこで停止し、そのサービスのログを有効にします。ログは、特定のリソースがタスクの実行に失敗する理由を説明します。

次のステップ

このチュートリアルでは、OCIイベント・サービス、OCI通知、OCI関数およびOCIストリーミングを統合して、イベント・ペイロードを安全に処理し、プライベート・ストリームに公開する方法を学習しました。セキュアなエンドポイントの設定、暗号化キーの管理、およびOCI IAMを使用したきめ細かいアクセス制御により、認可されたユーザーとサービスのみがストリーミング・データと対話できるようにします。

このソリューションにより、チームは堅牢なセキュリティ標準を維持しながら、リソース変更をリアルタイムで取得できます。組織は、プライベート・ストリームを使用して機密データを保護し、業界の規制に準拠し、業務ワークフローが組織のセキュリティやコンプライアンスの目標に適合していることを確認できます。このアプローチにより、セキュリティ体制が強化され、チームはシームレスで安全なイベントドリブンの自動化を実現できます。

OCI FunctionsおよびOCIプライベート・ストリーム機能の使用の詳細は、Oracle担当者に連絡するか、クラウド・セキュリティ・ソリューションを参照してください。

承認

作成者 - Aneel Kanuri (Distinguished Cloud Architect)

その他の学習リソース

docs.oracle.com/learnの他のラボを確認するか、Oracle Learning YouTubeチャネルで無料のラーニング・コンテンツにアクセスしてください。また、education.oracle.com/learning-explorerにアクセスしてOracle Learning Explorerになります。

製品ドキュメントは、Oracle Help Centerを参照してください。

タイトルおよび著作権情報

Write Events to an OCI Private Stream using OCI Events Service and OCI Functions

G23236-01

December 2024

Oracle and/or its affiliates.