주:

이 자습서에서는 Oracle Cloud에 액세스해야 합니다. 무료 계정에 등록하려면 Oracle Cloud Infrastructure Free Tier 시작하기를 참조하십시오.
Oracle Cloud Infrastructure 자격 증명, 테넌시 및 구획에 예제 값을 사용합니다. 실습을 완료했으면 이러한 값을 자신의 클라우드 환경과 관련된 값으로 대체하십시오.

Oracle Cloud Infrastructure Functions를 사용하여 Oracle Cloud Infrastructure Private Stream에 로그 작성

소개

로그는 최신 클라우드 운영의 초석이며 시스템 활동, 성능 및 보안에 대한 중요한 통찰력을 제공합니다. 중요한 데이터를 처리하는 기업의 경우 로그를 안전하고 효율적으로 관리하는 것이 가장 중요합니다. 이 사용지침서에서는 Oracle Cloud Infrastructure(OCI)를 사용하여 로그 관리를 위한 안전하고 확장 가능하며 이벤트 기반 아키텍처를 구축하는 방법을 안내합니다.

이 솔루션의 핵심은 공용 엔드포인트에 노출되지 않고 로그가 안전하게 전송되고 처리되도록 하는 OCI Functions입니다. 이 아키텍처는 OCI Connector Hub를 통해 OCI Object Storage로 수집 및 압축된 로그의 이벤트에 의해 구동됩니다. 로그 파일이 생성될 때마다 OCI Functions가 트리거되어 압축을 풀고 관리 가능한 조각으로 나누어 보안 저장 영역 또는 추가 처리를 위해 로그를 전용(private) 스트림에 게시합니다.

이 이벤트 기반 접근 방식은 구성요소가 변화에 동적으로 반응하여 대기 시간을 최소화하고 효율성을 극대화하는 원활하고 자동화된 파이프라인을 보장합니다. 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 Object Storage, OCI Connector Hub, OCI Event Service 규칙, OCI 로그, OCI Functions 및 OCI 스트림을 관리합니다. OCI Vault(선택사항) 및 네트워크 제품군에서 사용합니다.
서비스 정책: OCI Object Storage 버킷(읽은 객체)에서 메시지를 읽고 스트림에 쓸 수 있는 권한을 함수에 부여합니다(stream-push 사용). 동적 그룹이 필요합니다.

자세한 정책은 여기에서 확인할 수 있습니다.

작업 2: 전용 스트림 생성

OCI Streaming은 미사용 및 전송 중에 데이터가 암호화되어 메시지의 무결성과 보안을 보장하는 완전 관리형 OCI 서비스입니다. 향상된 보안을 위해 OCI Vault 서비스를 사용하여 고유의 암호화 키를 저장 및 관리하고 특정 준수 또는 보안 요구 사항을 충족할 수 있습니다. 프라이빗 끝점은 VCN(가상 클라우드 네트워크) 내에서 구성하여 스트림 보안을 강화하고 프라이빗 IP 주소를 스트림 풀에 연관시킬 수 있습니다. 따라서 OCI Streaming 트래픽이 VCN 내에 유지되므로 인터넷을 완전히 피할 수 있습니다. 그러나 전용 엔드포인트를 사용하는 스트림은 인터넷에서 액세스할 수 없으므로 콘솔을 통해 최신 메시지를 볼 수 있는 기능이 제한됩니다. 개인 스트림에서 메시지를 소비하려면 소비자가 개인 스트림이 호스팅되는 네트워크에 대한 경로 및 액세스 권한을 모두 가져야 합니다.

스트림 및 스트림 풀을 생성합니다. 스트림 이름을 입력하고 새 스트림 풀 생성을 선택하여 스트림 풀을 생성합니다. 스트림 풀 구성 섹션에서 스트림 풀 이름을 입력하고 프라이빗 끝점을 선택한 다음 VCN 및 서브넷과 그에 따라 네트워크 세부정보를 입력합니다. 선택 사항이지만 네트워크 보안 그룹에 해당 NSG 내의 모든 트래픽에 대한 수신 규칙을 제공하는 것이 좋습니다. 자세한 내용은 스트림 생성 및 스트림 풀 생성을 참조하십시오.

