附註：

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

使用 Oracle Cloud Infrastructure Functions 將日誌寫入 Oracle Cloud Infrastructure 專用串流

簡介

日誌是現代雲端作業的基石，提供對系統活動、效能和安全性的重要洞察力。對於處理機密資料的企業，安全且有效率地管理日誌至關重要。本教學課程將引導您使用 Oracle Cloud Infrastructure (OCI) 為日誌管理建立安全、可擴展且事件導向的架構。

此解決方案的核心是 OCI Functions，可確保在不暴露於公用端點的情況下安全地傳輸和處理日誌。此架構是由透過 OCI Connector Hub 收集日誌並壓縮至 OCI Object Storage 的事件所驅動。每次建立日誌檔時，它會觸發 OCI 函數來解壓縮、將其分成可管理的區塊，然後將日誌發布至專用串流以進行安全儲存或進一步處理。

此事件導向方法可確保順暢的自動化管線，讓元件能夠動態回應變化、將延遲降到最低並發揮最大效率。透過利用 OCI 的原生服務，您將實現安全且可擴展的架構，以即時處理大量日誌處理。

目標

導入專為想要保護機密日誌資料並最佳化雲端環境作業之企業設計的彈性安全日誌管理工作流程。

必要條件

存取 OCI 租用戶。
管理 OCI Object Storage、OCI 日誌、OCI Connector Hub、OCI Event Service 規則、Oracle Applications 和 OCI Streaming 服務的權限。

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

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

使用者原則：管理 OCI 物件儲存、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 Connector Hub 是安全的訊息匯流排，有助於在來源和目的地之間進行無縫可靠的資料傳輸。在此架構中，來源為 OCI 日誌記錄，目的地為 OCI 物件儲存，其中會壓縮並儲存這些日誌以供進一步處理。OCI Connector Hub 是中介服務，可確保資料流有效率，同時維持傳輸訊息的安全性和完整性。

本教學課程假設子網路已啟用流量日誌，並且提供 OCI 物件儲存的儲存桶。如需啟用流量日誌和建立儲存桶的詳細資訊，請參閱啟用流量日誌和建立物件儲存的儲存桶。

建立儲存桶時，請確定選取省略物件事件。這是我們事件導向架構的關鍵。

儲存桶的映像檔

設定 OCI 連線器中心，在 OCI 日誌記錄服務與 OCI 物件儲存的儲存桶之間建立資料流程。如需詳細資訊，請參閱使用記錄來源建立連線器。

SCH 的影像

透過調整批次變換詳細資訊，您可以設定日誌寫入 OCI 物件儲存的儲存桶的頻率。預設值為 100 MB 或 7 分鐘。

作業 4：開發並部署函數

此函數會從 OCI 物件儲存讀取物件，並將訊息寫入串流。為達到此目的，它會執行下列作業：

讀取儲存桶中的物件。
解壓縮物件。
檢查物件大小，並視需要建立 1 MB 的區塊。OCI Streaming 服務限制產生器可以發布到串流的唯一訊息大小上限為 1 MB。
編碼訊息。
發布至串流。

如需詳細資訊，請參閱建立函數。

func.py:

import io
import json
import logging
import oci
import gzip
from base64 import b64encode

def handler(ctx, data: io.BytesIO = None):
    try:
        # Parse the incoming data
        cfg = ctx.Config()
        body = json.loads(data.getvalue())
        bucket_name = body["data"]["additionalDetails"]["bucketName"]
        object_name = body["data"]["resourceName"]
        stream_ocid = cfg["stream_ocid"]
        stream_endpoint = cfg["stream_endpoint"]

        logging.getLogger().info(f'Function invoked for bucket upload: {bucket_name}')
    except (Exception, ValueError) as ex:
        logging.getLogger().error(f'Error parsing JSON payload: {str(ex)}')
        return {"status": 400, "message": "Bad Request"}

    try:
        # Get the object data from Object Storage
        object_content = get_object(bucket_name, object_name)

        # Check if the object is a .gz file and decompress it
        if object_name.endswith('.gz'):
            logging.getLogger().info(f'Decompressing object: {object_name}')
            object_content = gzip.decompress(object_content)

        logging.getLogger().info(f'Object Content: {object_content.decode("utf-8")[:100]}...')

        # Split content into message chunks ensuring no message is split
        messages = split_content_into_messages(object_content, max_size=1024*1024)

        # Publish messages to the stream
        for message in messages:
            publish_to_stream(stream_ocid, stream_endpoint, data.getvalue().decode('utf-8'), message)

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

