Note:

Este tutorial requiere acceso a Oracle Cloud. Para registrarse para obtener una cuenta gratuita, consulte Introducción a la cuenta gratuita de Oracle Cloud Infrastructure.
Utiliza valores de ejemplo para credenciales, arrendamiento y compartimentos de Oracle Cloud Infrastructure. Al finalizar la práctica, sustituya estos valores por otros específicos de su entorno en la nube.

Escritura de logs en el flujo privado de Oracle Cloud Infrastructure mediante Oracle Cloud Infrastructure Functions

Introducción

Los logs son la piedra angular de las operaciones modernas en la nube, ya que proporcionan estadísticas esenciales sobre la actividad, el rendimiento y la seguridad del sistema. Para las empresas que manejan datos confidenciales, la gestión de registros de forma segura y eficiente es primordial. Este tutorial le guiará en la creación de una arquitectura segura, escalable y controlada por eventos para la gestión de logs mediante Oracle Cloud Infrastructure (OCI).

En el corazón de esta solución se encuentra OCI Functions, que garantiza que los logs se transmitan y procesen de forma segura sin exposición a puntos finales públicos. La arquitectura se basa en eventos de logs recopilados y comprimidos en OCI Object Storage a través de OCI Connector Hub. Cada vez que se crea un archivo log, dispara OCI Functions, que lo descomprime, lo divide en fragmentos gestionables y publica los logs en un flujo privado para un almacenamiento seguro o un procesamiento posterior.

Este enfoque basado en eventos garantiza un pipeline sin fisuras y automatizado donde los componentes reaccionan dinámicamente a los cambios, minimizando la latencia y maximizando la eficiencia. Al aprovechar los servicios nativos de OCI, conseguirás un marco seguro y escalable para gestionar el procesamiento de logs de gran volumen en tiempo real.

Objetivo

Implante un flujo de trabajo de gestión de logs resistente y seguro diseñado para empresas que buscan proteger los datos de log confidenciales y optimizar las operaciones en su entorno en la nube.

Requisitos

Acceso a un arrendamiento de OCI.
Privilegios para gestionar OCI Object Storage, logs de OCI, OCI Connector Hub, reglas de OCI Event Service, Oracle Applications y servicios OCI Streaming.

Tarea 1: Configuración de las políticas necesarias y los permisos de Oracle Cloud Infrastructure Identity and Access Management (OCI IAM)

Cada componente de esta solución debe tener acceso a los recursos de OCI con los que interactúa. Para seguir este tutorial, se necesitan los siguientes permisos.

Políticas de usuario: gestiona OCI Object Storage, OCI Connector Hub, reglas de OCI Event Service, logs de OCI, OCI Functions y flujos de OCI. Se utiliza en OCI Vault (opcional) y en la familia de redes.
Política de servicio: otorgue a la función permiso para leer mensajes del cubo de OCI Object Storage (objetos de lectura) y escribir en un flujo (usar stream-push). Se necesita un grupo dinámico.

Las políticas detalladas se pueden encontrar aquí:

Tarea 2: Creación de un flujo privado

OCI Streaming es un servicio OCI totalmente gestionado en el que los datos se cifran estáticos y en tránsito, lo que garantiza la integridad y seguridad de los mensajes. Para mejorar la seguridad, puede utilizar el servicio OCI Vault para almacenar y gestionar sus propias claves de cifrado, cumpliendo requisitos específicos de conformidad o seguridad. Los puntos finales privados se pueden configurar en una red virtual en la nube (VCN) para proteger aún más sus flujos, asociando una dirección IP privada al pool de flujos. Esto garantiza que el tráfico de OCI Streaming permanezca dentro de la VCN, evitando Internet por completo. Sin embargo, tenga en cuenta que no se puede acceder a los flujos que utilizan puntos finales privados desde Internet, lo que limita la capacidad de ver sus últimos mensajes a través de la consola. Para consumir mensajes de un flujo privado, el consumidor debe tener acceso y ruta a la red donde se aloja el flujo privado.

Crear un flujo y un pool de flujos. Introduzca Stream Name y seleccione Create New Stream Pool para crear el pool de flujos. En la sección Configurar pool de flujos, introduzca Nombre de pool de flujos, seleccione Punto final privado e introduzca VCN y Subred y los detalles de red según corresponda. Aunque es opcional, recomendamos proporcionar a un grupo de seguridad de red una regla de entrada para todo el tráfico dentro de ese NSG. Para obtener más información, consulte Creación de flujo y Creación de pool de flujos.

Imagen de la regla de NSG

Imagen del flujo y del pool de flujos

Puede utilizar sus propias claves de cifrado para obtener un mayor control sobre el ciclo de vida de la clave. Tiene la opción de ajustar la retención de mensajes en el flujo. El valor por defecto es 1 día y el máximo es 7 días.

Imagen del cifrado de flujo

Imagen del pool de flujos terminado

Anote el OCID de flujo y el punto final de mensajes. Necesitamos pasar esta información a la función.

Tarea 3: Creación y configuración de OCI Connector Hub

