Hinweis:

Dieses Tutorial erfordert Zugriff auf Oracle Cloud. Informationen zur Registrierung für einen kostenlosen Account finden Sie unter Erste Schritte mit Oracle Cloud Infrastructure Free Tier.
Es verwendet Beispielwerte für Oracle Cloud Infrastructure-Zugangsdaten, -Mandanten und -Compartments. Ersetzen Sie diese Werte nach Abschluss der Übung durch Werte, die für Ihre Cloud-Umgebung spezifisch sind.

Logs mit Oracle Cloud Infrastructure Functions in privaten Oracle Cloud Infrastructure-Stream schreiben

Einführung

Logs sind ein Eckpfeiler moderner Cloud-Vorgänge und bieten wichtige Einblicke in Systemaktivität, Performance und Sicherheit. Für Unternehmen, die sensible Daten verarbeiten, ist die sichere und effiziente Verwaltung von Protokollen von größter Bedeutung. In diesem Tutorial erfahren Sie, wie Sie eine sichere, skalierbare und ereignisgesteuerte Architektur für das Logmanagement mit Oracle Cloud Infrastructure (OCI) erstellen.

Das Herzstück dieser Lösung ist OCI Functions, das sicherstellt, dass Logs sicher übertragen und verarbeitet werden, ohne öffentliche Endpunkte zu gefährden. Die Architektur wird durch Ereignisse aus Logs gesteuert, die über OCI Connector Hub in OCI Object Storage erfasst und komprimiert werden. Jedes Mal, wenn eine Logdatei erstellt wird, löst sie eine OCI Functions aus, die sie dekomprimiert, in verwaltbare Chunks aufteilt und die Logs in einem privaten Stream für sicheren Speicher oder weitere Verarbeitung veröffentlicht.

Dieser ereignisgesteuerte Ansatz gewährleistet eine nahtlose, automatisierte Pipeline, in der Komponenten dynamisch auf Änderungen reagieren, Latenz minimieren und die Effizienz maximieren. Durch die Nutzung der nativen Services von OCI erhalten Sie ein sicheres und skalierbares Framework für die Verarbeitung von Logs mit hohem Volumen in Echtzeit.

Zielsetzung

Implementieren Sie einen robusten und sicheren Logmanagement-Workflow, der für Unternehmen entwickelt wurde, die sensible Logdaten schützen und den Betrieb in ihrer Cloud-Umgebung optimieren möchten.

Voraussetzungen

Zugriff auf einen OCI-Mandanten.
Berechtigungen zur Verwaltung von OCI Object Storage, OCI-Logs, OCI Connector Hub, OCI Event Service-Regeln, Oracle Applications und OCI Streaming-Services.

Aufgabe 1: Erforderliche Policys und Oracle Cloud Infrastructure Identity and Access Management-(OCI IAM-)Berechtigungen einrichten

Jede Komponente dieser Lösung muss Zugriff auf die OCI-Ressourcen haben, mit denen sie interagiert. Um diesem Tutorial zu folgen, sind die folgenden Berechtigungen erforderlich.

Benutzer-Policys: Verwalten Sie OCI Object Storage, OCI Connector Hub, OCI Event Service-Regeln, OCI-Logs, OCI Functions und OCI-Streams. Verwendung in OCI Vault (optional) und der Netzwerkfamilie.
Service-Policy: Erteilen Sie der Funktionsberechtigung, Nachrichten aus dem OCI Object Storage-Bucket (Objekte lesen) zu lesen und in einen Stream zu schreiben (Stream-Push verwenden). Eine dynamische Gruppe ist erforderlich.

Detaillierte Richtlinien finden Sie hier:

Aufgabe 2: Privaten Stream erstellen

OCI Streaming ist ein vollständig verwalteter OCI-Service, in dem Daten im Ruhezustand und während der Übertragung verschlüsselt werden, um die Integrität und Sicherheit von Nachrichten sicherzustellen. Um die Sicherheit zu erhöhen, können Sie mit dem OCI Vault-Service Ihre eigenen Verschlüsselungsschlüssel speichern und verwalten, um bestimmte Compliance- oder Sicherheitsanforderungen zu erfüllen. Private Endpunkte können in einem virtuellen Cloud-Netzwerk (VCN) konfiguriert werden, um Ihre Streams weiter zu sichern und eine private IP-Adresse mit dem Streampool zu verknüpfen. Dadurch wird sichergestellt, dass der OCI Streaming-Traffic im VCN bleibt, sodass das Internet vollständig vermieden wird. Beachten Sie jedoch, dass Streams, die private Endpunkte verwenden, nicht über das Internet zugänglich sind, was die Anzeige der neuesten Nachrichten über die Konsole einschränkt. Um Nachrichten aus einem privaten Stream zu konsumieren, muss der Consumer sowohl über die Route als auch über Zugriff auf das Netzwerk verfügen, in dem der private Stream gehostet wird.

