Note:

This tutorial requires access to Oracle Cloud. To sign up for a free account, see Get started with Oracle Cloud Infrastructure Free Tier.
It uses example values for Oracle Cloud Infrastructure credentials, tenancy, and compartments. When completing your lab, substitute these values with ones specific to your cloud environment.

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

Introduction

Logs are a cornerstone of modern cloud operations, providing critical insights into system activity, performance, and security. For enterprises handling sensitive data, managing logs securely and efficiently is paramount. This tutorial will guide you in building a secure, scalable, and event-driven architecture for log management using Oracle Cloud Infrastructure (OCI).

At the heart of this solution is OCI Functions, which ensures logs are transmitted and processed securely without exposure to public endpoints. The architecture is driven by events from logs collected and compressed into OCI Object Storage through OCI Connector Hub. Each time a log file is created, it triggers an OCI Functions, which uncompresses it, breaks it into manageable chunks, and publishes the logs to a private stream for secure storage or further processing.

This event-driven approach ensures a seamless, automated pipeline where components react dynamically to changes, minimizing latency and maximizing efficiency. By leveraging OCI’s native services, you will achieve a secure and scalable framework to handle high-volume log processing in real-time.

Objective

Implement a resilient and secure log management workflow designed for enterprises looking to protect sensitive log data and optimize operations in their cloud environment.

Prerequisites

Access to an OCI tenancy.
Privileges to manage OCI Object Storage, OCI logs, OCI Connector Hub, OCI Event Service rules, Oracle Applications, and OCI Streaming services.

Task 1: Set Up the Required Policies and Oracle Cloud Infrastructure Identity and Access Management (OCI IAM) Permissions

Each component of this solution must have access to the OCI resources it interacts with. To follow this tutorial, the following permissions are required.

User Policies: Manage OCI Object Storage, OCI Connector Hub, OCI Event Service rules, OCI logs, OCI Functions, and OCI streams. Use on OCI Vault (optional) and network family.
Service Policy: Grant the function permission to read messages from the OCI Object Storage bucket (read objects) and write to a stream (use stream-push). A dynamic group is required.

Detailed policies can be found here:

Task 2: Create a Private Stream

OCI Streaming is a fully managed OCI service in which data is encrypted at rest and in transit, ensuring the integrity and security of messages. For enhanced security, you can use the OCI Vault service to store and manage your own encryption keys, meeting specific compliance or security requirements. Private endpoints can be configured within a Virtual Cloud Network (VCN) to further secure your streams, associating a private IP address to the stream pool. This ensures that OCI Streaming traffic stays within the VCN, avoiding the internet entirely. However, note that streams using private endpoints are not accessible from the internet, limiting the ability to view their latest messages through the console. To consume messages from a private stream, the consumer must have both route and access to the network where the private stream is hosted.

Create a stream and stream pool. Enter Stream Name and select Create New Stream Pool to create stream pool. In the Configure Stream Pool section, enter Stream Pool Name, select Private Endpoint and enter VCN and Subnet and network details accordingly. Though optional, we recommend providing a Network Security Group with an ingress rule for all traffic within that NSG. For more information, see Create Stream and Create Stream Pool.

Image of the NSG Rule

Image of the Stream and Stream Pool

You can use your own encryption keys, gaining greater control over the key’s lifecycle. You have an option to adjust the retention of messages within the stream. The default is 1 day, and the maximum is 7 days.

Image of the Stream Encryption

Image of the Completed Stream Pool

Note down the stream OCID and the messages endpoint. We need to pass this information on to the function.

Task 3: Create and Configure OCI Connector Hub

OCI Connector Hub acts as a secure message bus, facilitating the seamless and reliable data transfer between a source and a destination. In this architecture, the source is OCI Logging, and the destination is OCI Object Storage, where these logs are compressed and stored for further processing. By serving as an intermediary, the OCI Connector Hub ensures efficient data flow while maintaining the security and integrity of the transferred messages.