def get_object(bucket_name, object_name):
    signer = oci.auth.signers.get_resource_principals_signer()
    object_storage_client = oci.object_storage.ObjectStorageClient(config={}, signer=signer)
    namespace = object_storage_client.get_namespace().data

    try:
        logging.getLogger().info(f'Searching for bucket: {bucket_name}, object: {object_name}')
        obj = object_storage_client.get_object(namespace, bucket_name, object_name)
        logging.getLogger().info(f'Found object: {object_name}')
        return obj.data.content
    except Exception as ex:
        logging.getLogger().error(f'Failed to retrieve object: {str(ex)}')
        raise

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

    # Build the message list
    message_list = [
        oci.streaming.models.PutMessagesDetailsEntry(
            key=b64encode("partition-key-1".encode()).decode(),
            value=b64encode(object_content).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

def split_content_into_messages(content, max_size):
    """
    Splits content into messages of specified max size while ensuring no message is split.
    Attempts to split based on newline characters for text content.
    """
    messages = []
    current_chunk = []
    current_size = 0

    for line in content.decode('utf-8').splitlines(keepends=True):
        line_size = len(line.encode('utf-8'))
        if current_size + line_size > max_size:
            messages.append(''.join(current_chunk).encode('utf-8'))
            current_chunk = [line]
            current_size = line_size
        else:
            current_chunk.append(line)
            current_size += line_size

    if current_chunk:
        messages.append(''.join(current_chunk).encode('utf-8'))

    return messages

func.yaml:

schema_version: 20180708
name: logs-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
```

函數的影像

最後一步是告訴功能專用串流所在的位置。此函數使用組態參數，若想要在其他租用戶部署，可重複使用。

函數組態的影像

在此任務中，訂閱物件上傳事件的函數。建立事件類型的規則作為物件 - 建立，並將儲存桶名稱作為條件屬性。如需詳細資訊，請參閱建立事件規則。

事件組態的影像

驗證

有多個地方可以驗證資料流程。

請檢查日誌群組度量，以檢查是否已攝取流量日誌。
下一個躍點是連線器中心度量。OCI Connector Hub 會收集日誌並將其傳送到 OCI Object Storage。確定來源和目標沒有錯誤。
下一個躍點是 OCI Object Storage。確定物件數目增加。如有需要，請啟用讀取和寫入日誌以進一步除錯。
下一個躍點是 OCI 事件服務。請複查度量，以確定沒有任何傳遞失敗。
下一步是檢查函數呼叫測量結果。確定沒有錯誤，且函數不會進行節流。
檢查資料是否正攝取至專用串流。

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

接下來的步驟

恭喜您成功在 OCI 中導入安全且事件導向的日誌管理解決方案！透過結合 OCI Logging、OCI Connector Hub、OCI Object Storage 和 OCI 私有串流的強大功能，您建立了強大的架構，確保您的日誌安全地收集、處理和發佈在近乎即時。

此解決方案透過私有串流保護敏感記錄資料，並展示事件驅動自動化的效率。隨著系統擴展，此架構將無縫適應，使您只需最少的手動介入即可處理大量的日誌。

有了此架構，您就可以確保安全且有效率的日誌處理，同時維持符合隱私權需求的規範。此架構可讓您彈性建置專為企業需求量身打造的自訂處理管線。使用額外的分析或警示機制擴充此設定，可獲得更深入的系統事件洞察分析，並增強主動偵測和回應異常的能力。

如需有關使用 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 Logs to Oracle Cloud Infrastructure Private Stream using Oracle Cloud Infrastructure Functions

G23246-01

December 2024

Oracle 和 (或) 其關係企業。