Note :

Ce tutoriel nécessite l'accès à Oracle Cloud. Pour vous inscrire à un compte gratuit, voir Introduction à l' niveau gratuit d'Oracle Cloud Infrastructure.
Il utilise des exemples de valeurs pour les données d'identification, la location et les compartiments Oracle Cloud Infrastructure. À la fin de votre laboratoire, remplacez ces valeurs par celles propres à votre environnement en nuage.

Écrire des journaux dans un flux privé Oracle Cloud Infrastructure à l'aide d'Oracle Cloud Infrastructure Functions

Présentation

Les journaux sont la pierre angulaire des opérations infonuagiques modernes, fournissant des renseignements essentiels sur l'activité, la performance 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 basée sur les événements pour la gestion des journaux à l'aide d'Oracle Cloud Infrastructure (OCI).

Au cœur de cette solution se trouve le service des fonctions pour OCI, qui garantit que les journaux sont transmis et traités en toute sécurité sans être exposés à des points d'extrémité publics. L'architecture est guidée par des événements à partir des journaux collectés et compressés dans le stockage d'objets OCI au moyen du centre de connecteurs OCI. Chaque fois qu'un fichier journal est créé, il déclenche un service des fonctions pour OCI, qui le décompose, le décompose en fragments gérables et publie les journaux dans un flux privé pour un stockage sécurisé ou un traitement ultérieur.

Cette approche axée sur les événements garantit un pipeline automatisé et transparent où les composants réagissent dynamiquement aux changements, réduisant ainsi la latence et optimisant l'efficacité. En tirant parti des services natifs d'OCI, vous obtiendrez un cadre sécurisé et évolutif permettant de traiter des journaux à volume élevé en temps réel.

Objectif

Mettez en œuvre un flux de travail de gestion des journaux résilient et sécurisé conçu pour les entreprises qui cherchent à protéger les données de journalisation sensibles et à optimiser les opérations dans leur environnement en nuage.

Préalables

Accès à une location OCI.
Privilèges pour gérer le stockage d'objets OCI, les journaux OCI, le centre de connecteurs OCI, les règles du service d'événements OCI, les applications Oracle et les services de diffusion en continu OCI.

Tâche 1 : Configurer les politiques requises et les autorisations 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.

Politiques d'utilisateur : Gérer le stockage d'objets OCI, le centre de connecteurs OCI, les règles du service d'événements OCI, les journaux OCI, le service des fonctions pour OCI et les flux OCI. À utiliser dans la chambre forte OCI (facultatif) et la famille de réseaux.
Politique de service : Accordez à la fonction l'autorisation de lire des messages à partir du seau de stockage d'objets OCI (lire des objets) et d'écrire dans un flux (utiliser la poussée de flux). Un groupe dynamique est requis.

Vous trouverez des politiques détaillées ici :

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

Le service de diffusion en continu pour OCI est un service OCI entièrement géré dans lequel les données sont chiffrées au repos et en transit, ce qui assure l'intégrité et la sécurité des messages. Pour une sécurité améliorée, vous pouvez utiliser le service de chambre forte OCI pour stocker et gérer vos propres clés de chiffrement, en respectant des exigences de conformité ou de sécurité spécifiques. Des points d'extrémité privés peuvent être configurés dans un réseau en nuage virtuel (VCN) pour sécuriser davantage vos flux, en associant une adresse IP privée au groupe de flux. Cela garantit que le trafic du service de diffusion en continu pour OCI reste dans le VCN, évitant ainsi entièrement Internet. Toutefois, notez que les flux utilisant des points d'extrémité privés ne sont pas accessibles sur Internet, ce qui limite la possibilité de voir leurs derniers messages au moyen de la console. Pour consommer des messages à partir d'un flux privé, le consommateur doit avoir à la fois une route et un accès au réseau où le flux privé est hébergé.

Créez un flux et un groupe de flux. Entrez le nom du flux et sélectionnez Créer un nouveau groupe de flux pour créer un groupe de flux. Dans la section Configurer le groupe de flux, entrez Nom du groupe de flux, sélectionnez Point d'extrémité privé et entrez le VCN, le sous-réseau et les détails du réseau en conséquence. Bien que facultatif, nous vous recommandons de fournir un groupe de sécurité de réseau avec une règle de trafic entrant pour tout le trafic au sein de ce groupe de sécurité de réseau. Pour plus d'informations, voir Créer un flux et Créer un groupe de flux.