Erstellen Sie einen Stream und einen Streampool. Geben Sie den Streamnamen ein, und wählen Sie Neuen Streampool erstellen aus, um einen Streampool zu erstellen. Geben Sie im Abschnitt Streampool konfigurieren den Streampoolnamen ein, wählen Sie Privater Endpunkt aus, und geben Sie VCN und Subnetz sowie Netzwerkdetails entsprechend ein. Obwohl optional, wird empfohlen, eine Netzwerksicherheitsgruppe mit einer Ingress-Regel für den gesamten Traffic innerhalb dieser NSG bereitzustellen. Weitere Informationen finden Sie unter Stream erstellen und Streampool erstellen.

Bild der NSG-Regel

Image des Streams und des Streampools

Sie können Ihre eigenen Verschlüsselungsschlüssel verwenden, um mehr Kontrolle über den Lebenszyklus des Schlüssels zu erhalten. Sie können die Aufbewahrung von Nachrichten im Stream anpassen. Der Standardwert ist 1 Tag, der Höchstwert ist 7 Tage.

Image der Streamverschlüsselung

Bild des abgeschlossenen Streampools

Notieren Sie sich die Stream-OCID und den Nachrichtenendpunkt. Wir müssen diese Informationen an die Funktion weitergeben.

Aufgabe 3: OCI Connector Hub erstellen und konfigurieren

OCI Connector Hub fungiert als sicherer Nachrichtenbus und erleichtert die nahtlose und zuverlässige Datenübertragung zwischen einer Quelle und einem Ziel. In dieser Architektur ist die Quelle OCI Logging, und das Ziel ist OCI Object Storage, wo diese Logs komprimiert und zur weiteren Verarbeitung gespeichert werden. Der OCI Connector Hub dient als Vermittler und stellt einen effizienten Datenfluss sicher, während die Sicherheit und Integrität der übertragenen Nachrichten gewahrt bleibt.

In diesem Tutorial wird davon ausgegangen, dass Flowlogs in den Subnetzen aktiviert sind und ein OCI Object Storage-Bucket verfügbar ist. Weitere Informationen zum Aktivieren von Flowlogs und Erstellen eines Buckets finden Sie unter Flowlogs aktivieren und Object Storage-Bucket erstellen.

Stellen Sie beim Erstellen eines Buckets sicher, dass Sie Objektereignisse ausgeben auswählen. Dies ist der Schlüssel für unsere ereignisgesteuerte Architektur.

Bild des Buckets

Konfigurieren Sie den OCI Connector-Hub, und erstellen Sie einen Datenfluss zwischen dem OCI Logging-Service und dem OCI Object Storage-Bucket. Weitere Informationen finden Sie unter Connector mit einer Loggingquelle erstellen.

Bild der SCH

Durch Anpassen der Batch-Rollover-Details können Sie die Häufigkeit konfigurieren, mit der die Logs in den OCI Object Storage-Bucket geschrieben werden. Der Standardwert beträgt 100 MB oder 7 Minuten.

Aufgabe 4: Funktion entwickeln und bereitstellen

Diese Funktion liest das Objekt aus OCI Object Storage und schreibt die Nachricht in den Stream. Um dies zu erreichen, führt es die folgenden Vorgänge dazwischen aus:

Lesen Sie das Objekt aus dem Bucket.
Dekomprimieren Sie das Objekt.
Prüfen Sie die Objektgröße, und erstellen Sie bei Bedarf 1 MB Chunks. Der OCI Streaming-Service begrenzt die maximale Größe einer eindeutigen Nachricht, die Produzenten in einem Stream veröffentlichen können, auf 1 MB.
Codieren Sie die Nachricht.
Im Stream veröffentlichen.

Weitere Informationen finden Sie unter Funktionen erstellen.

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