OCI Connector Hub actúa como un bus de mensajes seguro, lo que facilita la transferencia de datos fluida y fiable entre un origen y un destino. En esta arquitectura, el origen es OCI Logging y el destino es OCI Object Storage, donde estos logs se comprimen y almacenan para su posterior procesamiento. Al servir como intermediario, OCI Connector Hub garantiza un flujo de datos eficiente al tiempo que mantiene la seguridad e integridad de los mensajes transferidos.

En este tutorial se asume que los logs de flujo están activados en las subredes y que hay disponible un cubo de OCI Object Storage. Para obtener más información sobre cómo activar logs de flujo y crear un cubo, consulte Activación de logs de flujo y Creación de un cubo de Object Storage.

Al crear un cubo, asegúrese de seleccionar Emitir eventos de objeto. Esta es la clave de nuestra arquitectura basada en eventos.

Imagen del cubo

Configure el hub de OCI Connector, creando un flujo de datos entre el servicio OCI Logging y el cubo de OCI Object Storage. Para obtener más información, consulte Creación de un conector con un origen de log.

Imagen de la SCH

Al ajustar los detalles de renovación de lotes, puede configurar la frecuencia con la que se escriben los logs en el cubo de OCI Object Storage. El valor predeterminado es 100 MB o 7 minutos.

Tarea 4: Desarrollo y despliegue de la función

Esta función leerá el objeto de OCI Object Storage y escribirá el mensaje en el flujo. Para lograrlo, realiza las siguientes operaciones entre:

Lea el objeto del cubo.
Descomprima el objeto.
Compruebe el tamaño del objeto y cree fragmentos de 1 MB, si es necesario. El servicio OCI Streaming limita el tamaño máximo de un único mensaje que los productores pueden publicar en un flujo a 1 MB.
Codifique el mensaje.
Publique en el flujo.

Para obtener más información, consulte Creación de funciones.

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

Imagen de la función

El paso final es indicar a la función dónde está el flujo privado. Esta función utiliza parámetros de configuración, por lo que es reutilizable si desea desplegar en otro arrendamiento.

Imagen de la configuración de función

En esta tarea, suscriba la función al evento de carga de objeto. Cree una regla para Tipo de evento como Objeto - Crear con el nombre del cubo como atributo condicional. Para obtener más información, consulte Creating an Events Rule.

Imagen de la configuración del evento

Verificación

Hay varios lugares donde se puede verificar el flujo de datos.

Verifique las métricas del grupo de logs para comprobar si se ingieren los logs de flujo.
El siguiente salto son las métricas del hub de conector. OCI Connector Hub recopila los logs y los envía a OCI Object Storage. Asegúrese de que no haya errores en el origen y el destino.
El siguiente salto es OCI Object Storage. Asegúrese de que el recuento de objetos está aumentando. Si es necesario, active los logs de lectura y escritura para continuar con la depuración.
El siguiente salto es el servicio OCI Events. Revise las métricas para asegurarse de que no hay fallos de entrega.
El siguiente paso es comprobar las métricas de la función Llamada. Asegúrese de que no hay errores y de que la función no está limitando.
Compruebe que los datos se están ingiriendo en el flujo privado.

Si los datos están ausentes en cualquiera de los siguientes gráficos, deténgase allí y active los logs para ese servicio. Los logs explicarán por qué un recurso específico no puede realizar la tarea.

Pasos Siguientes

¡Felicidades por implementar con éxito una solución de gestión de logs segura y basada en eventos en OCI! Al combinar la potencia de OCI Logging, OCI Connector Hub, OCI Object Storage y flujos privados de OCI, ha creado una arquitectura sólida que garantiza que los logs se recopilen, procesen y publiquen de forma segura casi en tiempo real.

Esta solución protege los datos confidenciales de log a través de flujos privados y demuestra la eficiencia de la automatización basada en eventos. A medida que el sistema se amplíe, esta arquitectura se adaptará sin problemas, lo que le permitirá manejar grandes volúmenes de logs con una intervención manual mínima.

Con este marco, puede garantizar un procesamiento de logs seguro y eficaz, al tiempo que mantiene el cumplimiento de los requisitos de privacidad. Esta arquitectura proporciona flexibilidad para crear pipelines de procesamiento personalizados adaptados a las necesidades de su empresa. La ampliación de esta configuración con análisis o mecanismos de alerta adicionales puede obtener estadísticas más detalladas sobre los eventos del sistema y mejorar su capacidad de detectar y responder a las anomalías de forma proactiva.

Para obtener más información sobre el uso de OCI Functions y las capacidades de flujos privados de OCI, póngase en contacto con su representante de Oracle o consulte Soluciones de seguridad en la nube.

Agradecimientos

Autor: Aneel Kanuri (arquitecto de nube distinguido)

Más recursos de aprendizaje

Explore otros laboratorios en docs.oracle.com/learn o acceda a más contenido de formación gratuita en el canal YouTube de Oracle Learning. Además, visita education.oracle.com/learning-explorer para convertirte en un Oracle Learning Explorer.

Para obtener documentación sobre el producto, visite Oracle Help Center.

Título e Información de Copyright

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

G23248-01

December 2024

Oracle y/o sus filiales.