Illustration de la règle de NSG

Image du flux et du groupe de flux

Vous pouvez utiliser vos propres clés de chiffrement, 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 chiffrement de flux

Image du groupe de flux terminé

Notez l'OCID du flux et le point d'extrémité des messages. Nous devons transmettre ces informations à la fonction.

Tâche 3 : Créer et configurer le centre de connecteurs OCI

Le centre de connecteurs OCI 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 le service de journalisation OCI et la destination est le service de stockage d'objets OCI, où ces journaux sont compressés et stockés pour un traitement ultérieur. En servant d'intermédiaire, le centre de connecteurs OCI assure un flux de données efficace tout en assurant 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 seau de stockage d'objets OCI est disponible. Pour plus d'informations sur l'activation des journaux de flux et la création d'un seau, voir Activation des journaux de flux et Création d'un seau de stockage d'objets.

Lors de la création d'un seau, assurez-vous de sélectionner Émettre des événements d'objet. C'est la clé de notre architecture événementielle.

Image du seau

Configurez le concentrateur de connecteurs OCI, en créant un flux de données entre le service de journalisation OCI et le seau de stockage d'objets OCI. Pour plus d'informations, voir Création d'un connecteur avec une source de journalisation.

Image du SCH

En ajustant les détails de report par lots, vous pouvez configurer la fréquence à laquelle les journaux sont écrits dans le seau de stockage d'objets OCI. La valeur par défaut est 100 Mo ou 7 minutes.

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

Cette fonction lit l'objet à partir du service de stockage d'objets OCI et écrit le message dans le flux. Pour ce faire, il effectue les opérations suivantes entre :

Lisez l'objet à partir du seau.
Décompressez l'objet.
Vérifiez la taille de l'objet et créez des fragments de 1 Mo, si nécessaire. Le service de diffusion en continu pour OCI limite la taille maximale d'un message unique que les producteurs peuvent publier dans un flux à 1 Mo.
Encodez le message.
Publier dans le flux.

Pour plus d'informations, voir 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
```

Illustration 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.

Illustration de la configuration de la fonction

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

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 du groupe de journaux pour vérifier si les journaux de flux sont ingérés.
Le saut suivant correspond aux mesures du concentrateur de connecteurs. Le centre de connecteurs OCI collecte les journaux et les envoie au service de stockage d'objets OCI. Assurez-vous qu'il n'y a aucune erreur à la source et à la cible.
Le saut suivant est le stockage d'objets OCI. 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 d'événements OCI. 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 aucune erreur et que la fonction n'est pas ralentie.
Vérifiez que les données sont ingérées dans le flux privé.

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

Étapes suivantes

Félicitations pour la mise en oeuvre réussie d'une solution de gestion des journaux sécurisée et basée sur les événements dans OCI! En combinant la puissance de la journalisation OCI, du centre de connecteurs OCI, du stockage d'objets OCI et des flux privés OCI, vous avez créé une architecture robuste qui garantit que vos journaux sont collectés, traités et publiés en temps quasi réel.

Cette solution protège les données de journalisation sensibles par le biais de 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 en toute transparence, ce qui vous permet de gérer de grands volumes de journaux avec une intervention manuelle minimale.

Grâce à ce cadre, vous pouvez assurer un traitement des journaux sécurisé et efficace tout en respectant les exigences en matière de confidentialité. Cette architecture offre la flexibilité nécessaire pour créer des pipelines de traitement personnalisés adaptés aux besoins de votre entreprise. Étendre cette configuration à l'aide d'analyses ou de mécanismes d'alerte supplémentaires peut vous aider à mieux comprendre les événements du système et à détecter les anomalies et à y réagir de manière proactive.

Pour plus d'informations sur l'utilisation du service des fonctions pour OCI et des capacités de flux privé OCI, communiquez avec votre représentant Oracle ou consultez Solutions de sécurité en nuage.

Confirmation

Auteur - Aneel Kanuri (architecte en nuage distinct)

Autres ressources d'apprentissage

Explorez d'autres laboratoires sur la page docs.oracle.com/learn ou accédez à plus de contenu d'apprentissage gratuit sur le canal YouTube d'Oracle Learning. De plus, visitez education.oracle.com/learning-explorer pour devenir un explorateur Oracle Learning.

Pour obtenir de la documentation sur le produit, visitez Oracle Help Center.

Titre et Copyright

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

G23252-01

December 2024

Oracle et/ou ses sociétés affiliées.