Remarques :

Ce tutoriel nécessite un accès à Oracle Cloud. Pour vous inscrire à un compte gratuit, reportez-vous à Introduction au niveau gratuit d'Oracle Cloud Infrastructure.
Il utilise des exemples de valeurs pour les informations d'identification, la location et les compartiments Oracle Cloud Infrastructure. Lorsque vous terminez votre atelier, remplacez ces valeurs par celles propres à votre environnement cloud.

Ecrire des journaux dans le flux privé Oracle Cloud Infrastructure à l'aide d'Oracle Cloud Infrastructure Functions

Introduction

Les journaux sont la pierre angulaire des opérations cloud innovantes, fournissant des informations essentielles sur l'activité, les performances et la sécurité du système. Pour les entreprises qui gèrent des données sensibles, la gestion sécurisée et efficace des journaux est primordiale. Ce tutoriel vous aidera à créer une architecture sécurisée, évolutive et orientée événements pour la gestion des journaux à l'aide d'Oracle Cloud Infrastructure (OCI).

OCI Functions est au cœur de cette solution. Elle garantit la transmission et le traitement sécurisés des journaux sans exposition aux adresses publiques. L'architecture est guidée par les événements provenant des journaux collectés et compressés dans OCI Object Storage via OCI Connector Hub. Chaque fois qu'un fichier journal est créé, il déclenche OCI Functions, qui le décompresse, le divise en blocs gérables et publie les journaux dans un flux privé pour un stockage sécurisé ou un traitement ultérieur.

Cette approche basée sur les événements garantit un pipeline transparent et automatisé où les composants réagissent dynamiquement aux changements, minimisant la latence et optimisant l'efficacité. En tirant parti des services natifs d'OCI, vous obtiendrez une structure sécurisée et évolutive pour gérer le traitement de journaux en grand volume en temps réel.

Objectif

Implémentez un workflow de gestion des journaux résilient et sécurisé conçu pour les entreprises qui cherchent à protéger les données de journaux sensibles et à optimiser les opérations dans leur environnement cloud.

Prérequis

Accès à une location OCI.
Privilèges permettant de gérer OCI Object Storage, les journaux OCI, OCI Connector Hub, les règles OCI Event Service, les applications Oracle et les services OCI Streaming.

Tâche 1 : configuration des stratégies requises et des droits d'accès Oracle Cloud Infrastructure Identity and Access Management (OCI IAM)

Chaque composant de cette solution doit avoir accès aux ressources OCI avec lesquelles il interagit. Pour suivre ce tutoriel, les autorisations suivantes sont requises.

Stratégies utilisateur : gérez OCI Object Storage, OCI Connector Hub, les règles OCI Event Service, les journaux OCI, OCI Functions et les flux OCI. Utiliser sur OCI Vault (facultatif) et la famille de réseaux.
Stratégie de service : autorisez la fonction à lire les messages du bucket OCI Object Storage (objets en lecture) et à écrire dans un flux de données (utilisez stream-push). Un groupe dynamique est requis.

Des politiques détaillées peuvent être trouvées ici :

Tâche 2 : créer un flux privé

OCI Streaming est un service OCI entièrement géré dans lequel les données sont cryptées au repos et en transit, garantissant ainsi l'intégrité et la sécurité des messages. Pour améliorer la sécurité, vous pouvez utiliser le service OCI Vault pour stocker et gérer vos propres clés de cryptage, en respectant des exigences de conformité ou de sécurité spécifiques. Les adresses privées peuvent être configurées au sein d'un réseau cloud virtuel (VCN) pour sécuriser davantage vos flux de données, en associant une adresse IP privée au pool de flux de données. Cela garantit que le trafic OCI Streaming reste dans le VCN, évitant ainsi entièrement Internet. Toutefois, les flux utilisant des adresses privées ne sont pas accessibles à partir d'Internet, ce qui limite la possibilité de visualiser leurs derniers messages via la console. Pour utiliser les messages d'un flux privé, le destinataire doit disposer à la fois d'un routage et d'un accès au réseau sur lequel le flux privé est hébergé.

Créez un flux de données et un pool de flux de données. Entrez le nom du flux de données et sélectionnez Créer un pool de flux de données pour créer un pool de flux de données. Dans la section Configurer le pool de flux de données, entrez le nom du pool de flux de données, sélectionnez Adresse privée et entrez le VCN et le sous-réseau, ainsi que les détails du réseau en conséquence. Bien que facultatif, nous vous recommandons de fournir à un groupe de sécurité réseau une règle entrante pour tout le trafic au sein de ce groupe de sécurité réseau. Pour plus d'informations, reportez-vous à Création d'un flux et à Création d'un pool de flux de données.

Image de la règle de groupe de sécurité réseau

Image du flux de données et du pool de flux de données

Vous pouvez utiliser vos propres clés de cryptage, ce qui vous permet de mieux contrôler le cycle de vie de la clé. Vous avez la possibilité d'ajuster la conservation des messages dans le flux. La valeur par défaut est 1 jour et la valeur maximale est 7 jours.

Image du cryptage de flux

Image du pool de flux de données terminé

Notez l'OCID de flux de données et l'adresse de messages. Nous devons transmettre ces informations à la fonction.

Tâche 3 : création et configuration d'OCI Connector Hub