This tutorial assumes that flow logs are enabled on the subnets and an OCI Object Storage bucket is available. For more information to enable flow logs and create a bucket, see Enabling Flow Logs and Creating an Object Storage Bucket.

While creating a bucket, ensure to select Emit Object Events. This is the key for our event-driven architecture.

Image of the Bucket

Configure the OCI Connector hub, creating a data flow between the OCI Logging service and the OCI Object storage bucket. For more information, see Creating a Connector with a Logging Source.

Image of the SCH

By adjusting the batch rollover details, you can configure the frequency with which the logs are written to the OCI Object Storage bucket. The default is 100 MB or 7 minutes.

Task 4: Develop and Deploy the Function

This function will read the object from the OCI Object Storage and write the message to the stream. To achieve that, it performs the following operations in between:

Read the object from the bucket.
Uncompress the object.
Check the object size and create 1 MB chunks, if needed. OCI Streaming service limits the maximum size of a unique message that producers can publish to a stream is 1 MB.
Encode the message.
Publish to the stream.

For more information, see Creating functions.

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
```

Image of the Function

The final step is to tell the function where the private stream is. This function uses configuration parameters, making it reusable if you want to deploy in another tenancy.

Image of the Function Config

In this task, subscribe the function to the object upload event. Create a rule for Event Type as Object - Create with the bucket name as a conditional attribute. For more information, see Creating an Events Rule.

Image of the Event Config

Verification

There are multiple places where the data flow can be verified.

Verify the log group metrics to check if the flow logs are ingested.
The next hop is the connector hub metrics. OCI Connector Hub gathers the logs and send them to the OCI Object Storage. Make sure there are no errors at source and target.
The next hop is the OCI Object Storage. Ensure the object count is increasing. If required, enable the read and write logs to debug further.
The next hop is the OCI Events Service. Review the metrics to ensure there are no delivery failures.
The next step is to check the function Invocation metrics. Ensure there are no errors and the function is not throttling.
Check that the data is being ingested into the private stream.

If the data is absent in any of the following charts, stop there and enable logs for that service. Logs will explain why a specific resource is failing to perform the task.

Next Steps

Congratulations on successfully implementing a secure and event-driven log management solution in OCI! By combining the power of OCI Logging, OCI Connector Hub, OCI Object Storage, and OCI private streams, you have created a robust architecture that ensures your logs are securely collected, processed, and published in near real-time.

This solution safeguards sensitive log data through private streams and demonstrates the efficiency of event-driven automation. As your system scales, this architecture will adapt seamlessly, enabling you to handle large volumes of logs with minimal manual intervention.

With this framework in place, you can ensure secure and efficient log processing while maintaining compliance with privacy requirements. This architecture provides flexibility for building custom processing pipelines tailored to your enterprise needs. Extending this setup with additional analytics or alerting mechanisms can gain deeper insights into system events and enhance your ability to detect and respond to anomalies proactively.

For more information about using OCI Functions and OCI private stream capabilities, contact your Oracle representative or see Cloud Security Solutions.

Acknowledgments

Author - Aneel Kanuri (Distinguished Cloud Architect)

More Learning Resources

Explore other labs on docs.oracle.com/learn or access more free learning content on the Oracle Learning YouTube channel. Additionally, visit education.oracle.com/learning-explorer to become an Oracle Learning Explorer.

For product documentation, visit Oracle Help Center.

Title and Copyright Information

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

G23046-01

December 2024

Oracle and/or its affiliates.

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

Introduction

Objective

Prerequisites

Task 1: Set Up the Required Policies and Oracle Cloud Infrastructure Identity and Access Management (OCI IAM) Permissions

Task 2: Create a Private Stream

Task 3: Create and Configure OCI Connector Hub

Task 4: Develop and Deploy the Function

Task 5: Create an Event and Subscribe to the Function

Verification

Next Steps

Acknowledgments

More Learning Resources