附註：

此教學課程需要存取 Oracle Cloud。若要註冊免費帳戶，請參閱 Oracle Cloud Infrastructure Free Tier 入門。
它使用 Oracle Cloud Infrastructure 證明資料、租用戶及區間的範例值。完成實驗室時，請將這些值取代為您雲端環境特定的值。

使用 OCI 事件服務和 OCI 函數將事件寫入 OCI 專用串流

簡介

Oracle Cloud Infrastructure (OCI) 事件服務規則會使用符合 Cloud Native Computing Foundation (CNCF) CloudEvents 標準的事件，追蹤 OCI 中的資源變更。這可讓開發人員使用 OCI Functions 觸發程式碼、將資料寫入 OCI Streaming 或使用 OCI Notifications 傳送警示，以回應即時變更。

原生支援將事件傳送至公用串流時，需要使用 OCI Functions 或自訂程式碼自訂方法，才能寫入專用串流。本教學課程示範使用 OCI 函數從 OCI 事件服務規則擷取事件有效負載，然後將它們寫入專用串流。

目標

自動設定處理 OCI 事件，並以可擴展的方式將它們傳送到 OCI 專用串流，以進行安全、自訂的處理。

必要條件

存取 OCI 租用戶。
管理 OCI 事件服務規則、Oracle Applications 以及 OCI 串流服務的權限。

作業 1：設定必要的原則和 Oracle Cloud Infrastructure Identity and Access Management (OCI IAM) 權限

此解決方案的每個元件都必須能夠存取與其互動的 OCI 資源。若要遵循此自學課程，開發人員必須具有下列權限。

使用者原則： 管理 OCI 事件服務規則、OCI 通知、OCI 函數、OCI 串流，以及 OCI 保存庫 (選擇性) 和網路系列的使用。
服務原則： 授予將訊息寫入 OCI 專用串流的功能權限。

可在此處找到詳細的原則：

作業 2：建立專用串流

OCI Streaming 是一項完全託管的 OCI 服務，在靜態和傳輸中加密資料，確保訊息的完整性和安全性。為了增強安全性，您可以使用 OCI Vault 服務來儲存和管理您自己的加密金鑰、符合特定規範或安全需求。您可以在虛擬雲端網路 (VCN) 內設定專用端點，以進一步保護串流，建立專用 IP 位址與串流集區的關聯。這樣可確保 OCI Streaming 流量保持在 VCN 內，進而避免整個網際網路。不過請注意，使用專用端點的串流無法從網際網路存取，這也限制了透過主控台檢視最新訊息的能力。若要使用來自專用串流的訊息，用戶必須擁有代管專用串流之網路的路由和存取權。

建立串流和串流集區。輸入串流名稱，然後選取建立新串流集區以建立串流集區。在設定串流集區段落中，輸入串流集區名稱，選取專用端點，然後相應地輸入 VCN 和子網路以及網路詳細資訊。雖然選用，但是建議您為 NSG 內的所有流量提供網路安全群組的傳入規則。如需詳細資訊，請參閱建立串流和建立串流集區。

NSG 規則的影像

串流與串流集區建立的映像檔

您可以使用自己的加密金鑰，更有效地控制金鑰的生命週期。您可以選擇調整串流內訊息的保留。預設值為 1 天，上限為 7 天。

串流加密的映像檔

串流集區的映像檔

請記下串流 OCID 和訊息端點。我們需要將此資訊傳遞至功能。

作業 3：建立 OCI 通知主題和 OCI 事件服務規則

重新整理者是事件的運作方式。OCI 服務會發出資源或資料的事件。我們會建立規則，其中包含指定相關事件的篩選。當篩選找到相符的事件時，規則必須觸發動作。此動作可以是 OCI 通知主題、OCI 串流服務或 OCI 函數。

約 1：如果需求是將事件有效負載傳送至公用串流，則不需要函數，因為可以原生設定流程。
約 2：如果需求是只將特定事件有效負載傳送至專用串流，則可直接將函數提供為動作。
約 3：如果要求將多個事件有效負載傳送至專用串流，則最理想的方式是將事件有效負載傳送至 OCI 通知主題，並將 OCI 函數訂閱 OCI 通知主題。

在本教學課程中，我們將探索最複雜的 Approach 3。

建立 OCI 通知主題。如需詳細資訊，請參閱建立主題。
建立事件規則。為了說明此方法，我們將建立兩個事件規則。如需詳細資訊，請參閱建立事件規則。

作業 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 通知主題度量。確定所有訊息都已傳遞；失敗的訊息圖表中沒有資料。
檢查函數呼叫測量結果。確定沒有錯誤，且函數不會進行節流。
檢查資料是否正攝取至專用串流。

串流測量結果的影像

如果下列任一圖表中沒有資料，請停止該處並啟用該服務的日誌。日誌將會說明特定資源無法執行作業的原因。

接下來的步驟

在本教學課程中，您將瞭解如何整合 OCI 事件服務、OCI 通知、OCI 函數及 OCI Streaming，以安全地處理事件有效負載並將其發布到私有串流。您探索了設定安全端點、管理加密金鑰，以及使用 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

G23241-01

December 2024

Oracle 和 (或) 其關係企業。