NSG 규칙 이미지

스트림 및 스트림 풀 이미지

자체 암호화 키를 사용하여 키의 수명 주기를 보다 효과적으로 제어할 수 있습니다. 스트림 내의 메시지 보존을 조정할 수 있는 옵션이 있습니다. 기본값은 1일이고 최대값은 7일입니다.

스트림 암호화 이미지

완료된 스트림 풀 이미지

스트림 OCID 및 메시지 끝점을 적어 둡니다. 이 정보를 함수에 전달해야 합니다.

작업 3: OCI Connector Hub 생성 및 구성

OCI Connector Hub는 소스와 대상 간의 원활하고 안정적인 데이터 전송을 지원하는 보안 메시지 버스 역할을 합니다. 이 아키텍처에서는 소스가 OCI Logging이고, 대상은 OCI Object Storage이며, 이후 처리를 위해 해당 로그가 압축되고 저장됩니다. 중개자 역할을 하는 OCI Connector Hub는 전송된 메시지의 보안 및 무결성을 유지하면서 효율적인 데이터 흐름을 보장합니다.

이 사용지침서에서는 서브넷에서 플로우 로그가 사용으로 설정되었고 OCI Object Storage 버킷을 사용할 수 있다고 가정합니다. 플로우 로그를 사용으로 설정하고 버킷을 생성하는 방법은 플로우 로그를 사용으로 설정 및 오브젝트 스토리지 버킷 생성을 참조하십시오.

버킷을 생성하는 동안 객체 이벤트 생략을 선택해야 합니다. 이것이 이벤트 중심 아키텍처의 핵심입니다.

버킷 이미지

OCI 커넥터 허브를 구성하여 OCI 로깅 서비스와 OCI 오브젝트 스토리지 버킷 간에 데이터 플로우를 생성합니다. 자세한 내용은 로깅 소스를 사용하여 커넥터 생성을 참조하십시오.

사진 SCH

일괄 처리 롤오버 세부정보를 조정하여 OCI Object Storage 버킷에 로그가 기록되는 빈도를 구성할 수 있습니다. 기본값은 100MB 또는 7분입니다.

작업 4: 함수 개발 및 배치

이 함수는 OCI Object Storage에서 객체를 읽고 스트림에 메시지를 기록합니다. 이를 위해 다음 작업을 수행합니다.

버킷에서 객체를 읽습니다.
객체의 압축을 풉니다.
필요한 경우 객체 크기를 확인하고 1MB 청크를 생성합니다. OCI Streaming 서비스는 생산자가 스트림에 게시할 수 있는 고유 메시지의 최대 크기를 1MB로 제한합니다.
메시지를 인코딩합니다.
스트림에 게시합니다.

자세한 내용은 함수 생성을 참조하십시오.

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 함수 및 OCI 프라이빗 스트림 기능 사용에 대한 자세한 내용은 Oracle 담당자에게 문의하거나 클라우드 보안 솔루션을 참조하십시오.

확인

작성자 - Aneel Kanuri(Distinguished Cloud Architect)

추가 학습 자원

docs.oracle.com/learn에서 다른 실습을 탐색하거나 Oracle Learning YouTube 채널에서 더 많은 무료 학습 콘텐츠에 액세스하세요. 또한 Oracle Learning Explorer가 되려면 education.oracle.com/learning-explorer을 방문하십시오.

제품 설명서는 Oracle Help Center를 참조하십시오.

제목 및 저작권 정보

Write Logs to Oracle Cloud Infrastructure Private Stream using Oracle Cloud Infrastructure Functions

G23250-01

December 2024

Oracle and/or its affiliates.