Nota

Questa esercitazione richiede l'accesso a Oracle Cloud. Per iscriverti a un account gratuito, consulta Inizia a utilizzare Oracle Cloud Infrastructure Free Tier.
Utilizza valori di esempio per le credenziali, la tenancy e i compartimenti di Oracle Cloud Infrastructure. Al termine del laboratorio, sostituisci questi valori con quelli specifici del tuo ambiente cloud.

Scrivere i log in Oracle Cloud Infrastructure Private Stream utilizzando Oracle Cloud Infrastructure Functions

Introduzione

I log sono una pietra miliare delle moderne operazioni cloud e forniscono insight strategici su attività, performance e sicurezza del sistema. Per le aziende che gestiscono dati sensibili, è fondamentale gestire i log in modo sicuro ed efficiente. Questa esercitazione descrive le procedure per creare un'architettura sicura, scalabile e basata sugli eventi per la gestione dei log mediante Oracle Cloud Infrastructure (OCI).

Al centro di questa soluzione ci sono le OCI Functions, che garantiscono la trasmissione e l'elaborazione sicura dei log senza esposizione agli endpoint pubblici. L'architettura è guidata dagli eventi provenienti dai log raccolti e compressi nello storage degli oggetti OCI tramite OCI Connector Hub. Ogni volta che un file di log viene creato, attiva un OCI Functions, che lo decomprime, lo suddivide in chunk gestibili e pubblica i log in un flusso privato per uno storage sicuro o un'ulteriore elaborazione.

Questo approccio basato sugli eventi garantisce una pipeline trasparente e automatizzata in cui i componenti reagiscono dinamicamente ai cambiamenti, riducendo al minimo la latenza e massimizzando l'efficienza. Sfruttando i servizi nativi di OCI, otterrai un framework sicuro e scalabile per gestire l'elaborazione dei log a volumi elevati in tempo reale.

Obiettivo

Implementa un flusso di lavoro di gestione dei log resiliente e sicuro progettato per le aziende che desiderano proteggere i dati di log riservati e ottimizzare le operazioni nel proprio ambiente cloud.

Prerequisiti

Accesso a una tenancy OCI.
Privilegi per gestire lo storage degli oggetti OCI, i log OCI, OCI Connector Hub, le regole del servizio eventi OCI, le applicazioni Oracle e i servizi di streaming OCI.

Task 1: impostare i criteri richiesti e le autorizzazioni Oracle Cloud Infrastructure Identity and Access Management (OCI IAM)

Ogni componente di questa soluzione deve avere accesso alle risorse OCI con cui interagisce. Per seguire questa esercitazione, sono necessarie le seguenti autorizzazioni.

Criteri utente: gestisci OCI Object Storage, OCI Connector Hub, regole OCI Event Service, log OCI, OCI Functions e flussi OCI. Usare su OCI Vault (facoltativo) e sulla famiglia di rete.
Criterio servizio: concedere l'autorizzazione della funzione per leggere i messaggi dal bucket di storage degli oggetti OCI (oggetti di lettura) e scrivere in un flusso (utilizzare stream-push). È necessario un gruppo dinamico.

Le politiche dettagliate possono essere trovate qui:

Task 2: Creare un flusso privato

OCI Streaming è un servizio OCI completamente gestito in cui i dati vengono cifrati in archivio e in transito, garantendo l'integrità e la sicurezza dei messaggi. Per una maggiore sicurezza, puoi utilizzare il servizio OCI Vault per memorizzare e gestire le tue chiavi di cifratura, soddisfacendo requisiti specifici di conformità o sicurezza. Gli endpoint privati possono essere configurati all'interno di una rete cloud virtuale (VCN) per proteggere ulteriormente i flussi, associando un indirizzo IP privato al pool di flussi. Ciò garantisce che il traffico di streaming OCI rimanga all'interno della VCN, evitando completamente Internet. Tuttavia, si noti che i flussi che utilizzano endpoint privati non sono accessibili da Internet, limitando la possibilità di visualizzare i messaggi più recenti tramite la console. Per utilizzare i messaggi da un flusso privato, il consumatore deve disporre sia dell'instradamento che dell'accesso alla rete in cui è ospitato il flusso privato.

Creare un flusso e un pool di flussi. Immettere Nome flusso e selezionare Crea nuovo pool di flussi per creare il pool di flussi. Nella sezione Configura pool di flussi, immettere Nome pool di streaming, selezionare Endpoint privato e immettere di conseguenza la VCN e la subnet e i dettagli di rete. Sebbene sia facoltativo, si consiglia di fornire un gruppo di sicurezza di rete con una regola di entrata per tutto il traffico all'interno di tale gruppo NSG. Per ulteriori informazioni, vedere Crea flusso e Crea pool di flussi.

Immagine della regola NSG

Immagine del flusso e del pool di flussi

Puoi utilizzare le tue chiavi di cifratura, ottenendo un maggiore controllo sul ciclo di vita della chiave. È possibile regolare la conservazione dei messaggi all'interno del flusso. L'impostazione predefinita è 1 giorno e il massimo è 7 giorni.

Immagine della cifratura del flusso

Immagine del pool di flussi completato

Prendere nota dell'OCID del flusso e dell'endpoint dei messaggi. Dobbiamo trasmettere queste informazioni alla funzione.

Task 3: Creare e configurare OCI Connector Hub