OCI Connector Hub agit comme un bus de messages sécurisé, facilitant le transfert de données transparent et fiable entre une source et une destination. Dans cette architecture, la source est OCI Logging et la destination est OCI Object Storage, où ces journaux sont compressés et stockés pour un traitement ultérieur. En servant d'intermédiaire, OCI Connector Hub garantit un flux de données efficace tout en maintenant la sécurité et l'intégrité des messages transférés.

Ce tutoriel suppose que les journaux de flux sont activés sur les sous-réseaux et qu'un bucket OCI Object Storage est disponible. Pour plus d'informations sur l'activation des journaux de flux et la création d'un bucket, reportez-vous à Activation des journaux de flux et à Création d'un bucket Object Storage.

Lors de la création d'un bucket, veillez à sélectionner Emettre des événements d'objet. C'est la clé de notre architecture événementielle.

Image du bucket

Configurez le hub de connecteur OCI en créant un flux de données entre le service OCI Logging et le bucket OCI Object Storage. Pour plus d'informations, reportez-vous à Création d'un connecteur avec une source Logging.

Image du SCH

En ajustant les détails de report de batch, vous pouvez configurer la fréquence à laquelle les journaux sont écrits dans le bucket OCI Object Storage. La valeur par défaut est de 100 Mo ou 7 minutes.

Tâche 4 : développer et déployer la fonction

Cette fonction lit l'objet à partir d'OCI Object Storage et écrit le message dans le flux. Pour cela, il effectue les opérations suivantes entre les deux :

Lisez l'objet du bucket.
Décompressez l'objet.
Vérifiez la taille de l'objet et créez des blocs de 1 Mo, si nécessaire. Le service OCI Streaming limite la taille maximale d'un message unique que les émetteurs peuvent publier dans un flux de données de 1 Mo.
Codez le message.
Publier dans le flux.

Pour plus d'informations, reportez-vous à la section Création de fonctions.

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 de la fonction

La dernière étape consiste à indiquer à la fonction où se trouve le flux privé. Cette fonction utilise des paramètres de configuration, ce qui la rend réutilisable si vous souhaitez effectuer un déploiement dans une autre location.

Image de la configuration de fonction

Dans cette tâche, abonnez la fonction à l'événement de téléchargement d'objet. Créez une règle pour Type d'événement en tant qu'Objet - Créer avec le nom de bucket en tant qu'attribut conditionnel. Pour plus d'informations, reportez-vous à Création d'une règle Events.

Image de la configuration de l'événement

Vérification

Il existe plusieurs endroits où le flux de données peut être vérifié.

Vérifiez les mesures de groupe de journaux pour vérifier si les journaux de flux sont inclus.
Le saut suivant correspond aux mesures du hub de connecteur. OCI Connector Hub collecte les journaux et les envoie à OCI Object Storage. Assurez-vous qu'il n'y a aucune erreur à la source et à la cible.
Le saut suivant est OCI Object Storage. Assurez-vous que le nombre d'objets augmente. Si nécessaire, activez les journaux de lecture et d'écriture pour poursuivre le débogage.
Le saut suivant est le service OCI Events. Vérifiez les mesures pour vous assurer qu'il n'y a pas d'échec de livraison.
L'étape suivante consiste à vérifier les mesures d'appel de la fonction. Assurez-vous qu'il n'y a pas d'erreur et que la fonction ne ralentit pas.
Vérifiez que les données sont en cours d'inclusion dans le flux privé.

Si les données sont absentes dans l'un des graphiques suivants, arrêtez-vous là et activez les journaux pour ce service. Les journaux expliquent pourquoi une ressource spécifique ne parvient pas à effectuer la tâche.

Etapes suivantes

Félicitations pour avoir implémenté avec succès une solution de gestion des journaux sécurisée et orientée événements dans OCI ! En combinant la puissance d'OCI Logging, d'OCI Connector Hub, d'OCI Object Storage et de flux privés OCI, vous avez créé une architecture robuste qui garantit que vos journaux sont collectés, traités et publiés de manière sécurisée en temps quasi réel.

Cette solution protège les données de journal sensibles via des flux privés et démontre l'efficacité de l'automatisation basée sur les événements. Au fur et à mesure que votre système évolue, cette architecture s'adapte de manière transparente, vous permettant de gérer de grands volumes de journaux avec une intervention manuelle minimale.

Avec cette structure en place, vous pouvez assurer un traitement des journaux sécurisé et efficace tout en maintenant la conformité aux exigences de confidentialité. Cette architecture permet de créer des pipelines de traitement personnalisés adaptés aux besoins de votre entreprise. L'extension de cette configuration avec des analyses ou des mécanismes d'alerte supplémentaires peut vous permettre d'obtenir des informations plus précises sur les événements système et d'améliorer votre capacité à détecter les anomalies et à y réagir de manière proactive.

Pour plus d'informations sur l'utilisation d'OCI Functions et des fonctionnalités de flux privé OCI, contactez votre représentant Oracle ou consultez les solutions de sécurité cloud.

Remerciements

Auteur - Aneel Kanuri (architecte cloud distingué)

Ressources de formation supplémentaires

Explorez d'autres ateliers sur docs.oracle.com/learn ou accédez à d'autres contenus de formation gratuits sur le canal Oracle Learning YouTube. De plus, visitez le site education.oracle.com/learning-explorer pour devenir un explorateur Oracle Learning.

Pour obtenir la documentation produit, consultez le site Oracle Help Center.

Informations relatives au titre et au copyright

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

G23247-01

December 2024

Oracle et/ou ses affiliés.