Bild der Funktion

Der letzte Schritt besteht darin, der Funktion mitzuteilen, wo sich der private Stream befindet. Diese Funktion verwendet Konfigurationsparameter, sodass sie wiederverwendbar ist, wenn Sie sie in einem anderen Mandanten bereitstellen möchten.

Bild der Funktionskonfiguration

Abonnieren Sie in dieser Aufgabe die Funktion für das Objektuploadereignis. Erstellen Sie eine Regel für den Ereignistyp als Objekt - Erstellen mit dem Bucket-Namen als bedingtes Attribut. Weitere Informationen finden Sie unter Ereignisregeln erstellen.

Bild der Veranstaltungskonfiguration

Verifizierung

Es gibt mehrere Stellen, an denen der Datenfluss überprüft werden kann.

Prüfen Sie die Loggruppenmetriken, um zu prüfen, ob die Flowlogs aufgenommen werden.
Der nächste Hop sind die Connector Hub-Metriken. OCI Connector Hub erfasst die Logs und sendet sie an den OCI Object Storage. Stellen Sie sicher, dass keine Fehler in Quelle und Ziel vorhanden sind.
Der nächste Hop ist OCI Object Storage. Stellen Sie sicher, dass die Objektanzahl steigt. Aktivieren Sie bei Bedarf die Lese- und Schreiblogs, um das Debugging fortzusetzen.
Der nächste Hop ist der OCI Events Service. Prüfen Sie die Metriken, um sicherzustellen, dass keine Zustellungsfehler vorliegen.
Im nächsten Schritt prüfen Sie die Metriken für den Aufruf der Funktion. Stellen Sie sicher, dass keine Fehler vorliegen und die Funktion nicht gedrosselt wird.
Prüfen Sie, ob die Daten im privaten Stream aufgenommen werden.

Wenn die Daten in einem der folgenden Diagramme fehlen, stoppen Sie den Vorgang, und aktivieren Sie Logs für diesen Service. In Logs wird erläutert, warum eine bestimmte Ressource die Aufgabe nicht ausführen kann.

Nächste Schritte

Herzlichen Glückwunsch zur erfolgreichen Implementierung einer sicheren und ereignisgesteuerten Protokollmanagementlösung in OCI! Durch die Kombination der Leistungsfähigkeit von OCI Logging, OCI Connector Hub, OCI Object Storage und privaten OCI-Streams haben Sie eine robuste Architektur erstellt, die sicherstellt, dass Ihre Logs nahezu in Echtzeit erfasst, verarbeitet und veröffentlicht werden.

Diese Lösung schützt sensible Protokolldaten über private Streams und demonstriert die Effizienz ereignisgesteuerter Automatisierung. Bei der Skalierung Ihres Systems passt sich diese Architektur nahtlos an, sodass Sie große Logvolumina mit minimalem manuellen Eingriff verarbeiten können.

Mit diesem Framework können Sie eine sichere und effiziente Protokollverarbeitung gewährleisten und gleichzeitig die Datenschutzanforderungen einhalten. Diese Architektur bietet Flexibilität beim Erstellen benutzerdefinierter Verarbeitungspipelines, die auf Ihre Unternehmensanforderungen zugeschnitten sind. Wenn Sie dieses Setup mit zusätzlichen Analyse- oder Alerting-Mechanismen erweitern, können Sie tiefere Einblicke in Systemereignisse erhalten und Ihre Fähigkeit verbessern, Anomalien proaktiv zu erkennen und darauf zu reagieren.

Weitere Informationen zur Verwendung von privaten OCI Functions- und OCI-Streamfunktionen erhalten Sie von Ihrem Oracle-Mitarbeiter oder unter Cloud-Sicherheitslösungen.

Danksagungen

Autor - Aneel Kanuri (Distinguished Cloud Architect)

Weitere Lernressourcen

Sehen Sie sich andere Übungen zu docs.oracle.com/learn an, oder greifen Sie im Oracle Learning YouTube-Channel auf weitere kostenlose Lerninhalte zu. Besuchen Sie außerdem education.oracle.com/learning-explorer, um Oracle Learning Explorer zu werden.

Die Produktdokumentation finden Sie im Oracle Help Center.

Titel und Copyright-Informationen

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

G23251-01

December 2024

Oracle und/oder verbundene Unternehmen.