OCI Connector Hub funge da bus di messaggi sicuro, facilitando il trasferimento dei dati trasparente e affidabile tra un'origine e una destinazione. In questa architettura, l'origine è OCI Logging e la destinazione è OCI Object Storage, dove questi log vengono compressi e memorizzati per un'ulteriore elaborazione. Facendo da intermediario, OCI Connector Hub garantisce un flusso di dati efficiente mantenendo la sicurezza e l'integrità dei messaggi trasferiti.

In questa esercitazione si presume che i log di flusso siano abilitati nelle subnet e che sia disponibile un bucket di storage degli oggetti OCI. Per ulteriori informazioni su come abilitare i log di flusso e creare un bucket, vedere Abilitazione dei log di flusso e Creazione di un bucket di storage degli oggetti.

Durante la creazione di un bucket, assicurarsi di selezionare Emetti eventi oggetto. Questa è la chiave per la nostra architettura basata sugli eventi.

Immagine del bucket

Configurare l'hub OCI Connector, creando un flusso di dati tra il servizio OCI Logging e il bucket di storage degli oggetti OCI. Per ulteriori informazioni, vedere Creazione di un connettore con un'origine di log.

Immagine della SCH

Modificando i dettagli del rollover batch, puoi configurare la frequenza di scrittura dei log nel bucket di storage degli oggetti OCI. L'impostazione predefinita è 100 MB o 7 minuti.

Task 4: Sviluppare e distribuire la funzione

Questa funzione leggerà l'oggetto dallo storage degli oggetti OCI e scriverà il messaggio nel flusso. Per raggiungere questo obiettivo, esegue le seguenti operazioni tra:

Leggere l'oggetto dal bucket.
Decomprimere l'oggetto.
Controllare la dimensione dell'oggetto e creare chunk da 1 MB, se necessario. Il servizio di streaming OCI limita la dimensione massima di un messaggio univoco che i producer possono pubblicare in un flusso pari a 1 MB.
Codificare il messaggio.
Pubblica nel flusso.

Per ulteriori informazioni, vedere Creazione di funzioni.

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

Immagine della funzione

L'ultimo passo è dire alla funzione dove si trova il flusso privato. Questa funzione utilizza i parametri di configurazione, rendendola riutilizzabile se si desidera eseguire la distribuzione in un'altra tenancy.

Immagine della configurazione funzione

In questo task, eseguire la sottoscrizione della funzione all'evento di caricamento dell'oggetto. Creare una regola per Tipo di evento come Oggetto - Crea con il nome del bucket come attributo condizionale. Per ulteriori informazioni, vedere Creazione di una regola di eventi.

Immagine della configurazione evento

Verifica

Esistono più punti in cui è possibile verificare il flusso di dati.

Verificare le metriche del gruppo di log per verificare se i log di flusso sono inclusi.
L'hop successivo sono le metriche dell'hub connettore. OCI Connector Hub raccoglie i log e li invia a OCI Object Storage. Assicurarsi che non vi siano errori nell'origine e nella destinazione.
L'hop successivo è lo storage degli oggetti OCI. Assicurarsi che il conteggio degli oggetti stia aumentando. Se necessario, abilitare i log di lettura e scrittura per eseguire ulteriormente il debug.
L'hop successivo è OCI Events Service. Esaminare le metriche per assicurarsi che non si verifichino errori di consegna.
Il passo successivo consiste nel controllare le metriche di richiamo della funzione. Assicurarsi che non vi siano errori e che la funzione non sia in fase di limitazione.
Controllare che i dati vengano inclusi nel flusso privato.

Se i dati sono assenti in uno dei grafici riportati di seguito, fermarsi e abilitare i log per tale servizio. I log spiegano perché una risorsa specifica non riesce a eseguire l'attività.

Passi successivi

Congratulazioni per aver implementato con successo una soluzione di gestione dei log sicura e basata sugli eventi in OCI! Combinando la potenza di OCI Logging, OCI Connector Hub, OCI Object Storage e i flussi privati OCI, hai creato un'architettura solida che garantisce che i log vengano raccolti, elaborati e pubblicati in modo sicuro quasi in tempo reale.

Questa soluzione salvaguarda i dati di log sensibili attraverso flussi privati e dimostra l'efficienza dell'automazione basata sugli eventi. Man mano che il sistema si ridimensiona, questa architettura si adatta perfettamente, consentendo di gestire grandi volumi di log con un intervento manuale minimo.

Grazie a questo framework, puoi garantire un'elaborazione dei log sicura ed efficiente mantenendo la conformità ai requisiti di privacy. Questa architettura offre la flessibilità necessaria per creare pipeline di elaborazione personalizzate in base alle esigenze aziendali. L'estensione di questa impostazione con ulteriori meccanismi di analisi o avviso può ottenere insight più approfonditi sugli eventi di sistema e migliorare la tua capacità di rilevare e rispondere alle anomalie in modo proattivo.

Per ulteriori informazioni sull'uso di OCI Functions e delle funzionalità di flusso privato OCI, contatta il tuo rappresentante Oracle o consulta Soluzioni di sicurezza cloud.

Conferme

Autore - Aneel Kanuri (architetto cloud distinto)

Altre risorse di apprendimento

Esplora altri laboratori su docs.oracle.com/learn o accedi a più contenuti gratuiti sulla formazione su Oracle Learning YouTube channel. Inoltre, visita education.oracle.com/learning-explorer per diventare un Oracle Learning Explorer.

Per la documentazione del prodotto, visita l'Oracle Help Center.

Titolo e informazioni sul copyright

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

G23249-01

December 2024

Oracle e/o relative consociate.