ノート:

OCI関数を使用したOCIコネクタ・ハブを介したプライベート・エンドポイント・ストリームへのログのプッシュ

イントロダクション

Oracle Cloud Infrastructure (OCI) Functionsは、完全に管理された、マルチテナントでスケーラビリティが高いオンデマンドのFunctions-as-a-Serviceプラットフォームです。ビジネス・ニーズを満たすためのコードの記述に集中する場合は、OCI Functionsを使用します。

OCI Connector Hubは、OCIのサービス間でデータを転送するのに役立ちます。OCI Connector Hubは、プライベート・エンドポイント・ストリームをターゲットとしてサポートしていません。

OCIストリーミング・サービスは、大量のデータ・ストリームをリアルタイムで収集および消費するための、スケーラブルで耐久性の高いフルマネージド・ソリューションを提供します。OCI Streamingは、パブリッシュ/サブスクライブ・メッセージング・モデルでデータが連続して順番に生成および処理されます。

このチュートリアルでは、OCI Connector HubのソースとターゲットはそれぞれOCIロギングとOCI関数になります。OCI Connector Hubは、新しいログをチェックします。検出されると、ログがOCI Functionsに転送され、ファンクションがトリガーされます。OCI Functionsは、それらのログをプライベート・エンドポイント・ストリームにプッシュします。

アーキテクチャ

目的

前提条件

タスク1: OCI Functionsを使用したファンクションの作成

次のコードを使用して、プライベート・ストリームが作成される同じプライベート・サブネットにファンクションを作成します。次のコードは、OCI Object Storageログの例です。コードでは、stream-ocidをプライベート・エンドポイント・ストリームのOracle Cloud Identifier (OCID)およびmessage-endpointをプライベート・ストリームのメッセージ・エンドポイントURLに置き換える必要があります。bucketName、message、clientIpAddressなどのログから抽出するパラメータはほとんどありません。詳細は、ファンクションの作成およびデプロイを参照してください。

ノート:このファンクション・コードは、OCIオブジェクト・ストレージ・ログをプライベート・エンドポイント・ストリームにプッシュする例です。異なるログで要件に従ってPythonスクリプトをカスタマイズする必要があります。

#!/bin/bash
# Copyright (c) 2016, 2021, Oracle and/or its affiliates.  All rights reserved.
# This software is dual-licensed to you under the Universal Permissive License (UPL) 1.0 as shown at https://oss.oracle.com/licenses/upl or Apache License 2.0 as shown at http://www.apache.org/licenses/LICENSE-2.0. You may choose either license.

import io
import oci
import logging
import json
import os
from fdk import response
from base64 import b64encode, b64decode

def handler(ctx, data: io.BytesIO = None):
    signer_auth = oci.auth.signers.get_resource_principals_signer()
    streaming_client = oci.streaming.StreamClient(config = {}, service_endpoint = "<message-endpoint>", signer=signer_auth)

    logger = logging.getLogger()
    logger.info("function start")

    try:
      logentries = json.loads(data.getvalue()) # deserialize the bytesstream input as JSON array
      if not isinstance(logentries, list):
          logger.error('Invalid connector payload. No log queries detected')
          raise
      logger.info("json input from SCH")
      logger.info(data.getvalue())

      for logEntry in logentries:
          logger.info("Extracting/Parse log details from the log entry json")
          bucketName = logEntry["data"]["bucketName"]
          message = logEntry["data"]["message"]
          clientIpAddress = logEntry["data"]["clientIpAddress"]
          log_line = bucketName + '\t' + message + '\t' + clientIpAddress
          encoded_message = b64encode(log_line.encode()).decode()
          put_messages_response = streaming_client.put_messages(
            stream_id="<stream-ocid>",
            put_messages_details=oci.streaming.models.PutMessagesDetails(
                messages=[
                    oci.streaming.models.PutMessagesDetailsEntry(value=encoded_message)]))

      logger.info("function end")
      return
    except Exception as e:
      logger.error("Failure in the function: {}".format(str(e)))
      raise

タスク2: OCIコネクタ・ハブの作成および設定

  1. 「コネクタの作成」をクリックします。

    コネクタの作成

  2. コンパートメント名を入力し、ソースとしてプッシュするログを選択します。

    ソース:

  3. 作成したファンクションをターゲットとして選択し、「作成」をクリックします。

    Target

OCI Connector Hubは、ソースの新しいログを探します。新しいログが検出されると、OCI Connector HubはそのログをOCI Functionsに送信します。ファンクションがトリガーされ、ログがプライベート・エンドポイント・ストリームにプッシュされます。

確認

その他の学習リソース

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

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